mirror of
https://github.com/multica-ai/multica.git
synced 2026-07-13 05:16:29 +02:00
* fix(lark): allow rebinding a revoked Feishu bot to a different agent When a Feishu/Lark Bot is disconnected from agent A (status → revoked), the row is preserved for audit but still holds the (channel_type, config->>app_id) unique index slot. Binding the same Bot to agent B would fail with: duplicate key value violates unique constraint "idx_channel_installation_type_appid" (SQLSTATE 23505) because UpsertChannelInstallation conflicts on (workspace_id, agent_id, channel_type) — a different agent_id means no conflict match, so it tries INSERT and hits the app_id unique index. Fix: before the upsert, inside the same transaction, hard-delete any revoked installation with the same app_id in the same workspace. The delete is fenced to status=revoked so an active installation can never be silently removed. If no revoked row exists the delete is a no-op (deletes zero rows, returns nil error) and the upsert proceeds normally. Co-Authored-By: Claude <noreply@anthropic.com> Co-authored-by: multica-agent <github@multica.ai> * fix(lark): preserve same-agent bindings when reconnecting a revoked Feishu bot The cleanup added in the previous commit hard-deletes every revoked channel_installation sharing the app_id in the workspace before the upsert — including the row belonging to the agent currently being (re)installed. That regresses the common "disconnect then reconnect the same bot to the same agent" flow: disconnect only flips status to 'revoked' (bindings are preserved), and UpsertChannelInstallation conflicts on (workspace_id, agent_id, channel_type), so before this the same agent's row was reactivated in place — installation_id and every channel_user_binding / channel_chat_session_binding kept. Deleting it first forces an INSERT with a fresh installation_id, orphaning every member's account link (they must re-link) and all chat-session continuity; only the installer is re-bound. Fence the delete with `agent_id <> $agent_id` so it only clears a DIFFERENT agent's revoked row (the genuine app_id-slot blocker). The same agent's revoked row is left for the upsert to reactivate losslessly. Since idx_channel_installation_type_appid is globally unique on (channel_type, app_id), at most one row ever holds a given app_id, so the excluded row is exactly the one the upsert will reuse. Adds DB-backed regression tests: same-agent revoked row preserved, different-agent revoked row deleted, active row never deleted, other workspace fenced, plus end-to-end reactivation semantics (same agent keeps installation_id + bindings; different agent gets a fresh id). Co-authored-by: multica-agent <github@multica.ai> * fix(lark): clean dependent rows when hard-deleting a rebound Feishu installation Addresses review on #4997 (MUL-4134). channel_* has no FK/cascade (MUL-3515 §4), so hard-deleting a different-agent revoked installation left application-owned rows dangling at a removed installation_id: - channel_chat_session_binding: the outbound patcher would resolve a binding, then fail loading the deleted installation — turning a clean no-op into error logs. - channel_binding_token: a still-unexpired bind link (15 min TTL) could be redeemed into the deleted installation, reporting "bound" against a bot that no longer reaches the user. - channel_inbound_audit: dangling installation_id, where migration 124 models the old ON DELETE SET NULL as an app-layer NULL. - channel_user_binding: dead member links (a different agent is a distinct connection; links do not follow and can never be reused). Rework RemoveRevokedInstallationByAppID to resolve the single row holding the app_id and act only when it is revoked, in this workspace, and owned by another agent; then, on the caller's transaction, clear chat-session bindings, pending binding tokens and member links, NULL the audit references, and finally delete the row via the fenced query (defense in depth). Same-agent reconnect and active/other-workspace rows are no-ops. Adds DeleteChannelUserBindingsByInstallation, DeleteChannelBindingTokensByInstallation, and NullChannelInboundAuditInstallationID queries, plus a DB-backed test (TestChannelStore_RebindCleansDependentRows) asserting every dependent is cleaned and the audit row survives detached. Verified the test fails when the cleanup is skipped. Co-authored-by: multica-agent <github@multica.ai> * fix(lark): make the rebind cleanup race-safe with a guarded delete gate Addresses the concurrency must-fix on #4997 (MUL-4134). The prior shape read the candidate installation, checked revoked/workspace/agent in Go, cleaned the dependent rows, then ran the fenced delete. That read-then- clean-then-delete order has a TOCTOU: while B is rebinding the bot to a different agent, A can reconnect to the SAME agent and reactivate the row to 'active' in between. B still wipes A's user/chat/token bindings and NULLs its audit based on the stale "it was revoked" read, then the fenced delete no-ops (status is no longer revoked) — so A's installation survives active but its bindings are gone. Concurrent same-agent data loss, reintroduced. Make the guarded DELETE the atomic gate. DeleteChannelInstallationByAppID becomes DeleteRevokedChannelInstallationByAppID `:one ... RETURNING id`, and RemoveRevokedInstallationByAppID keys all dependent cleanup off the id the delete actually claimed. No separate read. Under READ COMMITTED a concurrent reactivation makes the DELETE re-check status='revoked' against the live row (EvalPlanQual): it claims nothing, returns pgx.ErrNoRows, and no dependents are touched. With no FK the cleanup can follow the claiming delete in the same transaction; any failure rolls the whole thing back. Adds TestChannelStore_RebindGuardedDeleteRaceWithReactivation: two real transactions race on one revoked installation — one reactivates and holds the row lock, the other runs the rebind cleanup and blocks on the guarded delete — asserting the installation and every binding stay intact. Verified this test fails on the old read-then-clean-then-delete shape and passes (also under -race) on the gated version. Co-authored-by: multica-agent <github@multica.ai> --------- Co-authored-by: jiangliangyou <jiangliangyou@xiaomi.com> Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: multica-agent <github@multica.ai> Co-authored-by: J <j@multica.ai>
1339 lines
49 KiB
Go
1339 lines
49 KiB
Go
// Code generated by sqlc. DO NOT EDIT.
|
|
// versions:
|
|
// sqlc v1.31.1
|
|
// source: channel.sql
|
|
|
|
package db
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
)
|
|
|
|
const acquireChannelWSLease = `-- name: AcquireChannelWSLease :one
|
|
UPDATE channel_installation
|
|
SET ws_lease_token = $1,
|
|
ws_lease_expires_at = $2,
|
|
updated_at = now()
|
|
WHERE id = $3
|
|
AND status = 'active'
|
|
AND (
|
|
ws_lease_token IS NULL
|
|
OR ws_lease_expires_at < now()
|
|
OR ws_lease_token = $1
|
|
)
|
|
RETURNING id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at
|
|
`
|
|
|
|
type AcquireChannelWSLeaseParams struct {
|
|
NewToken pgtype.Text `json:"new_token"`
|
|
NewExpiresAt pgtype.Timestamptz `json:"new_expires_at"`
|
|
ID pgtype.UUID `json:"id"`
|
|
}
|
|
|
|
// Atomically claims the WebSocket lease. CAS predicate accepts when no
|
|
// holder exists, the holder expired, or the holder is us (renewal).
|
|
func (q *Queries) AcquireChannelWSLease(ctx context.Context, arg AcquireChannelWSLeaseParams) (ChannelInstallation, error) {
|
|
row := q.db.QueryRow(ctx, acquireChannelWSLease, arg.NewToken, arg.NewExpiresAt, arg.ID)
|
|
var i ChannelInstallation
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.AgentID,
|
|
&i.ChannelType,
|
|
&i.Config,
|
|
&i.Status,
|
|
&i.WsLeaseToken,
|
|
&i.WsLeaseExpiresAt,
|
|
&i.InstallerUserID,
|
|
&i.InstalledAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const backfillChannelInstallationRegionToFeishuLark = `-- name: BackfillChannelInstallationRegionToFeishuLark :execrows
|
|
UPDATE channel_installation
|
|
SET config = jsonb_set(config, '{region}', '"lark"'),
|
|
updated_at = now()
|
|
WHERE channel_type = 'feishu'
|
|
AND config ->> 'region' = 'feishu'
|
|
`
|
|
|
|
// Operator repair, feishu-only: flip every feishu installation still
|
|
// carrying region='feishu' to 'lark'. Called only on deployments whose
|
|
// legacy global base-URL override pointed at Lark international. Idempotent.
|
|
func (q *Queries) BackfillChannelInstallationRegionToFeishuLark(ctx context.Context) (int64, error) {
|
|
result, err := q.db.Exec(ctx, backfillChannelInstallationRegionToFeishuLark)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected(), nil
|
|
}
|
|
|
|
const claimChannelInboundDedup = `-- name: ClaimChannelInboundDedup :one
|
|
|
|
INSERT INTO channel_inbound_message_dedup (installation_id, message_id, claim_token)
|
|
VALUES ($1, $2, gen_random_uuid())
|
|
ON CONFLICT (installation_id, message_id) DO UPDATE
|
|
SET received_at = now(),
|
|
claim_token = gen_random_uuid()
|
|
WHERE channel_inbound_message_dedup.processed_at IS NULL
|
|
AND channel_inbound_message_dedup.received_at < now() - INTERVAL '60 seconds'
|
|
RETURNING installation_id, message_id, received_at, processed_at, claim_token
|
|
`
|
|
|
|
type ClaimChannelInboundDedupParams struct {
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
MessageID string `json:"message_id"`
|
|
}
|
|
|
|
// =====================
|
|
// channel_inbound_message_dedup
|
|
// =====================
|
|
// Two-phase idempotency gate with owner fencing. Returns the row when a
|
|
// claim is acquired (fresh insert, or stale-reclaim of an in-flight claim
|
|
// older than 60s); returns no rows when terminal (processed) or actively
|
|
// in-flight. Every claim mints a fresh claim_token; Mark/Release are
|
|
// fenced on it. See the table comment in migration 124 / the lark
|
|
// predecessor for the full invariant set.
|
|
func (q *Queries) ClaimChannelInboundDedup(ctx context.Context, arg ClaimChannelInboundDedupParams) (ChannelInboundMessageDedup, error) {
|
|
row := q.db.QueryRow(ctx, claimChannelInboundDedup, arg.InstallationID, arg.MessageID)
|
|
var i ChannelInboundMessageDedup
|
|
err := row.Scan(
|
|
&i.InstallationID,
|
|
&i.MessageID,
|
|
&i.ReceivedAt,
|
|
&i.ProcessedAt,
|
|
&i.ClaimToken,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const consumeChannelBindingToken = `-- name: ConsumeChannelBindingToken :one
|
|
UPDATE channel_binding_token
|
|
SET consumed_at = now()
|
|
WHERE token_hash = $1
|
|
AND consumed_at IS NULL
|
|
AND expires_at > now()
|
|
RETURNING token_hash, workspace_id, installation_id, channel_type, channel_user_id, expires_at, consumed_at, created_at
|
|
`
|
|
|
|
// Atomic redemption: returns the row only if the hash exists, is
|
|
// unconsumed, and unexpired. Two simultaneous redemptions cannot both win.
|
|
func (q *Queries) ConsumeChannelBindingToken(ctx context.Context, tokenHash string) (ChannelBindingToken, error) {
|
|
row := q.db.QueryRow(ctx, consumeChannelBindingToken, tokenHash)
|
|
var i ChannelBindingToken
|
|
err := row.Scan(
|
|
&i.TokenHash,
|
|
&i.WorkspaceID,
|
|
&i.InstallationID,
|
|
&i.ChannelType,
|
|
&i.ChannelUserID,
|
|
&i.ExpiresAt,
|
|
&i.ConsumedAt,
|
|
&i.CreatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const createChannelBindingToken = `-- name: CreateChannelBindingToken :one
|
|
|
|
INSERT INTO channel_binding_token (
|
|
token_hash, workspace_id, installation_id, channel_type,
|
|
channel_user_id, expires_at
|
|
) VALUES (
|
|
$1, $2, $3, $4, $5, $6
|
|
)
|
|
RETURNING token_hash, workspace_id, installation_id, channel_type, channel_user_id, expires_at, consumed_at, created_at
|
|
`
|
|
|
|
type CreateChannelBindingTokenParams struct {
|
|
TokenHash string `json:"token_hash"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
ChannelType string `json:"channel_type"`
|
|
ChannelUserID string `json:"channel_user_id"`
|
|
ExpiresAt pgtype.Timestamptz `json:"expires_at"`
|
|
}
|
|
|
|
// =====================
|
|
// channel_binding_token
|
|
// =====================
|
|
// Mints a single-use binding token for an unbound platform user. TTL cap
|
|
// (15 min) enforced by the table CHECK in lockstep with
|
|
// channel.BindingTokenTTL. The HASH is stored, never the raw token.
|
|
func (q *Queries) CreateChannelBindingToken(ctx context.Context, arg CreateChannelBindingTokenParams) (ChannelBindingToken, error) {
|
|
row := q.db.QueryRow(ctx, createChannelBindingToken,
|
|
arg.TokenHash,
|
|
arg.WorkspaceID,
|
|
arg.InstallationID,
|
|
arg.ChannelType,
|
|
arg.ChannelUserID,
|
|
arg.ExpiresAt,
|
|
)
|
|
var i ChannelBindingToken
|
|
err := row.Scan(
|
|
&i.TokenHash,
|
|
&i.WorkspaceID,
|
|
&i.InstallationID,
|
|
&i.ChannelType,
|
|
&i.ChannelUserID,
|
|
&i.ExpiresAt,
|
|
&i.ConsumedAt,
|
|
&i.CreatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const createChannelChatSessionBinding = `-- name: CreateChannelChatSessionBinding :one
|
|
|
|
INSERT INTO channel_chat_session_binding (
|
|
chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, config
|
|
) VALUES (
|
|
$1, $2, $3, $4, $5, $6
|
|
)
|
|
RETURNING id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at
|
|
`
|
|
|
|
type CreateChannelChatSessionBindingParams struct {
|
|
ChatSessionID pgtype.UUID `json:"chat_session_id"`
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
ChannelType string `json:"channel_type"`
|
|
ChannelChatID string `json:"channel_chat_id"`
|
|
ChatType string `json:"chat_type"`
|
|
Config []byte `json:"config"`
|
|
}
|
|
|
|
// =====================
|
|
// channel_chat_session_binding
|
|
// =====================
|
|
// channel_chat_id is the session-isolation key (one chat_session per
|
|
// (installation_id, channel_chat_id)): Feishu passes the chat id; Slack passes
|
|
// a stable key that, for channels, includes the thread root so each @bot thread
|
|
// is its own session. config carries any platform-specific outbound routing the
|
|
// key alone does not (e.g. Slack's real channel_id when the key is composite);
|
|
// it is opaque to the shared session service.
|
|
func (q *Queries) CreateChannelChatSessionBinding(ctx context.Context, arg CreateChannelChatSessionBindingParams) (ChannelChatSessionBinding, error) {
|
|
row := q.db.QueryRow(ctx, createChannelChatSessionBinding,
|
|
arg.ChatSessionID,
|
|
arg.InstallationID,
|
|
arg.ChannelType,
|
|
arg.ChannelChatID,
|
|
arg.ChatType,
|
|
arg.Config,
|
|
)
|
|
var i ChannelChatSessionBinding
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.ChatSessionID,
|
|
&i.InstallationID,
|
|
&i.ChannelType,
|
|
&i.ChannelChatID,
|
|
&i.ChatType,
|
|
&i.LastMessageID,
|
|
&i.LastThreadID,
|
|
&i.Config,
|
|
&i.CreatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const createChannelOutboundCardMessage = `-- name: CreateChannelOutboundCardMessage :one
|
|
|
|
INSERT INTO channel_outbound_card_message (
|
|
chat_session_id, task_id, channel_type, channel_chat_id,
|
|
channel_card_message_id, status
|
|
) VALUES (
|
|
$1, $6, $2, $3, $4, $5
|
|
)
|
|
RETURNING id, chat_session_id, task_id, channel_type, channel_chat_id, channel_card_message_id, status, last_patched_at, created_at
|
|
`
|
|
|
|
type CreateChannelOutboundCardMessageParams struct {
|
|
ChatSessionID pgtype.UUID `json:"chat_session_id"`
|
|
ChannelType string `json:"channel_type"`
|
|
ChannelChatID string `json:"channel_chat_id"`
|
|
ChannelCardMessageID string `json:"channel_card_message_id"`
|
|
Status string `json:"status"`
|
|
TaskID pgtype.UUID `json:"task_id"`
|
|
}
|
|
|
|
// =====================
|
|
// channel_outbound_card_message
|
|
// =====================
|
|
func (q *Queries) CreateChannelOutboundCardMessage(ctx context.Context, arg CreateChannelOutboundCardMessageParams) (ChannelOutboundCardMessage, error) {
|
|
row := q.db.QueryRow(ctx, createChannelOutboundCardMessage,
|
|
arg.ChatSessionID,
|
|
arg.ChannelType,
|
|
arg.ChannelChatID,
|
|
arg.ChannelCardMessageID,
|
|
arg.Status,
|
|
arg.TaskID,
|
|
)
|
|
var i ChannelOutboundCardMessage
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.ChatSessionID,
|
|
&i.TaskID,
|
|
&i.ChannelType,
|
|
&i.ChannelChatID,
|
|
&i.ChannelCardMessageID,
|
|
&i.Status,
|
|
&i.LastPatchedAt,
|
|
&i.CreatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const createChannelUserBinding = `-- name: CreateChannelUserBinding :one
|
|
|
|
INSERT INTO channel_user_binding (
|
|
workspace_id, multica_user_id, installation_id,
|
|
channel_type, channel_user_id, config
|
|
) VALUES (
|
|
$1, $2, $3, $4, $5, $6
|
|
)
|
|
ON CONFLICT (installation_id, channel_user_id) DO UPDATE SET
|
|
-- jsonb_strip_nulls(EXCLUDED.config) preserves the old lark semantics
|
|
-- ` + "`" + `union_id = COALESCE(EXCLUDED.union_id, lark_user_binding.union_id)` + "`" + `:
|
|
-- a re-bind that carries ` + "`" + `{"union_id": null}` + "`" + ` (or omits the key) must NOT
|
|
-- erase a union_id we already captured. Only non-null incoming keys win.
|
|
config = channel_user_binding.config || jsonb_strip_nulls(EXCLUDED.config),
|
|
bound_at = now()
|
|
WHERE channel_user_binding.multica_user_id = EXCLUDED.multica_user_id
|
|
RETURNING id, workspace_id, multica_user_id, installation_id, channel_type, channel_user_id, config, bound_at
|
|
`
|
|
|
|
type CreateChannelUserBindingParams struct {
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
MulticaUserID pgtype.UUID `json:"multica_user_id"`
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
ChannelType string `json:"channel_type"`
|
|
ChannelUserID string `json:"channel_user_id"`
|
|
Config []byte `json:"config"`
|
|
}
|
|
|
|
// =====================
|
|
// channel_user_binding
|
|
// =====================
|
|
// Records that a platform user id (per-installation; Feishu open_id) maps
|
|
// to a Multica user. The old composite member-FK is gone, so this no
|
|
// longer fails when the redeemer is not a workspace member — the caller
|
|
// (BindingTokenService.RedeemAndBind) validates membership explicitly
|
|
// before calling. ON CONFLICT DO UPDATE is still gated on multica_user_id
|
|
// matching, so a second redeemer cannot steal an already-bound user id;
|
|
// a cross-user conflict updates zero rows and the caller maps that to
|
|
// ErrBindingAlreadyAssigned. config carries secondary identity (union_id).
|
|
func (q *Queries) CreateChannelUserBinding(ctx context.Context, arg CreateChannelUserBindingParams) (ChannelUserBinding, error) {
|
|
row := q.db.QueryRow(ctx, createChannelUserBinding,
|
|
arg.WorkspaceID,
|
|
arg.MulticaUserID,
|
|
arg.InstallationID,
|
|
arg.ChannelType,
|
|
arg.ChannelUserID,
|
|
arg.Config,
|
|
)
|
|
var i ChannelUserBinding
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.MulticaUserID,
|
|
&i.InstallationID,
|
|
&i.ChannelType,
|
|
&i.ChannelUserID,
|
|
&i.Config,
|
|
&i.BoundAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const deleteChannelBindingTokensByInstallation = `-- name: DeleteChannelBindingTokensByInstallation :exec
|
|
DELETE FROM channel_binding_token
|
|
WHERE installation_id = $1
|
|
`
|
|
|
|
// Application-layer integrity (schema has no FK/cascade, MUL-3515 §4): drop
|
|
// every pending binding token for an installation that is being hard-deleted.
|
|
// A token stays redeemable for up to 15 min; without this a user who clicks a
|
|
// still-unexpired bind link right after the bot was rebound to another agent
|
|
// would consume the token and get a "bound" result written against a deleted
|
|
// installation — a link that never actually reaches the live bot.
|
|
func (q *Queries) DeleteChannelBindingTokensByInstallation(ctx context.Context, installationID pgtype.UUID) error {
|
|
_, err := q.db.Exec(ctx, deleteChannelBindingTokensByInstallation, installationID)
|
|
return err
|
|
}
|
|
|
|
const deleteChannelChatSessionBindingBySession = `-- name: DeleteChannelChatSessionBindingBySession :exec
|
|
DELETE FROM channel_chat_session_binding
|
|
WHERE chat_session_id = $1
|
|
`
|
|
|
|
// Application-layer integrity (replaces the old chat_session-FK ON DELETE
|
|
// CASCADE): drop the binding when its chat_session is deleted.
|
|
func (q *Queries) DeleteChannelChatSessionBindingBySession(ctx context.Context, chatSessionID pgtype.UUID) error {
|
|
_, err := q.db.Exec(ctx, deleteChannelChatSessionBindingBySession, chatSessionID)
|
|
return err
|
|
}
|
|
|
|
const deleteChannelChatSessionBindingsByInstallation = `-- name: DeleteChannelChatSessionBindingsByInstallation :exec
|
|
DELETE FROM channel_chat_session_binding
|
|
WHERE installation_id = $1 AND channel_type = $2
|
|
`
|
|
|
|
type DeleteChannelChatSessionBindingsByInstallationParams struct {
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
ChannelType string `json:"channel_type"`
|
|
}
|
|
|
|
// Retire every chat-session binding for an installation. Used when an
|
|
// installation is re-pointed to a different agent (Slack re-connect): each
|
|
// existing chat_session is permanently tied to the agent it was created under,
|
|
// so reusing it would keep routing the conversation to the OLD agent. Dropping
|
|
// the bindings forces the next inbound message to create a fresh session under
|
|
// the new agent. The chat_session rows are preserved for history; only the
|
|
// channel binding is removed.
|
|
func (q *Queries) DeleteChannelChatSessionBindingsByInstallation(ctx context.Context, arg DeleteChannelChatSessionBindingsByInstallationParams) error {
|
|
_, err := q.db.Exec(ctx, deleteChannelChatSessionBindingsByInstallation, arg.InstallationID, arg.ChannelType)
|
|
return err
|
|
}
|
|
|
|
const deleteChannelUserBindingsByInstallation = `-- name: DeleteChannelUserBindingsByInstallation :exec
|
|
DELETE FROM channel_user_binding
|
|
WHERE installation_id = $1
|
|
`
|
|
|
|
// Application-layer integrity (schema has no FK/cascade, MUL-3515 §4): drop
|
|
// every member account link for an installation that is being hard-deleted.
|
|
// Rebinding a Feishu bot to a DIFFERENT agent starts a fresh installation, so
|
|
// old links do not follow — a different agent is a distinct connection and
|
|
// members re-establish their link on first contact. The rows could never be
|
|
// reused anyway (every Feishu identity lookup is installation_id-scoped, and
|
|
// FindReusableChannelUserBinding is Slack-only), so removing them just keeps
|
|
// dead rows from accumulating.
|
|
func (q *Queries) DeleteChannelUserBindingsByInstallation(ctx context.Context, installationID pgtype.UUID) error {
|
|
_, err := q.db.Exec(ctx, deleteChannelUserBindingsByInstallation, installationID)
|
|
return err
|
|
}
|
|
|
|
const deleteChannelUserBindingsByWorkspaceMember = `-- name: DeleteChannelUserBindingsByWorkspaceMember :exec
|
|
DELETE FROM channel_user_binding
|
|
WHERE workspace_id = $1 AND multica_user_id = $2
|
|
`
|
|
|
|
type DeleteChannelUserBindingsByWorkspaceMemberParams struct {
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
MulticaUserID pgtype.UUID `json:"multica_user_id"`
|
|
}
|
|
|
|
// Application-layer integrity (replaces the old member-FK ON DELETE
|
|
// CASCADE): prune every binding for a user who has been removed from a
|
|
// workspace, across all installations in that workspace.
|
|
func (q *Queries) DeleteChannelUserBindingsByWorkspaceMember(ctx context.Context, arg DeleteChannelUserBindingsByWorkspaceMemberParams) error {
|
|
_, err := q.db.Exec(ctx, deleteChannelUserBindingsByWorkspaceMember, arg.WorkspaceID, arg.MulticaUserID)
|
|
return err
|
|
}
|
|
|
|
const deleteRevokedChannelInstallationByAppID = `-- name: DeleteRevokedChannelInstallationByAppID :one
|
|
DELETE FROM channel_installation
|
|
WHERE channel_type = $1
|
|
AND config ->> 'app_id' = $2::text
|
|
AND workspace_id = $3
|
|
AND agent_id <> $4
|
|
AND status = 'revoked'
|
|
RETURNING id
|
|
`
|
|
|
|
type DeleteRevokedChannelInstallationByAppIDParams struct {
|
|
ChannelType string `json:"channel_type"`
|
|
AppID string `json:"app_id"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
}
|
|
|
|
// Hard-delete a REVOKED installation that belongs to a DIFFERENT agent, keyed
|
|
// by its platform app identity, and RETURN the deleted id. When a Feishu/Lark
|
|
// Bot is disconnected from agent A (status → 'revoked') and the same Bot (same
|
|
// app_id) is later bound to a DIFFERENT agent B, agent A's revoked row still
|
|
// occupies the (channel_type, config->>'app_id') unique slot and blocks the
|
|
// UpsertChannelInstallation INSERT for B. Removing that placeholder first lets
|
|
// the upsert create a fresh row for the new agent.
|
|
//
|
|
// The `agent_id <> arg` fence is load-bearing: re-connecting the SAME agent
|
|
// must NOT delete its own revoked row. UpsertChannelInstallation conflicts on
|
|
// (workspace_id, agent_id, channel_type), so the same agent's revoked row is
|
|
// reactivated in place (status → 'active') with its installation_id — and every
|
|
// channel_user_binding / channel_chat_session_binding hanging off it —
|
|
// preserved. Deleting it here would force an INSERT with a fresh installation_id
|
|
// and orphan all of that agent's member links and chat sessions.
|
|
//
|
|
// Fenced to one workspace AND status='revoked' so an active installation can
|
|
// never be silently deleted through this path.
|
|
//
|
|
// This DELETE is the atomic gate for the whole rebind cleanup: the caller keys
|
|
// its dependent-row cleanup off the RETURNING id, so cleanup runs ONLY for a row
|
|
// this statement actually removed. Under READ COMMITTED, a concurrent same-agent
|
|
// reconnect that reactivates the row to 'active' first makes the WHERE re-check
|
|
// fail (EvalPlanQual), this deletes nothing (pgx.ErrNoRows), and no dependents
|
|
// are touched — closing the read-then-delete TOCTOU where a stale "it was
|
|
// revoked" read could wipe the bindings of a since-reactivated installation.
|
|
func (q *Queries) DeleteRevokedChannelInstallationByAppID(ctx context.Context, arg DeleteRevokedChannelInstallationByAppIDParams) (pgtype.UUID, error) {
|
|
row := q.db.QueryRow(ctx, deleteRevokedChannelInstallationByAppID,
|
|
arg.ChannelType,
|
|
arg.AppID,
|
|
arg.WorkspaceID,
|
|
arg.AgentID,
|
|
)
|
|
var id pgtype.UUID
|
|
err := row.Scan(&id)
|
|
return id, err
|
|
}
|
|
|
|
const findReusableChannelUserBinding = `-- name: FindReusableChannelUserBinding :one
|
|
SELECT b.id, b.workspace_id, b.multica_user_id, b.installation_id, b.channel_type, b.channel_user_id, b.config, b.bound_at FROM channel_user_binding b
|
|
JOIN channel_installation ci ON ci.id = b.installation_id
|
|
WHERE b.workspace_id = $1
|
|
AND b.channel_type = $2
|
|
AND b.channel_user_id = $3
|
|
AND ci.config ->> 'team_id' = $4::text
|
|
ORDER BY b.bound_at DESC
|
|
LIMIT 1
|
|
`
|
|
|
|
type FindReusableChannelUserBindingParams struct {
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
ChannelType string `json:"channel_type"`
|
|
ChannelUserID string `json:"channel_user_id"`
|
|
TeamID string `json:"team_id"`
|
|
}
|
|
|
|
// Cross-installation account-link reuse (MUL-3911). When a platform user
|
|
// messages an installation they have NOT linked, but the SAME user id is already
|
|
// bound to ANOTHER installation in the SAME Multica workspace + SAME Slack team,
|
|
// the inbound identity step reuses that link instead of re-prompting. Slack user
|
|
// ids are stable within a team, so an identical channel_user_id denotes the same
|
|
// human across that team's apps. The match is fenced to one workspace AND one
|
|
// team (installation config->>'team_id'): a Slack team can be connected to two
|
|
// different Multica workspaces, and a user may hold different Multica accounts in
|
|
// each, so reuse must cross neither boundary. Most-recently-bound wins. The
|
|
// caller re-checks membership and materializes a fresh per-installation binding.
|
|
//
|
|
// team_id is pinned ::text so sqlc types the arg as a string instead of
|
|
// attributing the bare param to the JSONB config column (mirrors
|
|
// GetChannelInstallationByAppID's app_id cast).
|
|
func (q *Queries) FindReusableChannelUserBinding(ctx context.Context, arg FindReusableChannelUserBindingParams) (ChannelUserBinding, error) {
|
|
row := q.db.QueryRow(ctx, findReusableChannelUserBinding,
|
|
arg.WorkspaceID,
|
|
arg.ChannelType,
|
|
arg.ChannelUserID,
|
|
arg.TeamID,
|
|
)
|
|
var i ChannelUserBinding
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.MulticaUserID,
|
|
&i.InstallationID,
|
|
&i.ChannelType,
|
|
&i.ChannelUserID,
|
|
&i.Config,
|
|
&i.BoundAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const getChannelChatSessionBinding = `-- name: GetChannelChatSessionBinding :one
|
|
SELECT id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at FROM channel_chat_session_binding
|
|
WHERE installation_id = $1 AND channel_chat_id = $2
|
|
`
|
|
|
|
type GetChannelChatSessionBindingParams struct {
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
ChannelChatID string `json:"channel_chat_id"`
|
|
}
|
|
|
|
// Lookup-by-channel-chat: the inbound dispatcher finds the existing
|
|
// chat_session before deciding whether to create one.
|
|
func (q *Queries) GetChannelChatSessionBinding(ctx context.Context, arg GetChannelChatSessionBindingParams) (ChannelChatSessionBinding, error) {
|
|
row := q.db.QueryRow(ctx, getChannelChatSessionBinding, arg.InstallationID, arg.ChannelChatID)
|
|
var i ChannelChatSessionBinding
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.ChatSessionID,
|
|
&i.InstallationID,
|
|
&i.ChannelType,
|
|
&i.ChannelChatID,
|
|
&i.ChatType,
|
|
&i.LastMessageID,
|
|
&i.LastThreadID,
|
|
&i.Config,
|
|
&i.CreatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const getChannelChatSessionBindingBySession = `-- name: GetChannelChatSessionBindingBySession :one
|
|
SELECT id, chat_session_id, installation_id, channel_type, channel_chat_id, chat_type, last_message_id, last_thread_id, config, created_at FROM channel_chat_session_binding
|
|
WHERE chat_session_id = $1
|
|
AND channel_type = $2
|
|
`
|
|
|
|
type GetChannelChatSessionBindingBySessionParams struct {
|
|
ChatSessionID pgtype.UUID `json:"chat_session_id"`
|
|
ChannelType string `json:"channel_type"`
|
|
}
|
|
|
|
// Reverse lookup for the outbound patcher: given a chat_session_id, find
|
|
// its channel binding to know which (installation, chat_id) to send to.
|
|
// Scoped by channel_type so a future non-Feishu binding on the same
|
|
// chat_session is never treated as a Feishu reply target.
|
|
func (q *Queries) GetChannelChatSessionBindingBySession(ctx context.Context, arg GetChannelChatSessionBindingBySessionParams) (ChannelChatSessionBinding, error) {
|
|
row := q.db.QueryRow(ctx, getChannelChatSessionBindingBySession, arg.ChatSessionID, arg.ChannelType)
|
|
var i ChannelChatSessionBinding
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.ChatSessionID,
|
|
&i.InstallationID,
|
|
&i.ChannelType,
|
|
&i.ChannelChatID,
|
|
&i.ChatType,
|
|
&i.LastMessageID,
|
|
&i.LastThreadID,
|
|
&i.Config,
|
|
&i.CreatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const getChannelInstallation = `-- name: GetChannelInstallation :one
|
|
SELECT id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at FROM channel_installation
|
|
WHERE id = $1 AND channel_type = $2
|
|
`
|
|
|
|
type GetChannelInstallationParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
ChannelType string `json:"channel_type"`
|
|
}
|
|
|
|
// Scoped by channel_type: a per-channel caller (e.g. the Feishu store)
|
|
// must never resolve another channel's installation by guessing its UUID.
|
|
func (q *Queries) GetChannelInstallation(ctx context.Context, arg GetChannelInstallationParams) (ChannelInstallation, error) {
|
|
row := q.db.QueryRow(ctx, getChannelInstallation, arg.ID, arg.ChannelType)
|
|
var i ChannelInstallation
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.AgentID,
|
|
&i.ChannelType,
|
|
&i.Config,
|
|
&i.Status,
|
|
&i.WsLeaseToken,
|
|
&i.WsLeaseExpiresAt,
|
|
&i.InstallerUserID,
|
|
&i.InstalledAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const getChannelInstallationByAppID = `-- name: GetChannelInstallationByAppID :one
|
|
SELECT id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at FROM channel_installation
|
|
WHERE channel_type = $1
|
|
AND config ->> 'app_id' = $2::text
|
|
`
|
|
|
|
type GetChannelInstallationByAppIDParams struct {
|
|
ChannelType string `json:"channel_type"`
|
|
AppID string `json:"app_id"`
|
|
}
|
|
|
|
// Inbound routing. The platform event carries only the channel's app
|
|
// identifier (Feishu app_id); the dispatcher's installation resolver routes
|
|
// on (channel_type, config->>'app_id'). Backed by the functional unique
|
|
// index idx_channel_installation_type_appid.
|
|
//
|
|
// Both params are named + explicitly typed: `config ->> 'app_id'` makes sqlc
|
|
// attribute a bare `$2` to the JSONB `config` column (it would emit
|
|
// `Config []byte`), so we pin the app_id arg to ::text to get AppID string.
|
|
func (q *Queries) GetChannelInstallationByAppID(ctx context.Context, arg GetChannelInstallationByAppIDParams) (ChannelInstallation, error) {
|
|
row := q.db.QueryRow(ctx, getChannelInstallationByAppID, arg.ChannelType, arg.AppID)
|
|
var i ChannelInstallation
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.AgentID,
|
|
&i.ChannelType,
|
|
&i.Config,
|
|
&i.Status,
|
|
&i.WsLeaseToken,
|
|
&i.WsLeaseExpiresAt,
|
|
&i.InstallerUserID,
|
|
&i.InstalledAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const getChannelInstallationInWorkspace = `-- name: GetChannelInstallationInWorkspace :one
|
|
SELECT id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at FROM channel_installation
|
|
WHERE id = $1
|
|
AND workspace_id = $2
|
|
AND channel_type = $3
|
|
`
|
|
|
|
type GetChannelInstallationInWorkspaceParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
ChannelType string `json:"channel_type"`
|
|
}
|
|
|
|
func (q *Queries) GetChannelInstallationInWorkspace(ctx context.Context, arg GetChannelInstallationInWorkspaceParams) (ChannelInstallation, error) {
|
|
row := q.db.QueryRow(ctx, getChannelInstallationInWorkspace, arg.ID, arg.WorkspaceID, arg.ChannelType)
|
|
var i ChannelInstallation
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.AgentID,
|
|
&i.ChannelType,
|
|
&i.Config,
|
|
&i.Status,
|
|
&i.WsLeaseToken,
|
|
&i.WsLeaseExpiresAt,
|
|
&i.InstallerUserID,
|
|
&i.InstalledAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const getChannelOutboundCardByTask = `-- name: GetChannelOutboundCardByTask :one
|
|
SELECT id, chat_session_id, task_id, channel_type, channel_chat_id, channel_card_message_id, status, last_patched_at, created_at FROM channel_outbound_card_message
|
|
WHERE task_id = $1
|
|
AND channel_type = $2
|
|
`
|
|
|
|
type GetChannelOutboundCardByTaskParams struct {
|
|
TaskID pgtype.UUID `json:"task_id"`
|
|
ChannelType string `json:"channel_type"`
|
|
}
|
|
|
|
// The partial unique index on (task_id) WHERE task_id IS NOT NULL
|
|
// guarantees at most one row. Scoped by channel_type so a future non-Feishu
|
|
// card for the same task is not patched as a Feishu card.
|
|
func (q *Queries) GetChannelOutboundCardByTask(ctx context.Context, arg GetChannelOutboundCardByTaskParams) (ChannelOutboundCardMessage, error) {
|
|
row := q.db.QueryRow(ctx, getChannelOutboundCardByTask, arg.TaskID, arg.ChannelType)
|
|
var i ChannelOutboundCardMessage
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.ChatSessionID,
|
|
&i.TaskID,
|
|
&i.ChannelType,
|
|
&i.ChannelChatID,
|
|
&i.ChannelCardMessageID,
|
|
&i.Status,
|
|
&i.LastPatchedAt,
|
|
&i.CreatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const getChannelUserBindingByUserID = `-- name: GetChannelUserBindingByUserID :one
|
|
SELECT id, workspace_id, multica_user_id, installation_id, channel_type, channel_user_id, config, bound_at FROM channel_user_binding
|
|
WHERE installation_id = $1 AND channel_user_id = $2
|
|
`
|
|
|
|
type GetChannelUserBindingByUserIDParams struct {
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
ChannelUserID string `json:"channel_user_id"`
|
|
}
|
|
|
|
// The inbound identity lookup: does this platform user id map to a Multica
|
|
// user for this installation? With the member-FK removed, a row's
|
|
// existence no longer proves current workspace membership — the dispatcher
|
|
// re-checks membership after this lookup.
|
|
func (q *Queries) GetChannelUserBindingByUserID(ctx context.Context, arg GetChannelUserBindingByUserIDParams) (ChannelUserBinding, error) {
|
|
row := q.db.QueryRow(ctx, getChannelUserBindingByUserID, arg.InstallationID, arg.ChannelUserID)
|
|
var i ChannelUserBinding
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.MulticaUserID,
|
|
&i.InstallationID,
|
|
&i.ChannelType,
|
|
&i.ChannelUserID,
|
|
&i.Config,
|
|
&i.BoundAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const listActiveChannelInstallations = `-- name: ListActiveChannelInstallations :many
|
|
SELECT ci.id, ci.workspace_id, ci.agent_id, ci.channel_type, ci.config, ci.status, ci.ws_lease_token, ci.ws_lease_expires_at, ci.installer_user_id, ci.installed_at, ci.created_at, ci.updated_at FROM channel_installation ci
|
|
JOIN workspace w ON w.id = ci.workspace_id
|
|
JOIN agent a ON a.id = ci.agent_id
|
|
WHERE ci.status = 'active'
|
|
AND ci.channel_type = $1
|
|
ORDER BY ci.created_at ASC
|
|
`
|
|
|
|
// Boot path for a per-channel-type inbound hub: every active installation of
|
|
// the given channel_type, so a hub claims leases and opens connections only
|
|
// for its own platform and never supervises another channel's installation.
|
|
//
|
|
// The JOINs require the owning workspace and agent rows to still exist.
|
|
// channel_installation has no FK (MUL-3515 §4), so unlike the old
|
|
// lark_installation (which cascaded away on workspace/agent deletion) an
|
|
// installation can be orphaned when its workspace is deleted or its agent is
|
|
// hard-deleted (e.g. runtime teardown). Without this guard the hub would keep
|
|
// opening a WebSocket for a bot whose workspace/agent is gone. The JOIN matches
|
|
// the old ON DELETE CASCADE semantics: it filters on row existence, not agent
|
|
// archival, so an archived-but-present agent's installation is still listed.
|
|
func (q *Queries) ListActiveChannelInstallations(ctx context.Context, channelType string) ([]ChannelInstallation, error) {
|
|
rows, err := q.db.Query(ctx, listActiveChannelInstallations, channelType)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ChannelInstallation{}
|
|
for rows.Next() {
|
|
var i ChannelInstallation
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.AgentID,
|
|
&i.ChannelType,
|
|
&i.Config,
|
|
&i.Status,
|
|
&i.WsLeaseToken,
|
|
&i.WsLeaseExpiresAt,
|
|
&i.InstallerUserID,
|
|
&i.InstalledAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listAllActiveChannelInstallations = `-- name: ListAllActiveChannelInstallations :many
|
|
SELECT ci.id, ci.workspace_id, ci.agent_id, ci.channel_type, ci.config, ci.status, ci.ws_lease_token, ci.ws_lease_expires_at, ci.installer_user_id, ci.installed_at, ci.created_at, ci.updated_at FROM channel_installation ci
|
|
JOIN workspace w ON w.id = ci.workspace_id
|
|
JOIN agent a ON a.id = ci.agent_id
|
|
WHERE ci.status = 'active'
|
|
ORDER BY ci.created_at ASC
|
|
`
|
|
|
|
// Boot path for the channel-agnostic engine Supervisor (MUL-3620): every
|
|
// active installation across ALL channel types, so one Supervisor drives every
|
|
// platform's connections rather than a per-platform hub. This is the de-
|
|
// hardcoded counterpart of ListActiveChannelInstallations — the Supervisor
|
|
// routes each row to its registered channel.Factory by channel_type, so it
|
|
// never needs to know which platforms exist. Same orphan guard as the per-type
|
|
// query: the workspace + agent JOINs drop installations whose owning rows are
|
|
// gone (channel_installation has no FK, MUL-3515 §4), matching the old ON
|
|
// DELETE CASCADE semantics (row existence, not agent archival).
|
|
func (q *Queries) ListAllActiveChannelInstallations(ctx context.Context) ([]ChannelInstallation, error) {
|
|
rows, err := q.db.Query(ctx, listAllActiveChannelInstallations)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ChannelInstallation{}
|
|
for rows.Next() {
|
|
var i ChannelInstallation
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.AgentID,
|
|
&i.ChannelType,
|
|
&i.Config,
|
|
&i.Status,
|
|
&i.WsLeaseToken,
|
|
&i.WsLeaseExpiresAt,
|
|
&i.InstallerUserID,
|
|
&i.InstalledAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listChannelInboundAuditByInstallation = `-- name: ListChannelInboundAuditByInstallation :many
|
|
SELECT id, installation_id, channel_type, channel_chat_id, event_type, channel_event_id, channel_message_id, drop_reason, received_at FROM channel_inbound_audit
|
|
WHERE installation_id = $1
|
|
ORDER BY received_at DESC
|
|
LIMIT $2 OFFSET $3
|
|
`
|
|
|
|
type ListChannelInboundAuditByInstallationParams struct {
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
Limit int32 `json:"limit"`
|
|
Offset int32 `json:"offset"`
|
|
}
|
|
|
|
func (q *Queries) ListChannelInboundAuditByInstallation(ctx context.Context, arg ListChannelInboundAuditByInstallationParams) ([]ChannelInboundAudit, error) {
|
|
rows, err := q.db.Query(ctx, listChannelInboundAuditByInstallation, arg.InstallationID, arg.Limit, arg.Offset)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ChannelInboundAudit{}
|
|
for rows.Next() {
|
|
var i ChannelInboundAudit
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.InstallationID,
|
|
&i.ChannelType,
|
|
&i.ChannelChatID,
|
|
&i.EventType,
|
|
&i.ChannelEventID,
|
|
&i.ChannelMessageID,
|
|
&i.DropReason,
|
|
&i.ReceivedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listChannelInstallationsByWorkspace = `-- name: ListChannelInstallationsByWorkspace :many
|
|
SELECT id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at FROM channel_installation
|
|
WHERE workspace_id = $1
|
|
AND channel_type = $2
|
|
ORDER BY created_at ASC
|
|
`
|
|
|
|
type ListChannelInstallationsByWorkspaceParams struct {
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
ChannelType string `json:"channel_type"`
|
|
}
|
|
|
|
// Scoped by channel_type so a per-channel management surface (e.g. the Lark
|
|
// installation list) only ever sees its own platform's installations.
|
|
func (q *Queries) ListChannelInstallationsByWorkspace(ctx context.Context, arg ListChannelInstallationsByWorkspaceParams) ([]ChannelInstallation, error) {
|
|
rows, err := q.db.Query(ctx, listChannelInstallationsByWorkspace, arg.WorkspaceID, arg.ChannelType)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []ChannelInstallation{}
|
|
for rows.Next() {
|
|
var i ChannelInstallation
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.AgentID,
|
|
&i.ChannelType,
|
|
&i.Config,
|
|
&i.Status,
|
|
&i.WsLeaseToken,
|
|
&i.WsLeaseExpiresAt,
|
|
&i.InstallerUserID,
|
|
&i.InstalledAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, i)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const markChannelInboundDedupProcessed = `-- name: MarkChannelInboundDedupProcessed :execrows
|
|
UPDATE channel_inbound_message_dedup
|
|
SET processed_at = now()
|
|
WHERE installation_id = $1
|
|
AND message_id = $2
|
|
AND claim_token = $3
|
|
AND processed_at IS NULL
|
|
`
|
|
|
|
type MarkChannelInboundDedupProcessedParams struct {
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
MessageID string `json:"message_id"`
|
|
ClaimToken pgtype.UUID `json:"claim_token"`
|
|
}
|
|
|
|
// Locks a claim in as permanently processed after a durable outcome.
|
|
// Invoked inside the chat_message tx (via qtx) on the ingest path so the
|
|
// durable write and the Mark commit atomically. Token mismatch returns
|
|
// zero rows (a reclaim happened); the caller rolls back its in-tx write.
|
|
func (q *Queries) MarkChannelInboundDedupProcessed(ctx context.Context, arg MarkChannelInboundDedupProcessedParams) (int64, error) {
|
|
result, err := q.db.Exec(ctx, markChannelInboundDedupProcessed, arg.InstallationID, arg.MessageID, arg.ClaimToken)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected(), nil
|
|
}
|
|
|
|
const nullChannelInboundAuditInstallationID = `-- name: NullChannelInboundAuditInstallationID :exec
|
|
UPDATE channel_inbound_audit
|
|
SET installation_id = NULL
|
|
WHERE installation_id = $1
|
|
`
|
|
|
|
// Application-layer stand-in for the old ON DELETE SET NULL (MUL-3515 §4,
|
|
// migration 124 keeps installation_id nullable for exactly this): before an
|
|
// installation row is hard-deleted, detach its inbound-audit rows by NULLing
|
|
// installation_id. The drop-audit history is preserved (channel_type,
|
|
// chat/message ids, drop_reason stay) without a dangling reference to a
|
|
// removed installation.
|
|
func (q *Queries) NullChannelInboundAuditInstallationID(ctx context.Context, installationID pgtype.UUID) error {
|
|
_, err := q.db.Exec(ctx, nullChannelInboundAuditInstallationID, installationID)
|
|
return err
|
|
}
|
|
|
|
const purgeChannelInboundDedup = `-- name: PurgeChannelInboundDedup :exec
|
|
DELETE FROM channel_inbound_message_dedup
|
|
WHERE received_at < $1
|
|
`
|
|
|
|
// Vacuum job: remove dedup rows older than the supplied cutoff (e.g. 24h).
|
|
func (q *Queries) PurgeChannelInboundDedup(ctx context.Context, receivedAt pgtype.Timestamptz) error {
|
|
_, err := q.db.Exec(ctx, purgeChannelInboundDedup, receivedAt)
|
|
return err
|
|
}
|
|
|
|
const purgeExpiredChannelBindingTokens = `-- name: PurgeExpiredChannelBindingTokens :exec
|
|
DELETE FROM channel_binding_token
|
|
WHERE expires_at < $1
|
|
`
|
|
|
|
func (q *Queries) PurgeExpiredChannelBindingTokens(ctx context.Context, expiresAt pgtype.Timestamptz) error {
|
|
_, err := q.db.Exec(ctx, purgeExpiredChannelBindingTokens, expiresAt)
|
|
return err
|
|
}
|
|
|
|
const recordChannelInboundDrop = `-- name: RecordChannelInboundDrop :exec
|
|
|
|
INSERT INTO channel_inbound_audit (
|
|
installation_id, channel_type, channel_chat_id, event_type,
|
|
channel_event_id, channel_message_id, drop_reason
|
|
) VALUES (
|
|
$4,
|
|
$1,
|
|
$5,
|
|
$2,
|
|
$6,
|
|
$7,
|
|
$3
|
|
)
|
|
`
|
|
|
|
type RecordChannelInboundDropParams struct {
|
|
ChannelType string `json:"channel_type"`
|
|
EventType string `json:"event_type"`
|
|
DropReason string `json:"drop_reason"`
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
ChannelChatID pgtype.Text `json:"channel_chat_id"`
|
|
ChannelEventID pgtype.Text `json:"channel_event_id"`
|
|
ChannelMessageID pgtype.Text `json:"channel_message_id"`
|
|
}
|
|
|
|
// =====================
|
|
// channel_inbound_audit
|
|
// =====================
|
|
// The only write path for dropped events. Deliberately carries no body
|
|
// column — only routing / identity / drop_reason / timestamp.
|
|
func (q *Queries) RecordChannelInboundDrop(ctx context.Context, arg RecordChannelInboundDropParams) error {
|
|
_, err := q.db.Exec(ctx, recordChannelInboundDrop,
|
|
arg.ChannelType,
|
|
arg.EventType,
|
|
arg.DropReason,
|
|
arg.InstallationID,
|
|
arg.ChannelChatID,
|
|
arg.ChannelEventID,
|
|
arg.ChannelMessageID,
|
|
)
|
|
return err
|
|
}
|
|
|
|
const releaseChannelInboundDedup = `-- name: ReleaseChannelInboundDedup :execrows
|
|
DELETE FROM channel_inbound_message_dedup
|
|
WHERE installation_id = $1
|
|
AND message_id = $2
|
|
AND claim_token = $3
|
|
AND processed_at IS NULL
|
|
`
|
|
|
|
type ReleaseChannelInboundDedupParams struct {
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
MessageID string `json:"message_id"`
|
|
ClaimToken pgtype.UUID `json:"claim_token"`
|
|
}
|
|
|
|
// Releases an in-flight claim when an infra error occurred before any
|
|
// durable side effect, so a retry can re-acquire immediately. Fenced on
|
|
// processed_at IS NULL and claim_token.
|
|
func (q *Queries) ReleaseChannelInboundDedup(ctx context.Context, arg ReleaseChannelInboundDedupParams) (int64, error) {
|
|
result, err := q.db.Exec(ctx, releaseChannelInboundDedup, arg.InstallationID, arg.MessageID, arg.ClaimToken)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected(), nil
|
|
}
|
|
|
|
const releaseChannelWSLease = `-- name: ReleaseChannelWSLease :exec
|
|
UPDATE channel_installation
|
|
SET ws_lease_token = NULL,
|
|
ws_lease_expires_at = NULL,
|
|
updated_at = now()
|
|
WHERE id = $1
|
|
AND ws_lease_token = $2
|
|
`
|
|
|
|
type ReleaseChannelWSLeaseParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
CurrentToken pgtype.Text `json:"current_token"`
|
|
}
|
|
|
|
// Drops the lease iff we are still the holder.
|
|
func (q *Queries) ReleaseChannelWSLease(ctx context.Context, arg ReleaseChannelWSLeaseParams) error {
|
|
_, err := q.db.Exec(ctx, releaseChannelWSLease, arg.ID, arg.CurrentToken)
|
|
return err
|
|
}
|
|
|
|
const setChannelInstallationConfig = `-- name: SetChannelInstallationConfig :exec
|
|
UPDATE channel_installation
|
|
SET config = $2, updated_at = now()
|
|
WHERE id = $1
|
|
`
|
|
|
|
type SetChannelInstallationConfigParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
Config []byte `json:"config"`
|
|
}
|
|
|
|
// Replaces the whole config blob for one installation. Used by the
|
|
// operator backfills (e.g. setting a freshly-fetched bot_union_id) that
|
|
// read-modify-write the JSON in Go and persist it back atomically by id.
|
|
func (q *Queries) SetChannelInstallationConfig(ctx context.Context, arg SetChannelInstallationConfigParams) error {
|
|
_, err := q.db.Exec(ctx, setChannelInstallationConfig, arg.ID, arg.Config)
|
|
return err
|
|
}
|
|
|
|
const setChannelInstallationStatus = `-- name: SetChannelInstallationStatus :exec
|
|
UPDATE channel_installation
|
|
SET status = $2, updated_at = now()
|
|
WHERE id = $1
|
|
`
|
|
|
|
type SetChannelInstallationStatusParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
Status string `json:"status"`
|
|
}
|
|
|
|
func (q *Queries) SetChannelInstallationStatus(ctx context.Context, arg SetChannelInstallationStatusParams) error {
|
|
_, err := q.db.Exec(ctx, setChannelInstallationStatus, arg.ID, arg.Status)
|
|
return err
|
|
}
|
|
|
|
const updateChannelChatSessionBindingReplyTarget = `-- name: UpdateChannelChatSessionBindingReplyTarget :exec
|
|
UPDATE channel_chat_session_binding
|
|
SET last_message_id = $2,
|
|
last_thread_id = $3
|
|
WHERE chat_session_id = $1
|
|
`
|
|
|
|
type UpdateChannelChatSessionBindingReplyTargetParams struct {
|
|
ChatSessionID pgtype.UUID `json:"chat_session_id"`
|
|
LastMessageID pgtype.Text `json:"last_message_id"`
|
|
LastThreadID pgtype.Text `json:"last_thread_id"`
|
|
}
|
|
|
|
// Records the most recent inbound trigger message + thread so the decoupled
|
|
// outbound patcher can thread its reply back into the originating topic.
|
|
func (q *Queries) UpdateChannelChatSessionBindingReplyTarget(ctx context.Context, arg UpdateChannelChatSessionBindingReplyTargetParams) error {
|
|
_, err := q.db.Exec(ctx, updateChannelChatSessionBindingReplyTarget, arg.ChatSessionID, arg.LastMessageID, arg.LastThreadID)
|
|
return err
|
|
}
|
|
|
|
const updateChannelOutboundCardStatus = `-- name: UpdateChannelOutboundCardStatus :exec
|
|
UPDATE channel_outbound_card_message
|
|
SET status = $2,
|
|
last_patched_at = now()
|
|
WHERE id = $1
|
|
`
|
|
|
|
type UpdateChannelOutboundCardStatusParams struct {
|
|
ID pgtype.UUID `json:"id"`
|
|
Status string `json:"status"`
|
|
}
|
|
|
|
func (q *Queries) UpdateChannelOutboundCardStatus(ctx context.Context, arg UpdateChannelOutboundCardStatusParams) error {
|
|
_, err := q.db.Exec(ctx, updateChannelOutboundCardStatus, arg.ID, arg.Status)
|
|
return err
|
|
}
|
|
|
|
const upsertChannelInstallation = `-- name: UpsertChannelInstallation :one
|
|
|
|
|
|
INSERT INTO channel_installation (
|
|
workspace_id, agent_id, channel_type, config, installer_user_id
|
|
) VALUES (
|
|
$1, $2, $3, $4, $5
|
|
)
|
|
ON CONFLICT (workspace_id, agent_id, channel_type) DO UPDATE SET
|
|
channel_type = EXCLUDED.channel_type,
|
|
config = EXCLUDED.config,
|
|
installer_user_id = EXCLUDED.installer_user_id,
|
|
status = 'active',
|
|
installed_at = now(),
|
|
updated_at = now()
|
|
RETURNING id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at
|
|
`
|
|
|
|
type UpsertChannelInstallationParams struct {
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
ChannelType string `json:"channel_type"`
|
|
Config []byte `json:"config"`
|
|
InstallerUserID pgtype.UUID `json:"installer_user_id"`
|
|
}
|
|
|
|
// Platform-agnostic inbound channel queries (MUL-3515). These operate on
|
|
// the channel_* tables created in migration 124. Each installation carries
|
|
// a `channel_type` discriminator and a JSONB `config` blob for
|
|
// platform-specific identifiers/credentials; the cross-platform columns
|
|
// stay flat. The Go layer owns building/parsing config — these queries
|
|
// treat it as opaque JSON except for the routing index on config->>'app_id'.
|
|
//
|
|
// No foreign keys exist on these tables (MUL-3515 §4): the integrity the
|
|
// old composite FKs enforced (binding workspace matches installation;
|
|
// binding dies with membership / chat_session) is maintained in the
|
|
// application layer via the membership check in the inbound identity step
|
|
// and the *DeleteChannel*BindingsBy* cleanup queries below.
|
|
// =====================
|
|
// channel_installation
|
|
// =====================
|
|
// Install / re-install path. `config` is the opaque per-channel JSONB the
|
|
// Go layer assembles (for feishu: app_id, app_secret_encrypted, tenant_key,
|
|
// bot_open_id, bot_union_id, region). Re-installing the same agent on the
|
|
// same channel_type replaces the whole config and forces status back to
|
|
// 'active'. The conflict key is (workspace_id, agent_id, channel_type) so an
|
|
// agent may hold one installation per channel_type (feishu + slack + ...)
|
|
// without one install clobbering another. The WS lease is intentionally NOT
|
|
// reset here — the inbound hub owns lease lifecycle.
|
|
func (q *Queries) UpsertChannelInstallation(ctx context.Context, arg UpsertChannelInstallationParams) (ChannelInstallation, error) {
|
|
row := q.db.QueryRow(ctx, upsertChannelInstallation,
|
|
arg.WorkspaceID,
|
|
arg.AgentID,
|
|
arg.ChannelType,
|
|
arg.Config,
|
|
arg.InstallerUserID,
|
|
)
|
|
var i ChannelInstallation
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.AgentID,
|
|
&i.ChannelType,
|
|
&i.Config,
|
|
&i.Status,
|
|
&i.WsLeaseToken,
|
|
&i.WsLeaseExpiresAt,
|
|
&i.InstallerUserID,
|
|
&i.InstalledAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|
|
|
|
const upsertChannelInstallationByAppID = `-- name: UpsertChannelInstallationByAppID :one
|
|
INSERT INTO channel_installation (
|
|
workspace_id, agent_id, channel_type, config, installer_user_id
|
|
) VALUES (
|
|
$1, $2, $3, $4, $5
|
|
)
|
|
ON CONFLICT (channel_type, (config ->> 'app_id')) DO UPDATE SET
|
|
agent_id = EXCLUDED.agent_id,
|
|
config = EXCLUDED.config,
|
|
installer_user_id = EXCLUDED.installer_user_id,
|
|
status = 'active',
|
|
installed_at = now(),
|
|
updated_at = now()
|
|
WHERE channel_installation.workspace_id = EXCLUDED.workspace_id
|
|
RETURNING id, workspace_id, agent_id, channel_type, config, status, ws_lease_token, ws_lease_expires_at, installer_user_id, installed_at, created_at, updated_at
|
|
`
|
|
|
|
type UpsertChannelInstallationByAppIDParams struct {
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
ChannelType string `json:"channel_type"`
|
|
Config []byte `json:"config"`
|
|
InstallerUserID pgtype.UUID `json:"installer_user_id"`
|
|
}
|
|
|
|
// Team-keyed install / re-install for channels whose natural identity is the
|
|
// platform workspace, not the (agent) pairing. Slack: one Slack workspace
|
|
// (team_id, stored as config->>'app_id') maps to exactly one installation, so
|
|
// re-connecting it — even to represent a DIFFERENT agent in the SAME Multica
|
|
// workspace — UPDATES the existing row (moving agent_id) instead of colliding
|
|
// with the (channel_type, app_id) unique index. Contrast UpsertChannelInstallation,
|
|
// whose conflict key is (workspace_id, agent_id, channel_type): right for Feishu
|
|
// (one app per agent), wrong for Slack.
|
|
//
|
|
// The `WHERE channel_installation.workspace_id = EXCLUDED.workspace_id` fences
|
|
// the conflict update to the SAME Multica workspace: a team already owned by a
|
|
// DIFFERENT workspace updates no row and RETURNING is empty (pgx.ErrNoRows),
|
|
// which the caller maps to ErrTeamOwnedByAnotherWorkspace. This is the ATOMIC
|
|
// cross-workspace guard — a plain SELECT before the upsert cannot stop two
|
|
// workspaces racing to OAuth the same team (both read no rows, then one inserts
|
|
// and the other's conflict-update would silently steal it). A re-connect that
|
|
// would move the team to an agent already holding a different Slack install in
|
|
// the same workspace still trips the (workspace_id, agent_id, channel_type)
|
|
// unique constraint — a genuine conflict the OAuth callback turns into a redirect.
|
|
func (q *Queries) UpsertChannelInstallationByAppID(ctx context.Context, arg UpsertChannelInstallationByAppIDParams) (ChannelInstallation, error) {
|
|
row := q.db.QueryRow(ctx, upsertChannelInstallationByAppID,
|
|
arg.WorkspaceID,
|
|
arg.AgentID,
|
|
arg.ChannelType,
|
|
arg.Config,
|
|
arg.InstallerUserID,
|
|
)
|
|
var i ChannelInstallation
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.WorkspaceID,
|
|
&i.AgentID,
|
|
&i.ChannelType,
|
|
&i.Config,
|
|
&i.Status,
|
|
&i.WsLeaseToken,
|
|
&i.WsLeaseExpiresAt,
|
|
&i.InstallerUserID,
|
|
&i.InstalledAt,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return i, err
|
|
}
|