fix(api): reject stale same-target merge reversals via active merge event pointer

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",

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(),
       })

packages/database/src/channel-spine-schema.ts

@@ -22,6 +22,7 @@ export async function ensureChannelSpineTenantSchema<Database>(
       ["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<Database>(
         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<Database>(
         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]}`;
 }

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<Date>;
   updated_at: Generated<Date>;

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<unknown>): Promise<void> {
+  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<void> {
+  throw new Error("migration 104 is forward-only");
+}

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",