multica/server/pkg/db/queries/lark.sql

-- 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;