Files
53c4bbe5cd MUL-8395 fix(lark): distinguish bound bots from verified message reception (#9709)
* fix(lark): verify message reception separately from binding

* fix(lark): preserve verification evidence during rollout

* fix(lark): remove stray space in Chinese verification hint

Co-authored-by: multica-agent <[email protected]>

---------

Co-authored-by: Multica Eve <[email protected]>
Co-authored-by: multica-agent <[email protected]>
2026-10-31 02:05:37 +08:00

3574 lines
140 KiB
Go

// 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
WITH due AS MATERIALIZED (
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
)
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 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.
// MATERIALIZED pins the bounded candidate set so a nested-loop rescan cannot
// move the LIMIT window to another row after an update.
//
// 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 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 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 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. Prefer field-level
// patches for metadata that can change concurrently with receipt evidence.
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
}