mirror of
https://github.com/multica-ai/multica.git
synced 2026-08-06 10:50:54 +02:00
* feat(runtime): unbind agents on runtime delete instead of destroying them Deleting a runtime archived its agents and then hard-deleted the rows, so the agents and every conversation with them disappeared — while the confirmation dialog said "archive", which a user reasonably reads as recoverable. Retiring a laptop is an ordinary action; losing the agents configured on it is not an ordinary consequence. An agent is now a persistent business object and a runtime is replaceable execution capacity: deleting a runtime unbinds its agents. `runtime_id IS NULL` means unbound — orthogonal to archived — and the agent keeps its instructions, skills, chats, labels, channel installations, autopilots and task history. service.AgentReadiness already refused an agent with no runtime, so the scheduling safety gate needed no change. Two columns become nullable, not one. Without `agent_task_queue.runtime_id`, deleting the runtime still cascades the task history away (and task_message / task_usage / task_token with it), so the agents would survive with no record of anything they did — the same class of loss. A NOT VALID CHECK keeps NULL confined to history: an active task must always have a runtime, so claim / dispatch / delivery-CAS paths can never observe one without. It is written against completed_at rather than a status list so a future non-terminal status fails closed instead of slipping through. Two prerequisites this depends on: - 'deferred' (migration 128) was missing from CancelAgentTasksByRuntimeOrAgent. It went unnoticed because the delete used to cascade those rows away; with the new CHECK it would abort the delete and make the runtime undeletable. - The channel-installation / label / chat-pin / invocation-target / draft-restore cleanups were scoped to "archived agents on this runtime". Archived user agents now survive, so that scope is narrowed to kind='system' — otherwise the fix would produce a subtler loss: agent alive, configuration wiped. Also removes the squad guard that refused (409) when an active squad's leader was an archived agent on the runtime, plus the archived-squad delete that existed only to get past squad.leader_id's RESTRICT FK. The leader is no longer deleted, so nothing needs to be given up to retire a machine. Autopilots are no longer paused either: their assignee survives, and a rebind restores them without the owner having to remember to re-enable. Reason codes: an unbound agent reports agent_runtime_required, not runtime_offline. The copy for runtime_offline tells users to reconnect a machine; an unbound agent has no machine to reconnect, and the fix is to bind a runtime. Chat's bare 409 string gains the same code so the composer can offer that action. API: agents gain runtime_bound. runtime_id stays a string (empty when unbound) so installed clients keep parsing and no gated two-release rollout is needed. The confirmed-delete endpoint is /unbind-agents-and-delete; /archive-agents-and-delete still routes to it, and the compared expected_active_agent_ids set is unchanged — widening it would 409 every older client forever. Co-authored-by: multica-agent <github@multica.ai> * fix: make runtime unbinding recoverable Co-authored-by: multica-agent <github@multica.ai> * fix: address runtime unbind review nits Co-authored-by: multica-agent <github@multica.ai> * fix: resolve runtime unbind review blockers Co-authored-by: multica-agent <github@multica.ai> * fix(migrations): renumber runtime unbind after main merge Co-authored-by: multica-agent <github@multica.ai> * test(daemon): avoid late-request lease flake Co-authored-by: multica-agent <github@multica.ai> * test(autopilots): bind validation fixture runtime Co-authored-by: multica-agent <github@multica.ai> --------- Co-authored-by: Eve <eve@multica-ai.local> Co-authored-by: multica-agent <github@multica.ai>
1799 lines
69 KiB
Go
1799 lines
69 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 channelMediaObjectIsReferenced = `-- name: ChannelMediaObjectIsReferenced :one
|
|
SELECT EXISTS (
|
|
SELECT 1 FROM attachment
|
|
WHERE chat_message_id = $1
|
|
AND workspace_id = $2
|
|
AND url = $3
|
|
) AS referenced
|
|
`
|
|
|
|
type ChannelMediaObjectIsReferencedParams struct {
|
|
ChatMessageID pgtype.UUID `json:"chat_message_id"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
StorageUrl string `json:"storage_url"`
|
|
}
|
|
|
|
// The post-claim reference check: an attachment row carrying this object's
|
|
// URL on the intended message. Only meaningful AFTER the claim flipped the
|
|
// row to 'deleting' — from that point a bind can no longer succeed on the
|
|
// key, so a negative answer is terminal, not a snapshot race. Re-run on every
|
|
// tombstone pass as well: a positive answer there is an invariant violation,
|
|
// and the object is kept and reported rather than deleted.
|
|
func (q *Queries) ChannelMediaObjectIsReferenced(ctx context.Context, arg ChannelMediaObjectIsReferencedParams) (bool, error) {
|
|
row := q.db.QueryRow(ctx, channelMediaObjectIsReferenced, arg.ChatMessageID, arg.WorkspaceID, arg.StorageUrl)
|
|
var referenced bool
|
|
err := row.Scan(&referenced)
|
|
return referenced, err
|
|
}
|
|
|
|
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 claimChannelMediaPendingObjectsForBind = `-- name: ClaimChannelMediaPendingObjectsForBind :many
|
|
DELETE FROM channel_media_pending_object
|
|
WHERE storage_key = ANY($1::text[])
|
|
AND workspace_id = $2
|
|
AND state = 'pending'
|
|
RETURNING storage_key
|
|
`
|
|
|
|
type ClaimChannelMediaPendingObjectsForBindParams struct {
|
|
StorageKeys []string `json:"storage_keys"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
}
|
|
|
|
// Runs inside the attachment-insert transaction: commit landed ⇔ the intents
|
|
// are gone, atomically, so an ambiguous COMMIT never needs adjudication. Only
|
|
// 'pending' rows can be claimed — a key the reconciler moved to 'deleting'
|
|
// is NOT returned, and the caller must skip attaching that object (the
|
|
// placeholder stays; the reconciler will delete the object).
|
|
func (q *Queries) ClaimChannelMediaPendingObjectsForBind(ctx context.Context, arg ClaimChannelMediaPendingObjectsForBindParams) ([]string, error) {
|
|
rows, err := q.db.Query(ctx, claimChannelMediaPendingObjectsForBind, arg.StorageKeys, arg.WorkspaceID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []string{}
|
|
for rows.Next() {
|
|
var storage_key string
|
|
if err := rows.Scan(&storage_key); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, storage_key)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const claimNextChannelMediaPendingObjectForReconcile = `-- name: ClaimNextChannelMediaPendingObjectForReconcile :one
|
|
UPDATE channel_media_pending_object AS obj
|
|
SET state = CASE WHEN obj.state = 'tombstoned' THEN 'tombstoned' ELSE 'deleting' END,
|
|
lease_token = $1,
|
|
lease_expires_at = now() + $2::interval,
|
|
attempt = obj.attempt + 1
|
|
FROM (
|
|
SELECT cand.storage_key FROM channel_media_pending_object AS cand
|
|
WHERE cand.next_attempt_at <= now()
|
|
AND (
|
|
(cand.state = 'pending' AND cand.created_at <= now() - $3::interval)
|
|
OR (cand.state = 'deleting' AND (cand.lease_expires_at IS NULL OR cand.lease_expires_at <= now()))
|
|
OR (cand.state = 'tombstoned' AND (cand.lease_expires_at IS NULL OR cand.lease_expires_at <= now()))
|
|
)
|
|
ORDER BY cand.next_attempt_at
|
|
LIMIT 1
|
|
FOR UPDATE SKIP LOCKED
|
|
) AS due
|
|
WHERE obj.storage_key = due.storage_key
|
|
RETURNING obj.storage_key, obj.workspace_id, obj.chat_message_id, obj.storage_url, obj.installation_id, obj.state, obj.lease_token, obj.lease_expires_at, obj.attempt, obj.next_attempt_at, obj.last_error, obj.tombstone_pass, obj.created_at
|
|
`
|
|
|
|
type ClaimNextChannelMediaPendingObjectForReconcileParams struct {
|
|
LeaseToken pgtype.UUID `json:"lease_token"`
|
|
Lease pgtype.Interval `json:"lease"`
|
|
SettleDelay pgtype.Interval `json:"settle_delay"`
|
|
}
|
|
|
|
// Short-transaction claim of ONE due row, taken immediately before that row is
|
|
// settled. Claiming a whole batch up front made the claim a promise the sweep
|
|
// might not keep: a tail row sat in 'deleting' with attempt already bumped
|
|
// while earlier rows ran, and could expire and be reclaimed before its own
|
|
// DELETE was ever tried, inflating attempt/backoff for work that never
|
|
// happened. One row per claim means attempt counts attempts.
|
|
//
|
|
// Due means (a) 'pending' rows older than the settle delay — an operational
|
|
// buffer only; correctness comes from the state flip, after which a bind can
|
|
// never succeed on the key — or (b) 'deleting' rows whose lease expired (a
|
|
// crashed or failed worker) — or (c) tombstones: the object was deleted, but a
|
|
// PUT the client abandoned may still materialize it afterwards, so each due
|
|
// tombstone gets another idempotent delete before the row is finally dropped.
|
|
// FOR UPDATE SKIP LOCKED keeps replicas off each other's row; the
|
|
// object-storage DELETE happens outside any transaction, gated by the lease
|
|
// token. No row (ErrNoRows) means nothing is due — the sweep is done.
|
|
func (q *Queries) ClaimNextChannelMediaPendingObjectForReconcile(ctx context.Context, arg ClaimNextChannelMediaPendingObjectForReconcileParams) (ChannelMediaPendingObject, error) {
|
|
row := q.db.QueryRow(ctx, claimNextChannelMediaPendingObjectForReconcile, arg.LeaseToken, arg.Lease, arg.SettleDelay)
|
|
var i ChannelMediaPendingObject
|
|
err := row.Scan(
|
|
&i.StorageKey,
|
|
&i.WorkspaceID,
|
|
&i.ChatMessageID,
|
|
&i.StorageUrl,
|
|
&i.InstallationID,
|
|
&i.State,
|
|
&i.LeaseToken,
|
|
&i.LeaseExpiresAt,
|
|
&i.Attempt,
|
|
&i.NextAttemptAt,
|
|
&i.LastError,
|
|
&i.TombstonePass,
|
|
&i.CreatedAt,
|
|
)
|
|
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 countChannelMediaPendingObjects = `-- name: CountChannelMediaPendingObjects :one
|
|
SELECT
|
|
count(*) FILTER (WHERE state <> 'tombstoned') AS pending_objects,
|
|
count(*) FILTER (WHERE state = 'tombstoned') AS tombstoned_objects
|
|
FROM channel_media_pending_object
|
|
`
|
|
|
|
type CountChannelMediaPendingObjectsRow struct {
|
|
PendingObjects int64 `json:"pending_objects"`
|
|
TombstonedObjects int64 `json:"tombstoned_objects"`
|
|
}
|
|
|
|
// Ledger backlog gauge for the reconciler's observability. Tombstones are
|
|
// reported separately: they are bounded bookkeeping for already-deleted
|
|
// objects, not a backlog of objects awaiting reclaim.
|
|
func (q *Queries) CountChannelMediaPendingObjects(ctx context.Context) (CountChannelMediaPendingObjectsRow, error) {
|
|
row := q.db.QueryRow(ctx, countChannelMediaPendingObjects)
|
|
var i CountChannelMediaPendingObjectsRow
|
|
err := row.Scan(&i.PendingObjects, &i.TombstonedObjects)
|
|
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,
|
|
LEAST($6::timestamptz, now() + INTERVAL '15 minutes')
|
|
)
|
|
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. Clamp against the database clock so small clock
|
|
// skew between an app node and Postgres cannot reject an otherwise valid
|
|
// 15-minute token. The HASH is stored, never the raw token.
|
|
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 deleteChannelInstallationsBySystemRuntimeAgents = `-- name: DeleteChannelInstallationsBySystemRuntimeAgents :exec
|
|
WITH doomed AS (
|
|
SELECT id FROM channel_installation
|
|
WHERE agent_id IN (
|
|
SELECT id FROM agent WHERE runtime_id = $1 AND kind = 'system'
|
|
)
|
|
),
|
|
cleared_chat_sessions AS (
|
|
DELETE FROM channel_chat_session_binding WHERE installation_id IN (SELECT id FROM doomed)
|
|
RETURNING chat_session_id
|
|
),
|
|
cleared_outbound_cards AS (
|
|
-- Reach channel_outbound_card_message (keyed by chat_session_id, no FK)
|
|
-- through the just-removed chat-session bindings, same as the reclaim path.
|
|
DELETE FROM channel_outbound_card_message
|
|
WHERE chat_session_id IN (SELECT chat_session_id FROM cleared_chat_sessions)
|
|
),
|
|
cleared_binding_tokens AS (
|
|
DELETE FROM channel_binding_token WHERE installation_id IN (SELECT id FROM doomed)
|
|
),
|
|
cleared_user_bindings AS (
|
|
DELETE FROM channel_user_binding WHERE installation_id IN (SELECT id FROM doomed)
|
|
),
|
|
cleared_inbound_dedup AS (
|
|
DELETE FROM channel_inbound_message_dedup WHERE installation_id IN (SELECT id FROM doomed)
|
|
),
|
|
cleared_audit AS (
|
|
-- Hard delete: purge audit rows rather than detaching them into permanently
|
|
-- unattributable NULL rows (channel_inbound_audit has no workspace_id / reaper).
|
|
DELETE FROM channel_inbound_audit WHERE installation_id IN (SELECT id FROM doomed)
|
|
)
|
|
DELETE FROM channel_installation WHERE id IN (SELECT id FROM doomed)
|
|
`
|
|
|
|
// Application-layer replacement for the (deliberately absent, MUL-3515 §4)
|
|
// workspace/agent ON DELETE CASCADE: on runtime teardown, before the system
|
|
// agents are hard-deleted, remove every channel installation they own — plus all
|
|
// of each installation's dependent rows — so no orphaned installation keeps
|
|
// occupying its bot's (channel_type, app_id) routing slot after its agent is gone
|
|
// (#4810). MUST run in the same tx as, and BEFORE, DeleteSystemAgentsByRuntime.
|
|
// Mirrors the agent hard-delete predicate (runtime_id, kind = 'system') exactly.
|
|
//
|
|
// Scoped to kind = 'system' since MUL-5559: a user agent now survives its
|
|
// runtime's deletion as an unbound agent, so tearing down its installations
|
|
// here would take a working bot away from an agent that is still there.
|
|
func (q *Queries) DeleteChannelInstallationsBySystemRuntimeAgents(ctx context.Context, runtimeID pgtype.UUID) error {
|
|
_, err := q.db.Exec(ctx, deleteChannelInstallationsBySystemRuntimeAgents, runtimeID)
|
|
return err
|
|
}
|
|
|
|
const deleteChannelMediaPendingObject = `-- name: DeleteChannelMediaPendingObject :execrows
|
|
DELETE FROM channel_media_pending_object
|
|
WHERE storage_key = $1
|
|
AND workspace_id = $2
|
|
AND lease_token = $3
|
|
`
|
|
|
|
type DeleteChannelMediaPendingObjectParams struct {
|
|
StorageKey string `json:"storage_key"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
LeaseToken pgtype.UUID `json:"lease_token"`
|
|
}
|
|
|
|
// Drops a claimed row for good: a durable attachment reference was found, or
|
|
// the tombstone's re-delete schedule is exhausted. Lease-token guarded so an
|
|
// expired-lease reclaim by another
|
|
// replica cannot be clobbered; workspace_id explicit per the tenancy rule.
|
|
func (q *Queries) DeleteChannelMediaPendingObject(ctx context.Context, arg DeleteChannelMediaPendingObjectParams) (int64, error) {
|
|
result, err := q.db.Exec(ctx, deleteChannelMediaPendingObject, arg.StorageKey, arg.WorkspaceID, arg.LeaseToken)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected(), nil
|
|
}
|
|
|
|
const deleteChannelOutboundCardMessagesBySession = `-- name: DeleteChannelOutboundCardMessagesBySession :exec
|
|
DELETE FROM channel_outbound_card_message
|
|
WHERE chat_session_id = $1
|
|
`
|
|
|
|
// Application-layer integrity (channel_* has no FK/cascade, MUL-3515 §4): drop the
|
|
// outbound card-message rows for a chat_session being deleted. They are keyed by
|
|
// chat_session_id with no FK and no reaper, so the standalone chat-session delete
|
|
// path must prune them here alongside DeleteChannelChatSessionBindingBySession —
|
|
// otherwise deleting a chat session leaves them as permanent orphans (Elon's
|
|
// follow-up on #4810; the workspace/agent/reclaim sweeps already cover their
|
|
// paths). A card that survived its session could only mis-route a later patch.
|
|
func (q *Queries) DeleteChannelOutboundCardMessagesBySession(ctx context.Context, chatSessionID pgtype.UUID) error {
|
|
_, err := q.db.Exec(ctx, deleteChannelOutboundCardMessagesBySession, chatSessionID)
|
|
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 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 getChannelInstallationOwnerByAppID = `-- name: GetChannelInstallationOwnerByAppID :one
|
|
SELECT ci.workspace_id, ci.agent_id, a.archived_at AS agent_archived_at
|
|
FROM channel_installation ci
|
|
JOIN agent a ON a.id = ci.agent_id
|
|
WHERE ci.channel_type = $1
|
|
AND ci.config ->> 'app_id' = $2::text
|
|
`
|
|
|
|
type GetChannelInstallationOwnerByAppIDParams struct {
|
|
ChannelType string `json:"channel_type"`
|
|
AppID string `json:"app_id"`
|
|
}
|
|
|
|
type GetChannelInstallationOwnerByAppIDRow struct {
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
AgentArchivedAt pgtype.Timestamptz `json:"agent_archived_at"`
|
|
}
|
|
|
|
// Identifies the LIVE owner of a (channel_type, config->>'app_id') routing slot
|
|
// so the install path can refuse a rebind with an ACCURATE message instead of the
|
|
// old catch-all "connected to a different Multica workspace". Meant to be read
|
|
// only after ReclaimDeadChannelInstallationByAppID has removed every DEAD owner,
|
|
// so a returned row is a live active owner. `agent_archived` distinguishes an
|
|
// archived (reversible) owner — its bot stays owned, recovered by unarchiving the
|
|
// agent or disconnecting the bot — from a plain active one. The JOIN drops a row
|
|
// whose agent no longer exists (an orphan the reclaim gate should already have
|
|
// cleared), so a missing row (pgx.ErrNoRows) means "no live owner". The caller
|
|
// reads agent_archived_at.Valid to tell an archived (reversible) owner apart.
|
|
func (q *Queries) GetChannelInstallationOwnerByAppID(ctx context.Context, arg GetChannelInstallationOwnerByAppIDParams) (GetChannelInstallationOwnerByAppIDRow, error) {
|
|
row := q.db.QueryRow(ctx, getChannelInstallationOwnerByAppID, arg.ChannelType, arg.AppID)
|
|
var i GetChannelInstallationOwnerByAppIDRow
|
|
err := row.Scan(&i.WorkspaceID, &i.AgentID, &i.AgentArchivedAt)
|
|
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 reclaimDeadChannelInstallationByAppID = `-- name: ReclaimDeadChannelInstallationByAppID :one
|
|
WITH dead AS (
|
|
DELETE FROM channel_installation ci
|
|
WHERE ci.channel_type = $1
|
|
AND ci.config ->> 'app_id' = $2::text
|
|
AND (
|
|
(ci.status = 'revoked'
|
|
AND NOT (ci.workspace_id = $3
|
|
AND ci.agent_id = $4))
|
|
OR NOT EXISTS (SELECT 1 FROM workspace w WHERE w.id = ci.workspace_id)
|
|
OR NOT EXISTS (SELECT 1 FROM agent a WHERE a.id = ci.agent_id)
|
|
)
|
|
RETURNING ci.id
|
|
),
|
|
cleared_chat_sessions AS (
|
|
DELETE FROM channel_chat_session_binding
|
|
WHERE installation_id IN (SELECT id FROM dead)
|
|
RETURNING chat_session_id
|
|
),
|
|
cleared_outbound_cards AS (
|
|
-- channel_outbound_card_message is keyed by chat_session_id (no installation_id,
|
|
-- no FK), so it is reached through the just-removed chat-session bindings. On an
|
|
-- orphan reclaim the chat_session row itself is already cascade-gone, but its
|
|
-- binding survived and still carries the id — the only reliable link back.
|
|
DELETE FROM channel_outbound_card_message
|
|
WHERE chat_session_id IN (SELECT chat_session_id FROM cleared_chat_sessions)
|
|
),
|
|
cleared_binding_tokens AS (
|
|
DELETE FROM channel_binding_token
|
|
WHERE installation_id IN (SELECT id FROM dead)
|
|
),
|
|
cleared_user_bindings AS (
|
|
DELETE FROM channel_user_binding
|
|
WHERE installation_id IN (SELECT id FROM dead)
|
|
),
|
|
cleared_inbound_dedup AS (
|
|
DELETE FROM channel_inbound_message_dedup
|
|
WHERE installation_id IN (SELECT id FROM dead)
|
|
),
|
|
detached_audit AS (
|
|
-- Reclaim keeps the DETACH semantics: the workspace still exists, so a
|
|
-- NULL-installation audit row stays meaningful for operator triage. The hard-
|
|
-- delete paths (DeleteWorkspace / runtime teardown) purge audit outright.
|
|
UPDATE channel_inbound_audit SET installation_id = NULL
|
|
WHERE installation_id IN (SELECT id FROM dead)
|
|
)
|
|
SELECT id FROM dead
|
|
`
|
|
|
|
type ReclaimDeadChannelInstallationByAppIDParams struct {
|
|
ChannelType string `json:"channel_type"`
|
|
AppID string `json:"app_id"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
AgentID pgtype.UUID `json:"agent_id"`
|
|
}
|
|
|
|
// Rebind cleanup gate. Frees the (channel_type, config->>'app_id') routing slot
|
|
// so a valid new agent can (re)bind a bot whose previous owner is DEAD, and, in
|
|
// the same statement, clears every application-owned dependent row of the removed
|
|
// installation (channel_* has no FK/cascade, MUL-3515 §4). Returns the removed id
|
|
// (pgx.ErrNoRows when nothing was dead — a no-op the caller treats as success).
|
|
//
|
|
// "Dead" is exactly one of:
|
|
// 1. a REVOKED placeholder held by ANY agent OTHER than the caller's own
|
|
// (workspace, agent) pair. Disconnect only flips status to 'revoked' — no
|
|
// product path ever hard-deletes the row — so a revoked row would otherwise
|
|
// pin the bot's app_id slot forever with no self-serve recovery, even across
|
|
// workspaces (workspace A disconnects; workspace B, which proves control by
|
|
// holding the same app credentials, rebinds). Revoke is the owner's explicit
|
|
// "I'm done with this bot", so any revoked row is reclaimable — only the
|
|
// caller's OWN revoked row is spared (reactivated in place; see below).
|
|
// 2. an ORPHAN whose owning workspace OR agent row no longer exists — the
|
|
// workspace was deleted, or the agent was hard-deleted on runtime teardown.
|
|
// With no FK the installation outlives its owner and keeps occupying the
|
|
// app_id slot: the "ghost binding" that made a bot un-rebindable (#4810).
|
|
//
|
|
// Deliberately NOT dead (the caller refuses these with an accurate conflict):
|
|
// - the SAME agent's own revoked row (agent_id = @agent_id): the upsert
|
|
// reactivates it in place, preserving its installation_id and every binding;
|
|
// - a live ACTIVE owner whose agent still exists — INCLUDING an ARCHIVED agent:
|
|
// archive is reversible, so its bot stays owned rather than being silently
|
|
// stolen. Only a hard delete frees the slot.
|
|
//
|
|
// The guard lives in the DELETE predicate (not a prior SELECT) so under READ
|
|
// COMMITTED the row is re-checked at execution (EvalPlanQual): a concurrent
|
|
// same-agent reconnect that flips the revoked row back to 'active' first makes
|
|
// the predicate re-check fail, this deletes nothing, and no dependents are
|
|
// touched — closing the read-then-delete TOCTOU. Dependent cleanup keys off the
|
|
// actually-deleted id (the `dead` CTE), so it runs ONLY for a row this statement
|
|
// removed. The (channel_type, app_id) unique index guarantees at most one match.
|
|
func (q *Queries) ReclaimDeadChannelInstallationByAppID(ctx context.Context, arg ReclaimDeadChannelInstallationByAppIDParams) (pgtype.UUID, error) {
|
|
row := q.db.QueryRow(ctx, reclaimDeadChannelInstallationByAppID,
|
|
arg.ChannelType,
|
|
arg.AppID,
|
|
arg.WorkspaceID,
|
|
arg.AgentID,
|
|
)
|
|
var id pgtype.UUID
|
|
err := row.Scan(&id)
|
|
return id, 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 recordChannelMediaPendingObject = `-- name: RecordChannelMediaPendingObject :one
|
|
|
|
INSERT INTO channel_media_pending_object (
|
|
storage_key, workspace_id, chat_message_id, storage_url, installation_id
|
|
)
|
|
VALUES ($1, $2, $3, $4, $5)
|
|
ON CONFLICT (storage_key) DO UPDATE
|
|
SET created_at = now(), next_attempt_at = now(),
|
|
chat_message_id = EXCLUDED.chat_message_id,
|
|
storage_url = EXCLUDED.storage_url
|
|
WHERE channel_media_pending_object.state = 'pending'
|
|
AND channel_media_pending_object.workspace_id = EXCLUDED.workspace_id
|
|
RETURNING storage_key
|
|
`
|
|
|
|
type RecordChannelMediaPendingObjectParams struct {
|
|
StorageKey string `json:"storage_key"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
ChatMessageID pgtype.UUID `json:"chat_message_id"`
|
|
StorageUrl string `json:"storage_url"`
|
|
InstallationID pgtype.UUID `json:"installation_id"`
|
|
}
|
|
|
|
// =====================
|
|
// channel_media_pending_object (media intent ledger)
|
|
// =====================
|
|
// Records upload intent BEFORE the PUT. A redelivered attempt refreshes the
|
|
// settle window, but only while the row is still 'pending' — a key the
|
|
// reconciler owns ('deleting') must never be resurrected — and only within
|
|
// the SAME workspace: a cross-workspace key collision (impossible via the
|
|
// derived key, but tenancy must never trust the key string) updates nothing
|
|
// and returns no row, so the caller skips the upload entirely.
|
|
func (q *Queries) RecordChannelMediaPendingObject(ctx context.Context, arg RecordChannelMediaPendingObjectParams) (string, error) {
|
|
row := q.db.QueryRow(ctx, recordChannelMediaPendingObject,
|
|
arg.StorageKey,
|
|
arg.WorkspaceID,
|
|
arg.ChatMessageID,
|
|
arg.StorageUrl,
|
|
arg.InstallationID,
|
|
)
|
|
var storage_key string
|
|
err := row.Scan(&storage_key)
|
|
return storage_key, 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 releaseChannelMediaPendingObject = `-- name: ReleaseChannelMediaPendingObject :exec
|
|
UPDATE channel_media_pending_object
|
|
SET lease_token = NULL,
|
|
lease_expires_at = NULL,
|
|
next_attempt_at = now() + $1::interval,
|
|
last_error = $2
|
|
WHERE storage_key = $3
|
|
AND workspace_id = $4
|
|
AND lease_token = $5
|
|
`
|
|
|
|
type ReleaseChannelMediaPendingObjectParams struct {
|
|
Backoff pgtype.Interval `json:"backoff"`
|
|
LastError pgtype.Text `json:"last_error"`
|
|
StorageKey string `json:"storage_key"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
LeaseToken pgtype.UUID `json:"lease_token"`
|
|
}
|
|
|
|
// Object-storage DELETE failed: keep the row in 'deleting' (bind must still
|
|
// never attach it), release the lease, and back off the next attempt.
|
|
// workspace_id is redundant with the storage_key PK but explicit per the
|
|
// tenancy rule: every query constrains the workspace column, never trusting
|
|
// the key string.
|
|
func (q *Queries) ReleaseChannelMediaPendingObject(ctx context.Context, arg ReleaseChannelMediaPendingObjectParams) error {
|
|
_, err := q.db.Exec(ctx, releaseChannelMediaPendingObject,
|
|
arg.Backoff,
|
|
arg.LastError,
|
|
arg.StorageKey,
|
|
arg.WorkspaceID,
|
|
arg.LeaseToken,
|
|
)
|
|
return err
|
|
}
|
|
|
|
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 tombstoneChannelMediaPendingObject = `-- name: TombstoneChannelMediaPendingObject :execrows
|
|
UPDATE channel_media_pending_object
|
|
SET state = 'tombstoned',
|
|
lease_token = NULL,
|
|
lease_expires_at = NULL,
|
|
next_attempt_at = now() + $1::interval,
|
|
-- The pass index lives in its own column: a failed re-delete writes
|
|
-- last_error, so carrying the schedule position there would reset the
|
|
-- walk on every failure and a flaky store could keep the row alive
|
|
-- indefinitely. The delete that got here succeeded, so any previous
|
|
-- failure text is stale.
|
|
tombstone_pass = $2,
|
|
last_error = NULL
|
|
WHERE storage_key = $3
|
|
AND workspace_id = $4
|
|
AND lease_token = $5
|
|
`
|
|
|
|
type TombstoneChannelMediaPendingObjectParams struct {
|
|
RedeleteDelay pgtype.Interval `json:"redelete_delay"`
|
|
TombstonePass int32 `json:"tombstone_pass"`
|
|
StorageKey string `json:"storage_key"`
|
|
WorkspaceID pgtype.UUID `json:"workspace_id"`
|
|
LeaseToken pgtype.UUID `json:"lease_token"`
|
|
}
|
|
|
|
// The object was deleted, but the row is KEPT as a tombstone: a PUT the client
|
|
// abandoned before the delete may still materialize the object afterwards, and
|
|
// no DELETE can be ordered against it. Each due tombstone re-runs the
|
|
// reference check and, only if still unreferenced, triggers another idempotent
|
|
// delete, so a late materialization is reclaimed by a later pass while an
|
|
// object something durably reads is never removed;
|
|
// only after the re-delete schedule is exhausted is the row dropped
|
|
// (DeleteChannelMediaPendingObject). Lease-token guarded like every other
|
|
// settle write; workspace_id explicit per the tenancy rule.
|
|
func (q *Queries) TombstoneChannelMediaPendingObject(ctx context.Context, arg TombstoneChannelMediaPendingObjectParams) (int64, error) {
|
|
result, err := q.db.Exec(ctx, tombstoneChannelMediaPendingObject,
|
|
arg.RedeleteDelay,
|
|
arg.TombstonePass,
|
|
arg.StorageKey,
|
|
arg.WorkspaceID,
|
|
arg.LeaseToken,
|
|
)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected(), nil
|
|
}
|
|
|
|
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
|
|
}
|