diff --git a/server/migrations/125_drop_lark_tables.down.sql b/server/migrations/125_drop_lark_tables.down.sql new file mode 100644 index 0000000000..9971915e04 --- /dev/null +++ b/server/migrations/125_drop_lark_tables.down.sql @@ -0,0 +1,195 @@ +-- Recreate the lark_* tables dropped by 125_drop_lark_tables.up.sql +-- (structure only — the data lives in channel_* after migration 124). This is +-- the authoritative pre-drop schema: lark_installation plus the deltas from +-- migrations 112 (bot_union_id), 113 (per-installation dedup), 116 (region), +-- and 122 (thread-reply columns). Foreign keys are added after all tables +-- exist, so table order does not matter. + + +CREATE TABLE IF NOT EXISTS lark_binding_token ( + token_hash text NOT NULL, + workspace_id uuid NOT NULL, + installation_id uuid NOT NULL, + lark_open_id text NOT NULL, + expires_at timestamp with time zone NOT NULL, + consumed_at timestamp with time zone, + created_at timestamp with time zone DEFAULT now() NOT NULL, + CONSTRAINT lark_binding_token_ttl_cap CHECK ((expires_at <= (created_at + '00:15:00'::interval))) +); + +CREATE TABLE IF NOT EXISTS lark_chat_session_binding ( + id uuid DEFAULT gen_random_uuid() NOT NULL, + chat_session_id uuid NOT NULL, + installation_id uuid NOT NULL, + lark_chat_id text NOT NULL, + lark_chat_type text NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + last_lark_message_id text, + last_lark_thread_id text, + CONSTRAINT lark_chat_session_binding_lark_chat_type_check CHECK ((lark_chat_type = ANY (ARRAY['p2p'::text, 'group'::text]))) +); + +CREATE TABLE IF NOT EXISTS lark_inbound_audit ( + id uuid DEFAULT gen_random_uuid() NOT NULL, + installation_id uuid, + lark_chat_id text, + event_type text NOT NULL, + lark_event_id text, + lark_message_id text, + drop_reason text NOT NULL, + received_at timestamp with time zone DEFAULT now() NOT NULL +); + +CREATE TABLE IF NOT EXISTS lark_inbound_message_dedup ( + installation_id uuid NOT NULL, + message_id text NOT NULL, + received_at timestamp with time zone DEFAULT now() NOT NULL, + processed_at timestamp with time zone, + claim_token uuid DEFAULT gen_random_uuid() NOT NULL +); + +CREATE TABLE IF NOT EXISTS lark_installation ( + id uuid DEFAULT gen_random_uuid() NOT NULL, + workspace_id uuid NOT NULL, + agent_id uuid NOT NULL, + app_id text NOT NULL, + app_secret_encrypted bytea NOT NULL, + tenant_key text, + bot_open_id text NOT NULL, + installer_user_id uuid NOT NULL, + status text DEFAULT 'active'::text NOT NULL, + ws_lease_token text, + ws_lease_expires_at timestamp with time zone, + installed_at timestamp with time zone DEFAULT now() NOT NULL, + created_at timestamp with time zone DEFAULT now() NOT NULL, + updated_at timestamp with time zone DEFAULT now() NOT NULL, + bot_union_id text, + region text DEFAULT 'feishu'::text NOT NULL, + CONSTRAINT lark_installation_region_check CHECK ((region = ANY (ARRAY['feishu'::text, 'lark'::text]))), + CONSTRAINT lark_installation_status_check CHECK ((status = ANY (ARRAY['active'::text, 'revoked'::text]))) +); + +CREATE TABLE IF NOT EXISTS lark_outbound_card_message ( + id uuid DEFAULT gen_random_uuid() NOT NULL, + chat_session_id uuid NOT NULL, + task_id uuid, + lark_chat_id text NOT NULL, + lark_card_message_id text NOT NULL, + status text DEFAULT 'pending'::text NOT NULL, + last_patched_at timestamp with time zone, + created_at timestamp with time zone DEFAULT now() NOT NULL, + CONSTRAINT lark_outbound_card_message_status_check CHECK ((status = ANY (ARRAY['pending'::text, 'streaming'::text, 'final'::text, 'error'::text]))) +); + +CREATE TABLE IF NOT EXISTS lark_user_binding ( + id uuid DEFAULT gen_random_uuid() NOT NULL, + workspace_id uuid NOT NULL, + multica_user_id uuid NOT NULL, + installation_id uuid NOT NULL, + lark_open_id text NOT NULL, + union_id text, + bound_at timestamp with time zone DEFAULT now() NOT NULL +); + +ALTER TABLE ONLY lark_binding_token + ADD CONSTRAINT lark_binding_token_pkey PRIMARY KEY (token_hash); + +ALTER TABLE ONLY lark_chat_session_binding + ADD CONSTRAINT lark_chat_session_binding_chat_session_id_key UNIQUE (chat_session_id); + +ALTER TABLE ONLY lark_chat_session_binding + ADD CONSTRAINT lark_chat_session_binding_installation_id_lark_chat_id_key UNIQUE (installation_id, lark_chat_id); + +ALTER TABLE ONLY lark_chat_session_binding + ADD CONSTRAINT lark_chat_session_binding_pkey PRIMARY KEY (id); + +ALTER TABLE ONLY lark_inbound_audit + ADD CONSTRAINT lark_inbound_audit_pkey PRIMARY KEY (id); + +ALTER TABLE ONLY lark_inbound_message_dedup + ADD CONSTRAINT lark_inbound_message_dedup_pkey PRIMARY KEY (installation_id, message_id); + +ALTER TABLE ONLY lark_installation + ADD CONSTRAINT lark_installation_app_id_key UNIQUE (app_id); + +ALTER TABLE ONLY lark_installation + ADD CONSTRAINT lark_installation_id_workspace_id_key UNIQUE (id, workspace_id); + +ALTER TABLE ONLY lark_installation + ADD CONSTRAINT lark_installation_pkey PRIMARY KEY (id); + +ALTER TABLE ONLY lark_installation + ADD CONSTRAINT lark_installation_workspace_id_agent_id_key UNIQUE (workspace_id, agent_id); + +ALTER TABLE ONLY lark_outbound_card_message + ADD CONSTRAINT lark_outbound_card_message_pkey PRIMARY KEY (id); + +ALTER TABLE ONLY lark_user_binding + ADD CONSTRAINT lark_user_binding_installation_id_lark_open_id_key UNIQUE (installation_id, lark_open_id); + +ALTER TABLE ONLY lark_user_binding + ADD CONSTRAINT lark_user_binding_pkey PRIMARY KEY (id); + +CREATE INDEX IF NOT EXISTS idx_lark_binding_token_installation ON lark_binding_token USING btree (installation_id, expires_at); + +CREATE INDEX IF NOT EXISTS idx_lark_chat_session_binding_session ON lark_chat_session_binding USING btree (chat_session_id); + +CREATE INDEX IF NOT EXISTS idx_lark_inbound_audit_installation ON lark_inbound_audit USING btree (installation_id, received_at DESC); + +CREATE INDEX IF NOT EXISTS idx_lark_inbound_audit_reason ON lark_inbound_audit USING btree (drop_reason, received_at DESC); + +CREATE INDEX IF NOT EXISTS idx_lark_inbound_dedup_received ON lark_inbound_message_dedup USING btree (received_at); + +CREATE INDEX IF NOT EXISTS idx_lark_installation_agent ON lark_installation USING btree (agent_id); + +CREATE INDEX IF NOT EXISTS idx_lark_installation_lease ON lark_installation USING btree (ws_lease_expires_at) WHERE (status = 'active'::text); + +CREATE INDEX IF NOT EXISTS idx_lark_installation_workspace ON lark_installation USING btree (workspace_id); + +CREATE INDEX IF NOT EXISTS idx_lark_outbound_card_session ON lark_outbound_card_message USING btree (chat_session_id, created_at DESC); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_lark_outbound_card_task ON lark_outbound_card_message USING btree (task_id) WHERE (task_id IS NOT NULL); + +CREATE INDEX IF NOT EXISTS idx_lark_user_binding_user ON lark_user_binding USING btree (multica_user_id, workspace_id); + +CREATE INDEX IF NOT EXISTS idx_lark_user_binding_workspace_open ON lark_user_binding USING btree (workspace_id, lark_open_id); + +ALTER TABLE ONLY lark_binding_token + ADD CONSTRAINT lark_binding_token_installation_id_fkey FOREIGN KEY (installation_id) REFERENCES lark_installation(id) ON DELETE CASCADE; + +ALTER TABLE ONLY lark_binding_token + ADD CONSTRAINT lark_binding_token_workspace_id_fkey FOREIGN KEY (workspace_id) REFERENCES workspace(id) ON DELETE CASCADE; + +ALTER TABLE ONLY lark_chat_session_binding + ADD CONSTRAINT lark_chat_session_binding_chat_session_id_fkey FOREIGN KEY (chat_session_id) REFERENCES chat_session(id) ON DELETE CASCADE; + +ALTER TABLE ONLY lark_chat_session_binding + ADD CONSTRAINT lark_chat_session_binding_installation_id_fkey FOREIGN KEY (installation_id) REFERENCES lark_installation(id) ON DELETE CASCADE; + +ALTER TABLE ONLY lark_inbound_audit + ADD CONSTRAINT lark_inbound_audit_installation_id_fkey FOREIGN KEY (installation_id) REFERENCES lark_installation(id) ON DELETE SET NULL; + +ALTER TABLE ONLY lark_inbound_message_dedup + ADD CONSTRAINT lark_inbound_message_dedup_installation_id_fkey FOREIGN KEY (installation_id) REFERENCES lark_installation(id) ON DELETE CASCADE; + +ALTER TABLE ONLY lark_installation + ADD CONSTRAINT lark_installation_agent_id_fkey FOREIGN KEY (agent_id) REFERENCES agent(id) ON DELETE CASCADE; + +ALTER TABLE ONLY lark_installation + ADD CONSTRAINT lark_installation_installer_user_id_fkey FOREIGN KEY (installer_user_id) REFERENCES "user"(id) ON DELETE RESTRICT; + +ALTER TABLE ONLY lark_installation + ADD CONSTRAINT lark_installation_workspace_id_fkey FOREIGN KEY (workspace_id) REFERENCES workspace(id) ON DELETE CASCADE; + +ALTER TABLE ONLY lark_outbound_card_message + ADD CONSTRAINT lark_outbound_card_message_chat_session_id_fkey FOREIGN KEY (chat_session_id) REFERENCES chat_session(id) ON DELETE CASCADE; + +ALTER TABLE ONLY lark_outbound_card_message + ADD CONSTRAINT lark_outbound_card_message_task_id_fkey FOREIGN KEY (task_id) REFERENCES agent_task_queue(id) ON DELETE SET NULL; + +ALTER TABLE ONLY lark_user_binding + ADD CONSTRAINT lark_user_binding_installation_fk FOREIGN KEY (installation_id, workspace_id) REFERENCES lark_installation(id, workspace_id) ON DELETE CASCADE; + +ALTER TABLE ONLY lark_user_binding + ADD CONSTRAINT lark_user_binding_member_fk FOREIGN KEY (workspace_id, multica_user_id) REFERENCES member(workspace_id, user_id) ON DELETE CASCADE; + diff --git a/server/migrations/125_drop_lark_tables.up.sql b/server/migrations/125_drop_lark_tables.up.sql new file mode 100644 index 0000000000..340f6b8ac8 --- /dev/null +++ b/server/migrations/125_drop_lark_tables.up.sql @@ -0,0 +1,15 @@ +-- Drop the lark_* tables. MUL-3515 generalized them into channel_* in +-- migration 124 and the Go cutover has landed (the lark package now reads and +-- writes only channel_*), so the old tables are dead. Per the design (§5: +-- replace, do not keep both) the old path is removed rather than left in place. +-- +-- Dropped in dependency order — every other lark_* table has a foreign key to +-- lark_installation, so it goes last. The down migration recreates the schema +-- (structure only; the data already lives in channel_*). +DROP TABLE IF EXISTS lark_binding_token; +DROP TABLE IF EXISTS lark_outbound_card_message; +DROP TABLE IF EXISTS lark_inbound_audit; +DROP TABLE IF EXISTS lark_inbound_message_dedup; +DROP TABLE IF EXISTS lark_user_binding; +DROP TABLE IF EXISTS lark_chat_session_binding; +DROP TABLE IF EXISTS lark_installation; diff --git a/server/pkg/db/generated/lark.sql.go b/server/pkg/db/generated/lark.sql.go deleted file mode 100644 index d84f9b0a5e..0000000000 --- a/server/pkg/db/generated/lark.sql.go +++ /dev/null @@ -1,1175 +0,0 @@ -// Code generated by sqlc. DO NOT EDIT. -// versions: -// sqlc v1.31.1 -// source: lark.sql - -package db - -import ( - "context" - - "github.com/jackc/pgx/v5/pgtype" -) - -const acquireLarkWSLease = `-- name: AcquireLarkWSLease :one -UPDATE lark_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, app_id, app_secret_encrypted, tenant_key, bot_open_id, installer_user_id, status, ws_lease_token, ws_lease_expires_at, installed_at, created_at, updated_at, bot_union_id, region -` - -type AcquireLarkWSLeaseParams struct { - NewToken pgtype.Text `json:"new_token"` - NewExpiresAt pgtype.Timestamptz `json:"new_expires_at"` - ID pgtype.UUID `json:"id"` -} - -// Atomically claims the WebSocket lease for an installation. The CAS -// predicate accepts the lease when (a) no current holder exists, (b) -// the holder's lease has expired, or (c) the holder is us (renewal). -// Returns the row when the lease was successfully claimed; returns no -// rows when another live holder still owns it. -func (q *Queries) AcquireLarkWSLease(ctx context.Context, arg AcquireLarkWSLeaseParams) (LarkInstallation, error) { - row := q.db.QueryRow(ctx, acquireLarkWSLease, arg.NewToken, arg.NewExpiresAt, arg.ID) - var i LarkInstallation - err := row.Scan( - &i.ID, - &i.WorkspaceID, - &i.AgentID, - &i.AppID, - &i.AppSecretEncrypted, - &i.TenantKey, - &i.BotOpenID, - &i.InstallerUserID, - &i.Status, - &i.WsLeaseToken, - &i.WsLeaseExpiresAt, - &i.InstalledAt, - &i.CreatedAt, - &i.UpdatedAt, - &i.BotUnionID, - &i.Region, - ) - return i, err -} - -const backfillLarkInstallationRegionToLark = `-- name: BackfillLarkInstallationRegionToLark :execrows -UPDATE lark_installation -SET region = 'lark', - updated_at = now() -WHERE region = 'feishu' -` - -// Upgrade repair: flip every installation still carrying the migration-116 -// default ('feishu') to 'lark'. Called ONLY by -// BackfillRegionFromLegacyOverride, and ONLY when the deployment's global -// base-URL override pointed at Lark international — on such a deployment the -// whole integration talked to open.larksuite.com, so every existing install -// is really Lark and the migration's mainland default mislabels it. -// Idempotent: once flipped there is nothing left at 'feishu' to update, and -// new installs already carry the device-flow-detected region. -func (q *Queries) BackfillLarkInstallationRegionToLark(ctx context.Context) (int64, error) { - result, err := q.db.Exec(ctx, backfillLarkInstallationRegionToLark) - if err != nil { - return 0, err - } - return result.RowsAffected(), nil -} - -const claimLarkInboundDedup = `-- name: ClaimLarkInboundDedup :one - -INSERT INTO lark_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 lark_inbound_message_dedup.processed_at IS NULL - AND lark_inbound_message_dedup.received_at < now() - INTERVAL '60 seconds' -RETURNING installation_id, message_id, received_at, processed_at, claim_token -` - -type ClaimLarkInboundDedupParams struct { - InstallationID pgtype.UUID `json:"installation_id"` - MessageID string `json:"message_id"` -} - -// ===================== -// lark_inbound_message_dedup -// ===================== -// The two-phase idempotency gate. The dispatcher uses this BEFORE -// group filter / identity check / chat-session lookup so a WebSocket -// reconnect that replays an event cannot re-trigger binding prompts, -// re-write drop audit rows, or re-touch chat_session. -// -// Returns the row when a claim is acquired: -// - newly inserted (first delivery of this message_id), OR -// - re-taken from a stale in-flight claim. A claim is stale when -// processed_at IS NULL AND received_at is older than 60 seconds — -// the previous worker crashed or lost its DB connection between -// claim and finalize, and a retry should be allowed to proceed. -// -// Returns NO rows (pgx.ErrNoRows) when the claim cannot be acquired: -// - the row exists with processed_at IS NOT NULL (terminal: prior -// attempt reached a durable outcome), OR -// - the row exists with processed_at IS NULL AND received_at within -// the last 60 seconds (another worker is actively processing). -// -// Owner fencing: every successful Claim mints a fresh UUID into -// `claim_token`. The Caller passes that token to MarkLarkInbound- -// DedupProcessed / ReleaseLarkInboundDedup; mismatched tokens are -// ignored. A stale-reclaim that re-takes the row ROTATES the token, -// so the previous (slow but still alive) worker can no longer Mark -// the row — its same-tx Mark returns zero rows and the chat_message -// write rolls back. See lark_inbound_message_dedup table comment. -// -// The dispatcher MUST follow up every successful claim with exactly one -// of MarkLarkInboundDedupProcessed (durable outcome) or -// ReleaseLarkInboundDedup (infra failure before durable outcome), -// supplying the returned claim_token. Otherwise the row sits as an -// in-flight claim and the next replay attempt must wait for the -// staleness TTL. -func (q *Queries) ClaimLarkInboundDedup(ctx context.Context, arg ClaimLarkInboundDedupParams) (LarkInboundMessageDedup, error) { - row := q.db.QueryRow(ctx, claimLarkInboundDedup, arg.InstallationID, arg.MessageID) - var i LarkInboundMessageDedup - err := row.Scan( - &i.InstallationID, - &i.MessageID, - &i.ReceivedAt, - &i.ProcessedAt, - &i.ClaimToken, - ) - return i, err -} - -const consumeLarkBindingToken = `-- name: ConsumeLarkBindingToken :one -UPDATE lark_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, lark_open_id, expires_at, consumed_at, created_at -` - -// Atomic redemption. Returns the row only if (a) the hash exists, (b) -// it has not been consumed, and (c) it has not expired. The UPDATE + -// RETURNING pattern guarantees that two simultaneous redemptions of -// the same token cannot both succeed — exactly one row update wins, -// the other sees zero rows. -func (q *Queries) ConsumeLarkBindingToken(ctx context.Context, tokenHash string) (LarkBindingToken, error) { - row := q.db.QueryRow(ctx, consumeLarkBindingToken, tokenHash) - var i LarkBindingToken - err := row.Scan( - &i.TokenHash, - &i.WorkspaceID, - &i.InstallationID, - &i.LarkOpenID, - &i.ExpiresAt, - &i.ConsumedAt, - &i.CreatedAt, - ) - return i, err -} - -const createLarkBindingToken = `-- name: CreateLarkBindingToken :one - -INSERT INTO lark_binding_token ( - token_hash, workspace_id, installation_id, lark_open_id, expires_at -) VALUES ( - $1, $2, $3, $4, $5 -) -RETURNING token_hash, workspace_id, installation_id, lark_open_id, expires_at, consumed_at, created_at -` - -type CreateLarkBindingTokenParams struct { - TokenHash string `json:"token_hash"` - WorkspaceID pgtype.UUID `json:"workspace_id"` - InstallationID pgtype.UUID `json:"installation_id"` - LarkOpenID string `json:"lark_open_id"` - ExpiresAt pgtype.Timestamptz `json:"expires_at"` -} - -// ===================== -// lark_binding_token -// ===================== -// Mints a single-use binding token for an unbound Lark user. The TTL -// cap (`expires_at <= created_at + INTERVAL '15 minutes'`) is enforced -// by the DB CHECK on the table, in lockstep with lark.BindingTokenTTL. -// We store the HASH, not the raw token; the raw value is returned to -// the caller exactly once (in the URL it embeds in the Bot's reply -// card) and never persisted server-side. -func (q *Queries) CreateLarkBindingToken(ctx context.Context, arg CreateLarkBindingTokenParams) (LarkBindingToken, error) { - row := q.db.QueryRow(ctx, createLarkBindingToken, - arg.TokenHash, - arg.WorkspaceID, - arg.InstallationID, - arg.LarkOpenID, - arg.ExpiresAt, - ) - var i LarkBindingToken - err := row.Scan( - &i.TokenHash, - &i.WorkspaceID, - &i.InstallationID, - &i.LarkOpenID, - &i.ExpiresAt, - &i.ConsumedAt, - &i.CreatedAt, - ) - return i, err -} - -const createLarkChatSessionBinding = `-- name: CreateLarkChatSessionBinding :one - -INSERT INTO lark_chat_session_binding ( - chat_session_id, installation_id, lark_chat_id, lark_chat_type -) VALUES ( - $1, $2, $3, $4 -) -RETURNING id, chat_session_id, installation_id, lark_chat_id, lark_chat_type, created_at, last_lark_message_id, last_lark_thread_id -` - -type CreateLarkChatSessionBindingParams struct { - ChatSessionID pgtype.UUID `json:"chat_session_id"` - InstallationID pgtype.UUID `json:"installation_id"` - LarkChatID string `json:"lark_chat_id"` - LarkChatType string `json:"lark_chat_type"` -} - -// ===================== -// lark_chat_session_binding -// ===================== -func (q *Queries) CreateLarkChatSessionBinding(ctx context.Context, arg CreateLarkChatSessionBindingParams) (LarkChatSessionBinding, error) { - row := q.db.QueryRow(ctx, createLarkChatSessionBinding, - arg.ChatSessionID, - arg.InstallationID, - arg.LarkChatID, - arg.LarkChatType, - ) - var i LarkChatSessionBinding - err := row.Scan( - &i.ID, - &i.ChatSessionID, - &i.InstallationID, - &i.LarkChatID, - &i.LarkChatType, - &i.CreatedAt, - &i.LastLarkMessageID, - &i.LastLarkThreadID, - ) - return i, err -} - -const createLarkInstallation = `-- name: CreateLarkInstallation :one - - -INSERT INTO lark_installation ( - workspace_id, agent_id, app_id, app_secret_encrypted, - tenant_key, bot_open_id, bot_union_id, installer_user_id -) VALUES ( - $1, $2, $3, $4, $7, $5, $8, $6 -) -RETURNING id, workspace_id, agent_id, app_id, app_secret_encrypted, tenant_key, bot_open_id, installer_user_id, status, ws_lease_token, ws_lease_expires_at, installed_at, created_at, updated_at, bot_union_id, region -` - -type CreateLarkInstallationParams struct { - WorkspaceID pgtype.UUID `json:"workspace_id"` - AgentID pgtype.UUID `json:"agent_id"` - AppID string `json:"app_id"` - AppSecretEncrypted []byte `json:"app_secret_encrypted"` - BotOpenID string `json:"bot_open_id"` - InstallerUserID pgtype.UUID `json:"installer_user_id"` - TenantKey pgtype.Text `json:"tenant_key"` - BotUnionID pgtype.Text `json:"bot_union_id"` -} - -// Lark (飞书) Bot integration queries. The migration that defines these -// tables lives at server/migrations/109_lark_integration.up.sql; the -// architectural boundaries the package enforces on top of them are -// documented in server/internal/integrations/lark/doc.go. -// -// Scoping convention: every public-facing read goes through a -// workspace-scoped variant where one exists. The lookups that take only -// a UUID PK (e.g. GetLarkInstallation) are reserved for internal trusted -// callers (the WS lease scanner, the inbound dispatcher after identity -// resolution); HTTP handlers should prefer the *InWorkspace forms. -// ===================== -// lark_installation -// ===================== -// Used by the OAuth callback. `app_secret_encrypted` is the ciphertext -// produced by internal/util/secretbox — never plaintext. The -// (workspace_id, agent_id) UNIQUE constraint enforces the spec rule -// "one Multica Agent ↔ one Lark Bot"; re-installing on the same agent -// goes through UpsertLarkInstallation instead. -func (q *Queries) CreateLarkInstallation(ctx context.Context, arg CreateLarkInstallationParams) (LarkInstallation, error) { - row := q.db.QueryRow(ctx, createLarkInstallation, - arg.WorkspaceID, - arg.AgentID, - arg.AppID, - arg.AppSecretEncrypted, - arg.BotOpenID, - arg.InstallerUserID, - arg.TenantKey, - arg.BotUnionID, - ) - var i LarkInstallation - err := row.Scan( - &i.ID, - &i.WorkspaceID, - &i.AgentID, - &i.AppID, - &i.AppSecretEncrypted, - &i.TenantKey, - &i.BotOpenID, - &i.InstallerUserID, - &i.Status, - &i.WsLeaseToken, - &i.WsLeaseExpiresAt, - &i.InstalledAt, - &i.CreatedAt, - &i.UpdatedAt, - &i.BotUnionID, - &i.Region, - ) - return i, err -} - -const createLarkOutboundCardMessage = `-- name: CreateLarkOutboundCardMessage :one - -INSERT INTO lark_outbound_card_message ( - chat_session_id, task_id, lark_chat_id, lark_card_message_id, status -) VALUES ( - $1, $5, $2, $3, $4 -) -RETURNING id, chat_session_id, task_id, lark_chat_id, lark_card_message_id, status, last_patched_at, created_at -` - -type CreateLarkOutboundCardMessageParams struct { - ChatSessionID pgtype.UUID `json:"chat_session_id"` - LarkChatID string `json:"lark_chat_id"` - LarkCardMessageID string `json:"lark_card_message_id"` - Status string `json:"status"` - TaskID pgtype.UUID `json:"task_id"` -} - -// ===================== -// lark_outbound_card_message -// ===================== -func (q *Queries) CreateLarkOutboundCardMessage(ctx context.Context, arg CreateLarkOutboundCardMessageParams) (LarkOutboundCardMessage, error) { - row := q.db.QueryRow(ctx, createLarkOutboundCardMessage, - arg.ChatSessionID, - arg.LarkChatID, - arg.LarkCardMessageID, - arg.Status, - arg.TaskID, - ) - var i LarkOutboundCardMessage - err := row.Scan( - &i.ID, - &i.ChatSessionID, - &i.TaskID, - &i.LarkChatID, - &i.LarkCardMessageID, - &i.Status, - &i.LastPatchedAt, - &i.CreatedAt, - ) - return i, err -} - -const createLarkUserBinding = `-- name: CreateLarkUserBinding :one - -INSERT INTO lark_user_binding ( - workspace_id, multica_user_id, installation_id, lark_open_id, union_id -) VALUES ( - $1, $2, $3, $4, $5 -) -ON CONFLICT (installation_id, lark_open_id) DO UPDATE SET - union_id = COALESCE(EXCLUDED.union_id, lark_user_binding.union_id), - bound_at = now() -WHERE lark_user_binding.multica_user_id = EXCLUDED.multica_user_id -RETURNING id, workspace_id, multica_user_id, installation_id, lark_open_id, union_id, bound_at -` - -type CreateLarkUserBindingParams struct { - WorkspaceID pgtype.UUID `json:"workspace_id"` - MulticaUserID pgtype.UUID `json:"multica_user_id"` - InstallationID pgtype.UUID `json:"installation_id"` - LarkOpenID string `json:"lark_open_id"` - UnionID pgtype.Text `json:"union_id"` -} - -// ===================== -// lark_user_binding -// ===================== -// Records that a Lark open_id (per-installation) maps to a Multica -// user. -// -// Two structural guarantees: -// 1. The composite FK to member(workspace_id, user_id) makes this -// statement fail when the redeemer is not (or no longer) a -// workspace member — that is §4.3 of the design. -// 2. ON CONFLICT DO UPDATE is gated on `multica_user_id` matching -// the existing binding, so a second redeemer holding their own -// valid binding token CANNOT silently steal an already-bound -// open_id. If the conflict row points at a different user, the -// UPDATE is skipped and the statement returns ZERO rows — the -// caller (lark.BindingTokenService.RedeemAndBind) translates -// that into ErrBindingAlreadyAssigned. -// -// The same-user case still updates metadata (union_id refresh, -// bound_at bump) so an idempotent re-bind by the original user -// continues to work; only a cross-user re-assignment is rejected. -// True account changes must go through an explicit unbind flow, not -// through a binding token. -func (q *Queries) CreateLarkUserBinding(ctx context.Context, arg CreateLarkUserBindingParams) (LarkUserBinding, error) { - row := q.db.QueryRow(ctx, createLarkUserBinding, - arg.WorkspaceID, - arg.MulticaUserID, - arg.InstallationID, - arg.LarkOpenID, - arg.UnionID, - ) - var i LarkUserBinding - err := row.Scan( - &i.ID, - &i.WorkspaceID, - &i.MulticaUserID, - &i.InstallationID, - &i.LarkOpenID, - &i.UnionID, - &i.BoundAt, - ) - return i, err -} - -const deleteLarkUserBinding = `-- name: DeleteLarkUserBinding :exec -DELETE FROM lark_user_binding WHERE id = $1 -` - -func (q *Queries) DeleteLarkUserBinding(ctx context.Context, id pgtype.UUID) error { - _, err := q.db.Exec(ctx, deleteLarkUserBinding, id) - return err -} - -const getLarkChatSessionBinding = `-- name: GetLarkChatSessionBinding :one -SELECT id, chat_session_id, installation_id, lark_chat_id, lark_chat_type, created_at, last_lark_message_id, last_lark_thread_id FROM lark_chat_session_binding -WHERE installation_id = $1 AND lark_chat_id = $2 -` - -type GetLarkChatSessionBindingParams struct { - InstallationID pgtype.UUID `json:"installation_id"` - LarkChatID string `json:"lark_chat_id"` -} - -// Lookup-by-Lark-chat path. Used by the inbound dispatcher to find the -// existing chat_session before deciding whether to create one. The -// UNIQUE (installation_id, lark_chat_id) constraint means at most one -// row matches. -func (q *Queries) GetLarkChatSessionBinding(ctx context.Context, arg GetLarkChatSessionBindingParams) (LarkChatSessionBinding, error) { - row := q.db.QueryRow(ctx, getLarkChatSessionBinding, arg.InstallationID, arg.LarkChatID) - var i LarkChatSessionBinding - err := row.Scan( - &i.ID, - &i.ChatSessionID, - &i.InstallationID, - &i.LarkChatID, - &i.LarkChatType, - &i.CreatedAt, - &i.LastLarkMessageID, - &i.LastLarkThreadID, - ) - return i, err -} - -const getLarkChatSessionBindingBySession = `-- name: GetLarkChatSessionBindingBySession :one -SELECT id, chat_session_id, installation_id, lark_chat_id, lark_chat_type, created_at, last_lark_message_id, last_lark_thread_id FROM lark_chat_session_binding -WHERE chat_session_id = $1 -` - -// Reverse lookup: given a chat_session_id, find its Lark binding. Used -// by the outbound card patcher to know which (installation, chat_id) -// to PATCH when an agent emits a stream event for this session. -func (q *Queries) GetLarkChatSessionBindingBySession(ctx context.Context, chatSessionID pgtype.UUID) (LarkChatSessionBinding, error) { - row := q.db.QueryRow(ctx, getLarkChatSessionBindingBySession, chatSessionID) - var i LarkChatSessionBinding - err := row.Scan( - &i.ID, - &i.ChatSessionID, - &i.InstallationID, - &i.LarkChatID, - &i.LarkChatType, - &i.CreatedAt, - &i.LastLarkMessageID, - &i.LastLarkThreadID, - ) - return i, err -} - -const getLarkInstallation = `-- name: GetLarkInstallation :one -SELECT id, workspace_id, agent_id, app_id, app_secret_encrypted, tenant_key, bot_open_id, installer_user_id, status, ws_lease_token, ws_lease_expires_at, installed_at, created_at, updated_at, bot_union_id, region FROM lark_installation WHERE id = $1 -` - -func (q *Queries) GetLarkInstallation(ctx context.Context, id pgtype.UUID) (LarkInstallation, error) { - row := q.db.QueryRow(ctx, getLarkInstallation, id) - var i LarkInstallation - err := row.Scan( - &i.ID, - &i.WorkspaceID, - &i.AgentID, - &i.AppID, - &i.AppSecretEncrypted, - &i.TenantKey, - &i.BotOpenID, - &i.InstallerUserID, - &i.Status, - &i.WsLeaseToken, - &i.WsLeaseExpiresAt, - &i.InstalledAt, - &i.CreatedAt, - &i.UpdatedAt, - &i.BotUnionID, - &i.Region, - ) - return i, err -} - -const getLarkInstallationByAgent = `-- name: GetLarkInstallationByAgent :one -SELECT id, workspace_id, agent_id, app_id, app_secret_encrypted, tenant_key, bot_open_id, installer_user_id, status, ws_lease_token, ws_lease_expires_at, installed_at, created_at, updated_at, bot_union_id, region FROM lark_installation -WHERE workspace_id = $1 AND agent_id = $2 -` - -type GetLarkInstallationByAgentParams struct { - WorkspaceID pgtype.UUID `json:"workspace_id"` - AgentID pgtype.UUID `json:"agent_id"` -} - -func (q *Queries) GetLarkInstallationByAgent(ctx context.Context, arg GetLarkInstallationByAgentParams) (LarkInstallation, error) { - row := q.db.QueryRow(ctx, getLarkInstallationByAgent, arg.WorkspaceID, arg.AgentID) - var i LarkInstallation - err := row.Scan( - &i.ID, - &i.WorkspaceID, - &i.AgentID, - &i.AppID, - &i.AppSecretEncrypted, - &i.TenantKey, - &i.BotOpenID, - &i.InstallerUserID, - &i.Status, - &i.WsLeaseToken, - &i.WsLeaseExpiresAt, - &i.InstalledAt, - &i.CreatedAt, - &i.UpdatedAt, - &i.BotUnionID, - &i.Region, - ) - return i, err -} - -const getLarkInstallationByAppID = `-- name: GetLarkInstallationByAppID :one -SELECT id, workspace_id, agent_id, app_id, app_secret_encrypted, tenant_key, bot_open_id, installer_user_id, status, ws_lease_token, ws_lease_expires_at, installed_at, created_at, updated_at, bot_union_id, region FROM lark_installation WHERE app_id = $1 -` - -// Used by the OAuth callback to detect re-install vs first-install, -// and by the inbound dispatcher to route an event payload (which only -// carries app_id) to its installation row. -func (q *Queries) GetLarkInstallationByAppID(ctx context.Context, appID string) (LarkInstallation, error) { - row := q.db.QueryRow(ctx, getLarkInstallationByAppID, appID) - var i LarkInstallation - err := row.Scan( - &i.ID, - &i.WorkspaceID, - &i.AgentID, - &i.AppID, - &i.AppSecretEncrypted, - &i.TenantKey, - &i.BotOpenID, - &i.InstallerUserID, - &i.Status, - &i.WsLeaseToken, - &i.WsLeaseExpiresAt, - &i.InstalledAt, - &i.CreatedAt, - &i.UpdatedAt, - &i.BotUnionID, - &i.Region, - ) - return i, err -} - -const getLarkInstallationInWorkspace = `-- name: GetLarkInstallationInWorkspace :one -SELECT id, workspace_id, agent_id, app_id, app_secret_encrypted, tenant_key, bot_open_id, installer_user_id, status, ws_lease_token, ws_lease_expires_at, installed_at, created_at, updated_at, bot_union_id, region FROM lark_installation -WHERE id = $1 AND workspace_id = $2 -` - -type GetLarkInstallationInWorkspaceParams struct { - ID pgtype.UUID `json:"id"` - WorkspaceID pgtype.UUID `json:"workspace_id"` -} - -func (q *Queries) GetLarkInstallationInWorkspace(ctx context.Context, arg GetLarkInstallationInWorkspaceParams) (LarkInstallation, error) { - row := q.db.QueryRow(ctx, getLarkInstallationInWorkspace, arg.ID, arg.WorkspaceID) - var i LarkInstallation - err := row.Scan( - &i.ID, - &i.WorkspaceID, - &i.AgentID, - &i.AppID, - &i.AppSecretEncrypted, - &i.TenantKey, - &i.BotOpenID, - &i.InstallerUserID, - &i.Status, - &i.WsLeaseToken, - &i.WsLeaseExpiresAt, - &i.InstalledAt, - &i.CreatedAt, - &i.UpdatedAt, - &i.BotUnionID, - &i.Region, - ) - return i, err -} - -const getLarkOutboundCardByTask = `-- name: GetLarkOutboundCardByTask :one -SELECT id, chat_session_id, task_id, lark_chat_id, lark_card_message_id, status, last_patched_at, created_at FROM lark_outbound_card_message -WHERE task_id = $1 -` - -// Most card patches arrive keyed by task_id (we're streaming an agent -// run's output). The partial unique index on (task_id) WHERE task_id IS -// NOT NULL guarantees this returns at most one row. -func (q *Queries) GetLarkOutboundCardByTask(ctx context.Context, taskID pgtype.UUID) (LarkOutboundCardMessage, error) { - row := q.db.QueryRow(ctx, getLarkOutboundCardByTask, taskID) - var i LarkOutboundCardMessage - err := row.Scan( - &i.ID, - &i.ChatSessionID, - &i.TaskID, - &i.LarkChatID, - &i.LarkCardMessageID, - &i.Status, - &i.LastPatchedAt, - &i.CreatedAt, - ) - return i, err -} - -const getLarkUserBindingByOpenID = `-- name: GetLarkUserBindingByOpenID :one -SELECT id, workspace_id, multica_user_id, installation_id, lark_open_id, union_id, bound_at FROM lark_user_binding -WHERE installation_id = $1 AND lark_open_id = $2 -` - -type GetLarkUserBindingByOpenIDParams struct { - InstallationID pgtype.UUID `json:"installation_id"` - LarkOpenID string `json:"lark_open_id"` -} - -// The inbound identity check. A row here means: this open_id maps to a -// Multica user who IS currently a workspace member (the composite FK -// cascades the binding away when membership is revoked, so a row's -// existence is itself the membership proof). -func (q *Queries) GetLarkUserBindingByOpenID(ctx context.Context, arg GetLarkUserBindingByOpenIDParams) (LarkUserBinding, error) { - row := q.db.QueryRow(ctx, getLarkUserBindingByOpenID, arg.InstallationID, arg.LarkOpenID) - var i LarkUserBinding - err := row.Scan( - &i.ID, - &i.WorkspaceID, - &i.MulticaUserID, - &i.InstallationID, - &i.LarkOpenID, - &i.UnionID, - &i.BoundAt, - ) - return i, err -} - -const listActiveLarkInstallations = `-- name: ListActiveLarkInstallations :many -SELECT id, workspace_id, agent_id, app_id, app_secret_encrypted, tenant_key, bot_open_id, installer_user_id, status, ws_lease_token, ws_lease_expires_at, installed_at, created_at, updated_at, bot_union_id, region FROM lark_installation -WHERE status = 'active' -ORDER BY created_at ASC -` - -// Boot path for the WebSocket hub: enumerate every active installation -// so the hub can claim leases and open long connections. Excludes -// revoked rows — their WS should already be torn down. -func (q *Queries) ListActiveLarkInstallations(ctx context.Context) ([]LarkInstallation, error) { - rows, err := q.db.Query(ctx, listActiveLarkInstallations) - if err != nil { - return nil, err - } - defer rows.Close() - items := []LarkInstallation{} - for rows.Next() { - var i LarkInstallation - if err := rows.Scan( - &i.ID, - &i.WorkspaceID, - &i.AgentID, - &i.AppID, - &i.AppSecretEncrypted, - &i.TenantKey, - &i.BotOpenID, - &i.InstallerUserID, - &i.Status, - &i.WsLeaseToken, - &i.WsLeaseExpiresAt, - &i.InstalledAt, - &i.CreatedAt, - &i.UpdatedAt, - &i.BotUnionID, - &i.Region, - ); err != nil { - return nil, err - } - items = append(items, i) - } - if err := rows.Err(); err != nil { - return nil, err - } - return items, nil -} - -const listLarkInboundAuditByInstallation = `-- name: ListLarkInboundAuditByInstallation :many -SELECT id, installation_id, lark_chat_id, event_type, lark_event_id, lark_message_id, drop_reason, received_at FROM lark_inbound_audit -WHERE installation_id = $1 -ORDER BY received_at DESC -LIMIT $2 OFFSET $3 -` - -type ListLarkInboundAuditByInstallationParams struct { - InstallationID pgtype.UUID `json:"installation_id"` - Limit int32 `json:"limit"` - Offset int32 `json:"offset"` -} - -// Ops debugging view; paged via the (installation_id, received_at) idx. -func (q *Queries) ListLarkInboundAuditByInstallation(ctx context.Context, arg ListLarkInboundAuditByInstallationParams) ([]LarkInboundAudit, error) { - rows, err := q.db.Query(ctx, listLarkInboundAuditByInstallation, arg.InstallationID, arg.Limit, arg.Offset) - if err != nil { - return nil, err - } - defer rows.Close() - items := []LarkInboundAudit{} - for rows.Next() { - var i LarkInboundAudit - if err := rows.Scan( - &i.ID, - &i.InstallationID, - &i.LarkChatID, - &i.EventType, - &i.LarkEventID, - &i.LarkMessageID, - &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 listLarkInstallationsByWorkspace = `-- name: ListLarkInstallationsByWorkspace :many -SELECT id, workspace_id, agent_id, app_id, app_secret_encrypted, tenant_key, bot_open_id, installer_user_id, status, ws_lease_token, ws_lease_expires_at, installed_at, created_at, updated_at, bot_union_id, region FROM lark_installation -WHERE workspace_id = $1 -ORDER BY created_at ASC -` - -func (q *Queries) ListLarkInstallationsByWorkspace(ctx context.Context, workspaceID pgtype.UUID) ([]LarkInstallation, error) { - rows, err := q.db.Query(ctx, listLarkInstallationsByWorkspace, workspaceID) - if err != nil { - return nil, err - } - defer rows.Close() - items := []LarkInstallation{} - for rows.Next() { - var i LarkInstallation - if err := rows.Scan( - &i.ID, - &i.WorkspaceID, - &i.AgentID, - &i.AppID, - &i.AppSecretEncrypted, - &i.TenantKey, - &i.BotOpenID, - &i.InstallerUserID, - &i.Status, - &i.WsLeaseToken, - &i.WsLeaseExpiresAt, - &i.InstalledAt, - &i.CreatedAt, - &i.UpdatedAt, - &i.BotUnionID, - &i.Region, - ); err != nil { - return nil, err - } - items = append(items, i) - } - if err := rows.Err(); err != nil { - return nil, err - } - return items, nil -} - -const listLarkUserBindingsByInstallation = `-- name: ListLarkUserBindingsByInstallation :many -SELECT id, workspace_id, multica_user_id, installation_id, lark_open_id, union_id, bound_at FROM lark_user_binding -WHERE installation_id = $1 -ORDER BY bound_at DESC -` - -func (q *Queries) ListLarkUserBindingsByInstallation(ctx context.Context, installationID pgtype.UUID) ([]LarkUserBinding, error) { - rows, err := q.db.Query(ctx, listLarkUserBindingsByInstallation, installationID) - if err != nil { - return nil, err - } - defer rows.Close() - items := []LarkUserBinding{} - for rows.Next() { - var i LarkUserBinding - if err := rows.Scan( - &i.ID, - &i.WorkspaceID, - &i.MulticaUserID, - &i.InstallationID, - &i.LarkOpenID, - &i.UnionID, - &i.BoundAt, - ); err != nil { - return nil, err - } - items = append(items, i) - } - if err := rows.Err(); err != nil { - return nil, err - } - return items, nil -} - -const markLarkInboundDedupProcessed = `-- name: MarkLarkInboundDedupProcessed :execrows -UPDATE lark_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 MarkLarkInboundDedupProcessedParams struct { - InstallationID pgtype.UUID `json:"installation_id"` - MessageID string `json:"message_id"` - ClaimToken pgtype.UUID `json:"claim_token"` -} - -// Locks in a claim as permanently processed. Called by the dispatcher -// after a durable outcome has been reached: -// - a drop audit row was persisted (group filter / unbound user / -// revoked / invalid event), OR -// - chat_message + chat_session.updated_at were committed (ingest -// path, including ingest paths that subsequently fail at issue -// creation / task enqueue — the user-visible message is already in -// the session). -// -// For the chat_message ingest path the dispatcher invokes this query -// INSIDE the chat_message+session transaction (via qtx), so the -// durable write and the Mark commit atomically. A token mismatch -// (another worker has re-claimed the row in the meantime) returns -// zero rows; the caller treats that as a lost claim and rolls back the -// in-tx invocation, so no second chat_message is written. -// -// Guarded by processed_at IS NULL so a successful Mark is itself -// idempotent: replaying it cannot resurrect a row that was already -// terminal. -func (q *Queries) MarkLarkInboundDedupProcessed(ctx context.Context, arg MarkLarkInboundDedupProcessedParams) (int64, error) { - result, err := q.db.Exec(ctx, markLarkInboundDedupProcessed, arg.InstallationID, arg.MessageID, arg.ClaimToken) - if err != nil { - return 0, err - } - return result.RowsAffected(), nil -} - -const purgeExpiredLarkBindingTokens = `-- name: PurgeExpiredLarkBindingTokens :exec -DELETE FROM lark_binding_token -WHERE expires_at < $1 -` - -// Tokens are tiny but unbounded over time. The same vacuum cron that -// handles dedup can sweep these too. -func (q *Queries) PurgeExpiredLarkBindingTokens(ctx context.Context, expiresAt pgtype.Timestamptz) error { - _, err := q.db.Exec(ctx, purgeExpiredLarkBindingTokens, expiresAt) - return err -} - -const purgeLarkInboundDedup = `-- name: PurgeLarkInboundDedup :exec -DELETE FROM lark_inbound_message_dedup -WHERE received_at < $1 -` - -// Removes dedup rows older than the supplied cutoff. The vacuum job -// (separate cron) calls this with cutoff = now() - INTERVAL '24h'. -// Sweeps both processed and (very old) abandoned in-flight rows. -func (q *Queries) PurgeLarkInboundDedup(ctx context.Context, receivedAt pgtype.Timestamptz) error { - _, err := q.db.Exec(ctx, purgeLarkInboundDedup, receivedAt) - return err -} - -const recordLarkInboundDrop = `-- name: RecordLarkInboundDrop :exec - -INSERT INTO lark_inbound_audit ( - installation_id, lark_chat_id, event_type, - lark_event_id, lark_message_id, drop_reason -) VALUES ( - $3, - $4, - $1, - $5, - $6, - $2 -) -` - -type RecordLarkInboundDropParams struct { - EventType string `json:"event_type"` - DropReason string `json:"drop_reason"` - InstallationID pgtype.UUID `json:"installation_id"` - LarkChatID pgtype.Text `json:"lark_chat_id"` - LarkEventID pgtype.Text `json:"lark_event_id"` - LarkMessageID pgtype.Text `json:"lark_message_id"` -} - -// ===================== -// lark_inbound_audit -// ===================== -// The ONLY write path for events that fail identity check or the -// group-mention filter. Deliberately accepts no body column — the -// AuditLogger interface in internal/integrations/lark mirrors that -// shape so a caller cannot accidentally hand a body to this row. -func (q *Queries) RecordLarkInboundDrop(ctx context.Context, arg RecordLarkInboundDropParams) error { - _, err := q.db.Exec(ctx, recordLarkInboundDrop, - arg.EventType, - arg.DropReason, - arg.InstallationID, - arg.LarkChatID, - arg.LarkEventID, - arg.LarkMessageID, - ) - return err -} - -const releaseLarkInboundDedup = `-- name: ReleaseLarkInboundDedup :execrows -DELETE FROM lark_inbound_message_dedup -WHERE installation_id = $1 - AND message_id = $2 - AND claim_token = $3 - AND processed_at IS NULL -` - -type ReleaseLarkInboundDedupParams struct { - InstallationID pgtype.UUID `json:"installation_id"` - MessageID string `json:"message_id"` - ClaimToken pgtype.UUID `json:"claim_token"` -} - -// Releases an in-flight claim. Called by the dispatcher when an infra -// error occurred BEFORE any durable side effect (e.g. EnsureChatSession -// or AppendUserMessage returned an error and its transaction rolled -// back). Deleting the row lets the WS adapter's retry re-acquire the -// claim immediately, instead of waiting for the 60-second staleness -// TTL. Guarded by processed_at IS NULL so an out-of-order Release -// cannot undo a Mark; guarded by claim_token so a slow-but-alive worker -// whose claim was reclaimed cannot delete the new holder's row. -func (q *Queries) ReleaseLarkInboundDedup(ctx context.Context, arg ReleaseLarkInboundDedupParams) (int64, error) { - result, err := q.db.Exec(ctx, releaseLarkInboundDedup, arg.InstallationID, arg.MessageID, arg.ClaimToken) - if err != nil { - return 0, err - } - return result.RowsAffected(), nil -} - -const releaseLarkWSLease = `-- name: ReleaseLarkWSLease :exec -UPDATE lark_installation -SET ws_lease_token = NULL, - ws_lease_expires_at = NULL, - updated_at = now() -WHERE id = $1 - AND ws_lease_token = $2 -` - -type ReleaseLarkWSLeaseParams struct { - ID pgtype.UUID `json:"id"` - CurrentToken pgtype.Text `json:"current_token"` -} - -// Drops the lease iff we're still the holder. A racing acquirer that -// already took over will not have its lease cleared. -func (q *Queries) ReleaseLarkWSLease(ctx context.Context, arg ReleaseLarkWSLeaseParams) error { - _, err := q.db.Exec(ctx, releaseLarkWSLease, arg.ID, arg.CurrentToken) - return err -} - -const setLarkInstallationBotUnionID = `-- name: SetLarkInstallationBotUnionID :exec -UPDATE lark_installation -SET bot_union_id = $2, - updated_at = now() -WHERE id = $1 -` - -type SetLarkInstallationBotUnionIDParams struct { - ID pgtype.UUID `json:"id"` - BotUnionID pgtype.Text `json:"bot_union_id"` -} - -// Operator-only backfill for installations created before the -// bot_union_id column existed (migration 112). Production reads do -// NOT use this — finishSuccess writes union_id during install, and -// the upsert path writes it on re-install. Kept as a focused single- -// column UPDATE so the backfill cannot accidentally overwrite app -// credentials, status, or lease state. -func (q *Queries) SetLarkInstallationBotUnionID(ctx context.Context, arg SetLarkInstallationBotUnionIDParams) error { - _, err := q.db.Exec(ctx, setLarkInstallationBotUnionID, arg.ID, arg.BotUnionID) - return err -} - -const setLarkInstallationStatus = `-- name: SetLarkInstallationStatus :exec -UPDATE lark_installation -SET status = $2, updated_at = now() -WHERE id = $1 -` - -type SetLarkInstallationStatusParams struct { - ID pgtype.UUID `json:"id"` - Status string `json:"status"` -} - -func (q *Queries) SetLarkInstallationStatus(ctx context.Context, arg SetLarkInstallationStatusParams) error { - _, err := q.db.Exec(ctx, setLarkInstallationStatus, arg.ID, arg.Status) - return err -} - -const updateLarkChatSessionBindingReplyTarget = `-- name: UpdateLarkChatSessionBindingReplyTarget :exec -UPDATE lark_chat_session_binding -SET last_lark_message_id = $2, - last_lark_thread_id = $3 -WHERE chat_session_id = $1 -` - -type UpdateLarkChatSessionBindingReplyTargetParams struct { - ChatSessionID pgtype.UUID `json:"chat_session_id"` - LastLarkMessageID pgtype.Text `json:"last_lark_message_id"` - LastLarkThreadID pgtype.Text `json:"last_lark_thread_id"` -} - -// Records the most recent inbound trigger message + the Lark topic -// (thread) it belongs to, so the decoupled outbound patcher can thread -// its reply back into the originating 话题. Called on every ingested -// message (in the same tx as the chat_message write). last_lark_thread_id -// is NULL for non-thread messages, which keeps the outbound on the -// existing chat-level send path; a non-NULL value flips the next reply -// into a thread reply targeting last_lark_message_id. -func (q *Queries) UpdateLarkChatSessionBindingReplyTarget(ctx context.Context, arg UpdateLarkChatSessionBindingReplyTargetParams) error { - _, err := q.db.Exec(ctx, updateLarkChatSessionBindingReplyTarget, arg.ChatSessionID, arg.LastLarkMessageID, arg.LastLarkThreadID) - return err -} - -const updateLarkOutboundCardStatus = `-- name: UpdateLarkOutboundCardStatus :exec -UPDATE lark_outbound_card_message -SET status = $2, - last_patched_at = now() -WHERE id = $1 -` - -type UpdateLarkOutboundCardStatusParams struct { - ID pgtype.UUID `json:"id"` - Status string `json:"status"` -} - -func (q *Queries) UpdateLarkOutboundCardStatus(ctx context.Context, arg UpdateLarkOutboundCardStatusParams) error { - _, err := q.db.Exec(ctx, updateLarkOutboundCardStatus, arg.ID, arg.Status) - return err -} - -const upsertLarkInstallation = `-- name: UpsertLarkInstallation :one -INSERT INTO lark_installation ( - workspace_id, agent_id, app_id, app_secret_encrypted, - tenant_key, bot_open_id, bot_union_id, installer_user_id, region -) VALUES ( - $1, $2, $3, $4, $7, $5, $8, $6, $9 -) -ON CONFLICT (workspace_id, agent_id) DO UPDATE SET - app_id = EXCLUDED.app_id, - app_secret_encrypted = EXCLUDED.app_secret_encrypted, - tenant_key = EXCLUDED.tenant_key, - bot_open_id = EXCLUDED.bot_open_id, - bot_union_id = EXCLUDED.bot_union_id, - installer_user_id = EXCLUDED.installer_user_id, - region = EXCLUDED.region, - status = 'active', - installed_at = now(), - updated_at = now() -RETURNING id, workspace_id, agent_id, app_id, app_secret_encrypted, tenant_key, bot_open_id, installer_user_id, status, ws_lease_token, ws_lease_expires_at, installed_at, created_at, updated_at, bot_union_id, region -` - -type UpsertLarkInstallationParams struct { - WorkspaceID pgtype.UUID `json:"workspace_id"` - AgentID pgtype.UUID `json:"agent_id"` - AppID string `json:"app_id"` - AppSecretEncrypted []byte `json:"app_secret_encrypted"` - BotOpenID string `json:"bot_open_id"` - InstallerUserID pgtype.UUID `json:"installer_user_id"` - TenantKey pgtype.Text `json:"tenant_key"` - BotUnionID pgtype.Text `json:"bot_union_id"` - Region string `json:"region"` -} - -// Re-install path: a user who already bound this agent to Lark scans -// the QR again (e.g. they rotated their Lark app secret, or revoked + -// reinstalled). We refresh the app credentials, bot identity, and -// installer attribution, and force status back to 'active'. The WS -// lease is intentionally NOT reset here — the inbound hub owns lease -// lifecycle. -func (q *Queries) UpsertLarkInstallation(ctx context.Context, arg UpsertLarkInstallationParams) (LarkInstallation, error) { - row := q.db.QueryRow(ctx, upsertLarkInstallation, - arg.WorkspaceID, - arg.AgentID, - arg.AppID, - arg.AppSecretEncrypted, - arg.BotOpenID, - arg.InstallerUserID, - arg.TenantKey, - arg.BotUnionID, - arg.Region, - ) - var i LarkInstallation - err := row.Scan( - &i.ID, - &i.WorkspaceID, - &i.AgentID, - &i.AppID, - &i.AppSecretEncrypted, - &i.TenantKey, - &i.BotOpenID, - &i.InstallerUserID, - &i.Status, - &i.WsLeaseToken, - &i.WsLeaseExpiresAt, - &i.InstalledAt, - &i.CreatedAt, - &i.UpdatedAt, - &i.BotUnionID, - &i.Region, - ) - return i, err -} diff --git a/server/pkg/db/generated/models.go b/server/pkg/db/generated/models.go index 58c4ecef9b..ac60617b67 100644 --- a/server/pkg/db/generated/models.go +++ b/server/pkg/db/generated/models.go @@ -523,86 +523,6 @@ type IssueToLabel struct { LabelID pgtype.UUID `json:"label_id"` } -type LarkBindingToken struct { - TokenHash string `json:"token_hash"` - WorkspaceID pgtype.UUID `json:"workspace_id"` - InstallationID pgtype.UUID `json:"installation_id"` - LarkOpenID string `json:"lark_open_id"` - ExpiresAt pgtype.Timestamptz `json:"expires_at"` - ConsumedAt pgtype.Timestamptz `json:"consumed_at"` - CreatedAt pgtype.Timestamptz `json:"created_at"` -} - -type LarkChatSessionBinding struct { - ID pgtype.UUID `json:"id"` - ChatSessionID pgtype.UUID `json:"chat_session_id"` - InstallationID pgtype.UUID `json:"installation_id"` - LarkChatID string `json:"lark_chat_id"` - LarkChatType string `json:"lark_chat_type"` - CreatedAt pgtype.Timestamptz `json:"created_at"` - LastLarkMessageID pgtype.Text `json:"last_lark_message_id"` - LastLarkThreadID pgtype.Text `json:"last_lark_thread_id"` -} - -type LarkInboundAudit struct { - ID pgtype.UUID `json:"id"` - InstallationID pgtype.UUID `json:"installation_id"` - LarkChatID pgtype.Text `json:"lark_chat_id"` - EventType string `json:"event_type"` - LarkEventID pgtype.Text `json:"lark_event_id"` - LarkMessageID pgtype.Text `json:"lark_message_id"` - DropReason string `json:"drop_reason"` - ReceivedAt pgtype.Timestamptz `json:"received_at"` -} - -type LarkInboundMessageDedup struct { - InstallationID pgtype.UUID `json:"installation_id"` - MessageID string `json:"message_id"` - ReceivedAt pgtype.Timestamptz `json:"received_at"` - ProcessedAt pgtype.Timestamptz `json:"processed_at"` - ClaimToken pgtype.UUID `json:"claim_token"` -} - -type LarkInstallation struct { - ID pgtype.UUID `json:"id"` - WorkspaceID pgtype.UUID `json:"workspace_id"` - AgentID pgtype.UUID `json:"agent_id"` - AppID string `json:"app_id"` - AppSecretEncrypted []byte `json:"app_secret_encrypted"` - TenantKey pgtype.Text `json:"tenant_key"` - BotOpenID string `json:"bot_open_id"` - InstallerUserID pgtype.UUID `json:"installer_user_id"` - Status string `json:"status"` - WsLeaseToken pgtype.Text `json:"ws_lease_token"` - WsLeaseExpiresAt pgtype.Timestamptz `json:"ws_lease_expires_at"` - InstalledAt pgtype.Timestamptz `json:"installed_at"` - CreatedAt pgtype.Timestamptz `json:"created_at"` - UpdatedAt pgtype.Timestamptz `json:"updated_at"` - BotUnionID pgtype.Text `json:"bot_union_id"` - Region string `json:"region"` -} - -type LarkOutboundCardMessage struct { - ID pgtype.UUID `json:"id"` - ChatSessionID pgtype.UUID `json:"chat_session_id"` - TaskID pgtype.UUID `json:"task_id"` - LarkChatID string `json:"lark_chat_id"` - LarkCardMessageID string `json:"lark_card_message_id"` - Status string `json:"status"` - LastPatchedAt pgtype.Timestamptz `json:"last_patched_at"` - CreatedAt pgtype.Timestamptz `json:"created_at"` -} - -type LarkUserBinding struct { - ID pgtype.UUID `json:"id"` - WorkspaceID pgtype.UUID `json:"workspace_id"` - MulticaUserID pgtype.UUID `json:"multica_user_id"` - InstallationID pgtype.UUID `json:"installation_id"` - LarkOpenID string `json:"lark_open_id"` - UnionID pgtype.Text `json:"union_id"` - BoundAt pgtype.Timestamptz `json:"bound_at"` -} - type Member struct { ID pgtype.UUID `json:"id"` WorkspaceID pgtype.UUID `json:"workspace_id"` diff --git a/server/pkg/db/queries/lark.sql b/server/pkg/db/queries/lark.sql deleted file mode 100644 index a9c418d01b..0000000000 --- a/server/pkg/db/queries/lark.sql +++ /dev/null @@ -1,420 +0,0 @@ --- Lark (飞书) Bot integration queries. The migration that defines these --- tables lives at server/migrations/109_lark_integration.up.sql; the --- architectural boundaries the package enforces on top of them are --- documented in server/internal/integrations/lark/doc.go. --- --- Scoping convention: every public-facing read goes through a --- workspace-scoped variant where one exists. The lookups that take only --- a UUID PK (e.g. GetLarkInstallation) are reserved for internal trusted --- callers (the WS lease scanner, the inbound dispatcher after identity --- resolution); HTTP handlers should prefer the *InWorkspace forms. - --- ===================== --- lark_installation --- ===================== - --- name: CreateLarkInstallation :one --- Used by the OAuth callback. `app_secret_encrypted` is the ciphertext --- produced by internal/util/secretbox — never plaintext. The --- (workspace_id, agent_id) UNIQUE constraint enforces the spec rule --- "one Multica Agent ↔ one Lark Bot"; re-installing on the same agent --- goes through UpsertLarkInstallation instead. -INSERT INTO lark_installation ( - workspace_id, agent_id, app_id, app_secret_encrypted, - tenant_key, bot_open_id, bot_union_id, installer_user_id -) VALUES ( - $1, $2, $3, $4, sqlc.narg('tenant_key'), $5, sqlc.narg('bot_union_id'), $6 -) -RETURNING *; - --- name: UpsertLarkInstallation :one --- Re-install path: a user who already bound this agent to Lark scans --- the QR again (e.g. they rotated their Lark app secret, or revoked + --- reinstalled). We refresh the app credentials, bot identity, and --- installer attribution, and force status back to 'active'. The WS --- lease is intentionally NOT reset here — the inbound hub owns lease --- lifecycle. -INSERT INTO lark_installation ( - workspace_id, agent_id, app_id, app_secret_encrypted, - tenant_key, bot_open_id, bot_union_id, installer_user_id, region -) VALUES ( - $1, $2, $3, $4, sqlc.narg('tenant_key'), $5, sqlc.narg('bot_union_id'), $6, sqlc.arg('region') -) -ON CONFLICT (workspace_id, agent_id) DO UPDATE SET - app_id = EXCLUDED.app_id, - app_secret_encrypted = EXCLUDED.app_secret_encrypted, - tenant_key = EXCLUDED.tenant_key, - bot_open_id = EXCLUDED.bot_open_id, - bot_union_id = EXCLUDED.bot_union_id, - installer_user_id = EXCLUDED.installer_user_id, - region = EXCLUDED.region, - status = 'active', - installed_at = now(), - updated_at = now() -RETURNING *; - --- name: BackfillLarkInstallationRegionToLark :execrows --- Upgrade repair: flip every installation still carrying the migration-116 --- default ('feishu') to 'lark'. Called ONLY by --- BackfillRegionFromLegacyOverride, and ONLY when the deployment's global --- base-URL override pointed at Lark international — on such a deployment the --- whole integration talked to open.larksuite.com, so every existing install --- is really Lark and the migration's mainland default mislabels it. --- Idempotent: once flipped there is nothing left at 'feishu' to update, and --- new installs already carry the device-flow-detected region. -UPDATE lark_installation -SET region = 'lark', - updated_at = now() -WHERE region = 'feishu'; - --- name: SetLarkInstallationBotUnionID :exec --- Operator-only backfill for installations created before the --- bot_union_id column existed (migration 112). Production reads do --- NOT use this — finishSuccess writes union_id during install, and --- the upsert path writes it on re-install. Kept as a focused single- --- column UPDATE so the backfill cannot accidentally overwrite app --- credentials, status, or lease state. -UPDATE lark_installation -SET bot_union_id = $2, - updated_at = now() -WHERE id = $1; - --- name: GetLarkInstallation :one -SELECT * FROM lark_installation WHERE id = $1; - --- name: GetLarkInstallationInWorkspace :one -SELECT * FROM lark_installation -WHERE id = $1 AND workspace_id = $2; - --- name: GetLarkInstallationByAgent :one -SELECT * FROM lark_installation -WHERE workspace_id = $1 AND agent_id = $2; - --- name: GetLarkInstallationByAppID :one --- Used by the OAuth callback to detect re-install vs first-install, --- and by the inbound dispatcher to route an event payload (which only --- carries app_id) to its installation row. -SELECT * FROM lark_installation WHERE app_id = $1; - --- name: ListLarkInstallationsByWorkspace :many -SELECT * FROM lark_installation -WHERE workspace_id = $1 -ORDER BY created_at ASC; - --- name: ListActiveLarkInstallations :many --- Boot path for the WebSocket hub: enumerate every active installation --- so the hub can claim leases and open long connections. Excludes --- revoked rows — their WS should already be torn down. -SELECT * FROM lark_installation -WHERE status = 'active' -ORDER BY created_at ASC; - --- name: SetLarkInstallationStatus :exec -UPDATE lark_installation -SET status = $2, updated_at = now() -WHERE id = $1; - --- name: AcquireLarkWSLease :one --- Atomically claims the WebSocket lease for an installation. The CAS --- predicate accepts the lease when (a) no current holder exists, (b) --- the holder's lease has expired, or (c) the holder is us (renewal). --- Returns the row when the lease was successfully claimed; returns no --- rows when another live holder still owns it. -UPDATE lark_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: ReleaseLarkWSLease :exec --- Drops the lease iff we're still the holder. A racing acquirer that --- already took over will not have its lease cleared. -UPDATE lark_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'); - --- ===================== --- lark_user_binding --- ===================== - --- name: CreateLarkUserBinding :one --- Records that a Lark open_id (per-installation) maps to a Multica --- user. --- --- Two structural guarantees: --- 1. The composite FK to member(workspace_id, user_id) makes this --- statement fail when the redeemer is not (or no longer) a --- workspace member — that is §4.3 of the design. --- 2. ON CONFLICT DO UPDATE is gated on `multica_user_id` matching --- the existing binding, so a second redeemer holding their own --- valid binding token CANNOT silently steal an already-bound --- open_id. If the conflict row points at a different user, the --- UPDATE is skipped and the statement returns ZERO rows — the --- caller (lark.BindingTokenService.RedeemAndBind) translates --- that into ErrBindingAlreadyAssigned. --- --- The same-user case still updates metadata (union_id refresh, --- bound_at bump) so an idempotent re-bind by the original user --- continues to work; only a cross-user re-assignment is rejected. --- True account changes must go through an explicit unbind flow, not --- through a binding token. -INSERT INTO lark_user_binding ( - workspace_id, multica_user_id, installation_id, lark_open_id, union_id -) VALUES ( - $1, $2, $3, $4, sqlc.narg('union_id') -) -ON CONFLICT (installation_id, lark_open_id) DO UPDATE SET - union_id = COALESCE(EXCLUDED.union_id, lark_user_binding.union_id), - bound_at = now() -WHERE lark_user_binding.multica_user_id = EXCLUDED.multica_user_id -RETURNING *; - --- name: GetLarkUserBindingByOpenID :one --- The inbound identity check. A row here means: this open_id maps to a --- Multica user who IS currently a workspace member (the composite FK --- cascades the binding away when membership is revoked, so a row's --- existence is itself the membership proof). -SELECT * FROM lark_user_binding -WHERE installation_id = $1 AND lark_open_id = $2; - --- name: ListLarkUserBindingsByInstallation :many -SELECT * FROM lark_user_binding -WHERE installation_id = $1 -ORDER BY bound_at DESC; - --- name: DeleteLarkUserBinding :exec -DELETE FROM lark_user_binding WHERE id = $1; - --- ===================== --- lark_chat_session_binding --- ===================== - --- name: CreateLarkChatSessionBinding :one -INSERT INTO lark_chat_session_binding ( - chat_session_id, installation_id, lark_chat_id, lark_chat_type -) VALUES ( - $1, $2, $3, $4 -) -RETURNING *; - --- name: GetLarkChatSessionBinding :one --- Lookup-by-Lark-chat path. Used by the inbound dispatcher to find the --- existing chat_session before deciding whether to create one. The --- UNIQUE (installation_id, lark_chat_id) constraint means at most one --- row matches. -SELECT * FROM lark_chat_session_binding -WHERE installation_id = $1 AND lark_chat_id = $2; - --- name: GetLarkChatSessionBindingBySession :one --- Reverse lookup: given a chat_session_id, find its Lark binding. Used --- by the outbound card patcher to know which (installation, chat_id) --- to PATCH when an agent emits a stream event for this session. -SELECT * FROM lark_chat_session_binding -WHERE chat_session_id = $1; - --- name: UpdateLarkChatSessionBindingReplyTarget :exec --- Records the most recent inbound trigger message + the Lark topic --- (thread) it belongs to, so the decoupled outbound patcher can thread --- its reply back into the originating 话题. Called on every ingested --- message (in the same tx as the chat_message write). last_lark_thread_id --- is NULL for non-thread messages, which keeps the outbound on the --- existing chat-level send path; a non-NULL value flips the next reply --- into a thread reply targeting last_lark_message_id. -UPDATE lark_chat_session_binding -SET last_lark_message_id = sqlc.narg('last_lark_message_id'), - last_lark_thread_id = sqlc.narg('last_lark_thread_id') -WHERE chat_session_id = $1; - --- ===================== --- lark_inbound_message_dedup --- ===================== - --- name: ClaimLarkInboundDedup :one --- The two-phase idempotency gate. The dispatcher uses this BEFORE --- group filter / identity check / chat-session lookup so a WebSocket --- reconnect that replays an event cannot re-trigger binding prompts, --- re-write drop audit rows, or re-touch chat_session. --- --- Returns the row when a claim is acquired: --- - newly inserted (first delivery of this message_id), OR --- - re-taken from a stale in-flight claim. A claim is stale when --- processed_at IS NULL AND received_at is older than 60 seconds — --- the previous worker crashed or lost its DB connection between --- claim and finalize, and a retry should be allowed to proceed. --- --- Returns NO rows (pgx.ErrNoRows) when the claim cannot be acquired: --- - the row exists with processed_at IS NOT NULL (terminal: prior --- attempt reached a durable outcome), OR --- - the row exists with processed_at IS NULL AND received_at within --- the last 60 seconds (another worker is actively processing). --- --- Owner fencing: every successful Claim mints a fresh UUID into --- `claim_token`. The Caller passes that token to MarkLarkInbound- --- DedupProcessed / ReleaseLarkInboundDedup; mismatched tokens are --- ignored. A stale-reclaim that re-takes the row ROTATES the token, --- so the previous (slow but still alive) worker can no longer Mark --- the row — its same-tx Mark returns zero rows and the chat_message --- write rolls back. See lark_inbound_message_dedup table comment. --- --- The dispatcher MUST follow up every successful claim with exactly one --- of MarkLarkInboundDedupProcessed (durable outcome) or --- ReleaseLarkInboundDedup (infra failure before durable outcome), --- supplying the returned claim_token. Otherwise the row sits as an --- in-flight claim and the next replay attempt must wait for the --- staleness TTL. -INSERT INTO lark_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 lark_inbound_message_dedup.processed_at IS NULL - AND lark_inbound_message_dedup.received_at < now() - INTERVAL '60 seconds' -RETURNING installation_id, message_id, received_at, processed_at, claim_token; - --- name: MarkLarkInboundDedupProcessed :execrows --- Locks in a claim as permanently processed. Called by the dispatcher --- after a durable outcome has been reached: --- - a drop audit row was persisted (group filter / unbound user / --- revoked / invalid event), OR --- - chat_message + chat_session.updated_at were committed (ingest --- path, including ingest paths that subsequently fail at issue --- creation / task enqueue — the user-visible message is already in --- the session). --- For the chat_message ingest path the dispatcher invokes this query --- INSIDE the chat_message+session transaction (via qtx), so the --- durable write and the Mark commit atomically. A token mismatch --- (another worker has re-claimed the row in the meantime) returns --- zero rows; the caller treats that as a lost claim and rolls back the --- in-tx invocation, so no second chat_message is written. --- --- Guarded by processed_at IS NULL so a successful Mark is itself --- idempotent: replaying it cannot resurrect a row that was already --- terminal. -UPDATE lark_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: ReleaseLarkInboundDedup :execrows --- Releases an in-flight claim. Called by the dispatcher when an infra --- error occurred BEFORE any durable side effect (e.g. EnsureChatSession --- or AppendUserMessage returned an error and its transaction rolled --- back). Deleting the row lets the WS adapter's retry re-acquire the --- claim immediately, instead of waiting for the 60-second staleness --- TTL. Guarded by processed_at IS NULL so an out-of-order Release --- cannot undo a Mark; guarded by claim_token so a slow-but-alive worker --- whose claim was reclaimed cannot delete the new holder's row. -DELETE FROM lark_inbound_message_dedup -WHERE installation_id = $1 - AND message_id = $2 - AND claim_token = $3 - AND processed_at IS NULL; - --- name: PurgeLarkInboundDedup :exec --- Removes dedup rows older than the supplied cutoff. The vacuum job --- (separate cron) calls this with cutoff = now() - INTERVAL '24h'. --- Sweeps both processed and (very old) abandoned in-flight rows. -DELETE FROM lark_inbound_message_dedup -WHERE received_at < $1; - --- ===================== --- lark_inbound_audit --- ===================== - --- name: RecordLarkInboundDrop :exec --- The ONLY write path for events that fail identity check or the --- group-mention filter. Deliberately accepts no body column — the --- AuditLogger interface in internal/integrations/lark mirrors that --- shape so a caller cannot accidentally hand a body to this row. -INSERT INTO lark_inbound_audit ( - installation_id, lark_chat_id, event_type, - lark_event_id, lark_message_id, drop_reason -) VALUES ( - sqlc.narg('installation_id'), - sqlc.narg('lark_chat_id'), - $1, - sqlc.narg('lark_event_id'), - sqlc.narg('lark_message_id'), - $2 -); - --- name: ListLarkInboundAuditByInstallation :many --- Ops debugging view; paged via the (installation_id, received_at) idx. -SELECT * FROM lark_inbound_audit -WHERE installation_id = $1 -ORDER BY received_at DESC -LIMIT $2 OFFSET $3; - --- ===================== --- lark_outbound_card_message --- ===================== - --- name: CreateLarkOutboundCardMessage :one -INSERT INTO lark_outbound_card_message ( - chat_session_id, task_id, lark_chat_id, lark_card_message_id, status -) VALUES ( - $1, sqlc.narg('task_id'), $2, $3, $4 -) -RETURNING *; - --- name: GetLarkOutboundCardByTask :one --- Most card patches arrive keyed by task_id (we're streaming an agent --- run's output). The partial unique index on (task_id) WHERE task_id IS --- NOT NULL guarantees this returns at most one row. -SELECT * FROM lark_outbound_card_message -WHERE task_id = $1; - --- name: UpdateLarkOutboundCardStatus :exec -UPDATE lark_outbound_card_message -SET status = $2, - last_patched_at = now() -WHERE id = $1; - --- ===================== --- lark_binding_token --- ===================== - --- name: CreateLarkBindingToken :one --- Mints a single-use binding token for an unbound Lark user. The TTL --- cap (`expires_at <= created_at + INTERVAL '15 minutes'`) is enforced --- by the DB CHECK on the table, in lockstep with lark.BindingTokenTTL. --- We store the HASH, not the raw token; the raw value is returned to --- the caller exactly once (in the URL it embeds in the Bot's reply --- card) and never persisted server-side. -INSERT INTO lark_binding_token ( - token_hash, workspace_id, installation_id, lark_open_id, expires_at -) VALUES ( - $1, $2, $3, $4, $5 -) -RETURNING *; - --- name: ConsumeLarkBindingToken :one --- Atomic redemption. Returns the row only if (a) the hash exists, (b) --- it has not been consumed, and (c) it has not expired. The UPDATE + --- RETURNING pattern guarantees that two simultaneous redemptions of --- the same token cannot both succeed — exactly one row update wins, --- the other sees zero rows. -UPDATE lark_binding_token -SET consumed_at = now() -WHERE token_hash = $1 - AND consumed_at IS NULL - AND expires_at > now() -RETURNING *; - --- name: PurgeExpiredLarkBindingTokens :exec --- Tokens are tiny but unbounded over time. The same vacuum cron that --- handles dedup can sweep these too. -DELETE FROM lark_binding_token -WHERE expires_at < $1;