merge: feat/phase1 into task/9

packages/core/src/entries.ts

@@ -579,6 +579,16 @@ async function moveEntry(
     await moveEngagements(k, userId, src.id, dst.id, actor);
     await writeEntry(k, src, {}, actor, { extra: { deleted_at: nowIso() } });
   }
+  if (result.id !== src.id) {
+    // The moved entry merged into another one: source records that mapped to it keep syncing into the
+    // survivor. Engagements keep their ids when re-parented, so their mappings stay valid.
+    await k
+      .updateTable("source_records")
+      .set({ target_id: result.id })
+      .where("target_entity", "=", "entry")
+      .where("target_id", "=", src.id)
+      .execute();
+  }
   if (opts.undo) return result;
 
   await repointCopiesAndListItems(k, userId, src.workId, target.id, actor);

packages/core/src/lists.ts

@@ -265,6 +265,28 @@ export async function addListItem(
   );
 }
 
+/** Replace a list's items with `next`, in order (one history row). Used by the sync engine. */
+export async function replaceListItems(
+  db: Db,
+  userId: string,
+  listId: string,
+  next: { workId: string; note?: string | null }[],
+  actor: Actor = "user",
+  tx?: Executor,
+): Promise<void> {
+  await inTx(db, tx, (k) =>
+    mutate(k, userId, listId, actor, async () => {
+      const rows: ItemRow[] = [];
+      for (const [i, it] of next.entries()) {
+        const work = await resolveVisibleWork(k, userId, it.workId);
+        rows.push({ id: newId(), list_id: listId, work_id: work.id, position: i, note: it.note == null ? null : checkText("note", it.note) });
+      }
+      await k.deleteFrom("list_items").where("list_id", "=", listId).execute();
+      if (rows.length > 0) await k.insertInto("list_items").values(rows).execute();
+    }),
+  );
+}
+
 function clampIndex(i: number, max: number): number {
   if (typeof i !== "number" || !Number.isInteger(i)) throw new ValidationError("position", "not_integer");
   return Math.max(0, Math.min(i, max));

packages/core/src/sync/apply.ts

@@ -1,11 +1,13 @@
+import { sql } from "kysely";
 import type { Db } from "@sttracker/db/db";
 import { newId, nowIso } from "@sttracker/db/ids";
 import { fromJson, toJson } from "@sttracker/db/json";
 import type { ChangeStatus, ChangeType, Database } from "@sttracker/db/schema";
-import type { EngagementFields, EntryFields, PartialDate, SourceRecord } from "@sttracker/plugin-sdk/types";
+import type { EngagementFields, EntryFields, PartialDate, SourceRecord, WorkHint } from "@sttracker/plugin-sdk/types";
 import { addEngagement, type EngagementPatch, updateEngagement } from "../engagements.ts";
 import { type EntryPatch, fromStoredDate, resolveVisibleWork, upsertEntry } from "../entries.ts";
 import type { Actor, Executor } from "../history.ts";
+import { createList, replaceListItems, updateList } from "../lists.ts";
 import { matchWork } from "../matching.ts";
 import { resolveAliasIn, updateWorkFromHintIn } from "../works.ts";
 import { decodeVersion, encodeVersion, hashRecord, type NormalizedRecord, normalizeRecord, sameValue } from "./snapshot.ts";
@@ -31,14 +33,33 @@ export interface SyncState {
   db: Db;
   userId: string;
   connectionId: string;
+  pluginId: string;
   runId: string;
   actor: Actor;
   counts: RunCounts;
   log: string[];
 }
 
+/**
+ * Run `fn` against scratch counts and log, and fold them into the run's only when `fn` resolves, so
+ * the run summary reflects committed work only.
+ */
+export async function committed(st: SyncState, fn: (local: SyncState) => Promise<void>): Promise<void> {
+  const local: SyncState = { ...st, counts: emptyCounts(), log: [] };
+  await fn(local);
+  for (const key of Object.keys(local.counts) as (keyof RunCounts)[]) st.counts[key] += local.counts[key];
+  st.log.push(...local.log);
+}
+
 type SourceRecordRow = Database["source_records"];
 type Fields = Record<string, unknown>;
+type ListItemValue = { workId: string; note: string | null };
+
+/**
+ * `source_records.current_hash` of a record that has been matched (its work id stored as
+ * `target_entity "work"`) but whose apply has not committed yet. A retry reuses the stored work.
+ */
+const PENDING_HASH = "";
 
 // ---------------------------------------------------------------------------------------------
 // Field mapping between SourceRecord fields and the user layer's patch keys
@@ -71,6 +92,14 @@ export function engagementPatchFrom(f: EngagementFields | Record<string, unknown
   return out;
 }
 
+/** A list record's own fields (items are matched separately). */
+function listPatchFrom(l: Record<string, unknown> | undefined): Fields {
+  const out: Fields = {};
+  if (typeof l?.name === "string") out.name = l.name;
+  if (typeof l?.description === "string") out.description = l.description;
+  return out;
+}
+
 /** An entry row's current values under the sync-managed patch keys. */
 export function entryValues(r: Database["entries"]): Fields {
   return {
@@ -95,10 +124,39 @@ export function engagementValues(r: Database["engagements"]): Fields {
   };
 }
 
+/** A list's current values: name, description and ordered items. */
+export async function listValues(k: Executor, r: Database["lists"]): Promise<Fields> {
+  const items = await k.selectFrom("list_items").select(["work_id", "note"]).where("list_id", "=", r.id).orderBy("position").orderBy("id").execute();
+  return { name: r.name, description: r.description, items: items.map((i): ListItemValue => ({ workId: i.work_id, note: i.note })) };
+}
+
+/**
+ * Lists carry no `field_sources`, so ownership is read from their history: `name`/`description` belong
+ * to whoever last changed them, and `items` to the user once any user edit exists, else to the source.
+ */
+export async function listFieldSources(k: Executor, listId: string, sourceActor: Actor): Promise<Record<string, string>> {
+  const rows = await k
+    .selectFrom("user_history")
+    .select(["after", "actor"])
+    .where("entity", "=", "list")
+    .where("entity_id", "=", listId)
+    .orderBy("created_at")
+    .orderBy("id")
+    .execute();
+  const fs: Record<string, string> = { items: sourceActor };
+  for (const r of rows) {
+    const after = fromJson<Fields>(r.after, {});
+    for (const f of ["name", "description"]) if (f in after) fs[f] = r.actor;
+    if (r.actor === "user") fs.items = "user";
+  }
+  return fs;
+}
+
 function patchOf(n: NormalizedRecord | undefined): Fields {
   if (!n) return {};
   if (n.kind === "entry") return entryPatchFrom(n.entry);
   if (n.kind === "engagement") return engagementPatchFrom(n.engagement);
+  if (n.kind === "list") return listPatchFrom(n.list);
   return {};
 }
 
@@ -109,12 +167,12 @@ function patchOf(n: NormalizedRecord | undefined): Fields {
 async function writeEvent(
   k: Executor,
   st: SyncState,
-  e: { recordId: string; type: ChangeType; status: ChangeStatus; targetEntity: string; targetId: string | null; payload: Fields },
+  e: { id?: string; recordId: string; type: ChangeType; status: ChangeStatus; targetEntity: string; targetId: string | null; payload: Fields },
 ): Promise<void> {
   await k
     .insertInto("change_events")
     .values({
-      id: newId(),
+      id: e.id ?? newId(),
       user_id: st.userId,
       connection_id: st.connectionId,
       run_id: st.runId,
@@ -133,35 +191,44 @@ async function writeEvent(
   if (e.status === "pending" && e.type === "conflict") st.counts.conflicts++;
 }
 
-/**
- * Anti-flip-flop: a pending changed/conflict event for the same (record, field, incoming value) is
- * never filed twice.
- */
-async function hasPendingFieldEvent(k: Executor, recordId: string, field: string, incoming: unknown): Promise<boolean> {
+const incomingOf = (p: Fields) => ("incoming" in p ? p.incoming : p.to);
+
+/** Pending changed/conflict events of a record for one field. */
+async function pendingFieldEvents(k: Executor, recordId: string, field: string) {
   const rows = await k
     .selectFrom("change_events")
-    .select("payload")
+    .select(["id", "payload"])
     .where("record_id", "=", recordId)
     .where("status", "=", "pending")
     .where("type", "in", ["changed", "conflict"])
     .execute();
-  return rows.some((r) => {
-    const p = fromJson<Fields>(r.payload, {});
-    return p.field === field && sameValue("incoming" in p ? p.incoming : p.to, incoming);
-  });
+  return rows.map((r) => ({ id: r.id, payload: fromJson<Fields>(r.payload, {}) })).filter((r) => r.payload.field === field);
+}
+
+/** Resolve stale pending events as `ignored`, recording why in their payload. */
+async function retire(k: Executor, stale: { id: string; payload: Fields }[], extra: Fields): Promise<void> {
+  for (const s of stale) {
+    await k
+      .updateTable("change_events")
+      .set({ status: "ignored", resolved_at: nowIso(), payload: toJson({ ...s.payload, ...extra }) })
+      .where("id", "=", s.id)
+      .execute();
+  }
 }
 
 interface Target {
-  entity: "entry" | "engagement";
+  entity: "entry" | "engagement" | "list";
   id: string;
   current: Fields;
   fieldSources: Record<string, string>;
 }
 
 /**
- * Decide each incoming field against the target's `field_sources` (spec §8.3):
+ * Decide each incoming field against the target's field ownership (spec §8.3):
  * unset or this connection → applied; `"user"` → pending `changed`; another source → pending `conflict`.
  * Values equal to the current one are skipped, so the plugin never takes ownership of a field it did not change.
+ * Anti-flip-flop: a pending event for the same (record, field, incoming value) is never filed twice, and a
+ * newer incoming value supersedes an older pending one (`ignored`, `payload.supersededBy`).
  * Returns the patch that may be applied, with each applied field's previous value.
  */
 async function decideFields(
@@ -175,16 +242,24 @@ async function decideFields(
   const from: Fields = {};
   for (const [field, value] of Object.entries(incoming)) {
     const current = t.current[field];
-    if (sameValue(current, value)) continue;
+    const pending = await pendingFieldEvents(k, recordId, field);
+    if (sameValue(current, value)) {
+      // The source now agrees with what STTracker holds: its older proposals are moot.
+      await retire(k, pending, { reason: "source_agrees" });
+      continue;
+    }
     const source = t.fieldSources[field];
     if (source === undefined || source === st.actor) {
       patch[field] = value;
       from[field] = current ?? null;
       continue;
     }
-    if (await hasPendingFieldEvent(k, recordId, field, value)) continue;
+    if (pending.some((p) => sameValue(incomingOf(p.payload), value))) continue;
+    const id = newId();
+    await retire(k, pending, { supersededBy: id });
     if (source === "user") {
       await writeEvent(k, st, {
+        id,
         recordId,
         type: "changed",
         status: "pending",
@@ -194,6 +269,7 @@ async function decideFields(
       });
     } else {
       await writeEvent(k, st, {
+        id,
         recordId,
         type: "conflict",
         status: "pending",
@@ -206,25 +282,30 @@ async function decideFields(
   return { patch, from };
 }
 
-async function applyPatch(k: Executor, st: SyncState, t: Target, patch: Fields): Promise<void> {
+/** Write an allowed patch to the target as this connection's actor. */
+export async function applyPatch(k: Executor, db: Db, userId: string, t: Pick<Target, "entity" | "id">, patch: Fields, actor: Actor): Promise<void> {
   if (Object.keys(patch).length === 0) return;
   if (t.entity === "entry") {
     const row = await k.selectFrom("entries").select("work_id").where("id", "=", t.id).executeTakeFirstOrThrow();
-    await upsertEntry(st.db, st.userId, row.work_id, patch as Partial<EntryPatch>, st.actor, k);
+    await upsertEntry(db, userId, row.work_id, patch as Partial<EntryPatch>, actor, k);
+  } else if (t.entity === "engagement") {
+    await updateEngagement(db, userId, t.id, patch as Partial<EngagementPatch>, actor, k);
   } else {
-    await updateEngagement(st.db, st.userId, t.id, patch as Partial<EngagementPatch>, st.actor, k);
+    const { items, ...own } = patch;
+    if (Object.keys(own).length > 0) await updateList(db, userId, t.id, own, actor, k);
+    if (items !== undefined) await replaceListItems(db, userId, t.id, items as ListItemValue[], actor, k);
   }
 }
 
 /** The live target of a mapped record, or null (unmapped, gone, or soft-deleted: never revived). */
-async function loadTarget(k: Executor, userId: string, entity: string | null, id: string | null): Promise<Target | null> {
+async function loadTarget(k: Executor, st: SyncState, entity: string | null, id: string | null): Promise<Target | null> {
   if (!id) return null;
   if (entity === "entry") {
     const r = await k
       .selectFrom("entries")
       .selectAll()
       .where("id", "=", id)
-      .where("user_id", "=", userId)
+      .where("user_id", "=", st.userId)
       .where("deleted_at", "is", null)
       .executeTakeFirst();
     return r ? { entity, id, current: entryValues(r), fieldSources: fromJson(r.field_sources, {}) } : null;
@@ -235,12 +316,22 @@ async function loadTarget(k: Executor, userId: string, entity: string | null, id
       .innerJoin("entries as e", "e.id", "g.entry_id")
       .selectAll("g")
       .where("g.id", "=", id)
-      .where("e.user_id", "=", userId)
+      .where("e.user_id", "=", st.userId)
       .where("g.deleted_at", "is", null)
       .where("e.deleted_at", "is", null)
       .executeTakeFirst();
     return r ? { entity, id, current: engagementValues(r), fieldSources: fromJson(r.field_sources, {}) } : null;
   }
+  if (entity === "list") {
+    const r = await k
+      .selectFrom("lists")
+      .selectAll()
+      .where("id", "=", id)
+      .where("user_id", "=", st.userId)
+      .where("deleted_at", "is", null)
+      .executeTakeFirst();
+    return r ? { entity, id, current: await listValues(k, r), fieldSources: await listFieldSources(k, id, st.actor) } : null;
+  }
   return null;
 }
 
@@ -320,7 +411,10 @@ export function sameStartInstant(a: PartialDate, b: PartialDate): boolean {
   return Number.isNaN(x) || Number.isNaN(y) ? a.value === b.value : x === y;
 }
 
-/** A live engagement on the entry that is the same play, not already reported by this connection. */
+/**
+ * A live engagement on the entry that is the same play, not already reported by this connection.
+ * Also returns the date field that matched (its values differ only within the dedupe tolerance).
+ */
 async function findDuplicateEngagement(k: Executor, st: SyncState, entryId: string, incoming: Fields) {
   const fin = incoming.finishedAt as PartialDate | null | undefined;
   const start = incoming.startedAt as PartialDate | null | undefined;
@@ -347,8 +441,8 @@ async function findDuplicateEngagement(k: Executor, st: SyncState, entryId: stri
     const v = engagementValues(r);
     const f = v.finishedAt as PartialDate | null;
     const s = v.startedAt as PartialDate | null;
-    if (fin && f && finishedWithin24h(fin, f)) return r;
-    if (start && s && sameStartInstant(start, s)) return r;
+    if (fin && f && finishedWithin24h(fin, f)) return { row: r, matchedOn: "finishedAt" };
+    if (start && s && sameStartInstant(start, s)) return { row: r, matchedOn: "startedAt" };
   }
   return null;
 }
@@ -356,72 +450,147 @@ async function findDuplicateEngagement(k: Executor, st: SyncState, entryId: stri
 const isEmptyValue = (v: unknown) =>
   v === null || v === undefined || (typeof v === "object" && !Array.isArray(v) && Object.keys(v as object).length === 0);
 
+// ---------------------------------------------------------------------------------------------
+// List items
+// ---------------------------------------------------------------------------------------------
+
+/**
+ * Match list item hints to works, outside any transaction. An ID-less hint first reuses the user's own
+ * live private work with the same media type and title, so re-matching never piles up drafts.
+ */
+async function matchListItems(st: SyncState, items: { work: WorkHint; note?: string }[]): Promise<ListItemValue[]> {
+  const out: ListItemValue[] = [];
+  const seen = new Set<string>();
+  for (const it of items) {
+    let workId: string | undefined;
+    if (Object.keys(it.work.externalIds).length === 0) {
+      workId = (
+        await st.db.k
+          .selectFrom("works")
+          .select("id")
+          .where("owner_user_id", "=", st.userId)
+          .where("merged_into_id", "is", null)
+          .where("media_type", "=", it.work.mediaType)
+          .where(sql<boolean>`lower(title) = ${it.work.title.toLowerCase()}`)
+          .orderBy("id")
+          .executeTakeFirst()
+      )?.id;
+    }
+    workId ??= (await matchWork(st.db, st.userId, it.work, st.actor)).workId;
+    if (seen.has(workId)) continue;
+    seen.add(workId);
+    out.push({ workId, note: it.note ?? null });
+  }
+  return out;
+}
+
 // ---------------------------------------------------------------------------------------------
 // Records
 // ---------------------------------------------------------------------------------------------
 
-/** Snapshot one non-listen record and apply it. Call outside any transaction (matchWork has its own). */
-export async function applyRecord(st: SyncState, rec: SourceRecord): Promise<void> {
-  const hash = hashRecord(rec);
-  const existing = await st.db.k
-    .selectFrom("source_records")
-    .selectAll()
-    .where("connection_id", "=", st.connectionId)
-    .where("key", "=", rec.key)
-    .executeTakeFirst();
-  if (!existing) return addRecord(st, rec, hash);
-  if (existing.current_hash === hash) {
-    await st.db.k.transaction().execute(async (tx) => {
-      await tx.updateTable("source_records").set({ last_seen: nowIso() }).where("id", "=", existing.id).execute();
-      if (existing.deleted_at !== null && rec.deleted !== true) await reappeared(tx, existing);
-    });
-    return;
-  }
-  return changeRecord(st, existing, rec, hash);
-}
-
-async function addRecord(st: SyncState, rec: SourceRecord, hash: string): Promise<void> {
-  const applicable = (rec.kind === "entry" || rec.kind === "engagement") && rec.deleted !== true && !rec.mergedIntoKey;
-  // Matching runs in its own transaction (it retries on unique violations), before the apply transaction.
-  const workId =
-    applicable && (rec.kind === "entry" || rec.kind === "engagement") ? (await matchWork(st.db, st.userId, rec.work, st.actor)).workId : null;
-  await st.db.k.transaction().execute(async (tx) => {
-    const now = nowIso();
-    const recordId = newId();
-    await tx
-      .insertInto("source_records")
-      .values({
-        id: recordId,
-        connection_id: st.connectionId,
-        key: rec.key,
-        kind: rec.kind,
-        current_hash: hash,
-        first_seen: now,
-        last_seen: now,
-        deleted_at: rec.deleted === true ? now : null,
-        merged_into_key: rec.mergedIntoKey ?? null,
-        target_entity: rec.kind === "entry" || rec.kind === "engagement" ? rec.kind : null,
-        target_id: null,
-      })
-      .execute();
-    await tx.insertInto("source_record_versions").values({ id: newId(), record_id: recordId, hash, raw_gz: encodeVersion(rec), seen_at: now }).execute();
-    if (!applicable || workId === null) {
-      if (rec.kind === "list") st.log.push(`record ${rec.key}: list records are stored but not applied`);
+/**
+ * Snapshot one non-listen record and apply it. Call outside any transaction (matchWork has its own).
+ * Counts land in `st` only once the record's work has committed.
+ */
+export function applyRecord(st: SyncState, rec: SourceRecord): Promise<void> {
+  return committed(st, async (local) => {
+    const hash = hashRecord(rec);
+    const existing = await local.db.k
+      .selectFrom("source_records")
+      .selectAll()
+      .where("connection_id", "=", local.connectionId)
+      .where("key", "=", rec.key)
+      .executeTakeFirst();
+    if (!existing || existing.current_hash === PENDING_HASH) return addRecord(local, rec, hash, existing ?? null);
+    if (existing.current_hash === hash) {
+      await local.db.k.transaction().execute(async (tx) => {
+        await tx.updateTable("source_records").set({ last_seen: nowIso() }).where("id", "=", existing.id).execute();
+        if (existing.deleted_at !== null && rec.deleted !== true) await reappeared(tx, existing);
+      });
       return;
     }
-    const targetId =
-      rec.kind === "entry"
-        ? await addEntryRecord(tx, st, recordId, rec.key, workId, entryPatchFrom(rec.entry))
-        : rec.kind === "engagement"
-          ? await addEngagementRecord(tx, st, recordId, rec.key, workId, engagementPatchFrom(rec.engagement))
-          : null;
+    return changeRecord(local, existing, rec, hash);
+  });
+}
+
+async function addRecord(st: SyncState, rec: SourceRecord, hash: string, pendingRow: SourceRecordRow | null): Promise<void> {
+  const applicable = rec.deleted !== true && !rec.mergedIntoKey && rec.kind !== "listen";
+  let recordId = pendingRow?.id ?? null;
+  let workId: string | null = null;
+  let items: ListItemValue[] = [];
+  if (applicable && (rec.kind === "entry" || rec.kind === "engagement")) {
+    if (pendingRow?.target_entity === "work" && pendingRow.target_id) {
+      workId = pendingRow.target_id;
+    } else {
+      // Matching runs in its own transaction (it retries on unique violations). Its result is committed
+      // right away, so a failed apply is retried against the same work instead of a fresh draft.
+      workId = (await matchWork(st.db, st.userId, rec.work, st.actor)).workId;
+      recordId = await persistMatched(st, rec, workId, recordId);
+    }
+  } else if (applicable && rec.kind === "list") {
+    items = await matchListItems(st, rec.list.items);
+  }
+
+  await st.db.k.transaction().execute(async (tx) => {
+    const now = nowIso();
+    const row = {
+      current_hash: hash,
+      last_seen: now,
+      deleted_at: rec.deleted === true ? now : null,
+      merged_into_key: rec.mergedIntoKey ?? null,
+      target_entity: rec.kind === "listen" ? null : rec.kind,
+      target_id: null,
+    };
+    if (recordId) {
+      await tx.updateTable("source_records").set(row).where("id", "=", recordId).execute();
+    } else {
+      recordId = newId();
+      await tx
+        .insertInto("source_records")
+        .values({ id: recordId, connection_id: st.connectionId, key: rec.key, kind: rec.kind, first_seen: now, ...row })
+        .execute();
+    }
+    const id = recordId;
+    await tx.insertInto("source_record_versions").values({ id: newId(), record_id: id, hash, raw_gz: encodeVersion(rec), seen_at: now }).execute();
+    if (!applicable) return;
+    let targetId: string | null = null;
+    if (rec.kind === "entry" && workId) targetId = await addEntryRecord(tx, st, id, rec.key, workId, entryPatchFrom(rec.entry));
+    else if (rec.kind === "engagement" && workId) targetId = await addEngagementRecord(tx, st, id, rec.key, workId, engagementPatchFrom(rec.engagement));
+    else if (rec.kind === "list") targetId = await addListRecord(tx, st, id, rec.key, rec.list.name, items);
     if (targetId) {
-      await tx.updateTable("source_records").set({ target_id: targetId }).where("id", "=", recordId).execute();
+      await tx.updateTable("source_records").set({ target_id: targetId }).where("id", "=", id).execute();
       st.counts.added++;
     }
   });
 }
 
+/** Commit the record row with its matched work (`target_entity "work"`, pending hash) before applying. */
+async function persistMatched(st: SyncState, rec: SourceRecord, workId: string, recordId: string | null): Promise<string> {
+  const now = nowIso();
+  if (recordId) {
+    await st.db.k.updateTable("source_records").set({ target_entity: "work", target_id: workId, last_seen: now }).where("id", "=", recordId).execute();
+    return recordId;
+  }
+  const id = newId();
+  await st.db.k
+    .insertInto("source_records")
+    .values({
+      id,
+      connection_id: st.connectionId,
+      key: rec.key,
+      kind: rec.kind,
+      current_hash: PENDING_HASH,
+      first_seen: now,
+      last_seen: now,
+      deleted_at: null,
+      merged_into_key: null,
+      target_entity: "work",
+      target_id: workId,
+    })
+    .execute();
+  return id;
+}
+
 async function addEntryRecord(k: Executor, st: SyncState, recordId: string, key: string, workId: string, incoming: Fields): Promise<string | null> {
   const live = await liveEntryFor(k, st, workId, incoming);
   if (!live) {
@@ -433,7 +602,7 @@ async function addEntryRecord(k: Executor, st: SyncState, recordId: string, key:
     // The user (or another source) already has this entry: decide field by field.
     const t: Target = { entity: "entry", id: live.row.id, current: entryValues(live.row), fieldSources: fromJson(live.row.field_sources, {}) };
     const { patch } = await decideFields(k, st, recordId, t, incoming);
-    await applyPatch(k, st, t, patch);
+    await applyPatch(k, st.db, st.userId, t, patch, st.actor);
     applied = Object.keys(patch);
   }
   await writeEvent(k, st, {
@@ -455,25 +624,32 @@ async function addEngagementRecord(k: Executor, st: SyncState, recordId: string,
   }
   const dup = await findDuplicateEngagement(k, st, live.row.id, incoming);
   let engagementId: string;
-  let applied: string[];
   let payload: Fields;
   if (dup) {
-    // Same play from another source: attach, and only fill fields nobody has set yet.
-    const current = engagementValues(dup);
-    const sources = fromJson<Record<string, string>>(dup.field_sources, {});
+    // Same play from another source: attach. Fields nobody has set yet are filled; any other differing
+    // field goes through the per-field rules (conflict / pending changed). The date that identified the
+    // play differs only within the dedupe tolerance and is left alone.
+    const t: Target = {
+      entity: "engagement",
+      id: dup.row.id,
+      current: engagementValues(dup.row),
+      fieldSources: fromJson(dup.row.field_sources, {}),
+    };
     const fill: Fields = {};
+    const rest: Fields = {};
     for (const [f, v] of Object.entries(incoming)) {
-      if (sources[f] === undefined && isEmptyValue(current[f]) && !isEmptyValue(v)) fill[f] = v;
+      if (f === dup.matchedOn) continue;
+      if (t.fieldSources[f] === undefined && isEmptyValue(t.current[f]) && !isEmptyValue(v)) fill[f] = v;
+      else rest[f] = v;
     }
-    if (Object.keys(fill).length > 0) await updateEngagement(st.db, st.userId, dup.id, fill as Partial<EngagementPatch>, st.actor, k);
-    engagementId = dup.id;
-    applied = Object.keys(fill);
-    payload = { key, kind: "engagement", fields: applied, dedupedInto: dup.id };
+    const { patch } = await decideFields(k, st, recordId, t, rest);
+    await applyPatch(k, st.db, st.userId, t, { ...fill, ...patch }, st.actor);
+    engagementId = dup.row.id;
+    payload = { key, kind: "engagement", fields: Object.keys({ ...fill, ...patch }), dedupedInto: dup.row.id };
   } else {
     const eg = await addEngagement(st.db, st.userId, live.row.id, incoming as Partial<EngagementPatch>, st.actor, k);
     engagementId = eg.id;
-    applied = Object.keys(incoming);
-    payload = { key, kind: "engagement", fields: applied };
+    payload = { key, kind: "engagement", fields: Object.keys(incoming) };
   }
   await k
     .insertInto("engagement_sources")
@@ -484,6 +660,20 @@ async function addEngagementRecord(k: Executor, st: SyncState, recordId: string,
   return engagementId;
 }
 
+async function addListRecord(k: Executor, st: SyncState, recordId: string, key: string, name: string, items: ListItemValue[]): Promise<string> {
+  const list = await createList(st.db, st.userId, { name, description: `Imported from ${st.pluginId}`, kind: "list" }, st.actor, k);
+  if (items.length > 0) await replaceListItems(st.db, st.userId, list.id, items, st.actor, k);
+  await writeEvent(k, st, {
+    recordId,
+    type: "added",
+    status: "auto_applied",
+    targetEntity: "list",
+    targetId: list.id,
+    payload: { key, kind: "list", fields: ["name", "description", "items"] },
+  });
+  return list.id;
+}
+
 async function changeRecord(st: SyncState, existing: SourceRecordRow, rec: SourceRecord, hash: string): Promise<void> {
   const prevRow = await st.db.k
     .selectFrom("source_record_versions")
@@ -495,18 +685,23 @@ async function changeRecord(st: SyncState, existing: SourceRecordRow, rec: Sourc
     .executeTakeFirst();
   const prev = prevRow ? decodeVersion(prevRow.raw_gz).normalized : undefined;
   const next = normalizeRecord(rec);
+  // List items are matched before the transaction (matchWork has its own), and only when they changed.
+  const itemsChanged = rec.kind === "list" && !rec.deleted && !sameValue(prev?.list?.items, next.list?.items);
+  const items = itemsChanged && rec.kind === "list" ? await matchListItems(st, rec.list.items) : null;
 
   await st.db.k.transaction().execute(async (tx) => {
     const now = nowIso();
+    // A merged record that is later reported deleted stays "merged": its disappearance is expected.
+    const mergedInto = rec.mergedIntoKey ?? (rec.deleted === true ? existing.merged_into_key : null);
     await tx.insertInto("source_record_versions").values({ id: newId(), record_id: existing.id, hash, raw_gz: encodeVersion(rec), seen_at: now }).execute();
     await tx
       .updateTable("source_records")
-      .set({ current_hash: hash, last_seen: now, merged_into_key: rec.mergedIntoKey ?? null })
+      .set({ current_hash: hash, last_seen: now, merged_into_key: mergedInto })
       .where("id", "=", existing.id)
       .execute();
 
     if (rec.deleted === true) {
-      if (existing.deleted_at === null) await markRemoved(tx, st, existing);
+      if (existing.deleted_at === null) await markRemoved(tx, st, { ...existing, merged_into_key: mergedInto });
       return;
     }
     st.counts.changed++;
@@ -514,14 +709,15 @@ async function changeRecord(st: SyncState, existing: SourceRecordRow, rec: Sourc
 
     if (rec.mergedIntoKey && rec.mergedIntoKey !== (prev?.mergedIntoKey ?? null)) await fileMerged(tx, st, existing, rec.mergedIntoKey);
 
-    const target = await loadTarget(tx, st.userId, existing.target_entity, existing.target_id);
+    const target = await loadTarget(tx, st, existing.target_entity, existing.target_id);
     if (!target) return; // unmapped, or the user deleted it: never revived
     const before = patchOf(prev);
     const after = patchOf(next);
     const incoming: Fields = {};
     for (const [f, v] of Object.entries(after)) if (!sameValue(before[f], v)) incoming[f] = v;
+    if (items) incoming.items = items;
     const { patch, from } = await decideFields(tx, st, existing.id, target, incoming);
-    await applyPatch(tx, st, target, patch);
+    await applyPatch(tx, st.db, st.userId, target, patch, st.actor);
     for (const [field, to] of Object.entries(patch)) {
       await writeEvent(tx, st, {
         recordId: existing.id,
@@ -541,6 +737,7 @@ async function changeRecord(st: SyncState, existing: SourceRecordRow, rec: Sourc
 }
 
 async function targetWorkId(k: Executor, t: Target): Promise<string | null> {
+  if (t.entity === "list") return null;
   const entryId =
     t.entity === "entry"
       ? t.id
@@ -561,11 +758,24 @@ async function reappeared(k: Executor, rec: SourceRecordRow): Promise<void> {
     .execute();
 }
 
-/** Set `source_records.deleted_at` and file a pending `removed` event when there is a live target. */
+/**
+ * Set `source_records.deleted_at` and file a pending `removed` event when there is a live target.
+ * A record merged into another (its `merged_into_key` set, or a pending `merged` event) is expected
+ * to vanish: no removal is counted or filed for it, ever.
+ */
 export async function markRemoved(k: Executor, st: SyncState, rec: SourceRecordRow): Promise<void> {
   await k.updateTable("source_records").set({ deleted_at: nowIso() }).where("id", "=", rec.id).execute();
+  if (rec.merged_into_key) return;
+  const pendingMerge = await k
+    .selectFrom("change_events")
+    .select("id")
+    .where("record_id", "=", rec.id)
+    .where("type", "=", "merged")
+    .where("status", "=", "pending")
+    .executeTakeFirst();
+  if (pendingMerge) return;
   st.counts.removed++;
-  const target = await loadTarget(k, st.userId, rec.target_entity, rec.target_id);
+  const target = await loadTarget(k, st, rec.target_entity, rec.target_id);
   if (!target) return;
   const already = await k
     .selectFrom("change_events")
@@ -588,8 +798,8 @@ export async function markRemoved(k: Executor, st: SyncState, rec: SourceRecordR
 /** A pending `merged` event on the entry this record maps to (an engagement record's entry). */
 async function fileMerged(k: Executor, st: SyncState, rec: SourceRecordRow, intoKey: string): Promise<void> {
   st.counts.merged++;
-  const target = await loadTarget(k, st.userId, rec.target_entity, rec.target_id);
-  if (!target) return;
+  const target = await loadTarget(k, st, rec.target_entity, rec.target_id);
+  if (!target || target.entity === "list") return;
   const entryId =
     target.entity === "entry"
       ? target.id

packages/core/src/sync/changes.ts

@@ -1,14 +1,16 @@
 import type { Db } from "@sttracker/db/db";
 import { nowIso } from "@sttracker/db/ids";
-import { fromJson } from "@sttracker/db/json";
+import { fromJson, toJson } from "@sttracker/db/json";
 import type { ChangeStatus, ChangeType, Database } from "@sttracker/db/schema";
 import type { WorkHint } from "@sttracker/plugin-sdk/types";
 import { type EngagementPatch, softDeleteEngagement, updateEngagement } from "../engagements.ts";
 import { type EntryPatch, moveEntryToWork, softDeleteEntry, upsertEntry } from "../entries.ts";
 import { NotFoundError, ValidationError } from "../errors.ts";
 import type { Actor, Executor } from "../history.ts";
+import { softDeleteList } from "../lists.ts";
 import { matchWork } from "../matching.ts";
-import { engagementValues, entryValues } from "./apply.ts";
+import { resolveAliasIn } from "../works.ts";
+import { applyPatch, engagementValues, entryValues } from "./apply.ts";
 import { decodeVersion } from "./snapshot.ts";
 
 export interface ChangeEvent {
@@ -65,10 +67,12 @@ const STATUS_FOR = { accept: "accepted", ignore: "ignored", keep_mine: "kept_min
 /**
  * Resolve a pending event.
  * - `accept`: changed/conflict applies the incoming value as the event's plugin actor (which then owns
- *   the field); removed soft-deletes the target; merged moves the entry to the merge target's work.
+ *   the field); removed soft-deletes the target (an engagement other sources still report only loses
+ *   this connection's `engagement_sources` row); merged moves the entry to the merge target's work.
  * - `keep_mine`: changed/conflict set `field_sources[field] = "user"`, so that source stops overriding it.
  * - `ignore`: data untouched.
- * All three set `status` and `resolved_at`.
+ * All three set `status` and `resolved_at`. Accept/keep_mine on a target that is gone (deleted, or merged
+ * away with no live survivor) resolves the event as `ignored` with `payload.reason = "target_gone"`.
  */
 export async function resolveChange(
   db: Db,
@@ -76,7 +80,7 @@ export async function resolveChange(
   changeId: string,
   action: "accept" | "ignore" | "keep_mine",
 ): Promise<void> {
-  if (!(action in STATUS_FOR)) throw new ValidationError("action", "invalid");
+  if (!Object.hasOwn(STATUS_FOR, action)) throw new ValidationError("action", "invalid");
   const row = await db.k.selectFrom("change_events").selectAll().where("id", "=", changeId).where("user_id", "=", userId).executeTakeFirst();
   if (!row) throw new NotFoundError("change");
   if (row.status !== "pending") throw new ValidationError("status", "not_pending");
@@ -96,19 +100,26 @@ export async function resolveChange(
       .executeTakeFirst();
     if (Number(claimed.numUpdatedRows) === 0) throw new ValidationError("status", "not_pending");
     if (action === "ignore") return;
-    if (ev.type === "changed" || ev.type === "conflict") {
-      const field = String(ev.payload.field);
-      if (action === "accept") {
-        const value = ev.type === "conflict" ? ev.payload.incoming : ev.payload.to;
-        await setField(db, tx, userId, ev, field, value, actor);
-      } else {
-        await setField(db, tx, userId, ev, field, undefined, "user");
-      }
+    const isField = ev.type === "changed" || ev.type === "conflict";
+    if (action === "keep_mine" && !isField) return;
+
+    const targetId = await liveTargetId(tx, userId, ev);
+    if (!targetId) {
+      await tx
+        .updateTable("change_events")
+        .set({ status: "ignored", payload: toJson({ ...ev.payload, reason: "target_gone" }) })
+        .where("id", "=", ev.id)
+        .execute();
       return;
     }
-    if (action !== "accept") return;
-    if (ev.type === "removed") return removeTarget(db, tx, userId, ev, actor);
-    if (ev.type === "merged" && mergeTarget) return moveTo(db, tx, userId, ev, mergeTarget, actor);
+    if (isField) {
+      const field = String(ev.payload.field);
+      const value = action === "keep_mine" ? undefined : ev.type === "conflict" ? ev.payload.incoming : ev.payload.to;
+      await setField(db, tx, userId, ev.targetEntity, targetId, field, value, action === "keep_mine" ? "user" : actor);
+      return;
+    }
+    if (ev.type === "removed") return removeTarget(db, tx, userId, ev, targetId, actor);
+    if (ev.type === "merged" && mergeTarget) return moveTo(db, tx, userId, targetId, mergeTarget, actor);
   });
 }
 
@@ -118,51 +129,29 @@ async function actorFor(db: Db, connectionId: string | null): Promise<Actor> {
   return c ? `plugin:${c.plugin_id}:${c.id}` : "user";
 }
 
-/** Write `value` to the target's field as `actor`; with `value` undefined, re-stamp the current value. */
-async function setField(db: Db, k: Executor, userId: string, ev: ChangeEvent, field: string, value: unknown, actor: Actor): Promise<void> {
-  if (!ev.targetId) throw new NotFoundError(ev.targetEntity);
+/**
+ * The event's target if it is still live. A soft-deleted entry that was merged into another (the user's
+ * work alias leads to a work with a live entry) resolves to that survivor. Otherwise null.
+ */
+async function liveTargetId(k: Executor, userId: string, ev: ChangeEvent): Promise<string | null> {
+  if (!ev.targetId) return null;
   if (ev.targetEntity === "entry") {
-    const r = await k
+    const r = await k.selectFrom("entries").select(["id", "work_id", "deleted_at"]).where("id", "=", ev.targetId).where("user_id", "=", userId).executeTakeFirst();
+    if (!r) return null;
+    if (r.deleted_at === null) return r.id;
+    const to = await resolveAliasIn(k, userId, r.work_id);
+    if (to === r.work_id) return null;
+    const survivor = await k
       .selectFrom("entries")
-      .selectAll()
-      .where("id", "=", ev.targetId)
+      .select("id")
       .where("user_id", "=", userId)
+      .where("work_id", "=", to)
       .where("deleted_at", "is", null)
       .executeTakeFirst();
-    if (!r) throw new NotFoundError("entry");
-    const v = value === undefined ? entryValues(r)[field] : value;
-    await upsertEntry(db, userId, r.work_id, { [field]: v } as Partial<EntryPatch>, actor, k);
-    return;
+    return survivor?.id ?? null;
   }
   if (ev.targetEntity === "engagement") {
     const r = await k
-      .selectFrom("engagements as g")
-      .innerJoin("entries as e", "e.id", "g.entry_id")
-      .selectAll("g")
-      .where("g.id", "=", ev.targetId)
-      .where("e.user_id", "=", userId)
-      .executeTakeFirst();
-    if (!r) throw new NotFoundError("engagement");
-    const v = value === undefined ? engagementValues(r)[field] : value;
-    await updateEngagement(db, userId, r.id, { [field]: v } as Partial<EngagementPatch>, actor, k);
-    return;
-  }
-  throw new ValidationError("targetEntity", "unsupported");
-}
-
-async function removeTarget(db: Db, k: Executor, userId: string, ev: ChangeEvent, actor: Actor): Promise<void> {
-  if (!ev.targetId) return;
-  if (ev.targetEntity === "entry") {
-    const live = await k
-      .selectFrom("entries")
-      .select("id")
-      .where("id", "=", ev.targetId)
-      .where("user_id", "=", userId)
-      .where("deleted_at", "is", null)
-      .executeTakeFirst();
-    if (live) await softDeleteEntry(db, userId, live.id, actor, k);
-  } else if (ev.targetEntity === "engagement") {
-    const live = await k
       .selectFrom("engagements as g")
       .innerJoin("entries as e", "e.id", "g.entry_id")
       .select("g.id")
@@ -171,7 +160,51 @@ async function removeTarget(db: Db, k: Executor, userId: string, ev: ChangeEvent
       .where("g.deleted_at", "is", null)
       .where("e.deleted_at", "is", null)
       .executeTakeFirst();
-    if (live) await softDeleteEngagement(db, userId, live.id, actor, k);
+    return r?.id ?? null;
+  }
+  if (ev.targetEntity === "list") {
+    const r = await k
+      .selectFrom("lists")
+      .select("id")
+      .where("id", "=", ev.targetId)
+      .where("user_id", "=", userId)
+      .where("deleted_at", "is", null)
+      .executeTakeFirst();
+    return r?.id ?? null;
+  }
+  return null;
+}
+
+/** Write `value` to a live target's field as `actor`; with `value` undefined, re-stamp the current value. */
+async function setField(db: Db, k: Executor, userId: string, entity: string, id: string, field: string, value: unknown, actor: Actor): Promise<void> {
+  if (entity === "entry") {
+    const r = await k.selectFrom("entries").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
+    const v = value === undefined ? entryValues(r)[field] : value;
+    await upsertEntry(db, userId, r.work_id, { [field]: v } as Partial<EntryPatch>, actor, k);
+  } else if (entity === "engagement") {
+    const r = await k.selectFrom("engagements").selectAll().where("id", "=", id).executeTakeFirstOrThrow();
+    const v = value === undefined ? engagementValues(r)[field] : value;
+    await updateEngagement(db, userId, id, { [field]: v } as Partial<EngagementPatch>, actor, k);
+  } else if (entity === "list") {
+    // Lists have no field_sources: ownership comes from history, so keep_mine needs no write.
+    if (value === undefined) return;
+    await applyPatch(k, db, userId, { entity: "list", id }, { [field]: value }, actor);
+  } else {
+    throw new ValidationError("targetEntity", "unsupported");
+  }
+}
+
+async function removeTarget(db: Db, k: Executor, userId: string, ev: ChangeEvent, targetId: string, actor: Actor): Promise<void> {
+  if (ev.targetEntity === "entry") return softDeleteEntry(db, userId, targetId, actor, k);
+  if (ev.targetEntity === "list") return softDeleteList(db, userId, targetId, actor, k);
+  if (ev.targetEntity === "engagement") {
+    // The same play may still be reported by other sources: detach this one, delete only when none remain.
+    if (ev.connectionId) {
+      await k.deleteFrom("engagement_sources").where("engagement_id", "=", targetId).where("connection_id", "=", ev.connectionId).execute();
+    }
+    const left = await k.selectFrom("engagement_sources").select("id").where("engagement_id", "=", targetId).executeTakeFirst();
+    if (left) return;
+    await softDeleteEngagement(db, userId, targetId, actor, k);
   }
 }
 
@@ -191,7 +224,9 @@ async function mergeTargetWork(db: Db, userId: string, ev: ChangeEvent, actor: A
     .where("key", "=", intoKey)
     .executeTakeFirst();
   if (!target) throw new ValidationError("intoKey", "unknown_record");
-  if (target.target_id) {
+  // A record matched but not yet applied stores its work directly.
+  if (target.target_entity === "work" && target.target_id) return target.target_id;
+  if (target.target_id && (target.target_entity === "entry" || target.target_entity === "engagement")) {
     const entryId =
       target.target_entity === "entry"
         ? target.target_id
@@ -214,25 +249,9 @@ async function mergeTargetWork(db: Db, userId: string, ev: ChangeEvent, actor: A
   return (await matchWork(db, userId, hint, actor)).workId;
 }
 
-async function moveTo(db: Db, k: Executor, userId: string, ev: ChangeEvent, workId: string, actor: Actor): Promise<void> {
-  if (!ev.targetId) return;
-  const entry = await k
-    .selectFrom("entries")
-    .select(["id", "work_id"])
-    .where("id", "=", ev.targetId)
-    .where("user_id", "=", userId)
-    .where("deleted_at", "is", null)
-    .executeTakeFirst();
-  if (!entry) throw new NotFoundError("entry");
+/** Move the live entry to the work; `moveEntryToWork` re-points source records if it merges. */
+async function moveTo(db: Db, k: Executor, userId: string, entryId: string, workId: string, actor: Actor): Promise<void> {
+  const entry = await k.selectFrom("entries").select("work_id").where("id", "=", entryId).executeTakeFirstOrThrow();
   if (entry.work_id === workId) return;
-  const result = await moveEntryToWork(db, userId, entry.id, workId, actor, k);
-  if (result.id !== entry.id) {
-    // The entry merged into an existing one: records that mapped to it now map to the survivor.
-    await k
-      .updateTable("source_records")
-      .set({ target_id: result.id })
-      .where("target_entity", "=", "entry")
-      .where("target_id", "=", entry.id)
-      .execute();
-  }
+  await moveEntryToWork(db, userId, entryId, workId, actor, k);
 }

packages/core/src/sync/listens.ts

@@ -31,10 +31,14 @@ async function trackWork(db: Db, recording: string, l: ListenFields, actor: Acto
   }
 }
 
+/** Listens per query/transaction. 500 rows × 12 columns stays well under both dialects' parameter limits. */
+const BATCH = 500;
+
 /**
  * Append listens (spec §6.4): insert-or-ignore on (user, source, source_key), where `source_key` is the
  * record key. A work is linked only through `mbids.recording`. Non-listen records are ignored.
- * No change events are written.
+ * No change events are written. Per batch: one lookup of existing keys, each distinct recording resolved
+ * once (outside the transaction; `createWork` has its own), then one multi-row insert.
  */
 export async function ingestListens(
   db: Db,
@@ -43,38 +47,48 @@ export async function ingestListens(
   records: SourceRecord[],
   actor: Actor = `plugin:${source}:ingest`,
 ): Promise<{ inserted: number }> {
+  const listens = records.filter((r): r is Extract<SourceRecord, { kind: "listen" }> => r.kind === "listen");
   let inserted = 0;
-  for (const r of records) {
-    if (r.kind !== "listen") continue;
-    const exists = await db.k
+  for (let i = 0; i < listens.length; i += BATCH) {
+    const batch = new Map<string, ListenFields>();
+    for (const r of listens.slice(i, i + BATCH)) if (!batch.has(r.key)) batch.set(r.key, r.listen);
+    const existing = await db.k
       .selectFrom("listens")
-      .select("id")
+      .select("source_key")
       .where("user_id", "=", userId)
       .where("source", "=", source)
-      .where("source_key", "=", r.key)
-      .executeTakeFirst();
-    if (exists) continue;
-    const l = r.listen;
-    const recording = l.mbids?.recording;
-    const workId = recording ? await trackWork(db, recording, l, actor) : null;
-    const res = await db.k
-      .insertInto("listens")
-      .values({
-        id: newId(),
-        user_id: userId,
-        listened_at: l.listenedAt,
-        artist: l.artist,
-        track: l.track,
-        album: l.album ?? null,
-        duration_ms: l.durationMs ?? null,
-        mbids: toJson(l.mbids ?? {}),
-        work_id: workId,
-        source,
-        source_key: r.key,
-        created_at: nowIso(),
-      })
-      .onConflict((oc) => oc.columns(["user_id", "source", "source_key"]).doNothing())
-      .executeTakeFirst();
+      .where("source_key", "in", [...batch.keys()])
+      .execute();
+    for (const e of existing) batch.delete(e.source_key);
+    if (batch.size === 0) continue;
+
+    const works = new Map<string, string>();
+    for (const l of batch.values()) {
+      const rec = l.mbids?.recording;
+      if (rec && !works.has(rec)) works.set(rec, await trackWork(db, rec, l, actor));
+    }
+    const now = nowIso();
+    const rows = [...batch].map(([key, l]) => ({
+      id: newId(),
+      user_id: userId,
+      listened_at: l.listenedAt,
+      artist: l.artist,
+      track: l.track,
+      album: l.album ?? null,
+      duration_ms: l.durationMs ?? null,
+      mbids: toJson(l.mbids ?? {}),
+      work_id: l.mbids?.recording ? works.get(l.mbids.recording)! : null,
+      source,
+      source_key: key,
+      created_at: now,
+    }));
+    const res = await db.k.transaction().execute((tx) =>
+      tx
+        .insertInto("listens")
+        .values(rows)
+        .onConflict((oc) => oc.columns(["user_id", "source", "source_key"]).doNothing())
+        .executeTakeFirst(),
+    );
     inserted += Number(res.numInsertedOrUpdatedRows ?? 0n);
   }
   return { inserted };

packages/core/src/sync/run.ts

@@ -5,7 +5,7 @@ import type { ConnectionStatus, Database, RunMode } from "@sttracker/db/schema";
 import type { PluginContext, SourcePlugin, SourceRecord } from "@sttracker/plugin-sdk/types";
 import { NotFoundError, ValidationError } from "../errors.ts";
 import type { Actor } from "../history.ts";
-import { applyRecord, emptyCounts, markRemoved, type RunCounts, type SyncState } from "./apply.ts";
+import { applyRecord, committed, emptyCounts, markRemoved, type RunCounts, type SyncState } from "./apply.ts";
 import { ingestListens } from "./listens.ts";
 
 /** A `source_connections` row with JSON parsed. Credentials stay encrypted; the caller's `makeContext` decrypts. */
@@ -92,7 +92,7 @@ export async function runSync(deps: SyncDeps, plugin: SourcePlugin, connectionId
     .execute();
 
   const actor: Actor = `plugin:${conn.pluginId}:${conn.id}`;
-  const st: SyncState = { db, userId: conn.userId, connectionId: conn.id, runId, actor, counts: emptyCounts(), log: [] };
+  const st: SyncState = { db, userId: conn.userId, connectionId: conn.id, pluginId: conn.pluginId, runId, actor, counts: emptyCounts(), log: [] };
   const seen = new Set<string>();
   const errors: string[] = [];
   let pullError: string | null = null;
@@ -140,7 +140,7 @@ export async function runSync(deps: SyncDeps, plugin: SourcePlugin, connectionId
         .execute();
       for (const r of live) {
         if (seen.has(r.key)) continue;
-        await db.k.transaction().execute((tx) => markRemoved(tx, st, r));
+        await committed(st, (local) => db.k.transaction().execute((tx) => markRemoved(tx, local, r)));
       }
     } catch (e) {
       errors.push(`removal: ${message(e)}`);