fix(core): atomic restore with aliases, snapshot export, hardened backups

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

packages/core/src/export.ts

@@ -4,10 +4,11 @@ import { toCsv } from "@sttracker/plugin-sdk/csv";
 import type { Db } from "@sttracker/db/db";
 import { newId } from "@sttracker/db/ids";
 import { fromJson, toJson } from "@sttracker/db/json";
-import type { AttachmentEntity } from "@sttracker/db/schema";
-import { storeAttachment } from "./attachments.ts";
+import type { AttachmentEntity, EntryStatus, Visibility } from "@sttracker/db/schema";
+import { MAX_ATTACHMENT_BYTES, mimeAllowed, normalizeMime, sha256Hex, writeAttachmentFile } from "./attachments.ts";
+import { isUniqueViolation } from "./matching.ts";
 import { ImportError } from "./errors.ts";
-import { createWorkIn, findWorkByExternalIdsIn, type Position } from "./works.ts";
+import { createWorkIn, findWorkByExternalIdsIn, type K, type Position } from "./works.ts";
 
 export const EXPORT_FORMAT = "sttracker-export/v1";
 
@@ -59,9 +60,26 @@ function readJsonl(files: Record<string, Uint8Array>, name: string): Rec[] {
     .map((l) => JSON.parse(l) as Rec);
 }
 
-/** Per-user export bundle (zip). Includes soft-deleted rows so a restore is lossless; never credentials. */
-export async function exportUserBundle(db: Db, dataDir: string, userId: string): Promise<Uint8Array> {
-  const k = db.k;
+const CHUNK = 500;
+const MAX_UNCOMPRESSED = 2 * 1024 * 1024 * 1024;
+
+/** Run `fn` over `ids` in batches (keeps `IN (...)` lists under driver parameter limits). */
+async function inChunks<T>(ids: readonly string[], fn: (chunk: string[]) => Promise<T[]>): Promise<T[]> {
+  const all: T[] = [];
+  for (let i = 0; i < ids.length; i += CHUNK) all.push(...(await fn(ids.slice(i, i + CHUNK))));
+  return all;
+}
+
+/**
+ * Per-user export bundle (zip). Includes soft-deleted rows so a restore is lossless; never credentials.
+ * All reads happen in one transaction (repeatable read on Postgres) so the bundle is a consistent snapshot.
+ */
+export function exportUserBundle(db: Db, dataDir: string, userId: string): Promise<Uint8Array> {
+  const tb = db.k.transaction();
+  return (db.dialect === "postgres" ? tb.setIsolationLevel("repeatable read") : tb).execute((tx) => buildBundle(tx, dataDir, userId));
+}
+
+async function buildBundle(k: K, dataDir: string, userId: string): Promise<Uint8Array> {
   const user = await k.selectFrom("users").select(["username", "display_name"]).where("id", "=", userId).executeTakeFirstOrThrow();
 
   const entries = await k.selectFrom("entries").selectAll().where("user_id", "=", userId).orderBy("id").execute();
@@ -70,16 +88,15 @@ export async function exportUserBundle(db: Db, dataDir: string, userId: string):
   );
   const listens = await k.selectFrom("listens").selectAll().where("user_id", "=", userId).orderBy("id").execute();
   const lists = await k.selectFrom("lists").selectAll().where("user_id", "=", userId).orderBy("id").execute();
-  const items = lists.length
-    ? await k.selectFrom("list_items").selectAll().where("list_id", "in", lists.map((l) => l.id)).orderBy("position").orderBy("id").execute()
-    : [];
+  const items = await inChunks(lists.map((l) => l.id), (c) =>
+    k.selectFrom("list_items").selectAll().where("list_id", "in", c).orderBy("position").orderBy("id").execute(),
+  );
   const copies = await k.selectFrom("copies").selectAll().where("user_id", "=", userId).orderBy("id").execute();
   const defs = await k.selectFrom("custom_field_definitions").selectAll().where("user_id", "=", userId).orderBy("id").execute();
   const history = await k.selectFrom("user_history").selectAll().where("user_id", "=", userId).orderBy("created_at").orderBy("id").execute();
   const atts = await k.selectFrom("attachments").selectAll().where("user_id", "=", userId).orderBy("id").execute();
-  const links = atts.length
-    ? await k.selectFrom("attachment_links").selectAll().where("attachment_id", "in", atts.map((a) => a.id)).execute()
-    : [];
+  const links = await inChunks(atts.map((a) => a.id), (c) => k.selectFrom("attachment_links").selectAll().where("attachment_id", "in", c).execute());
+  const aliases = await k.selectFrom("work_aliases").selectAll().where("user_id", "=", userId).orderBy("id").execute();
 
   // Works: everything referenced, plus ancestors.
   const workIds = new Set<string>();
@@ -87,19 +104,22 @@ export async function exportUserBundle(db: Db, dataDir: string, userId: string):
   for (const i of items) workIds.add(i.work_id);
   for (const c of copies) if (c.work_id) workIds.add(c.work_id);
   for (const l of listens) if (l.work_id) workIds.add(l.work_id);
+  for (const a of aliases) {
+    workIds.add(a.from_work_id);
+    workIds.add(a.to_work_id);
+  }
   const works = new Map<string, Rec>();
   let frontier = [...workIds];
   while (frontier.length) {
-    const rows = await k.selectFrom("works").selectAll().where("id", "in", frontier).execute();
+    const rows = await inChunks(frontier, (c) => k.selectFrom("works").selectAll().where("id", "in", c).execute());
     frontier = [];
     for (const w of rows) {
       works.set(w.id, w as Rec);
-      if (w.parent_id && !works.has(w.parent_id)) frontier.push(w.parent_id);
+      // Ancestors and merge survivors are part of the bundle too.
+      for (const ref of [w.parent_id, w.merged_into_id]) if (ref && !works.has(ref)) frontier.push(ref);
     }
   }
-  const extIds = works.size
-    ? await k.selectFrom("external_ids").selectAll().where("work_id", "in", [...works.keys()]).orderBy("namespace").execute()
-    : [];
+  const extIds = await inChunks([...works.keys()], (c) => k.selectFrom("external_ids").selectAll().where("work_id", "in", c).orderBy("namespace").execute());
   const extByWork = new Map<string, Record<string, string>>();
   for (const x of extIds) {
     const m = extByWork.get(x.work_id) ?? {};
@@ -109,7 +129,7 @@ export async function exportUserBundle(db: Db, dataDir: string, userId: string):
   const workRecs = [...works.values()]
     .sort((a, b) => (a.id < b.id ? -1 : 1))
     .map((w) => ({
-      ...out(w, JSON_COLS.works, ["owner_user_id", "merged_into_id", "cover_attachment_id"]),
+      ...out(w, JSON_COLS.works, ["owner_user_id", "cover_attachment_id"]),
       private: w.owner_user_id != null,
       externalIds: extByWork.get(w.id) ?? {},
     }));
@@ -145,6 +165,7 @@ export async function exportUserBundle(db: Db, dataDir: string, userId: string):
     "lists.jsonl": jsonl(listRecs),
     "copies.jsonl": jsonl(copies.map((c) => out(c, JSON_COLS.copies, ["user_id"]))),
     "custom_fields.jsonl": jsonl(defs.map((d) => out(d, JSON_COLS.defs, ["user_id"]))),
+    "work_aliases.jsonl": jsonl(aliases.map((a) => ({ fromWorkId: a.from_work_id, toWorkId: a.to_work_id }))),
     "history.jsonl": jsonl(history.map((h) => out(h, JSON_COLS.history, ["user_id"]))),
     "attachments.jsonl": jsonl(attRecs),
     ...files,
@@ -192,8 +213,29 @@ export async function exportUserCsv(db: Db, userId: string): Promise<Record<stri
 }
 
 const UUID_RE = /[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/g;
+const STATUSES: readonly EntryStatus[] = ["planned", "in_progress", "paused", "completed", "dropped", "none"];
+const VISIBILITIES: readonly Visibility[] = ["private", "friends", "public"];
 
-/** Restore a bundle into an account with no entries. Works are re-matched by external id. */
+function corrupt(): never {
+  throw new ImportError("corrupt_bundle");
+}
+
+function checkRatingValue(v: unknown): void {
+  if (v != null && !(typeof v === "number" && Number.isInteger(v) && v >= 0 && v <= 100)) corrupt();
+}
+
+function checkRows(rows: Rec[], opts: { status?: boolean; visibility?: boolean; rating?: boolean }): void {
+  for (const r of rows) {
+    if (opts.rating) checkRatingValue(r.rating);
+    if (opts.status && !STATUSES.includes(r.status)) corrupt();
+    if (opts.visibility && !VISIBILITIES.includes(r.visibility)) corrupt();
+  }
+}
+
+/**
+ * Restore a bundle into an account with no entries. Works are re-matched by external id. Everything is
+ * validated before anything is written; rows (attachments included) are inserted in one transaction.
+ */
 export async function importUserBundle(
   db: Db,
   dataDir: string,
@@ -201,9 +243,17 @@ export async function importUserBundle(
   zip: Uint8Array,
 ): Promise<{ works: number; entries: number }> {
   let files: Record<string, Uint8Array>;
+  let total = 0;
   try {
-    files = unzipSync(zip);
-  } catch {
+    files = unzipSync(zip, {
+      filter: (f) => {
+        total += f.originalSize;
+        if (total > MAX_UNCOMPRESSED) throw new ImportError("bundle_too_large");
+        return true;
+      },
+    });
+  } catch (e) {
+    if (e instanceof ImportError) throw e;
     throw new ImportError("unsupported_format");
   }
   const mf = files["manifest.json"];
@@ -218,15 +268,39 @@ export async function importUserBundle(
   const existing = await db.k.selectFrom("entries").select("id").where("user_id", "=", userId).limit(1).executeTakeFirst();
   if (existing) throw new ImportError("account_not_empty");
 
-  const works = readJsonl(files, "works.jsonl");
-  const entries = readJsonl(files, "entries.jsonl");
-  const engagements = readJsonl(files, "engagements.jsonl");
-  const listens = readJsonl(files, "listens.jsonl");
-  const lists = readJsonl(files, "lists.jsonl");
-  const copies = readJsonl(files, "copies.jsonl");
-  const defs = readJsonl(files, "custom_fields.jsonl");
-  const history = readJsonl(files, "history.jsonl");
-  const atts = readJsonl(files, "attachments.jsonl");
+  let works: Rec[], entries: Rec[], engagements: Rec[], listens: Rec[], lists: Rec[], copies: Rec[];
+  let defs: Rec[], history: Rec[], atts: Rec[], aliases: Rec[];
+  try {
+    works = readJsonl(files, "works.jsonl");
+    entries = readJsonl(files, "entries.jsonl");
+    engagements = readJsonl(files, "engagements.jsonl");
+    listens = readJsonl(files, "listens.jsonl");
+    lists = readJsonl(files, "lists.jsonl");
+    copies = readJsonl(files, "copies.jsonl");
+    defs = readJsonl(files, "custom_fields.jsonl");
+    history = readJsonl(files, "history.jsonl");
+    atts = readJsonl(files, "attachments.jsonl");
+    aliases = readJsonl(files, "work_aliases.jsonl");
+  } catch {
+    return corrupt();
+  }
+  checkRows(entries, { rating: true, status: true, visibility: true });
+  checkRows(engagements, { rating: true });
+  checkRows(lists, { rating: true, visibility: true });
+
+  // Attachments: validate everything before writing anything.
+  for (const a of atts) {
+    const bytes = files[`attachments/${a.sha256}`];
+    if (
+      !bytes ||
+      !mimeAllowed(normalizeMime(String(a.mime ?? ""))) ||
+      bytes.byteLength > MAX_ATTACHMENT_BYTES ||
+      sha256Hex(bytes) !== a.sha256
+    ) {
+      throw new ImportError("invalid_attachment");
+    }
+  }
+  for (const a of atts) await writeAttachmentFile(dataDir, files[`attachments/${a.sha256}`]!);
 
   const ids = new Map<string, string>();
   const remap = (old: unknown): string | null => (typeof old === "string" ? (ids.get(old) ?? null) : null);
@@ -249,134 +323,152 @@ export async function importUserBundle(
   };
   const ordered = works.map((w) => ({ w, d: depth(w) })).sort((a, b) => a.d - b.d).map((x) => x.w);
 
-  await db.k.transaction().execute(async (tx) => {
-    for (const w of ordered) {
-      const externalIds = (w.externalIds ?? {}) as Record<string, string>;
-      const hasExt = Object.keys(externalIds).length > 0;
-      const matched = hasExt ? await findWorkByExternalIdsIn(tx, externalIds, userId) : null;
-      if (matched) {
-        ids.set(w.id, matched);
-        continue;
-      }
-      const newWorkId = await createWorkIn(
-        tx,
-        {
-          hint: {
-            title: w.title,
-            mediaType: w.mediaType as MediaType,
-            year: w.year ?? undefined,
-            creators: w.creators ?? [],
-            externalIds,
-            coverUrl: w.coverUrl ?? "",
-            description: w.description ?? "",
-            position: (w.position ?? undefined) as Position | undefined,
+  try {
+    await db.k.transaction().execute(async (tx) => {
+      const created = new Set<string>();
+      for (const w of ordered) {
+        const externalIds = (w.externalIds ?? {}) as Record<string, string>;
+        const hasExt = Object.keys(externalIds).length > 0;
+        const matched = hasExt ? await findWorkByExternalIdsIn(tx, externalIds, userId) : null;
+        if (matched) {
+          ids.set(w.id, matched);
+          continue;
+        }
+        const newWorkId = await createWorkIn(
+          tx,
+          {
+            hint: {
+              title: w.title,
+              mediaType: w.mediaType as MediaType,
+              year: w.year ?? undefined,
+              creators: w.creators ?? [],
+              externalIds,
+              coverUrl: w.coverUrl ?? "",
+              description: w.description ?? "",
+              position: (w.position ?? undefined) as Position | undefined,
+            },
+            ownerUserId: hasExt ? null : userId,
+            parentId: remap(w.parentId),
           },
-          ownerUserId: hasExt ? null : userId,
-          parentId: remap(w.parentId),
-        },
-        "import",
-      );
-      await tx
-        .updateTable("works")
-        .set({
-          original_title: w.originalTitle ?? "",
-          release_date: w.releaseDate ?? null,
-          metadata: toJson(w.metadata ?? {}),
-          created_at: w.createdAt,
-          updated_at: w.updatedAt,
-        })
-        .where("id", "=", newWorkId)
-        .execute();
-      ids.set(w.id, newWorkId);
-    }
-
-    for (const d of defs) {
-      const id = newId();
-      ids.set(d.id, id);
-      await tx.insertInto("custom_field_definitions").values({ ...inn(d, JSON_COLS.defs), id, user_id: userId } as never).execute();
-    }
-
-    for (const e of entries) {
-      const id = newId();
-      ids.set(e.id, id);
-      const workId = remap(e.workId);
-      if (!workId) throw new ImportError("corrupt_bundle");
-      await tx
-        .insertInto("entries")
-        .values({ ...inn(e, JSON_COLS.entries), id, user_id: userId, work_id: workId, favorite: e.favorite ? 1 : 0, custom_fields: toJson(remapCustom(e.customFields)) } as never)
-        .execute();
-    }
-
-    for (const g of engagements) {
-      const id = newId();
-      ids.set(g.id, id);
-      const entryId = remap(g.entryId);
-      if (!entryId) throw new ImportError("corrupt_bundle");
-      await tx
-        .insertInto("engagements")
-        .values({ ...inn(g, JSON_COLS.engagements), id, entry_id: entryId, custom_fields: toJson(remapCustom(g.customFields)) } as never)
-        .execute();
-    }
-
-    for (const l of listens) {
-      await tx
-        .insertInto("listens")
-        .values({ ...inn(l, JSON_COLS.listens), id: newId(), user_id: userId, work_id: remap(l.workId) } as never)
-        .onConflict((oc) => oc.doNothing())
-        .execute();
-    }
-
-    for (const l of lists) {
-      const id = newId();
-      ids.set(l.id, id);
-      const { items, ...rest } = l;
-      await tx.insertInto("lists").values({ ...inn(rest, []), id, user_id: userId } as never).execute();
-      for (const it of (items ?? []) as Rec[]) {
-        const workId = remap(it.workId);
-        if (!workId) continue;
-        await tx.insertInto("list_items").values({ ...inn(it, []), id: newId(), list_id: id, work_id: workId } as never).execute();
+          "import",
+        );
+        await tx
+          .updateTable("works")
+          .set({
+            original_title: w.originalTitle ?? "",
+            release_date: w.releaseDate ?? null,
+            metadata: toJson(w.metadata ?? {}),
+            created_at: w.createdAt,
+            updated_at: w.updatedAt,
+          })
+          .where("id", "=", newWorkId)
+          .execute();
+        ids.set(w.id, newWorkId);
+        created.add(w.id);
       }
-    }
 
-    for (const c of copies) {
-      const id = newId();
-      ids.set(c.id, id);
-      await tx
-        .insertInto("copies")
-        .values({ ...inn(c, JSON_COLS.copies), id, user_id: userId, work_id: remap(c.workId), custom_fields: toJson(remapCustom(c.customFields)) } as never)
-        .execute();
-    }
+      // Tombstones: only for works created here (a matched existing work keeps its own state).
+      for (const w of works) {
+        const target = remap(w.mergedIntoId);
+        if (!created.has(w.id) || !target) continue;
+        await tx.updateTable("works").set({ merged_into_id: target }).where("id", "=", ids.get(w.id)!).execute();
+      }
+      for (const a of aliases) {
+        const from = remap(a.fromWorkId);
+        const to = remap(a.toWorkId);
+        if (!from || !to || from === to) continue;
+        await tx.insertInto("work_aliases").values({ id: newId(), user_id: userId, from_work_id: from, to_work_id: to }).execute();
+      }
 
-    for (const h of history) {
-      const sub = (v: unknown) => toJson(JSON.parse(JSON.stringify(v).replace(UUID_RE, (m) => ids.get(m) ?? m)));
-      await tx
-        .insertInto("user_history")
-        .values({ ...inn(h, []), id: newId(), user_id: userId, entity_id: remap(h.entityId) ?? h.entityId, before: sub(h.before), after: sub(h.after) } as never)
-        .execute();
-    }
-  });
+      for (const d of defs) {
+        const id = newId();
+        ids.set(d.id, id);
+        await tx.insertInto("custom_field_definitions").values({ ...inn(d, JSON_COLS.defs), id, user_id: userId } as never).execute();
+      }
 
-  // Attachments go through storeAttachment (own transaction), so after the main one.
-  for (const a of atts) {
-    const bytes = files[`attachments/${a.sha256}`];
-    if (!bytes) continue;
-    const links = ((a.links ?? []) as Rec[])
-      .map((l) => ({ entity: l.entity as AttachmentEntity, entityId: remap(l.entityId) }))
-      .filter((l): l is { entity: AttachmentEntity; entityId: string } => l.entityId != null);
-    const [first, ...rest] = links;
-    if (!first) continue;
-    const row = await storeAttachment(
-      db,
-      dataDir,
-      userId,
-      { bytes, mime: a.mime, filename: a.filename, caption: a.caption ?? undefined, takenAt: a.takenAt ?? undefined },
-      first,
-    );
-    await db.k.updateTable("attachments").set({ visibility: a.visibility ?? null, created_at: a.createdAt }).where("id", "=", row.id).execute();
-    for (const l of rest) {
-      await db.k.insertInto("attachment_links").values({ id: newId(), attachment_id: row.id, entity: l.entity, entity_id: l.entityId }).execute();
-    }
+      for (const e of entries) {
+        const id = newId();
+        ids.set(e.id, id);
+        const workId = remap(e.workId);
+        if (!workId) corrupt();
+        await tx
+          .insertInto("entries")
+          .values({ ...inn(e, JSON_COLS.entries), id, user_id: userId, work_id: workId, favorite: e.favorite ? 1 : 0, custom_fields: toJson(remapCustom(e.customFields)) } as never)
+          .execute();
+      }
+
+      for (const g of engagements) {
+        const id = newId();
+        ids.set(g.id, id);
+        const entryId = remap(g.entryId);
+        if (!entryId) corrupt();
+        await tx
+          .insertInto("engagements")
+          .values({ ...inn(g, JSON_COLS.engagements), id, entry_id: entryId, custom_fields: toJson(remapCustom(g.customFields)) } as never)
+          .execute();
+      }
+
+      for (const l of listens) {
+        await tx
+          .insertInto("listens")
+          .values({ ...inn(l, JSON_COLS.listens), id: newId(), user_id: userId, work_id: remap(l.workId) } as never)
+          .onConflict((oc) => oc.doNothing())
+          .execute();
+      }
+
+      for (const l of lists) {
+        const id = newId();
+        ids.set(l.id, id);
+        const { items, ...rest } = l;
+        await tx.insertInto("lists").values({ ...inn(rest, []), id, user_id: userId } as never).execute();
+        for (const it of (items ?? []) as Rec[]) {
+          const workId = remap(it.workId);
+          if (!workId) continue;
+          await tx.insertInto("list_items").values({ ...inn(it, []), id: newId(), list_id: id, work_id: workId } as never).execute();
+        }
+      }
+
+      for (const c of copies) {
+        const id = newId();
+        ids.set(c.id, id);
+        await tx
+          .insertInto("copies")
+          .values({ ...inn(c, JSON_COLS.copies), id, user_id: userId, work_id: remap(c.workId), custom_fields: toJson(remapCustom(c.customFields)) } as never)
+          .execute();
+      }
+
+      for (const a of atts) {
+        const links = ((a.links ?? []) as Rec[])
+          .map((l) => ({ entity: l.entity as AttachmentEntity, entityId: remap(l.entityId) }))
+          .filter((l): l is { entity: AttachmentEntity; entityId: string } => l.entityId != null);
+        if (links.length === 0) continue;
+        const id = newId();
+        const { links: _links, ...attRow } = a;
+        await tx
+          .insertInto("attachments")
+          .values({ ...inn(attRow, []), id, user_id: userId, mime: normalizeMime(a.mime), size: files[`attachments/${a.sha256}`]!.byteLength } as never)
+          .execute();
+        for (const l of links) {
+          await tx.insertInto("attachment_links").values({ id: newId(), attachment_id: id, entity: l.entity, entity_id: l.entityId }).execute();
+        }
+      }
+
+      // History last, so every referenced id has been remapped.
+      for (const h of history) {
+        const sub = (v: unknown) => toJson(JSON.parse(JSON.stringify(v).replace(UUID_RE, (m) => ids.get(m) ?? m)));
+        await tx
+          .insertInto("user_history")
+          .values({ ...inn(h, []), id: newId(), user_id: userId, entity_id: remap(h.entityId) ?? h.entityId, before: sub(h.before), after: sub(h.after) } as never)
+          .execute();
+      }
+    });
+  } catch (e) {
+    if (e instanceof ImportError) throw e;
+    if (isUniqueViolation(e)) throw new ImportError("conflict");
+    throw e;
   }
+  // An unlink of identical bytes between the pre-write and the commit may have removed a file.
+  for (const a of atts) await writeAttachmentFile(dataDir, files[`attachments/${a.sha256}`]!);
 
   return { works: works.length, entries: entries.length };
 }