// 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 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 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 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 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 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, $6 ) 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. 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 INSERT INTO channel_chat_session_binding ( chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, config ) VALUES ( $1, $2, $3, $4, $5, $6 ) RETURNING id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at ` 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"` } // ===================== // 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) (ChannelChatSessionBinding, error) { row := q.db.QueryRow(ctx, createChannelChatSessionBinding, arg.ChatSessionID, arg.InstallationID, arg.ChannelType, arg.ChannelChatID, arg.ChatType, arg.Config, ) 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, ) 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 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 deleteChannelChatSessionBindingBySession = `-- name: DeleteChannelChatSessionBindingBySession :exec DELETE FROM channel_chat_session_binding WHERE chat_session_id = $1 ` // Application-layer integrity (replaces the old chat_session-FK ON DELETE // CASCADE): drop the binding when its chat_session is deleted. func (q *Queries) DeleteChannelChatSessionBindingBySession(ctx context.Context, chatSessionID pgtype.UUID) error { _, err := q.db.Exec(ctx, deleteChannelChatSessionBindingBySession, chatSessionID) 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 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 FROM channel_chat_session_binding WHERE installation_id = $1 AND channel_chat_id = $2 ` 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, ) 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 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, ) 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 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 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 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 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 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 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 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 ) VALUES ( $4, $1, $5, $2, $6, $7, $3 ) ` 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"` } // ===================== // 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, ) return err } 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 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 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 updateChannelChatSessionBindingReplyTarget = `-- name: UpdateChannelChatSessionBindingReplyTarget :exec UPDATE channel_chat_session_binding SET last_message_id = $2, last_thread_id = $3 WHERE chat_session_id = $1 ` type UpdateChannelChatSessionBindingReplyTargetParams struct { ChatSessionID pgtype.UUID `json:"chat_session_id"` LastMessageID pgtype.Text `json:"last_message_id"` LastThreadID pgtype.Text `json:"last_thread_id"` } // Records the most recent inbound trigger message + thread so the decoupled // outbound patcher can thread its reply back into the originating topic. func (q *Queries) UpdateChannelChatSessionBindingReplyTarget(ctx context.Context, arg UpdateChannelChatSessionBindingReplyTargetParams) error { _, err := q.db.Exec(ctx, updateChannelChatSessionBindingReplyTarget, arg.ChatSessionID, arg.LastMessageID, arg.LastThreadID) 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 }