Files
multica/server/pkg/db/generated/subscriber.sql.go
Jiayuan Zhang 28b6105edc fix(subscribers): notify the human an agent files sub-issues for (MUL-5483) (#6209)
When an agent created a sub-issue while working on a human's behalf, that human
received no notifications for it at all. issue_subscriber modelled ACTOR
identity, so an agent-created, agent-assigned issue had a full subscriber list
and zero members to deliver to. The platform already knew who the work was for
(agent_task_queue.originator_user_id, MUL-4302); notification never asked.

- attribution.DelegatedSubscriber: one shared rule over the same origin
  waterfall ClassifyDirect uses. agent_create subscribes the originator as
  'delegated'; quick_create keeps the direct 'creator' tier; autopilot and
  degraded attribution subscribe nobody.
- Delegated is a reduced delivery tier: in_review/done/cancelled/blocked plus
  failures and mentions. Routine churn is suppressed, and the parent bubble
  cannot re-deliver what the tier dropped.
- Unsubscribe becomes stateful: an unsubscribed_at tombstone survives later
  rule passes, and opt_out_scope distinguishes "this issue" from "this subtree"
  so a narrow opt-out no longer silently suppresses future children.
- Subtree unsubscribe is its own endpoint. A body flag cannot fail loudly
  against an older backend (Go drops unknown fields); an unknown route 404s,
  which the UI now surfaces with a distinct message.
- Eligibility and the write share one statement under a (workspace, user)
  advisory lock that subtree unsubscribe and member revoke also take, closing
  the check-then-insert races. Revoke additionally clears the departing
  member's subscriptions in the same tx.
- UI explains a delegated subscription and offers both unsubscribe scopes.

Migrations 249/250 add the delegated reason, the opt-out tombstone, and the
opt-out scope, using NOT VALID + VALIDATE CONSTRAINT so the widened CHECK does
not scan issue_subscriber under an exclusive lock.

Reviewed across eight rounds; an earlier write-time subtree roll-up was built
and then removed in full once it proved unfixable without serializing every
topology mutation. The parent's own status transition already carries that
signal.

Closes MUL-5483.
2026-07-31 16:52:17 +08:00

447 lines
16 KiB
Go

// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: subscriber.sql
package db
import (
"context"
"github.com/jackc/pgx/v5/pgtype"
)
const addDelegatedSubscriber = `-- name: AddDelegatedSubscriber :execrows
WITH RECURSIVE ancestors(node_id, parent_id, depth) AS (
SELECT root.id, root.parent_issue_id, 0 FROM issue root WHERE root.id = $1
UNION ALL
SELECT i.id, i.parent_issue_id, a.depth + 1
FROM issue i JOIN ancestors a ON i.id = a.parent_id
),
active_member AS (
SELECT m.user_id FROM member m
WHERE m.user_id = $2 AND m.workspace_id = $4
FOR SHARE
)
INSERT INTO issue_subscriber (issue_id, user_type, user_id, reason)
SELECT $1, 'member', am.user_id, $3
FROM active_member am
WHERE NOT EXISTS (
SELECT 1
FROM issue_subscriber s
JOIN ancestors a ON a.node_id = s.issue_id
WHERE s.user_type = 'member' AND s.user_id = $2
AND s.unsubscribed_at IS NOT NULL
AND (a.depth = 0 OR s.opt_out_scope = 'subtree')
)
ON CONFLICT (issue_id, user_type, user_id) DO UPDATE
SET reason = EXCLUDED.reason
WHERE issue_subscriber.unsubscribed_at IS NULL
AND issue_subscriber.reason = 'delegated'
AND EXCLUDED.reason <> 'delegated'
`
type AddDelegatedSubscriberParams struct {
IssueID pgtype.UUID `json:"issue_id"`
UserID pgtype.UUID `json:"user_id"`
Reason string `json:"reason"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
}
// The delegated rule's write (MUL-5483). Eligibility and insert are ONE
// statement so they share a snapshot; callers must hold LockSubscriberWrites
// for the same (workspace, user), which is what makes that snapshot stable.
//
// Eligibility is two conditions the previous version checked in separate
// round trips:
//
// 1. active_member — the originator must STILL be a workspace member.
// originator_user_id is stamped when the task is queued and never
// revisited, so a revoked member's still-running tasks would otherwise
// keep filing ghost subscribers. FOR SHARE also blocks a concurrent
// DELETE of that member row for the rest of this transaction, so the
// check cannot go stale between here and the insert.
//
// 2. NOT EXISTS (...) — the ancestor opt-out walk from HasAncestorOptOut,
// inlined for the same reason: a subtree tombstone committed between a
// separate check and this insert would silently undo the user's opt-out
// on every child the agent files next.
//
// ON CONFLICT keeps AddIssueSubscriber's two rules: never resurrect a
// tombstone, and upgrade an active 'delegated' row when the user becomes
// directly involved.
func (q *Queries) AddDelegatedSubscriber(ctx context.Context, arg AddDelegatedSubscriberParams) (int64, error) {
result, err := q.db.Exec(ctx, addDelegatedSubscriber,
arg.IssueID,
arg.UserID,
arg.Reason,
arg.WorkspaceID,
)
if err != nil {
return 0, err
}
return result.RowsAffected(), nil
}
const addIssueSubscriber = `-- name: AddIssueSubscriber :execrows
INSERT INTO issue_subscriber (issue_id, user_type, user_id, reason)
VALUES ($1, $2, $3, $4)
ON CONFLICT (issue_id, user_type, user_id) DO UPDATE
SET reason = EXCLUDED.reason
WHERE issue_subscriber.unsubscribed_at IS NULL
AND issue_subscriber.reason = 'delegated'
AND EXCLUDED.reason <> 'delegated'
`
type AddIssueSubscriberParams struct {
IssueID pgtype.UUID `json:"issue_id"`
UserType string `json:"user_type"`
UserID pgtype.UUID `json:"user_id"`
Reason string `json:"reason"`
}
// Auto-subscribe path (creator / assignee / commenter / mentioned / autopilot /
// delegated).
//
// Two behaviors are load-bearing here:
//
// 1. A tombstoned row is NEVER resurrected. The WHERE on the DO UPDATE fails
// for an opted-out row, which degrades to DO NOTHING — so unsubscribe still
// sticks on a tree an agent keeps adding to (MUL-5483).
//
// 2. An ACTIVE 'delegated' row is upgraded when the user becomes directly
// involved (assigned / mentioned / commented). Without this the reason
// stays 'delegated' forever and someone who is now a real participant keeps
// getting the reduced delivery tier.
//
// Returns rows affected so the caller only broadcasts subscriber:added when an
// active subscription actually changed. Publishing unconditionally made the
// frontend insert a subscriber the DB had refused to write.
func (q *Queries) AddIssueSubscriber(ctx context.Context, arg AddIssueSubscriberParams) (int64, error) {
result, err := q.db.Exec(ctx, addIssueSubscriber,
arg.IssueID,
arg.UserType,
arg.UserID,
arg.Reason,
)
if err != nil {
return 0, err
}
return result.RowsAffected(), nil
}
const deleteSubscriptionsByMember = `-- name: DeleteSubscriptionsByMember :exec
DELETE FROM issue_subscriber s
USING issue i
WHERE s.issue_id = i.id
AND i.workspace_id = $1
AND s.user_type = 'member'
AND s.user_id = $2
`
type DeleteSubscriptionsByMemberParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
UserID pgtype.UUID `json:"user_id"`
}
// Drop a departing member's subscriptions across the workspace, in the same tx
// as the member-row delete (MUL-5483 review round 7).
//
// Same application-layer cleanup rule the surrounding revoke path already
// applies to channel bindings and invocation grants: issue_subscriber carries
// no FK, so nothing removes these implicitly. Without it a revoked member keeps
// accruing inbox rows, and re-inviting them silently restores visibility of
// every issue they used to watch.
//
// A hard DELETE, not a tombstone: a tombstone means "this person chose to
// leave this issue" and would suppress re-subscription if they rejoin, which
// is not what a workspace revoke means.
func (q *Queries) DeleteSubscriptionsByMember(ctx context.Context, arg DeleteSubscriptionsByMemberParams) error {
_, err := q.db.Exec(ctx, deleteSubscriptionsByMember, arg.WorkspaceID, arg.UserID)
return err
}
const hasAncestorOptOut = `-- name: HasAncestorOptOut :one
WITH RECURSIVE ancestors(node_id, parent_id, depth) AS (
SELECT root.id, root.parent_issue_id, 0 FROM issue root WHERE root.id = $1
UNION ALL
SELECT i.id, i.parent_issue_id, a.depth + 1 FROM issue i JOIN ancestors a ON i.id = a.parent_id
)
SELECT EXISTS(
SELECT 1
FROM issue_subscriber s
JOIN ancestors a ON a.node_id = s.issue_id
WHERE s.user_type = $2 AND s.user_id = $3
AND s.unsubscribed_at IS NOT NULL
AND (a.depth = 0 OR s.opt_out_scope = 'subtree')
) AS opted_out
`
type HasAncestorOptOutParams struct {
ID pgtype.UUID `json:"id"`
UserType string `json:"user_type"`
UserID pgtype.UUID `json:"user_id"`
}
// True when the user has opted out in a way that should keep them off THIS
// issue.
//
// Two distinct cases, and the distinction is the whole point of opt_out_scope:
//
// - a tombstone on this issue itself, at any scope — they left this issue;
// - a 'subtree'-scoped tombstone on a STRICT ancestor — they left a tree this
// issue belongs to, so a child created an hour later is still covered.
//
// An 'issue'-scoped tombstone on an ancestor deliberately does NOT match: the
// user declined that one issue, not everything the agent files beneath it.
func (q *Queries) HasAncestorOptOut(ctx context.Context, arg HasAncestorOptOutParams) (bool, error) {
row := q.db.QueryRow(ctx, hasAncestorOptOut, arg.ID, arg.UserType, arg.UserID)
var opted_out bool
err := row.Scan(&opted_out)
return opted_out, err
}
const isIssueSubscriber = `-- name: IsIssueSubscriber :one
SELECT EXISTS(
SELECT 1 FROM issue_subscriber
WHERE issue_id = $1 AND user_type = $2 AND user_id = $3
AND unsubscribed_at IS NULL
) AS subscribed
`
type IsIssueSubscriberParams struct {
IssueID pgtype.UUID `json:"issue_id"`
UserType string `json:"user_type"`
UserID pgtype.UUID `json:"user_id"`
}
func (q *Queries) IsIssueSubscriber(ctx context.Context, arg IsIssueSubscriberParams) (bool, error) {
row := q.db.QueryRow(ctx, isIssueSubscriber, arg.IssueID, arg.UserType, arg.UserID)
var subscribed bool
err := row.Scan(&subscribed)
return subscribed, err
}
const listIssueSubscribers = `-- name: ListIssueSubscribers :many
SELECT issue_id, user_type, user_id, reason, created_at, unsubscribed_at, opt_out_scope FROM issue_subscriber
WHERE issue_id = $1 AND unsubscribed_at IS NULL
ORDER BY created_at
`
func (q *Queries) ListIssueSubscribers(ctx context.Context, issueID pgtype.UUID) ([]IssueSubscriber, error) {
rows, err := q.db.Query(ctx, listIssueSubscribers, issueID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []IssueSubscriber{}
for rows.Next() {
var i IssueSubscriber
if err := rows.Scan(
&i.IssueID,
&i.UserType,
&i.UserID,
&i.Reason,
&i.CreatedAt,
&i.UnsubscribedAt,
&i.OptOutScope,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const lockActiveMember = `-- name: LockActiveMember :one
SELECT id FROM member
WHERE user_id = $1 AND workspace_id = $2
FOR SHARE
`
type LockActiveMemberParams struct {
UserID pgtype.UUID `json:"user_id"`
WorkspaceID pgtype.UUID `json:"workspace_id"`
}
// Re-assert workspace membership INSIDE a serialized transaction, holding the
// row so a concurrent revoke cannot complete underneath the caller
// (MUL-5483 review round 8).
//
// The subtree-unsubscribe handler validates membership from its own MVCC
// snapshot before it opens a transaction, which is only a statement about the
// past. A revoke that commits in between clears the user's subscriptions and
// deletes the member row, and the handler would then write a fresh tombstone
// behind it — leaving an opt-out that outlives the membership and is inherited
// if the person is ever re-invited.
func (q *Queries) LockActiveMember(ctx context.Context, arg LockActiveMemberParams) (pgtype.UUID, error) {
row := q.db.QueryRow(ctx, lockActiveMember, arg.UserID, arg.WorkspaceID)
var id pgtype.UUID
err := row.Scan(&id)
return id, err
}
const lockSubscriberWrites = `-- name: LockSubscriberWrites :exec
SELECT pg_advisory_xact_lock(
hashtext(($1::uuid)::text),
hashtext(($2::uuid)::text)
)
`
type LockSubscriberWritesParams struct {
WorkspaceID pgtype.UUID `json:"workspace_id"`
UserID pgtype.UUID `json:"user_id"`
}
// Transaction-scoped serialization boundary for (workspace, user) subscriber
// state (MUL-5483 review round 7).
//
// Three paths race over the same question — "should this user be an active
// watcher?" — and each one is a check in one statement followed by a write in
// another:
//
// - the delegated rule: is the user a member / not opted out, then insert;
// - subtree unsubscribe: tombstone the tree, which must cover children the
// agent files a moment later;
// - member revoke: delete the member row and their subscriptions.
//
// Row locks cannot close this. The interleaving that breaks the escape hatch
// inserts a subscriber for an issue that did not exist when the opt-out was
// written, so there is no row to lock — a phantom, which under READ COMMITTED
// needs an explicit lock object. Every path above takes THIS lock first, so
// their lock ordering is identical and they cannot deadlock against each other.
//
// Keyed on (workspace, user) rather than the issue: an opt-out is about a
// person and a tree, and the tree's membership is exactly what is changing.
//
// The key is derived from the UUID VALUE, not from however the caller happened
// to spell it. ::uuid parses the input and the outer ::text renders PostgreSQL's
// canonical lowercase form, so 'AB…' and 'ab…' — the same UUID everywhere else
// in this file, and to every membership query — also produce the same lock.
// Hashing the raw string instead let an uppercase request take a DIFFERENT lock
// from the canonical holder, which silently reopened the very race this lock
// exists to close. Casting at the DB boundary keeps that guarantee even if a
// future caller forgets to canonicalize.
func (q *Queries) LockSubscriberWrites(ctx context.Context, arg LockSubscriberWritesParams) error {
_, err := q.db.Exec(ctx, lockSubscriberWrites, arg.WorkspaceID, arg.UserID)
return err
}
const removeIssueSubscriber = `-- name: RemoveIssueSubscriber :exec
UPDATE issue_subscriber
SET unsubscribed_at = now(), opt_out_scope = 'issue'
WHERE issue_id = $1 AND user_type = $2 AND user_id = $3 AND unsubscribed_at IS NULL
`
type RemoveIssueSubscriberParams struct {
IssueID pgtype.UUID `json:"issue_id"`
UserType string `json:"user_type"`
UserID pgtype.UUID `json:"user_id"`
}
// Leave THIS issue only. Tombstone rather than delete, so auto-subscribe rules
// can tell "never subscribed" (no row) from "chose to leave" (row with
// unsubscribed_at). scope='issue' keeps the opt-out from reaching descendants:
// future children of this issue are still allowed to subscribe the user.
func (q *Queries) RemoveIssueSubscriber(ctx context.Context, arg RemoveIssueSubscriberParams) error {
_, err := q.db.Exec(ctx, removeIssueSubscriber, arg.IssueID, arg.UserType, arg.UserID)
return err
}
const subscribeToIssueExplicitly = `-- name: SubscribeToIssueExplicitly :exec
INSERT INTO issue_subscriber (issue_id, user_type, user_id, reason)
VALUES ($1, $2, $3, $4)
ON CONFLICT (issue_id, user_type, user_id)
DO UPDATE SET unsubscribed_at = NULL, opt_out_scope = NULL, reason = EXCLUDED.reason
`
type SubscribeToIssueExplicitlyParams struct {
IssueID pgtype.UUID `json:"issue_id"`
UserType string `json:"user_type"`
UserID pgtype.UUID `json:"user_id"`
Reason string `json:"reason"`
}
// Explicit user action (the Subscribe button). Unlike the rule-driven path this
// CLEARS an existing opt-out tombstone and its scope: the user is overriding
// their own earlier unsubscribe, which is the one thing that should bring them
// back.
func (q *Queries) SubscribeToIssueExplicitly(ctx context.Context, arg SubscribeToIssueExplicitlyParams) error {
_, err := q.db.Exec(ctx, subscribeToIssueExplicitly,
arg.IssueID,
arg.UserType,
arg.UserID,
arg.Reason,
)
return err
}
const unsubscribeFromIssueSubtree = `-- name: UnsubscribeFromIssueSubtree :many
WITH RECURSIVE subtree(node_id) AS (
SELECT root.id FROM issue root WHERE root.id = $1
UNION ALL
SELECT i.id FROM issue i JOIN subtree s ON i.parent_issue_id = s.node_id
),
retire_descendants AS (
UPDATE issue_subscriber sub
SET unsubscribed_at = now(), opt_out_scope = 'subtree'
WHERE sub.issue_id IN (SELECT node_id FROM subtree WHERE node_id <> $1)
AND sub.user_type = $2 AND sub.user_id = $3
AND sub.unsubscribed_at IS NULL
RETURNING sub.issue_id
),
retire_root AS (
INSERT INTO issue_subscriber (issue_id, user_type, user_id, reason, unsubscribed_at, opt_out_scope)
VALUES ($1, $2, $3, 'manual', now(), 'subtree')
ON CONFLICT (issue_id, user_type, user_id)
DO UPDATE SET unsubscribed_at = now(), opt_out_scope = 'subtree'
RETURNING issue_id
)
SELECT issue_id FROM retire_descendants
UNION ALL
SELECT issue_id FROM retire_root
`
type UnsubscribeFromIssueSubtreeParams struct {
ID pgtype.UUID `json:"id"`
UserType string `json:"user_type"`
UserID pgtype.UUID `json:"user_id"`
}
// Leave an issue AND every descendant in one action. An agent-built tree is
// the unit a user actually wants to stop watching; leaving 30 sub-issues one
// at a time is not a real escape hatch.
//
// The root gets an UPSERTED tombstone rather than a conditional UPDATE, because
// "I don't want this tree" has to persist even when the user holds no
// subscription on the root itself — otherwise the opt-out records nothing and
// the next child the agent files re-subscribes them. That root tombstone is
// also what covers FUTURE descendants: HasAncestorOptOut walks up to it and
// honors it because its scope is 'subtree'.
//
// Returns every issue id it actually tombstoned so the caller can broadcast one
// subscriber:removed per issue; publishing only the root left other open tabs
// showing a stale subscription on the children.
func (q *Queries) UnsubscribeFromIssueSubtree(ctx context.Context, arg UnsubscribeFromIssueSubtreeParams) ([]pgtype.UUID, error) {
rows, err := q.db.Query(ctx, unsubscribeFromIssueSubtree, arg.ID, arg.UserType, arg.UserID)
if err != nil {
return nil, err
}
defer rows.Close()
items := []pgtype.UUID{}
for rows.Next() {
var issue_id pgtype.UUID
if err := rows.Scan(&issue_id); err != nil {
return nil, err
}
items = append(items, issue_id)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}