Files
multica/server/pkg/db/generated/channel.sql.go
091d3ce5de MUL-7209 fix(lark): reply natively and mention the asker in ordinary groups (#8238)
* fix(lark): reply natively and mention the asker in ordinary groups

In an ordinary Feishu group the agent's answer arrived as a standalone
message with a plain-text "@name" the model had written itself, so it
neither attached to the question nor notified the member who asked
(#8234). Two independent gaps:

Native reply. The trigger message id was already there — the engine
records it on every inbound turn and the delivery row freezes it per
task — but threadReplyTarget required a thread id alongside it, so an
ordinary group (which has none) discarded a perfectly usable target and
fell through to the chat-level send. Now the thread id only decides
reply_in_thread; the message id alone is enough to reply. p2p is left on
the chat-level send: a 1:1 has no competing questions for a quote to
disambiguate.

Native mention. The sender's open_id is parsed on inbound but never
persisted, so the outbound patcher only had a Multica UUID. It now
resolves the open_id back from the task's own initiator_user_id through
a new installation-scoped channel_user_binding lookup, and renders a
real mention — <at user_id=...> for text, <at id=...> for the schema-2.0
card, since the two wire shapes spell it differently. Reading identity
from the task (not from the chat's most recent sender) is what keeps two
members' concurrent runs from cross-mentioning each other.

Identity is never inferred from the "@name" in the agent's prose:
duplicate and guessed names would ping the wrong colleague. With no
open_id — unbound member, missing row, failed lookup — the answer goes
out unmentioned rather than mentioning someone else.

sendWithThreadFallback is renamed sendWithReplyFallback; it now guards
ordinary replies too, and its classified error set (recalled trigger,
invisible message) already covered them.

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

* fix(lark): freeze the trigger's sender per task instead of re-deriving it

The mention added in the previous commit resolved the open_id at send time
by looking the task's initiator_user_id back up in channel_user_binding.
That is unsound. channel_user_binding is unique on (installation_id,
channel_user_id), NOT on (installation_id, multica_user_id), and the redeem
path plain-INSERTs when a member binds a second Feishu account on the same
installation — the ON CONFLICT only fires for a repeat of the same
channel_user_id. With two rows for one member the member-keyed lookup has no
ordering and no reverse uniqueness to lean on, so it could return either: a
reply to account B carrying account A's mention. That is reachable from the
normal binding flow, not a corrupt-database state, and it defeats the
acceptance criterion the mention exists for.

The sender is per-trigger data, exactly like the message id it accompanies,
so it now travels the same path: channel_chat_session_binding.last_sender_id
is written in the same statement as last_message_id / last_thread_id on every
inbound turn, and CreateChannelTaskDeliveryFromSession freezes it onto the
task's delivery row. The Patcher reads that snapshot and the Multica member
stops being an input to the mention entirely, so the sibling-identity case is
now unrepresentable rather than merely unlikely. Writing all three in one
statement is what guarantees the sender always describes the message the
reply targets.

GetChannelUserBindingByMember and its store wrapper are removed with the
lookup they existed for.

Only the Lark adapter populates SenderChannelID today; other channels leave
it NULL, which reads as "no identity" and sends unmentioned — the same
degradation as pre-migration rows, which are not backfilled.

Tests: the sender rides along in the reply-target write and is NULL when
absent; a DB-backed test proves a delivery row keeps its own sender after a
second member messages the same group; and the Patcher mentions the account
that sent the trigger even when the task's initiator holds another open_id.

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

* fix(lark): snapshot the reply trigger per context generation, not per session

The previous commit froze the trigger onto the task's delivery row, but it
sampled that trigger from channel_chat_session_binding — one row per
chat_session, holding only the LATEST trigger. The debouncer keys its timers
on (chat_session, context revision), so a run batched in revision 1 is
enqueued after a /clear has opened revision 2 and revision 2's own message has
committed. The delivery row then froze the newer generation's message and
sender: A's answer quoting and @-mentioning B. It needs no unusual
interleaving — only B's append committing inside A's 3s debounce window.

channel_chat_context_generation is keyed (chat_session_id, revision) and
already snapshots initiator_user_id against precisely this hazard;
SetChannelChatContextInitiator's comment spells it out for crash recovery. The
trigger belongs beside it, so the message/thread/sender triple moves there and
CreateChannelTaskDeliveryFromSession takes the context revision and reads that
generation. The snapshot is written under the same `cmd == nil` guard as the
initiator: a message excluded from agent input is not the question being
answered, so it must not become the message the answer quotes.

The binding keeps last_message_id / last_thread_id, which remain a genuine
per-session cursor driving the history-boundary bookkeeping, and loses the
last_sender_id the previous commit added — nothing reads it now. Migration 457
is rewritten rather than superseded, since it has not shipped; it seeds each
session's CURRENT generation from the binding cursor so live sessions keep
replying correctly across the deploy. Older generations are deliberately not
seeded: for those the cursor has already moved on and would be exactly the
wrong value.

This also fixes the same latent mis-attribution for the reply target on every
other channel — Telegram's quote-reply and Slack's thread had it too; only
Lark's ordinary-group reply made it visible.

Tests: two generations are appended with no delivery created in between, then
enqueued, and each delivery keeps its own trigger — the test first asserts the
session cursor holds the OTHER generation's message, so the wrong answer is
demonstrably reachable. Immutability after creation is kept as a separate case.

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

* fix(lark): drop the migration's trigger seed, leaving pre-migration rows NULL

The seed copied channel_chat_session_binding's cursor into each session's
current generation. It cannot be right: AppendUserMessage advances that cursor
for channel commands too, while the generation trigger is deliberately written
only when cmd == nil. A session whose last pre-deploy message was /issue would
have that command seeded as a generation's reply target — the exact value the
guard exists to keep out.

It also buys almost nothing. The trigger is recorded during AppendUserMessage,
which commits well before the debounced flush that enqueues the task and
creates its delivery, so the first inbound turn after deploy already supplies
a correct value. The only rows a seed could serve are generations enqueued
with no post-deploy append — the recovered-run path — and for exactly those
the cursor has already advanced past the generation, so seeding hands them a
confidently wrong message and sender instead of no answer. The benefit set and
the wrong-value set are nearly the same set.

Leaving them NULL degrades to a chat-level send with no mention, which for
those runs is strictly better than the pre-change behaviour of quoting the
session's newest trigger, i.e. some other member's message. This also matches
451_agent_task_comment_thread's convention: pre-migration rows drain without
rewriting historical data.

Migration 457 is now purely additive DDL.

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

* fix(lark): keep the topic route on the binding, and never leak a topic reply

Moving the whole message/thread/sender triple to the context generation
conflated two different things. The TRIGGER — which message an answer quotes,
which account it mentions — is per-generation. The ROUTE — which chat or topic
the answer goes to — is per-binding: a thread-isolated session has one binding
per topic, so its thread id names that topic for every generation of it.

Sourcing the route from the generation left pre-migration generations with no
thread. For Slack and Telegram that silently relocated a recovered run's answer
from its thread to the parent channel; for Lark, outboundChatID resolves the
composite "chat:thread" key to the parent group, so a topic conversation would
surface in front of the whole group. That is a visibility change, not the safe
degradation the previous commit claimed.

So channel_thread_id goes back to the binding — where it was before this PR —
and the generation carries only last_message_id and last_sender_id. Slack and
Telegram route correctly again with no adapter change, since both can place a
message in a thread from the thread id alone.

Lark cannot: its only route into a topic is replying to a message inside it.
With a topic route and no trigger message it now declines to send at all
rather than falling through to the parent group. Silence is recoverable — the
member asks again, and the answer is still in Multica — while a leak is not.
This is deliberately narrower than sendWithReplyFallback's chat-level retry,
which fires only when Lark reports the topic itself cannot receive the reply;
there the topic is unusable and delivering beats losing the reply. An ordinary
group keeps replying without a quote or mention, because a chat-level send
lands in the very chat the session belongs to.

Reachable only for generations predating migration 457 that are recovered
after deploy without a new inbound turn.

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

* fix(lark): keep the reply thread with the trigger, and renumber to 460

The previous commit called the thread "route" and read it from the binding.
That only holds for thread-ISOLATED sessions, where there is one binding per
thread. Slack DMs are the counterexample the repo already encodes:
slackSessionRouting keeps ONE binding for the whole DM channel while replying
into whichever thread the member used. Reading the thread from the binding
there hands a debounced revision-1 run the thread of a later revision — the
same cross-generation hazard this snapshot exists to close for the message and
sender, reintroduced through the one field I moved out.

That is a live regression on ordinary DM traffic, not a legacy-only edge: A
asks inside thread T1, a /clear within the 3s window opens revision 2 at top
level, and revision 1's answer lands wherever revision 2 spoke.

So last_thread_id goes back onto the generation with the message and sender.
The three are one trigger and travel together. The binding keeps its own
cursor for the history-boundary bookkeeping, and route stays what it always
was: chat, type, config, revision.

Lark's fail-closed guard now identifies a topic-isolated binding by its
config, which is what actually marks one — larkSessionRouting writes a config
only for topic sessions — rather than by a thread id that is now trigger data
and legitimately absent for pre-migration generations.

Migration renumbered 457 -> 460: main has since taken 457, 458 and 459.

The route-survives-missing-trigger test is replaced rather than repaired. It
ran on a p2p, non-composite fixture with a hand-inserted thread, so it proved
nothing about isolated bindings and would have locked in the wrong
generalization. Its replacement is the DM shape: T1 in revision 1, top level
in revision 2, both appended before either delivery is created, each keeping
its own thread, with the binding cursor asserted to hold the other one.

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

* fix(lark): recover the thread for isolated bindings when a generation has none

Putting the thread back on the generation fixed Slack DMs but reopened the
legacy hole on the channels that route by thread. A pre-460 generation carries
no thread, so a recovered run in a Slack channel thread or a Telegram forum
topic answered in the parent conversation: both adapters read
delivery.channel_thread_id directly and neither recovers a thread from the
composite key. Lark was covered by its config guard; these two were not.

CreateChannelTaskDeliveryFromSession now expresses both semantics instead of
one. When the generation recorded a trigger its thread wins, INCLUDING when
that thread is legitimately empty — which is what distinguishes a Slack DM
answered at top level from one answered inside a thread, the distinction the
previous commit restored. Only when there is no trigger at all does it fall
back, and only for a thread-isolated binding, where the thread is a stable
property of the session rather than a moving cursor.

Isolation is detected from the marker every adapter already writes: a binding
whose key is the composite "chat:thread" stores the real chat id in its config,
so a config chat id that differs from channel_chat_id IS the isolation flag
(larkSessionRouting, slackSessionRouting, telegramSessionRouting). A DM or
plain chat stores a config id equal to its key and recovers nothing, because
its cursor only ever names whoever spoke last.

Done in the shared query rather than in three adapters: the delivery row is
where route and trigger are frozen, so fixing it there covers Slack, Telegram
and any future thread-routed channel at once. Slack's historyTarget already
recovers a thread root from the composite key for reads; this is the same idea
for the write, minus the key parsing.

Lark still declines to send in this case regardless — a recovered thread is not
enough when the only way into a topic is replying to a message inside it.

Also corrects the migration's closing note, which still described NULL as
degrading uniformly to a chat-level send.

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

---------

Co-authored-by: J <[email protected]>
Co-authored-by: multica-agent <[email protected]>
2026-09-10 16:45:55 +08:00

2990 lines
117 KiB
Go

// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.31.1
// source: channel.sql
package db
import (
"context"
"github.com/jackc/pgx/v5/pgtype"
)
const 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 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)
), 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
)
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_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 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 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 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 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_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 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 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 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 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
}