// Code generated by sqlc. DO NOT EDIT. // versions: // sqlc v1.31.1 // source: channel.sql package db import ( "context" "github.com/jackc/pgx/v5/pgtype" ) const acquireChannelReplyDelivery = `-- name: AcquireChannelReplyDelivery :one INSERT INTO channel_reply_delivery ( turn_id, task_id, attempt_depth, binding_id, installation_id, channel_type, chat_id, phase, send_state, owner_token, owner_expires_at ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, 'none', $9, now() + make_interval(secs => $10::double precision) ) ON CONFLICT (turn_id) DO UPDATE SET task_id = EXCLUDED.task_id, attempt_depth = EXCLUDED.attempt_depth, phase = CASE WHEN EXCLUDED.phase = 'terminal' THEN 'terminal' ELSE channel_reply_delivery.phase END, owner_token = EXCLUDED.owner_token, owner_expires_at = EXCLUDED.owner_expires_at, updated_at = now() WHERE channel_reply_delivery.phase <> 'settled' AND (channel_reply_delivery.owner_token IS NULL OR channel_reply_delivery.owner_expires_at <= now()) AND NOT (EXCLUDED.phase = 'streaming' AND channel_reply_delivery.phase = 'terminal') -- An attempt the retry chain has already moved past may not take the turn -- back: its late frames would rewrite what the user is reading with content -- from a run that was superseded. AND EXCLUDED.attempt_depth >= channel_reply_delivery.attempt_depth RETURNING turn_id, task_id, binding_id, installation_id, channel_type, chat_id, phase, send_state, message_id, chunks_sent, owner_token, owner_expires_at, settled_reason, created_at, updated_at, attempt_depth ` type AcquireChannelReplyDeliveryParams struct { TurnID pgtype.UUID `json:"turn_id"` TaskID pgtype.UUID `json:"task_id"` AttemptDepth int32 `json:"attempt_depth"` BindingID pgtype.UUID `json:"binding_id"` InstallationID pgtype.UUID `json:"installation_id"` ChannelType string `json:"channel_type"` ChatID string `json:"chat_id"` Phase string `json:"phase"` OwnerToken pgtype.UUID `json:"owner_token"` LeaseSeconds float64 `json:"lease_seconds"` } // Take the turn's delivery lease, creating the row on first use. Returns no // row when the turn is settled, when a live owner holds it, or when a // streaming path asks for a reply the final answer has taken over — the three // cases where this caller must not touch the provider. // // An expired lease is a process that died mid-delivery. Its successor gets the // turn, but never a clean slate: send_state survives, so an outstanding send // stays outstanding rather than being silently retried. func (q *Queries) AcquireChannelReplyDelivery(ctx context.Context, arg AcquireChannelReplyDeliveryParams) (ChannelReplyDelivery, error) { row := q.db.QueryRow(ctx, acquireChannelReplyDelivery, arg.TurnID, arg.TaskID, arg.AttemptDepth, arg.BindingID, arg.InstallationID, arg.ChannelType, arg.ChatID, arg.Phase, arg.OwnerToken, arg.LeaseSeconds, ) var i ChannelReplyDelivery err := row.Scan( &i.TurnID, &i.TaskID, &i.BindingID, &i.InstallationID, &i.ChannelType, &i.ChatID, &i.Phase, &i.SendState, &i.MessageID, &i.ChunksSent, &i.OwnerToken, &i.OwnerExpiresAt, &i.SettledReason, &i.CreatedAt, &i.UpdatedAt, &i.AttemptDepth, ) return i, err } const acquireChannelWSLease = `-- name: AcquireChannelWSLease :one UPDATE channel_installation SET ws_lease_token = $1, ws_lease_expires_at = $2, updated_at = now() WHERE id = $3 AND status = 'active' AND ( ws_lease_token IS NULL OR ws_lease_expires_at < now() OR ws_lease_token = $1 ) RETURNING id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at ` type AcquireChannelWSLeaseParams struct { NewToken pgtype.Text `json:"new_token"` NewExpiresAt pgtype.Timestamptz `json:"new_expires_at"` ID pgtype.UUID `json:"id"` } // Atomically claims the WebSocket lease. CAS predicate accepts when no // holder exists, the holder expired, or the holder is us (renewal). func (q *Queries) AcquireChannelWSLease(ctx context.Context, arg AcquireChannelWSLeaseParams) (ChannelInstallation, error) { row := q.db.QueryRow(ctx, acquireChannelWSLease, arg.NewToken, arg.NewExpiresAt, arg.ID) var i ChannelInstallation err := row.Scan( &i.ID, &i.WorkspaceID, &i.AgentID, &i.ChannelType, &i.Config, &i.Status, &i.WsLeaseToken, &i.WsLeaseExpiresAt, &i.InstallerUserID, &i.InstalledAt, &i.CreatedAt, &i.UpdatedAt, ) return i, err } const advanceChannelChatContextGeneration = `-- name: AdvanceChannelChatContextGeneration :one WITH closed AS ( UPDATE channel_chat_context_generation AS generation SET history_end_message_id = $1 WHERE generation.chat_session_id = $2 AND generation.revision = $3 ), advanced AS ( UPDATE channel_chat_session_binding AS binding SET context_revision = binding.context_revision + 1, pending_fresh = TRUE WHERE binding.chat_session_id = $2 AND binding.context_revision = $3 RETURNING binding.id, binding.chat_session_id, binding.installation_id, binding.channel_type, binding.channel_chat_id, binding.chat_type, binding.last_message_id, binding.last_thread_id, binding.config, binding.created_at, binding.pending_fresh, binding.context_revision, binding.route_revision, binding.retired_at, binding.history_start_message_id, binding.history_end_message_id, binding.history_boundary_pending ), opened AS ( INSERT INTO channel_chat_context_generation ( chat_session_id, revision, history_start_message_id, history_boundary_pending, pending_fresh ) SELECT chat_session_id, context_revision, CASE WHEN $4::boolean THEN $1 END, NOT $4::boolean, TRUE FROM advanced RETURNING chat_session_id, revision, history_start_message_id, history_end_message_id, history_boundary_pending, pending_fresh, initiator_user_id, created_at, last_message_id, last_thread_id, last_sender_id ) SELECT chat_session_id, revision, history_start_message_id, history_end_message_id, history_boundary_pending, pending_fresh, initiator_user_id, created_at, last_message_id, last_thread_id, last_sender_id FROM opened ` type AdvanceChannelChatContextGenerationParams struct { HistoryBoundaryMessageID pgtype.Text `json:"history_boundary_message_id"` ChatSessionID pgtype.UUID `json:"chat_session_id"` CurrentRevision int64 `json:"current_revision"` HasMessageBody bool `json:"has_message_body"` } type AdvanceChannelChatContextGenerationRow struct { ChatSessionID pgtype.UUID `json:"chat_session_id"` Revision int64 `json:"revision"` HistoryStartMessageID pgtype.Text `json:"history_start_message_id"` HistoryEndMessageID pgtype.Text `json:"history_end_message_id"` HistoryBoundaryPending bool `json:"history_boundary_pending"` PendingFresh bool `json:"pending_fresh"` InitiatorUserID pgtype.UUID `json:"initiator_user_id"` CreatedAt pgtype.Timestamptz `json:"created_at"` LastMessageID pgtype.Text `json:"last_message_id"` LastThreadID pgtype.Text `json:"last_thread_id"` LastSenderID pgtype.Text `json:"last_sender_id"` } // Opens a new agent-visible context while retaining the same Multica Chat. // The triggering platform message is the exclusive end of the old generation // and, when it has a body, the inclusive start of the new one. func (q *Queries) AdvanceChannelChatContextGeneration(ctx context.Context, arg AdvanceChannelChatContextGenerationParams) (AdvanceChannelChatContextGenerationRow, error) { row := q.db.QueryRow(ctx, advanceChannelChatContextGeneration, arg.HistoryBoundaryMessageID, arg.ChatSessionID, arg.CurrentRevision, arg.HasMessageBody, ) var i AdvanceChannelChatContextGenerationRow err := row.Scan( &i.ChatSessionID, &i.Revision, &i.HistoryStartMessageID, &i.HistoryEndMessageID, &i.HistoryBoundaryPending, &i.PendingFresh, &i.InitiatorUserID, &i.CreatedAt, &i.LastMessageID, &i.LastThreadID, &i.LastSenderID, ) return i, err } const backfillChannelInstallationRegionToFeishuLark = `-- name: BackfillChannelInstallationRegionToFeishuLark :execrows UPDATE channel_installation SET config = jsonb_set(config, '{region}', '"lark"'), updated_at = now() WHERE channel_type = 'feishu' AND config ->> 'region' = 'feishu' ` // Operator repair, feishu-only: flip every feishu installation still // carrying region='feishu' to 'lark'. Called only on deployments whose // legacy global base-URL override pointed at Lark international. Idempotent. func (q *Queries) BackfillChannelInstallationRegionToFeishuLark(ctx context.Context) (int64, error) { result, err := q.db.Exec(ctx, backfillChannelInstallationRegionToFeishuLark) if err != nil { return 0, err } return result.RowsAffected(), nil } const channelMediaObjectIsReferenced = `-- name: ChannelMediaObjectIsReferenced :one SELECT EXISTS ( SELECT 1 FROM attachment WHERE chat_message_id = $1 AND workspace_id = $2 AND url = $3 ) AS referenced ` type ChannelMediaObjectIsReferencedParams struct { ChatMessageID pgtype.UUID `json:"chat_message_id"` WorkspaceID pgtype.UUID `json:"workspace_id"` StorageUrl string `json:"storage_url"` } // The post-claim reference check: an attachment row carrying this object's // URL on the intended message. Only meaningful AFTER the claim flipped the // row to 'deleting' — from that point a bind can no longer succeed on the // key, so a negative answer is terminal, not a snapshot race. Re-run on every // tombstone pass as well: a positive answer there is an invariant violation, // and the object is kept and reported rather than deleted. func (q *Queries) ChannelMediaObjectIsReferenced(ctx context.Context, arg ChannelMediaObjectIsReferencedParams) (bool, error) { row := q.db.QueryRow(ctx, channelMediaObjectIsReferenced, arg.ChatMessageID, arg.WorkspaceID, arg.StorageUrl) var referenced bool err := row.Scan(&referenced) return referenced, err } const claimChannelInboundDedup = `-- name: ClaimChannelInboundDedup :one INSERT INTO channel_inbound_message_dedup (installation_id, message_id, claim_token) VALUES ($1, $2, gen_random_uuid()) ON CONFLICT (installation_id, message_id) DO UPDATE SET received_at = now(), claim_token = gen_random_uuid() WHERE channel_inbound_message_dedup.processed_at IS NULL AND channel_inbound_message_dedup.received_at < now() - INTERVAL '60 seconds' RETURNING installation_id, message_id, received_at, processed_at, claim_token ` type ClaimChannelInboundDedupParams struct { InstallationID pgtype.UUID `json:"installation_id"` MessageID string `json:"message_id"` } // ===================== // channel_inbound_message_dedup // ===================== // Two-phase idempotency gate with owner fencing. Returns the row when a // claim is acquired (fresh insert, or stale-reclaim of an in-flight claim // older than 60s); returns no rows when terminal (processed) or actively // in-flight. Every claim mints a fresh claim_token; Mark/Release are // fenced on it. See the table comment in migration 124 / the lark // predecessor for the full invariant set. func (q *Queries) ClaimChannelInboundDedup(ctx context.Context, arg ClaimChannelInboundDedupParams) (ChannelInboundMessageDedup, error) { row := q.db.QueryRow(ctx, claimChannelInboundDedup, arg.InstallationID, arg.MessageID) var i ChannelInboundMessageDedup err := row.Scan( &i.InstallationID, &i.MessageID, &i.ReceivedAt, &i.ProcessedAt, &i.ClaimToken, ) return i, err } const claimChannelMediaPendingObjectsForBind = `-- name: ClaimChannelMediaPendingObjectsForBind :many DELETE FROM channel_media_pending_object WHERE storage_key = ANY($1::text[]) AND workspace_id = $2 AND state = 'pending' RETURNING storage_key ` type ClaimChannelMediaPendingObjectsForBindParams struct { StorageKeys []string `json:"storage_keys"` WorkspaceID pgtype.UUID `json:"workspace_id"` } // Runs inside the attachment-insert transaction: commit landed ⇔ the intents // are gone, atomically, so an ambiguous COMMIT never needs adjudication. Only // 'pending' rows can be claimed — a key the reconciler moved to 'deleting' // is NOT returned, and the caller must skip attaching that object (the // placeholder stays; the reconciler will delete the object). func (q *Queries) ClaimChannelMediaPendingObjectsForBind(ctx context.Context, arg ClaimChannelMediaPendingObjectsForBindParams) ([]string, error) { rows, err := q.db.Query(ctx, claimChannelMediaPendingObjectsForBind, arg.StorageKeys, arg.WorkspaceID) if err != nil { return nil, err } defer rows.Close() items := []string{} for rows.Next() { var storage_key string if err := rows.Scan(&storage_key); err != nil { return nil, err } items = append(items, storage_key) } if err := rows.Err(); err != nil { return nil, err } return items, nil } const claimNextChannelMediaPendingObjectForReconcile = `-- name: ClaimNextChannelMediaPendingObjectForReconcile :one UPDATE channel_media_pending_object AS obj SET state = CASE WHEN obj.state = 'tombstoned' THEN 'tombstoned' ELSE 'deleting' END, lease_token = $1, lease_expires_at = now() + $2::interval, attempt = obj.attempt + 1 FROM ( SELECT cand.storage_key FROM channel_media_pending_object AS cand WHERE cand.next_attempt_at <= now() AND ( (cand.state = 'pending' AND cand.created_at <= now() - $3::interval) OR (cand.state = 'deleting' AND (cand.lease_expires_at IS NULL OR cand.lease_expires_at <= now())) OR (cand.state = 'tombstoned' AND (cand.lease_expires_at IS NULL OR cand.lease_expires_at <= now())) ) ORDER BY cand.next_attempt_at LIMIT 1 FOR UPDATE SKIP LOCKED ) AS due WHERE obj.storage_key = due.storage_key RETURNING obj.storage_key, obj.workspace_id, obj.chat_message_id, obj.storage_url, obj.installation_id, obj.state, obj.lease_token, obj.lease_expires_at, obj.attempt, obj.next_attempt_at, obj.last_error, obj.tombstone_pass, obj.created_at ` type ClaimNextChannelMediaPendingObjectForReconcileParams struct { LeaseToken pgtype.UUID `json:"lease_token"` Lease pgtype.Interval `json:"lease"` SettleDelay pgtype.Interval `json:"settle_delay"` } // Short-transaction claim of ONE due row, taken immediately before that row is // settled. Claiming a whole batch up front made the claim a promise the sweep // might not keep: a tail row sat in 'deleting' with attempt already bumped // while earlier rows ran, and could expire and be reclaimed before its own // DELETE was ever tried, inflating attempt/backoff for work that never // happened. One row per claim means attempt counts attempts. // // Due means (a) 'pending' rows older than the settle delay — an operational // buffer only; correctness comes from the state flip, after which a bind can // never succeed on the key — or (b) 'deleting' rows whose lease expired (a // crashed or failed worker) — or (c) tombstones: the object was deleted, but a // PUT the client abandoned may still materialize it afterwards, so each due // tombstone gets another idempotent delete before the row is finally dropped. // FOR UPDATE SKIP LOCKED keeps replicas off each other's row; the // object-storage DELETE happens outside any transaction, gated by the lease // token. No row (ErrNoRows) means nothing is due — the sweep is done. func (q *Queries) ClaimNextChannelMediaPendingObjectForReconcile(ctx context.Context, arg ClaimNextChannelMediaPendingObjectForReconcileParams) (ChannelMediaPendingObject, error) { row := q.db.QueryRow(ctx, claimNextChannelMediaPendingObjectForReconcile, arg.LeaseToken, arg.Lease, arg.SettleDelay) var i ChannelMediaPendingObject err := row.Scan( &i.StorageKey, &i.WorkspaceID, &i.ChatMessageID, &i.StorageUrl, &i.InstallationID, &i.State, &i.LeaseToken, &i.LeaseExpiresAt, &i.Attempt, &i.NextAttemptAt, &i.LastError, &i.TombstonePass, &i.CreatedAt, ) return i, err } const clearChannelChatContextPendingFresh = `-- name: ClearChannelChatContextPendingFresh :exec UPDATE channel_chat_context_generation SET pending_fresh = FALSE WHERE chat_session_id = $1 AND revision = $2 ` type ClearChannelChatContextPendingFreshParams struct { ChatSessionID pgtype.UUID `json:"chat_session_id"` Revision int64 `json:"revision"` } func (q *Queries) ClearChannelChatContextPendingFresh(ctx context.Context, arg ClearChannelChatContextPendingFreshParams) error { _, err := q.db.Exec(ctx, clearChannelChatContextPendingFresh, arg.ChatSessionID, arg.Revision) return err } const clearChannelChatSessionPendingFresh = `-- name: ClearChannelChatSessionPendingFresh :exec UPDATE channel_chat_session_binding SET pending_fresh = FALSE WHERE chat_session_id = $1 ` func (q *Queries) ClearChannelChatSessionPendingFresh(ctx context.Context, chatSessionID pgtype.UUID) error { _, err := q.db.Exec(ctx, clearChannelChatSessionPendingFresh, chatSessionID) return err } const clearChannelChatSessionPendingFreshForRevision = `-- name: ClearChannelChatSessionPendingFreshForRevision :exec UPDATE channel_chat_session_binding SET pending_fresh = FALSE WHERE chat_session_id = $1 AND context_revision = $2 ` type ClearChannelChatSessionPendingFreshForRevisionParams struct { ChatSessionID pgtype.UUID `json:"chat_session_id"` Revision int64 `json:"revision"` } func (q *Queries) ClearChannelChatSessionPendingFreshForRevision(ctx context.Context, arg ClearChannelChatSessionPendingFreshForRevisionParams) error { _, err := q.db.Exec(ctx, clearChannelChatSessionPendingFreshForRevision, arg.ChatSessionID, arg.Revision) return err } const clearChannelInstallationBotScopedRows = `-- name: ClearChannelInstallationBotScopedRows :one WITH cleared_chat_sessions AS ( DELETE FROM channel_chat_session_binding AS binding WHERE binding.installation_id = $1 RETURNING binding.chat_session_id ), cleared_outbound_cards AS ( DELETE FROM channel_outbound_card_message AS card WHERE card.chat_session_id IN (SELECT chat_session_id FROM cleared_chat_sessions) RETURNING card.id ), cleared_task_deliveries AS ( DELETE FROM channel_task_delivery AS delivery WHERE delivery.installation_id = $1 RETURNING delivery.task_id ), cleared_outbound_messages AS ( DELETE FROM channel_outbound_message AS outbound WHERE outbound.installation_id = $1 RETURNING outbound.channel_message_id ), cleared_binding_tokens AS ( DELETE FROM channel_binding_token AS token WHERE token.installation_id = $1 RETURNING token.token_hash ), cleared_inbound_dedup AS ( DELETE FROM channel_inbound_message_dedup AS dedup WHERE dedup.installation_id = $1 RETURNING dedup.message_id ), cleared_user_bindings AS ( DELETE FROM channel_user_binding AS user_binding WHERE user_binding.installation_id = $1 RETURNING user_binding.id ) SELECT (SELECT count(*) FROM cleared_user_bindings)::bigint AS user_bindings, (SELECT count(*) FROM cleared_chat_sessions)::bigint AS chat_session_bindings, (SELECT count(*) FROM cleared_task_deliveries)::bigint AS task_deliveries, (SELECT count(*) FROM cleared_outbound_messages)::bigint AS outbound_messages ` type ClearChannelInstallationBotScopedRowsRow struct { UserBindings int64 `json:"user_bindings"` ChatSessionBindings int64 `json:"chat_session_bindings"` TaskDeliveries int64 `json:"task_deliveries"` OutboundMessages int64 `json:"outbound_messages"` } // Bot-swap cleanup. Pointing an existing installation at a DIFFERENT bot keeps // the installation row and its id — UpsertChannelInstallation conflicts on // (workspace_id, agent_id, channel_type) and only rewrites config — so every // dependent row keyed on installation_id survives into a bot it does not belong // to. This clears the ones that are scoped to the OLD bot (#6547). // // Why they are invalid rather than merely stale: a WeCom aibot userid is // anonymized per (bot, user), so a carried-over channel_user_binding addresses // the new bot with an id from a namespace it does not share — Outbound's inbox // push resolves the sender by installation_id (reused, hence the LIVE new bot) // and then sends to the OLD bot's userid. The same reasoning covers a p2p chat // binding, whose channel_chat_id IS that userid, and every queued outbound row // carrying one. // // The installation itself, and the audit trail beneath it, deliberately stay: // the row is the same connection under new credentials, and a channel_inbound_ // audit row still records something an operator can act on. Reclaim DETACHes // audit only because the installation is going away, which here it is not. // // chat_session is likewise NOT touched. Deleting the binding detaches the WeCom // route from the Chat, the same shape a disconnect leaves behind; the Chat and // its history are the user's work product and outlive whichever bot carried // them. channel_outbound_card_message has no installation_id and no FK, so it // is reached through the just-removed bindings — the only link back — and goes // because the platform message ids on it belong to the old bot. // // Returns what it removed. A queued channel_task_delivery is a RUNNING task's // answer: processEvent finds no row and returns nil, so the answer is dropped // with no counter and no log line of its own. Deleting it is still right — the // address on it is the old bot's userid and unreachable either way — but // whoever is waiting for that answer deserves one line saying where it went. // channel_outbound_message is counted for the same reason and not because it is // expected to be non-zero: a dropped queued send is a reply that never arrives, // and a count nobody has to read costs nothing next to guessing later. func (q *Queries) ClearChannelInstallationBotScopedRows(ctx context.Context, installationID pgtype.UUID) (ClearChannelInstallationBotScopedRowsRow, error) { row := q.db.QueryRow(ctx, clearChannelInstallationBotScopedRows, installationID) var i ClearChannelInstallationBotScopedRowsRow err := row.Scan( &i.UserBindings, &i.ChatSessionBindings, &i.TaskDeliveries, &i.OutboundMessages, ) return i, err } const closeChannelReplyDeliveryTurn = `-- name: CloseChannelReplyDeliveryTurn :one INSERT INTO channel_reply_delivery ( turn_id, task_id, attempt_depth, binding_id, installation_id, channel_type, chat_id, phase, send_state, settled_reason ) VALUES ( $1, $2, $3, $4, $5, $6, $7, 'settled', 'none', $8 ) ON CONFLICT (turn_id) DO UPDATE SET phase = 'settled', settled_reason = $8, owner_token = NULL, owner_expires_at = NULL, updated_at = now() WHERE channel_reply_delivery.phase <> 'settled' AND (channel_reply_delivery.owner_token IS NULL OR channel_reply_delivery.owner_expires_at <= now()) -- Same one-way rule the claim follows. A cancellation or empty completion -- belonging to an attempt the retry chain has moved past must not end the -- turn: the attempt that superseded it is still delivering, and its answer -- would be dropped as "already settled". AND EXCLUDED.attempt_depth >= channel_reply_delivery.attempt_depth RETURNING turn_id, task_id, binding_id, installation_id, channel_type, chat_id, phase, send_state, message_id, chunks_sent, owner_token, owner_expires_at, settled_reason, created_at, updated_at, attempt_depth ` type CloseChannelReplyDeliveryTurnParams struct { TurnID pgtype.UUID `json:"turn_id"` TaskID pgtype.UUID `json:"task_id"` AttemptDepth int32 `json:"attempt_depth"` BindingID pgtype.UUID `json:"binding_id"` InstallationID pgtype.UUID `json:"installation_id"` ChannelType string `json:"channel_type"` ChatID string `json:"chat_id"` SettledReason string `json:"settled_reason"` } // End a turn that has no answer to deliver — cancelled, or completed empty — // creating the row when the turn never reached the provider at all. Without // the insert, a first text frame arriving after the cancellation would find // nothing, open a placeholder, and leave it there forever. func (q *Queries) CloseChannelReplyDeliveryTurn(ctx context.Context, arg CloseChannelReplyDeliveryTurnParams) (ChannelReplyDelivery, error) { row := q.db.QueryRow(ctx, closeChannelReplyDeliveryTurn, arg.TurnID, arg.TaskID, arg.AttemptDepth, arg.BindingID, arg.InstallationID, arg.ChannelType, arg.ChatID, arg.SettledReason, ) var i ChannelReplyDelivery err := row.Scan( &i.TurnID, &i.TaskID, &i.BindingID, &i.InstallationID, &i.ChannelType, &i.ChatID, &i.Phase, &i.SendState, &i.MessageID, &i.ChunksSent, &i.OwnerToken, &i.OwnerExpiresAt, &i.SettledReason, &i.CreatedAt, &i.UpdatedAt, &i.AttemptDepth, ) return i, err } const consumeChannelBindingToken = `-- name: ConsumeChannelBindingToken :one UPDATE channel_binding_token SET consumed_at = now() WHERE token_hash = $1 AND consumed_at IS NULL AND expires_at > now() RETURNING token_hash, workspace_id, installation_id, channel_type, channel_user_id, expires_at, consumed_at, created_at ` // Atomic redemption: returns the row only if the hash exists, is // unconsumed, and unexpired. Two simultaneous redemptions cannot both win. func (q *Queries) ConsumeChannelBindingToken(ctx context.Context, tokenHash string) (ChannelBindingToken, error) { row := q.db.QueryRow(ctx, consumeChannelBindingToken, tokenHash) var i ChannelBindingToken err := row.Scan( &i.TokenHash, &i.WorkspaceID, &i.InstallationID, &i.ChannelType, &i.ChannelUserID, &i.ExpiresAt, &i.ConsumedAt, &i.CreatedAt, ) return i, err } const copyChannelTaskDelivery = `-- name: CopyChannelTaskDelivery :exec INSERT INTO channel_task_delivery ( task_id, binding_id, installation_id, channel_type, channel_chat_id, chat_type, channel_message_id, channel_thread_id, channel_sender_id, route_revision, config ) SELECT $1, delivery.binding_id, delivery.installation_id, delivery.channel_type, delivery.channel_chat_id, delivery.chat_type, delivery.channel_message_id, delivery.channel_thread_id, delivery.channel_sender_id, delivery.route_revision, delivery.config FROM channel_task_delivery AS delivery WHERE delivery.task_id = $2 ` type CopyChannelTaskDeliveryParams struct { ChildTaskID pgtype.UUID `json:"child_task_id"` ParentTaskID pgtype.UUID `json:"parent_task_id"` } func (q *Queries) CopyChannelTaskDelivery(ctx context.Context, arg CopyChannelTaskDeliveryParams) error { _, err := q.db.Exec(ctx, copyChannelTaskDelivery, arg.ChildTaskID, arg.ParentTaskID) return err } const createChannelBindingToken = `-- name: CreateChannelBindingToken :one INSERT INTO channel_binding_token ( token_hash, workspace_id, installation_id, channel_type, channel_user_id, expires_at ) VALUES ( $1, $2, $3, $4, $5, LEAST($6::timestamptz, now() + INTERVAL '15 minutes') ) RETURNING token_hash, workspace_id, installation_id, channel_type, channel_user_id, expires_at, consumed_at, created_at ` type CreateChannelBindingTokenParams struct { TokenHash string `json:"token_hash"` WorkspaceID pgtype.UUID `json:"workspace_id"` InstallationID pgtype.UUID `json:"installation_id"` ChannelType string `json:"channel_type"` ChannelUserID string `json:"channel_user_id"` ExpiresAt pgtype.Timestamptz `json:"expires_at"` } // ===================== // channel_binding_token // ===================== // Mints a single-use binding token for an unbound platform user. TTL cap // (15 min) enforced by the table CHECK in lockstep with // channel.BindingTokenTTL. Clamp against the database clock so small clock // skew between an app node and Postgres cannot reject an otherwise valid // 15-minute token. The HASH is stored, never the raw token. func (q *Queries) CreateChannelBindingToken(ctx context.Context, arg CreateChannelBindingTokenParams) (ChannelBindingToken, error) { row := q.db.QueryRow(ctx, createChannelBindingToken, arg.TokenHash, arg.WorkspaceID, arg.InstallationID, arg.ChannelType, arg.ChannelUserID, arg.ExpiresAt, ) var i ChannelBindingToken err := row.Scan( &i.TokenHash, &i.WorkspaceID, &i.InstallationID, &i.ChannelType, &i.ChannelUserID, &i.ExpiresAt, &i.ConsumedAt, &i.CreatedAt, ) return i, err } const createChannelChatSessionBinding = `-- name: CreateChannelChatSessionBinding :one WITH next_route AS ( SELECT COALESCE(MAX(route_revision) + 1, 1)::bigint AS route_revision FROM channel_chat_session_binding AS existing WHERE existing.installation_id = $2 AND existing.channel_chat_id = $4 ), binding AS ( INSERT INTO channel_chat_session_binding ( chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, config, route_revision ) SELECT $1, $2, $3, $4, $5, $6, next_route.route_revision FROM next_route RETURNING id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at, pending_fresh, context_revision, route_revision, retired_at, history_start_message_id, history_end_message_id, history_boundary_pending ), generation AS ( INSERT INTO channel_chat_context_generation (chat_session_id, revision) SELECT chat_session_id, context_revision FROM binding ) SELECT id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at, pending_fresh, context_revision, route_revision, retired_at, history_start_message_id, history_end_message_id, history_boundary_pending FROM binding ` type CreateChannelChatSessionBindingParams struct { ChatSessionID pgtype.UUID `json:"chat_session_id"` InstallationID pgtype.UUID `json:"installation_id"` ChannelType string `json:"channel_type"` ChannelChatID string `json:"channel_chat_id"` ChatType string `json:"chat_type"` Config []byte `json:"config"` } type CreateChannelChatSessionBindingRow struct { ID pgtype.UUID `json:"id"` ChatSessionID pgtype.UUID `json:"chat_session_id"` InstallationID pgtype.UUID `json:"installation_id"` ChannelType string `json:"channel_type"` ChannelChatID string `json:"channel_chat_id"` ChatType string `json:"chat_type"` LastMessageID pgtype.Text `json:"last_message_id"` LastThreadID pgtype.Text `json:"last_thread_id"` Config []byte `json:"config"` CreatedAt pgtype.Timestamptz `json:"created_at"` PendingFresh bool `json:"pending_fresh"` ContextRevision int64 `json:"context_revision"` RouteRevision int64 `json:"route_revision"` RetiredAt pgtype.Timestamptz `json:"retired_at"` HistoryStartMessageID pgtype.Text `json:"history_start_message_id"` HistoryEndMessageID pgtype.Text `json:"history_end_message_id"` HistoryBoundaryPending bool `json:"history_boundary_pending"` } // ===================== // channel_chat_session_binding // ===================== // channel_chat_id is the session-isolation key (one chat_session per // (installation_id, channel_chat_id)): Feishu passes the chat id; Slack passes // a stable key that, for channels, includes the thread root so each @bot thread // is its own session. config carries any platform-specific outbound routing the // key alone does not (e.g. Slack's real channel_id when the key is composite); // it is opaque to the shared session service. func (q *Queries) CreateChannelChatSessionBinding(ctx context.Context, arg CreateChannelChatSessionBindingParams) (CreateChannelChatSessionBindingRow, error) { row := q.db.QueryRow(ctx, createChannelChatSessionBinding, arg.ChatSessionID, arg.InstallationID, arg.ChannelType, arg.ChannelChatID, arg.ChatType, arg.Config, ) var i CreateChannelChatSessionBindingRow err := row.Scan( &i.ID, &i.ChatSessionID, &i.InstallationID, &i.ChannelType, &i.ChannelChatID, &i.ChatType, &i.LastMessageID, &i.LastThreadID, &i.Config, &i.CreatedAt, &i.PendingFresh, &i.ContextRevision, &i.RouteRevision, &i.RetiredAt, &i.HistoryStartMessageID, &i.HistoryEndMessageID, &i.HistoryBoundaryPending, ) return i, err } const createChannelChatSessionBindingGeneration = `-- name: CreateChannelChatSessionBindingGeneration :one WITH next_route AS ( SELECT GREATEST( $1::bigint, COALESCE(MAX(route_revision) + 1, 1)::bigint ) AS route_revision FROM channel_chat_session_binding AS existing WHERE existing.installation_id = $2 AND existing.channel_chat_id = $3 ), binding AS ( INSERT INTO channel_chat_session_binding ( chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, config, route_revision, history_start_message_id, history_boundary_pending ) SELECT $4, $2, $5, $3, $6, $7, next_route.route_revision, $8, $9 FROM next_route RETURNING id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at, pending_fresh, context_revision, route_revision, retired_at, history_start_message_id, history_end_message_id, history_boundary_pending ), generation AS ( INSERT INTO channel_chat_context_generation ( chat_session_id, revision, history_start_message_id, history_boundary_pending ) SELECT chat_session_id, context_revision, history_start_message_id, history_boundary_pending FROM binding ) SELECT id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at, pending_fresh, context_revision, route_revision, retired_at, history_start_message_id, history_end_message_id, history_boundary_pending FROM binding ` type CreateChannelChatSessionBindingGenerationParams struct { RouteRevision int64 `json:"route_revision"` InstallationID pgtype.UUID `json:"installation_id"` ChannelChatID string `json:"channel_chat_id"` ChatSessionID pgtype.UUID `json:"chat_session_id"` ChannelType string `json:"channel_type"` ChatType string `json:"chat_type"` Config []byte `json:"config"` HistoryStartMessageID pgtype.Text `json:"history_start_message_id"` HistoryBoundaryPending bool `json:"history_boundary_pending"` } type CreateChannelChatSessionBindingGenerationRow struct { ID pgtype.UUID `json:"id"` ChatSessionID pgtype.UUID `json:"chat_session_id"` InstallationID pgtype.UUID `json:"installation_id"` ChannelType string `json:"channel_type"` ChannelChatID string `json:"channel_chat_id"` ChatType string `json:"chat_type"` LastMessageID pgtype.Text `json:"last_message_id"` LastThreadID pgtype.Text `json:"last_thread_id"` Config []byte `json:"config"` CreatedAt pgtype.Timestamptz `json:"created_at"` PendingFresh bool `json:"pending_fresh"` ContextRevision int64 `json:"context_revision"` RouteRevision int64 `json:"route_revision"` RetiredAt pgtype.Timestamptz `json:"retired_at"` HistoryStartMessageID pgtype.Text `json:"history_start_message_id"` HistoryEndMessageID pgtype.Text `json:"history_end_message_id"` HistoryBoundaryPending bool `json:"history_boundary_pending"` } func (q *Queries) CreateChannelChatSessionBindingGeneration(ctx context.Context, arg CreateChannelChatSessionBindingGenerationParams) (CreateChannelChatSessionBindingGenerationRow, error) { row := q.db.QueryRow(ctx, createChannelChatSessionBindingGeneration, arg.RouteRevision, arg.InstallationID, arg.ChannelChatID, arg.ChatSessionID, arg.ChannelType, arg.ChatType, arg.Config, arg.HistoryStartMessageID, arg.HistoryBoundaryPending, ) var i CreateChannelChatSessionBindingGenerationRow err := row.Scan( &i.ID, &i.ChatSessionID, &i.InstallationID, &i.ChannelType, &i.ChannelChatID, &i.ChatType, &i.LastMessageID, &i.LastThreadID, &i.Config, &i.CreatedAt, &i.PendingFresh, &i.ContextRevision, &i.RouteRevision, &i.RetiredAt, &i.HistoryStartMessageID, &i.HistoryEndMessageID, &i.HistoryBoundaryPending, ) return i, err } const createChannelOutboundCardMessage = `-- name: CreateChannelOutboundCardMessage :one INSERT INTO channel_outbound_card_message ( chat_session_id, task_id, channel_type, channel_chat_id, channel_card_message_id, status ) VALUES ( $1, $6, $2, $3, $4, $5 ) RETURNING id, chat_session_id, task_id, channel_type, channel_chat_id, channel_card_message_id, status, last_patched_at, created_at ` type CreateChannelOutboundCardMessageParams struct { ChatSessionID pgtype.UUID `json:"chat_session_id"` ChannelType string `json:"channel_type"` ChannelChatID string `json:"channel_chat_id"` ChannelCardMessageID string `json:"channel_card_message_id"` Status string `json:"status"` TaskID pgtype.UUID `json:"task_id"` } // ===================== // channel_outbound_card_message // ===================== func (q *Queries) CreateChannelOutboundCardMessage(ctx context.Context, arg CreateChannelOutboundCardMessageParams) (ChannelOutboundCardMessage, error) { row := q.db.QueryRow(ctx, createChannelOutboundCardMessage, arg.ChatSessionID, arg.ChannelType, arg.ChannelChatID, arg.ChannelCardMessageID, arg.Status, arg.TaskID, ) var i ChannelOutboundCardMessage err := row.Scan( &i.ID, &i.ChatSessionID, &i.TaskID, &i.ChannelType, &i.ChannelChatID, &i.ChannelCardMessageID, &i.Status, &i.LastPatchedAt, &i.CreatedAt, ) return i, err } const createChannelTaskDeliveryFromSession = `-- name: CreateChannelTaskDeliveryFromSession :one INSERT INTO channel_task_delivery ( task_id, binding_id, installation_id, channel_type, channel_chat_id, chat_type, channel_message_id, channel_thread_id, channel_sender_id, route_revision, config ) SELECT $1, binding.id, binding.installation_id, binding.channel_type, binding.channel_chat_id, binding.chat_type, generation.last_message_id, CASE -- This generation recorded a trigger, so its thread is authoritative — -- including when it is legitimately empty, which is how a Slack DM -- answered at top level is distinguished from one answered inside a -- thread. WHEN generation.last_message_id IS NOT NULL THEN generation.last_thread_id -- No trigger: pre-migration, or a generation whose only messages were -- channel commands. Recover the thread ONLY for a thread-isolated -- binding, where it is a stable property of the session rather than a -- moving cursor. Such a binding is exactly the one whose key is the -- composite "chat:thread", which is why it carries the real chat id in -- its config — the marker every adapter writes (larkSessionRouting, -- slackSessionRouting, telegramSessionRouting). Without this, a -- recovered pre-migration run in a Slack channel thread or a Telegram -- forum topic would answer in the parent conversation. WHEN COALESCE(binding.config ->> 'chat_id', binding.config ->> 'channel_id', '') NOT IN ('', binding.channel_chat_id) THEN binding.last_thread_id -- A non-isolated binding's cursor names whichever thread spoke last, -- which is not this run's. Nothing we can justify. ELSE NULL END, generation.last_sender_id, binding.route_revision, binding.config FROM channel_chat_session_binding AS binding JOIN channel_chat_context_generation AS generation ON generation.chat_session_id = binding.chat_session_id AND generation.revision = $2 WHERE binding.chat_session_id = $3 RETURNING task_id, binding_id, installation_id, channel_type, channel_chat_id, chat_type, channel_message_id, channel_thread_id, route_revision, config, created_at, channel_sender_id ` type CreateChannelTaskDeliveryFromSessionParams struct { TaskID pgtype.UUID `json:"task_id"` ContextRevision int64 `json:"context_revision"` ChatSessionID pgtype.UUID `json:"chat_session_id"` } // ===================== // channel_task_delivery // ===================== // Freezes one task's outbound delivery, from two different sources on purpose. // // ROUTE — chat, type, config, revision — comes from the session binding. It is // a property of the session and does not vary by generation. // // TRIGGER — the message an answer quotes, the thread it replies into, and the // account it @-mentions — comes from the generation this task answers, NOT // from the binding's latest-trigger cursor, which a newer generation may // already have advanced past. // // The thread sits with the trigger rather than the route deliberately. For a // thread-isolated session (Lark topic, Slack channel thread) the two coincide, // so it is easy to mistake the thread for route data — but Slack DMs keep ONE // binding per channel while replying into whichever thread the member used // (slackSessionRouting), so the binding cursor there names the latest thread, // not this run's. Taking it from the generation is correct for both shapes. // // The one exception is the CASE below: when a generation recorded no trigger // at all, a thread-isolated binding can still say which thread the session // lives in, and must, or the answer surfaces in the parent channel. // // A NULL trigger means "we cannot attribute this run": callers reply without a // quote or mention rather than inventing one. INNER JOIN on the generation: a // task whose generation row is missing entirely has no context to deliver // against at all. func (q *Queries) CreateChannelTaskDeliveryFromSession(ctx context.Context, arg CreateChannelTaskDeliveryFromSessionParams) (ChannelTaskDelivery, error) { row := q.db.QueryRow(ctx, createChannelTaskDeliveryFromSession, arg.TaskID, arg.ContextRevision, arg.ChatSessionID) var i ChannelTaskDelivery err := row.Scan( &i.TaskID, &i.BindingID, &i.InstallationID, &i.ChannelType, &i.ChannelChatID, &i.ChatType, &i.ChannelMessageID, &i.ChannelThreadID, &i.RouteRevision, &i.Config, &i.CreatedAt, &i.ChannelSenderID, ) return i, err } const createChannelUserBinding = `-- name: CreateChannelUserBinding :one INSERT INTO channel_user_binding ( workspace_id, multica_user_id, installation_id, channel_type, channel_user_id, config ) VALUES ( $1, $2, $3, $4, $5, $6 ) ON CONFLICT (installation_id, channel_user_id) DO UPDATE SET -- jsonb_strip_nulls(EXCLUDED.config) preserves the old lark semantics -- ` + "`" + `union_id = COALESCE(EXCLUDED.union_id, lark_user_binding.union_id)` + "`" + `: -- a re-bind that carries ` + "`" + `{"union_id": null}` + "`" + ` (or omits the key) must NOT -- erase a union_id we already captured. Only non-null incoming keys win. config = channel_user_binding.config || jsonb_strip_nulls(EXCLUDED.config), bound_at = now() WHERE channel_user_binding.multica_user_id = EXCLUDED.multica_user_id RETURNING id, workspace_id, multica_user_id, installation_id, channel_type, channel_user_id, config, bound_at ` type CreateChannelUserBindingParams struct { WorkspaceID pgtype.UUID `json:"workspace_id"` MulticaUserID pgtype.UUID `json:"multica_user_id"` InstallationID pgtype.UUID `json:"installation_id"` ChannelType string `json:"channel_type"` ChannelUserID string `json:"channel_user_id"` Config []byte `json:"config"` } // ===================== // channel_user_binding // ===================== // Records that a platform user id (per-installation; Feishu open_id) maps // to a Multica user. The old composite member-FK is gone, so this no // longer fails when the redeemer is not a workspace member — the caller // (BindingTokenService.RedeemAndBind) validates membership explicitly // before calling. ON CONFLICT DO UPDATE is still gated on multica_user_id // matching, so a second redeemer cannot steal an already-bound user id; // a cross-user conflict updates zero rows and the caller maps that to // ErrBindingAlreadyAssigned. config carries secondary identity (union_id). func (q *Queries) CreateChannelUserBinding(ctx context.Context, arg CreateChannelUserBindingParams) (ChannelUserBinding, error) { row := q.db.QueryRow(ctx, createChannelUserBinding, arg.WorkspaceID, arg.MulticaUserID, arg.InstallationID, arg.ChannelType, arg.ChannelUserID, arg.Config, ) var i ChannelUserBinding err := row.Scan( &i.ID, &i.WorkspaceID, &i.MulticaUserID, &i.InstallationID, &i.ChannelType, &i.ChannelUserID, &i.Config, &i.BoundAt, ) return i, err } const deleteChannelBindingTokensByInstallation = `-- name: DeleteChannelBindingTokensByInstallation :exec DELETE FROM channel_binding_token WHERE installation_id = $1 ` // Application-layer integrity (schema has no FK/cascade, MUL-3515 §4): drop // every pending binding token for an installation that is being hard-deleted. // A token stays redeemable for up to 15 min; without this a user who clicks a // still-unexpired bind link right after the bot was rebound to another agent // would consume the token and get a "bound" result written against a deleted // installation — a link that never actually reaches the live bot. func (q *Queries) DeleteChannelBindingTokensByInstallation(ctx context.Context, installationID pgtype.UUID) error { _, err := q.db.Exec(ctx, deleteChannelBindingTokensByInstallation, installationID) return err } const deleteChannelChatSessionBindingBySession = `-- name: DeleteChannelChatSessionBindingBySession :exec WITH target AS ( SELECT binding.id FROM channel_chat_session_binding AS binding WHERE binding.chat_session_id = $1 ), cleared_deliveries AS ( DELETE FROM channel_task_delivery AS delivery WHERE delivery.binding_id IN (SELECT id FROM target) ), cleared_outbound AS ( DELETE FROM channel_outbound_message AS outbound WHERE outbound.binding_id IN (SELECT id FROM target) ), cleared_reply_deliveries AS ( DELETE FROM channel_reply_delivery AS reply WHERE reply.binding_id IN (SELECT id FROM target) ), deleted_binding AS ( DELETE FROM channel_chat_session_binding AS binding WHERE binding.id IN (SELECT id FROM target) ) DELETE FROM channel_chat_context_generation AS generation WHERE generation.chat_session_id = $1 ` // Application-layer integrity (replaces the old chat_session-FK ON DELETE // CASCADE): drop the binding, its immutable delivery/history dependents, // and the Chat's context generations. func (q *Queries) DeleteChannelChatSessionBindingBySession(ctx context.Context, chatSessionID pgtype.UUID) error { _, err := q.db.Exec(ctx, deleteChannelChatSessionBindingBySession, chatSessionID) return err } const deleteChannelChatSessionBindingsByInstallation = `-- name: DeleteChannelChatSessionBindingsByInstallation :exec WITH cleared_deliveries AS ( DELETE FROM channel_task_delivery AS delivery WHERE delivery.installation_id = $1 AND delivery.channel_type = $2 ), cleared_outbound AS ( DELETE FROM channel_outbound_message AS outbound WHERE outbound.installation_id = $1 AND outbound.channel_type = $2 ), cleared_reply_deliveries AS ( DELETE FROM channel_reply_delivery AS reply WHERE reply.installation_id = $1 AND reply.channel_type = $2 ) DELETE FROM channel_chat_session_binding AS binding WHERE binding.installation_id = $1 AND binding.channel_type = $2 ` type DeleteChannelChatSessionBindingsByInstallationParams struct { InstallationID pgtype.UUID `json:"installation_id"` ChannelType string `json:"channel_type"` } // Retire every chat-session binding for an installation. Used when an // installation is re-pointed to a different agent (Slack re-connect): each // existing chat_session is permanently tied to the agent it was created under, // so reusing it would keep routing the conversation to the OLD agent. Dropping // the bindings forces the next inbound message to create a fresh session under // the new agent. The chat_session rows are preserved for history; only the // channel binding is removed. Context generations belong to the preserved Chat, // not to the retired binding: in-flight task history snapshots may still read // them after this statement commits. func (q *Queries) DeleteChannelChatSessionBindingsByInstallation(ctx context.Context, arg DeleteChannelChatSessionBindingsByInstallationParams) error { _, err := q.db.Exec(ctx, deleteChannelChatSessionBindingsByInstallation, arg.InstallationID, arg.ChannelType) return err } const deleteChannelInstallationsBySystemRuntimeAgents = `-- name: DeleteChannelInstallationsBySystemRuntimeAgents :exec WITH system_agents AS ( SELECT system_agent.id FROM agent AS system_agent WHERE system_agent.runtime_id = $1 AND system_agent.kind = 'system' ), doomed_sessions AS ( SELECT id FROM chat_session WHERE agent_id IN (SELECT id FROM system_agents) ), doomed AS ( SELECT id FROM channel_installation WHERE agent_id IN (SELECT id FROM system_agents) ), cleared_dingtalk_group_presence AS ( DELETE FROM dingtalk_group_presence WHERE installation_id IN (SELECT id FROM doomed) ), cleared_dingtalk_bot_identity AS ( DELETE FROM dingtalk_bot_identity WHERE installation_id IN (SELECT id FROM doomed) ), cleared_dingtalk_group_routes AS ( DELETE FROM dingtalk_group_route WHERE installation_id IN (SELECT id FROM doomed) ), cleared_reply_deliveries AS ( DELETE FROM channel_reply_delivery WHERE installation_id IN (SELECT id FROM doomed) ), cleared_task_deliveries AS ( DELETE FROM channel_task_delivery WHERE installation_id IN (SELECT id FROM doomed) ), cleared_outbound_messages AS ( DELETE FROM channel_outbound_message WHERE installation_id IN (SELECT id FROM doomed) ), cleared_chat_sessions AS ( DELETE FROM channel_chat_session_binding WHERE installation_id IN (SELECT id FROM doomed) RETURNING chat_session_id ), cleared_chat_contexts AS ( DELETE FROM channel_chat_context_generation WHERE chat_session_id IN (SELECT chat_session_id FROM cleared_chat_sessions) OR chat_session_id IN (SELECT id FROM doomed_sessions) ), cleared_outbound_cards AS ( -- Reach channel_outbound_card_message (keyed by chat_session_id, no FK) -- through the just-removed chat-session bindings, same as the reclaim path. DELETE FROM channel_outbound_card_message WHERE chat_session_id IN (SELECT chat_session_id FROM cleared_chat_sessions) ), cleared_binding_tokens AS ( DELETE FROM channel_binding_token WHERE installation_id IN (SELECT id FROM doomed) ), cleared_user_bindings AS ( DELETE FROM channel_user_binding WHERE installation_id IN (SELECT id FROM doomed) ), cleared_inbound_dedup AS ( DELETE FROM channel_inbound_message_dedup WHERE installation_id IN (SELECT id FROM doomed) ), cleared_audit AS ( -- Hard delete: purge audit rows rather than detaching them into permanently -- unattributable NULL rows (channel_inbound_audit has no workspace_id / reaper). DELETE FROM channel_inbound_audit WHERE installation_id IN (SELECT id FROM doomed) ) DELETE FROM channel_installation WHERE id IN (SELECT id FROM doomed) ` // Application-layer replacement for the (deliberately absent, MUL-3515 §4) // workspace/agent ON DELETE CASCADE: on runtime teardown, before the system // agents are hard-deleted, remove every channel installation they own — plus all // of each installation's dependent rows — so no orphaned installation keeps // occupying its bot's (channel_type, app_id) routing slot after its agent is gone // (#4810). MUST run in the same tx as, and BEFORE, DeleteSystemAgentsByRuntime. // Mirrors the agent hard-delete predicate (runtime_id, kind = 'system') exactly. // // Scoped to kind = 'system' since MUL-5559: a user agent now survives its // runtime's deletion as an unbound agent, so tearing down its installations // here would take a working bot away from an agent that is still there. func (q *Queries) DeleteChannelInstallationsBySystemRuntimeAgents(ctx context.Context, runtimeID pgtype.UUID) error { _, err := q.db.Exec(ctx, deleteChannelInstallationsBySystemRuntimeAgents, runtimeID) return err } const deleteChannelMediaPendingObject = `-- name: DeleteChannelMediaPendingObject :execrows DELETE FROM channel_media_pending_object WHERE storage_key = $1 AND workspace_id = $2 AND lease_token = $3 ` type DeleteChannelMediaPendingObjectParams struct { StorageKey string `json:"storage_key"` WorkspaceID pgtype.UUID `json:"workspace_id"` LeaseToken pgtype.UUID `json:"lease_token"` } // Drops a claimed row for good: a durable attachment reference was found, or // the tombstone's re-delete schedule is exhausted. Lease-token guarded so an // expired-lease reclaim by another // replica cannot be clobbered; workspace_id explicit per the tenancy rule. func (q *Queries) DeleteChannelMediaPendingObject(ctx context.Context, arg DeleteChannelMediaPendingObjectParams) (int64, error) { result, err := q.db.Exec(ctx, deleteChannelMediaPendingObject, arg.StorageKey, arg.WorkspaceID, arg.LeaseToken) if err != nil { return 0, err } return result.RowsAffected(), nil } const deleteChannelOutboundCardMessagesBySession = `-- name: DeleteChannelOutboundCardMessagesBySession :exec DELETE FROM channel_outbound_card_message WHERE chat_session_id = $1 ` // Application-layer integrity (channel_* has no FK/cascade, MUL-3515 §4): drop the // outbound card-message rows for a chat_session being deleted. They are keyed by // chat_session_id with no FK and no reaper, so the standalone chat-session delete // path must prune them here alongside DeleteChannelChatSessionBindingBySession — // otherwise deleting a chat session leaves them as permanent orphans (Elon's // follow-up on #4810; the workspace/agent/reclaim sweeps already cover their // paths). A card that survived its session could only mis-route a later patch. func (q *Queries) DeleteChannelOutboundCardMessagesBySession(ctx context.Context, chatSessionID pgtype.UUID) error { _, err := q.db.Exec(ctx, deleteChannelOutboundCardMessagesBySession, chatSessionID) return err } const deleteChannelUserBindingsByInstallation = `-- name: DeleteChannelUserBindingsByInstallation :exec DELETE FROM channel_user_binding WHERE installation_id = $1 ` // Application-layer integrity (schema has no FK/cascade, MUL-3515 §4): drop // every member account link for an installation that is being hard-deleted. // Rebinding a Feishu bot to a DIFFERENT agent starts a fresh installation, so // old links do not follow — a different agent is a distinct connection and // members re-establish their link on first contact. The rows could never be // reused anyway (every Feishu identity lookup is installation_id-scoped, and // FindReusableChannelUserBinding is Slack-only), so removing them just keeps // dead rows from accumulating. func (q *Queries) DeleteChannelUserBindingsByInstallation(ctx context.Context, installationID pgtype.UUID) error { _, err := q.db.Exec(ctx, deleteChannelUserBindingsByInstallation, installationID) return err } const deleteChannelUserBindingsByWorkspaceMember = `-- name: DeleteChannelUserBindingsByWorkspaceMember :exec DELETE FROM channel_user_binding WHERE workspace_id = $1 AND multica_user_id = $2 ` type DeleteChannelUserBindingsByWorkspaceMemberParams struct { WorkspaceID pgtype.UUID `json:"workspace_id"` MulticaUserID pgtype.UUID `json:"multica_user_id"` } // Application-layer integrity (replaces the old member-FK ON DELETE // CASCADE): prune every binding for a user who has been removed from a // workspace, across all installations in that workspace. func (q *Queries) DeleteChannelUserBindingsByWorkspaceMember(ctx context.Context, arg DeleteChannelUserBindingsByWorkspaceMemberParams) error { _, err := q.db.Exec(ctx, deleteChannelUserBindingsByWorkspaceMember, arg.WorkspaceID, arg.MulticaUserID) return err } const findChannelBindingForMember = `-- name: FindChannelBindingForMember :one SELECT b.id, b.workspace_id, b.multica_user_id, b.installation_id, b.channel_type, b.channel_user_id, b.config, b.bound_at FROM channel_user_binding b JOIN channel_installation ci ON ci.id = b.installation_id WHERE b.workspace_id = $1 AND b.multica_user_id = $2 AND b.channel_type = $3 AND ci.status = 'active' ORDER BY b.bound_at DESC LIMIT 1 ` type FindChannelBindingForMemberParams struct { WorkspaceID pgtype.UUID `json:"workspace_id"` MulticaUserID pgtype.UUID `json:"multica_user_id"` ChannelType string `json:"channel_type"` } // Outbound notification lookup: given a Multica member and a channel_type, // return the (installation, channel_user_id) that outbound push should // target. The wecom smart-bot inbox-notification path uses this to decide // whether to deliver via the bot at all — no row means "unbound member, // fall back to the legacy path (TOF/RTX)". // // If a member has bound multiple installations of the same channel_type in // one workspace (multi-bot org), the most-recently-bound wins — matches // FindReusableChannelUserBinding's tiebreak so the two lookups agree. func (q *Queries) FindChannelBindingForMember(ctx context.Context, arg FindChannelBindingForMemberParams) (ChannelUserBinding, error) { row := q.db.QueryRow(ctx, findChannelBindingForMember, arg.WorkspaceID, arg.MulticaUserID, arg.ChannelType) var i ChannelUserBinding err := row.Scan( &i.ID, &i.WorkspaceID, &i.MulticaUserID, &i.InstallationID, &i.ChannelType, &i.ChannelUserID, &i.Config, &i.BoundAt, ) return i, err } const findChannelBindingForMemberOnAgentInstallation = `-- name: FindChannelBindingForMemberOnAgentInstallation :one SELECT b.id, b.workspace_id, b.multica_user_id, b.installation_id, b.channel_type, b.channel_user_id, b.config, b.bound_at FROM channel_user_binding b JOIN channel_installation ci ON ci.id = b.installation_id WHERE b.workspace_id = $1 AND b.multica_user_id = $2 AND b.channel_type = $3 AND ci.channel_type = $3 AND ci.agent_id = $4 AND ci.status = 'active' ORDER BY b.bound_at DESC LIMIT 1 ` type FindChannelBindingForMemberOnAgentInstallationParams struct { WorkspaceID pgtype.UUID `json:"workspace_id"` MulticaUserID pgtype.UUID `json:"multica_user_id"` ChannelType string `json:"channel_type"` AgentID pgtype.UUID `json:"agent_id"` } // Agent-attributed notification lookup: the same member lookup as // FindChannelBindingForMember, pinned to the ONE installation the agent whose // activity raised the notification owns. // // Why this exists. FindChannelBindingForMember answers by the // RECIPIENT alone, so in a multi-bot org — one bot per agent, several of them // bound by the same person — every notification a member receives is pushed // through whichever bot they bound most recently, no matter which agent the // notification is about. The recipient reads three agents' activity as if it // all came from one of them. Pinning the lookup to the acting agent's bot is // what makes the notification follow the bot↔agent mapping the sender is // reading. // // The tiebreak stays for the case where one member holds two channel users on // the same bot (two WeCom accounts, one Multica account): both rows answer, // and `:one` cannot take more than one, so the newest binding wins here // exactly as it does in the recipient-wide lookup. // // channel_type is named once and used for both columns: the installation's // own discriminator and the binding's must agree, or installs of two // platforms could cross-answer here. func (q *Queries) FindChannelBindingForMemberOnAgentInstallation(ctx context.Context, arg FindChannelBindingForMemberOnAgentInstallationParams) (ChannelUserBinding, error) { row := q.db.QueryRow(ctx, findChannelBindingForMemberOnAgentInstallation, arg.WorkspaceID, arg.MulticaUserID, arg.ChannelType, arg.AgentID, ) var i ChannelUserBinding err := row.Scan( &i.ID, &i.WorkspaceID, &i.MulticaUserID, &i.InstallationID, &i.ChannelType, &i.ChannelUserID, &i.Config, &i.BoundAt, ) return i, err } const findLiveChannelBindingToken = `-- name: FindLiveChannelBindingToken :one SELECT token_hash, workspace_id, installation_id, channel_type, channel_user_id, expires_at, consumed_at, created_at FROM channel_binding_token WHERE installation_id = $1 AND channel_type = $2 AND channel_user_id = $3 AND consumed_at IS NULL AND expires_at > now() AND created_at >= now() - $4::interval ORDER BY created_at DESC LIMIT 1 ` type FindLiveChannelBindingTokenParams struct { InstallationID pgtype.UUID `json:"installation_id"` ChannelType string `json:"channel_type"` ChannelUserID string `json:"channel_user_id"` MintInterval pgtype.Interval `json:"mint_interval"` } // Mint guard: the newest token for this platform user that is still // unconsumed, unexpired, and recent enough that the link already sitting in // their chat is the one to point back at. Without it every message from an // unbound user mints another row, so a user who keeps typing at a bot they // have not linked yet writes one row per message. This narrows that to // roughly one row per window; it is not a hard guarantee, since the caller // runs this and the insert as two statements. // // `mint_interval` is the caller's throttle window (see // wecom.BindingTokenMintInterval). It is subtracted from now() rather than // passed in as an absolute cutoff so the whole window is measured on the // database clock: created_at is stamped by the column default, and comparing // it against an application-side timestamp would let clock skew between the // two stretch or shrink the window. The consumed_at / expires_at predicates // keep an already-redeemed or stale token from suppressing a mint the user // actually needs. // // idx_channel_binding_token_installation covers the installation_id prefix; // the rest is a filter over that installation's live tokens, which is a small // set because nothing here outlives the 15-minute TTL. func (q *Queries) FindLiveChannelBindingToken(ctx context.Context, arg FindLiveChannelBindingTokenParams) (ChannelBindingToken, error) { row := q.db.QueryRow(ctx, findLiveChannelBindingToken, arg.InstallationID, arg.ChannelType, arg.ChannelUserID, arg.MintInterval, ) var i ChannelBindingToken err := row.Scan( &i.TokenHash, &i.WorkspaceID, &i.InstallationID, &i.ChannelType, &i.ChannelUserID, &i.ExpiresAt, &i.ConsumedAt, &i.CreatedAt, ) return i, err } const findReusableChannelUserBinding = `-- name: FindReusableChannelUserBinding :one SELECT b.id, b.workspace_id, b.multica_user_id, b.installation_id, b.channel_type, b.channel_user_id, b.config, b.bound_at FROM channel_user_binding b JOIN channel_installation ci ON ci.id = b.installation_id WHERE b.workspace_id = $1 AND b.channel_type = $2 AND b.channel_user_id = $3 AND ci.config ->> 'team_id' = $4::text ORDER BY b.bound_at DESC LIMIT 1 ` type FindReusableChannelUserBindingParams struct { WorkspaceID pgtype.UUID `json:"workspace_id"` ChannelType string `json:"channel_type"` ChannelUserID string `json:"channel_user_id"` TeamID string `json:"team_id"` } // Cross-installation account-link reuse (MUL-3911). When a platform user // messages an installation they have NOT linked, but the SAME user id is already // bound to ANOTHER installation in the SAME Multica workspace + SAME Slack team, // the inbound identity step reuses that link instead of re-prompting. Slack user // ids are stable within a team, so an identical channel_user_id denotes the same // human across that team's apps. The match is fenced to one workspace AND one // team (installation config->>'team_id'): a Slack team can be connected to two // different Multica workspaces, and a user may hold different Multica accounts in // each, so reuse must cross neither boundary. Most-recently-bound wins. The // caller re-checks membership and materializes a fresh per-installation binding. // // team_id is pinned ::text so sqlc types the arg as a string instead of // attributing the bare param to the JSONB config column (mirrors // GetChannelInstallationByAppID's app_id cast). func (q *Queries) FindReusableChannelUserBinding(ctx context.Context, arg FindReusableChannelUserBindingParams) (ChannelUserBinding, error) { row := q.db.QueryRow(ctx, findReusableChannelUserBinding, arg.WorkspaceID, arg.ChannelType, arg.ChannelUserID, arg.TeamID, ) var i ChannelUserBinding err := row.Scan( &i.ID, &i.WorkspaceID, &i.MulticaUserID, &i.InstallationID, &i.ChannelType, &i.ChannelUserID, &i.Config, &i.BoundAt, ) return i, err } const getChannelChatContextGeneration = `-- name: GetChannelChatContextGeneration :one SELECT chat_session_id, revision, history_start_message_id, history_end_message_id, history_boundary_pending, pending_fresh, initiator_user_id, created_at, last_message_id, last_thread_id, last_sender_id FROM channel_chat_context_generation WHERE chat_session_id = $1 AND revision = $2 ` type GetChannelChatContextGenerationParams struct { ChatSessionID pgtype.UUID `json:"chat_session_id"` Revision int64 `json:"revision"` } func (q *Queries) GetChannelChatContextGeneration(ctx context.Context, arg GetChannelChatContextGenerationParams) (ChannelChatContextGeneration, error) { row := q.db.QueryRow(ctx, getChannelChatContextGeneration, arg.ChatSessionID, arg.Revision) var i ChannelChatContextGeneration err := row.Scan( &i.ChatSessionID, &i.Revision, &i.HistoryStartMessageID, &i.HistoryEndMessageID, &i.HistoryBoundaryPending, &i.PendingFresh, &i.InitiatorUserID, &i.CreatedAt, &i.LastMessageID, &i.LastThreadID, &i.LastSenderID, ) return i, err } const getChannelChatSessionBinding = `-- name: GetChannelChatSessionBinding :one SELECT id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at, pending_fresh, context_revision, route_revision, retired_at, history_start_message_id, history_end_message_id, history_boundary_pending FROM channel_chat_session_binding WHERE installation_id = $1 AND channel_chat_id = $2 AND retired_at IS NULL ` type GetChannelChatSessionBindingParams struct { InstallationID pgtype.UUID `json:"installation_id"` ChannelChatID string `json:"channel_chat_id"` } // Lookup-by-channel-chat: the inbound dispatcher finds the existing // chat_session before deciding whether to create one. func (q *Queries) GetChannelChatSessionBinding(ctx context.Context, arg GetChannelChatSessionBindingParams) (ChannelChatSessionBinding, error) { row := q.db.QueryRow(ctx, getChannelChatSessionBinding, arg.InstallationID, arg.ChannelChatID) var i ChannelChatSessionBinding err := row.Scan( &i.ID, &i.ChatSessionID, &i.InstallationID, &i.ChannelType, &i.ChannelChatID, &i.ChatType, &i.LastMessageID, &i.LastThreadID, &i.Config, &i.CreatedAt, &i.PendingFresh, &i.ContextRevision, &i.RouteRevision, &i.RetiredAt, &i.HistoryStartMessageID, &i.HistoryEndMessageID, &i.HistoryBoundaryPending, ) return i, err } const getChannelChatSessionBindingBySession = `-- name: GetChannelChatSessionBindingBySession :one SELECT id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at, pending_fresh, context_revision, route_revision, retired_at, history_start_message_id, history_end_message_id, history_boundary_pending FROM channel_chat_session_binding WHERE chat_session_id = $1 AND channel_type = $2 ` type GetChannelChatSessionBindingBySessionParams struct { ChatSessionID pgtype.UUID `json:"chat_session_id"` ChannelType string `json:"channel_type"` } // Reverse lookup for the outbound patcher: given a chat_session_id, find // its channel binding to know which (installation, chat_id) to send to. // Scoped by channel_type so a future non-Feishu binding on the same // chat_session is never treated as a Feishu reply target. func (q *Queries) GetChannelChatSessionBindingBySession(ctx context.Context, arg GetChannelChatSessionBindingBySessionParams) (ChannelChatSessionBinding, error) { row := q.db.QueryRow(ctx, getChannelChatSessionBindingBySession, arg.ChatSessionID, arg.ChannelType) var i ChannelChatSessionBinding err := row.Scan( &i.ID, &i.ChatSessionID, &i.InstallationID, &i.ChannelType, &i.ChannelChatID, &i.ChatType, &i.LastMessageID, &i.LastThreadID, &i.Config, &i.CreatedAt, &i.PendingFresh, &i.ContextRevision, &i.RouteRevision, &i.RetiredAt, &i.HistoryStartMessageID, &i.HistoryEndMessageID, &i.HistoryBoundaryPending, ) return i, err } const getChannelChatSessionBindingBySessionAny = `-- name: GetChannelChatSessionBindingBySessionAny :one SELECT id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at, pending_fresh, context_revision, route_revision, retired_at, history_start_message_id, history_end_message_id, history_boundary_pending FROM channel_chat_session_binding WHERE chat_session_id = $1 ` // Channel-agnostic reverse lookup: which channel, if any, is behind this // chat_session? UNIQUE (chat_session_id) guarantees at most one row, so a // caller that only needs to READ the binding never has to name the channel it // is hoping for — and therefore cannot go blind on a channel added later. // The channel_type-scoped variant above stays for the outbound senders, which // are per-platform by construction and must not deliver into a foreign one. func (q *Queries) GetChannelChatSessionBindingBySessionAny(ctx context.Context, chatSessionID pgtype.UUID) (ChannelChatSessionBinding, error) { row := q.db.QueryRow(ctx, getChannelChatSessionBindingBySessionAny, chatSessionID) var i ChannelChatSessionBinding err := row.Scan( &i.ID, &i.ChatSessionID, &i.InstallationID, &i.ChannelType, &i.ChannelChatID, &i.ChatType, &i.LastMessageID, &i.LastThreadID, &i.Config, &i.CreatedAt, &i.PendingFresh, &i.ContextRevision, &i.RouteRevision, &i.RetiredAt, &i.HistoryStartMessageID, &i.HistoryEndMessageID, &i.HistoryBoundaryPending, ) return i, err } const getChannelInstallation = `-- name: GetChannelInstallation :one SELECT id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at FROM channel_installation WHERE id = $1 AND channel_type = $2 ` type GetChannelInstallationParams struct { ID pgtype.UUID `json:"id"` ChannelType string `json:"channel_type"` } // Scoped by channel_type: a per-channel caller (e.g. the Feishu store) // must never resolve another channel's installation by guessing its UUID. func (q *Queries) GetChannelInstallation(ctx context.Context, arg GetChannelInstallationParams) (ChannelInstallation, error) { row := q.db.QueryRow(ctx, getChannelInstallation, arg.ID, arg.ChannelType) var i ChannelInstallation err := row.Scan( &i.ID, &i.WorkspaceID, &i.AgentID, &i.ChannelType, &i.Config, &i.Status, &i.WsLeaseToken, &i.WsLeaseExpiresAt, &i.InstallerUserID, &i.InstalledAt, &i.CreatedAt, &i.UpdatedAt, ) return i, err } const getChannelInstallationByAppID = `-- name: GetChannelInstallationByAppID :one SELECT id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at FROM channel_installation WHERE channel_type = $1 AND config ->> 'app_id' = $2::text ` type GetChannelInstallationByAppIDParams struct { ChannelType string `json:"channel_type"` AppID string `json:"app_id"` } // Inbound routing. The platform event carries only the channel's app // identifier (Feishu app_id); the dispatcher's installation resolver routes // on (channel_type, config->>'app_id'). Backed by the functional unique // index idx_channel_installation_type_appid. // // Both params are named + explicitly typed: `config ->> 'app_id'` makes sqlc // attribute a bare `$2` to the JSONB `config` column (it would emit // `Config []byte`), so we pin the app_id arg to ::text to get AppID string. func (q *Queries) GetChannelInstallationByAppID(ctx context.Context, arg GetChannelInstallationByAppIDParams) (ChannelInstallation, error) { row := q.db.QueryRow(ctx, getChannelInstallationByAppID, arg.ChannelType, arg.AppID) var i ChannelInstallation err := row.Scan( &i.ID, &i.WorkspaceID, &i.AgentID, &i.ChannelType, &i.Config, &i.Status, &i.WsLeaseToken, &i.WsLeaseExpiresAt, &i.InstallerUserID, &i.InstalledAt, &i.CreatedAt, &i.UpdatedAt, ) return i, err } const getChannelInstallationInWorkspace = `-- name: GetChannelInstallationInWorkspace :one SELECT id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at FROM channel_installation WHERE id = $1 AND workspace_id = $2 AND channel_type = $3 ` type GetChannelInstallationInWorkspaceParams struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` ChannelType string `json:"channel_type"` } func (q *Queries) GetChannelInstallationInWorkspace(ctx context.Context, arg GetChannelInstallationInWorkspaceParams) (ChannelInstallation, error) { row := q.db.QueryRow(ctx, getChannelInstallationInWorkspace, arg.ID, arg.WorkspaceID, arg.ChannelType) var i ChannelInstallation err := row.Scan( &i.ID, &i.WorkspaceID, &i.AgentID, &i.ChannelType, &i.Config, &i.Status, &i.WsLeaseToken, &i.WsLeaseExpiresAt, &i.InstallerUserID, &i.InstalledAt, &i.CreatedAt, &i.UpdatedAt, ) return i, err } const getChannelInstallationOwnerByAppID = `-- name: GetChannelInstallationOwnerByAppID :one SELECT ci.workspace_id, ci.agent_id, a.archived_at AS agent_archived_at FROM channel_installation ci JOIN agent a ON a.id = ci.agent_id WHERE ci.channel_type = $1 AND ci.config ->> 'app_id' = $2::text ` type GetChannelInstallationOwnerByAppIDParams struct { ChannelType string `json:"channel_type"` AppID string `json:"app_id"` } type GetChannelInstallationOwnerByAppIDRow struct { WorkspaceID pgtype.UUID `json:"workspace_id"` AgentID pgtype.UUID `json:"agent_id"` AgentArchivedAt pgtype.Timestamptz `json:"agent_archived_at"` } // Identifies the LIVE owner of a (channel_type, config->>'app_id') routing slot // so the install path can refuse a rebind with an ACCURATE message instead of the // old catch-all "connected to a different Multica workspace". Meant to be read // only after ReclaimDeadChannelInstallationByAppID has removed every DEAD owner, // so a returned row is a live active owner. `agent_archived` distinguishes an // archived (reversible) owner — its bot stays owned, recovered by unarchiving the // agent or disconnecting the bot — from a plain active one. The JOIN drops a row // whose agent no longer exists (an orphan the reclaim gate should already have // cleared), so a missing row (pgx.ErrNoRows) means "no live owner". The caller // reads agent_archived_at.Valid to tell an archived (reversible) owner apart. func (q *Queries) GetChannelInstallationOwnerByAppID(ctx context.Context, arg GetChannelInstallationOwnerByAppIDParams) (GetChannelInstallationOwnerByAppIDRow, error) { row := q.db.QueryRow(ctx, getChannelInstallationOwnerByAppID, arg.ChannelType, arg.AppID) var i GetChannelInstallationOwnerByAppIDRow err := row.Scan(&i.WorkspaceID, &i.AgentID, &i.AgentArchivedAt) return i, err } const getChannelInstallationSlotOwnerByAppID = `-- name: GetChannelInstallationSlotOwnerByAppID :one SELECT ci.id, ci.workspace_id, ci.agent_id, ci.status, a.archived_at AS agent_archived_at, (a.id IS NOT NULL)::boolean AS agent_exists, (w.id IS NOT NULL)::boolean AS workspace_exists FROM channel_installation ci LEFT JOIN agent a ON a.id = ci.agent_id LEFT JOIN workspace w ON w.id = ci.workspace_id WHERE ci.channel_type = $1 AND ci.config ->> 'app_id' = $2::text ` type GetChannelInstallationSlotOwnerByAppIDParams struct { ChannelType string `json:"channel_type"` AppID string `json:"app_id"` } type GetChannelInstallationSlotOwnerByAppIDRow struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` AgentID pgtype.UUID `json:"agent_id"` Status string `json:"status"` AgentArchivedAt pgtype.Timestamptz `json:"agent_archived_at"` AgentExists bool `json:"agent_exists"` WorkspaceExists bool `json:"workspace_exists"` } // Everything the install path needs to classify the current holder of a // (channel_type, config->>'app_id') slot BEFORE it acts on it, in one read. // Distinct from GetChannelInstallationOwnerByAppID, which is the after-the-fact // "name the conflict" read and INNER JOINs the agent away. // // Here the joins are LEFT so an ORPHAN row survives the read: with no FKs // (MUL-3515 §4) an installation outlives a deleted workspace or agent, and the // caller has to tell "orphan, reclaimable" apart from "live owner, refuse". // workspace_exists / agent_exists carry that; status and agent_archived_at // carry the rest of ReclaimDeadChannelInstallationByAppID's own definition of // dead, so the caller can predict what the reclaim would do without running it. // pgx.ErrNoRows means the slot is free. func (q *Queries) GetChannelInstallationSlotOwnerByAppID(ctx context.Context, arg GetChannelInstallationSlotOwnerByAppIDParams) (GetChannelInstallationSlotOwnerByAppIDRow, error) { row := q.db.QueryRow(ctx, getChannelInstallationSlotOwnerByAppID, arg.ChannelType, arg.AppID) var i GetChannelInstallationSlotOwnerByAppIDRow err := row.Scan( &i.ID, &i.WorkspaceID, &i.AgentID, &i.Status, &i.AgentArchivedAt, &i.AgentExists, &i.WorkspaceExists, ) return i, err } const getChannelOutboundCardByTask = `-- name: GetChannelOutboundCardByTask :one SELECT id, chat_session_id, task_id, channel_type, channel_chat_id, channel_card_message_id, status, last_patched_at, created_at FROM channel_outbound_card_message WHERE task_id = $1 AND channel_type = $2 ` type GetChannelOutboundCardByTaskParams struct { TaskID pgtype.UUID `json:"task_id"` ChannelType string `json:"channel_type"` } // The partial unique index on (task_id) WHERE task_id IS NOT NULL // guarantees at most one row. Scoped by channel_type so a future non-Feishu // card for the same task is not patched as a Feishu card. func (q *Queries) GetChannelOutboundCardByTask(ctx context.Context, arg GetChannelOutboundCardByTaskParams) (ChannelOutboundCardMessage, error) { row := q.db.QueryRow(ctx, getChannelOutboundCardByTask, arg.TaskID, arg.ChannelType) var i ChannelOutboundCardMessage err := row.Scan( &i.ID, &i.ChatSessionID, &i.TaskID, &i.ChannelType, &i.ChannelChatID, &i.ChannelCardMessageID, &i.Status, &i.LastPatchedAt, &i.CreatedAt, ) return i, err } const getChannelReplyDelivery = `-- name: GetChannelReplyDelivery :one SELECT turn_id, task_id, binding_id, installation_id, channel_type, chat_id, phase, send_state, message_id, chunks_sent, owner_token, owner_expires_at, settled_reason, created_at, updated_at, attempt_depth FROM channel_reply_delivery WHERE turn_id = $1 ` // Read without taking the lease, so a caller that lost the race can tell // "someone else is working on it" from "this turn is finished". func (q *Queries) GetChannelReplyDelivery(ctx context.Context, turnID pgtype.UUID) (ChannelReplyDelivery, error) { row := q.db.QueryRow(ctx, getChannelReplyDelivery, turnID) var i ChannelReplyDelivery err := row.Scan( &i.TurnID, &i.TaskID, &i.BindingID, &i.InstallationID, &i.ChannelType, &i.ChatID, &i.Phase, &i.SendState, &i.MessageID, &i.ChunksSent, &i.OwnerToken, &i.OwnerExpiresAt, &i.SettledReason, &i.CreatedAt, &i.UpdatedAt, &i.AttemptDepth, ) return i, err } const getChannelReplyTurn = `-- name: GetChannelReplyTurn :one WITH RECURSIVE chain(task_id, parent_task_id, depth) AS ( SELECT attempt.id, attempt.retry_of_task_id, 0 FROM agent_task_queue attempt WHERE attempt.id = $1 UNION ALL SELECT parent.id, parent.retry_of_task_id, chain.depth + 1 FROM agent_task_queue parent JOIN chain ON parent.id = chain.parent_task_id ) SELECT (SELECT root.task_id FROM chain root WHERE root.parent_task_id IS NULL LIMIT 1) AS turn_id, (SELECT COALESCE(MAX(step.depth), 0) FROM chain step)::int AS attempt_depth ` type GetChannelReplyTurnRow struct { TurnID pgtype.UUID `json:"turn_id"` AttemptDepth int32 `json:"attempt_depth"` } // --------------------------------------------------------------------------- // Reply delivery ownership (channel_reply_delivery). // // One user turn, one owner, one reply. Every path that can put a message in a // chat proves it holds the turn's lease before it calls the provider, and // proves it still holds the lease when it records what happened. // --------------------------------------------------------------------------- // The root of a task's automatic-retry chain — the user turn all its attempts // belong to — and how far down the chain this attempt sits. A retry runs under // a new task id and must finish the reply its previous attempt started, so // ownership is keyed by the root; depth is what stops an earlier attempt's late // frame from taking the turn back off the retry that superseded it. func (q *Queries) GetChannelReplyTurn(ctx context.Context, id pgtype.UUID) (GetChannelReplyTurnRow, error) { row := q.db.QueryRow(ctx, getChannelReplyTurn, id) var i GetChannelReplyTurnRow err := row.Scan(&i.TurnID, &i.AttemptDepth) return i, err } const getChannelTaskDelivery = `-- name: GetChannelTaskDelivery :one SELECT task_id, binding_id, installation_id, channel_type, channel_chat_id, chat_type, channel_message_id, channel_thread_id, route_revision, config, created_at, channel_sender_id FROM channel_task_delivery WHERE task_id = $1 ` func (q *Queries) GetChannelTaskDelivery(ctx context.Context, taskID pgtype.UUID) (ChannelTaskDelivery, error) { row := q.db.QueryRow(ctx, getChannelTaskDelivery, taskID) var i ChannelTaskDelivery err := row.Scan( &i.TaskID, &i.BindingID, &i.InstallationID, &i.ChannelType, &i.ChannelChatID, &i.ChatType, &i.ChannelMessageID, &i.ChannelThreadID, &i.RouteRevision, &i.Config, &i.CreatedAt, &i.ChannelSenderID, ) return i, err } const getChannelUserBindingByUserID = `-- name: GetChannelUserBindingByUserID :one SELECT id, workspace_id, multica_user_id, installation_id, channel_type, channel_user_id, config, bound_at FROM channel_user_binding WHERE installation_id = $1 AND channel_user_id = $2 ` type GetChannelUserBindingByUserIDParams struct { InstallationID pgtype.UUID `json:"installation_id"` ChannelUserID string `json:"channel_user_id"` } // The inbound identity lookup: does this platform user id map to a Multica // user for this installation? With the member-FK removed, a row's // existence no longer proves current workspace membership — the dispatcher // re-checks membership after this lookup. func (q *Queries) GetChannelUserBindingByUserID(ctx context.Context, arg GetChannelUserBindingByUserIDParams) (ChannelUserBinding, error) { row := q.db.QueryRow(ctx, getChannelUserBindingByUserID, arg.InstallationID, arg.ChannelUserID) var i ChannelUserBinding err := row.Scan( &i.ID, &i.WorkspaceID, &i.MulticaUserID, &i.InstallationID, &i.ChannelType, &i.ChannelUserID, &i.Config, &i.BoundAt, ) return i, err } const isChannelMessageTypingActive = `-- name: IsChannelMessageTypingActive :one SELECT EXISTS ( SELECT 1 FROM chat_message AS message JOIN chat_session AS session ON session.id = message.chat_session_id JOIN channel_chat_session_binding AS binding ON binding.chat_session_id = session.id JOIN channel_installation AS installation ON installation.id = binding.installation_id LEFT JOIN agent_task_queue AS task ON task.id = message.task_id LEFT JOIN agent AS task_agent ON task_agent.id = task.agent_id WHERE message.id = $1 AND session.id = $2 AND session.workspace_id = $3 AND installation.workspace_id = $3 AND installation.id = $4 AND installation.channel_type = $5 AND binding.channel_type = $5 AND installation.status = 'active' AND session.status = 'active' AND binding.retired_at IS NULL AND message.role = 'user' AND message.channel_ingested AND NOT message.channel_typing_settled AND (message.task_id IS NULL OR ( task_agent.workspace_id = $3 AND task.chat_session_id = session.id AND task.status IN ('queued', 'dispatched', 'running', 'waiting_local_directory', 'deferred') )) ) ` type IsChannelMessageTypingActiveParams struct { MessageID pgtype.UUID `json:"message_id"` ChatSessionID pgtype.UUID `json:"chat_session_id"` WorkspaceID pgtype.UUID `json:"workspace_id"` InstallationID pgtype.UUID `json:"installation_id"` ChannelType string `json:"channel_type"` } // Revalidate the exact input after a remote reaction Add. A pending debounce // input has no task yet; a sealed input follows its immutable owning task. // Do not use the binding's latest message or another turn's active task. func (q *Queries) IsChannelMessageTypingActive(ctx context.Context, arg IsChannelMessageTypingActiveParams) (bool, error) { row := q.db.QueryRow(ctx, isChannelMessageTypingActive, arg.MessageID, arg.ChatSessionID, arg.WorkspaceID, arg.InstallationID, arg.ChannelType, ) var exists bool err := row.Scan(&exists) return exists, err } const listActiveChannelInstallations = `-- name: ListActiveChannelInstallations :many SELECT ci.id, ci.workspace_id, ci.agent_id, ci.channel_type, ci.config, ci.status, ci.ws_lease_token, ci.ws_lease_expires_at, ci.installer_user_id, ci.installed_at, ci.created_at, ci.updated_at FROM channel_installation ci JOIN workspace w ON w.id = ci.workspace_id JOIN agent a ON a.id = ci.agent_id WHERE ci.status = 'active' AND ci.channel_type = $1 ORDER BY ci.created_at ASC ` // Boot path for a per-channel-type inbound hub: every active installation of // the given channel_type, so a hub claims leases and opens connections only // for its own platform and never supervises another channel's installation. // // The JOINs require the owning workspace and agent rows to still exist. // channel_installation has no FK (MUL-3515 §4), so unlike the old // lark_installation (which cascaded away on workspace/agent deletion) an // installation can be orphaned when its workspace is deleted or its agent is // hard-deleted (e.g. runtime teardown). Without this guard the hub would keep // opening a WebSocket for a bot whose workspace/agent is gone. The JOIN matches // the old ON DELETE CASCADE semantics: it filters on row existence, not agent // archival, so an archived-but-present agent's installation is still listed. func (q *Queries) ListActiveChannelInstallations(ctx context.Context, channelType string) ([]ChannelInstallation, error) { rows, err := q.db.Query(ctx, listActiveChannelInstallations, channelType) if err != nil { return nil, err } defer rows.Close() items := []ChannelInstallation{} for rows.Next() { var i ChannelInstallation if err := rows.Scan( &i.ID, &i.WorkspaceID, &i.AgentID, &i.ChannelType, &i.Config, &i.Status, &i.WsLeaseToken, &i.WsLeaseExpiresAt, &i.InstallerUserID, &i.InstalledAt, &i.CreatedAt, &i.UpdatedAt, ); err != nil { return nil, err } items = append(items, i) } if err := rows.Err(); err != nil { return nil, err } return items, nil } const listAllActiveChannelInstallations = `-- name: ListAllActiveChannelInstallations :many SELECT ci.id, ci.workspace_id, ci.agent_id, ci.channel_type, ci.config, ci.status, ci.ws_lease_token, ci.ws_lease_expires_at, ci.installer_user_id, ci.installed_at, ci.created_at, ci.updated_at FROM channel_installation ci JOIN workspace w ON w.id = ci.workspace_id JOIN agent a ON a.id = ci.agent_id WHERE ci.status = 'active' ORDER BY ci.created_at ASC ` // Boot path for the channel-agnostic engine Supervisor (MUL-3620): every // active installation across ALL channel types, so one Supervisor drives every // platform's connections rather than a per-platform hub. This is the de- // hardcoded counterpart of ListActiveChannelInstallations — the Supervisor // routes each row to its registered channel.Factory by channel_type, so it // never needs to know which platforms exist. Same orphan guard as the per-type // query: the workspace + agent JOINs drop installations whose owning rows are // gone (channel_installation has no FK, MUL-3515 §4), matching the old ON // DELETE CASCADE semantics (row existence, not agent archival). func (q *Queries) ListAllActiveChannelInstallations(ctx context.Context) ([]ChannelInstallation, error) { rows, err := q.db.Query(ctx, listAllActiveChannelInstallations) if err != nil { return nil, err } defer rows.Close() items := []ChannelInstallation{} for rows.Next() { var i ChannelInstallation if err := rows.Scan( &i.ID, &i.WorkspaceID, &i.AgentID, &i.ChannelType, &i.Config, &i.Status, &i.WsLeaseToken, &i.WsLeaseExpiresAt, &i.InstallerUserID, &i.InstalledAt, &i.CreatedAt, &i.UpdatedAt, ); err != nil { return nil, err } items = append(items, i) } if err := rows.Err(); err != nil { return nil, err } return items, nil } const listChannelChatSessionBindingsBySessions = `-- name: ListChannelChatSessionBindingsBySessions :many SELECT id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at, pending_fresh, context_revision, route_revision, retired_at, history_start_message_id, history_end_message_id, history_boundary_pending FROM channel_chat_session_binding WHERE chat_session_id = ANY($1::uuid[]) ` // Batch projection for Chat list responses. Historical bindings are retained, // so retired Chats keep their channel source while only the active generation // reports itself as the current route. func (q *Queries) ListChannelChatSessionBindingsBySessions(ctx context.Context, chatSessionIds []pgtype.UUID) ([]ChannelChatSessionBinding, error) { rows, err := q.db.Query(ctx, listChannelChatSessionBindingsBySessions, chatSessionIds) if err != nil { return nil, err } defer rows.Close() items := []ChannelChatSessionBinding{} for rows.Next() { var i ChannelChatSessionBinding if err := rows.Scan( &i.ID, &i.ChatSessionID, &i.InstallationID, &i.ChannelType, &i.ChannelChatID, &i.ChatType, &i.LastMessageID, &i.LastThreadID, &i.Config, &i.CreatedAt, &i.PendingFresh, &i.ContextRevision, &i.RouteRevision, &i.RetiredAt, &i.HistoryStartMessageID, &i.HistoryEndMessageID, &i.HistoryBoundaryPending, ); err != nil { return nil, err } items = append(items, i) } if err := rows.Err(); err != nil { return nil, err } return items, nil } const listChannelInboundAuditByInstallation = `-- name: ListChannelInboundAuditByInstallation :many SELECT id, installation_id, channel_type, channel_chat_id, event_type, channel_event_id, channel_message_id, drop_reason, received_at FROM channel_inbound_audit WHERE installation_id = $1 ORDER BY received_at DESC LIMIT $2 OFFSET $3 ` type ListChannelInboundAuditByInstallationParams struct { InstallationID pgtype.UUID `json:"installation_id"` Limit int32 `json:"limit"` Offset int32 `json:"offset"` } func (q *Queries) ListChannelInboundAuditByInstallation(ctx context.Context, arg ListChannelInboundAuditByInstallationParams) ([]ChannelInboundAudit, error) { rows, err := q.db.Query(ctx, listChannelInboundAuditByInstallation, arg.InstallationID, arg.Limit, arg.Offset) if err != nil { return nil, err } defer rows.Close() items := []ChannelInboundAudit{} for rows.Next() { var i ChannelInboundAudit if err := rows.Scan( &i.ID, &i.InstallationID, &i.ChannelType, &i.ChannelChatID, &i.EventType, &i.ChannelEventID, &i.ChannelMessageID, &i.DropReason, &i.ReceivedAt, ); err != nil { return nil, err } items = append(items, i) } if err := rows.Err(); err != nil { return nil, err } return items, nil } const listChannelInstallationsByWorkspace = `-- name: ListChannelInstallationsByWorkspace :many SELECT id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at FROM channel_installation WHERE workspace_id = $1 AND channel_type = $2 ORDER BY created_at ASC ` type ListChannelInstallationsByWorkspaceParams struct { WorkspaceID pgtype.UUID `json:"workspace_id"` ChannelType string `json:"channel_type"` } // Scoped by channel_type so a per-channel management surface (e.g. the Lark // installation list) only ever sees its own platform's installations. func (q *Queries) ListChannelInstallationsByWorkspace(ctx context.Context, arg ListChannelInstallationsByWorkspaceParams) ([]ChannelInstallation, error) { rows, err := q.db.Query(ctx, listChannelInstallationsByWorkspace, arg.WorkspaceID, arg.ChannelType) if err != nil { return nil, err } defer rows.Close() items := []ChannelInstallation{} for rows.Next() { var i ChannelInstallation if err := rows.Scan( &i.ID, &i.WorkspaceID, &i.AgentID, &i.ChannelType, &i.Config, &i.Status, &i.WsLeaseToken, &i.WsLeaseExpiresAt, &i.InstallerUserID, &i.InstalledAt, &i.CreatedAt, &i.UpdatedAt, ); err != nil { return nil, err } items = append(items, i) } if err := rows.Err(); err != nil { return nil, err } return items, nil } const listChannelOutboundMessageIDsForBinding = `-- name: ListChannelOutboundMessageIDsForBinding :many SELECT outbound.channel_message_id FROM channel_outbound_message AS outbound WHERE outbound.binding_id = $1 ORDER BY outbound.created_at ASC ` func (q *Queries) ListChannelOutboundMessageIDsForBinding(ctx context.Context, bindingID pgtype.UUID) ([]string, error) { rows, err := q.db.Query(ctx, listChannelOutboundMessageIDsForBinding, bindingID) if err != nil { return nil, err } defer rows.Close() items := []string{} for rows.Next() { var channel_message_id string if err := rows.Scan(&channel_message_id); err != nil { return nil, err } items = append(items, channel_message_id) } if err := rows.Err(); err != nil { return nil, err } return items, nil } const listChannelOutboundMessagesByIDs = `-- name: ListChannelOutboundMessagesByIDs :many SELECT outbound.installation_id, outbound.channel_type, outbound.channel_message_id, outbound.binding_id, outbound.route_revision, outbound.task_id, outbound.outbound_kind, outbound.created_at FROM channel_outbound_message AS outbound WHERE outbound.installation_id = $1 AND outbound.channel_message_id = ANY($2::text[]) ` type ListChannelOutboundMessagesByIDsParams struct { InstallationID pgtype.UUID `json:"installation_id"` ChannelMessageIds []string `json:"channel_message_ids"` } func (q *Queries) ListChannelOutboundMessagesByIDs(ctx context.Context, arg ListChannelOutboundMessagesByIDsParams) ([]ChannelOutboundMessage, error) { rows, err := q.db.Query(ctx, listChannelOutboundMessagesByIDs, arg.InstallationID, arg.ChannelMessageIds) if err != nil { return nil, err } defer rows.Close() items := []ChannelOutboundMessage{} for rows.Next() { var i ChannelOutboundMessage if err := rows.Scan( &i.InstallationID, &i.ChannelType, &i.ChannelMessageID, &i.BindingID, &i.RouteRevision, &i.TaskID, &i.OutboundKind, &i.CreatedAt, ); err != nil { return nil, err } items = append(items, i) } if err := rows.Err(); err != nil { return nil, err } return items, nil } const lockChannelChatContextGenerationByRevision = `-- name: LockChannelChatContextGenerationByRevision :one SELECT chat_session_id, revision, history_start_message_id, history_end_message_id, history_boundary_pending, pending_fresh, initiator_user_id, created_at, last_message_id, last_thread_id, last_sender_id FROM channel_chat_context_generation WHERE chat_session_id = $1 AND revision = $2 FOR UPDATE ` type LockChannelChatContextGenerationByRevisionParams struct { ChatSessionID pgtype.UUID `json:"chat_session_id"` Revision int64 `json:"revision"` } func (q *Queries) LockChannelChatContextGenerationByRevision(ctx context.Context, arg LockChannelChatContextGenerationByRevisionParams) (ChannelChatContextGeneration, error) { row := q.db.QueryRow(ctx, lockChannelChatContextGenerationByRevision, arg.ChatSessionID, arg.Revision) var i ChannelChatContextGeneration err := row.Scan( &i.ChatSessionID, &i.Revision, &i.HistoryStartMessageID, &i.HistoryEndMessageID, &i.HistoryBoundaryPending, &i.PendingFresh, &i.InitiatorUserID, &i.CreatedAt, &i.LastMessageID, &i.LastThreadID, &i.LastSenderID, ) return i, err } const lockChannelChatSessionBindingForContext = `-- name: LockChannelChatSessionBindingForContext :one SELECT id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at, pending_fresh, context_revision, route_revision, retired_at, history_start_message_id, history_end_message_id, history_boundary_pending FROM channel_chat_session_binding WHERE chat_session_id = $1 FOR UPDATE ` // Context mutations acquire this row after chat_session and before any // channel_chat_context_generation row. Keeping the statements separate makes // the lock order explicit and identical for append, /clear, and task enqueue. func (q *Queries) LockChannelChatSessionBindingForContext(ctx context.Context, chatSessionID pgtype.UUID) (ChannelChatSessionBinding, error) { row := q.db.QueryRow(ctx, lockChannelChatSessionBindingForContext, chatSessionID) var i ChannelChatSessionBinding err := row.Scan( &i.ID, &i.ChatSessionID, &i.InstallationID, &i.ChannelType, &i.ChannelChatID, &i.ChatType, &i.LastMessageID, &i.LastThreadID, &i.Config, &i.CreatedAt, &i.PendingFresh, &i.ContextRevision, &i.RouteRevision, &i.RetiredAt, &i.HistoryStartMessageID, &i.HistoryEndMessageID, &i.HistoryBoundaryPending, ) return i, err } const lockChannelChatSessionPendingFresh = `-- name: LockChannelChatSessionPendingFresh :one SELECT pending_fresh FROM channel_chat_session_binding WHERE chat_session_id = $1 FOR UPDATE ` // EnqueueChatTask reads this under the same row lock and transaction that // creates the task. A concurrent `/clear` therefore lands either before this // task and is consumed by it, or after this task and remains for the next one. func (q *Queries) LockChannelChatSessionPendingFresh(ctx context.Context, chatSessionID pgtype.UUID) (bool, error) { row := q.db.QueryRow(ctx, lockChannelChatSessionPendingFresh, chatSessionID) var pending_fresh bool err := row.Scan(&pending_fresh) return pending_fresh, err } const lockChannelInstallationAppIDSlot = `-- name: LockChannelInstallationAppIDSlot :exec SELECT pg_advisory_xact_lock( hashtext($1::text), hashtext($2::text) ) ` type LockChannelInstallationAppIDSlotParams struct { ChannelType string `json:"channel_type"` AppID string `json:"app_id"` } // Serializes everything an install does to one (channel_type, config->>'app_id') // routing slot: read the current owner, decide, reclaim, upsert. Taken as the // first statement of the install transaction and released by COMMIT/ROLLBACK, // so the owner read below cannot go stale under a concurrent install or // reconnect — a plain read-then-write leaves a TOCTOU window in which two // callers both see "no live owner" and both go on to touch the slot. // // Two-key form: the first key namespaces by channel so a feishu app_id and a // wecom bot id that hash alike do not serialize against each other. hashtext // collisions inside one channel only cost extra serialization, never // correctness. pg_advisory_xact_lock (not pg_try_) so a second caller waits // its turn rather than failing. func (q *Queries) LockChannelInstallationAppIDSlot(ctx context.Context, arg LockChannelInstallationAppIDSlotParams) error { _, err := q.db.Exec(ctx, lockChannelInstallationAppIDSlot, arg.ChannelType, arg.AppID) return err } const lockCurrentChannelChatSessionBinding = `-- name: LockCurrentChannelChatSessionBinding :one SELECT id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at, pending_fresh, context_revision, route_revision, retired_at, history_start_message_id, history_end_message_id, history_boundary_pending FROM channel_chat_session_binding WHERE installation_id = $1 AND channel_chat_id = $2 AND retired_at IS NULL FOR UPDATE ` type LockCurrentChannelChatSessionBindingParams struct { InstallationID pgtype.UUID `json:"installation_id"` ChannelChatID string `json:"channel_chat_id"` } func (q *Queries) LockCurrentChannelChatSessionBinding(ctx context.Context, arg LockCurrentChannelChatSessionBindingParams) (ChannelChatSessionBinding, error) { row := q.db.QueryRow(ctx, lockCurrentChannelChatSessionBinding, arg.InstallationID, arg.ChannelChatID) var i ChannelChatSessionBinding err := row.Scan( &i.ID, &i.ChatSessionID, &i.InstallationID, &i.ChannelType, &i.ChannelChatID, &i.ChatType, &i.LastMessageID, &i.LastThreadID, &i.Config, &i.CreatedAt, &i.PendingFresh, &i.ContextRevision, &i.RouteRevision, &i.RetiredAt, &i.HistoryStartMessageID, &i.HistoryEndMessageID, &i.HistoryBoundaryPending, ) return i, err } const lockCurrentChannelChatSessionBindingBySession = `-- name: LockCurrentChannelChatSessionBindingBySession :one SELECT id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at, pending_fresh, context_revision, route_revision, retired_at, history_start_message_id, history_end_message_id, history_boundary_pending FROM channel_chat_session_binding WHERE chat_session_id = $1 AND retired_at IS NULL FOR UPDATE ` // Shared append fence. A message may have resolved a Chat immediately before // /new retired it; locking only chat_session would then let the stale append // land after the route switch. Requiring the binding to remain current makes // every adapter retry against the winning generation. func (q *Queries) LockCurrentChannelChatSessionBindingBySession(ctx context.Context, chatSessionID pgtype.UUID) (ChannelChatSessionBinding, error) { row := q.db.QueryRow(ctx, lockCurrentChannelChatSessionBindingBySession, chatSessionID) var i ChannelChatSessionBinding err := row.Scan( &i.ID, &i.ChatSessionID, &i.InstallationID, &i.ChannelType, &i.ChannelChatID, &i.ChatType, &i.LastMessageID, &i.LastThreadID, &i.Config, &i.CreatedAt, &i.PendingFresh, &i.ContextRevision, &i.RouteRevision, &i.RetiredAt, &i.HistoryStartMessageID, &i.HistoryEndMessageID, &i.HistoryBoundaryPending, ) return i, err } const markChannelChatSessionPendingFresh = `-- name: MarkChannelChatSessionPendingFresh :one UPDATE channel_chat_session_binding SET pending_fresh = TRUE WHERE chat_session_id = $1 AND retired_at IS NULL RETURNING pending_fresh ` // Persists a channel `/clear` intent until the next chat task is successfully // created. RETURNING makes a missing binding an error instead of silently // acknowledging a fresh start that was never stored. func (q *Queries) MarkChannelChatSessionPendingFresh(ctx context.Context, chatSessionID pgtype.UUID) (bool, error) { row := q.db.QueryRow(ctx, markChannelChatSessionPendingFresh, chatSessionID) var pending_fresh bool err := row.Scan(&pending_fresh) return pending_fresh, err } const markChannelInboundDedupProcessed = `-- name: MarkChannelInboundDedupProcessed :execrows UPDATE channel_inbound_message_dedup SET processed_at = now() WHERE installation_id = $1 AND message_id = $2 AND claim_token = $3 AND processed_at IS NULL ` type MarkChannelInboundDedupProcessedParams struct { InstallationID pgtype.UUID `json:"installation_id"` MessageID string `json:"message_id"` ClaimToken pgtype.UUID `json:"claim_token"` } // Locks a claim in as permanently processed after a durable outcome. // Invoked inside the chat_message tx (via qtx) on the ingest path so the // durable write and the Mark commit atomically. Token mismatch returns // zero rows (a reclaim happened); the caller rolls back its in-tx write. func (q *Queries) MarkChannelInboundDedupProcessed(ctx context.Context, arg MarkChannelInboundDedupProcessedParams) (int64, error) { result, err := q.db.Exec(ctx, markChannelInboundDedupProcessed, arg.InstallationID, arg.MessageID, arg.ClaimToken) if err != nil { return 0, err } return result.RowsAffected(), nil } const markChannelReplyDeliverySendUnknown = `-- name: MarkChannelReplyDeliverySendUnknown :execrows UPDATE channel_reply_delivery SET send_state = 'unknown', updated_at = now() WHERE turn_id = $1 AND send_state = 'in_flight' ` // The response was lost. Recorded without the owner check: this is the write // that must survive a caller whose lease expired while its own request hung, // because the alternative is a successor assuming nothing was ever sent. func (q *Queries) MarkChannelReplyDeliverySendUnknown(ctx context.Context, turnID pgtype.UUID) (int64, error) { result, err := q.db.Exec(ctx, markChannelReplyDeliverySendUnknown, turnID) if err != nil { return 0, err } return result.RowsAffected(), nil } const markChannelReplyDeliverySending = `-- name: MarkChannelReplyDeliverySending :execrows UPDATE channel_reply_delivery SET send_state = 'in_flight', updated_at = now() WHERE turn_id = $1 AND owner_token = $2 AND send_state NOT IN ('in_flight', 'unknown') ` type MarkChannelReplyDeliverySendingParams struct { TurnID pgtype.UUID `json:"turn_id"` OwnerToken pgtype.UUID `json:"owner_token"` } // Publish a send before making it, so any other process reads "a send is // outstanding" rather than "nothing has been sent". A resolved earlier send // does not block the next part of a multi-part answer; an unresolved one does, // because that is the case where nobody knows what is already in the chat. func (q *Queries) MarkChannelReplyDeliverySending(ctx context.Context, arg MarkChannelReplyDeliverySendingParams) (int64, error) { result, err := q.db.Exec(ctx, markChannelReplyDeliverySending, arg.TurnID, arg.OwnerToken) if err != nil { return 0, err } return result.RowsAffected(), nil } const nullChannelInboundAuditInstallationID = `-- name: NullChannelInboundAuditInstallationID :exec UPDATE channel_inbound_audit SET installation_id = NULL WHERE installation_id = $1 ` // Application-layer stand-in for the old ON DELETE SET NULL (MUL-3515 §4, // migration 124 keeps installation_id nullable for exactly this): before an // installation row is hard-deleted, detach its inbound-audit rows by NULLing // installation_id. The drop-audit history is preserved (channel_type, // chat/message ids, drop_reason stay) without a dangling reference to a // removed installation. func (q *Queries) NullChannelInboundAuditInstallationID(ctx context.Context, installationID pgtype.UUID) error { _, err := q.db.Exec(ctx, nullChannelInboundAuditInstallationID, installationID) return err } const purgeChannelInboundDedup = `-- name: PurgeChannelInboundDedup :exec DELETE FROM channel_inbound_message_dedup WHERE received_at < $1 ` // Vacuum job: remove dedup rows older than the supplied cutoff (e.g. 24h). func (q *Queries) PurgeChannelInboundDedup(ctx context.Context, receivedAt pgtype.Timestamptz) error { _, err := q.db.Exec(ctx, purgeChannelInboundDedup, receivedAt) return err } const purgeExpiredChannelBindingTokens = `-- name: PurgeExpiredChannelBindingTokens :exec DELETE FROM channel_binding_token WHERE expires_at < $1 ` func (q *Queries) PurgeExpiredChannelBindingTokens(ctx context.Context, expiresAt pgtype.Timestamptz) error { _, err := q.db.Exec(ctx, purgeExpiredChannelBindingTokens, expiresAt) return err } const reclaimDeadChannelInstallationByAppID = `-- name: ReclaimDeadChannelInstallationByAppID :one WITH dead AS ( DELETE FROM channel_installation ci WHERE ci.channel_type = $1 AND ci.config ->> 'app_id' = $2::text AND ( (ci.status = 'revoked' AND NOT (ci.workspace_id = $3 AND ci.agent_id = $4)) OR NOT EXISTS (SELECT 1 FROM workspace w WHERE w.id = ci.workspace_id) OR NOT EXISTS (SELECT 1 FROM agent a WHERE a.id = ci.agent_id) ) RETURNING ci.id ), cleared_dingtalk_group_presence AS ( DELETE FROM dingtalk_group_presence WHERE installation_id IN (SELECT id FROM dead) ), cleared_dingtalk_bot_identity AS ( DELETE FROM dingtalk_bot_identity WHERE installation_id IN (SELECT id FROM dead) ), cleared_dingtalk_group_routes AS ( DELETE FROM dingtalk_group_route WHERE installation_id IN (SELECT id FROM dead) ), cleared_task_deliveries AS ( DELETE FROM channel_task_delivery WHERE installation_id IN (SELECT id FROM dead) ), cleared_outbound_messages AS ( DELETE FROM channel_outbound_message WHERE installation_id IN (SELECT id FROM dead) ), cleared_reply_deliveries AS ( DELETE FROM channel_reply_delivery WHERE installation_id IN (SELECT id FROM dead) ), cleared_chat_sessions AS ( DELETE FROM channel_chat_session_binding WHERE installation_id IN (SELECT id FROM dead) RETURNING chat_session_id ), cleared_chat_contexts AS ( -- A revoked installation can still own preserved Chat history and an -- in-flight task snapshot. Remove only generations whose Chat was already -- cascade-deleted by an earlier orphan teardown. DELETE FROM channel_chat_context_generation AS generation WHERE generation.chat_session_id IN (SELECT chat_session_id FROM cleared_chat_sessions) AND NOT EXISTS ( SELECT 1 FROM chat_session AS session WHERE session.id = generation.chat_session_id ) ), cleared_outbound_cards AS ( -- channel_outbound_card_message is keyed by chat_session_id (no installation_id, -- no FK), so it is reached through the just-removed chat-session bindings. On an -- orphan reclaim the chat_session row itself is already cascade-gone, but its -- binding survived and still carries the id — the only reliable link back. DELETE FROM channel_outbound_card_message WHERE chat_session_id IN (SELECT chat_session_id FROM cleared_chat_sessions) ), cleared_binding_tokens AS ( DELETE FROM channel_binding_token WHERE installation_id IN (SELECT id FROM dead) ), cleared_user_bindings AS ( DELETE FROM channel_user_binding WHERE installation_id IN (SELECT id FROM dead) ), cleared_inbound_dedup AS ( DELETE FROM channel_inbound_message_dedup WHERE installation_id IN (SELECT id FROM dead) ), detached_audit AS ( -- Reclaim keeps the DETACH semantics: the workspace still exists, so a -- NULL-installation audit row stays meaningful for operator triage. The hard- -- delete paths (DeleteWorkspace / runtime teardown) purge audit outright. UPDATE channel_inbound_audit SET installation_id = NULL WHERE installation_id IN (SELECT id FROM dead) ) SELECT id FROM dead ` type ReclaimDeadChannelInstallationByAppIDParams struct { ChannelType string `json:"channel_type"` AppID string `json:"app_id"` WorkspaceID pgtype.UUID `json:"workspace_id"` AgentID pgtype.UUID `json:"agent_id"` } // Rebind cleanup gate. Frees the (channel_type, config->>'app_id') routing slot // so a valid new agent can (re)bind a bot whose previous owner is DEAD, and, in // the same statement, clears every application-owned dependent row of the removed // installation (channel_* has no FK/cascade, MUL-3515 §4). Returns the removed id // (pgx.ErrNoRows when nothing was dead — a no-op the caller treats as success). // // "Dead" is exactly one of: // 1. a REVOKED placeholder held by ANY agent OTHER than the caller's own // (workspace, agent) pair. Disconnect only flips status to 'revoked' — no // product path ever hard-deletes the row — so a revoked row would otherwise // pin the bot's app_id slot forever with no self-serve recovery, even across // workspaces (workspace A disconnects; workspace B, which proves control by // holding the same app credentials, rebinds). Revoke is the owner's explicit // "I'm done with this bot", so any revoked row is reclaimable — only the // caller's OWN revoked row is spared (reactivated in place; see below). // 2. an ORPHAN whose owning workspace OR agent row no longer exists — the // workspace was deleted, or the agent was hard-deleted on runtime teardown. // With no FK the installation outlives its owner and keeps occupying the // app_id slot: the "ghost binding" that made a bot un-rebindable (#4810). // // Deliberately NOT dead (the caller refuses these with an accurate conflict): // - the SAME agent's own revoked row (agent_id = @agent_id): the upsert // reactivates it in place, preserving its installation_id and every binding; // - a live ACTIVE owner whose agent still exists — INCLUDING an ARCHIVED agent: // archive is reversible, so its bot stays owned rather than being silently // stolen. Only a hard delete frees the slot. // // The guard lives in the DELETE predicate (not a prior SELECT) so under READ // COMMITTED the row is re-checked at execution (EvalPlanQual): a concurrent // same-agent reconnect that flips the revoked row back to 'active' first makes // the predicate re-check fail, this deletes nothing, and no dependents are // touched — closing the read-then-delete TOCTOU. Dependent cleanup keys off the // actually-deleted id (the `dead` CTE), so it runs ONLY for a row this statement // removed. The (channel_type, app_id) unique index guarantees at most one match. func (q *Queries) ReclaimDeadChannelInstallationByAppID(ctx context.Context, arg ReclaimDeadChannelInstallationByAppIDParams) (pgtype.UUID, error) { row := q.db.QueryRow(ctx, reclaimDeadChannelInstallationByAppID, arg.ChannelType, arg.AppID, arg.WorkspaceID, arg.AgentID, ) var id pgtype.UUID err := row.Scan(&id) return id, err } const recordChannelInboundDrop = `-- name: RecordChannelInboundDrop :exec INSERT INTO channel_inbound_audit ( installation_id, channel_type, channel_chat_id, event_type, channel_event_id, channel_message_id, drop_reason, id ) VALUES ( $4, $1, $5, $2, $6, $7, $3, COALESCE($8::uuid, gen_random_uuid()) ) ` type RecordChannelInboundDropParams struct { ChannelType string `json:"channel_type"` EventType string `json:"event_type"` DropReason string `json:"drop_reason"` InstallationID pgtype.UUID `json:"installation_id"` ChannelChatID pgtype.Text `json:"channel_chat_id"` ChannelEventID pgtype.Text `json:"channel_event_id"` ChannelMessageID pgtype.Text `json:"channel_message_id"` ID pgtype.UUID `json:"id"` } // ===================== // channel_inbound_audit // ===================== // The only write path for dropped events. Deliberately carries no body // column — only routing / identity / drop_reason / timestamp. func (q *Queries) RecordChannelInboundDrop(ctx context.Context, arg RecordChannelInboundDropParams) error { _, err := q.db.Exec(ctx, recordChannelInboundDrop, arg.ChannelType, arg.EventType, arg.DropReason, arg.InstallationID, arg.ChannelChatID, arg.ChannelEventID, arg.ChannelMessageID, arg.ID, ) return err } const recordChannelMediaPendingObject = `-- name: RecordChannelMediaPendingObject :one INSERT INTO channel_media_pending_object ( storage_key, workspace_id, chat_message_id, storage_url, installation_id ) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (storage_key) DO UPDATE SET created_at = now(), next_attempt_at = now(), chat_message_id = EXCLUDED.chat_message_id, storage_url = EXCLUDED.storage_url WHERE channel_media_pending_object.state = 'pending' AND channel_media_pending_object.workspace_id = EXCLUDED.workspace_id RETURNING storage_key ` type RecordChannelMediaPendingObjectParams struct { StorageKey string `json:"storage_key"` WorkspaceID pgtype.UUID `json:"workspace_id"` ChatMessageID pgtype.UUID `json:"chat_message_id"` StorageUrl string `json:"storage_url"` InstallationID pgtype.UUID `json:"installation_id"` } // ===================== // channel_media_pending_object (media intent ledger) // ===================== // Records upload intent BEFORE the PUT. A redelivered attempt refreshes the // settle window, but only while the row is still 'pending' — a key the // reconciler owns ('deleting') must never be resurrected — and only within // the SAME workspace: a cross-workspace key collision (impossible via the // derived key, but tenancy must never trust the key string) updates nothing // and returns no row, so the caller skips the upload entirely. func (q *Queries) RecordChannelMediaPendingObject(ctx context.Context, arg RecordChannelMediaPendingObjectParams) (string, error) { row := q.db.QueryRow(ctx, recordChannelMediaPendingObject, arg.StorageKey, arg.WorkspaceID, arg.ChatMessageID, arg.StorageUrl, arg.InstallationID, ) var storage_key string err := row.Scan(&storage_key) return storage_key, err } const recordChannelOutboundMessage = `-- name: RecordChannelOutboundMessage :exec INSERT INTO channel_outbound_message ( installation_id, channel_type, channel_message_id, binding_id, route_revision, task_id, outbound_kind ) VALUES ( $1, $2, $3, $4, $5, $6, $7 ) ON CONFLICT (installation_id, channel_message_id) DO NOTHING ` type RecordChannelOutboundMessageParams struct { OutboundInstallationID pgtype.UUID `json:"outbound_installation_id"` OutboundChannelType string `json:"outbound_channel_type"` OutboundMessageID string `json:"outbound_message_id"` OutboundBindingID pgtype.UUID `json:"outbound_binding_id"` OutboundRouteRevision int64 `json:"outbound_route_revision"` OutboundTaskID pgtype.UUID `json:"outbound_task_id"` OutboundKind string `json:"outbound_kind"` } // ===================== // channel_outbound_message // ===================== func (q *Queries) RecordChannelOutboundMessage(ctx context.Context, arg RecordChannelOutboundMessageParams) error { _, err := q.db.Exec(ctx, recordChannelOutboundMessage, arg.OutboundInstallationID, arg.OutboundChannelType, arg.OutboundMessageID, arg.OutboundBindingID, arg.OutboundRouteRevision, arg.OutboundTaskID, arg.OutboundKind, ) return err } const recordChannelReplyDeliveryChunk = `-- name: RecordChannelReplyDeliveryChunk :execrows UPDATE channel_reply_delivery SET send_state = 'known', message_id = CASE WHEN message_id = '' THEN $1::text ELSE message_id END, chunks_sent = GREATEST(chunks_sent, $2::int), updated_at = now() WHERE turn_id = $3 AND owner_token = $4 ` type RecordChannelReplyDeliveryChunkParams struct { MessageID string `json:"message_id"` ChunksSent int32 `json:"chunks_sent"` TurnID pgtype.UUID `json:"turn_id"` OwnerToken pgtype.UUID `json:"owner_token"` } // A part of the final answer landed. message_id is only adopted when the turn // has no editable message yet, so later parts never retarget the first one. func (q *Queries) RecordChannelReplyDeliveryChunk(ctx context.Context, arg RecordChannelReplyDeliveryChunkParams) (int64, error) { result, err := q.db.Exec(ctx, recordChannelReplyDeliveryChunk, arg.MessageID, arg.ChunksSent, arg.TurnID, arg.OwnerToken, ) if err != nil { return 0, err } return result.RowsAffected(), nil } const recordChannelReplyDeliveryPlaceholder = `-- name: RecordChannelReplyDeliveryPlaceholder :execrows UPDATE channel_reply_delivery SET send_state = 'known', message_id = $3, updated_at = now() WHERE turn_id = $1 AND owner_token = $2 AND send_state = 'in_flight' ` type RecordChannelReplyDeliveryPlaceholderParams struct { TurnID pgtype.UUID `json:"turn_id"` OwnerToken pgtype.UUID `json:"owner_token"` MessageID string `json:"message_id"` } // The placeholder landed. It gives the turn an editable message; it delivers // no part of the final answer, so chunks_sent stays where it is. func (q *Queries) RecordChannelReplyDeliveryPlaceholder(ctx context.Context, arg RecordChannelReplyDeliveryPlaceholderParams) (int64, error) { result, err := q.db.Exec(ctx, recordChannelReplyDeliveryPlaceholder, arg.TurnID, arg.OwnerToken, arg.MessageID) if err != nil { return 0, err } return result.RowsAffected(), nil } const releaseChannelInboundDedup = `-- name: ReleaseChannelInboundDedup :execrows DELETE FROM channel_inbound_message_dedup WHERE installation_id = $1 AND message_id = $2 AND claim_token = $3 AND processed_at IS NULL ` type ReleaseChannelInboundDedupParams struct { InstallationID pgtype.UUID `json:"installation_id"` MessageID string `json:"message_id"` ClaimToken pgtype.UUID `json:"claim_token"` } // Releases an in-flight claim when an infra error occurred before any // durable side effect, so a retry can re-acquire immediately. Fenced on // processed_at IS NULL and claim_token. func (q *Queries) ReleaseChannelInboundDedup(ctx context.Context, arg ReleaseChannelInboundDedupParams) (int64, error) { result, err := q.db.Exec(ctx, releaseChannelInboundDedup, arg.InstallationID, arg.MessageID, arg.ClaimToken) if err != nil { return 0, err } return result.RowsAffected(), nil } const releaseChannelMediaPendingObject = `-- name: ReleaseChannelMediaPendingObject :exec UPDATE channel_media_pending_object SET lease_token = NULL, lease_expires_at = NULL, next_attempt_at = now() + $1::interval, last_error = $2 WHERE storage_key = $3 AND workspace_id = $4 AND lease_token = $5 ` type ReleaseChannelMediaPendingObjectParams struct { Backoff pgtype.Interval `json:"backoff"` LastError pgtype.Text `json:"last_error"` StorageKey string `json:"storage_key"` WorkspaceID pgtype.UUID `json:"workspace_id"` LeaseToken pgtype.UUID `json:"lease_token"` } // Object-storage DELETE failed: keep the row in 'deleting' (bind must still // never attach it), release the lease, and back off the next attempt. // workspace_id is redundant with the storage_key PK but explicit per the // tenancy rule: every query constrains the workspace column, never trusting // the key string. func (q *Queries) ReleaseChannelMediaPendingObject(ctx context.Context, arg ReleaseChannelMediaPendingObjectParams) error { _, err := q.db.Exec(ctx, releaseChannelMediaPendingObject, arg.Backoff, arg.LastError, arg.StorageKey, arg.WorkspaceID, arg.LeaseToken, ) return err } const releaseChannelReplyDelivery = `-- name: ReleaseChannelReplyDelivery :execrows UPDATE channel_reply_delivery SET owner_token = NULL, owner_expires_at = NULL, updated_at = now() WHERE turn_id = $1 AND owner_token = $2 ` type ReleaseChannelReplyDeliveryParams struct { TurnID pgtype.UUID `json:"turn_id"` OwnerToken pgtype.UUID `json:"owner_token"` } // Hand the turn back so the next path does not wait out the lease. func (q *Queries) ReleaseChannelReplyDelivery(ctx context.Context, arg ReleaseChannelReplyDeliveryParams) (int64, error) { result, err := q.db.Exec(ctx, releaseChannelReplyDelivery, arg.TurnID, arg.OwnerToken) if err != nil { return 0, err } return result.RowsAffected(), nil } const releaseChannelWSLease = `-- name: ReleaseChannelWSLease :exec UPDATE channel_installation SET ws_lease_token = NULL, ws_lease_expires_at = NULL, updated_at = now() WHERE id = $1 AND ws_lease_token = $2 ` type ReleaseChannelWSLeaseParams struct { ID pgtype.UUID `json:"id"` CurrentToken pgtype.Text `json:"current_token"` } // Drops the lease iff we are still the holder. func (q *Queries) ReleaseChannelWSLease(ctx context.Context, arg ReleaseChannelWSLeaseParams) error { _, err := q.db.Exec(ctx, releaseChannelWSLease, arg.ID, arg.CurrentToken) return err } const renewChannelReplyDelivery = `-- name: RenewChannelReplyDelivery :execrows UPDATE channel_reply_delivery SET owner_expires_at = now() + make_interval(secs => $1::double precision), updated_at = now() WHERE turn_id = $2 AND owner_token = $3 AND phase <> 'settled' ` type RenewChannelReplyDeliveryParams struct { LeaseSeconds float64 `json:"lease_seconds"` TurnID pgtype.UUID `json:"turn_id"` OwnerToken pgtype.UUID `json:"owner_token"` } // Prove the turn is still ours and push the lease out. Terminal delivery holds // a turn across several scheduler rounds — edit pacing, a 429 backoff — so it // re-proves ownership before each Telegram call rather than trusting a lease // taken minutes earlier. No rows means another process took the turn over, and // this one must stop. func (q *Queries) RenewChannelReplyDelivery(ctx context.Context, arg RenewChannelReplyDeliveryParams) (int64, error) { result, err := q.db.Exec(ctx, renewChannelReplyDelivery, arg.LeaseSeconds, arg.TurnID, arg.OwnerToken) if err != nil { return 0, err } return result.RowsAffected(), nil } const resetChannelReplyDeliverySend = `-- name: ResetChannelReplyDeliverySend :execrows UPDATE channel_reply_delivery SET send_state = CASE WHEN message_id = '' THEN 'none' ELSE 'known' END, updated_at = now() WHERE turn_id = $1 AND owner_token = $2 AND send_state = 'in_flight' ` type ResetChannelReplyDeliverySendParams struct { TurnID pgtype.UUID `json:"turn_id"` OwnerToken pgtype.UUID `json:"owner_token"` } // The provider answered and refused: nothing is in the chat, so the turn may // be attempted again. func (q *Queries) ResetChannelReplyDeliverySend(ctx context.Context, arg ResetChannelReplyDeliverySendParams) (int64, error) { result, err := q.db.Exec(ctx, resetChannelReplyDeliverySend, arg.TurnID, arg.OwnerToken) if err != nil { return 0, err } return result.RowsAffected(), nil } const resolveChannelChatContextHistoryStart = `-- name: ResolveChannelChatContextHistoryStart :exec UPDATE channel_chat_context_generation SET history_start_message_id = $1, history_boundary_pending = FALSE WHERE chat_session_id = $2 AND revision = $3 AND history_boundary_pending ` type ResolveChannelChatContextHistoryStartParams struct { HistoryStartMessageID pgtype.Text `json:"history_start_message_id"` ChatSessionID pgtype.UUID `json:"chat_session_id"` Revision int64 `json:"revision"` } func (q *Queries) ResolveChannelChatContextHistoryStart(ctx context.Context, arg ResolveChannelChatContextHistoryStartParams) error { _, err := q.db.Exec(ctx, resolveChannelChatContextHistoryStart, arg.HistoryStartMessageID, arg.ChatSessionID, arg.Revision) return err } const retireChannelChatSessionBinding = `-- name: RetireChannelChatSessionBinding :one UPDATE channel_chat_session_binding SET retired_at = now(), history_end_message_id = $1 WHERE id = $2 AND retired_at IS NULL RETURNING id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at, pending_fresh, context_revision, route_revision, retired_at, history_start_message_id, history_end_message_id, history_boundary_pending ` type RetireChannelChatSessionBindingParams struct { HistoryEndMessageID pgtype.Text `json:"history_end_message_id"` ID pgtype.UUID `json:"id"` } func (q *Queries) RetireChannelChatSessionBinding(ctx context.Context, arg RetireChannelChatSessionBindingParams) (ChannelChatSessionBinding, error) { row := q.db.QueryRow(ctx, retireChannelChatSessionBinding, arg.HistoryEndMessageID, arg.ID) var i ChannelChatSessionBinding err := row.Scan( &i.ID, &i.ChatSessionID, &i.InstallationID, &i.ChannelType, &i.ChannelChatID, &i.ChatType, &i.LastMessageID, &i.LastThreadID, &i.Config, &i.CreatedAt, &i.PendingFresh, &i.ContextRevision, &i.RouteRevision, &i.RetiredAt, &i.HistoryStartMessageID, &i.HistoryEndMessageID, &i.HistoryBoundaryPending, ) return i, err } const setChannelChatContextInitiator = `-- name: SetChannelChatContextInitiator :one UPDATE channel_chat_context_generation SET initiator_user_id = $1 WHERE chat_session_id = $2 AND revision = $3 RETURNING initiator_user_id ` type SetChannelChatContextInitiatorParams struct { InitiatorUserID pgtype.UUID `json:"initiator_user_id"` ChatSessionID pgtype.UUID `json:"chat_session_id"` Revision int64 `json:"revision"` } // Snapshots the latest authenticated sender whose durable input belongs to a // generation. Crash recovery must use this identity rather than the sender of // a later generation that happened to re-arm the lost debounce timer. func (q *Queries) SetChannelChatContextInitiator(ctx context.Context, arg SetChannelChatContextInitiatorParams) (pgtype.UUID, error) { row := q.db.QueryRow(ctx, setChannelChatContextInitiator, arg.InitiatorUserID, arg.ChatSessionID, arg.Revision) var initiator_user_id pgtype.UUID err := row.Scan(&initiator_user_id) return initiator_user_id, err } const setChannelChatContextReplyTarget = `-- name: SetChannelChatContextReplyTarget :exec UPDATE channel_chat_context_generation SET last_message_id = $1, last_thread_id = $2, last_sender_id = $3 WHERE chat_session_id = $4 AND revision = $5 ` type SetChannelChatContextReplyTargetParams struct { LastMessageID pgtype.Text `json:"last_message_id"` LastThreadID pgtype.Text `json:"last_thread_id"` LastSenderID pgtype.Text `json:"last_sender_id"` ChatSessionID pgtype.UUID `json:"chat_session_id"` Revision int64 `json:"revision"` } // Snapshots the trigger this generation will be answered on: the message an // outbound reply targets, its thread, and the channel-native id of whoever // sent it. The sibling of SetChannelChatContextInitiator, and load-bearing for // the same reason — a debounced run flushes against ITS generation, which may // no longer be the session's newest, so reading the trigger from the session // would answer one member's question quoting and @-mentioning another's. // // All three move together in one statement: a sender or thread that described // a different message than the reply targets is precisely the cross-attribution // this exists to prevent. func (q *Queries) SetChannelChatContextReplyTarget(ctx context.Context, arg SetChannelChatContextReplyTargetParams) error { _, err := q.db.Exec(ctx, setChannelChatContextReplyTarget, arg.LastMessageID, arg.LastThreadID, arg.LastSenderID, arg.ChatSessionID, arg.Revision, ) return err } const setChannelInstallationConfig = `-- name: SetChannelInstallationConfig :exec UPDATE channel_installation SET config = $2, updated_at = now() WHERE id = $1 ` type SetChannelInstallationConfigParams struct { ID pgtype.UUID `json:"id"` Config []byte `json:"config"` } // Replaces the whole config blob for one installation. Used by the // operator backfills (e.g. setting a freshly-fetched bot_union_id) that // read-modify-write the JSON in Go and persist it back atomically by id. func (q *Queries) SetChannelInstallationConfig(ctx context.Context, arg SetChannelInstallationConfigParams) error { _, err := q.db.Exec(ctx, setChannelInstallationConfig, arg.ID, arg.Config) return err } const setChannelInstallationStatus = `-- name: SetChannelInstallationStatus :exec UPDATE channel_installation SET status = $2, updated_at = now() WHERE id = $1 ` type SetChannelInstallationStatusParams struct { ID pgtype.UUID `json:"id"` Status string `json:"status"` } func (q *Queries) SetChannelInstallationStatus(ctx context.Context, arg SetChannelInstallationStatusParams) error { _, err := q.db.Exec(ctx, setChannelInstallationStatus, arg.ID, arg.Status) return err } const settleChannelReplyDelivery = `-- name: SettleChannelReplyDelivery :execrows UPDATE channel_reply_delivery SET phase = 'settled', settled_reason = $3, owner_token = NULL, owner_expires_at = NULL, updated_at = now() WHERE turn_id = $1 AND owner_token = $2 AND phase <> 'settled' ` type SettleChannelReplyDeliveryParams struct { TurnID pgtype.UUID `json:"turn_id"` OwnerToken pgtype.UUID `json:"owner_token"` SettledReason string `json:"settled_reason"` } // Delivery is over. Nothing sends or edits for this turn afterwards. func (q *Queries) SettleChannelReplyDelivery(ctx context.Context, arg SettleChannelReplyDeliveryParams) (int64, error) { result, err := q.db.Exec(ctx, settleChannelReplyDelivery, arg.TurnID, arg.OwnerToken, arg.SettledReason) if err != nil { return 0, err } return result.RowsAffected(), nil } const tombstoneChannelMediaPendingObject = `-- name: TombstoneChannelMediaPendingObject :execrows UPDATE channel_media_pending_object SET state = 'tombstoned', lease_token = NULL, lease_expires_at = NULL, next_attempt_at = now() + $1::interval, -- The pass index lives in its own column: a failed re-delete writes -- last_error, so carrying the schedule position there would reset the -- walk on every failure and a flaky store could keep the row alive -- indefinitely. The delete that got here succeeded, so any previous -- failure text is stale. tombstone_pass = $2, last_error = NULL WHERE storage_key = $3 AND workspace_id = $4 AND lease_token = $5 ` type TombstoneChannelMediaPendingObjectParams struct { RedeleteDelay pgtype.Interval `json:"redelete_delay"` TombstonePass int32 `json:"tombstone_pass"` StorageKey string `json:"storage_key"` WorkspaceID pgtype.UUID `json:"workspace_id"` LeaseToken pgtype.UUID `json:"lease_token"` } // The object was deleted, but the row is KEPT as a tombstone: a PUT the client // abandoned before the delete may still materialize the object afterwards, and // no DELETE can be ordered against it. Each due tombstone re-runs the // reference check and, only if still unreferenced, triggers another idempotent // delete, so a late materialization is reclaimed by a later pass while an // object something durably reads is never removed; // only after the re-delete schedule is exhausted is the row dropped // (DeleteChannelMediaPendingObject). Lease-token guarded like every other // settle write; workspace_id explicit per the tenancy rule. func (q *Queries) TombstoneChannelMediaPendingObject(ctx context.Context, arg TombstoneChannelMediaPendingObjectParams) (int64, error) { result, err := q.db.Exec(ctx, tombstoneChannelMediaPendingObject, arg.RedeleteDelay, arg.TombstonePass, arg.StorageKey, arg.WorkspaceID, arg.LeaseToken, ) if err != nil { return 0, err } return result.RowsAffected(), nil } const updateChannelChatSessionBindingReplyTarget = `-- name: UpdateChannelChatSessionBindingReplyTarget :exec WITH current_route AS ( SELECT current_binding.id, current_binding.chat_session_id, current_binding.installation_id, current_binding.channel_type, current_binding.channel_chat_id, current_binding.chat_type, current_binding.last_message_id, current_binding.last_thread_id, current_binding.config, current_binding.created_at, current_binding.pending_fresh, current_binding.context_revision, current_binding.route_revision, current_binding.retired_at, current_binding.history_start_message_id, current_binding.history_end_message_id, current_binding.history_boundary_pending FROM channel_chat_session_binding AS current_binding WHERE current_binding.chat_session_id = $3 FOR UPDATE OF current_binding ), closed_previous AS ( UPDATE channel_chat_session_binding AS previous SET history_end_message_id = $1 FROM current_route AS current WHERE current.history_boundary_pending AND $1::text IS NOT NULL AND previous.installation_id = current.installation_id AND previous.channel_chat_id = current.channel_chat_id -- Multiple native Slack /new commands can rotate through empty -- generations before another public message supplies a platform cursor. -- That cursor closes every still-open retired generation. AND previous.route_revision < current.route_revision AND previous.retired_at IS NOT NULL AND previous.history_end_message_id IS NULL ) UPDATE channel_chat_session_binding AS binding SET last_message_id = $1, last_thread_id = $2, history_start_message_id = CASE WHEN binding.history_boundary_pending AND $1::text IS NOT NULL THEN $1 ELSE binding.history_start_message_id END, history_boundary_pending = CASE WHEN $1::text IS NOT NULL THEN FALSE ELSE binding.history_boundary_pending END FROM current_route WHERE binding.id = current_route.id ` type UpdateChannelChatSessionBindingReplyTargetParams struct { LastMessageID pgtype.Text `json:"last_message_id"` LastThreadID pgtype.Text `json:"last_thread_id"` ReplyChatSessionID pgtype.UUID `json:"reply_chat_session_id"` } // Advances the session's latest-trigger cursor, which drives the history // boundary bookkeeping below. NOT the outbound reply target: that is frozen // per context generation (SetChannelChatContextReplyTarget), because this row // only ever remembers the newest trigger and a debounced run can be enqueued // after a later generation has already moved it. func (q *Queries) UpdateChannelChatSessionBindingReplyTarget(ctx context.Context, arg UpdateChannelChatSessionBindingReplyTargetParams) error { _, err := q.db.Exec(ctx, updateChannelChatSessionBindingReplyTarget, arg.LastMessageID, arg.LastThreadID, arg.ReplyChatSessionID) return err } const updateChannelOutboundCardStatus = `-- name: UpdateChannelOutboundCardStatus :exec UPDATE channel_outbound_card_message SET status = $2, last_patched_at = now() WHERE id = $1 ` type UpdateChannelOutboundCardStatusParams struct { ID pgtype.UUID `json:"id"` Status string `json:"status"` } func (q *Queries) UpdateChannelOutboundCardStatus(ctx context.Context, arg UpdateChannelOutboundCardStatusParams) error { _, err := q.db.Exec(ctx, updateChannelOutboundCardStatus, arg.ID, arg.Status) return err } const upsertChannelInstallation = `-- name: UpsertChannelInstallation :one INSERT INTO channel_installation ( workspace_id, agent_id, channel_type, config, installer_user_id ) VALUES ( $1, $2, $3, $4, $5 ) ON CONFLICT (workspace_id, agent_id, channel_type) DO UPDATE SET channel_type = EXCLUDED.channel_type, config = EXCLUDED.config, installer_user_id = EXCLUDED.installer_user_id, status = 'active', installed_at = now(), updated_at = now() RETURNING id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at ` type UpsertChannelInstallationParams struct { WorkspaceID pgtype.UUID `json:"workspace_id"` AgentID pgtype.UUID `json:"agent_id"` ChannelType string `json:"channel_type"` Config []byte `json:"config"` InstallerUserID pgtype.UUID `json:"installer_user_id"` } // Platform-agnostic inbound channel queries (MUL-3515). These operate on // the channel_* tables created in migration 124. Each installation carries // a `channel_type` discriminator and a JSONB `config` blob for // platform-specific identifiers/credentials; the cross-platform columns // stay flat. The Go layer owns building/parsing config — these queries // treat it as opaque JSON except for the routing index on config->>'app_id'. // // No foreign keys exist on these tables (MUL-3515 §4): the integrity the // old composite FKs enforced (binding workspace matches installation; // binding dies with membership / chat_session) is maintained in the // application layer via the membership check in the inbound identity step // and the *DeleteChannel*BindingsBy* cleanup queries below. // ===================== // channel_installation // ===================== // Install / re-install path. `config` is the opaque per-channel JSONB the // Go layer assembles (for feishu: app_id, app_secret_encrypted, tenant_key, // bot_open_id, bot_union_id, region). Re-installing the same agent on the // same channel_type replaces the whole config and forces status back to // 'active'. The conflict key is (workspace_id, agent_id, channel_type) so an // agent may hold one installation per channel_type (feishu + slack + ...) // without one install clobbering another. The WS lease is intentionally NOT // reset here — the inbound hub owns lease lifecycle. func (q *Queries) UpsertChannelInstallation(ctx context.Context, arg UpsertChannelInstallationParams) (ChannelInstallation, error) { row := q.db.QueryRow(ctx, upsertChannelInstallation, arg.WorkspaceID, arg.AgentID, arg.ChannelType, arg.Config, arg.InstallerUserID, ) var i ChannelInstallation err := row.Scan( &i.ID, &i.WorkspaceID, &i.AgentID, &i.ChannelType, &i.Config, &i.Status, &i.WsLeaseToken, &i.WsLeaseExpiresAt, &i.InstallerUserID, &i.InstalledAt, &i.CreatedAt, &i.UpdatedAt, ) return i, err } const upsertChannelInstallationByAppID = `-- name: UpsertChannelInstallationByAppID :one INSERT INTO channel_installation ( workspace_id, agent_id, channel_type, config, installer_user_id ) VALUES ( $1, $2, $3, $4, $5 ) ON CONFLICT (channel_type, (config ->> 'app_id')) DO UPDATE SET agent_id = EXCLUDED.agent_id, config = EXCLUDED.config, installer_user_id = EXCLUDED.installer_user_id, status = 'active', installed_at = now(), updated_at = now() WHERE channel_installation.workspace_id = EXCLUDED.workspace_id RETURNING id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at ` type UpsertChannelInstallationByAppIDParams struct { WorkspaceID pgtype.UUID `json:"workspace_id"` AgentID pgtype.UUID `json:"agent_id"` ChannelType string `json:"channel_type"` Config []byte `json:"config"` InstallerUserID pgtype.UUID `json:"installer_user_id"` } // Team-keyed install / re-install for channels whose natural identity is the // platform workspace, not the (agent) pairing. Slack: one Slack workspace // (team_id, stored as config->>'app_id') maps to exactly one installation, so // re-connecting it — even to represent a DIFFERENT agent in the SAME Multica // workspace — UPDATES the existing row (moving agent_id) instead of colliding // with the (channel_type, app_id) unique index. Contrast UpsertChannelInstallation, // whose conflict key is (workspace_id, agent_id, channel_type): right for Feishu // (one app per agent), wrong for Slack. // // The `WHERE channel_installation.workspace_id = EXCLUDED.workspace_id` fences // the conflict update to the SAME Multica workspace: a team already owned by a // DIFFERENT workspace updates no row and RETURNING is empty (pgx.ErrNoRows), // which the caller maps to ErrTeamOwnedByAnotherWorkspace. This is the ATOMIC // cross-workspace guard — a plain SELECT before the upsert cannot stop two // workspaces racing to OAuth the same team (both read no rows, then one inserts // and the other's conflict-update would silently steal it). A re-connect that // would move the team to an agent already holding a different Slack install in // the same workspace still trips the (workspace_id, agent_id, channel_type) // unique constraint — a genuine conflict the OAuth callback turns into a redirect. func (q *Queries) UpsertChannelInstallationByAppID(ctx context.Context, arg UpsertChannelInstallationByAppIDParams) (ChannelInstallation, error) { row := q.db.QueryRow(ctx, upsertChannelInstallationByAppID, arg.WorkspaceID, arg.AgentID, arg.ChannelType, arg.Config, arg.InstallerUserID, ) var i ChannelInstallation err := row.Scan( &i.ID, &i.WorkspaceID, &i.AgentID, &i.ChannelType, &i.Config, &i.Status, &i.WsLeaseToken, &i.WsLeaseExpiresAt, &i.InstallerUserID, &i.InstalledAt, &i.CreatedAt, &i.UpdatedAt, ) return i, err }