mirror of
https://github.com/multica-ai/multica.git
synced 2026-07-12 12:18:55 +02:00
0aa3b53c25
* feat(lark): reply inside the originating thread (话题) instead of the group When a user @-mentions the bot inside a Lark topic/thread, the bot now replies back into that thread rather than posting a fresh message at the chat level. Behavior is automatic and scoped: only triggers that were themselves inside a thread get a threaded reply, so normal group/p2p chats are unchanged. The outbound path is event-driven and decoupled from the inbound message, so the trigger message_id + thread_id are persisted on lark_chat_session_binding (migration 122) at ingest time. The patcher then routes the agent reply (text / markdown card / error card) and the OutcomeReplier notices (/issue confirmation, offline/archived) through Lark's reply endpoint with reply_in_thread=true when a thread is present, falling back to a chat-level send if the threaded reply fails. Co-authored-by: Cursor <cursoragent@cursor.com> * fix(lark): classify thread-reply failures before chat-level fallback Only retry a threaded reply at the chat level when Lark returns an explicit "this message/topic cannot receive a threaded reply" error (recalled trigger, topic gone, topics disabled, aggregated message, etc.). Transport errors, 5xx, timeouts, rate limits, and ambiguous failures are now logged and returned as failures instead of being retried, so we never duplicate a reply or leak a thread-only reply into the main group chat. The three reply-capable send methods now return a structured *APIError carrying the Lark business code, and isThreadReplyUnsupported drives the fallback via an allowlist. sendWithThreadFallback is promoted to a package-level function so the immediate OutcomeReplier sends (/issue confirmation, offline/archived notices) share the same classified fallback path instead of silently swallowing thread-reply failures. Co-authored-by: Cursor <cursoragent@cursor.com> --------- Co-authored-by: kun <kuen@micous.com> Co-authored-by: Cursor <cursoragent@cursor.com>
1176 lines
39 KiB
Go
1176 lines
39 KiB
Go
// 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
|
|
}
|