From 33bd8aeaa97dcb2a625b5fc29a869f455462f634 Mon Sep 17 00:00:00 2001 From: Bohan Jiang <52446949+Bohan-J@users.noreply.github.com> Date: Tue, 7 Jul 2026 15:07:02 +0800 Subject: [PATCH] MUL-4134: fix(lark): preserve same-agent bindings when reconnecting a revoked Feishu bot (#4997) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 Co-authored-by: multica-agent * 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 * 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 * 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 --------- Co-authored-by: jiangliangyou Co-authored-by: Claude Co-authored-by: multica-agent Co-authored-by: J --- .../integrations/lark/channel_store.go | 57 +++ .../lark/channel_store_rebind_test.go | 454 ++++++++++++++++++ .../integrations/lark/registration_service.go | 16 + server/pkg/db/generated/channel.sql.go | 106 ++++ server/pkg/db/queries/channel.sql | 68 +++ 5 files changed, 701 insertions(+) create mode 100644 server/internal/integrations/lark/channel_store_rebind_test.go diff --git a/server/internal/integrations/lark/channel_store.go b/server/internal/integrations/lark/channel_store.go index 4afb246c62..ddc267360f 100644 --- a/server/internal/integrations/lark/channel_store.go +++ b/server/internal/integrations/lark/channel_store.go @@ -142,6 +142,63 @@ func (s *ChannelStore) UpsertLarkInstallation(ctx context.Context, arg UpsertIns return installationFromRow(row) } +// RemoveRevokedInstallationByAppID clears a revoked installation that belongs to +// a DIFFERENT agent in the same workspace and holds the (channel_type, +// config->>'app_id') unique slot, so the caller can re-install the same +// Lark/Feishu app against a new agent without tripping the functional unique +// index. The guarded DELETE is the atomic gate: it removes the row only when it +// is still (this workspace, another agent, revoked) at delete time and RETURNS +// its id. Only when a row is actually claimed do we clean its dependents. +// +// - the SAME agent's revoked row (agent_id = agentID) never matches, so the +// upsert reactivates it in place (installation_id + all bindings preserved); +// - an ACTIVE row (bot still connected) never matches, so the install surfaces +// as a conflict instead of silently stealing it; +// - a row in ANOTHER workspace never matches. +// +// Keying cleanup off the RETURNING id — rather than a prior read — closes the +// read-then-delete race: a concurrent same-agent reconnect that flips the row +// to 'active' between a read and the delete would previously still trigger the +// dependent cleanup on a stale id and wipe a since-reactivated installation's +// bindings. Under READ COMMITTED the DELETE re-checks status='revoked' against +// the live row, so it claims nothing in that case and we touch no dependents. +// +// channel_* has no FK/cascade (MUL-3515 §4), so every application-owned row that +// referenced the deleted installation is cleared explicitly — chat-session +// bindings, pending binding tokens, member links — and inbound-audit rows are +// detached by NULLing installation_id (the app-layer stand-in for the old +// ON DELETE SET NULL). With no FK the order is free, so cleanup runs after the +// claiming delete; all of it is on the caller's transaction-bound queries, so +// delete + cleanup + the follow-up upsert commit (or roll back) atomically. +func (s *ChannelStore) RemoveRevokedInstallationByAppID(ctx context.Context, workspaceID, agentID pgtype.UUID, appID string) error { + installID, err := s.Queries.DeleteRevokedChannelInstallationByAppID(ctx, db.DeleteRevokedChannelInstallationByAppIDParams{ + ChannelType: channelTypeFeishu, + AppID: appID, + WorkspaceID: workspaceID, + AgentID: agentID, + }) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil // no blocking row claimed (same agent / active / other workspace / none) + } + return err + } + + if err := s.Queries.DeleteChannelChatSessionBindingsByInstallation(ctx, db.DeleteChannelChatSessionBindingsByInstallationParams{ + InstallationID: installID, + ChannelType: channelTypeFeishu, + }); err != nil { + return err + } + if err := s.Queries.DeleteChannelBindingTokensByInstallation(ctx, installID); err != nil { + return err + } + if err := s.Queries.DeleteChannelUserBindingsByInstallation(ctx, installID); err != nil { + return err + } + return s.Queries.NullChannelInboundAuditInstallationID(ctx, installID) +} + func (s *ChannelStore) SetLarkInstallationStatus(ctx context.Context, arg SetInstallationStatusParams) error { return s.Queries.SetChannelInstallationStatus(ctx, db.SetChannelInstallationStatusParams{ ID: arg.ID, diff --git a/server/internal/integrations/lark/channel_store_rebind_test.go b/server/internal/integrations/lark/channel_store_rebind_test.go new file mode 100644 index 0000000000..f46ea0fbd9 --- /dev/null +++ b/server/internal/integrations/lark/channel_store_rebind_test.go @@ -0,0 +1,454 @@ +package lark + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + + "github.com/multica-ai/multica/server/internal/util" + db "github.com/multica-ai/multica/server/pkg/db/generated" +) + +// Rebind regression fixtures. Namespaced away from the scope test's ids so a +// shared test database never cross-contaminates. channel_* has no foreign keys, +// so these rows need no parent records; the test cleans up by deterministic key +// before and after (a killed prior run must not leave colliding rows behind). +const ( + rbWS = "5c09e100-0000-4000-8000-000000000001" + rbWS2 = "5c09e100-0000-4000-8000-000000000002" + rbAgentA = "5c09e100-0000-4000-8000-00000000000a" + rbAgentB = "5c09e100-0000-4000-8000-00000000000b" + rbInstaller = "5c09e100-0000-4000-8000-000000000005" + rbUser = "5c09e100-0000-4000-8000-000000000006" + rbChatSess = "5c09e100-0000-4000-8000-000000000007" + + rbAppSame = "cli_rb_same" + rbAppDiff = "cli_rb_diff" + rbAppActive = "cli_rb_active" + rbAppWsFence = "cli_rb_wsfence" + rbAppReactivate = "cli_rb_reactivate" + rbAppMove = "cli_rb_move" +) + +// TestChannelStore_RemoveRevokedInstallationByAppID guards the WHERE clause of +// the DeleteRevokedChannelInstallationByAppID gate: it must claim ONLY a revoked +// row that belongs to a DIFFERENT agent in the SAME workspace. The same agent's +// own revoked row, any active row, and rows in another workspace must survive. +func TestChannelStore_RemoveRevokedInstallationByAppID(t *testing.T) { + pool := channelScopeTestDB(t) + ctx := context.Background() + store := NewChannelStore(db.New(pool)) + + apps := []string{rbAppSame, rbAppDiff, rbAppActive, rbAppWsFence, rbAppReactivate, rbAppMove} + clean := func() { + _, _ = pool.Exec(ctx, `DELETE FROM channel_installation WHERE config->>'app_id' = ANY($1)`, apps) + _, _ = pool.Exec(ctx, `DELETE FROM channel_user_binding WHERE multica_user_id = $1`, rbUser) + _, _ = pool.Exec(ctx, `DELETE FROM channel_chat_session_binding WHERE chat_session_id = $1`, rbChatSess) + } + clean() + t.Cleanup(clean) + + // insert an installation and return its id. + insert := func(app, ws, agent, status string) pgtype.UUID { + var id string + if err := pool.QueryRow(ctx, ` +INSERT INTO channel_installation (workspace_id, agent_id, channel_type, config, installer_user_id, status) +VALUES ($1, $2, 'feishu', jsonb_build_object('app_id', $3::text), $4, $5) +RETURNING id +`, ws, agent, app, rbInstaller, status).Scan(&id); err != nil { + t.Fatalf("insert installation app=%s status=%s: %v", app, status, err) + } + return util.MustParseUUID(id) + } + exists := func(id pgtype.UUID) bool { + _, err := store.GetLarkInstallation(ctx, id) + if err == nil { + return true + } + if errors.Is(err, pgx.ErrNoRows) { + return false + } + t.Fatalf("GetLarkInstallation: %v", err) + return false + } + + wsUUID := util.MustParseUUID(rbWS) + agentAUUID := util.MustParseUUID(rbAgentA) + agentBUUID := util.MustParseUUID(rbAgentB) + + t.Run("same agent revoked row is preserved", func(t *testing.T) { + clean() + id := insert(rbAppSame, rbWS, rbAgentA, "revoked") + if err := store.RemoveRevokedInstallationByAppID(ctx, wsUUID, agentAUUID, rbAppSame); err != nil { + t.Fatalf("RemoveRevokedInstallationByAppID: %v", err) + } + if !exists(id) { + t.Fatal("same agent's own revoked row was deleted; it must be reactivated in place by the upsert, not orphaned") + } + }) + + t.Run("different agent revoked row is deleted", func(t *testing.T) { + clean() + id := insert(rbAppDiff, rbWS, rbAgentA, "revoked") + if err := store.RemoveRevokedInstallationByAppID(ctx, wsUUID, agentBUUID, rbAppDiff); err != nil { + t.Fatalf("RemoveRevokedInstallationByAppID: %v", err) + } + if exists(id) { + t.Fatal("a different agent's revoked row was not deleted; it would keep blocking the app_id unique slot") + } + }) + + t.Run("active row is never deleted", func(t *testing.T) { + clean() + id := insert(rbAppActive, rbWS, rbAgentA, "active") + if err := store.RemoveRevokedInstallationByAppID(ctx, wsUUID, agentBUUID, rbAppActive); err != nil { + t.Fatalf("RemoveRevokedInstallationByAppID: %v", err) + } + if !exists(id) { + t.Fatal("an active installation was deleted through the revoked-cleanup path") + } + }) + + t.Run("other workspace revoked row is preserved", func(t *testing.T) { + clean() + id := insert(rbAppWsFence, rbWS2, rbAgentA, "revoked") + if err := store.RemoveRevokedInstallationByAppID(ctx, wsUUID, agentBUUID, rbAppWsFence); err != nil { + t.Fatalf("RemoveRevokedInstallationByAppID: %v", err) + } + if !exists(id) { + t.Fatal("a revoked row in another workspace was deleted; the delete must stay workspace-scoped") + } + }) +} + +// TestChannelStore_ReinstallReactivationSemantics exercises the full +// finishSuccess ordering (cleanup-then-upsert) against a real database and +// pins the product behavior the fix protects: +// +// - SAME agent reconnect: the revoked row is reactivated in place, keeping its +// installation_id and every member/chat binding hanging off it. +// - DIFFERENT agent rebind: a fresh installation_id is created and the old +// agent's revoked row is removed so it no longer blocks the app_id slot. +func TestChannelStore_ReinstallReactivationSemantics(t *testing.T) { + pool := channelScopeTestDB(t) + ctx := context.Background() + store := NewChannelStore(db.New(pool)) + + apps := []string{rbAppReactivate, rbAppMove} + clean := func() { + _, _ = pool.Exec(ctx, `DELETE FROM channel_installation WHERE config->>'app_id' = ANY($1)`, apps) + _, _ = pool.Exec(ctx, `DELETE FROM channel_user_binding WHERE multica_user_id = $1`, rbUser) + _, _ = pool.Exec(ctx, `DELETE FROM channel_chat_session_binding WHERE chat_session_id = $1`, rbChatSess) + } + clean() + t.Cleanup(clean) + + insertRevoked := func(app, agent string) pgtype.UUID { + var id string + if err := pool.QueryRow(ctx, ` +INSERT INTO channel_installation (workspace_id, agent_id, channel_type, config, installer_user_id, status) +VALUES ($1, $2, 'feishu', jsonb_build_object('app_id', $3::text), $4, 'revoked') +RETURNING id +`, rbWS, agent, app, rbInstaller).Scan(&id); err != nil { + t.Fatalf("insert revoked installation: %v", err) + } + return util.MustParseUUID(id) + } + // Attach a member binding + chat-session binding to an installation, the way + // a real workspace accumulates them while the bot is connected. + attachBindings := func(installID pgtype.UUID) { + if _, err := pool.Exec(ctx, ` +INSERT INTO channel_user_binding (workspace_id, multica_user_id, installation_id, channel_type, channel_user_id) +VALUES ($1, $2, $3, 'feishu', 'ou_rb_user') +`, rbWS, rbUser, installID); err != nil { + t.Fatalf("insert user binding: %v", err) + } + if _, err := pool.Exec(ctx, ` +INSERT INTO channel_chat_session_binding (chat_session_id, installation_id, channel_type, channel_chat_id, chat_type) +VALUES ($1, $2, 'feishu', 'oc_rb_chat', 'p2p') +`, rbChatSess, installID); err != nil { + t.Fatalf("insert chat-session binding: %v", err) + } + } + countBindingsOn := func(installID pgtype.UUID) (users, chats int) { + if err := pool.QueryRow(ctx, `SELECT count(*) FROM channel_user_binding WHERE installation_id = $1`, installID).Scan(&users); err != nil { + t.Fatalf("count user bindings: %v", err) + } + if err := pool.QueryRow(ctx, `SELECT count(*) FROM channel_chat_session_binding WHERE installation_id = $1`, installID).Scan(&chats); err != nil { + t.Fatalf("count chat bindings: %v", err) + } + return + } + + upsert := func(agent, app string) Installation { + inst, err := store.UpsertLarkInstallation(ctx, UpsertInstallationParams{ + WorkspaceID: util.MustParseUUID(rbWS), + AgentID: util.MustParseUUID(agent), + AppID: app, + AppSecretEncrypted: []byte{1, 2, 3}, + BotOpenID: "ou_rb_bot", + InstallerUserID: util.MustParseUUID(rbInstaller), + Region: "feishu", + }) + if err != nil { + t.Fatalf("UpsertLarkInstallation: %v", err) + } + return inst + } + + t.Run("same agent reconnect keeps installation_id and bindings", func(t *testing.T) { + clean() + oldID := insertRevoked(rbAppReactivate, rbAgentA) + attachBindings(oldID) + + // finishSuccess order: cleanup for the current agent (a no-op for the + // same agent), then upsert. + if err := store.RemoveRevokedInstallationByAppID(ctx, util.MustParseUUID(rbWS), util.MustParseUUID(rbAgentA), rbAppReactivate); err != nil { + t.Fatalf("cleanup: %v", err) + } + inst := upsert(rbAgentA, rbAppReactivate) + + if inst.ID != oldID { + t.Fatalf("same agent reconnect changed installation_id: got %v, want %v (in-place reactivation lost)", inst.ID, oldID) + } + if inst.Status != "active" { + t.Fatalf("reactivated installation status=%q, want active", inst.Status) + } + if users, chats := countBindingsOn(oldID); users != 1 || chats != 1 { + t.Fatalf("bindings not preserved on reconnect: users=%d chats=%d, want 1/1", users, chats) + } + }) + + t.Run("different agent rebind gets a fresh installation_id", func(t *testing.T) { + clean() + oldID := insertRevoked(rbAppMove, rbAgentA) + attachBindings(oldID) + + if err := store.RemoveRevokedInstallationByAppID(ctx, util.MustParseUUID(rbWS), util.MustParseUUID(rbAgentB), rbAppMove); err != nil { + t.Fatalf("cleanup: %v", err) + } + inst := upsert(rbAgentB, rbAppMove) + + if inst.ID == oldID { + t.Fatal("different agent rebind reused the old installation_id; the blocking revoked row was not cleared") + } + if inst.Status != "active" { + t.Fatalf("new installation status=%q, want active", inst.Status) + } + // The old revoked row is gone (its unique app_id slot is freed for B). + if _, err := store.GetLarkInstallation(ctx, oldID); !errors.Is(err, pgx.ErrNoRows) { + t.Fatalf("old agent's revoked row still present after rebind: err=%v", err) + } + }) +} + +// TestChannelStore_RebindCleansDependentRows pins the application-layer cleanup: +// channel_* has no FK/cascade, so hard-deleting the blocking revoked +// installation of a DIFFERENT agent must also clear every row that references it +// (chat-session bindings, pending binding tokens, member links) and detach +// inbound-audit rows by NULLing installation_id — not leave dangling dead rows. +func TestChannelStore_RebindCleansDependentRows(t *testing.T) { + pool := channelScopeTestDB(t) + ctx := context.Background() + store := NewChannelStore(db.New(pool)) + + const ( + app = "cli_rb_cleanup" + tokenHash = "rb_token_hash_cleanup" + auditEvent = "ev_rb_cleanup" + ) + clean := func() { + _, _ = pool.Exec(ctx, `DELETE FROM channel_installation WHERE config->>'app_id' = $1`, app) + _, _ = pool.Exec(ctx, `DELETE FROM channel_user_binding WHERE multica_user_id = $1`, rbUser) + _, _ = pool.Exec(ctx, `DELETE FROM channel_chat_session_binding WHERE chat_session_id = $1`, rbChatSess) + _, _ = pool.Exec(ctx, `DELETE FROM channel_binding_token WHERE token_hash = $1`, tokenHash) + _, _ = pool.Exec(ctx, `DELETE FROM channel_inbound_audit WHERE channel_event_id = $1`, auditEvent) + } + clean() + t.Cleanup(clean) + + // A revoked installation for agent A carrying the full spread of dependents. + var oldID string + if err := pool.QueryRow(ctx, ` +INSERT INTO channel_installation (workspace_id, agent_id, channel_type, config, installer_user_id, status) +VALUES ($1, $2, 'feishu', jsonb_build_object('app_id', $3::text), $4, 'revoked') +RETURNING id +`, rbWS, rbAgentA, app, rbInstaller).Scan(&oldID); err != nil { + t.Fatalf("insert revoked installation: %v", err) + } + seed := func(q string, args ...any) { + if _, err := pool.Exec(ctx, q, args...); err != nil { + t.Fatalf("seed dependent row: %v", err) + } + } + seed(`INSERT INTO channel_user_binding (workspace_id, multica_user_id, installation_id, channel_type, channel_user_id) +VALUES ($1, $2, $3, 'feishu', 'ou_rb_user')`, rbWS, rbUser, oldID) + seed(`INSERT INTO channel_chat_session_binding (chat_session_id, installation_id, channel_type, channel_chat_id, chat_type) +VALUES ($1, $2, 'feishu', 'oc_rb_chat', 'p2p')`, rbChatSess, oldID) + seed(`INSERT INTO channel_binding_token (token_hash, workspace_id, installation_id, channel_type, channel_user_id, expires_at) +VALUES ($1, $2, $3, 'feishu', 'ou_rb_user', now() + interval '10 minutes')`, tokenHash, rbWS, oldID) + seed(`INSERT INTO channel_inbound_audit (installation_id, channel_type, event_type, channel_event_id, drop_reason) +VALUES ($1, 'feishu', 'im.message.receive_v1', $2, 'revoked_installation')`, oldID, auditEvent) + + // Rebind the app to a DIFFERENT agent. + if err := store.RemoveRevokedInstallationByAppID(ctx, util.MustParseUUID(rbWS), util.MustParseUUID(rbAgentB), app); err != nil { + t.Fatalf("RemoveRevokedInstallationByAppID: %v", err) + } + + count := func(q string, args ...any) int { + var n int + if err := pool.QueryRow(ctx, q, args...).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + return n + } + if n := count(`SELECT count(*) FROM channel_installation WHERE id = $1`, oldID); n != 0 { + t.Fatalf("blocking installation not deleted: %d rows remain", n) + } + if n := count(`SELECT count(*) FROM channel_user_binding WHERE installation_id = $1`, oldID); n != 0 { + t.Fatalf("member links not cleaned: %d dangling rows", n) + } + if n := count(`SELECT count(*) FROM channel_chat_session_binding WHERE installation_id = $1`, oldID); n != 0 { + t.Fatalf("chat-session bindings not cleaned: %d dangling rows (outbound patcher would error)", n) + } + if n := count(`SELECT count(*) FROM channel_binding_token WHERE installation_id = $1`, oldID); n != 0 { + t.Fatalf("binding tokens not cleaned: %d redeemable rows into a deleted installation", n) + } + // Audit history is preserved but detached: no row still points at the + // deleted installation, and our audit row survives with a NULL reference. + if n := count(`SELECT count(*) FROM channel_inbound_audit WHERE installation_id = $1`, oldID); n != 0 { + t.Fatalf("audit rows still reference the deleted installation: %d dangling ids", n) + } + if n := count(`SELECT count(*) FROM channel_inbound_audit WHERE channel_event_id = $1 AND installation_id IS NULL`, auditEvent); n != 1 { + t.Fatalf("audit row should survive detached (installation_id NULL), got %d", n) + } +} + +// TestChannelStore_RebindGuardedDeleteRaceWithReactivation exercises the real +// concurrency the guarded delete protects against. Two transactions race on one +// revoked installation: +// +// - txReconnect (agent A reconnecting to the SAME agent) reactivates the row +// to 'active' but holds the row lock uncommitted; +// - txRebind (agent B rebinding to a DIFFERENT agent) runs the full cleanup +// via RemoveRevokedInstallationByAppID. +// +// The old read-then-clean-then-delete shape would read the still-committed +// 'revoked' row, wipe its dependents, then no-op on the fenced delete — losing +// A's bindings even though A's installation survives. The guarded delete instead +// blocks on A's row lock; once A commits the reactivation it re-checks +// status='revoked' (EvalPlanQual under READ COMMITTED), claims nothing, and the +// cleanup keyed off the RETURNING id never runs. The installation and every +// binding must be intact. The assertion is timing-independent — it holds for the +// fixed code in every interleaving — so the test can never fail spuriously. +func TestChannelStore_RebindGuardedDeleteRaceWithReactivation(t *testing.T) { + pool := channelScopeTestDB(t) + ctx := context.Background() + store := NewChannelStore(db.New(pool)) + + const ( + app = "cli_rb_race" + tokenHash = "rb_token_hash_race" + auditEvent = "ev_rb_race" + ) + clean := func() { + _, _ = pool.Exec(ctx, `DELETE FROM channel_installation WHERE config->>'app_id' = $1`, app) + _, _ = pool.Exec(ctx, `DELETE FROM channel_user_binding WHERE multica_user_id = $1`, rbUser) + _, _ = pool.Exec(ctx, `DELETE FROM channel_chat_session_binding WHERE chat_session_id = $1`, rbChatSess) + _, _ = pool.Exec(ctx, `DELETE FROM channel_binding_token WHERE token_hash = $1`, tokenHash) + _, _ = pool.Exec(ctx, `DELETE FROM channel_inbound_audit WHERE channel_event_id = $1`, auditEvent) + } + clean() + t.Cleanup(clean) + + // A revoked installation for agent A, with the full spread of dependents. + var idStr string + if err := pool.QueryRow(ctx, ` +INSERT INTO channel_installation (workspace_id, agent_id, channel_type, config, installer_user_id, status) +VALUES ($1, $2, 'feishu', jsonb_build_object('app_id', $3::text), $4, 'revoked') +RETURNING id +`, rbWS, rbAgentA, app, rbInstaller).Scan(&idStr); err != nil { + t.Fatalf("insert revoked installation: %v", err) + } + seed := func(q string, args ...any) { + if _, err := pool.Exec(ctx, q, args...); err != nil { + t.Fatalf("seed dependent row: %v", err) + } + } + seed(`INSERT INTO channel_user_binding (workspace_id, multica_user_id, installation_id, channel_type, channel_user_id) +VALUES ($1, $2, $3, 'feishu', 'ou_rb_user')`, rbWS, rbUser, idStr) + seed(`INSERT INTO channel_chat_session_binding (chat_session_id, installation_id, channel_type, channel_chat_id, chat_type) +VALUES ($1, $2, 'feishu', 'oc_rb_chat', 'p2p')`, rbChatSess, idStr) + seed(`INSERT INTO channel_binding_token (token_hash, workspace_id, installation_id, channel_type, channel_user_id, expires_at) +VALUES ($1, $2, $3, 'feishu', 'ou_rb_user', now() + interval '10 minutes')`, tokenHash, rbWS, idStr) + seed(`INSERT INTO channel_inbound_audit (installation_id, channel_type, event_type, channel_event_id, drop_reason) +VALUES ($1, 'feishu', 'im.message.receive_v1', $2, 'revoked_installation')`, idStr, auditEvent) + + // txReconnect: agent A reactivates the row and holds the lock uncommitted. + txReconnect, err := pool.Begin(ctx) + if err != nil { + t.Fatalf("begin txReconnect: %v", err) + } + defer txReconnect.Rollback(ctx) + if _, err := txReconnect.Exec(ctx, `UPDATE channel_installation SET status = 'active' WHERE id = $1`, idStr); err != nil { + t.Fatalf("reactivate in txReconnect: %v", err) + } + + // txRebind: agent B's cleanup runs on its own transaction. Its guarded delete + // blocks on txReconnect's row lock. + done := make(chan error, 1) + go func() { + txRebind, err := pool.Begin(ctx) + if err != nil { + done <- err + return + } + defer txRebind.Rollback(ctx) + if err := store.WithTx(txRebind).RemoveRevokedInstallationByAppID(ctx, util.MustParseUUID(rbWS), util.MustParseUUID(rbAgentB), app); err != nil { + done <- err + return + } + done <- txRebind.Commit(ctx) + }() + + // Let txRebind reach and block on the guarded delete, then let A win the race. + time.Sleep(300 * time.Millisecond) + if err := txReconnect.Commit(ctx); err != nil { + t.Fatalf("commit txReconnect: %v", err) + } + select { + case err := <-done: + if err != nil { + t.Fatalf("txRebind: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("txRebind did not complete — possible deadlock in the guarded delete") + } + + count := func(q string, args ...any) int { + var n int + if err := pool.QueryRow(ctx, q, args...).Scan(&n); err != nil { + t.Fatalf("count: %v", err) + } + return n + } + if n := count(`SELECT count(*) FROM channel_installation WHERE id = $1 AND status = 'active'`, idStr); n != 1 { + t.Fatalf("reactivated installation was deleted by the racing rebind: %d active rows", n) + } + if n := count(`SELECT count(*) FROM channel_user_binding WHERE installation_id = $1`, idStr); n != 1 { + t.Fatalf("member link wiped by the racing rebind: got %d, want 1", n) + } + if n := count(`SELECT count(*) FROM channel_chat_session_binding WHERE installation_id = $1`, idStr); n != 1 { + t.Fatalf("chat-session binding wiped by the racing rebind: got %d, want 1", n) + } + if n := count(`SELECT count(*) FROM channel_binding_token WHERE installation_id = $1`, idStr); n != 1 { + t.Fatalf("binding token wiped by the racing rebind: got %d, want 1", n) + } + if n := count(`SELECT count(*) FROM channel_inbound_audit WHERE installation_id = $1`, idStr); n != 1 { + t.Fatalf("audit reference detached by the racing rebind: got %d, want 1", n) + } +} diff --git a/server/internal/integrations/lark/registration_service.go b/server/internal/integrations/lark/registration_service.go index 5191f29c77..9776659dae 100644 --- a/server/internal/integrations/lark/registration_service.go +++ b/server/internal/integrations/lark/registration_service.go @@ -563,6 +563,22 @@ func (s *RegistrationService) finishSuccess(ctx context.Context, sess *registrat defer tx.Rollback(ctx) qtx := s.queries.WithTx(tx) + // If the same Feishu app (app_id) was previously bound to a DIFFERENT + // agent in this workspace and later revoked, that revoked row still + // holds the (channel_type, config->>'app_id') unique index slot and + // blocks the UpsertChannelInstallation INSERT below. Remove the + // revoked placeholder first — the transaction wraps both the delete + // and the upsert so a failure between them rolls back cleanly. The + // current agent (sess.agentID) is excluded: re-connecting the SAME + // agent reactivates its own revoked row in place via the upsert's + // ON CONFLICT, keeping its installation_id and every binding intact. + if err := qtx.RemoveRevokedInstallationByAppID(ctx, sess.workspaceID, sess.agentID, res.ClientID); err != nil { + s.cfg.Logger.Warn("lark registration: cleanup revoked installation", + "session_id", sess.id, "err", err) + sess.markError(RegistrationReasonInternalError, err.Error(), s.gcDeadline()) + return + } + inst, err := qtx.UpsertLarkInstallation(ctx, UpsertInstallationParams{ WorkspaceID: sess.workspaceID, AgentID: sess.agentID, diff --git a/server/pkg/db/generated/channel.sql.go b/server/pkg/db/generated/channel.sql.go index 0f770a41fb..edd1439bc8 100644 --- a/server/pkg/db/generated/channel.sql.go +++ b/server/pkg/db/generated/channel.sql.go @@ -350,6 +350,22 @@ func (q *Queries) CreateChannelUserBinding(ctx context.Context, arg CreateChanne 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 @@ -384,6 +400,24 @@ func (q *Queries) DeleteChannelChatSessionBindingsByInstallation(ctx context.Con 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 @@ -402,6 +436,61 @@ func (q *Queries) DeleteChannelUserBindingsByWorkspaceMember(ctx context.Context 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 @@ -904,6 +993,23 @@ func (q *Queries) MarkChannelInboundDedupProcessed(ctx context.Context, arg Mark 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 diff --git a/server/pkg/db/queries/channel.sql b/server/pkg/db/queries/channel.sql index 71de3af193..b95bbeb5c7 100644 --- a/server/pkg/db/queries/channel.sql +++ b/server/pkg/db/queries/channel.sql @@ -98,6 +98,41 @@ SELECT * FROM channel_installation WHERE channel_type = sqlc.arg('channel_type') AND config ->> 'app_id' = sqlc.arg('app_id')::text; +-- name: DeleteRevokedChannelInstallationByAppID :one +-- 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. +DELETE FROM channel_installation +WHERE channel_type = sqlc.arg('channel_type') + AND config ->> 'app_id' = sqlc.arg('app_id')::text + AND workspace_id = sqlc.arg('workspace_id') + AND agent_id <> sqlc.arg('agent_id') + AND status = 'revoked' +RETURNING id; + -- name: ListChannelInstallationsByWorkspace :many -- Scoped by channel_type so a per-channel management surface (e.g. the Lark -- installation list) only ever sees its own platform's installations. @@ -258,6 +293,18 @@ LIMIT 1; DELETE FROM channel_user_binding WHERE workspace_id = $1 AND multica_user_id = $2; +-- name: DeleteChannelUserBindingsByInstallation :exec +-- 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. +DELETE FROM channel_user_binding +WHERE installation_id = $1; + -- ===================== -- channel_chat_session_binding -- ===================== @@ -389,6 +436,17 @@ WHERE installation_id = $1 ORDER BY received_at DESC LIMIT $2 OFFSET $3; +-- name: NullChannelInboundAuditInstallationID :exec +-- 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. +UPDATE channel_inbound_audit +SET installation_id = NULL +WHERE installation_id = $1; + -- ===================== -- channel_outbound_card_message -- ===================== @@ -445,3 +503,13 @@ RETURNING *; -- name: PurgeExpiredChannelBindingTokens :exec DELETE FROM channel_binding_token WHERE expires_at < $1; + +-- name: DeleteChannelBindingTokensByInstallation :exec +-- 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. +DELETE FROM channel_binding_token +WHERE installation_id = $1;