Delivery threads: dedupe and enforce a unique key

packages/chat/src/migrations.ts

@@ -468,6 +468,57 @@ export const chatMigrations: readonly ChatMigration[] = [
         WHERE "kind" = 'reply';
     `,
   },
+  {
+    // CL-7199: `createDeliveryThread` select-then-inserted with no
+    // unique constraint backing the read, same class of bug as 0025's
+    // root/reply case — a redelivered routine event could insert a
+    // second delivery thread for the same run. Dedupe existing
+    // duplicates — keeping the oldest row per (tenant, workbench,
+    // run_ref) — before the partial unique index below makes a repeat
+    // impossible, repointing every reference to a dropped duplicate's
+    // id first. Rows with a null `run_ref` are excluded from the
+    // dedupe: the window partition below groups nulls together, but
+    // the partial index treats them as distinct, so collapsing them
+    // here would delete rows the index was never going to reject.
+    name: "0026_workbench_threads_delivery_key",
+    sql: `
+      CREATE TEMP TABLE "delivery_thread_dedupe_map" ON COMMIT DROP AS
+      SELECT "id" AS "drop_id", "keep_id" FROM (
+        SELECT
+          "id",
+          first_value("id") OVER (
+            PARTITION BY "tenant_id", "workbench_id", "run_ref"
+            ORDER BY "created_at", "id"
+          ) AS "keep_id"
+        FROM "chat"."workbench_threads"
+        WHERE "kind" = 'delivery' AND "run_ref" IS NOT NULL
+      ) "ranked"
+      WHERE "id" <> "keep_id";
+
+      UPDATE "chat"."workbench_thread_messages" "wtm"
+        SET "thread_id" = "m"."keep_id"
+        FROM "delivery_thread_dedupe_map" "m"
+        WHERE "wtm"."thread_id" = "m"."drop_id";
+
+      UPDATE "chat"."workbench_messages" "wm"
+        SET "thread_id" = "m"."keep_id"
+        FROM "delivery_thread_dedupe_map" "m"
+        WHERE "wm"."thread_id" = "m"."drop_id";
+
+      UPDATE "chat"."workbench_threads" "wt"
+        SET "parent_thread_id" = "m"."keep_id"
+        FROM "delivery_thread_dedupe_map" "m"
+        WHERE "wt"."parent_thread_id" = "m"."drop_id";
+
+      DELETE FROM "chat"."workbench_threads" "wt"
+        USING "delivery_thread_dedupe_map" "m"
+        WHERE "wt"."id" = "m"."drop_id";
+
+      CREATE UNIQUE INDEX IF NOT EXISTS "workbench_threads_delivery_key"
+        ON "chat"."workbench_threads" ("tenant_id", "workbench_id", "run_ref")
+        WHERE "kind" = 'delivery' AND "run_ref" IS NOT NULL;
+    `,
+  },
 ];
 
 /**

packages/chat/src/schema.ts

@@ -237,6 +237,13 @@ export const workbenchThreads = chatSchema.table(
     uniqueIndex("workbench_threads_reply_key")
       .on(table.tenantId, table.workbenchId, table.parentMessageId)
       .where(sql`${table.kind} = 'reply'`),
+    // `run_ref` is nullable (root/reply rows never set it), and a
+    // partial unique index treats NULLs as distinct rather than equal
+    // — so the predicate excludes them explicitly rather than relying
+    // on every delivery-thread caller to always supply one.
+    uniqueIndex("workbench_threads_delivery_key")
+      .on(table.tenantId, table.workbenchId, table.runRef)
+      .where(sql`${table.kind} = 'delivery' AND ${table.runRef} IS NOT NULL`),
   ],
 );
 

packages/chat/src/threads.ts

@@ -7,7 +7,7 @@
 // mention fan-out (see codec.ts). Message-id reply correlation is a
 // workbench concern here — do not fork Interchange mail to change that.
 
-import { and, eq, asc } from "drizzle-orm";
+import { and, eq, asc, sql } from "drizzle-orm";
 import type { PostgresJsDatabase } from "drizzle-orm/postgres-js";
 
 import { workbenchThreadMessages, workbenchThreads } from "./schema";
@@ -358,14 +358,6 @@ export type ThreadDb<
   TSchema extends Record<string, unknown> = Record<string, never>,
 > = PostgresJsDatabase<TSchema>;
 
-function requireReturningRow<T>(rows: readonly T[], what: string): T {
-  const row = rows[0];
-  if (row === undefined) {
-    throw new Error(`expected ${what} row from returning()`);
-  }
-  return row;
-}
-
 function mapThreadRow(
   row: typeof workbenchThreads.$inferSelect,
 ): WorkbenchThread {
@@ -556,23 +548,36 @@ export function createDrizzleThreadStore<
     return reselected;
   }
 
+  async function selectDeliveryThread(
+    tenantId: string,
+    workbenchId: string,
+    runRef: string,
+  ): Promise<WorkbenchThread | undefined> {
+    const rows = await db
+      .select()
+      .from(workbenchThreads)
+      .where(
+        and(
+          eq(workbenchThreads.tenantId, tenantId),
+          eq(workbenchThreads.workbenchId, workbenchId),
+          eq(workbenchThreads.kind, "delivery"),
+          eq(workbenchThreads.runRef, runRef),
+        ),
+      )
+      .limit(1);
+    return rows[0] ? mapThreadRow(rows[0]) : undefined;
+  }
+
   return {
     ensureRootThread,
 
     async createDeliveryThread(input) {
-      const existing = await db
-        .select()
-        .from(workbenchThreads)
-        .where(
-          and(
-            eq(workbenchThreads.tenantId, input.tenantId),
-            eq(workbenchThreads.workbenchId, input.workbenchId),
-            eq(workbenchThreads.kind, "delivery"),
-            eq(workbenchThreads.runRef, input.runRef),
-          ),
-        )
-        .limit(1);
-      if (existing[0]) return mapThreadRow(existing[0]);
+      // Insert-first, not select-then-insert: two concurrent first
+      // deliveries of the same run both attempt the insert, the
+      // partial unique index (tenant_id, workbench_id, run_ref) WHERE
+      // kind = 'delivery' AND run_ref IS NOT NULL serializes them, and
+      // the loser's empty `returning()` re-selects the winner's row
+      // rather than creating a duplicate delivery thread.
       const id = newThreadId();
       const inserted = await db
         .insert(workbenchThreads)
@@ -586,8 +591,32 @@ export function createDrizzleThreadStore<
           runRef: input.runRef,
           title: input.title ?? null,
         })
+        .onConflictDoNothing({
+          target: [
+            workbenchThreads.tenantId,
+            workbenchThreads.workbenchId,
+            workbenchThreads.runRef,
+          ],
+          // Matches `workbench_threads_delivery_key`'s predicate in
+          // ./schema.ts exactly — the ON CONFLICT arbiter predicate
+          // must match the partial index's own predicate, not just be
+          // implied by it.
+          where: sql`${workbenchThreads.kind} = 'delivery' AND ${workbenchThreads.runRef} IS NOT NULL`,
+        })
         .returning();
-      return mapThreadRow(requireReturningRow(inserted, "delivery thread"));
+      const row = inserted[0];
+      if (row) return mapThreadRow(row);
+      const reselected = await selectDeliveryThread(
+        input.tenantId,
+        input.workbenchId,
+        input.runRef,
+      );
+      if (!reselected) {
+        throw new Error(
+          "expected delivery thread row after conflicting insert",
+        );
+      }
+      return reselected;
     },
 
     openReplyThread: (input) => anchoredReplyThread(input, "reply"),

packages/chat/test/migrations.test.ts

@@ -46,6 +46,7 @@ const migrationNames = [
   "0023_drop_workbench_host_arm",
   "0024_workbench_launch_sources_digest",
   "0025_workbench_threads_unique_key",
+  "0026_workbench_threads_delivery_key",
 ];
 
 describeIfDb("applyChatMigrations", () => {
@@ -173,6 +174,12 @@ describeIfDb("applyChatMigrations", () => {
       );
       expect(threadIndexNames).toContain("workbench_threads_root_key");
       expect(threadIndexNames).toContain("workbench_threads_reply_key");
+
+      // CL-7199: `createDeliveryThread` insert-then-reselects on
+      // conflict the same way; this partial unique index is what makes
+      // the conflict possible instead of a silent duplicate delivery
+      // thread per run.
+      expect(threadIndexNames).toContain("workbench_threads_delivery_key");
     } finally {
       await sql.end();
     }
@@ -385,3 +392,182 @@ describeIfDb("0025_workbench_threads_unique_key dedupe", () => {
     }
   }, 120000);
 });
+
+describeIfDb("0026_workbench_threads_delivery_key dedupe", () => {
+  const scratchUrl = scratchUrlFor(
+    databaseUrl ?? "postgres://localhost:5432/unused",
+  ).replace("_chat_migrations_test", "_chat_migrations_delivery_dedupe_test");
+  const scratchTarget = new URL(scratchUrl);
+  const scratchDatabase = scratchTarget.pathname.replace(/^\//, "");
+
+  beforeAll(async () => {
+    const maintenanceUrl = new URL(scratchUrl);
+    maintenanceUrl.pathname = "/postgres";
+    const maintenance = postgres(maintenanceUrl.toString(), {
+      max: 1,
+      onnotice: () => undefined,
+    });
+    try {
+      await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`);
+      await maintenance.unsafe(`CREATE DATABASE "${scratchDatabase}"`);
+    } finally {
+      await maintenance.end();
+    }
+  }, 20000);
+
+  afterAll(async () => {
+    const maintenanceUrl = new URL(scratchUrl);
+    maintenanceUrl.pathname = "/postgres";
+    const maintenance = postgres(maintenanceUrl.toString(), {
+      max: 1,
+      onnotice: () => undefined,
+    });
+    try {
+      await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`);
+    } finally {
+      await maintenance.end();
+    }
+  }, 20000);
+
+  test("collapses duplicate delivery threads per run ref, repoints thread_id and parent_thread_id references, leaves null-run-ref rows untouched, and rejects a repeat insert", async () => {
+    // Replay every migration through 0025 by hand, seeding the
+    // duplicates 0026 must clean up before it exists to forbid them —
+    // then let `applyChatMigrations` run 0026 for real.
+    const preDeliveryKeyMigrations = chatMigrations.filter(
+      (migration) => migration.name !== "0026_workbench_threads_delivery_key",
+    );
+    const seed = postgres(scratchUrl, { max: 1, onnotice: () => undefined });
+    try {
+      await seed.unsafe(`CREATE SCHEMA IF NOT EXISTS "chat"`);
+      await seed.unsafe(
+        `CREATE TABLE IF NOT EXISTS "chat"."chat_migrations" ` +
+          `(name text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`,
+      );
+      for (const migration of preDeliveryKeyMigrations) {
+        await seed.begin(async (tx) => {
+          await tx.unsafe(migration.sql);
+          await tx.unsafe(
+            `INSERT INTO "chat"."chat_migrations" (name) VALUES ($1)`,
+            [migration.name],
+          );
+        });
+      }
+
+      await seed.unsafe(`
+        INSERT INTO "chat"."workbench_threads"
+          (id, tenant_id, workbench_id, kind, run_ref, created_at)
+        VALUES
+          ('thr_delivery_old', 'tnt_ddedupe', 'wb_ddedupe', 'delivery', 'run_1', now() - interval '1 hour'),
+          ('thr_delivery_new', 'tnt_ddedupe', 'wb_ddedupe', 'delivery', 'run_1', now())
+      `);
+      // Two delivery rows with a null run_ref: the window partition in
+      // the dedupe groups nulls together (unlike the partial unique
+      // index it backstops, which treats nulls as distinct), so these
+      // must survive untouched rather than being collapsed to one row.
+      await seed.unsafe(`
+        INSERT INTO "chat"."workbench_threads"
+          (id, tenant_id, workbench_id, kind, run_ref, created_at)
+        VALUES
+          ('thr_delivery_null_a', 'tnt_ddedupe', 'wb_ddedupe', 'delivery', NULL, now() - interval '1 hour'),
+          ('thr_delivery_null_b', 'tnt_ddedupe', 'wb_ddedupe', 'delivery', NULL, now())
+      `);
+      await seed.unsafe(`
+        INSERT INTO "chat"."workbench_thread_messages"
+          (tenant_id, workbench_id, thread_id, message_id)
+        VALUES ('tnt_ddedupe', 'wb_ddedupe', 'thr_delivery_new', 'msg_ddedupe')
+      `);
+      // A message parked in the dropped delivery duplicate — proves
+      // `workbench_messages.thread_id` is repointed too.
+      await seed.unsafe(`
+        INSERT INTO "chat"."workbench_messages"
+          (id, tenant_id, workbench_id, sender_address, thread_id, parts)
+        VALUES (
+          'msg_in_dropped_delivery', 'tnt_ddedupe', 'wb_ddedupe', 'addr_1',
+          'thr_delivery_new', '[]'::jsonb
+        )
+      `);
+      // A reply thread anchored directly on the dropped delivery
+      // duplicate (a reply to a message that lived in it) — proves
+      // `workbench_threads.parent_thread_id` is repointed, matching
+      // `containerThreadFor`/`resolveThreadAnchor` letting a delivery
+      // thread act as a reply's anchor.
+      await seed.unsafe(`
+        INSERT INTO "chat"."workbench_threads"
+          (id, tenant_id, workbench_id, kind, parent_message_id, parent_thread_id, created_at)
+        VALUES (
+          'thr_child_of_dropped_delivery', 'tnt_ddedupe', 'wb_ddedupe', 'reply',
+          'msg_child', 'thr_delivery_new', now()
+        )
+      `);
+    } finally {
+      await seed.end();
+    }
+
+    const report = await applyChatMigrations(scratchUrl);
+    expect(report.applied).toEqual(["0026_workbench_threads_delivery_key"]);
+
+    const verify = postgres(scratchUrl, { max: 1, onnotice: () => undefined });
+    try {
+      const deliveries = await verify.unsafe(
+        `SELECT id FROM "chat"."workbench_threads" ` +
+          `WHERE tenant_id = 'tnt_ddedupe' AND workbench_id = 'wb_ddedupe' ` +
+          `AND kind = 'delivery' AND run_ref = 'run_1'`,
+      );
+      expect(deliveries.map((row) => String(row["id"]))).toEqual([
+        "thr_delivery_old",
+      ]);
+
+      const nullRunRefRows = await verify.unsafe(
+        `SELECT id FROM "chat"."workbench_threads" ` +
+          `WHERE tenant_id = 'tnt_ddedupe' AND workbench_id = 'wb_ddedupe' ` +
+          `AND kind = 'delivery' AND run_ref IS NULL`,
+      );
+      expect(nullRunRefRows.map((row) => String(row["id"])).sort()).toEqual([
+        "thr_delivery_null_a",
+        "thr_delivery_null_b",
+      ]);
+
+      const membership = await verify.unsafe(
+        `SELECT thread_id FROM "chat"."workbench_thread_messages" WHERE message_id = 'msg_ddedupe'`,
+      );
+      expect(String(membership[0]?.["thread_id"])).toBe("thr_delivery_old");
+
+      const messageThreadId = await verify.unsafe(
+        `SELECT thread_id FROM "chat"."workbench_messages" WHERE id = 'msg_in_dropped_delivery'`,
+      );
+      expect(String(messageThreadId[0]?.["thread_id"])).toBe(
+        "thr_delivery_old",
+      );
+
+      const childParentThreadId = await verify.unsafe(
+        `SELECT parent_thread_id FROM "chat"."workbench_threads" WHERE id = 'thr_child_of_dropped_delivery'`,
+      );
+      expect(String(childParentThreadId[0]?.["parent_thread_id"])).toBe(
+        "thr_delivery_old",
+      );
+
+      // The delivery unique index holds: a second delivery row for the
+      // same (tenant, workbench, run_ref) key is now rejected rather
+      // than silently accepted as a second duplicate.
+      await expect(
+        (async () => {
+          await verify.unsafe(
+            `INSERT INTO "chat"."workbench_threads" ` +
+              `(id, tenant_id, workbench_id, kind, run_ref) ` +
+              `VALUES ('thr_delivery_conflict', 'tnt_ddedupe', 'wb_ddedupe', 'delivery', 'run_1')`,
+          );
+        })(),
+      ).rejects.toThrow();
+
+      // A null run_ref never conflicts, matching the partial index's
+      // predicate.
+      await verify.unsafe(
+        `INSERT INTO "chat"."workbench_threads" ` +
+          `(id, tenant_id, workbench_id, kind, run_ref) ` +
+          `VALUES ('thr_delivery_null_c', 'tnt_ddedupe', 'wb_ddedupe', 'delivery', NULL)`,
+      );
+    } finally {
+      await verify.end();
+    }
+  }, 120000);
+});