fix(core): race-safe matching, visibility-checked ID matches, child-preserving merges

- migration 0002: unique index on works (parent_id, position)
- matchWork retries the whole transaction on a unique violation (3 attempts)
- addMissingExternalIds uses ON CONFLICT DO NOTHING
- ID matches only return shared works or the user's own; private works cannot carry external ids
- adminMergeWorks requires shared works and repoints non-colliding children
- draft is computed for the aliased work id

Co-Authored-By: Claude Opus 5.5 <[email protected]>

docs/DECISIONS.md

@@ -81,3 +81,26 @@ instead of 26. Where `AGENTS.md` or the spec say "ULID", read "UUIDv7".
 
 **Consequences:** Rule changes happen in one file, and the pointer files must stay thin. Plans are working
 artefacts: kept after merge, they go stale and contradict the code, so the spec is the record.
+
+## 2026-10-30: Catalogue matching is race-safe through unique keys plus whole-transaction retry
+**Status:** accepted
+**Context:** Two syncs matching the same new work at once both miss the lookup and both insert. On
+Postgres the loser hit the `external_ids (namespace, value)` unique key and crashed the sync; seasons and
+episodes had no key at all and were duplicated.
+**Decision:**
+- Migration `0002_works_position_unique` adds a unique index on `works (parent_id, position)`. NULLs are
+  distinct on both dialects, so top-level works and position-less children are unaffected.
+- `matchWork` re-runs the whole resolve in a fresh transaction on a unique violation (SQLite
+  "UNIQUE constraint failed", Postgres SQLSTATE 23505), up to 3 attempts. Postgres aborts the transaction
+  on the error, so a statement-level retry is not possible.
+- Adding missing external IDs to an existing work uses `ON CONFLICT DO NOTHING`.
+- An ID match only counts if the work is shared or owned by the matching user, and private works can
+  never carry external IDs (`createWork` refuses; global merges accept only shared works).
+- A global merge repoints children of `from` to `to` unless `to` already has a child at that position.
+
+**Alternatives:**
+- Advisory or table locks: dialect-specific, and SQLite already serialises writes.
+- Upserts throughout `createWork`: would hide genuine conflicts and need per-dialect `RETURNING` handling.
+
+**Consequences:** Parents resolved without IDs or position (title/year reuse) still have no unique key
+and can be duplicated under concurrency; tracked in `docs/TODO.md`.

docs/TODO.md

@@ -23,4 +23,6 @@ Small, specific, independently actionable. Remove only when actually done.
     file, and compare steps and token use.
 
 ## Known issues
+- [ ] Matching: a shared parent resolved without external IDs or position (title/year reuse in
+  `packages/core/src/matching.ts`) has no unique key, so concurrent matches can create it twice.
 - [ ] `bun run typecheck` fails with TS18003 until source files exist under `apps/`, `packages/` or `plugins/`.

packages/core/src/matching.ts

@@ -67,12 +67,12 @@ interface Resolved {
 }
 
 async function resolve(k: K, userId: string | null, hint: WorkHint, actor: string): Promise<Resolved> {
-  // 1. Any external ID.
-  const byId = await findWorkByExternalIdsIn(k, hint.externalIds);
+  // 1. Any external ID on a work this user may see (parents resolve with userId null: shared only).
+  const byId = await findWorkByExternalIdsIn(k, hint.externalIds, userId);
   if (byId) {
     await updateWorkFromHintIn(k, byId, hint, actor);
-    const owner = await k.selectFrom("works").select("owner_user_id").where("id", "=", byId).executeTakeFirstOrThrow();
     const workId = userId ? await resolveAliasIn(k, userId, byId) : byId;
+    const owner = await k.selectFrom("works").select("owner_user_id").where("id", "=", workId).executeTakeFirstOrThrow();
     return { workId, created: false, draft: owner.owner_user_id !== null };
   }
 
@@ -160,7 +160,31 @@ async function suggest(k: K, userId: string, draftId: string, hint: WorkHint): P
   }
 }
 
-/** Resolve a hint to a work per spec §8.4. Never auto-links on fuzzy matches; those become suggestions. */
-export function matchWork(db: Db, userId: string, hint: WorkHint, actor: string): Promise<Resolved> {
-  return db.k.transaction().execute((tx) => resolve(tx, userId, hint, actor));
+const MAX_ATTEMPTS = 3;
+
+/** SQLite "UNIQUE constraint failed", or Postgres SQLSTATE 23505, anywhere in the cause chain. */
+export function isUniqueViolation(err: unknown): boolean {
+  for (let e: unknown = err, depth = 0; e && typeof e === "object" && depth < 5; e = (e as { cause?: unknown }).cause, depth++) {
+    const { code, errno, message } = e as { code?: unknown; errno?: unknown; message?: unknown };
+    if (code === "23505" || errno === "23505") return true;
+    if (typeof message === "string" && (message.includes("UNIQUE constraint failed") || message.includes("duplicate key"))) {
+      return true;
+    }
+  }
+  return false;
+}
+
+/**
+ * Resolve a hint to a work per spec §8.4. Never auto-links on fuzzy matches; those become suggestions.
+ * A concurrent match may create the same work first (unique external id or parent+position); the losing
+ * transaction is discarded and the whole resolve re-run, which then finds the winner's rows.
+ */
+export async function matchWork(db: Db, userId: string, hint: WorkHint, actor: string): Promise<Resolved> {
+  for (let attempt = 1; ; attempt++) {
+    try {
+      return await db.k.transaction().execute((tx) => resolve(tx, userId, hint, actor));
+    } catch (err) {
+      if (attempt >= MAX_ATTEMPTS || !isUniqueViolation(err)) throw err;
+    }
+  }
 }

packages/core/src/works.ts

@@ -45,6 +45,9 @@ export async function createWorkIn(
 ): Promise<string> {
   const { hint } = input;
   if (hint.title.trim() === "") throw new ValidationError("title", "required");
+  if (input.ownerUserId && Object.keys(hint.externalIds).length > 0) {
+    throw new ValidationError("externalIds", "private works cannot carry external ids");
+  }
   const id = newId();
   const now = nowIso();
   await k
@@ -111,7 +114,15 @@ export async function finalWorkId(k: K, id: string): Promise<string> {
   return (await getWorkIn(k, id))?.id ?? id;
 }
 
-export async function findWorkByExternalIdsIn(k: K, ids: Record<string, string>): Promise<string | null> {
+/**
+ * First work matching any (namespace, value) pair, following merges. Only works visible to `userId`
+ * count (shared, or private to that user); with no user, only shared works. Others are treated as absent.
+ */
+export async function findWorkByExternalIdsIn(
+  k: K,
+  ids: Record<string, string>,
+  userId: string | null = null,
+): Promise<string | null> {
   for (const [namespace, value] of Object.entries(ids)) {
     const row = await k
       .selectFrom("external_ids")
@@ -119,14 +130,16 @@ export async function findWorkByExternalIdsIn(k: K, ids: Record<string, string>)
       .where("namespace", "=", namespace)
       .where("value", "=", value)
       .executeTakeFirst();
-    if (row) return finalWorkId(k, row.work_id);
+    if (!row) continue;
+    const work = await getWorkIn(k, row.work_id);
+    if (work && (work.owner_user_id === null || work.owner_user_id === userId)) return work.id;
   }
   return null;
 }
 
-/** First work matching any (namespace, value) pair, following merges. */
-export function findWorkByExternalIds(db: Db, ids: Record<string, string>): Promise<string | null> {
-  return findWorkByExternalIdsIn(db.k, ids);
+/** First work visible to `userId` (shared works only when omitted) matching any (namespace, value) pair. */
+export function findWorkByExternalIds(db: Db, ids: Record<string, string>, userId: string | null = null): Promise<string | null> {
+  return findWorkByExternalIdsIn(db.k, ids, userId);
 }
 
 async function addMissingExternalIds(k: K, workId: string, ids: Record<string, string>): Promise<string[]> {
@@ -139,8 +152,13 @@ async function addMissingExternalIds(k: K, workId: string, ids: Record<string, s
       .where("value", "=", value)
       .executeTakeFirst();
     if (exists) continue;
-    await k.insertInto("external_ids").values({ id: newId(), work_id: workId, namespace, value }).execute();
-    added.push(`${namespace}:${value}`);
+    // A concurrent sync may have added the same pair since the check: skip it rather than fail.
+    const res = await k
+      .insertInto("external_ids")
+      .values({ id: newId(), work_id: workId, namespace, value })
+      .onConflict((oc) => oc.columns(["namespace", "value"]).doNothing())
+      .executeTakeFirst();
+    if (Number(res.numInsertedOrUpdatedRows ?? 0n) > 0) added.push(`${namespace}:${value}`);
   }
   return added;
 }
@@ -231,7 +249,8 @@ export function resolveAlias(db: Db, userId: string, workId: string): Promise<st
 
 /**
  * Admin-only global merge of shared work `fromId` into `toId`. Leaves a `merged_into_id` tombstone and
- * moves external IDs. Users without an entry on `toId` get their entry repointed; users who already
+ * moves external IDs and the children whose position `toId` does not already have. Both works must
+ * be shared. Users without an entry on `toId` get their entry repointed; users who already
  * have one keep their `from` entry and get a pending `merged` change event.
  */
 export async function adminMergeWorks(db: Db, adminUserId: string, fromId: string, toId: string): Promise<void> {
@@ -245,12 +264,29 @@ export async function adminMergeWorks(db: Db, adminUserId: string, fromId: strin
     if (!to) throw new NotFoundError("work");
     if (from.merged_into_id) throw new ValidationError("fromId", "already merged");
     if (to.merged_into_id) throw new ValidationError("toId", "already merged");
+    if (from.owner_user_id !== null) throw new ValidationError("fromId", "only shared works can be merged");
+    if (to.owner_user_id !== null) throw new ValidationError("toId", "only shared works can be merged");
     const now = nowIso();
 
     await tx.updateTable("works").set({ merged_into_id: toId, updated_at: now }).where("id", "=", fromId).execute();
     const moved = await tx.selectFrom("external_ids").select(["namespace", "value"]).where("work_id", "=", fromId).execute();
     await tx.updateTable("external_ids").set({ work_id: toId }).where("work_id", "=", fromId).execute();
 
+    // Repoint children so matching through `toId` finds them. A child whose position `toId` already
+    // has stays under `fromId` (unique parent_id+position); it remains reachable via the tombstone.
+    const toPositions = new Set(
+      (await tx.selectFrom("works").select("position").where("parent_id", "=", toId).where("position", "is not", null).execute()).map(
+        (r) => r.position,
+      ),
+    );
+    const children = await tx.selectFrom("works").select(["id", "position"]).where("parent_id", "=", fromId).execute();
+    const movedChildren: string[] = [];
+    for (const c of children) {
+      if (c.position !== null && toPositions.has(c.position)) continue;
+      await tx.updateTable("works").set({ parent_id: toId, updated_at: now }).where("id", "=", c.id).execute();
+      movedChildren.push(c.id);
+    }
+
     const entries = await tx.selectFrom("entries").select(["id", "user_id"]).where("work_id", "=", fromId).execute();
     for (const e of entries) {
       const existing = await tx
@@ -288,7 +324,7 @@ export async function adminMergeWorks(db: Db, adminUserId: string, fromId: strin
       .execute();
     await tx
       .insertInto("work_history")
-      .values({ id: newId(), work_id: toId, before: "{}", after: toJson({ mergedFrom: fromId, externalIds: moved }), actor: adminUserId, created_at: now })
+      .values({ id: newId(), work_id: toId, before: "{}", after: toJson({ mergedFrom: fromId, externalIds: moved, children: movedChildren }), actor: adminUserId, created_at: now })
       .execute();
   });
 }

packages/core/test/matching.test.ts

@@ -99,4 +99,50 @@ forEachDialect("matching", (getDb) => {
     expect(sug[0]).toMatchObject({ user_id: u, work_id: r.workId, candidate_work_id: shared, status: "pending" });
     expect(sug[0]!.score).toBeGreaterThan(0.6);
   });
+
+  test("concurrent matches of the same new external id create one work", async () => {
+    const db = getDb();
+    const u = await addUser(db);
+    const hint = () => ({ title: "Heat", mediaType: "movie" as const, externalIds: { tmdb_movie: "949" } });
+    const rs = await Promise.all([matchWork(db, u, hint(), "p"), matchWork(db, u, hint(), "p"), matchWork(db, u, hint(), "p")]);
+    expect(new Set(rs.map((r) => r.workId)).size).toBe(1);
+    expect(await db.k.selectFrom("works").select("id").execute()).toHaveLength(1);
+    expect(await db.k.selectFrom("external_ids").select("id").execute()).toHaveLength(1);
+  });
+
+  test("concurrent matches of the same new episode chain create it once", async () => {
+    const db = getDb();
+    const u = await addUser(db);
+    const rs = await Promise.all([matchWork(db, u, chain(), "p"), matchWork(db, u, chain(), "p"), matchWork(db, u, chain(), "p")]);
+    expect(new Set(rs.map((r) => r.workId)).size).toBe(1);
+    expect(await db.k.selectFrom("works").select("id").execute()).toHaveLength(3);
+  });
+
+  test("concurrent matches of new episodes under an existing show create one season", async () => {
+    const db = getDb();
+    const u = await addUser(db);
+    await createWork(db, { hint: { title: "Lost", mediaType: "show", externalIds: { tmdb_show: "4607" } } }, "t");
+    const rs = await Promise.all([matchWork(db, u, chain(), "p"), matchWork(db, u, chain(), "p")]);
+    expect(rs[0]!.workId).toBe(rs[1]!.workId);
+    expect(await db.k.selectFrom("works").select("id").execute()).toHaveLength(3);
+  });
+
+  test("concurrent updates adding the same external id do not fail", async () => {
+    const db = getDb();
+    const u = await addUser(db);
+    const id = await createWork(db, { hint: { title: "Dune", mediaType: "book", externalIds: { isbn13: "1" } } }, "t");
+    const hint = () => ({ title: "Dune", mediaType: "book" as const, externalIds: { isbn13: "1", asin: "B1" } });
+    const rs = await Promise.all([matchWork(db, u, hint(), "p"), matchWork(db, u, hint(), "p"), matchWork(db, u, hint(), "p")]);
+    expect(rs.map((r) => r.workId)).toEqual([id, id, id]);
+    expect(await db.k.selectFrom("external_ids").select("id").where("namespace", "=", "asin").execute()).toHaveLength(1);
+  });
+
+  test("draft reflects the aliased work, not the matched one", async () => {
+    const db = getDb();
+    const u = await addUser(db);
+    const shared = await createWork(db, { hint: { title: "Dune", mediaType: "book", externalIds: { isbn13: "1" } } }, "t");
+    const priv = await createWork(db, { hint: { title: "My Dune", mediaType: "book", externalIds: {} }, ownerUserId: u }, "t");
+    await db.k.insertInto("work_aliases").values({ id: newId(), user_id: u, from_work_id: shared, to_work_id: priv }).execute();
+    expect(await matchWork(db, u, { title: "Dune", mediaType: "book", externalIds: { isbn13: "1" } }, "p")).toEqual({ workId: priv, created: false, draft: true });
+  });
 });

packages/core/test/works.test.ts

@@ -3,7 +3,8 @@ import type { Db } from "@sttracker/db/db";
 import { newId, nowIso } from "@sttracker/db/ids";
 import { fromJson } from "@sttracker/db/json";
 import { forEachDialect } from "@sttracker/db/testing";
-import { AuthError } from "../src/errors.ts";
+import { AuthError, ValidationError } from "../src/errors.ts";
+import { matchWork } from "../src/matching.ts";
 import {
   adminMergeWorks,
   createWork,
@@ -109,4 +110,87 @@ forEachDialect("works", (getDb) => {
     expect(fromJson(ev[0]!.payload, {})).toEqual({ fromWorkId: from, toWorkId: to });
     expect(await db.k.selectFrom("work_history").select("id").where("work_id", "=", from).where("actor", "=", admin).execute()).toHaveLength(1);
   });
+
+  test("createWork refuses external ids on a private work", async () => {
+    const db = getDb();
+    const u = await addUser(db);
+    const p = createWork(db, { hint: { title: "Mine", mediaType: "book", externalIds: { isbn13: "1" } }, ownerUserId: u }, "t");
+    await expect(p).rejects.toBeInstanceOf(ValidationError);
+    await expect(p).rejects.toMatchObject({ field: "externalIds", reason: "private works cannot carry external ids" });
+    expect(await db.k.selectFrom("works").select("id").execute()).toHaveLength(0);
+  });
+
+  test("findWorkByExternalIds only returns works visible to the user", async () => {
+    const db = getDb();
+    const a = await addUser(db);
+    const b = await addUser(db);
+    const priv = await createWork(db, { hint: { title: "Mine", mediaType: "book", externalIds: {} }, ownerUserId: a }, "t");
+    // Legacy data: a private work carrying an external id (no longer creatable through createWork).
+    await db.k.insertInto("external_ids").values({ id: newId(), work_id: priv, namespace: "isbn13", value: "42" }).execute();
+    expect(await findWorkByExternalIds(db, { isbn13: "42" }, a)).toBe(priv);
+    expect(await findWorkByExternalIds(db, { isbn13: "42" }, b)).toBeNull();
+    expect(await findWorkByExternalIds(db, { isbn13: "42" })).toBeNull();
+    const shared = await createWork(db, { hint: { title: "Theirs", mediaType: "book", externalIds: { asin: "X" } } }, "t");
+    const hint = { title: "Mine", mediaType: "book" as const, externalIds: { isbn13: "42", asin: "X" } };
+    expect(await matchWork(db, b, hint, "p")).toEqual({ workId: shared, created: false, draft: false });
+    expect(await matchWork(db, a, hint, "p")).toEqual({ workId: priv, created: false, draft: true });
+  });
+
+  test("getWork returns null on a merged_into cycle", async () => {
+    const db = getDb();
+    const a = await createWork(db, { hint: { title: "A", mediaType: "book", externalIds: { x: "1" } } }, "t");
+    const b = await createWork(db, { hint: { title: "B", mediaType: "book", externalIds: { x: "2" } } }, "t");
+    await db.k.updateTable("works").set({ merged_into_id: b }).where("id", "=", a).execute();
+    await db.k.updateTable("works").set({ merged_into_id: a }).where("id", "=", b).execute();
+    expect(await getWork(db, a)).toBeNull();
+  });
+
+  test("search treats a literal _ as a character, not a wildcard", async () => {
+    const db = getDb();
+    const u = await addUser(db);
+    await createWork(db, { hint: { title: "snake_case", mediaType: "book", externalIds: { x: "1" } } }, "t");
+    await createWork(db, { hint: { title: "snakeXcase", mediaType: "book", externalIds: { x: "2" } } }, "t");
+    expect((await searchWorks(db, u, "e_c")).map((w) => w.title)).toEqual(["snake_case"]);
+  });
+
+  test("adminMergeWorks refuses private works", async () => {
+    const db = getDb();
+    const admin = await addUser(db, "admin");
+    const shared = await createWork(db, { hint: { title: "A", mediaType: "book", externalIds: { x: "1" } } }, "t");
+    const priv = await createWork(db, { hint: { title: "B", mediaType: "book", externalIds: {} }, ownerUserId: admin }, "t");
+    await expect(adminMergeWorks(db, admin, priv, shared)).rejects.toBeInstanceOf(ValidationError);
+    await expect(adminMergeWorks(db, admin, shared, priv)).rejects.toBeInstanceOf(ValidationError);
+    expect((await getWork(db, priv))?.id).toBe(priv);
+  });
+
+  test("adminMergeWorks moves children that do not collide, so matching does not recreate them", async () => {
+    const db = getDb();
+    const admin = await addUser(db, "admin");
+    const u = await addUser(db);
+    const show = (id: string) => ({ title: "Lost", mediaType: "show" as const, externalIds: { tmdb_show: id } });
+    const ep = (showId: string, season: number) => ({
+      title: "Pilot",
+      mediaType: "episode" as const,
+      externalIds: {},
+      position: { season, episode: 1 },
+      parent: { title: `Season ${season}`, mediaType: "season" as const, externalIds: {}, position: { season }, parent: show(showId) },
+    });
+    const a1 = await matchWork(db, u, ep("A", 1), "p");
+    const a2 = await matchWork(db, u, ep("A", 2), "p");
+    const b2 = await matchWork(db, u, ep("B", 2), "p");
+    const showA = (await findWorkByExternalIds(db, { tmdb_show: "A" }))!;
+    const showB = (await findWorkByExternalIds(db, { tmdb_show: "B" }))!;
+    const seasonA1 = (await getWork(db, a1.workId))!.parent_id!;
+    const seasonA2 = (await getWork(db, a2.workId))!.parent_id!;
+    const before = (await db.k.selectFrom("works").select("id").execute()).length;
+
+    await adminMergeWorks(db, admin, showA, showB);
+
+    expect((await getWork(db, seasonA1))?.parent_id).toBe(showB); // moved: B had no season 1
+    expect((await getWork(db, seasonA2))?.parent_id).toBe(showA); // left: B already had season 2
+    const again = await matchWork(db, u, ep("A", 1), "p");
+    expect(again).toEqual({ workId: a1.workId, created: false, draft: false });
+    expect((await matchWork(db, u, ep("A", 2), "p")).workId).toBe(b2.workId);
+    expect((await db.k.selectFrom("works").select("id").execute()).length).toBe(before);
+  });
 });

packages/db/src/migrations/0002_works_position_unique.ts

@@ -0,0 +1,16 @@
+import type { Kysely } from "kysely";
+import type { Migration } from "kysely/migration";
+
+/**
+ * One child per (parent, position), so concurrent matches cannot duplicate seasons or episodes.
+ * NULLs are distinct on both dialects: top-level works and children without a position are unaffected.
+ */
+export const worksPositionUnique: Migration = {
+  async up(db: Kysely<any>) {
+    await db.schema.createIndex("works_parent_id_position_uq").unique().on("works").columns(["parent_id", "position"]).execute();
+  },
+
+  async down(db: Kysely<any>) {
+    await db.schema.dropIndex("works_parent_id_position_uq").execute();
+  },
+};

packages/db/src/migrations/index.ts

@@ -1,7 +1,9 @@
 import type { Migration } from "kysely/migration";
 import { initial } from "./0001_initial.ts";
+import { worksPositionUnique } from "./0002_works_position_unique.ts";
 
 /** All migrations, keyed by sortable name. Imported statically so bundles need no filesystem. */
 export const migrations: Record<string, Migration> = {
   "0001_initial": initial,
+  "0002_works_position_unique": worksPositionUnique,
 };

packages/db/test/schema.test.ts

@@ -1,6 +1,6 @@
 import { expect, test } from "bun:test";
 import { sql, type Kysely } from "kysely";
-import { Migrator } from "kysely/migration";
+import { Migrator, NO_MIGRATIONS } from "kysely/migration";
 import { newId, nowIso } from "../src/ids.ts";
 import { toJson } from "../src/json.ts";
 import { migrations } from "../src/migrations/index.ts";
@@ -79,10 +79,29 @@ forEachDialect("schema", (getDb) => {
     await expect(k.insertInto("entries").values({ id: newId(), ...entry }).execute()).rejects.toThrow();
   });
 
+  test("works (parent_id, position) is unique; null positions and null parents are not", async () => {
+    const k = getDb().k;
+    const now = nowIso();
+    const work = (id: string, parent: string | null, position: string | null) => ({
+      id, media_type: "show", title: "T", original_title: "", year: null, release_date: null, creators: "[]",
+      description: "", cover_url: "", cover_attachment_id: null, metadata: "{}", parent_id: parent, position,
+      owner_user_id: null, merged_into_id: null, created_at: now, updated_at: now,
+    });
+    const show = newId();
+    await k.insertInto("works").values(work(show, null, null)).execute();
+    await k.insertInto("works").values(work(newId(), null, null)).execute();
+    await k.insertInto("works").values(work(newId(), null, '{"season":1}')).execute();
+    await k.insertInto("works").values(work(newId(), null, '{"season":1}')).execute();
+    await k.insertInto("works").values(work(newId(), show, null)).execute();
+    await k.insertInto("works").values(work(newId(), show, null)).execute();
+    await k.insertInto("works").values(work(newId(), show, '{"season":1}')).execute();
+    await expect(k.insertInto("works").values(work(newId(), show, '{"season":1}')).execute()).rejects.toThrow();
+  });
+
   test("migrating down removes the tables", async () => {
     const k = getDb().k as Kysely<any>;
     const migrator = new Migrator({ db: k, provider: { getMigrations: async () => migrations } });
-    const { error } = await migrator.migrateDown();
+    const { error } = await migrator.migrateTo(NO_MIGRATIONS);
     expect(error).toBeUndefined();
     const probe = sql`select 1 from users`.execute(k);
     await expect(probe).rejects.toThrow();