diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index 2a7b147..b1787f4 100644 --- a/docs/DECISIONS.md +++ b/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`. diff --git a/docs/TODO.md b/docs/TODO.md index d737840..8d583a3 100644 --- a/docs/TODO.md +++ b/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/`. diff --git a/packages/core/src/matching.ts b/packages/core/src/matching.ts index ddc0563..3cf5a52 100644 --- a/packages/core/src/matching.ts +++ b/packages/core/src/matching.ts @@ -67,12 +67,12 @@ interface Resolved { } async function resolve(k: K, userId: string | null, hint: WorkHint, actor: string): Promise { - // 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 { - 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 { + 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; + } + } } diff --git a/packages/core/src/works.ts b/packages/core/src/works.ts index 2e9ccca..b5d68ae 100644 --- a/packages/core/src/works.ts +++ b/packages/core/src/works.ts @@ -45,6 +45,9 @@ export async function createWorkIn( ): Promise { 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 { return (await getWorkIn(k, id))?.id ?? id; } -export async function findWorkByExternalIdsIn(k: K, ids: Record): Promise { +/** + * 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, + userId: string | null = null, +): Promise { 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) .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): Promise { - 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, userId: string | null = null): Promise { + return findWorkByExternalIdsIn(db.k, ids, userId); } async function addMissingExternalIds(k: K, workId: string, ids: Record): Promise { @@ -139,8 +152,13 @@ async function addMissingExternalIds(k: K, workId: string, ids: Record 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 { @@ -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(); }); } diff --git a/packages/core/test/matching.test.ts b/packages/core/test/matching.test.ts index f460c82..b47c09e 100644 --- a/packages/core/test/matching.test.ts +++ b/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 }); + }); }); diff --git a/packages/core/test/works.test.ts b/packages/core/test/works.test.ts index 1a7a4aa..6ae7ed4 100644 --- a/packages/core/test/works.test.ts +++ b/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); + }); }); diff --git a/packages/db/src/migrations/0002_works_position_unique.ts b/packages/db/src/migrations/0002_works_position_unique.ts new file mode 100644 index 0000000..759af4f --- /dev/null +++ b/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) { + await db.schema.createIndex("works_parent_id_position_uq").unique().on("works").columns(["parent_id", "position"]).execute(); + }, + + async down(db: Kysely) { + await db.schema.dropIndex("works_parent_id_position_uq").execute(); + }, +}; diff --git a/packages/db/src/migrations/index.ts b/packages/db/src/migrations/index.ts index c8c4ed7..4dc1ca0 100644 --- a/packages/db/src/migrations/index.ts +++ b/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 = { "0001_initial": initial, + "0002_works_position_unique": worksPositionUnique, }; diff --git a/packages/db/test/schema.test.ts b/packages/db/test/schema.test.ts index 6bef1c2..e19cbd5 100644 --- a/packages/db/test/schema.test.ts +++ b/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; 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();