klinsmaya/multica
1
0
Fork 0
mirror of https://gh-proxy.org/https://github.com/klinsmaya/multica.git synced 2026-11-08 01:16:45 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
admin-workspace-create2
multica/server/pkg/db/generated/channel.sql.go
T
Basuotian 3647ec9a9c MUL-7687 fix(wecom): push an issue notification through the acting agent's bot (#8803) 
In a multi-bot organization, every issue notification a member received arrived
through whichever WeCom bot they had bound most recently, because
FindChannelBindingForMember keys on the recipient alone. Three agents' comments
all looked like they came from one of them, and the apparent sender drifted
whenever the member re-bound an account.

An inbox item already carries the actor that raised it. When that actor is an
agent, the WeCom push now resolves the bot as the binding the recipient holds
on the acting agent's active installation
(FindChannelBindingForMemberOnAgentInstallation), keeping the recipient-wide
lookup as the fallback — for an agent this member never linked, and for
member-authored notifications. Delivery is never narrowed, which is the
guarantee #7136 named.

When the preferred bot provably put nothing on the wire, the recipient-wide
binding is tried too. Proving that needed a four-state relay verdict
(relayNotWired / relayAccepted / relayNoHolder / relayUncertain) in place of a
bool, plus a non-fencing DedupeStore.ClaimState read so the publisher can
observe a delivery claim without fencing a worker that is mid-delivery. An
unclaimed frame is fenced under a token no delivery worker ever presents, so a
late or replaying dispatcher cannot add a second card after the fallback ran.
An uncertain send is deliberately not retried.

The acceptance wait runs on the request's own goroutine, so it is paid only on
the agent-attributed attempt — the one with a second route behind it — and is
bounded by the caller's context rather than by the window alone.

Reviewed-by: seacen
Reviewed-by: Eve (MUL-7687)
2026-10-12 11:32:52 +08:00

3616 lines
142 KiB
Go
Raw Permalink Blame History

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