mirror of
https://github.com/multica-ai/multica.git
synced 2026-08-08 20:00:02 +02:00
* fix: ingest feishu media as chat attachments * fix: ingest feishu post embedded media * fix(lark): make inbound media retries safe * fix lark media resource limit * fix(lark): move inbound media off ack path * fix(channel): make inbound media runs durable * fix(channel): close enqueue-vs-append race on media deferral EnqueueChatTask read the session-wide media deadline in one statement and sealed the input batch in a later one. Under READ COMMITTED a media message committing between the two got sealed into a task the deadline read had already decided was 'queued', so the daemon could claim it before its attachment bound — the agent received the bare placeholder, and the later media-ready promotion was a no-op against a non-deferred task. After the seal, re-derive the deferral from the sealed batch itself in the same transaction (DeferChatTaskForSealedPendingMedia): if any sealed message still carries an unexpired media marker, flip the task to deferred with fire_at aligned to the latest marker. The existing post-commit promote fence already covers the opposite direction (marker cleared mid-transaction). Adds a deterministic regression test that injects the media append between the deadline read and the seal via a wrapped pgx.Tx. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): keep committed chat task out of enqueue error path The post-commit media-ready fence returned its error from EnqueueChatTask even though the deferred task was already durably committed. The router flush treats any enqueue error as "no task exists": it clears the typing indicator and logs an enqueue failure while the run still happens at its fire_at deadline. Log the fence failure instead — the claim-path deferred promoter re-queues the task regardless. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): cap global media resolution concurrency Media jobs were serialized per session but unbounded across sessions: a burst could open arbitrarily many concurrent 45s Lark downloads, and each unknown-length upload may buffer up to the 100 MiB resource cap in memory. Gate resolveAndBindMedia behind a global slot semaphore (default 8, RouterConfig.MediaConcurrency). Per-session ordering is unchanged; on shutdown a job cancelled while waiting for a slot proceeds straight to the bounded DB finalize so marker clearing stays prompt. Also document that the per-message media budget spans queue/slot waits (it must match the persisted fire_at) and why timed-out uploads cannot leak unbounded orphans. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(chat): keep channel-sealed user messages on task cancel Sealing the channel input batch stamps task_id onto channel user messages, which exposed them to the cancel draft-restore path: an empty-transcript cancel would DeleteUserChatMessageByTask the sealed Feishu/Slack messages and detach their attachments. Those messages are the durable record of what the platform sender wrote — the sender has no Multica composer to restore a draft into. Gate the restore-delete on ChatSessionHasChannelBinding in both the synchronous finalize and the deferred finalize (the latter covers markers left by an older replica during a rolling deploy); a bound session now settles as "Stopped." instead. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): skip the media pipeline for messages without media Every inbound message on a Media-enabled platform persisted a 45s media deadline and queued a resolution job, so a plain text message could wait behind the global media semaphore (its task deferred while other sessions download 100 MiB videos) and a crash between append and clear delayed a pure-text run to the full 45s fallback. Add MediaResolver.HasMedia — a pure in-memory probe the Router calls on the ACK path — and only persist the deadline / enqueue the job when the message actually references platform media. The Feishu resolver decodes the already-received payload and reports standalone image or video keys and post-embedded img/media spans. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(chat): gate cancel restore on immutable channel provenance The previous guard keyed the cancel restore-delete off ChatSessionHasChannelBinding, but a binding only proves routing exists right now: archiving a session and rebinding an installation both delete the binding while preserving chat history, so a still-cancellable sealed task could again restore-delete the original inbound messages. Persist provenance on the message instead: migration 203 adds chat_message.channel_ingested, stamped inside the channel append transaction and never mutated, and both cancel finalize paths now gate on TaskHasChannelIngestedMessages over the task's sealed batch. The binding-existence query is removed. Regression tests cover ingest -> archive/unbind -> cancel for a queued and a started task. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): reclaim media uploads that never gain an attachment row Deadline expiry dropped already-resolved refs and a BindMedia failure was log-only, leaving uploaded objects with no attachment row and no reclaim path — the dedup mark commits with the message before media runs, so a redelivery is dropped as a duplicate and never re-resolves (and thus never overwrites) those keys, and workspace/session deletion only enumerates the attachment table. Add MediaResolver.DiscardMedia — a best-effort delete by StorageKey — and call it from both failure paths in resolveAndBindMedia. The Feishu resolver forwards to the storage backend's Delete. Tests cover a partial upload discarded at the deadline, discard on bind failure, and key-level deletion in the resolver. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs(server): refresh comments stale after detached media ingestion Channel tasks now seal a self-owned input batch, media ingestion is no longer out of scope for the flattener, and MediaRefs are filled by the detached resolver after append rather than by feishuChannel pre-engine. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(chat): stop keying channel empty-completion silence off chat_input_task_id Sealing gave channel tasks a self input-owner, which broke writeChatCompletionOutcome's discriminator: it treated any owned task as direct, so an empty channel completion wrote the no_response fallback row and the outbound patcher — which forwards any non-empty chat:done content verbatim — pushed the English fallback body to Feishu/Slack, violating the MUL-4351 contract. Silence is now decided by the immutable channel_ingested provenance of the task's input batch, looked up by the batch OWNER id (chat_input_task_id): auto-retry clones inherit the owner while their sealed messages stay tagged with the parent's id, so keying off the task's own id would misread a channel retry as direct. The cancel-path provenance gates switch to the same owner key via chatInputOwnerID. chat_input_task_id is back to meaning only "input batch owner". Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore(migrations): renumber to 203/204 after upstream took 202 Upstream main merged 202_runtime_profile_add_qwen while this branch held 202/203, tripping TestMigrationNumericPrefixesStayUniqueAfterLegacySet on the CI merge tree. channel_media_pending becomes 203 and channel_ingested becomes 204; no content changes. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channels): gate outbound delivery on channel provenance, not owner Merging main brought #5645 (keep direct chat replies in Multica), whose outbound gate assumed channel tasks leave chat_input_task_id NULL. Sealed channel tasks own an input batch too, so on the merge tree every channel reply and failure notice was classified as direct and silently dropped — agents stopped replying in Feishu/Slack. Both outbound gates now call engine.TaskInputIsChannelIngested: a NULL owner keeps #5645's deliver-by-default for pre-sealing tasks, an owned batch delivers only when it carries the immutable channel_ingested stamp (keyed by the owner id, so auto-retry clones inherit the verdict). Direct replies stay in Multica; sealed channel replies reach the platform. Tests cover both directions on both platforms. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): discard media orphans on a fresh context, after finalize DiscardMedia shared finalizeCtx with BindMedia, so a bind that failed because the finalize deadline expired handed the storage deletes an already-dead context — the compensation silently no-opped and the orphans leaked anyway. The deadline path also ran S3 deletes before the marker clear, eating the same 5s budget the user-facing bind/promotion needed. Collect the refs from both failure paths, run bind + promotion on the finalize budget first, then delete on a fresh discard context. The bind-failure test now pins that discard receives a live context. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): compensate result-uncertain media uploads and commits The compensation protocol treated "the call returned an error" as "the side effect did not happen", which is wrong in both directions across the result-uncertain windows: - An upload error can follow a server-side write (lost response, deadline mid-write). The attempted key never reached the router, so nothing could reclaim it and dedup guarantees no re-resolve. The resolver now idempotently deletes the deterministic key on a fresh budget right at the failure site. - A commit error is not a rollback guarantee: a lost ack can report failure after Postgres durably committed the attachment rows, and the router's discard would then delete objects those rows reference. BindMediaRefs now converges the ambiguity on a fresh budget — any of the batch's URLs present proves the atomic commit landed (bind reports success); none proves the rollback (discard stays safe); a failed verification returns ErrMediaBindResultUnknown and the router keeps the uploads, preferring a rare orphan over a broken attachment. Fault-injection coverage: an upload error deletes the attempted key; a lost-ack commit keeps the bound attachment and reports success; a verified rollback stays a discardable error; the router keeps uploads on the unknown-outcome sentinel. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore(migrations): renumber to 207/208 after upstream took 203-206 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs(channel): note DiscardMedia self-invocation and the unknown-outcome skip Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore(migrations): renumber to 212/213 after upstream took 207-211 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(channel): replace inline media compensation with an intent ledger and reconciler Inline best-effort compensation cannot answer "did my side effect happen?" at the moment it needs the answer — the DELETE/PUT reordering and the empty-read-vs-in-flight-COMMIT gaps were both instances of the same two-system atomicity problem. Persist the intent instead and let an asynchronous reconciler settle it: - channel_media_pending_object (migration 214; claim index 215 as its own single-statement CONCURRENTLY migration): a state machine row ('pending' -> 'deleting') with lease, attempt, and backoff columns. - The resolver upserts the row BEFORE each PUT, state-guarded so a key the reconciler owns is never resurrected (the resource is skipped). ObjectURL is a pure function of configuration, so the row carries the attachment URL pre-upload. - BindMediaRefs deletes the batch's rows INSIDE the attachment-insert transaction: commit landed <=> intents gone, atomically, so an ambiguous COMMIT never needs adjudication. A key already claimed to 'deleting' is skipped (placeholder stays). - Nothing is ever deleted inline. The reconciler — an independent worker so storage latency cannot starve other sweepers — claims due rows ('pending' past the settle delay, or expired leases) under a fresh lease, checks for a durable attachment reference only AFTER the claim (race-free: bind can no longer succeed on the key), deletes unreferenced objects outside any transaction, and backs off failed deletes with attempt-based retry. Crash windows converge for free. - The settle delay is a fixed constant carrying NO correctness weight; invariant tests pin it at >=10x every pipeline budget. Metrics cover deletes, referenced clears, delete failures, and ledger backlog. Removed: MediaResolver.DiscardMedia, ErrMediaBindResultUnknown, the post-commit verification, and both router discard branches. Tests: intent-before-upload ordering; upload error leaves the row and deletes nothing; bind-wins vs reconciler-wins on the same key; lost-ack and rolled-back commit injections (intent cleared iff the attachment landed); reconciler three-state settle; expired-lease reclaim; delete failure backoff and retry; settle invariants. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): never build or sweep the media reconciler without storage store is nil when S3 is unconfigured AND the local upload dir fails to initialize, but the reconciler was constructed unconditionally and main only gates the goroutine on the reconciler pointer — the first unreferenced ledger row (rows can pre-exist from a boot where storage worked) would nil-pointer panic a bare goroutine and take down the process. Construct the reconciler only when a storage backend exists, and guard RunOnce defensively: with no deleter it skips the sweep without claiming, so rows are not stranded in 'deleting' until lease expiry. Test covers the pre-existing-row + missing-storage boot. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore(migrations): renumber to 213-216 after upstream took 212 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * refactor(channel): remove the dead pre-resolved MediaRefs ingress path lark.InboundMessage.MediaRefs and the resolver's early-returns for pre-populated refs were vestiges of the pre-detached synchronous design — no producer fills them before the router anymore. Worse, the intent ledger made the path actively misleading: refs arriving without ledger rows would be silently skipped at bind (with a log blaming the reconciler), contradicting the field's "already persisted" contract. Delete the field, its channelMessageFromLark mapping, and both early-returns; channel.InboundMessage.MediaRefs is now documented as what it actually is — ResolveMedia's output channel, always empty on ingress, attachable only through a claimed ledger intent. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): enforce workspace tenancy on every ledger query The intent-ledger upsert's conflict branch guarded only on state, so a cross-workspace storage_key collision could rewrite the row's workspace/message/url ownership; release and delete keyed on (storage_key, lease_token) alone. The derived key embeds the workspace UUID so none of this is reachable today — but tenancy must be enforced by the workspace column in every query, never derived from the key string (MUL-3515 rule, restated in this PR's review). The upsert now updates only within the same workspace (a cross-tenant conflict updates nothing, returns no row, and the resolver skips the upload — the fail-safe direction), and release/delete take (workspace_id, storage_key, lease_token). Tests pin that a foreign workspace can neither steal, release, nor delete a row. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(migrations): build the ledger primary key via a concurrent index storage_key TEXT PRIMARY KEY created its unique index implicitly at CREATE TABLE, against the repo convention that every migration index — including a new table's unique index — is built CONCURRENTLY in its own single-statement migration (the exact three-step pattern client_usage_daily shipped in 207-209). The table now declares storage_key NOT NULL, 216 builds the unique index concurrently, and 217 attaches the primary key USING INDEX; the claim index moves to 218. ON CONFLICT (storage_key) still resolves against the constraint, and the full down/up round-trip is verified. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): bound each reconciler object delete with its own timeout DeleteObject ran on the worker-lifetime context and the SDK's default HTTP client has no overall request timeout, so one black-holed connection would wedge the sequential sweep loop — and with it every later batch and the backlog gauge — forever; a single-replica deployment has no other worker to reclaim the lease. Each delete now gets a 30s timeout (well under the 2min lease), and a timed-out delete takes the existing release/backoff path. Covered by a blocking-deleter test with an injectable timeout. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): anchor the media deadline to the DB clock and bound queue waits by it Two deadline gaps from review: - The persisted marker was an application-clock timestamp compared against SQL now() everywhere it is read, so a skewed app node could shrink the fallback window and hand the agent a placeholder before the resolver's local budget ended. The append transaction now anchors a relative budget (MediaPendingSeconds) with now() + make_interval, writer and readers sharing one clock; the local resolve budget stays monotonic app-side. A DB test pins that the remaining budget measured by the DB clock equals the requested one. - enqueueMedia's waits (per-session order, global slot) only watched shutdown, so in a burst an already-expired job kept its goroutine and payload until it reached the front. Both waits now also watch the message's deadline; on expiry the job skips the resolver entirely and runs only the empty finalize (marker clear + promotion), which also unblocks the session's later messages. Covered by a queued-expiry test that finalizes while the only slot is deterministically held. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore(migrations): renumber to 216-221 after upstream took 213-215 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): start the local media budget before the append transaction The DB anchors the durable fallback at insert-time now(), but the local monotonic budget started only after AppendMessage returned — so the resolver outlived the fallback by the append/commit latency, a window where the deferred task is already claimable while the resolver still runs and the agent reads a placeholder that binds moments later. Capture the local deadline before calling AppendMessage, restoring the ordering local-gives-up <= durable-fallback-fires. A slow-append test pins that the resolver's context deadline is measured from the pre-append instant. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore(migrations): renumber to 224-229 after upstream took 216-223 Verified against the merged tree: the numeric-prefix uniqueness test passes and the full migration set applies cleanly from scratch. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): heartbeat the reconciler lease per row One claim covers up to 50 rows under a single 2-minute lease, but the batch is processed sequentially and each delete may run its full 30s timeout — a few stalled deletes could outlive the lease mid-batch, letting another replica reclaim the tail: duplicate concurrent deletes, inflated attempt/backoff on rows whose owner was alive, and skewed metrics. The lease is now renewed before EACH row's settle work, so it only ever needs to cover one row's worst case (invariant-tested: lease >= 2x the per-delete timeout). A renewal that matches no row means another worker reclaimed it after a genuine expiry — the row is skipped, leaving the new owner's state untouched. Test simulates a mid-batch reclaim and pins the skip. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): dedup post media resources and make local writes atomic A rich post may reference the same image_key/file_key in several spans. The object key derives from (message, type, key), so duplicates uploaded to the SAME key twice: LocalStorage.UploadStream truncated the destination up front and removed it outright on a copy error, so a second failing attempt destroyed the object the first success had produced — leaving an attachment row pointing at nothing. A second succeeding attempt instead produced two attachment rows for one object. Collapse duplicate spans by (fetch type, platform key) before the upload loop, and write local uploads through a temp file renamed into place so a failed write can only discard its own temp file. Tests cover a duplicated span uploading once and a failed re-upload leaving the previous object intact. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): fence late-materializing PUTs with a tombstone schedule A DELETE cannot be ordered against a PUT the client already abandoned: the store may materialize the object after the delete completes. The reconciler cleared the ledger row right after deleting, so such an object had no row and nothing to reclaim it — which made the settle delay the de-facto correctness barrier for the PUT/DELETE race, exactly what the design says it must not be. The row is now kept as a tombstone ('tombstoned' state, migration 226's CHECK) and re-deleted on a widening schedule (15m, 1h, 6h, 24h, the pass index carried in last_error), so a late materialization is reclaimed by a later pass; only after the schedule is exhausted is the row dropped. Claim, heartbeat, lease, and tenancy predicates are unchanged — a tombstone is claimed exactly like any other due row. A separate gauge reports tombstones so they cannot be mistaken for a backlog of objects awaiting reclaim, and the header comment now states precisely what state fences (bind/commit) versus what the schedule fences (late PUTs). Tests: the reviewer's interleaving — DELETE completes, the abandoned PUT materializes right after, and the object is gone by the end of the schedule — plus a full schedule walk asserting the object is counted once and the row clears at the end. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): tombstones must re-delete, not re-ask the reference question A tombstone revisit ran the same reference check as a first settle, so an attachment carrying the same URL — a re-ingested copy of the object — sent the row down the "referenced, keep it" branch: the object was kept and the row cleared, abandoning the re-delete schedule that fences the ORIGINAL object against an abandoned PUT. A tombstone has already been judged unreferenced and deleted; it exists only to re-delete whatever materializes later, so it now goes straight to the delete + schedule tail (extracted as settleDeletedObject, shared with the first settle). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(channel): keep the tombstone schedule position in its own column The re-delete pass index was encoded into last_error, which the failure path also writes: one failed re-delete erased the position and restarted the walk. A store failing intermittently could therefore keep a tombstone alive indefinitely — every recovery would resume at pass 1 and the row would never reach the end of the schedule to be dropped. tombstone_pass is now its own column (the table is introduced in this PR, so migration 226 carries it), advanced only by a successful delete, and the tombstone write clears the now-stale last_error. Test walks the schedule across a failed re-delete and asserts it resumes rather than restarts, and that the row still terminates. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(lark): derive media object keys per chat message The object key was derived from the platform message alone, so a second ingest of the same Feishu message reused the first ingest's ledger row. That row can be a tombstone (up to ~31h while the re-delete schedule runs), and the intent upsert refuses anything that has left 'pending', so the second ingest skipped the upload and silently produced a placeholder with no attachment. A re-ingest is reachable: the inbound dedup claim is reclaimable once 60s stale and the dedup row is only vacuumed after 24h. Keying on the chat message the object will attach to keeps the two ingests independent, and nothing leaks: each one's objects are covered by its own ledger row. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * refactor(storage): route both local upload paths through one atomic write UploadStream wrote through a temp file and renamed into place, but the buffered Upload path still truncated the destination up front — the destructive shape the stream path exists to avoid, one caller away from coming back. Both now share writeAtomic, which also restores the 0644 the direct write used (CreateTemp makes files 0600). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * chore(channel): gofmt the media-pending append fields Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(storage): keep the local upload chmod best-effort The rename-into-place rewrite made a failed chmod fail the whole upload. CreateTemp's 0600 has to be widened to the 0644 the direct write used, but an upload dir on a mount that ignores chmod (SMB/NFS/FUSE) accepted the old direct write fine — turning those deployments' uploads into hard errors would be a regression for a cosmetic property. Log and continue. Tests pin 0644 on both upload paths, and that a failed buffered upload leaves no temp litter and no damage to a previous object. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(channel): never re-delete an object an attachment references The tombstone pass skipped the reference check and deleted unconditionally, so a durable attachment carrying that URL lost the only object it can read — the dangling attachment the intent ledger exists to prevent, and the opposite of the posture every other path here takes ("a reclaimable orphan beats a broken attachment"). The check now runs on every pass. A positive result on a tombstone is unreachable by design — keys are per (chat message, resource) and a bind cannot attach a key that has left 'pending' — so reaching it means an invariant broke: keep the object, clear the row, log it, and count it on a dedicated reconciler_tombstone_referenced_total counter. The test's contract is flipped to assert the referenced object survives. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(storage): make the local staging file reclaimable after a crash os.CreateTemp's random suffix meant a crash between the staging write and the rename left a file nothing could name: the ledger records only the final storage key, and DeleteObject removed only the object and its sidecar. Each leftover can approach the 100 MiB resource cap and they accumulate without bound. The staging path is now derived from the object key, so DeleteObject removes it alongside the object — which makes the media reconciler reclaim it too, since the intent row is written before the upload. Opening it 0644 directly also drops the chmod the previous commit had to make best-effort. Both read paths refuse the staging name (keys come from the request URL, and a half-written body should not be readable); a user-supplied ".tmp" extension is unaffected, since object keys are generated and never dot-prefixed. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * chore(channel): renumber the media migrations after merging main main took 224 (agent_task_session_rollout_missing), so the ledger group moves to 225-230 and the cross-references inside the table migration follow. main's CompleteTask also grew a sessionRolloutMissing parameter; the three call sites this PR added to chat_input_ownership_test.go pass false. Verified the way the numbering is meant to be verified: full migration set applied from scratch on the merged tree, and the whole server suite run against that database. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(channel): let Postgres compute every reconciler deadline The reconciler built settle cutoffs, lease expiry, backoff and re-delete times from the process clock and compared them against the database's now(). A replica whose clock had drifted would therefore settle rows whose upload was still in flight (the object is deleted and the bind then refuses to attach — media silently lost), hand out leases that are born expired (rows churn between workers, attempt/backoff inflate), or compress the tombstone schedule that fences a late-materializing PUT. The four settle queries now take durations and derive their timestamps from now(), so every replica reads one clock. The parameter types are the guard: an app-side timestamp can no longer be passed. Test asserts the persisted lease, backoff and re-delete deadlines all track the database's now(). The generated code also picks up main's new agent_task_queue column in the two RETURNING task.* queries this PR adds. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * chore(lark): drop unrelated gofmt-only churn from this PR Six files carried whitespace/comment-reformatting with no functional change, unrelated to the inbound media pipeline. Reverting them to the base revision keeps the diff focused on the feature (75 -> 69 files): server/internal/service/empty_claim_cache.go server/internal/integrations/lark/markdown_detect.go server/internal/integrations/lark/ws_chunk_assembler.go server/internal/integrations/lark/ws_chunk_assembler_test.go server/internal/integrations/lark/ws_frame_test.go server/internal/integrations/lark/registration_test.go Verified: `git diff -w` against these files was already empty, so no behavior is affected. go vet clean; tests covering these files pass. Co-authored-by: multica-agent <github@multica.ai> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: Bohan-J <bohan@devv.ai> Co-authored-by: multica-agent <github@multica.ai>
789 lines
37 KiB
SQL
789 lines
37 KiB
SQL
-- 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
|
|
-- =====================
|
|
|
|
-- name: UpsertChannelInstallation :one
|
|
-- 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.
|
|
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 *;
|
|
|
|
-- name: UpsertChannelInstallationByAppID :one
|
|
-- 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.
|
|
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 *;
|
|
|
|
-- name: GetChannelInstallation :one
|
|
-- Scoped by channel_type: a per-channel caller (e.g. the Feishu store)
|
|
-- must never resolve another channel's installation by guessing its UUID.
|
|
SELECT * FROM channel_installation
|
|
WHERE id = sqlc.arg('id') AND channel_type = sqlc.arg('channel_type');
|
|
|
|
-- name: GetChannelInstallationInWorkspace :one
|
|
SELECT * FROM channel_installation
|
|
WHERE id = sqlc.arg('id')
|
|
AND workspace_id = sqlc.arg('workspace_id')
|
|
AND channel_type = sqlc.arg('channel_type');
|
|
|
|
-- name: GetChannelInstallationByAppID :one
|
|
-- 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.
|
|
SELECT * FROM channel_installation
|
|
WHERE channel_type = sqlc.arg('channel_type')
|
|
AND config ->> 'app_id' = sqlc.arg('app_id')::text;
|
|
|
|
-- name: GetChannelInstallationOwnerByAppID :one
|
|
-- 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.
|
|
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 = sqlc.arg('channel_type')
|
|
AND ci.config ->> 'app_id' = sqlc.arg('app_id')::text;
|
|
|
|
-- name: ReclaimDeadChannelInstallationByAppID :one
|
|
-- 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.
|
|
WITH dead AS (
|
|
DELETE FROM channel_installation ci
|
|
WHERE ci.channel_type = sqlc.arg('channel_type')
|
|
AND ci.config ->> 'app_id' = sqlc.arg('app_id')::text
|
|
AND (
|
|
(ci.status = 'revoked'
|
|
AND NOT (ci.workspace_id = sqlc.arg('workspace_id')
|
|
AND ci.agent_id = sqlc.arg('agent_id')))
|
|
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_chat_sessions AS (
|
|
DELETE FROM channel_chat_session_binding
|
|
WHERE installation_id IN (SELECT id FROM dead)
|
|
RETURNING 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;
|
|
|
|
-- name: DeleteChannelInstallationsByArchivedRuntimeAgents :exec
|
|
-- Application-layer replacement for the (deliberately absent, MUL-3515 §4)
|
|
-- workspace/agent ON DELETE CASCADE: on runtime teardown, before the archived
|
|
-- 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, DeleteArchivedAgentsByRuntime.
|
|
-- Mirrors the agent hard-delete predicate (runtime_id, archived_at IS NOT NULL)
|
|
-- exactly.
|
|
WITH doomed AS (
|
|
SELECT id FROM channel_installation
|
|
WHERE agent_id IN (
|
|
SELECT id FROM agent WHERE runtime_id = sqlc.arg('runtime_id') AND archived_at IS NOT NULL
|
|
)
|
|
),
|
|
cleared_chat_sessions AS (
|
|
DELETE FROM channel_chat_session_binding WHERE installation_id IN (SELECT id FROM doomed)
|
|
RETURNING chat_session_id
|
|
),
|
|
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);
|
|
|
|
-- name: ListChannelInstallationsByWorkspace :many
|
|
-- Scoped by channel_type so a per-channel management surface (e.g. the Lark
|
|
-- installation list) only ever sees its own platform's installations.
|
|
SELECT * FROM channel_installation
|
|
WHERE workspace_id = sqlc.arg('workspace_id')
|
|
AND channel_type = sqlc.arg('channel_type')
|
|
ORDER BY created_at ASC;
|
|
|
|
-- name: ListActiveChannelInstallations :many
|
|
-- 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.
|
|
SELECT ci.* 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 = sqlc.arg('channel_type')
|
|
ORDER BY ci.created_at ASC;
|
|
|
|
-- name: ListAllActiveChannelInstallations :many
|
|
-- 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).
|
|
SELECT ci.* 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;
|
|
|
|
-- name: SetChannelInstallationStatus :exec
|
|
UPDATE channel_installation
|
|
SET status = $2, updated_at = now()
|
|
WHERE id = $1;
|
|
|
|
-- name: SetChannelInstallationConfig :exec
|
|
-- 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.
|
|
UPDATE channel_installation
|
|
SET config = $2, updated_at = now()
|
|
WHERE id = $1;
|
|
|
|
-- name: BackfillChannelInstallationRegionToFeishuLark :execrows
|
|
-- 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.
|
|
UPDATE channel_installation
|
|
SET config = jsonb_set(config, '{region}', '"lark"'),
|
|
updated_at = now()
|
|
WHERE channel_type = 'feishu'
|
|
AND config ->> 'region' = 'feishu';
|
|
|
|
-- name: AcquireChannelWSLease :one
|
|
-- Atomically claims the WebSocket lease. CAS predicate accepts when no
|
|
-- holder exists, the holder expired, or the holder is us (renewal).
|
|
UPDATE channel_installation
|
|
SET ws_lease_token = sqlc.arg('new_token'),
|
|
ws_lease_expires_at = sqlc.arg('new_expires_at'),
|
|
updated_at = now()
|
|
WHERE id = sqlc.arg('id')
|
|
AND status = 'active'
|
|
AND (
|
|
ws_lease_token IS NULL
|
|
OR ws_lease_expires_at < now()
|
|
OR ws_lease_token = sqlc.arg('new_token')
|
|
)
|
|
RETURNING *;
|
|
|
|
-- name: ReleaseChannelWSLease :exec
|
|
-- Drops the lease iff we are still the holder.
|
|
UPDATE channel_installation
|
|
SET ws_lease_token = NULL,
|
|
ws_lease_expires_at = NULL,
|
|
updated_at = now()
|
|
WHERE id = $1
|
|
AND ws_lease_token = sqlc.arg('current_token');
|
|
|
|
-- =====================
|
|
-- channel_user_binding
|
|
-- =====================
|
|
|
|
-- name: CreateChannelUserBinding :one
|
|
-- 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).
|
|
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 *;
|
|
|
|
-- name: GetChannelUserBindingByUserID :one
|
|
-- 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.
|
|
SELECT * FROM channel_user_binding
|
|
WHERE installation_id = $1 AND channel_user_id = $2;
|
|
|
|
-- name: FindReusableChannelUserBinding :one
|
|
-- 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).
|
|
SELECT b.* FROM channel_user_binding b
|
|
JOIN channel_installation ci ON ci.id = b.installation_id
|
|
WHERE b.workspace_id = sqlc.arg('workspace_id')
|
|
AND b.channel_type = sqlc.arg('channel_type')
|
|
AND b.channel_user_id = sqlc.arg('channel_user_id')
|
|
AND ci.config ->> 'team_id' = sqlc.arg('team_id')::text
|
|
ORDER BY b.bound_at DESC
|
|
LIMIT 1;
|
|
|
|
-- name: DeleteChannelUserBindingsByWorkspaceMember :exec
|
|
-- 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.
|
|
DELETE FROM channel_user_binding
|
|
WHERE workspace_id = $1 AND multica_user_id = $2;
|
|
|
|
-- name: DeleteChannelUserBindingsByInstallation :exec
|
|
-- 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.
|
|
DELETE FROM channel_user_binding
|
|
WHERE installation_id = $1;
|
|
|
|
-- =====================
|
|
-- channel_chat_session_binding
|
|
-- =====================
|
|
|
|
-- name: CreateChannelChatSessionBinding :one
|
|
-- 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.
|
|
INSERT INTO channel_chat_session_binding (
|
|
chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, config
|
|
) VALUES (
|
|
$1, $2, $3, $4, $5, $6
|
|
)
|
|
RETURNING *;
|
|
|
|
-- name: GetChannelChatSessionBinding :one
|
|
-- Lookup-by-channel-chat: the inbound dispatcher finds the existing
|
|
-- chat_session before deciding whether to create one.
|
|
SELECT * FROM channel_chat_session_binding
|
|
WHERE installation_id = $1 AND channel_chat_id = $2;
|
|
|
|
-- name: GetChannelChatSessionBindingBySession :one
|
|
-- 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.
|
|
SELECT * FROM channel_chat_session_binding
|
|
WHERE chat_session_id = sqlc.arg('chat_session_id')
|
|
AND channel_type = sqlc.arg('channel_type');
|
|
|
|
-- name: UpdateChannelChatSessionBindingReplyTarget :exec
|
|
-- Records the most recent inbound trigger message + thread so the decoupled
|
|
-- outbound patcher can thread its reply back into the originating topic.
|
|
UPDATE channel_chat_session_binding
|
|
SET last_message_id = sqlc.narg('last_message_id'),
|
|
last_thread_id = sqlc.narg('last_thread_id')
|
|
WHERE chat_session_id = $1;
|
|
|
|
-- name: DeleteChannelChatSessionBindingBySession :exec
|
|
-- Application-layer integrity (replaces the old chat_session-FK ON DELETE
|
|
-- CASCADE): drop the binding when its chat_session is deleted.
|
|
DELETE FROM channel_chat_session_binding
|
|
WHERE chat_session_id = $1;
|
|
|
|
-- name: DeleteChannelChatSessionBindingsByInstallation :exec
|
|
-- 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.
|
|
DELETE FROM channel_chat_session_binding
|
|
WHERE installation_id = $1 AND channel_type = $2;
|
|
|
|
-- =====================
|
|
-- channel_inbound_message_dedup
|
|
-- =====================
|
|
|
|
-- name: ClaimChannelInboundDedup :one
|
|
-- 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.
|
|
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;
|
|
|
|
-- name: MarkChannelInboundDedupProcessed :execrows
|
|
-- 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.
|
|
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;
|
|
|
|
-- name: ReleaseChannelInboundDedup :execrows
|
|
-- 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.
|
|
DELETE FROM channel_inbound_message_dedup
|
|
WHERE installation_id = $1
|
|
AND message_id = $2
|
|
AND claim_token = $3
|
|
AND processed_at IS NULL;
|
|
|
|
-- name: PurgeChannelInboundDedup :exec
|
|
-- Vacuum job: remove dedup rows older than the supplied cutoff (e.g. 24h).
|
|
DELETE FROM channel_inbound_message_dedup
|
|
WHERE received_at < $1;
|
|
|
|
-- =====================
|
|
-- channel_inbound_audit
|
|
-- =====================
|
|
|
|
-- name: RecordChannelInboundDrop :exec
|
|
-- The only write path for dropped events. Deliberately carries no body
|
|
-- column — only routing / identity / drop_reason / timestamp.
|
|
INSERT INTO channel_inbound_audit (
|
|
installation_id, channel_type, channel_chat_id, event_type,
|
|
channel_event_id, channel_message_id, drop_reason
|
|
) VALUES (
|
|
sqlc.narg('installation_id'),
|
|
$1,
|
|
sqlc.narg('channel_chat_id'),
|
|
$2,
|
|
sqlc.narg('channel_event_id'),
|
|
sqlc.narg('channel_message_id'),
|
|
$3
|
|
);
|
|
|
|
-- name: ListChannelInboundAuditByInstallation :many
|
|
SELECT * FROM channel_inbound_audit
|
|
WHERE installation_id = $1
|
|
ORDER BY received_at DESC
|
|
LIMIT $2 OFFSET $3;
|
|
|
|
-- name: NullChannelInboundAuditInstallationID :exec
|
|
-- 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.
|
|
UPDATE channel_inbound_audit
|
|
SET installation_id = NULL
|
|
WHERE installation_id = $1;
|
|
|
|
-- =====================
|
|
-- channel_outbound_card_message
|
|
-- =====================
|
|
|
|
-- 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, sqlc.narg('task_id'), $2, $3, $4, $5
|
|
)
|
|
RETURNING *;
|
|
|
|
-- name: GetChannelOutboundCardByTask :one
|
|
-- 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.
|
|
SELECT * FROM channel_outbound_card_message
|
|
WHERE task_id = sqlc.arg('task_id')
|
|
AND channel_type = sqlc.arg('channel_type');
|
|
|
|
-- name: UpdateChannelOutboundCardStatus :exec
|
|
UPDATE channel_outbound_card_message
|
|
SET status = $2,
|
|
last_patched_at = now()
|
|
WHERE id = $1;
|
|
|
|
-- name: DeleteChannelOutboundCardMessagesBySession :exec
|
|
-- 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.
|
|
DELETE FROM channel_outbound_card_message
|
|
WHERE chat_session_id = $1;
|
|
|
|
-- =====================
|
|
-- channel_binding_token
|
|
-- =====================
|
|
|
|
-- name: CreateChannelBindingToken :one
|
|
-- 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.
|
|
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(sqlc.arg('expires_at')::timestamptz, now() + INTERVAL '15 minutes')
|
|
)
|
|
RETURNING *;
|
|
|
|
-- name: ConsumeChannelBindingToken :one
|
|
-- Atomic redemption: returns the row only if the hash exists, is
|
|
-- unconsumed, and unexpired. Two simultaneous redemptions cannot both win.
|
|
UPDATE channel_binding_token
|
|
SET consumed_at = now()
|
|
WHERE token_hash = $1
|
|
AND consumed_at IS NULL
|
|
AND expires_at > now()
|
|
RETURNING *;
|
|
|
|
-- name: PurgeExpiredChannelBindingTokens :exec
|
|
DELETE FROM channel_binding_token
|
|
WHERE expires_at < $1;
|
|
|
|
-- name: DeleteChannelBindingTokensByInstallation :exec
|
|
-- 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.
|
|
DELETE FROM channel_binding_token
|
|
WHERE installation_id = $1;
|
|
|
|
-- =====================
|
|
-- channel_media_pending_object (media intent ledger)
|
|
-- =====================
|
|
|
|
-- name: RecordChannelMediaPendingObject :one
|
|
-- 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.
|
|
INSERT INTO channel_media_pending_object (
|
|
storage_key, workspace_id, chat_message_id, storage_url, installation_id
|
|
)
|
|
VALUES ($1, $2, $3, $4, sqlc.narg(installation_id))
|
|
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;
|
|
|
|
-- name: ClaimChannelMediaPendingObjectsForBind :many
|
|
-- 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).
|
|
DELETE FROM channel_media_pending_object
|
|
WHERE storage_key = ANY(@storage_keys::text[])
|
|
AND workspace_id = @workspace_id
|
|
AND state = 'pending'
|
|
RETURNING storage_key;
|
|
|
|
-- name: ClaimChannelMediaPendingObjectsForReconcile :many
|
|
-- Short-transaction claim: flips due rows to 'deleting' under a fresh lease.
|
|
-- 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). FOR UPDATE SKIP LOCKED keeps replicas from
|
|
-- claiming the same rows; the object-storage DELETE happens outside any
|
|
-- transaction, gated by the lease token.
|
|
UPDATE channel_media_pending_object AS obj
|
|
SET state = CASE WHEN obj.state = 'tombstoned' THEN 'tombstoned' ELSE 'deleting' END,
|
|
lease_token = @lease_token,
|
|
lease_expires_at = now() + @lease::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() - @settle_delay::interval)
|
|
OR (cand.state = 'deleting' AND (cand.lease_expires_at IS NULL OR cand.lease_expires_at <= now()))
|
|
-- 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.
|
|
OR (cand.state = 'tombstoned' AND (cand.lease_expires_at IS NULL OR cand.lease_expires_at <= now()))
|
|
)
|
|
ORDER BY cand.next_attempt_at
|
|
LIMIT @batch_limit
|
|
FOR UPDATE SKIP LOCKED
|
|
) AS due
|
|
WHERE obj.storage_key = due.storage_key
|
|
RETURNING obj.*;
|
|
|
|
-- name: RenewChannelMediaPendingObjectLease :execrows
|
|
-- Per-row heartbeat: the batch shares one claim, so the lease must be
|
|
-- extended before EACH row's settle work — otherwise a few storage deletes
|
|
-- running at their full timeout could outlive the lease mid-batch and a
|
|
-- second replica would reclaim the tail, duplicating deletes and inflating
|
|
-- attempt/backoff. Zero rows affected means another worker already reclaimed
|
|
-- this row: the caller must skip it. workspace_id explicit per the tenancy
|
|
-- rule.
|
|
UPDATE channel_media_pending_object
|
|
SET lease_expires_at = now() + @lease::interval
|
|
WHERE storage_key = @storage_key
|
|
AND workspace_id = @workspace_id
|
|
AND lease_token = @lease_token;
|
|
|
|
-- name: ReleaseChannelMediaPendingObject :exec
|
|
-- 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.
|
|
UPDATE channel_media_pending_object
|
|
SET lease_token = NULL,
|
|
lease_expires_at = NULL,
|
|
next_attempt_at = now() + @backoff::interval,
|
|
last_error = @last_error
|
|
WHERE storage_key = @storage_key
|
|
AND workspace_id = @workspace_id
|
|
AND lease_token = @lease_token;
|
|
|
|
-- name: TombstoneChannelMediaPendingObject :execrows
|
|
-- 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.
|
|
UPDATE channel_media_pending_object
|
|
SET state = 'tombstoned',
|
|
lease_token = NULL,
|
|
lease_expires_at = NULL,
|
|
next_attempt_at = now() + @redelete_delay::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 = @tombstone_pass,
|
|
last_error = NULL
|
|
WHERE storage_key = @storage_key
|
|
AND workspace_id = @workspace_id
|
|
AND lease_token = @lease_token;
|
|
|
|
-- name: DeleteChannelMediaPendingObject :execrows
|
|
-- 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.
|
|
DELETE FROM channel_media_pending_object
|
|
WHERE storage_key = @storage_key
|
|
AND workspace_id = @workspace_id
|
|
AND lease_token = @lease_token;
|
|
|
|
-- name: ChannelMediaObjectIsReferenced :one
|
|
-- 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.
|
|
SELECT EXISTS (
|
|
SELECT 1 FROM attachment
|
|
WHERE chat_message_id = @chat_message_id
|
|
AND workspace_id = @workspace_id
|
|
AND url = @storage_url
|
|
) AS referenced;
|
|
|
|
-- name: CountChannelMediaPendingObjects :one
|
|
-- Ledger backlog gauge for the reconciler's observability. Tombstones are
|
|
-- reported separately: they are bounded bookkeeping for already-deleted
|
|
-- objects, not a backlog of objects awaiting reclaim.
|
|
SELECT
|
|
count(*) FILTER (WHERE state <> 'tombstoned') AS pending_objects,
|
|
count(*) FILTER (WHERE state = 'tombstoned') AS tombstoned_objects
|
|
FROM channel_media_pending_object;
|