package message import ( "context" "database/sql" "errors" "fmt" "strings" "time" "github.com/tidbcloud/AgentTeam/internal/coredb" ) type scopeStoreBase struct { execer coredb.QueryExecer } // ScopeStore owns authoritative scope topology and profile persistence. type ScopeStore struct{ scopeStoreBase } // MembershipStore owns scope membership and read-state persistence. type MembershipStore struct{ scopeStoreBase } // NotificationStore owns scope notification-state persistence. type NotificationStore struct{ scopeStoreBase } // MessageStore owns authoritative scope message writes and history-page queries. type MessageStore struct{ scopeStoreBase } // SourceStore owns source-ref persistence. type SourceStore struct{ scopeStoreBase } // ActivityEventStore owns non-message scope activity events such as reactions. // Message rows are already the message activity source and must not be copied // into this store. type ActivityEventStore struct{ scopeStoreBase } // PrivateVisibilityStore owns private authority grant facts. type PrivateVisibilityStore struct{ scopeStoreBase } // OutboxStore owns post-transaction fanout outbox facts. type OutboxStore struct{ scopeStoreBase } // SyncDirtyStore owns sparse per-member changed-scope markers for opaque // message sync cursors. type SyncDirtyStore struct{ scopeStoreBase } // ReactionStore owns scope reaction facts. type ReactionStore struct{ scopeStoreBase } // PinStore owns scope message pin and per-viewer banner hide facts. type PinStore struct{ scopeStoreBase } // SearchStore owns scope search projection facts. type SearchStore struct{ scopeStoreBase } // ShareCardStore owns channel-share-card message action state in scope form. type ShareCardStore struct{ scopeStoreBase } // TaskContextIngressStore owns task-native ingress queries for receive. // // T-B1 only reserves this owner boundary. T-D2 will add task-native reads here; // this store must not synthesize task ingress from message rows or non-message // activity rows. type TaskContextIngressStore struct{ scopeStoreBase } // ScopeRepositories groups the scope repository owners so workers can inject // only the ports needed for their task card. type ScopeRepositories struct { Scopes ScopeStore Memberships MembershipStore Notifications NotificationStore Messages MessageStore Sources SourceStore ActivityEvents ActivityEventStore PrivateVisibility PrivateVisibilityStore Outbox OutboxStore SyncDirty SyncDirtyStore Reactions ReactionStore Pins PinStore Search SearchStore ShareCards ShareCardStore TaskContextIngress TaskContextIngressStore } func NewScopeRepositories(execer coredb.QueryExecer) ScopeRepositories { base := scopeStoreBase{execer: execer} return ScopeRepositories{ Scopes: ScopeStore{scopeStoreBase: base}, Memberships: MembershipStore{scopeStoreBase: base}, Notifications: NotificationStore{scopeStoreBase: base}, Messages: MessageStore{scopeStoreBase: base}, Sources: SourceStore{scopeStoreBase: base}, ActivityEvents: ActivityEventStore{scopeStoreBase: base}, PrivateVisibility: PrivateVisibilityStore{scopeStoreBase: base}, Outbox: OutboxStore{scopeStoreBase: base}, SyncDirty: SyncDirtyStore{scopeStoreBase: base}, Reactions: ReactionStore{scopeStoreBase: base}, Pins: PinStore{scopeStoreBase: base}, Search: SearchStore{scopeStoreBase: base}, ShareCards: ShareCardStore{scopeStoreBase: base}, TaskContextIngress: TaskContextIngressStore{scopeStoreBase: base}, } } func NewScopeStore(execer coredb.QueryExecer) ScopeStore { return ScopeStore{scopeStoreBase: scopeStoreBase{execer: execer}} } type scopeQueryDescriptor struct { Name string TargetIndex string AmplificationRisk string SQL string } var ( scopeMessagesBeforeSeqQuery = scopeQueryDescriptor{ Name: "scope_messages_before_seq", TargetIndex: "scope_messages_scope_seq_key(scope_id, seq)", AmplificationRisk: "bounded by one scope and caller limit; used by history pages", SQL: ` SELECT id, workspace_id, scope_id, seq, workspace_global_seq, author_member_id, author_type, COALESCE(NULLIF(notify_principal_member_id, ''), author_member_id), COALESCE(NULLIF(idempotency_subject_member_id, ''), author_member_id), COALESCE(client_message_id, ''), COALESCE(idempotency_payload_digest, ''), message_type, content, metadata, created_at, updated_at, edited_at, COALESCE(edited_by_member_id, ''), edit_revision FROM scope_messages WHERE scope_id = $1 AND seq < $2 ORDER BY seq DESC LIMIT $3 `, } scopeActivityEventsAfterCursorQuery = scopeQueryDescriptor{ Name: "scope_activity_events_after_cursor", TargetIndex: "scope_activity_events_scope_seq_key(scope_id, seq)", AmplificationRisk: "bounded by one workspace/scope-local activity cursor and caller limit; used by non-message activity pagination", SQL: ` SELECT id, workspace_id, scope_id, seq, activity_type, activity_ref_type, activity_ref_id, activity_global_seq, COALESCE(source_message_id, ''), COALESCE(actor_member_id, ''), reason, payload, created_at, suppressed_at FROM scope_activity_events WHERE workspace_id = $1 AND ($2 = '' OR scope_id = $2) AND seq > $3 AND suppressed_at IS NULL ORDER BY seq ASC, created_at ASC, id ASC LIMIT $4 `, } scopePendingFanoutOutboxQuery = scopeQueryDescriptor{ Name: "scope_pending_fanout_outbox", TargetIndex: "scope_fanout_outbox_pending_idx(status, next_attempt_at, id)", AmplificationRisk: "bounded by pending status, due timestamp, and caller limit; used by outbox worker leasing", SQL: ` SELECT id, workspace_id, scope_id, COALESCE(message_id, ''), event_ref_type, event_ref_id, fanout_type, dedupe_key, payload, status, attempts, max_attempts, next_attempt_at, COALESCE(lease_owner, ''), lease_expires_at, created_at, delivered_at, suppressed_at, suppression_reason, last_error FROM scope_fanout_outbox WHERE status IN ('pending', 'pending_v2') AND next_attempt_at <= CURRENT_TIMESTAMP ORDER BY next_attempt_at ASC, id ASC LIMIT $1 `, } scopeLeasePendingFanoutOutboxQuery = scopeQueryDescriptor{ Name: "scope_lease_pending_fanout_outbox", TargetIndex: "scope_fanout_outbox_pending_idx(status, next_attempt_at, id)", AmplificationRisk: "bounded by pending status, due timestamp, and caller limit; used by one outbox worker lease attempt", SQL: ` SELECT id, workspace_id, scope_id, COALESCE(message_id, ''), event_ref_type, event_ref_id, fanout_type, dedupe_key, payload, status, attempts, max_attempts, next_attempt_at, COALESCE(lease_owner, ''), lease_expires_at, created_at, delivered_at, suppressed_at, suppression_reason, last_error FROM scope_fanout_outbox WHERE status IN ('pending', 'pending_v2') AND next_attempt_at <= $1 ORDER BY next_attempt_at ASC, id ASC LIMIT $2 `, } scopeLeasePendingFanoutOutboxForWorkspaceGroupQuery = scopeQueryDescriptor{ Name: "scope_lease_pending_fanout_outbox_for_workspace_group", TargetIndex: "scope_fanout_outbox_pending_idx(status, next_attempt_at, id) + workspace_placements_group_state_idx(workspace_group_id, state, workspace_id)", AmplificationRisk: "bounded by one workspace placement group, pending status, due timestamp, and caller limit; used by one workspace-profile outbox worker lease attempt", SQL: ` SELECT outbox.id, outbox.workspace_id, outbox.scope_id, COALESCE(outbox.message_id, ''), outbox.event_ref_type, outbox.event_ref_id, outbox.fanout_type, outbox.dedupe_key, outbox.payload, outbox.status, outbox.attempts, outbox.max_attempts, outbox.next_attempt_at, COALESCE(outbox.lease_owner, ''), outbox.lease_expires_at, outbox.created_at, outbox.delivered_at, outbox.suppressed_at, outbox.suppression_reason, outbox.last_error FROM scope_fanout_outbox outbox JOIN workspace_placements placement ON placement.workspace_id = outbox.workspace_id AND placement.workspace_group_id = $2 AND placement.state IN ('active', 'draining', 'moving') WHERE outbox.status IN ('pending', 'pending_v2') AND outbox.next_attempt_at <= $1 ORDER BY outbox.next_attempt_at ASC, outbox.id ASC LIMIT $3 `, } scopeLeaseExpiredFanoutOutboxQuery = scopeQueryDescriptor{ Name: "scope_lease_expired_fanout_outbox", TargetIndex: "scope_fanout_outbox_lease_idx(lease_expires_at, status)", AmplificationRisk: "bounded by expired lease timestamp, leased status, and caller limit; used by one outbox worker lease attempt", SQL: ` SELECT id, workspace_id, scope_id, COALESCE(message_id, ''), event_ref_type, event_ref_id, fanout_type, dedupe_key, payload, status, attempts, max_attempts, next_attempt_at, COALESCE(lease_owner, ''), lease_expires_at, created_at, delivered_at, suppressed_at, suppression_reason, last_error FROM scope_fanout_outbox WHERE status IN ('leased', 'leased_v2') AND lease_expires_at <= $1 ORDER BY next_attempt_at ASC, id ASC LIMIT $2 `, } scopeLeaseExpiredFanoutOutboxForWorkspaceGroupQuery = scopeQueryDescriptor{ Name: "scope_lease_expired_fanout_outbox_for_workspace_group", TargetIndex: "scope_fanout_outbox_lease_idx(lease_expires_at, status) + workspace_placements_group_state_idx(workspace_group_id, state, workspace_id)", AmplificationRisk: "bounded by one workspace placement group, expired lease timestamp, leased status, and caller limit; used by one workspace-profile outbox worker lease attempt", SQL: ` SELECT outbox.id, outbox.workspace_id, outbox.scope_id, COALESCE(outbox.message_id, ''), outbox.event_ref_type, outbox.event_ref_id, outbox.fanout_type, outbox.dedupe_key, outbox.payload, outbox.status, outbox.attempts, outbox.max_attempts, outbox.next_attempt_at, COALESCE(outbox.lease_owner, ''), outbox.lease_expires_at, outbox.created_at, outbox.delivered_at, outbox.suppressed_at, outbox.suppression_reason, outbox.last_error FROM scope_fanout_outbox outbox JOIN workspace_placements placement ON placement.workspace_id = outbox.workspace_id AND placement.workspace_group_id = $2 AND placement.state IN ('active', 'draining', 'moving') WHERE outbox.status IN ('leased', 'leased_v2') AND outbox.lease_expires_at <= $1 ORDER BY outbox.next_attempt_at ASC, outbox.id ASC LIMIT $3 `, } scopeMarkFanoutOutboxLeasedQuery = scopeQueryDescriptor{ Name: "scope_mark_fanout_outbox_leased", TargetIndex: "scope_fanout_outbox PRIMARY(id) + scope_fanout_outbox_pending_idx/status lease indexes", AmplificationRisk: "bounded by one explicit outbox id selected by the lease query", SQL: ` UPDATE scope_fanout_outbox SET status = CASE WHEN status IN ('pending_v2', 'leased_v2') THEN 'leased_v2' ELSE 'leased' END, attempts = attempts + 1, lease_owner = $2, lease_expires_at = $3, last_error = '' WHERE id = $1 AND ((status IN ('pending', 'pending_v2') AND next_attempt_at <= $4) OR (status IN ('leased', 'leased_v2') AND lease_expires_at <= $4)) `, } scopeMessageFanoutOutboxIDQuery = scopeQueryDescriptor{ Name: "scope_message_fanout_outbox_id", TargetIndex: "scope_fanout_outbox_event_ref_idx(event_ref_type, event_ref_id, fanout_type)", AmplificationRisk: "bounded by one authoritative message source and fanout type before an inline lease claim", SQL: ` SELECT id FROM scope_fanout_outbox WHERE event_ref_type = 'message' AND event_ref_id = $1 AND fanout_type = 'scope_fanout' LIMIT 1 `, } scopeMessageFanoutOutboxPayloadTargetCountQuery = scopeQueryDescriptor{ Name: "scope_message_fanout_outbox_payload_target_count", TargetIndex: "scope_fanout_outbox_event_ref_idx(event_ref_type, event_ref_id, fanout_type)", AmplificationRisk: "always-on payload-target preflight, plus a zero-row idempotency recheck; index-bounded by one authoritative message source and fanout type", SQL: ` SELECT COUNT(*) FROM scope_fanout_outbox WHERE message_id = $1 AND event_ref_type = 'message' AND event_ref_id = $1 AND fanout_type = 'scope_fanout' `, } scopeMarkFanoutOutboxInlineLeasedQuery = scopeQueryDescriptor{ Name: "scope_mark_fanout_outbox_inline_leased", TargetIndex: "scope_fanout_outbox PRIMARY(id)", AmplificationRisk: "bounded by one authoritative message outbox id after the synchronous writer commits", SQL: ` UPDATE scope_fanout_outbox SET status = CASE WHEN status IN ('pending_v2', 'leased_v2') THEN 'leased_v2' ELSE 'leased' END, attempts = attempts + 1, lease_owner = $2, lease_expires_at = $3, last_error = '' WHERE id = $1 AND (status IN ('pending', 'pending_v2') OR (status IN ('leased', 'leased_v2') AND lease_expires_at <= $4)) `, } scopeRenewFanoutOutboxLeaseQuery = scopeQueryDescriptor{ Name: "scope_renew_fanout_outbox_lease", TargetIndex: "scope_fanout_outbox PRIMARY(id)", AmplificationRisk: "bounded by one leased outbox id immediately before an external fanout effect", SQL: ` UPDATE scope_fanout_outbox SET lease_expires_at = $3 WHERE id = $1 AND status IN ('leased', 'leased_v2') AND lease_owner = $2 `, } scopeMarkFanoutOutboxDeliveredQuery = scopeQueryDescriptor{ Name: "scope_mark_fanout_outbox_delivered", TargetIndex: "scope_fanout_outbox PRIMARY(id)", AmplificationRisk: "bounded by one leased outbox id after adapter success", SQL: ` UPDATE scope_fanout_outbox SET status = 'delivered', delivered_at = $2, lease_owner = NULL, lease_expires_at = NULL, last_error = '' WHERE id = $1 AND status IN ('leased', 'leased_v2') AND lease_owner = $3 `, } scopeMarkFanoutOutboxRetryQuery = scopeQueryDescriptor{ Name: "scope_mark_fanout_outbox_retry_or_failed", TargetIndex: "scope_fanout_outbox PRIMARY(id)", AmplificationRisk: "bounded by one leased outbox id after adapter failure", SQL: ` UPDATE scope_fanout_outbox SET status = CASE WHEN attempts >= max_attempts THEN 'failed' WHEN status = 'leased_v2' THEN 'pending_v2' ELSE 'pending' END, next_attempt_at = CASE WHEN attempts >= max_attempts THEN next_attempt_at ELSE $2 END, lease_owner = NULL, lease_expires_at = NULL, last_error = $3 WHERE id = $1 AND status IN ('leased', 'leased_v2') AND lease_owner = $4 `, } scopeMarkFanoutOutboxDeferredQuery = scopeQueryDescriptor{ Name: "scope_mark_fanout_outbox_deferred", TargetIndex: "scope_fanout_outbox PRIMARY(id)", AmplificationRisk: "bounded by one leased v2 outbox id waiting for owner-local prerequisite state", SQL: ` UPDATE scope_fanout_outbox SET status = 'pending_v2', attempts = GREATEST(attempts - 1, 0), next_attempt_at = $2, lease_owner = NULL, lease_expires_at = NULL, last_error = $3 WHERE id = $1 AND status = 'leased_v2' AND lease_owner = $4 `, } scopeMarkFanoutOutboxSuppressedQuery = scopeQueryDescriptor{ Name: "scope_mark_fanout_outbox_suppressed", TargetIndex: "scope_fanout_outbox PRIMARY(id)", AmplificationRisk: "bounded by one leased outbox id after suppression secondary check", SQL: ` UPDATE scope_fanout_outbox SET status = 'suppressed', suppressed_at = $2, suppression_reason = $3, lease_owner = NULL, lease_expires_at = NULL WHERE id = $1 AND status IN ('leased', 'leased_v2') AND lease_owner = $4 `, } scopeLeasedFanoutOutboxByOwnerQuery = scopeQueryDescriptor{ Name: "scope_leased_fanout_outbox_by_owner", TargetIndex: "scope_fanout_outbox PRIMARY(id)", AmplificationRisk: "bounded by one explicit outbox id and lease owner immediately before fanout", SQL: ` SELECT id, workspace_id, scope_id, COALESCE(message_id, ''), event_ref_type, event_ref_id, fanout_type, dedupe_key, payload, status, attempts, max_attempts, next_attempt_at, COALESCE(lease_owner, ''), lease_expires_at, created_at, delivered_at, suppressed_at, suppression_reason, last_error FROM scope_fanout_outbox WHERE id = $1 AND status IN ('leased', 'leased_v2') AND lease_owner = $2 LIMIT 1 `, } scopeMessageFanoutOutboxForMutationQuery = scopeQueryDescriptor{ Name: "scope_message_fanout_outbox_for_mutation", TargetIndex: "scope_fanout_outbox_event_ref_idx(event_ref_type, event_ref_id, fanout_type)", AmplificationRisk: "bounded by one authoritative message source and fanout type inside the message mutation transaction", SQL: ` SELECT id, payload, status FROM scope_fanout_outbox WHERE event_ref_type = 'message' AND event_ref_id = $1 AND fanout_type = 'scope_fanout' LIMIT 1 FOR UPDATE `, } scopeMessagesVisibleToMemberQuery = scopeQueryDescriptor{ Name: "scope_messages_visible_to_member", TargetIndex: "scope_messages_scope_seq_key(scope_id, seq) + scope_memberships PRIMARY(scope_id, member_id) + members PRIMARY(id) + scope_private_grants_scope_owner_idx + task_scope_executions_scope_key + task_scope_participant_metadata_execution_member_idx + agents + scope_memberships PRIMARY(scope_id, member_id)", AmplificationRisk: "bounded by one scope, one viewer, caller batch limit, and the SQL-readable predicate; budget exhaustion returns HasMore with NextCursor continuation", SQL: ` SELECT m.id, m.workspace_id, m.scope_id, m.seq, m.workspace_global_seq, m.author_member_id, m.author_type, COALESCE(NULLIF(m.notify_principal_member_id, ''), m.author_member_id), COALESCE(NULLIF(m.idempotency_subject_member_id, ''), m.author_member_id), COALESCE(m.client_message_id, ''), COALESCE(m.idempotency_payload_digest, ''), m.message_type, m.content, m.metadata, m.created_at, m.updated_at, m.edited_at, COALESCE(m.edited_by_member_id, ''), m.edit_revision, sm.scope_id, sm.member_id, sm.visibility, COALESCE(sm.owner_member_id, ''), sm.last_read_seq, sm.last_read_global_seq, sm.last_activity_global_seq, sm.owner_context_dismissed_through_global_seq, sm.effective_from_seq, sm.effective_to_seq, sm.created_at, sm.updated_at, sm.removed_at, sm.archived_at, CASE WHEN ` + scopePrivateAuthorityGrantIDSQL + ` <> '' THEN ` + scopePrivateAuthorityGrantIDSQL + ` ELSE '' END, ` + scopeReadableMessageReasonSQL + `, '' FROM scope_messages m JOIN scopes message_scope ON message_scope.id = m.scope_id AND message_scope.workspace_id = m.workspace_id ` + scopeReadableMembershipJoinSQL("$2") + ` ` + scopePrivateAuthorityGrantJoinsSQL + ` WHERE m.scope_id = $1 AND m.seq < $3 AND ` + scopeReadableMessagePredicateSQL + ` ORDER BY m.seq DESC LIMIT $4 `, } scopeAdvanceMembershipReadStateQuery = scopeQueryDescriptor{ Name: "scope_advance_membership_read_state", TargetIndex: "scope_memberships PRIMARY(scope_id, member_id, effective_from_seq)", AmplificationRisk: "bounded by one scope membership row; used by receive/history cursor advancement", SQL: ` UPDATE scope_memberships SET unread_generation = CASE WHEN GREATEST(last_read_seq, $3) <> last_read_seq OR GREATEST(last_read_global_seq, $4) <> last_read_global_seq THEN unread_generation + 1 ELSE unread_generation END, last_read_seq = GREATEST(last_read_seq, $3), last_read_global_seq = GREATEST(last_read_global_seq, $4), updated_at = $5 WHERE scope_id = $1 AND member_id = $2 AND removed_at IS NULL AND effective_to_seq IS NULL `, } scopeActiveMembershipByIDQuery = scopeQueryDescriptor{ Name: "scope_active_membership_by_id", TargetIndex: "scope_memberships PRIMARY(scope_id, member_id, effective_from_seq)", AmplificationRisk: "bounded by one scope membership row; used by agent-facing history read-state reporting before optional advancement", SQL: ` SELECT scope_id, member_id, visibility, COALESCE(owner_member_id, ''), last_read_seq, last_read_global_seq, last_activity_global_seq, owner_context_dismissed_through_global_seq, effective_from_seq, effective_to_seq, created_at, updated_at, removed_at, archived_at FROM scope_memberships WHERE scope_id = $1 AND member_id = $2 AND removed_at IS NULL AND effective_to_seq IS NULL ORDER BY effective_from_seq DESC LIMIT 1 `, } scopeActivityEventByIDQuery = scopeQueryDescriptor{ Name: "scope_activity_event_by_id", TargetIndex: "scope_activity_events PRIMARY(id)", AmplificationRisk: "bounded by one explicit non-message activity event id referenced by fanout outbox", SQL: ` SELECT id, workspace_id, scope_id, seq, activity_type, activity_ref_type, activity_ref_id, activity_global_seq, COALESCE(source_message_id, ''), COALESCE(actor_member_id, ''), reason, payload, created_at, suppressed_at FROM scope_activity_events WHERE id = $1 `, } ) type ScopeRecord struct { ID string WorkspaceID string Type string ParentScopeID string RootMessageID string RootMessageShortID string IdentityKey string LatestSeq int64 LatestGlobalSeq int64 LatestActivitySeq int64 LatestMessageAt *time.Time VisibleMessageCount int64 VisibleLastMessageAt *time.Time ArchivedUntilSeq int64 Status string CreatedByMemberID string CreatedAt time.Time UpdatedAt time.Time ArchivedAt *time.Time } type ScopeChannelProfileRecord struct { ScopeID string WorkspaceID string Slug string DisplayName string Description string Visibility string Subsystem string OwnerMemberID string IsAllChannel bool CreatedAt time.Time UpdatedAt time.Time } type ScopeMembershipRecord struct { ScopeID string MemberID string Visibility string OwnerMemberID string LastReadSeq int64 LastReadGlobalSeq int64 LastActivityGlobalSeq int64 OwnerContextDismissedThroughGlobalSeq int64 EffectiveFromSeq int64 EffectiveToSeq *int64 CreatedAt time.Time UpdatedAt time.Time RemovedAt *time.Time ArchivedAt *time.Time } type ScopeNotificationSettingRecord struct { ScopeID string MemberID string State string CreatedAt time.Time UpdatedAt time.Time } type ScopePrivateGrantRecord struct { ID string WorkspaceID string ScopeID string OwnerMemberID string GranteeMemberID string GrantedByMemberID string CreatedAt time.Time ExpiresAt *time.Time RevokedAt *time.Time } type ScopeMessageRecord struct { ID string WorkspaceID string ScopeID string Seq int64 WorkspaceGlobalSeq int64 AuthorMemberID string AuthorType string NotifyPrincipalMemberID string IdempotencySubjectMemberID string ClientMessageID string IdempotencyPayloadDigest string MessageType string Content string Metadata []byte CreatedAt time.Time UpdatedAt time.Time EditedAt *time.Time EditedByMemberID string EditRevision int64 } type ScopeMessageInsertResult struct { Inserted bool Message ScopeMessageRecord } type ScopeMessageContentUpdate struct { WorkspaceID string ScopeID string MessageID string Content string Metadata []byte EditedByMemberID string EditedAt time.Time } type ScopeMessageRecallUpdate struct { WorkspaceID string ScopeID string MessageID string Content string Metadata []byte RecalledByMemberID string ExpectedAuthorType string ExpectedAuthorID string ExpectedMessageType string RecalledAt time.Time } type ScopeMessageVisibilityCandidate struct { Message ScopeMessageRecord Membership ScopeMembershipRecord GrantID string Reason string SearchRank int } type ScopeMessageSourceRecord struct { TargetScopeID string MessageID string Ordinal int SourceScopeID string SourceSeq int64 SourceMessageID string SourceKind string SourceDedupeKey string CreatedAt time.Time } type ScopeActivityEventRecord struct { ID string WorkspaceID string ScopeID string Seq int64 ActivityType string ActivityRefType string ActivityRefID string ActivityGlobalSeq int64 SourceMessageID string ActorMemberID string Reason string Payload []byte CreatedAt time.Time SuppressedAt *time.Time } type ScopeFanoutOutboxRecord struct { ID string WorkspaceID string ScopeID string MessageID string EventRefType string EventRefID string FanoutType string DedupeKey string Payload []byte Status string Attempts int MaxAttempts int NextAttemptAt time.Time LeaseOwner string LeaseExpiresAt *time.Time CreatedAt time.Time DeliveredAt *time.Time SuppressedAt *time.Time SuppressionReason string LastError string } type ScopeMessageReactionRecord struct { ID string WorkspaceID string ScopeID string MessageID string Emoji string ReactorMemberID string ActivitySeq int64 ActivityGlobalSeq int64 CreatedAt time.Time } type ScopeSearchDocRecord struct { MessageID string WorkspaceID string ScopeID string GlobalSeq int64 Content string CreatedAt time.Time } type ScopeSearchStateRecord struct { WorkspaceID string ScopeID string ProjectionName string MigrationStatus string BackfillStatus string BackfillCursorGlobalSeq int64 FinalizeStatus string ReadReady bool CreatedAt time.Time UpdatedAt time.Time } type ScopeShareCardRecord struct { ID string WorkspaceID string ScopeID string MessageID string SourceScopeID string SourceMessageID string TargetScopeID string CreatedByMemberID string RevokedAt *time.Time RevokedByMemberID string Payload []byte CreatedAt time.Time UpdatedAt time.Time } func (store ScopeStore) UpsertScope(ctx context.Context, record ScopeRecord) error { if err := store.requireExecer(); err != nil { return err } if strings.TrimSpace(record.ID) == "" || strings.TrimSpace(record.WorkspaceID) == "" || strings.TrimSpace(record.Type) == "" { return invalidArgument("scope id, workspace id, and type are required") } if _, err := store.execer.Exec(ctx, ` INSERT INTO scopes ( id, workspace_id, type, parent_scope_id, root_message_id, root_message_short_id, identity_key, latest_seq, latest_global_seq, latest_activity_seq, latest_message_at, visible_message_count, visible_last_message_at, archived_until_seq, status, created_by_member_id, created_at, updated_at, archived_at ) VALUES ($1, $2, $3, NULLIF($4, ''), NULLIF($5, ''), NULLIF($6, ''), NULLIF($7, ''), $8, $9, $10, $11, $12, $13, $14, $15, NULLIF($16, ''), $17, $18, $19) ON DUPLICATE KEY UPDATE parent_scope_id = VALUES(parent_scope_id), root_message_id = VALUES(root_message_id), root_message_short_id = VALUES(root_message_short_id), identity_key = VALUES(identity_key), latest_seq = GREATEST(scopes.latest_seq, VALUES(latest_seq)), latest_global_seq = GREATEST(scopes.latest_global_seq, VALUES(latest_global_seq)), latest_activity_seq = GREATEST(scopes.latest_activity_seq, VALUES(latest_activity_seq)), latest_message_at = COALESCE(GREATEST(scopes.latest_message_at, VALUES(latest_message_at)), scopes.latest_message_at, VALUES(latest_message_at)), visible_message_count = GREATEST(scopes.visible_message_count, VALUES(visible_message_count)), visible_last_message_at = COALESCE(GREATEST(scopes.visible_last_message_at, VALUES(visible_last_message_at)), scopes.visible_last_message_at, VALUES(visible_last_message_at)), archived_until_seq = GREATEST(scopes.archived_until_seq, VALUES(archived_until_seq)), status = VALUES(status), created_by_member_id = VALUES(created_by_member_id), updated_at = VALUES(updated_at), archived_at = VALUES(archived_at) `, strings.TrimSpace(record.ID), strings.TrimSpace(record.WorkspaceID), strings.TrimSpace(record.Type), strings.TrimSpace(record.ParentScopeID), strings.TrimSpace(record.RootMessageID), strings.TrimSpace(record.RootMessageShortID), strings.TrimSpace(record.IdentityKey), record.LatestSeq, record.LatestGlobalSeq, record.LatestActivitySeq, record.LatestMessageAt, record.VisibleMessageCount, record.VisibleLastMessageAt, record.ArchivedUntilSeq, normalizedScopeStatus(record.Status), strings.TrimSpace(record.CreatedByMemberID), record.CreatedAt, record.UpdatedAt, record.ArchivedAt); err != nil { return fmt.Errorf("upsert scope: %w", err) } return nil } func (store ScopeStore) LoadScope(ctx context.Context, scopeID string) (ScopeRecord, error) { if err := store.requireExecer(); err != nil { return ScopeRecord{}, err } record, err := scanScopeRecord(store.execer.QueryRow(ctx, ` SELECT id, workspace_id, type, parent_scope_id, root_message_id, root_message_short_id, identity_key, latest_seq, latest_global_seq, latest_activity_seq, latest_message_at, visible_message_count, visible_last_message_at, archived_until_seq, status, created_by_member_id, created_at, updated_at, archived_at FROM scopes WHERE id = $1 `, strings.TrimSpace(scopeID))) if err != nil { return ScopeRecord{}, err } return record, nil } func (store ScopeStore) LoadActiveScopeByIdentityKey(ctx context.Context, workspaceID string, scopeType string, identityKey string) (ScopeRecord, error) { if err := store.requireExecer(); err != nil { return ScopeRecord{}, err } trimmedIdentityKey := strings.TrimSpace(identityKey) if strings.TrimSpace(workspaceID) == "" || strings.TrimSpace(scopeType) == "" || trimmedIdentityKey == "" { return ScopeRecord{}, invalidArgument("scope workspace, type, and identity key are required") } record, err := scanScopeRecord(store.execer.QueryRow(ctx, ` SELECT id, workspace_id, type, parent_scope_id, root_message_id, root_message_short_id, identity_key, latest_seq, latest_global_seq, latest_activity_seq, latest_message_at, visible_message_count, visible_last_message_at, archived_until_seq, status, created_by_member_id, created_at, updated_at, archived_at FROM scopes WHERE workspace_id = $1 AND type = $2 AND identity_key = $3 AND status = 'active' AND archived_at IS NULL LIMIT 1 `, strings.TrimSpace(workspaceID), strings.TrimSpace(scopeType), trimmedIdentityKey)) if err != nil { return ScopeRecord{}, err } return record, nil } func scanScopeRecord(scanner interface{ Scan(...any) error }) (ScopeRecord, error) { var record ScopeRecord var parentScopeID, rootMessageID, rootMessageShortID, identityKey, createdByMemberID sql.NullString var latestMessageAt, visibleLastMessageAt, archivedAt sql.NullTime err := scanner.Scan( &record.ID, &record.WorkspaceID, &record.Type, &parentScopeID, &rootMessageID, &rootMessageShortID, &identityKey, &record.LatestSeq, &record.LatestGlobalSeq, &record.LatestActivitySeq, &latestMessageAt, &record.VisibleMessageCount, &visibleLastMessageAt, &record.ArchivedUntilSeq, &record.Status, &createdByMemberID, &record.CreatedAt, &record.UpdatedAt, &archivedAt, ) if errors.Is(err, sql.ErrNoRows) { return ScopeRecord{}, notFound("scope not found") } if err != nil { return ScopeRecord{}, fmt.Errorf("load scope: %w", err) } record.ParentScopeID = parentScopeID.String record.RootMessageID = rootMessageID.String record.RootMessageShortID = rootMessageShortID.String record.IdentityKey = identityKey.String record.CreatedByMemberID = createdByMemberID.String record.LatestMessageAt = nullableTimePtr(latestMessageAt) record.VisibleLastMessageAt = nullableTimePtr(visibleLastMessageAt) record.ArchivedAt = nullableTimePtr(archivedAt) return record, nil } func (store ScopeStore) UpsertChannelProfile(ctx context.Context, record ScopeChannelProfileRecord) error { if err := store.requireExecer(); err != nil { return err } if strings.TrimSpace(record.ScopeID) == "" || strings.TrimSpace(record.WorkspaceID) == "" || strings.TrimSpace(record.Slug) == "" { return invalidArgument("channel profile scope, workspace, and slug are required") } if _, err := store.execer.Exec(ctx, ` INSERT INTO scope_channel_profiles ( scope_id, workspace_id, slug, display_name, description, visibility, subsystem, owner_member_id, is_all_channel, created_at, updated_at ) VALUES ($1, $2, $3, $4, $5, $6, $7, NULLIF($8, ''), $9, $10, $11) ON DUPLICATE KEY UPDATE slug = VALUES(slug), display_name = VALUES(display_name), description = VALUES(description), visibility = VALUES(visibility), subsystem = VALUES(subsystem), owner_member_id = VALUES(owner_member_id), is_all_channel = VALUES(is_all_channel), updated_at = VALUES(updated_at) `, strings.TrimSpace(record.ScopeID), strings.TrimSpace(record.WorkspaceID), strings.TrimSpace(record.Slug), strings.TrimSpace(record.DisplayName), record.Description, normalizedChannelVisibility(record.Visibility), NormalizeChannelSubsystem(record.Subsystem), strings.TrimSpace(record.OwnerMemberID), record.IsAllChannel, record.CreatedAt, record.UpdatedAt); err != nil { return fmt.Errorf("upsert scope channel profile: %w", err) } return nil } func (store ScopeStore) AdvanceMessageStats(ctx context.Context, scopeID string, seq int64, globalSeq int64, messageAt time.Time, humanVisible bool, recentAuthorMemberID string) error { if err := store.requireExecer(); err != nil { return err } visibleIncrement := int64(0) if humanVisible { visibleIncrement = 1 } recentAuthorMemberID = strings.TrimSpace(recentAuthorMemberID) result, err := store.execer.Exec(ctx, ` UPDATE scopes SET latest_seq = GREATEST(latest_seq, $2), latest_global_seq = GREATEST(latest_global_seq, $3), latest_message_at = CASE WHEN latest_message_at IS NULL OR latest_message_at < $4 THEN $4 ELSE latest_message_at END, visible_message_count = visible_message_count + $5, visible_last_message_at = CASE WHEN $5 > 0 AND (visible_last_message_at IS NULL OR visible_last_message_at < $4) THEN $4 ELSE visible_last_message_at END, recent_author_member_ids = CASE WHEN type <> 'thread' OR $6 = '' THEN recent_author_member_ids WHEN JSON_SEARCH(COALESCE(recent_author_member_ids, JSON_ARRAY()), 'one', $6) IS NULL THEN JSON_REMOVE( JSON_ARRAY_INSERT( COALESCE(recent_author_member_ids, JSON_ARRAY()), '$[0]', $6 ), '$[5]' ) ELSE JSON_ARRAY_INSERT( JSON_REMOVE( recent_author_member_ids, JSON_UNQUOTE(JSON_SEARCH(recent_author_member_ids, 'one', $6)) ), '$[0]', $6 ) END, updated_at = CASE WHEN updated_at < $4 THEN $4 ELSE updated_at END WHERE id = $1 `, strings.TrimSpace(scopeID), seq, globalSeq, messageAt, visibleIncrement, recentAuthorMemberID) if err != nil { return fmt.Errorf("advance scope message stats: %w", err) } if result.RowsAffected() == 0 { return fmt.Errorf("advance scope message stats: %w", sql.ErrNoRows) } return nil } func (store MembershipStore) Upsert(ctx context.Context, record ScopeMembershipRecord) error { if err := store.requireExecer(); err != nil { return err } if strings.TrimSpace(record.ScopeID) == "" || strings.TrimSpace(record.MemberID) == "" { return invalidArgument("membership scope and member are required") } if _, err := store.execer.Exec(ctx, ` INSERT INTO scope_memberships ( scope_id, member_id, visibility, owner_member_id, last_read_seq, last_read_global_seq, last_activity_global_seq, owner_context_dismissed_through_global_seq, effective_from_seq, effective_to_seq, authority_from_at, authority_to_at, created_at, updated_at, removed_at, archived_at ) VALUES ($1, $2, $3, NULLIF($4, ''), $5, $6, $7, $8, $9, $10, LEAST($11, CURRENT_TIMESTAMP(6)), $13, $11, $12, $13, $14) ON DUPLICATE KEY UPDATE unread_generation = CASE WHEN scope_memberships.visibility <> VALUES(visibility) OR COALESCE(scope_memberships.owner_member_id, '') <> COALESCE(VALUES(owner_member_id), '') OR GREATEST(scope_memberships.last_read_seq, VALUES(last_read_seq)) <> scope_memberships.last_read_seq OR GREATEST(scope_memberships.last_read_global_seq, VALUES(last_read_global_seq)) <> scope_memberships.last_read_global_seq OR GREATEST(scope_memberships.last_activity_global_seq, VALUES(last_activity_global_seq)) <> scope_memberships.last_activity_global_seq OR scope_memberships.effective_to_seq IS NOT NULL OR scope_memberships.removed_at IS NOT NULL OR scope_memberships.archived_at IS NOT NULL THEN scope_memberships.unread_generation + 1 ELSE scope_memberships.unread_generation END, visibility = VALUES(visibility), owner_member_id = VALUES(owner_member_id), last_read_seq = GREATEST(scope_memberships.last_read_seq, VALUES(last_read_seq)), last_read_global_seq = GREATEST(scope_memberships.last_read_global_seq, VALUES(last_read_global_seq)), last_activity_global_seq = GREATEST(scope_memberships.last_activity_global_seq, VALUES(last_activity_global_seq)), owner_context_dismissed_through_global_seq = GREATEST(scope_memberships.owner_context_dismissed_through_global_seq, VALUES(owner_context_dismissed_through_global_seq)), effective_to_seq = VALUES(effective_to_seq), authority_from_at = CASE WHEN scope_memberships.removed_at IS NOT NULL OR scope_memberships.archived_at IS NOT NULL OR scope_memberships.effective_to_seq IS NOT NULL THEN CASE WHEN VALUES(removed_at) IS NULL AND VALUES(archived_at) IS NULL THEN LEAST(VALUES(updated_at), CURRENT_TIMESTAMP(6)) ELSE scope_memberships.authority_from_at END ELSE COALESCE(scope_memberships.authority_from_at, scope_memberships.created_at) END, authority_to_at = CASE WHEN VALUES(removed_at) IS NULL AND VALUES(archived_at) IS NULL AND VALUES(effective_to_seq) IS NULL THEN NULL ELSE COALESCE(scope_memberships.authority_to_at, VALUES(removed_at)) END, updated_at = VALUES(updated_at), removed_at = VALUES(removed_at), archived_at = VALUES(archived_at) `, strings.TrimSpace(record.ScopeID), strings.TrimSpace(record.MemberID), normalizedMembershipVisibility(record.Visibility), strings.TrimSpace(record.OwnerMemberID), record.LastReadSeq, record.LastReadGlobalSeq, record.LastActivityGlobalSeq, record.OwnerContextDismissedThroughGlobalSeq, record.EffectiveFromSeq, record.EffectiveToSeq, record.CreatedAt, record.UpdatedAt, record.RemovedAt, record.ArchivedAt); err != nil { return fmt.Errorf("upsert scope membership: %w", err) } return nil } func (store MembershipStore) LoadActive(ctx context.Context, scopeID string, memberID string) (ScopeMembershipRecord, error) { if err := store.requireExecer(); err != nil { return ScopeMembershipRecord{}, err } scopeID = strings.TrimSpace(scopeID) memberID = strings.TrimSpace(memberID) if scopeID == "" || memberID == "" { return ScopeMembershipRecord{}, invalidArgument("active membership scope and member are required") } var record ScopeMembershipRecord var effectiveToSeq sql.NullInt64 var removedAt, archivedAt sql.NullTime err := store.execer.QueryRow(ctx, scopeActiveMembershipByIDQuery.SQL, scopeID, memberID).Scan( &record.ScopeID, &record.MemberID, &record.Visibility, &record.OwnerMemberID, &record.LastReadSeq, &record.LastReadGlobalSeq, &record.LastActivityGlobalSeq, &record.OwnerContextDismissedThroughGlobalSeq, &record.EffectiveFromSeq, &effectiveToSeq, &record.CreatedAt, &record.UpdatedAt, &removedAt, &archivedAt, ) if errors.Is(err, sql.ErrNoRows) { return ScopeMembershipRecord{}, notFound("scope membership not found") } if err != nil { return ScopeMembershipRecord{}, fmt.Errorf("load active scope membership: %w", err) } if effectiveToSeq.Valid { value := effectiveToSeq.Int64 record.EffectiveToSeq = &value } record.RemovedAt = nullableTimePtr(removedAt) record.ArchivedAt = nullableTimePtr(archivedAt) return record, nil } func (store MembershipStore) AdvanceReadState(ctx context.Context, scopeID string, memberID string, seq int64, globalSeq int64, updatedAt time.Time) error { if err := store.requireExecer(); err != nil { return err } if strings.TrimSpace(scopeID) == "" || strings.TrimSpace(memberID) == "" { return invalidArgument("read-state scope and member are required") } result, err := store.execer.Exec(ctx, scopeAdvanceMembershipReadStateQuery.SQL, strings.TrimSpace(scopeID), strings.TrimSpace(memberID), seq, globalSeq, updatedAt) if err != nil { return fmt.Errorf("advance scope membership read state: %w", err) } if result.RowsAffected() == 0 { return notFound("scope membership not found") } return nil } func (store NotificationStore) Upsert(ctx context.Context, record ScopeNotificationSettingRecord) error { if err := store.requireExecer(); err != nil { return err } if strings.TrimSpace(record.ScopeID) == "" || strings.TrimSpace(record.MemberID) == "" { return invalidArgument("notification scope and member are required") } if _, err := store.execer.Exec(ctx, ` INSERT INTO scope_notification_settings ( scope_id, member_id, state, created_at, updated_at, unread_generation ) VALUES ($1, $2, $3, $4, $5, 1) ON DUPLICATE KEY UPDATE unread_generation = CASE WHEN state <> VALUES(state) THEN unread_generation + 1 ELSE unread_generation END, updated_at = CASE WHEN state <> VALUES(state) THEN VALUES(updated_at) ELSE updated_at END, state = CASE WHEN state <> VALUES(state) THEN VALUES(state) ELSE state END `, strings.TrimSpace(record.ScopeID), strings.TrimSpace(record.MemberID), strings.TrimSpace(record.State), record.CreatedAt, record.UpdatedAt); err != nil { return fmt.Errorf("upsert scope notification setting: %w", err) } return nil } func (store PrivateVisibilityStore) InsertGrant(ctx context.Context, record ScopePrivateGrantRecord) error { if err := store.requireExecer(); err != nil { return err } if strings.TrimSpace(record.ID) == "" || strings.TrimSpace(record.ScopeID) == "" || strings.TrimSpace(record.OwnerMemberID) == "" || strings.TrimSpace(record.GranteeMemberID) == "" || strings.TrimSpace(record.GrantedByMemberID) == "" { return invalidArgument("private grant id, scope, owner, grantee, and grant actor are required") } if _, err := store.execer.Exec(ctx, ` INSERT INTO scope_private_grants ( id, workspace_id, scope_id, owner_member_id, grantee_member_id, granted_by_member_id, created_at, expires_at, revoked_at ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) `, strings.TrimSpace(record.ID), strings.TrimSpace(record.WorkspaceID), strings.TrimSpace(record.ScopeID), strings.TrimSpace(record.OwnerMemberID), strings.TrimSpace(record.GranteeMemberID), strings.TrimSpace(record.GrantedByMemberID), record.CreatedAt, record.ExpiresAt, record.RevokedAt); err != nil { return fmt.Errorf("insert scope private grant: %w", err) } return nil } func (store MessageStore) Insert(ctx context.Context, record ScopeMessageRecord) error { if err := store.requireExecer(); err != nil { return err } var err error record, err = normalizeScopeMessageCreateContract(record) if err != nil { return err } if strings.TrimSpace(record.ID) == "" || strings.TrimSpace(record.ScopeID) == "" || strings.TrimSpace(record.AuthorMemberID) == "" { return invalidArgument("message id, scope, and author are required") } if _, err := store.execer.Exec(ctx, ` INSERT INTO scope_messages ( id, workspace_id, scope_id, seq, workspace_global_seq, author_member_id, author_type, notify_principal_member_id, idempotency_subject_member_id, client_message_id, idempotency_payload_digest, message_type, content, metadata, created_at, updated_at ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NULLIF($10, ''), NULLIF($11, ''), $12, $13, $14, $15, $16) `, strings.TrimSpace(record.ID), strings.TrimSpace(record.WorkspaceID), strings.TrimSpace(record.ScopeID), record.Seq, record.WorkspaceGlobalSeq, record.AuthorMemberID, strings.TrimSpace(record.AuthorType), record.NotifyPrincipalMemberID, record.IdempotencySubjectMemberID, record.ClientMessageID, record.IdempotencyPayloadDigest, strings.TrimSpace(record.MessageType), record.Content, nullableJSON(record.Metadata), record.CreatedAt, record.UpdatedAt); err != nil { return fmt.Errorf("insert scope message: %w", err) } return nil } func (store MessageStore) InsertIdempotent(ctx context.Context, record ScopeMessageRecord) (ScopeMessageInsertResult, error) { var err error record, err = normalizeScopeMessageCreateContract(record) if err != nil { return ScopeMessageInsertResult{}, err } if record.ClientMessageID == "" { return ScopeMessageInsertResult{}, invalidArgument("client message id is required for idempotent scope message insert") } if err := store.Insert(ctx, record); err == nil { return ScopeMessageInsertResult{Inserted: true, Message: record}, nil } else if !uniqueViolationMatches(err, "scope_messages_client_message_id_key") { return ScopeMessageInsertResult{}, err } // A duplicate insert may have waited for a concurrent transaction whose // winner was not visible to this transaction's earlier snapshot. Use a // locking current read here so the duplicate-key branch always compares // against the committed winner instead of reporting a spurious not-found. existing, err := store.loadByClientMessageID(ctx, record.ScopeID, record.AuthorMemberID, record.IdempotencySubjectMemberID, record.ClientMessageID, true) if err != nil { return ScopeMessageInsertResult{}, err } if err := requireScopeMessageReplayDigest(existing, record); err != nil { return ScopeMessageInsertResult{}, err } return ScopeMessageInsertResult{Inserted: false, Message: existing}, nil } func (store MessageStore) LoadByClientMessageID(ctx context.Context, scopeID string, authorMemberID string, idempotencySubjectMemberID string, clientMessageID string) (ScopeMessageRecord, error) { return store.loadByClientMessageID(ctx, scopeID, authorMemberID, idempotencySubjectMemberID, clientMessageID, false) } // LoadByClientMessageIDForUpdate performs the idempotency lookup as a current // locking read inside the caller's transaction. func (store MessageStore) LoadByClientMessageIDForUpdate(ctx context.Context, scopeID string, authorMemberID string, idempotencySubjectMemberID string, clientMessageID string) (ScopeMessageRecord, error) { return store.loadByClientMessageID(ctx, scopeID, authorMemberID, idempotencySubjectMemberID, clientMessageID, true) } func (store MessageStore) loadByClientMessageID(ctx context.Context, scopeID string, authorMemberID string, idempotencySubjectMemberID string, clientMessageID string, forUpdate bool) (ScopeMessageRecord, error) { if err := store.requireExecer(); err != nil { return ScopeMessageRecord{}, err } idempotencySubjectMemberID = strings.TrimSpace(idempotencySubjectMemberID) if idempotencySubjectMemberID == "" { idempotencySubjectMemberID = strings.TrimSpace(authorMemberID) } var record ScopeMessageRecord var clientMessageIDValue sql.NullString var idempotencyPayloadDigest sql.NullString var editedAt sql.NullTime var metadata []byte query := ` SELECT id, workspace_id, scope_id, seq, workspace_global_seq, author_member_id, author_type, COALESCE(NULLIF(notify_principal_member_id, ''), author_member_id), COALESCE(NULLIF(idempotency_subject_member_id, ''), author_member_id), client_message_id, idempotency_payload_digest, message_type, content, metadata, created_at, updated_at, edited_at, COALESCE(edited_by_member_id, ''), edit_revision FROM scope_messages WHERE scope_id = $1 AND author_member_id = $2 AND COALESCE(NULLIF(idempotency_subject_member_id, ''), author_member_id) = $3 AND client_message_id = $4 ` if forUpdate { query += "FOR UPDATE\n" } err := store.execer.QueryRow(ctx, query, strings.TrimSpace(scopeID), strings.TrimSpace(authorMemberID), idempotencySubjectMemberID, strings.TrimSpace(clientMessageID)).Scan( &record.ID, &record.WorkspaceID, &record.ScopeID, &record.Seq, &record.WorkspaceGlobalSeq, &record.AuthorMemberID, &record.AuthorType, &record.NotifyPrincipalMemberID, &record.IdempotencySubjectMemberID, &clientMessageIDValue, &idempotencyPayloadDigest, &record.MessageType, &record.Content, &metadata, &record.CreatedAt, &record.UpdatedAt, &editedAt, &record.EditedByMemberID, &record.EditRevision, ) if errors.Is(err, sql.ErrNoRows) { return ScopeMessageRecord{}, notFound("scope message not found") } if err != nil { return ScopeMessageRecord{}, fmt.Errorf("load scope message by client id: %w", err) } record.ClientMessageID = clientMessageIDValue.String record.IdempotencyPayloadDigest = idempotencyPayloadDigest.String record.EditedAt = nullableTimePtr(editedAt) record.Metadata = append([]byte(nil), metadata...) return record, nil } func (store MessageStore) UpdateContent(ctx context.Context, request ScopeMessageContentUpdate) (ScopeMessageRecord, error) { if err := store.requireExecer(); err != nil { return ScopeMessageRecord{}, err } workspaceID := strings.TrimSpace(request.WorkspaceID) scopeID := strings.TrimSpace(request.ScopeID) messageID := strings.TrimSpace(request.MessageID) editedByMemberID := strings.TrimSpace(request.EditedByMemberID) if workspaceID == "" || scopeID == "" || messageID == "" || editedByMemberID == "" { return ScopeMessageRecord{}, invalidArgument("message edit workspace, scope, message, and editor are required") } editedAt := request.EditedAt.UTC() if editedAt.IsZero() { return ScopeMessageRecord{}, invalidArgument("message edit timestamp is required") } result, err := store.execer.Exec(ctx, ` UPDATE scope_messages SET content = $4, metadata = $5, updated_at = $6, edited_at = $6, edited_by_member_id = $7, edit_revision = edit_revision + 1 WHERE workspace_id = $1 AND scope_id = $2 AND id = $3 AND author_type = 'human' AND author_member_id = $7 AND COALESCE(message_type, '') IN ('', 'task_discussion') `, workspaceID, scopeID, messageID, request.Content, nullableJSON(request.Metadata), editedAt, editedByMemberID) if err != nil { return ScopeMessageRecord{}, fmt.Errorf("update scope message content: %w", err) } if result.RowsAffected() == 0 { return ScopeMessageRecord{}, notFound("scope message not found") } record, err := store.LoadByID(ctx, workspaceID, scopeID, messageID) if err != nil { return ScopeMessageRecord{}, err } return record, nil } func (store MessageStore) LoadByID(ctx context.Context, workspaceID string, scopeID string, messageID string) (ScopeMessageRecord, error) { if err := store.requireExecer(); err != nil { return ScopeMessageRecord{}, err } var record ScopeMessageRecord var clientMessageID sql.NullString var idempotencyPayloadDigest sql.NullString var editedAt sql.NullTime var metadata []byte err := store.execer.QueryRow(ctx, ` SELECT id, workspace_id, scope_id, seq, workspace_global_seq, author_member_id, author_type, COALESCE(NULLIF(notify_principal_member_id, ''), author_member_id), COALESCE(NULLIF(idempotency_subject_member_id, ''), author_member_id), client_message_id, idempotency_payload_digest, message_type, content, metadata, created_at, updated_at, edited_at, COALESCE(edited_by_member_id, ''), edit_revision FROM scope_messages WHERE workspace_id = $1 AND scope_id = $2 AND id = $3 `, strings.TrimSpace(workspaceID), strings.TrimSpace(scopeID), strings.TrimSpace(messageID)).Scan( &record.ID, &record.WorkspaceID, &record.ScopeID, &record.Seq, &record.WorkspaceGlobalSeq, &record.AuthorMemberID, &record.AuthorType, &record.NotifyPrincipalMemberID, &record.IdempotencySubjectMemberID, &clientMessageID, &idempotencyPayloadDigest, &record.MessageType, &record.Content, &metadata, &record.CreatedAt, &record.UpdatedAt, &editedAt, &record.EditedByMemberID, &record.EditRevision, ) if errors.Is(err, sql.ErrNoRows) { return ScopeMessageRecord{}, notFound("scope message not found") } if err != nil { return ScopeMessageRecord{}, fmt.Errorf("load scope message by id: %w", err) } record.ClientMessageID = clientMessageID.String record.IdempotencyPayloadDigest = idempotencyPayloadDigest.String record.EditedAt = nullableTimePtr(editedAt) record.Metadata = append([]byte(nil), metadata...) return record, nil } func (store MessageStore) Recall(ctx context.Context, request ScopeMessageRecallUpdate) (ScopeMessageRecord, error) { if err := store.requireExecer(); err != nil { return ScopeMessageRecord{}, err } workspaceID := strings.TrimSpace(request.WorkspaceID) scopeID := strings.TrimSpace(request.ScopeID) messageID := strings.TrimSpace(request.MessageID) recalledByMemberID := strings.TrimSpace(request.RecalledByMemberID) if workspaceID == "" || scopeID == "" || messageID == "" || recalledByMemberID == "" { return ScopeMessageRecord{}, invalidArgument("message recall workspace, scope, message, and actor are required") } recalledAt := request.RecalledAt.UTC() if recalledAt.IsZero() { return ScopeMessageRecord{}, invalidArgument("message recall timestamp is required") } expectedAuthorType := firstNonEmpty(strings.TrimSpace(request.ExpectedAuthorType), ScopeActorHuman) expectedAuthorID := firstNonEmpty(strings.TrimSpace(request.ExpectedAuthorID), recalledByMemberID) expectedMessageType := strings.TrimSpace(request.ExpectedMessageType) var result coredb.Result var err error if expectedMessageType != "" { result, err = store.execer.Exec(ctx, ` UPDATE scope_messages SET content = $4, metadata = $5, updated_at = $6 WHERE workspace_id = $1 AND scope_id = $2 AND id = $3 AND author_type = $7 AND author_member_id = $8 AND COALESCE(message_type, '') = $9 `, workspaceID, scopeID, messageID, request.Content, nullableJSON(request.Metadata), recalledAt, expectedAuthorType, expectedAuthorID, expectedMessageType) } else { result, err = store.execer.Exec(ctx, ` UPDATE scope_messages SET content = $4, metadata = $5, updated_at = $6 WHERE workspace_id = $1 AND scope_id = $2 AND id = $3 AND author_type = $7 AND author_member_id = $8 AND COALESCE(message_type, '') IN ('', 'sticker') `, workspaceID, scopeID, messageID, request.Content, nullableJSON(request.Metadata), recalledAt, expectedAuthorType, expectedAuthorID) } if err != nil { return ScopeMessageRecord{}, fmt.Errorf("recall scope message: %w", err) } if result.RowsAffected() == 0 { return ScopeMessageRecord{}, notFound("scope message not found") } record, err := store.LoadByID(ctx, workspaceID, scopeID, messageID) if err != nil { return ScopeMessageRecord{}, err } return record, nil } func UpdateScopeMessageContentAndSearchDoc(ctx context.Context, tx coredb.Tx, request ScopeMessageContentUpdate) (ScopeMessageRecord, error) { if tx == nil { return ScopeMessageRecord{}, invalidArgument("message edit transaction is required") } repos := NewScopeRepositories(tx) record, err := repos.Messages.UpdateContent(ctx, request) if err != nil { return ScopeMessageRecord{}, err } if err := repos.Search.InsertDoc(ctx, ScopeSearchDocRecord{ MessageID: record.ID, WorkspaceID: record.WorkspaceID, ScopeID: record.ScopeID, GlobalSeq: record.WorkspaceGlobalSeq, Content: record.Content, CreatedAt: record.CreatedAt, }); err != nil { return ScopeMessageRecord{}, fmt.Errorf("update scope message search doc: %w", err) } return record, nil } func (store SourceStore) Insert(ctx context.Context, record ScopeMessageSourceRecord) error { if err := store.requireExecer(); err != nil { return err } if _, err := store.execer.Exec(ctx, ` INSERT INTO scope_message_sources ( target_scope_id, message_id, ordinal, source_scope_id, source_seq, source_message_id, source_kind, source_dedupe_key, created_at ) VALUES ($1, $2, $3, $4, $5, NULLIF($6, ''), $7, NULLIF($8, ''), $9) ON DUPLICATE KEY UPDATE message_id = scope_message_sources.message_id `, strings.TrimSpace(record.TargetScopeID), strings.TrimSpace(record.MessageID), record.Ordinal, strings.TrimSpace(record.SourceScopeID), record.SourceSeq, strings.TrimSpace(record.SourceMessageID), strings.TrimSpace(record.SourceKind), strings.TrimSpace(record.SourceDedupeKey), record.CreatedAt); err != nil { return fmt.Errorf("insert scope message source: %w", err) } return nil } func (store ActivityEventStore) Insert(ctx context.Context, record ScopeActivityEventRecord) error { if err := store.requireExecer(); err != nil { return err } if record.ActivityGlobalSeq <= 0 { record.ActivityGlobalSeq = record.Seq } if err := validateScopeActivityEventRecord(record); err != nil { return err } if _, err := store.execer.Exec(ctx, ` INSERT INTO scope_activity_events ( id, workspace_id, scope_id, seq, activity_type, activity_ref_type, activity_ref_id, activity_global_seq, source_message_id, actor_member_id, reason, payload, created_at, suppressed_at ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NULLIF($9, ''), NULLIF($10, ''), $11, $12, $13, $14) ON DUPLICATE KEY UPDATE id = scope_activity_events.id `, strings.TrimSpace(record.ID), strings.TrimSpace(record.WorkspaceID), strings.TrimSpace(record.ScopeID), record.Seq, strings.TrimSpace(record.ActivityType), strings.TrimSpace(record.ActivityRefType), strings.TrimSpace(record.ActivityRefID), record.ActivityGlobalSeq, strings.TrimSpace(record.SourceMessageID), strings.TrimSpace(record.ActorMemberID), strings.TrimSpace(record.Reason), nullableJSON(record.Payload), record.CreatedAt, record.SuppressedAt); err != nil { return fmt.Errorf("insert scope activity event: %w", err) } return nil } func validateScopeActivityEventRecord(record ScopeActivityEventRecord) error { if strings.TrimSpace(record.ID) == "" || strings.TrimSpace(record.WorkspaceID) == "" || strings.TrimSpace(record.ScopeID) == "" || strings.TrimSpace(record.ActivityType) == "" || strings.TrimSpace(record.ActivityRefType) == "" || strings.TrimSpace(record.ActivityRefID) == "" { return invalidArgument("activity id, workspace, scope, type, and ref are required") } if record.Seq <= 0 { return invalidArgument("activity seq is required") } if record.CreatedAt.IsZero() { return invalidArgument("activity created timestamp is required") } if strings.EqualFold(strings.TrimSpace(record.ActivityRefType), "message") || strings.EqualFold(strings.TrimSpace(record.ActivityType), "message_created") { return invalidArgument("message-backed activity events are not allowed") } return nil } func (store OutboxStore) Insert(ctx context.Context, record ScopeFanoutOutboxRecord) error { if err := store.requireExecer(); err != nil { return err } if strings.TrimSpace(record.ID) == "" || strings.TrimSpace(record.WorkspaceID) == "" || strings.TrimSpace(record.ScopeID) == "" || strings.TrimSpace(record.EventRefType) == "" || strings.TrimSpace(record.EventRefID) == "" || strings.TrimSpace(record.FanoutType) == "" || strings.TrimSpace(record.DedupeKey) == "" { return invalidArgument("outbox id, workspace, scope, event ref, fanout type, and dedupe key are required") } if _, err := store.execer.Exec(ctx, ` INSERT INTO scope_fanout_outbox ( id, workspace_id, scope_id, message_id, event_ref_type, event_ref_id, fanout_type, dedupe_key, payload, status, attempts, max_attempts, next_attempt_at, lease_owner, lease_expires_at, created_at, delivered_at, suppressed_at, suppression_reason, last_error ) VALUES ($1, $2, $3, NULLIF($4, ''), $5, $6, $7, $8, $9, $10, $11, $12, $13, NULLIF($14, ''), $15, $16, $17, $18, $19, $20) ON DUPLICATE KEY UPDATE id = scope_fanout_outbox.id `, strings.TrimSpace(record.ID), strings.TrimSpace(record.WorkspaceID), strings.TrimSpace(record.ScopeID), strings.TrimSpace(record.MessageID), strings.TrimSpace(record.EventRefType), strings.TrimSpace(record.EventRefID), strings.TrimSpace(record.FanoutType), strings.TrimSpace(record.DedupeKey), nullableJSON(record.Payload), normalizedOutboxStatus(record.Status), record.Attempts, normalizedMaxAttempts(record.MaxAttempts), record.NextAttemptAt, strings.TrimSpace(record.LeaseOwner), record.LeaseExpiresAt, record.CreatedAt, record.DeliveredAt, record.SuppressedAt, strings.TrimSpace(record.SuppressionReason), record.LastError); err != nil { return fmt.Errorf("insert scope fanout outbox: %w", err) } return nil } // SetPayloadForMessage finalizes the authoritative message fanout payload. The // store must be bound to the transaction that created the message/outbox; a // duplicate-target conflict is an invariant failure and relies on that caller // rolling the transaction back. func (store OutboxStore) SetPayloadForMessage(ctx context.Context, messageID string, payload []byte) error { if err := store.requireExecer(); err != nil { return err } trimmedMessageID := strings.TrimSpace(messageID) if trimmedMessageID == "" { return invalidArgument("outbox message id is required") } if len(payload) == 0 { return invalidArgument("outbox payload is required") } decoded, err := UnmarshalScopeFanoutPayload(payload) if err != nil { return err } if strings.TrimSpace(decoded.AppMessage.ID) != trimmedMessageID || strings.TrimSpace(decoded.Delivery.MessageID) != trimmedMessageID { return invalidArgument("scope fanout payload message id does not match its outbox target") } matchingRows, err := store.messageFanoutPayloadTargetCount(ctx, trimmedMessageID) if err != nil { return err } if err := requireExactlyOneMessageFanoutPayloadTarget(matchingRows); err != nil { return err } result, err := store.execer.Exec(ctx, ` UPDATE scope_fanout_outbox SET payload = $2, status = CASE WHEN $3 AND status = 'pending' THEN 'pending_v2' ELSE status END WHERE message_id = $1 AND event_ref_type = 'message' AND event_ref_id = $1 AND fanout_type = 'scope_fanout' `, trimmedMessageID, payload, decoded.Delivery.HumanRecipientsTargeted) if err != nil { return fmt.Errorf("set scope fanout outbox payload: %w", err) } if result.RowsAffected() > 1 { return conflict(fmt.Sprintf("multiple scope fanout outbox payload targets found: %d", result.RowsAffected())) } if result.RowsAffected() == 0 { // PostgreSQL reports a matched row for an UPDATE that writes the same // values, while TiDB/MySQL may report zero changed rows. Recheck after the // write so an idempotent same-payload update succeeds without turning a // concurrent missing or duplicated outbox target into success. matchingRows, err = store.messageFanoutPayloadTargetCount(ctx, trimmedMessageID) if err != nil { return err } if err := requireExactlyOneMessageFanoutPayloadTarget(matchingRows); err != nil { return err } } return nil } func (store OutboxStore) messageFanoutPayloadTargetCount(ctx context.Context, messageID string) (int64, error) { var matchingRows int64 if err := store.execer.QueryRow(ctx, scopeMessageFanoutOutboxPayloadTargetCountQuery.SQL, messageID).Scan(&matchingRows); err != nil { return 0, fmt.Errorf("count scope fanout outbox payload targets: %w", err) } return matchingRows, nil } func requireExactlyOneMessageFanoutPayloadTarget(matchingRows int64) error { switch { case matchingRows == 0: return notFound("scope fanout outbox payload target not found") case matchingRows != 1: return conflict(fmt.Sprintf("multiple scope fanout outbox payload targets found: %d", matchingRows)) default: return nil } } // RefreshUndeliveredMessagePayload replaces the message.created snapshot while // it is still pending or leased. The worker reloads leased rows immediately // before fanout, so an update committed before that recheck is delivered as one // current create; an update committed after the recheck is ordered after the // already-started create by its own mutation realtime event. func (store OutboxStore) RefreshUndeliveredMessagePayload(ctx context.Context, record ScopeMessageRecord) (bool, error) { if err := store.requireExecer(); err != nil { return false, err } messageID := strings.TrimSpace(record.ID) if messageID == "" || strings.TrimSpace(record.WorkspaceID) == "" || strings.TrimSpace(record.ScopeID) == "" { return false, invalidArgument("message outbox refresh requires message, workspace, and scope") } var outboxID string var payload []byte var status string if err := store.execer.QueryRow(ctx, scopeMessageFanoutOutboxForMutationQuery.SQL, messageID).Scan(&outboxID, &payload, &status); err != nil { if coredb.IsNotFound(err) { return false, ErrNotFound } return false, fmt.Errorf("lock message fanout outbox for mutation: %w", err) } status = strings.TrimSpace(status) if status != scopeFanoutOutboxStatusPending && status != scopeFanoutOutboxStatusLeased && status != scopeFanoutOutboxStatusPendingV2 && status != scopeFanoutOutboxStatusLeasedV2 { return false, nil } refreshed, err := refreshScopeFanoutPayloadForMessage(payload, record) if err != nil { return false, err } result, err := store.execer.Exec(ctx, ` UPDATE scope_fanout_outbox SET payload = $3 WHERE id = $1 AND status = $2 `, strings.TrimSpace(outboxID), status, refreshed) if err != nil { return false, fmt.Errorf("refresh undelivered message fanout payload: %w", err) } if result.RowsAffected() != 1 { return false, conflict("message fanout outbox changed concurrently") } return true, nil } func (store OutboxStore) MarkDeliveredForSource(ctx context.Context, eventRefType string, eventRefID string, fanoutType string, deliveredAt time.Time) error { if err := store.requireExecer(); err != nil { return err } trimmedEventRefType := strings.TrimSpace(eventRefType) trimmedEventRefID := strings.TrimSpace(eventRefID) trimmedFanoutType := strings.TrimSpace(fanoutType) if trimmedFanoutType == "" { trimmedFanoutType = ScopeOutboxFanoutScope } if trimmedEventRefType == "" || trimmedEventRefID == "" || trimmedFanoutType == "" { return invalidArgument("outbox source ref and fanout type are required") } if deliveredAt.IsZero() { deliveredAt = time.Now().UTC() } result, err := store.execer.Exec(ctx, ` UPDATE scope_fanout_outbox SET status = 'delivered', delivered_at = $4, lease_owner = NULL, lease_expires_at = NULL, last_error = '' WHERE event_ref_type = $1 AND event_ref_id = $2 AND fanout_type = $3 AND status IN ('pending', 'leased', 'pending_v2', 'leased_v2') `, trimmedEventRefType, trimmedEventRefID, trimmedFanoutType, deliveredAt.UTC()) if err != nil { return fmt.Errorf("mark scope fanout outbox delivered for source: %w", err) } if result.RowsAffected() == 0 { return notFound("pending or leased scope fanout outbox not found for source") } return nil } func (store OutboxStore) MarkSuppressedForSource(ctx context.Context, eventRefType string, eventRefID string, fanoutType string, suppressedAt time.Time, reason string) error { if err := store.requireExecer(); err != nil { return err } trimmedEventRefType := strings.TrimSpace(eventRefType) trimmedEventRefID := strings.TrimSpace(eventRefID) trimmedFanoutType := strings.TrimSpace(fanoutType) if trimmedFanoutType == "" { trimmedFanoutType = ScopeOutboxFanoutScope } if trimmedEventRefType == "" || trimmedEventRefID == "" || trimmedFanoutType == "" { return invalidArgument("outbox source ref and fanout type are required") } if suppressedAt.IsZero() { suppressedAt = time.Now().UTC() } result, err := store.execer.Exec(ctx, ` UPDATE scope_fanout_outbox SET status = 'suppressed', suppressed_at = $4, suppression_reason = $5, lease_owner = NULL, lease_expires_at = NULL, last_error = '' WHERE event_ref_type = $1 AND event_ref_id = $2 AND fanout_type = $3 AND status IN ('pending', 'leased', 'pending_v2', 'leased_v2') `, trimmedEventRefType, trimmedEventRefID, trimmedFanoutType, suppressedAt.UTC(), strings.TrimSpace(reason)) if err != nil { return fmt.Errorf("mark scope fanout outbox suppressed for source: %w", err) } if result.RowsAffected() == 0 { return notFound("pending or leased scope fanout outbox not found for source") } return nil } func (store OutboxStore) MarkDeliveredForMessage(ctx context.Context, messageID string, deliveredAt time.Time) error { if err := store.requireExecer(); err != nil { return err } trimmedMessageID := strings.TrimSpace(messageID) if trimmedMessageID == "" { return invalidArgument("outbox message id is required") } if deliveredAt.IsZero() { deliveredAt = time.Now().UTC() } result, err := store.execer.Exec(ctx, ` UPDATE scope_fanout_outbox SET status = 'delivered', delivered_at = $2, lease_owner = NULL, lease_expires_at = NULL, last_error = '' WHERE message_id = $1 AND event_ref_type = 'message' AND fanout_type = 'scope_fanout' AND status = 'pending' `, trimmedMessageID, deliveredAt.UTC()) if err != nil { return fmt.Errorf("mark scope fanout outbox delivered for message: %w", err) } if result.RowsAffected() == 0 { return notFound("pending scope fanout outbox not found for message") } return nil } // ClaimForMessageInline atomically transfers one committed message fanout row // from pending (or an expired lease) to the synchronous post-commit dispatcher. // It deliberately ignores next_attempt_at: that timestamp is only the worker // grace period, while the request path may claim immediately after commit. // // A false claimed result means the row exists but another dispatcher already // owns it, or it reached a terminal state. ErrNotFound means this message has no // authoritative scope-fanout row and lets legacy callers fall back to their direct // delivery path during rolling upgrades. func (store OutboxStore) ClaimForMessageInline( ctx context.Context, messageID string, leaseOwner string, now time.Time, leaseUntil time.Time, ) (ScopeFanoutOutboxRecord, bool, error) { if err := store.requireExecer(); err != nil { return emptyScopeFanoutOutboxRecord(), false, err } messageID = strings.TrimSpace(messageID) leaseOwner = strings.TrimSpace(leaseOwner) if messageID == "" || leaseOwner == "" { return emptyScopeFanoutOutboxRecord(), false, invalidArgument("message id and inline outbox lease owner are required") } if now.IsZero() { now = time.Now().UTC() } else { now = now.UTC() } if !leaseUntil.After(now) { return emptyScopeFanoutOutboxRecord(), false, invalidArgument("inline outbox lease expiry must be after claim time") } var outboxID string if err := store.execer.QueryRow(ctx, scopeMessageFanoutOutboxIDQuery.SQL, messageID).Scan(&outboxID); err != nil { if coredb.IsNotFound(err) { return emptyScopeFanoutOutboxRecord(), false, ErrNotFound } return emptyScopeFanoutOutboxRecord(), false, fmt.Errorf("load message fanout outbox id for inline claim: %w", err) } result, err := store.execer.Exec( ctx, scopeMarkFanoutOutboxInlineLeasedQuery.SQL, strings.TrimSpace(outboxID), leaseOwner, leaseUntil.UTC(), now, ) if err != nil { return emptyScopeFanoutOutboxRecord(), false, fmt.Errorf("claim message fanout outbox inline: %w", err) } if result.RowsAffected() == 0 { return emptyScopeFanoutOutboxRecord(), false, nil } record, err := store.LoadLeasedForOwner(ctx, outboxID, leaseOwner) if err != nil { return emptyScopeFanoutOutboxRecord(), false, err } return record, true, nil } // RenewLease fences every external fanout phase and Agent handoff. A stale // dispatcher cannot extend or continue another owner's lease. func (store OutboxStore) RenewLease(ctx context.Context, outboxID string, leaseOwner string, leaseUntil time.Time) error { if err := store.requireExecer(); err != nil { return err } outboxID = strings.TrimSpace(outboxID) leaseOwner = strings.TrimSpace(leaseOwner) if outboxID == "" || leaseOwner == "" || leaseUntil.IsZero() { return invalidArgument("outbox id, lease owner, and lease expiry are required") } result, err := store.execer.Exec(ctx, scopeRenewFanoutOutboxLeaseQuery.SQL, outboxID, leaseOwner, leaseUntil.UTC()) if err != nil { return fmt.Errorf("renew scope fanout outbox lease: %w", err) } if result.RowsAffected() == 0 { // MySQL/TiDB reports changed rows by default. A renewal inside the // timestamp precision window can therefore match the owned lease while // writing the same expiry and return zero affected rows. Re-read the // fenced row before deciding ownership was lost. if _, loadErr := store.LoadLeasedForOwner(ctx, outboxID, leaseOwner); loadErr == nil { return nil } else if !errors.Is(loadErr, ErrNotFound) { return fmt.Errorf("verify unchanged scope fanout lease renewal: %w", loadErr) } return notFound("leased scope fanout outbox not found") } return nil } // HasAgentDeliveryReceipt reports whether one immutable source event was // already accepted for an Agent. The source identity is independent of the // outbox row so an embedded delivery and its own authoritative row cannot wake the // same Agent twice. func (store OutboxStore) HasAgentDeliveryReceipt( ctx context.Context, sourceEventType string, sourceEventID string, agentID string, ) (bool, error) { if err := store.requireExecer(); err != nil { return false, err } sourceEventType = strings.TrimSpace(sourceEventType) sourceEventID = strings.TrimSpace(sourceEventID) agentID = strings.TrimSpace(agentID) if sourceEventType == "" || sourceEventID == "" || agentID == "" { return false, invalidArgument("source event type, event id, and agent id are required") } var found int if err := store.execer.QueryRow(ctx, ` SELECT 1 FROM scope_fanout_agent_deliveries WHERE source_event_type = $1 AND source_event_id = $2 AND agent_id = $3 LIMIT 1 `, sourceEventType, sourceEventID, agentID).Scan(&found); err != nil { if coredb.IsNotFound(err) { return false, nil } return false, fmt.Errorf("load scope fanout Agent delivery receipt: %w", err) } return found == 1, nil } // RecordAgentDeliveryReceipt records an accepted Agent handoff only while the // caller still owns the outbox lease. The insert-after-ACK bias can duplicate // only in the unavoidable process-crash gap; AgentTeam daemon delivery ownership is // durably keyed by the same message/activity event ID and rejects that replay. func (store OutboxStore) RecordAgentDeliveryReceipt( ctx context.Context, outboxID string, leaseOwner string, sourceEventType string, sourceEventID string, agentID string, deliveredAt time.Time, ) error { if err := store.requireExecer(); err != nil { return err } outboxID = strings.TrimSpace(outboxID) leaseOwner = strings.TrimSpace(leaseOwner) sourceEventType = strings.TrimSpace(sourceEventType) sourceEventID = strings.TrimSpace(sourceEventID) agentID = strings.TrimSpace(agentID) if outboxID == "" || leaseOwner == "" || sourceEventType == "" || sourceEventID == "" || agentID == "" { return invalidArgument("outbox, lease owner, source event, and agent are required") } if deliveredAt.IsZero() { deliveredAt = time.Now().UTC() } else { deliveredAt = deliveredAt.UTC() } result, err := store.execer.Exec(ctx, ` INSERT INTO scope_fanout_agent_deliveries ( source_event_type, source_event_id, agent_id, outbox_id, delivered_at ) SELECT $3, $4, $5, outbox.id, $6 FROM scope_fanout_outbox AS outbox WHERE outbox.id = $1 AND outbox.status IN ('leased', 'leased_v2') AND outbox.lease_owner = $2 ON DUPLICATE KEY UPDATE outbox_id = VALUES(outbox_id), delivered_at = LEAST(scope_fanout_agent_deliveries.delivered_at, VALUES(delivered_at)) `, outboxID, leaseOwner, sourceEventType, sourceEventID, agentID, deliveredAt) if err != nil { return fmt.Errorf("record scope fanout Agent delivery receipt: %w", err) } if result.RowsAffected() == 0 { // MySQL/TiDB may report zero for an idempotent ON DUPLICATE KEY // no-op. Distinguish that clean replay from a stale lease. current, loadErr := store.LoadLeasedForOwner(ctx, outboxID, leaseOwner) if loadErr != nil { return loadErr } if strings.TrimSpace(current.ID) == "" { return notFound("leased scope fanout outbox not found") } recorded, loadErr := store.HasAgentDeliveryReceipt(ctx, sourceEventType, sourceEventID, agentID) if loadErr != nil { return loadErr } if !recorded { return notFound("scope fanout Agent delivery receipt was not recorded") } } return nil } func (store SyncDirtyStore) MarkMessageDirtyForScope(ctx context.Context, workspaceID string, scopeID string, latestMessageSeq int64, reason string, updatedAt time.Time) error { if err := store.requireExecer(); err != nil { return err } if strings.TrimSpace(workspaceID) == "" || strings.TrimSpace(scopeID) == "" { return invalidArgument("dirty sync workspace and scope are required") } if latestMessageSeq <= 0 { return invalidArgument("dirty sync latest message seq must be positive") } if updatedAt.IsZero() { updatedAt = time.Now().UTC() } normalizedReason := strings.TrimSpace(reason) if normalizedReason == "" { normalizedReason = "message" } if _, err := store.execer.Exec(ctx, ` INSERT INTO scope_sync_dirty ( workspace_id, scope_id, latest_message_seq, latest_activity_seq, reason, updated_at ) VALUES ($1, $2, $3, 0, $4, $5) ON DUPLICATE KEY UPDATE latest_message_seq = GREATEST(scope_sync_dirty.latest_message_seq, VALUES(latest_message_seq)), latest_activity_seq = GREATEST(scope_sync_dirty.latest_activity_seq, VALUES(latest_activity_seq)), reason = VALUES(reason), updated_at = VALUES(updated_at) `, strings.TrimSpace(workspaceID), strings.TrimSpace(scopeID), latestMessageSeq, normalizedReason, updatedAt); err != nil { return fmt.Errorf("mark message sync dirty scope: %w", err) } return nil } func (store SyncDirtyStore) MarkMessageDirtyForEligibleMembers(ctx context.Context, workspaceID string, scopeID string, authorMemberID string, latestMessageSeq int64, reason string, updatedAt time.Time) error { if err := store.requireExecer(); err != nil { return err } if strings.TrimSpace(workspaceID) == "" || strings.TrimSpace(scopeID) == "" { return invalidArgument("dirty sync workspace and scope are required") } if latestMessageSeq <= 0 { return invalidArgument("dirty sync latest message seq must be positive") } if updatedAt.IsZero() { updatedAt = time.Now().UTC() } normalizedReason := strings.TrimSpace(reason) if normalizedReason == "" { normalizedReason = "message" } if _, err := store.execer.Exec(ctx, ` INSERT INTO member_scope_sync_dirty ( workspace_id, member_id, scope_id, latest_message_seq, latest_activity_seq, reason, updated_at ) SELECT s.workspace_id, sm.member_id, s.id AS scope_id, $4 AS latest_message_seq, 0 AS latest_activity_seq, $5 AS reason, $6 AS updated_at FROM scopes s JOIN scope_memberships sm ON sm.scope_id = s.id AND sm.removed_at IS NULL AND sm.archived_at IS NULL AND sm.effective_to_seq IS NULL LEFT JOIN scope_notification_settings sns ON sns.scope_id = sm.scope_id AND sns.member_id = sm.member_id WHERE s.workspace_id = $1 AND s.id = $2 AND s.status = 'active' AND sm.member_id <> $3 AND (sns.state IS NULL OR sns.state IN ('active', 'quiet')) ON DUPLICATE KEY UPDATE latest_message_seq = GREATEST(member_scope_sync_dirty.latest_message_seq, VALUES(latest_message_seq)), latest_activity_seq = GREATEST(member_scope_sync_dirty.latest_activity_seq, VALUES(latest_activity_seq)), reason = VALUES(reason), updated_at = VALUES(updated_at) `, strings.TrimSpace(workspaceID), strings.TrimSpace(scopeID), strings.TrimSpace(authorMemberID), latestMessageSeq, normalizedReason, updatedAt); err != nil { return fmt.Errorf("mark message sync dirty scope: %w", err) } return nil } func (store ReactionStore) Insert(ctx context.Context, record ScopeMessageReactionRecord) error { if err := store.requireExecer(); err != nil { return err } if _, err := store.execer.Exec(ctx, ` INSERT INTO scope_message_reactions (id, workspace_id, scope_id, message_id, emoji, reactor_member_id, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7) ON DUPLICATE KEY UPDATE id = scope_message_reactions.id `, strings.TrimSpace(record.ID), strings.TrimSpace(record.WorkspaceID), strings.TrimSpace(record.ScopeID), strings.TrimSpace(record.MessageID), strings.TrimSpace(record.Emoji), strings.TrimSpace(record.ReactorMemberID), record.CreatedAt); err != nil { return fmt.Errorf("insert scope message reaction: %w", err) } return nil } func (store SearchStore) InsertDoc(ctx context.Context, record ScopeSearchDocRecord) error { if err := store.requireExecer(); err != nil { return err } if _, err := store.execer.Exec(ctx, ` INSERT INTO scope_message_search_docs (message_id, workspace_id, scope_id, global_seq, content, created_at) VALUES ($1, $2, $3, $4, $5, $6) ON DUPLICATE KEY UPDATE workspace_id = VALUES(workspace_id), scope_id = VALUES(scope_id), global_seq = VALUES(global_seq), content = VALUES(content) `, strings.TrimSpace(record.MessageID), strings.TrimSpace(record.WorkspaceID), strings.TrimSpace(record.ScopeID), record.GlobalSeq, record.Content, record.CreatedAt); err != nil { return fmt.Errorf("insert scope search doc: %w", err) } return nil } func (store SearchStore) UpsertState(ctx context.Context, record ScopeSearchStateRecord) error { if err := store.requireExecer(); err != nil { return err } if _, err := store.execer.Exec(ctx, ` INSERT INTO scope_message_search_state ( workspace_id, scope_id, projection_name, migration_status, backfill_status, backfill_cursor_global_seq, finalize_status, read_ready, created_at, updated_at ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) ON DUPLICATE KEY UPDATE migration_status = VALUES(migration_status), backfill_status = VALUES(backfill_status), backfill_cursor_global_seq = VALUES(backfill_cursor_global_seq), finalize_status = VALUES(finalize_status), read_ready = VALUES(read_ready), updated_at = VALUES(updated_at) `, strings.TrimSpace(record.WorkspaceID), strings.TrimSpace(record.ScopeID), strings.TrimSpace(record.ProjectionName), strings.TrimSpace(record.MigrationStatus), strings.TrimSpace(record.BackfillStatus), record.BackfillCursorGlobalSeq, strings.TrimSpace(record.FinalizeStatus), record.ReadReady, record.CreatedAt, record.UpdatedAt); err != nil { return fmt.Errorf("upsert scope search state: %w", err) } return nil } func (store ShareCardStore) Insert(ctx context.Context, record ScopeShareCardRecord) error { if err := store.requireExecer(); err != nil { return err } if _, err := store.execer.Exec(ctx, ` INSERT INTO scope_share_cards ( id, workspace_id, scope_id, message_id, source_scope_id, source_message_id, target_scope_id, created_by_member_id, revoked_at, revoked_by_member_id, payload, created_at, updated_at ) VALUES ($1, $2, $3, $4, $5, NULLIF($6, ''), NULLIF($7, ''), $8, $9, NULLIF($10, ''), $11, $12, $13) ON DUPLICATE KEY UPDATE revoked_at = VALUES(revoked_at), revoked_by_member_id = VALUES(revoked_by_member_id), payload = VALUES(payload), updated_at = VALUES(updated_at) `, strings.TrimSpace(record.ID), strings.TrimSpace(record.WorkspaceID), strings.TrimSpace(record.ScopeID), strings.TrimSpace(record.MessageID), strings.TrimSpace(record.SourceScopeID), strings.TrimSpace(record.SourceMessageID), strings.TrimSpace(record.TargetScopeID), strings.TrimSpace(record.CreatedByMemberID), record.RevokedAt, strings.TrimSpace(record.RevokedByMemberID), nullableJSON(record.Payload), record.CreatedAt, record.UpdatedAt); err != nil { return fmt.Errorf("insert scope share card: %w", err) } return nil } func (store MessageStore) ListBeforeSeq(ctx context.Context, scopeID string, beforeSeq int64, limit int) ([]ScopeMessageRecord, error) { if err := store.requireExecer(); err != nil { return nil, err } if limit <= 0 { limit = 50 } if beforeSeq <= 0 { beforeSeq = 1<<63 - 1 } rows, err := store.execer.Query(ctx, scopeMessagesBeforeSeqQuery.SQL, strings.TrimSpace(scopeID), beforeSeq, limit) if err != nil { return nil, fmt.Errorf("list scope messages: %w", err) } defer rows.Close() messages := make([]ScopeMessageRecord, 0) for rows.Next() { var record ScopeMessageRecord var metadata []byte var editedAt sql.NullTime if err := rows.Scan( &record.ID, &record.WorkspaceID, &record.ScopeID, &record.Seq, &record.WorkspaceGlobalSeq, &record.AuthorMemberID, &record.AuthorType, &record.NotifyPrincipalMemberID, &record.IdempotencySubjectMemberID, &record.ClientMessageID, &record.IdempotencyPayloadDigest, &record.MessageType, &record.Content, &metadata, &record.CreatedAt, &record.UpdatedAt, &editedAt, &record.EditedByMemberID, &record.EditRevision, ); err != nil { return nil, fmt.Errorf("scan scope message: %w", err) } record.EditedAt = nullableTimePtr(editedAt) record.Metadata = append([]byte(nil), metadata...) messages = append(messages, record) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterate scope messages: %w", err) } return messages, nil } func (store MessageStore) ListVisibilityCandidatesBeforeSeq(ctx context.Context, scopeID string, viewerMemberID string, beforeSeq int64, limit int) ([]ScopeMessageVisibilityCandidate, error) { if err := store.requireExecer(); err != nil { return nil, err } if limit <= 0 { limit = 50 } if beforeSeq <= 0 { beforeSeq = 1<<63 - 1 } rows, err := store.execer.Query(ctx, scopeMessagesVisibleToMemberQuery.SQL, strings.TrimSpace(scopeID), strings.TrimSpace(viewerMemberID), beforeSeq, limit) if err != nil { return nil, fmt.Errorf("list scope visible message candidates: %w", err) } defer rows.Close() return scanScopeMessageVisibilityCandidates(rows, false) } func scanScopeMessageVisibilityCandidates(rows coredb.Rows, withSearchRank bool) ([]ScopeMessageVisibilityCandidate, error) { candidates := make([]ScopeMessageVisibilityCandidate, 0) for rows.Next() { var candidate ScopeMessageVisibilityCandidate var metadata []byte var effectiveToSeq sql.NullInt64 var editedAt sql.NullTime var removedAt, archivedAt sql.NullTime var ignoredActivityID string dest := []any{ &candidate.Message.ID, &candidate.Message.WorkspaceID, &candidate.Message.ScopeID, &candidate.Message.Seq, &candidate.Message.WorkspaceGlobalSeq, &candidate.Message.AuthorMemberID, &candidate.Message.AuthorType, &candidate.Message.NotifyPrincipalMemberID, &candidate.Message.IdempotencySubjectMemberID, &candidate.Message.ClientMessageID, &candidate.Message.IdempotencyPayloadDigest, &candidate.Message.MessageType, &candidate.Message.Content, &metadata, &candidate.Message.CreatedAt, &candidate.Message.UpdatedAt, &editedAt, &candidate.Message.EditedByMemberID, &candidate.Message.EditRevision, &candidate.Membership.ScopeID, &candidate.Membership.MemberID, &candidate.Membership.Visibility, &candidate.Membership.OwnerMemberID, &candidate.Membership.LastReadSeq, &candidate.Membership.LastReadGlobalSeq, &candidate.Membership.LastActivityGlobalSeq, &candidate.Membership.OwnerContextDismissedThroughGlobalSeq, &candidate.Membership.EffectiveFromSeq, &effectiveToSeq, &candidate.Membership.CreatedAt, &candidate.Membership.UpdatedAt, &removedAt, &archivedAt, &candidate.GrantID, &candidate.Reason, } if withSearchRank { dest = append(dest, &candidate.SearchRank) } dest = append(dest, &ignoredActivityID, ) if err := rows.Scan(dest...); err != nil { return nil, fmt.Errorf("scan scope visible message candidate: %w", err) } candidate.Message.Metadata = append([]byte(nil), metadata...) candidate.Message.EditedAt = nullableTimePtr(editedAt) if effectiveToSeq.Valid { value := effectiveToSeq.Int64 candidate.Membership.EffectiveToSeq = &value } candidate.Membership.RemovedAt = nullableTimePtr(removedAt) candidate.Membership.ArchivedAt = nullableTimePtr(archivedAt) candidates = append(candidates, candidate) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterate scope visible message candidates: %w", err) } return candidates, nil } func (store ActivityEventStore) ListAfterCursor(ctx context.Context, workspaceID string, scopeID string, afterSeq int64, limit int) ([]ScopeActivityEventRecord, error) { if err := store.requireExecer(); err != nil { return nil, err } workspaceID = strings.TrimSpace(workspaceID) if workspaceID == "" { return nil, invalidArgument("activity workspace is required") } if limit <= 0 { limit = 50 } rows, err := store.execer.Query(ctx, scopeActivityEventsAfterCursorQuery.SQL, workspaceID, strings.TrimSpace(scopeID), afterSeq, limit) if err != nil { return nil, fmt.Errorf("list scope activity events after cursor: %w", err) } defer rows.Close() return scanScopeActivityEvents(rows) } func (store ActivityEventStore) Load(ctx context.Context, eventID string) (ScopeActivityEventRecord, error) { if err := store.requireExecer(); err != nil { return ScopeActivityEventRecord{}, err } rows, err := store.execer.Query(ctx, scopeActivityEventByIDQuery.SQL, strings.TrimSpace(eventID)) if err != nil { return ScopeActivityEventRecord{}, fmt.Errorf("load scope activity event: %w", err) } defer rows.Close() records, err := scanScopeActivityEvents(rows) if err != nil { return ScopeActivityEventRecord{}, err } if len(records) == 0 { return ScopeActivityEventRecord{}, notFound("scope activity event not found") } return records[0], nil } func (store OutboxStore) ListPending(ctx context.Context, limit int) ([]ScopeFanoutOutboxRecord, error) { if err := store.requireExecer(); err != nil { return nil, err } if limit <= 0 { limit = 50 } rows, err := store.execer.Query(ctx, scopePendingFanoutOutboxQuery.SQL, limit) if err != nil { return nil, fmt.Errorf("list pending scope fanout outbox: %w", err) } defer rows.Close() return scanScopeFanoutOutboxRows(rows) } func (store OutboxStore) LeaseDue(ctx context.Context, now time.Time, leaseOwner string, leaseUntil time.Time, limit int) ([]ScopeFanoutOutboxRecord, error) { return store.leaseDue(ctx, now, leaseOwner, leaseUntil, "", limit) } func (store OutboxStore) leaseDue(ctx context.Context, now time.Time, leaseOwner string, leaseUntil time.Time, workspaceGroupID string, limit int) ([]ScopeFanoutOutboxRecord, error) { if err := store.requireExecer(); err != nil { return nil, err } if strings.TrimSpace(leaseOwner) == "" { return nil, invalidArgument("outbox lease owner is required") } if limit <= 0 { limit = 50 } workspaceGroupID = strings.TrimSpace(workspaceGroupID) pending, err := store.listFanoutLeaseCandidates(ctx, scopeLeasePendingFanoutOutboxQuery, scopeLeasePendingFanoutOutboxForWorkspaceGroupQuery, now, workspaceGroupID, limit) if err != nil { return nil, err } expired, err := store.listFanoutLeaseCandidates(ctx, scopeLeaseExpiredFanoutOutboxQuery, scopeLeaseExpiredFanoutOutboxForWorkspaceGroupQuery, now, workspaceGroupID, limit) if err != nil { return nil, err } candidates := mergeScopeFanoutLeaseCandidates(pending, expired, limit) leased := make([]ScopeFanoutOutboxRecord, 0, len(candidates)) for _, candidate := range candidates { result, err := store.execer.Exec(ctx, scopeMarkFanoutOutboxLeasedQuery.SQL, candidate.ID, strings.TrimSpace(leaseOwner), leaseUntil, now) if err != nil { return nil, fmt.Errorf("mark scope fanout outbox leased: %w", err) } if result.RowsAffected() == 0 { continue } if candidate.Status == scopeFanoutOutboxStatusPendingV2 || candidate.Status == scopeFanoutOutboxStatusLeasedV2 { candidate.Status = scopeFanoutOutboxStatusLeasedV2 } else { candidate.Status = scopeFanoutOutboxStatusLeased } candidate.Attempts++ candidate.LeaseOwner = strings.TrimSpace(leaseOwner) candidate.LeaseExpiresAt = &leaseUntil candidate.LastError = "" leased = append(leased, candidate) } return leased, nil } // LoadLeasedForOwner reloads the authoritative payload immediately before a // worker performs fanout. Mutation transactions may refresh or suppress a // leased message outbox after LeaseDue returned its snapshot; this recheck is // the ordering fence that prevents replaying stale message.created content. func (store OutboxStore) LoadLeasedForOwner(ctx context.Context, outboxID string, leaseOwner string) (ScopeFanoutOutboxRecord, error) { if err := store.requireExecer(); err != nil { return emptyScopeFanoutOutboxRecord(), err } outboxID = strings.TrimSpace(outboxID) leaseOwner = strings.TrimSpace(leaseOwner) if outboxID == "" || leaseOwner == "" { return emptyScopeFanoutOutboxRecord(), invalidArgument("outbox id and lease owner are required") } rows, err := store.execer.Query(ctx, scopeLeasedFanoutOutboxByOwnerQuery.SQL, outboxID, leaseOwner) if err != nil { return emptyScopeFanoutOutboxRecord(), fmt.Errorf("load leased scope fanout outbox: %w", err) } defer rows.Close() records, err := scanScopeFanoutOutboxRows(rows) if err != nil { return emptyScopeFanoutOutboxRecord(), err } if len(records) == 0 { return emptyScopeFanoutOutboxRecord(), ErrNotFound } return records[0], nil } func emptyScopeFanoutOutboxRecord() ScopeFanoutOutboxRecord { var record ScopeFanoutOutboxRecord return record } func (store OutboxStore) listFanoutLeaseCandidates(ctx context.Context, descriptor scopeQueryDescriptor, groupedDescriptor scopeQueryDescriptor, now time.Time, workspaceGroupID string, limit int) ([]ScopeFanoutOutboxRecord, error) { query := descriptor.SQL args := []any{now, limit} if workspaceGroupID != "" { descriptor = groupedDescriptor query = descriptor.SQL args = []any{now, workspaceGroupID, limit} } rows, err := store.execer.Query(ctx, query, args...) if err != nil { return nil, fmt.Errorf("list %s candidates: %w", descriptor.Name, err) } defer rows.Close() candidates, err := scanScopeFanoutOutboxRows(rows) if err != nil { return nil, err } return candidates, nil } func mergeScopeFanoutLeaseCandidates(pending []ScopeFanoutOutboxRecord, expired []ScopeFanoutOutboxRecord, limit int) []ScopeFanoutOutboxRecord { if limit <= 0 || (len(pending) == 0 && len(expired) == 0) { return nil } capacity := len(pending) + len(expired) if capacity > limit { capacity = limit } merged := make([]ScopeFanoutOutboxRecord, 0, capacity) pendingIndex := 0 expiredIndex := 0 for len(merged) < limit && (pendingIndex < len(pending) || expiredIndex < len(expired)) { if expiredIndex >= len(expired) || (pendingIndex < len(pending) && scopeFanoutOutboxLeaseCandidateBefore(pending[pendingIndex], expired[expiredIndex])) { merged = append(merged, pending[pendingIndex]) pendingIndex++ continue } merged = append(merged, expired[expiredIndex]) expiredIndex++ } return merged } func scopeFanoutOutboxLeaseCandidateBefore(left ScopeFanoutOutboxRecord, right ScopeFanoutOutboxRecord) bool { if left.NextAttemptAt.Equal(right.NextAttemptAt) { return strings.TrimSpace(left.ID) < strings.TrimSpace(right.ID) } return left.NextAttemptAt.Before(right.NextAttemptAt) } func (store OutboxStore) MarkDelivered(ctx context.Context, outboxID string, leaseOwner string, deliveredAt time.Time) error { if err := store.requireExecer(); err != nil { return err } if strings.TrimSpace(outboxID) == "" || strings.TrimSpace(leaseOwner) == "" { return invalidArgument("outbox id and lease owner are required") } result, err := store.execer.Exec(ctx, scopeMarkFanoutOutboxDeliveredQuery.SQL, strings.TrimSpace(outboxID), deliveredAt, strings.TrimSpace(leaseOwner)) if err != nil { return fmt.Errorf("mark scope fanout outbox delivered: %w", err) } if result.RowsAffected() == 0 { return notFound("leased scope fanout outbox not found") } return nil } func (store OutboxStore) MarkRetryOrFailed(ctx context.Context, outboxID string, leaseOwner string, nextAttemptAt time.Time, lastError string) error { if err := store.requireExecer(); err != nil { return err } if strings.TrimSpace(outboxID) == "" || strings.TrimSpace(leaseOwner) == "" { return invalidArgument("outbox id and lease owner are required") } result, err := store.execer.Exec(ctx, scopeMarkFanoutOutboxRetryQuery.SQL, strings.TrimSpace(outboxID), nextAttemptAt, strings.TrimSpace(lastError), strings.TrimSpace(leaseOwner)) if err != nil { return fmt.Errorf("mark scope fanout outbox retry or failed: %w", err) } if result.RowsAffected() == 0 { return notFound("leased scope fanout outbox not found") } return nil } // MarkDeferred releases a targeted-consumer v2 lease without charging the // current lease attempt against the row's real delivery-failure budget. // Restricting the transition to leased_v2 preserves the rolling consumer // fence: a deferred payload can never become visible to a legacy v1 worker. func (store OutboxStore) MarkDeferred(ctx context.Context, outboxID string, leaseOwner string, nextAttemptAt time.Time, lastError string) error { if err := store.requireExecer(); err != nil { return err } if strings.TrimSpace(outboxID) == "" || strings.TrimSpace(leaseOwner) == "" { return invalidArgument("outbox id and lease owner are required") } result, err := store.execer.Exec( ctx, scopeMarkFanoutOutboxDeferredQuery.SQL, strings.TrimSpace(outboxID), nextAttemptAt, strings.TrimSpace(lastError), strings.TrimSpace(leaseOwner), ) if err != nil { return fmt.Errorf("mark scope fanout outbox deferred: %w", err) } if result.RowsAffected() == 0 { return notFound("leased v2 scope fanout outbox not found") } return nil } func (store OutboxStore) MarkSuppressed(ctx context.Context, outboxID string, leaseOwner string, suppressedAt time.Time, reason string) error { if err := store.requireExecer(); err != nil { return err } if strings.TrimSpace(outboxID) == "" || strings.TrimSpace(leaseOwner) == "" { return invalidArgument("outbox id and lease owner are required") } result, err := store.execer.Exec(ctx, scopeMarkFanoutOutboxSuppressedQuery.SQL, strings.TrimSpace(outboxID), suppressedAt, strings.TrimSpace(reason), strings.TrimSpace(leaseOwner)) if err != nil { return fmt.Errorf("mark scope fanout outbox suppressed: %w", err) } if result.RowsAffected() == 0 { return notFound("leased scope fanout outbox not found") } return nil } func (store scopeStoreBase) requireExecer() error { if store.execer == nil { return fmt.Errorf("scope store execer is nil") } return nil } func scanScopeActivityEvents(rows coredb.Rows) ([]ScopeActivityEventRecord, error) { records := make([]ScopeActivityEventRecord, 0) for rows.Next() { var record ScopeActivityEventRecord var sourceMessageID, actorMemberID sql.NullString var payload []byte var suppressedAt sql.NullTime if err := rows.Scan( &record.ID, &record.WorkspaceID, &record.ScopeID, &record.Seq, &record.ActivityType, &record.ActivityRefType, &record.ActivityRefID, &record.ActivityGlobalSeq, &sourceMessageID, &actorMemberID, &record.Reason, &payload, &record.CreatedAt, &suppressedAt, ); err != nil { return nil, fmt.Errorf("scan scope activity event: %w", err) } record.SourceMessageID = sourceMessageID.String record.ActorMemberID = actorMemberID.String record.Payload = append([]byte(nil), payload...) record.SuppressedAt = nullableTimePtr(suppressedAt) records = append(records, record) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterate scope activity events: %w", err) } return records, nil } func scanScopeFanoutOutboxRows(rows coredb.Rows) ([]ScopeFanoutOutboxRecord, error) { records := make([]ScopeFanoutOutboxRecord, 0) for rows.Next() { var record ScopeFanoutOutboxRecord var payload []byte var messageID sql.NullString var leaseExpiresAt, deliveredAt, suppressedAt sql.NullTime if err := rows.Scan( &record.ID, &record.WorkspaceID, &record.ScopeID, &messageID, &record.EventRefType, &record.EventRefID, &record.FanoutType, &record.DedupeKey, &payload, &record.Status, &record.Attempts, &record.MaxAttempts, &record.NextAttemptAt, &record.LeaseOwner, &leaseExpiresAt, &record.CreatedAt, &deliveredAt, &suppressedAt, &record.SuppressionReason, &record.LastError, ); err != nil { return nil, fmt.Errorf("scan scope fanout outbox: %w", err) } record.MessageID = messageID.String record.Payload = append([]byte(nil), payload...) record.LeaseExpiresAt = nullableTimePtr(leaseExpiresAt) record.DeliveredAt = nullableTimePtr(deliveredAt) record.SuppressedAt = nullableTimePtr(suppressedAt) records = append(records, record) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterate scope fanout outbox: %w", err) } return records, nil } func nullableTimePtr(value sql.NullTime) *time.Time { if !value.Valid { return nil } result := value.Time return &result } func nullableInt64Ptr(value sql.NullInt64) *int64 { if !value.Valid { return nil } result := value.Int64 return &result } func nullableJSON(value []byte) any { if len(value) == 0 { return nil } return value } func normalizedScopeStatus(status string) string { trimmed := strings.TrimSpace(status) if trimmed == "" { return "active" } return trimmed } func normalizedChannelVisibility(visibility string) string { trimmed := strings.TrimSpace(visibility) if trimmed == "" { return "private" } return trimmed } func normalizedMembershipVisibility(visibility string) string { trimmed := strings.TrimSpace(visibility) if trimmed == "" { return "public" } return trimmed } func normalizedOutboxStatus(status string) string { trimmed := strings.TrimSpace(status) if trimmed == "" { return "pending" } return trimmed } func normalizedMaxAttempts(maxAttempts int) int { if maxAttempts <= 0 { return 10 } return maxAttempts }