From 2daf746957ccf1d915fa42b34c2dfaabd3a8dde6 Mon Sep 17 00:00:00 2001 From: "detail-app[bot]" <180357370+detail-app[bot]@users.noreply.github.com> Date: Fri, 11 Sep 2026 21:56:48 +0000 Subject: [PATCH] fix(api): reject stale same-target merge reversals via active merge event pointer --- .../contact-merge.service.integration.test.ts | 214 ++++++++++++++++++ .../api/src/services/contact-merge.service.ts | 16 +- packages/database/src/channel-spine-schema.ts | 18 ++ packages/database/src/client.ts | 1 + .../104_track_active_merge_event.ts | 59 +++++ packages/database/src/tenant-schema.ts | 1 + 6 files changed, 304 insertions(+), 5 deletions(-) create mode 100644 packages/database/src/migrations/104_track_active_merge_event.ts diff --git a/apps/api/src/services/contact-merge.service.integration.test.ts b/apps/api/src/services/contact-merge.service.integration.test.ts index ad09e119..41428290 100644 --- a/apps/api/src/services/contact-merge.service.integration.test.ts +++ b/apps/api/src/services/contact-merge.service.integration.test.ts @@ -550,6 +550,220 @@ describe("unmergeContacts", () => { ); }); +describe("unmergeContacts same-target re-merge", () => { + integrationTest( + "rejects reversing a stale merge after the same (source, target) pair was re-merged", + async () => { + const companyId = crypto.randomUUID(); + const schemaName = getSchemaName(companyId); + const ownerId = crypto.randomUUID(); + try { + await db + .insertInto("users") + .values({ + id: ownerId, + email: `contact-stale-${ownerId}@example.com`, + password_hash: "test", + }) + .execute(); + await db + .insertInto("companies") + .values({ + id: companyId, + name: "Contact stale unmerge test", + schema_name: schemaName, + status: "active", + }) + .execute(); + await db + .insertInto("sla_policies") + .values({ + company_id: companyId, + target_minutes: 60, + direct_resolution_target_minutes: 480, + group_response_target_minutes: 120, + group_resolution_target_minutes: 960, + timezone: "UTC", + weekly_schedule: JSON.stringify(DEFAULT_SLA_WEEKLY_SCHEDULE), + exceptions: JSON.stringify([]), + effective_from: new Date("1970-01-01T00:00:00Z"), + created_by: ownerId, + }) + .execute(); + await createTenantSchema(companyId); + await reconcileChannelSpineConcurrentIndexes(db, schemaName); + const tenantDb = getTenantConnection(companyId); + + const account = crypto.randomUUID(); + await tenantDb + .insertInto("channel_accounts") + .values({ + id: account, + channel: "telegram", + provider: "telegram_bot", + display_name: "Bot", + status: "connected", + }) + .execute(); + const target = await tenantDb + .insertInto("contacts") + .values({ jid: "60123456789@s.whatsapp.net", push_name: "Ada" }) + .returning("id") + .executeTakeFirstOrThrow(); + const source = await tenantDb + .insertInto("contacts") + .values({ jid: null, push_name: "Ada (Telegram)" }) + .returning("id") + .executeTakeFirstOrThrow(); + const sourceEndpoint = await tenantDb + .insertInto("contact_endpoints") + .values({ + contact_id: source.id, + channel: "telegram", + provider: "telegram_bot", + channel_account_id: account, + endpoint_kind: "person", + external_id: "tg-stale", + identity_scope: "telegram:user", + }) + .returning("id") + .executeTakeFirstOrThrow(); + + // merge A -> unmerge A -> merge B, reusing the same (source, target). + const first = await mergeContacts(tenantDb, { + sourceContactId: source.id, + targetContactId: target.id, + actorUserId: ownerId, + reason: "first merge", + }); + await unmergeContacts(tenantDb, { + mergeEventId: first.mergeEventId, + actorUserId: ownerId, + reason: "reverse the first merge", + }); + const second = await mergeContacts(tenantDb, { + sourceContactId: source.id, + targetContactId: target.id, + actorUserId: ownerId, + reason: "re-merge the same pair", + }); + + // B is the merge in effect: the source points at the survivor and the + // active merge event pointer names B, not A. + const afterReMerge = await tenantDb + .selectFrom("contacts") + .select(["merged_into_contact_id", "active_merge_event_id"]) + .where("id", "=", source.id) + .executeTakeFirstOrThrow(); + expect(afterReMerge.merged_into_contact_id).toBe(target.id); + expect(afterReMerge.active_merge_event_id).toBe(second.mergeEventId); + + // Reversing the stale event A must be rejected as superseded, even + // though A and B share the same survivor. Before the fix this passed + // and orphaned B's audit rows. + await expect( + unmergeContacts(tenantDb, { + mergeEventId: first.mergeEventId, + actorUserId: ownerId, + reason: "reverse the stale first merge", + }), + ).rejects.toBeInstanceOf(ValidationError); + + // The rejection left the in-effect merge untouched: the endpoint is + // still on the survivor, the source is still archived, and B is still + // the active merge event. + expect( + ( + await tenantDb + .selectFrom("contact_endpoints") + .select("contact_id") + .where("id", "=", sourceEndpoint.id) + .executeTakeFirstOrThrow() + ).contact_id, + ).toBe(target.id); + const untouched = await tenantDb + .selectFrom("contacts") + .select([ + "merged_into_contact_id", + "active_merge_event_id", + "archived_at", + ]) + .where("id", "=", source.id) + .executeTakeFirstOrThrow(); + expect(untouched.merged_into_contact_id).toBe(target.id); + expect(untouched.active_merge_event_id).toBe(second.mergeEventId); + expect(untouched.archived_at).not.toBeNull(); + + // The in-effect event B is still correctable, and correcting it + // restores the endpoint and revives the source as a first unmerge + // would. + const undoneB = await unmergeContacts(tenantDb, { + mergeEventId: second.mergeEventId, + actorUserId: ownerId, + reason: "correct the in-effect merge", + }); + expect(undoneB.mergeEventId).toBe(second.mergeEventId); + expect(undoneB.restoredEndpoints).toBe(1); + expect(undoneB.skippedEndpoints).toBe(0); + expect( + ( + await tenantDb + .selectFrom("contact_endpoints") + .select("contact_id") + .where("id", "=", sourceEndpoint.id) + .executeTakeFirstOrThrow() + ).contact_id, + ).toBe(source.id); + const revived = await tenantDb + .selectFrom("contacts") + .select([ + "merged_into_contact_id", + "active_merge_event_id", + "archived_at", + ]) + .where("id", "=", source.id) + .executeTakeFirstOrThrow(); + expect(revived.merged_into_contact_id).toBeNull(); + expect(revived.active_merge_event_id).toBeNull(); + expect(revived.archived_at).toBeNull(); + expect(await resolveCanonicalContactId(tenantDb, source.id)).toBe( + source.id, + ); + + // After B is unmerged, A remains superseded (its pointer was cleared + // by the first unmerge and never re-pointed at A), so reversing A a + // second time is still rejected. + await expect( + unmergeContacts(tenantDb, { + mergeEventId: first.mergeEventId, + actorUserId: ownerId, + reason: "reverse A after B", + }), + ).rejects.toBeInstanceOf(ValidationError); + await expect( + unmergeContacts(tenantDb, { + mergeEventId: second.mergeEventId, + actorUserId: ownerId, + reason: "B again", + }), + ).rejects.toBeInstanceOf(ValidationError); + } finally { + await clearTenantConnection(companyId); + await sql + .raw(`DROP SCHEMA IF EXISTS "${schemaName}" CASCADE`) + .execute(db); + await db + .deleteFrom("sla_policies") + .where("company_id", "=", companyId) + .execute(); + await db.deleteFrom("companies").where("id", "=", companyId).execute(); + await db.deleteFrom("users").where("id", "=", ownerId).execute(); + } + }, + 60_000, + ); +}); + describe("suggestContactMerges placeholder addresses", () => { integrationTest( "never proposes merging two contacts that only share a placeholder number", diff --git a/apps/api/src/services/contact-merge.service.ts b/apps/api/src/services/contact-merge.service.ts index eb9e7966..ae860eda 100644 --- a/apps/api/src/services/contact-merge.service.ts +++ b/apps/api/src/services/contact-merge.service.ts @@ -140,6 +140,7 @@ export async function mergeContacts( .updateTable("contacts") .set({ merged_into_contact_id: target.id, + active_merge_event_id: mergeEvent.id, archived_at: new Date(), updated_at: new Date(), }) @@ -201,16 +202,20 @@ export async function unmergeContacts( } const source = await trx .selectFrom("contacts") - .select(["id", "merged_into_contact_id"]) + .select(["id", "merged_into_contact_id", "active_merge_event_id"]) .where("id", "=", mergeEvent.source_contact_id) + .forUpdate() .executeTakeFirst(); if (!source) { throw new ValidationError("The merged-away contact no longer exists"); } - // Only the merge that is currently in effect can be corrected. If the - // source was merged again afterwards, undoing this older event would - // revive it into a state that no longer describes anything. - if (source.merged_into_contact_id !== mergeEvent.target_contact_id) { + // Only the merge that is currently in effect can be corrected. The + // `active_merge_event_id` pointer names that specific event: once it is + // cleared (after an unmerge) or replaced (after a re-merge, even into the + // same survivor), the older event is superseded. Comparing the survivor + // id is not enough, because two events for the same `(source, target)` + // pair share a target. + if (source.active_merge_event_id !== mergeEvent.id) { throw new ValidationError( "This merge has already been superseded and cannot be reversed", ); @@ -265,6 +270,7 @@ export async function unmergeContacts( .updateTable("contacts") .set({ merged_into_contact_id: null, + active_merge_event_id: null, archived_at: null, updated_at: new Date(), }) diff --git a/packages/database/src/channel-spine-schema.ts b/packages/database/src/channel-spine-schema.ts index 7412f962..1cd60579 100644 --- a/packages/database/src/channel-spine-schema.ts +++ b/packages/database/src/channel-spine-schema.ts @@ -22,6 +22,7 @@ export async function ensureChannelSpineTenantSchema( ["record_kind", "TEXT"], ["merged_into_contact_id", "UUID"], ["archived_at", "TIMESTAMPTZ"], + ["active_merge_event_id", "UUID"], ]); await sql`CREATE TABLE IF NOT EXISTS ${table("channel_accounts")} ( @@ -558,6 +559,16 @@ export async function ensureChannelSpineTenantSchema( FOREIGN KEY (merged_into_contact_id) REFERENCES ${table("contacts")}(id) ON DELETE RESTRICT NOT VALID`, ); + await addConstraintIfMissing( + db, + schemaName, + "contacts", + "contacts_active_merge_event_fk", + sql`ALTER TABLE ${table("contacts")} + ADD CONSTRAINT contacts_active_merge_event_fk + FOREIGN KEY (active_merge_event_id) REFERENCES ${table("contact_merge_events")}(id) + ON DELETE RESTRICT NOT VALID`, + ); await addConstraintIfMissing( db, schemaName, @@ -711,6 +722,12 @@ async function ensureIndexes( sql`CREATE INDEX ${sql.ref(`${schemaName}_csrj_due_idx`)} ON ${table("channel_spine_reconciliation_journal")} (status, next_attempt_at, created_at)`, ], + [ + "contacts_active_merge_event_uidx", + sql`CREATE UNIQUE INDEX ${sql.ref(`${schemaName}_c_ame_uidx`)} + ON ${table("contacts")} (active_merge_event_id) + WHERE active_merge_event_id IS NOT NULL`, + ], ]; const existing = await sql<{ indexname: string }>` @@ -738,6 +755,7 @@ function indexNameFor(logicalName: string, schemaName: string): string { outbound_intents_send_message_uidx: "omi_send_msg_uidx", contact_suppressions_active_idx: "csup_active_idx", channel_spine_reconciliation_due_idx: "csrj_due_idx", + contacts_active_merge_event_uidx: "c_ame_uidx", }; return `${schemaName}_${suffixes[logicalName]}`; } diff --git a/packages/database/src/client.ts b/packages/database/src/client.ts index 77f80221..fee9fddc 100644 --- a/packages/database/src/client.ts +++ b/packages/database/src/client.ts @@ -770,6 +770,7 @@ export interface ContactsTable { avatar_url: string | null; record_kind: "customer" | "legacy_group_projection" | null; merged_into_contact_id: string | null; + active_merge_event_id: string | null; archived_at: Date | null; created_at: Generated; updated_at: Generated; diff --git a/packages/database/src/migrations/104_track_active_merge_event.ts b/packages/database/src/migrations/104_track_active_merge_event.ts new file mode 100644 index 00000000..c9e9b829 --- /dev/null +++ b/packages/database/src/migrations/104_track_active_merge_event.ts @@ -0,0 +1,59 @@ +import { type Kysely, sql } from "kysely"; +import { ensureChannelSpineTenantSchema } from "../channel-spine-schema.js"; +import { executeOnAllTenants } from "./migration-helpers.js"; + +/** + * Track the merge event currently in effect for a merged-away contact. + * + * `contacts.merged_into_contact_id` only names the survivor, not the specific + * `contact_merge_events` row that is active. Two merge events can share the + * same `(source, target)` pair (merge A -> unmerge A -> merge B), so the + * service-level supersede guard in `unmergeContacts` cannot tell A and B apart + * from the survivor id alone and would allow reversing the stale event A + * after B is in effect. `active_merge_event_id` is the single source of truth + * for "which merge event is in effect", set by `mergeContacts` and cleared by + * `unmergeContacts`. + * + * The reconcile/`ensureChannelSpineTenantSchema` path adds the column, the FK + * to `contact_merge_events(id)`, and a partial unique index (at most one + * active merge event per contact) for newly-created tenants. This migration + * applies the same additive schema to every existing tenant and backfills + * the column for contacts merged before it existed, so the service guard can + * identify the in-effect merge event for pre-existing merges too. + */ +export async function up(db: Kysely): Promise { + await executeOnAllTenants(db, async (schemaName) => { + await ensureChannelSpineTenantSchema(db, schemaName); + + const table = (name: string) => sql.raw(`"${schemaName}"."${name}"`); + + // Backfill `active_merge_event_id` for contacts that were merged before + // this column existed. The in-effect event for a merged source is the one + // whose target matches `merged_into_contact_id`; ties on `created_at` are + // broken by `id` so the choice is deterministic. Rows with no surviving + // matching event (none expected, since setting `merged_into_contact_id` + // always inserted a `contact_merge_events` row) are left null. + await sql` + UPDATE ${table("contacts")} AS source + SET active_merge_event_id = ( + SELECT merge_event.id + FROM ${table("contact_merge_events")} AS merge_event + WHERE merge_event.source_contact_id = source.id + AND merge_event.target_contact_id = source.merged_into_contact_id + ORDER BY merge_event.created_at DESC, merge_event.id DESC + LIMIT 1 + ) + WHERE source.merged_into_contact_id IS NOT NULL + AND source.active_merge_event_id IS NULL + `.execute(db); + }); +} + +/** + * Production migrations are forward-only. The down path is intentionally + * blocked because dropping the in-effect merge event pointer would strand + * existing merged contacts and break the unmerge correction path. + */ +export async function down(): Promise { + throw new Error("migration 104 is forward-only"); +} diff --git a/packages/database/src/tenant-schema.ts b/packages/database/src/tenant-schema.ts index 4fbb4c80..43bf8da3 100644 --- a/packages/database/src/tenant-schema.ts +++ b/packages/database/src/tenant-schema.ts @@ -365,6 +365,7 @@ export const TENANT_SCHEMA_CONTRACT = { "avatar_url", "record_kind", "merged_into_contact_id", + "active_merge_event_id", "archived_at", "created_at", "updated_at",