mirror of
https://gh-proxy.org/https://github.com/klinsmaya/multica.git
synced 2026-11-09 01:46:46 +08:00
* 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]>
2990 lines
117 KiB
Go
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
|
|
}
|