diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 3ff0f22..60d2e7c 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -16,20 +16,34 @@ user layer or files a Changes-inbox event. |---|---|---| | `apps/server` | `Bun.serve`: `/api/v1`, `/ingest/`, job worker, static web | planned | | `apps/web` | React app (HTML imports) | planned | -| `packages/core` | domain services, one module per area (import by path, no barrel): `auth`, `crypto`, `works`, `matching`, `entries` (entries, library listing, quick add, per-user moves, match suggestions), `engagements`, `history` (`user_history`, restore), `lists`, `copies`, `custom-fields`, `attachments`, `export` (per-user bundle, CSV, restore), `backup` (scheduled bundles + SQLite `VACUUM INTO`). Change engine planned | exists (partial) | +| `packages/core` | domain services, one module per area (import by path, no barrel): `auth`, `crypto`, `works`, `matching`, `entries` (entries, library listing, quick add, per-user moves, match suggestions), `engagements`, `history` (`user_history`, restore), `lists`, `copies`, `custom-fields`, `attachments`, `export` (per-user bundle, CSV, restore), `backup` (scheduled bundles + SQLite `VACUUM INTO`), `sync/` (the change engine: `snapshot` hashing and versions, `apply` per-record diff and apply, `run` = `runSync`, `changes` = Changes inbox `listChanges`/`resolveChange`, `listens` = `ingestListens`) | exists (partial) | | `packages/db` | Kysely over one Bun.SQL dialect (SQLite + Postgres), ids, JSON helpers, migrations, `forEachDialect` test harness | exists (schema pending) | | `packages/plugin-sdk` | plugin contract, `SourceRecord`, http/rate-limit helpers, fixture test harness | planned | | `plugins/` | one source integration each (hardcover, trakt, mangabaka, listenbrainz, lastfm, csv) | planned | | `docs/design/mockups` | approved UI mockups (reference only) | exists | ## Data flow -Planned sync lifecycle: +Sync lifecycle (`packages/core/src/sync/`; steps 1–2 are Task 9's job worker, planned): 1. The scheduler enqueues a job. -2. The worker loads the plugin and calls `pull` or `pullChanges` with an SDK context. -3. The plugin returns `SourceRecord[]`. -4. The core stores a new `source_record_versions` row for each changed hash and diffs the records into `change_events`. -5. Each event is either auto-applied (new record, or a field never edited by the user) or sent to the inbox. -6. Every user-layer write appends to `user_history`. +2. The worker loads the connection and plugin and calls `runSync(deps, plugin, connectionId, mode)`, injecting + `makeContext` (http client, credentials, cursor). `mode` picks `pull`, `pullChanges` or `importFile`. +3. `runSync` opens a `source_runs` row and iterates the plugin's `SourceRecord`s one at a time. + - `listen` records are batched into `ingestListens` (insert-or-ignore into `listens`; no snapshot, no events). + - Every other record goes through `applyRecord`: hash its normalised form; same hash as + `source_records.current_hash` → only `last_seen` moves. Otherwise a new `source_record_versions` row + (gzipped `{normalized, raw}`) is written, and: + - **new record**: `matchWork` (own transaction, result committed on the record row so retries reuse it), + then in one apply transaction create/patch the entry, engagement (de-duplicating plays across sources) + or list, and write an `added` `auto_applied` event; + - **changed record**: fields that differ from the previous version are decided against `field_sources` + → applied (`changed` `auto_applied`) or filed as `pending` `changed`/`conflict` events. Mapped records + are updated through `source_records.target_*`, never re-matched; + - `deleted: true` → `pending` `removed`; a new `mergedIntoKey` → `pending` `merged`. +4. After a `full` pull that completed and applied every record, records not seen are marked removed + (`source_records.deleted_at` + `pending` `removed`). +5. The run's counts, log and error land in `source_runs`; the connection gets `last_run_at` and `status`. +6. The user resolves `pending` events with `resolveChange` (accept / ignore / keep_mine). +7. Every user-layer write appends to `user_history` with the plugin actor `plugin::`. ## External dependencies - Database: SQLite file under `DATA_DIR` (default) or Postgres via `DATABASE_URL`. If unavailable, nothing works. diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index 6afe5f0..5c55193 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -4,6 +4,75 @@ Append-only. Never edit or delete an entry. Supersede it with a new one that lin One entry per decision a future maintainer would otherwise have to reverse-engineer. No index: `grep -n '^## ' docs/DECISIONS.md` lists every entry (date, title, line number). +## 2026-10-30: Sync engine: hash composition, anti-flip-flop, no revive, root ancestor, dedupe +**Status:** accepted +**Context:** Task 8 built the change engine (`packages/core/src/sync/`). Spec §8.3 fixes the outcomes; these +are the mechanics a maintainer would otherwise reverse-engineer. +**Decision:** +- **Hash composition.** sha256 of the stable-key-ordered JSON of `{kind, work, entry|engagement|listen|list, + deleted, mergedIntoKey}`. `key` and `raw` are excluded; `deleted` is always a boolean and `mergedIntoKey` + always present (null), so an omitted flag and an explicit `false` hash alike. A version row stores + base64(`Bun.gzipSync`) of `{normalized, raw}`: the normalised record is what the next change is diffed + against (field-level diff = previous version vs. incoming), the raw payload is kept for the record. +- **Per-field decision.** Only fields that differ from the previous version are considered, and of those + only values that differ from the current STTracker value. `field_sources[f]` unset or this connection's + actor → applied; `"user"` → `pending changed {field, from: current, to: incoming}`; another source → + `pending conflict {field, current, currentSource, incoming, incomingSource}`. A new record landing on an + entry that already exists goes through the same per-field decision, so a first sync never overwrites + what the user or another source wrote. Unchanged values are never sent, so a plugin never takes + ownership of a field it did not change. +- **Anti-flip-flop and superseding.** A pending `changed`/`conflict` for the same (record, field, incoming + value) is never filed twice. A different newer value for the same (record, field) resolves the older + pending event as `ignored` with `payload.supersededBy = `, so at most one is pending per + field. If the source comes round to the value STTracker already holds, its pending events for that field + are resolved `ignored` with `payload.reason = "source_agrees"`. `keep_mine` re-stamps + `field_sources[f] = "user"`, so later values from any source become a user-decided `changed`, never + another source-vs-source conflict. +- **No revive.** If the mapped entry/engagement is soft-deleted, the record is still snapshotted but nothing + is applied and no event is filed; a new record on a work whose entry the user deleted adds nothing. + Only the user revives. +- **Root ancestor.** Creating an entry on a child work also creates a `status "none"` entry (plugin actor, + empty `field_sources`) on the root ancestor work, unless the user has one, live or deleted. +- **Mapped records are never re-matched.** Only new records go through `matchWork`. Its result is committed + at once as the record's row (`current_hash ""`, `target_entity "work"`), so a failed apply is retried + against the same work and an ID-less hint never leaks a second draft. A changed work hint refreshes the + mapped work via `updateWorkFromHint` (shared works only). `moveEntryToWork` re-points `source_records` + from a merged-away entry to the survivor in the same transaction, so a user's merge never orphans a source. +- **Engagement dedupe.** Among live engagements on the entry not already sourced by this connection: + finish dates match when both are datetimes ≤ 24 h apart as instants, or when both are day precision and + their dates (first 10 chars, computed with `Date.UTC`, no timezone shift) are ≤ 1 day apart; or start + datetimes are the same instant. A match attaches an `engagement_sources` row and fills fields that are + empty and that no source has set. Every other differing field goes through the per-field decision + (so a disagreeing rating is a `conflict`), except the date that identified the play, whose difference is + within the tolerance by definition. +- **Removal.** Absence counts only after a `full` pull whose iteration finished and whose records all + applied without error; any record error fails the run and skips removal. A removed record that comes + back clears `deleted_at` and retires its still-pending `removed` event as `ignored`. A merged-away record + (`merged_into_key` set, or a pending `merged` event) is expected to vanish: no removal, ever. Accepting + `removed` on an engagement other sources still report only deletes this connection's + `engagement_sources` row; the engagement is soft-deleted when no source remains. Run counts include only + records whose transaction committed. +- **Listens skip snapshots.** Listen records are not stored as `source_records` (they are high-volume and + append-only). `listens` insert-or-ignore is the idempotence; they are never removed. Per batch of 500: + one lookup of existing keys, each distinct recording resolved once, one multi-row insert. +- **Lists.** A new `list` record creates a user list (kind `list`, description `Imported from `), + items matched outside the transaction (an ID-less item first reuses the user's own private work with the + same media type and title). Lists have no `field_sources`; ownership is derived from `user_history`: + `name`/`description` belong to whoever last changed them, `items` to the user once any user edit exists. + A user-owned field becomes a pending `changed` (for items, the whole incoming item set); otherwise the + fields are applied and items replaced (`replaceListItems`, one history row). +- **Merge accept.** The target work is the merge target record's mapped entry's work, else `matchWork` on + its latest hint. Admin-merge events (`toWorkId`, no connection) resolve through the same `resolveChange`. +- **Gone targets.** Accept/keep_mine on an event whose target is deleted (or merged away with no live + survivor, found through the user's work alias) resolves it `ignored` with `payload.reason = "target_gone"`. + +**Alternatives:** +- Diffing against the live entry instead of the previous version would re-file decided conflicts each run. +- Hashing `raw` would turn every cosmetic API change into a new version. +- Keeping one run "ok" despite failing records risks marking those records removed. + +**Consequences:** Concurrent runs of one connection are not locked here (Task 9's worker; `docs/TODO.md`). + ## 2026-10-30: Export format `sttracker-export/v1`: per-user zip, camelCase JSONL, deleted rows included, no secrets **Status:** accepted **Context:** Task 15 (`packages/core/src/{export,backup}.ts`) fixed the on-disk bundle format that backups and restore share. diff --git a/docs/TODO.md b/docs/TODO.md index 6ee3b41..a4b21ac 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -24,6 +24,8 @@ Small, specific, independently actionable. Remove only when actually done. file, and compare steps and token use. ## Known issues +- [ ] Sync: two concurrent runs of the same connection can both insert a new `source_records` row for one key + (one fails on the unique key and is logged as a record error). The job worker should lock per connection. - [ ] Restore: a new child work whose parent+position collides with an existing child of a re-matched parent fails with `ImportError("conflict")` instead of being merged or repositioned. - [ ] Restore: an unlink of identical attachment bytes between the pre-write and the commit could remove the file; diff --git a/packages/core/src/entries.ts b/packages/core/src/entries.ts index 28d5110..94dbcef 100644 --- a/packages/core/src/entries.ts +++ b/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); diff --git a/packages/core/src/lists.ts b/packages/core/src/lists.ts index 53b8e86..b139107 100644 --- a/packages/core/src/lists.ts +++ b/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 { + 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)); diff --git a/packages/core/src/sync/apply.ts b/packages/core/src/sync/apply.ts new file mode 100644 index 0000000..c9c8e8c --- /dev/null +++ b/packages/core/src/sync/apply.ts @@ -0,0 +1,815 @@ +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, 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"; + +export interface RunCounts { + seen: number; + added: number; + changed: number; + removed: number; + merged: number; + conflicts: number; + autoApplied: number; + pending: number; + listens: number; +} + +export function emptyCounts(): RunCounts { + return { seen: 0, added: 0, changed: 0, removed: 0, merged: 0, conflicts: 0, autoApplied: 0, pending: 0, listens: 0 }; +} + +/** Everything one run needs while applying records. */ +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): Promise { + 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; +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 +// --------------------------------------------------------------------------------------------- + +/** SDK `EntryFields` → `EntryPatch` keys (`notes` is the entry's private notes). Absent keys stay absent. */ +export function entryPatchFrom(f: EntryFields | Record | undefined): Fields { + const src = (f ?? {}) as EntryFields; + const out: Fields = {}; + if (src.status !== undefined) out.status = src.status; + if (src.rating !== undefined) out.rating = src.rating; + if (src.review !== undefined) out.review = src.review; + if (src.notes !== undefined) out.privateNotes = src.notes; + if (src.favorite !== undefined) out.favorite = src.favorite; + if (src.tags !== undefined) out.tags = src.tags; + return out; +} + +/** SDK `EngagementFields` → `EngagementPatch` keys. An "unknown"-precision date is `null`. */ +export function engagementPatchFrom(f: EngagementFields | Record | undefined): Fields { + const src = (f ?? {}) as EngagementFields; + const date = (d: PartialDate | null) => (d === null || d.precision === "unknown" ? null : d); + const out: Fields = {}; + if (src.startedAt !== undefined) out.startedAt = date(src.startedAt); + if (src.finishedAt !== undefined) out.finishedAt = date(src.finishedAt); + if (src.progress !== undefined) out.progress = src.progress; + if (src.platform !== undefined) out.platform = src.platform; + if (src.rating !== undefined) out.rating = src.rating; + if (src.notes !== undefined) out.notes = src.notes; + return out; +} + +/** A list record's own fields (items are matched separately). */ +function listPatchFrom(l: Record | 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 { + status: r.status, + rating: r.rating, + review: r.review, + privateNotes: r.private_notes, + favorite: Boolean(r.favorite), + tags: fromJson(r.tags, []), + }; +} + +/** An engagement row's current values under the sync-managed patch keys. */ +export function engagementValues(r: Database["engagements"]): Fields { + return { + startedAt: fromStoredDate(r.started_at, r.started_precision), + finishedAt: fromStoredDate(r.finished_at, r.finished_precision), + progress: fromJson>(r.progress, {}), + platform: r.platform, + rating: r.rating, + notes: r.notes, + }; +} + +/** A list's current values: name, description and ordered items. */ +export async function listValues(k: Executor, r: Database["lists"]): Promise { + 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> { + 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 = { items: sourceActor }; + for (const r of rows) { + const after = fromJson(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 {}; +} + +// --------------------------------------------------------------------------------------------- +// Change events +// --------------------------------------------------------------------------------------------- + +async function writeEvent( + k: Executor, + st: SyncState, + e: { id?: string; recordId: string; type: ChangeType; status: ChangeStatus; targetEntity: string; targetId: string | null; payload: Fields }, +): Promise { + await k + .insertInto("change_events") + .values({ + id: e.id ?? newId(), + user_id: st.userId, + connection_id: st.connectionId, + run_id: st.runId, + record_id: e.recordId, + type: e.type, + target_entity: e.targetEntity, + target_id: e.targetId, + payload: toJson(e.payload), + status: e.status, + created_at: nowIso(), + resolved_at: null, + }) + .execute(); + if (e.status === "auto_applied") st.counts.autoApplied++; + if (e.status === "pending") st.counts.pending++; + if (e.status === "pending" && e.type === "conflict") st.counts.conflicts++; +} + +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(["id", "payload"]) + .where("record_id", "=", recordId) + .where("status", "=", "pending") + .where("type", "in", ["changed", "conflict"]) + .execute(); + return rows.map((r) => ({ id: r.id, payload: fromJson(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 { + 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" | "list"; + id: string; + current: Fields; + fieldSources: Record; +} + +/** + * 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( + k: Executor, + st: SyncState, + recordId: string, + t: Target, + incoming: Fields, +): Promise<{ patch: Fields; from: Fields }> { + const patch: Fields = {}; + const from: Fields = {}; + for (const [field, value] of Object.entries(incoming)) { + const current = t.current[field]; + 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 (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", + targetEntity: t.entity, + targetId: t.id, + payload: { field, from: current ?? null, to: value }, + }); + } else { + await writeEvent(k, st, { + id, + recordId, + type: "conflict", + status: "pending", + targetEntity: t.entity, + targetId: t.id, + payload: { field, current: current ?? null, currentSource: source, incoming: value, incomingSource: st.actor }, + }); + } + } + return { patch, from }; +} + +/** Write an allowed patch to the target as this connection's actor. */ +export async function applyPatch(k: Executor, db: Db, userId: string, t: Pick, patch: Fields, actor: Actor): Promise { + 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(db, userId, row.work_id, patch as Partial, actor, k); + } else if (t.entity === "engagement") { + await updateEngagement(db, userId, t.id, patch as Partial, actor, k); + } else { + 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, st: SyncState, entity: string | null, id: string | null): Promise { + if (!id) return null; + if (entity === "entry") { + const r = await k + .selectFrom("entries") + .selectAll() + .where("id", "=", id) + .where("user_id", "=", st.userId) + .where("deleted_at", "is", null) + .executeTakeFirst(); + return r ? { entity, id, current: entryValues(r), fieldSources: fromJson(r.field_sources, {}) } : null; + } + if (entity === "engagement") { + const r = await k + .selectFrom("engagements as g") + .innerJoin("entries as e", "e.id", "g.entry_id") + .selectAll("g") + .where("g.id", "=", id) + .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; +} + +// --------------------------------------------------------------------------------------------- +// Entry creation helpers +// --------------------------------------------------------------------------------------------- + +async function findEntryRow(k: Executor, userId: string, workId: string) { + return k.selectFrom("entries").selectAll().where("user_id", "=", userId).where("work_id", "=", workId).executeTakeFirst(); +} + +/** + * Hierarchy ruling: when the engine creates an entry on a child work, the user also gets an entry on + * the root ancestor (status "none", plugin actor) unless one exists. An existing root entry, live or + * deleted, is never touched. + */ +async function ensureRootEntry(k: Executor, st: SyncState, workId: string): Promise { + let root = workId; + for (let hop = 0; hop < 32; hop++) { + const w = await k.selectFrom("works").select("parent_id").where("id", "=", root).executeTakeFirst(); + if (!w?.parent_id) break; + root = w.parent_id; + } + if (root === workId) return; + root = await resolveAliasIn(k, st.userId, root); + if (await findEntryRow(k, st.userId, root)) return; + await upsertEntry(st.db, st.userId, root, {}, st.actor, k); +} + +/** + * The user's live entry for a work, created (with root ancestor) when missing. Null when the user + * deleted it: syncs never revive. + */ +async function liveEntryFor( + k: Executor, + st: SyncState, + workId: string, + patch: Fields, +): Promise<{ row: Database["entries"]; created: boolean } | null> { + const work = await resolveVisibleWork(k, st.userId, workId); + const row = await findEntryRow(k, st.userId, work.id); + if (row) return row.deleted_at === null ? { row, created: false } : null; + const e = await upsertEntry(st.db, st.userId, work.id, patch as Partial, st.actor, k); + await ensureRootEntry(k, st, work.id); + const created = await k.selectFrom("entries").selectAll().where("id", "=", e.id).executeTakeFirstOrThrow(); + return { row: created, created: true }; +} + +// --------------------------------------------------------------------------------------------- +// Engagement de-duplication (spec §8.3: same play from two sources) +// --------------------------------------------------------------------------------------------- + +const DAY_MS = 86_400_000; +const hasTime = (v: string) => v.length > 10 && v[10] === "T"; +const dayNumber = (v: string) => Date.UTC(Number(v.slice(0, 4)), Number(v.slice(5, 7)) - 1, Number(v.slice(8, 10))) / DAY_MS; + +/** + * Finish dates "within 24 h". Two datetimes compare as instants. Otherwise day-precision values compare + * as calendar dates (a datetime's date part is its first 10 characters) and match when at most one day + * apart. Month/year/unknown precision never matches. + */ +export function finishedWithin24h(a: PartialDate, b: PartialDate): boolean { + if (a.precision !== "day" || b.precision !== "day") return false; + if (hasTime(a.value) && hasTime(b.value)) { + const x = Date.parse(a.value); + const y = Date.parse(b.value); + if (!Number.isNaN(x) && !Number.isNaN(y)) return Math.abs(x - y) <= DAY_MS; + } + return Math.abs(dayNumber(a.value) - dayNumber(b.value)) <= 1; +} + +/** Equal start datetimes (both carry a time). Compared as instants, falling back to string equality. */ +export function sameStartInstant(a: PartialDate, b: PartialDate): boolean { + if (!hasTime(a.value) || !hasTime(b.value)) return false; + const x = Date.parse(a.value); + const y = Date.parse(b.value); + 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. + * 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; + if (!fin && !start) return null; + const rows = await k + .selectFrom("engagements") + .selectAll() + .where("entry_id", "=", entryId) + .where("deleted_at", "is", null) + .where((eb) => + eb.not( + eb.exists( + eb + .selectFrom("engagement_sources as s") + .select("s.id") + .whereRef("s.engagement_id", "=", "engagements.id") + .where("s.connection_id", "=", st.connectionId), + ), + ), + ) + .orderBy("id") + .execute(); + for (const r of rows) { + 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 { row: r, matchedOn: "finishedAt" }; + if (start && s && sameStartInstant(start, s)) return { row: r, matchedOn: "startedAt" }; + } + return null; +} + +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 { + const out: ListItemValue[] = []; + const seen = new Set(); + 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`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). + * Counts land in `st` only once the record's work has committed. + */ +export function applyRecord(st: SyncState, rec: SourceRecord): Promise { + 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; + } + return changeRecord(local, existing, rec, hash); + }); +} + +async function addRecord(st: SyncState, rec: SourceRecord, hash: string, pendingRow: SourceRecordRow | null): Promise { + 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", "=", 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 { + 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 { + const live = await liveEntryFor(k, st, workId, incoming); + if (!live) { + st.log.push(`record ${key}: the user deleted this entry; not revived`); + return null; + } + let applied = Object.keys(incoming); + if (!live.created) { + // 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.db, st.userId, t, patch, st.actor); + applied = Object.keys(patch); + } + await writeEvent(k, st, { + recordId, + type: "added", + status: "auto_applied", + targetEntity: "entry", + targetId: live.row.id, + payload: { key, kind: "entry", fields: applied }, + }); + return live.row.id; +} + +async function addEngagementRecord(k: Executor, st: SyncState, recordId: string, key: string, workId: string, incoming: Fields): Promise { + const live = await liveEntryFor(k, st, workId, {}); + if (!live) { + st.log.push(`record ${key}: the user deleted this entry; engagement not added`); + return null; + } + const dup = await findDuplicateEngagement(k, st, live.row.id, incoming); + let engagementId: string; + let payload: Fields; + if (dup) { + // 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 (f === dup.matchedOn) continue; + if (t.fieldSources[f] === undefined && isEmptyValue(t.current[f]) && !isEmptyValue(v)) fill[f] = v; + else rest[f] = v; + } + 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, st.actor, k); + engagementId = eg.id; + payload = { key, kind: "engagement", fields: Object.keys(incoming) }; + } + await k + .insertInto("engagement_sources") + .values({ id: newId(), engagement_id: engagementId, connection_id: st.connectionId, record_key: key }) + .onConflict((oc) => oc.columns(["connection_id", "record_key"]).doNothing()) + .execute(); + await writeEvent(k, st, { recordId, type: "added", status: "auto_applied", targetEntity: "engagement", targetId: engagementId, payload }); + return engagementId; +} + +async function addListRecord(k: Executor, st: SyncState, recordId: string, key: string, name: string, items: ListItemValue[]): Promise { + 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 { + const prevRow = await st.db.k + .selectFrom("source_record_versions") + .select("raw_gz") + .where("record_id", "=", existing.id) + .where("hash", "=", existing.current_hash) + .orderBy("seen_at", "desc") + .orderBy("id", "desc") + .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: mergedInto }) + .where("id", "=", existing.id) + .execute(); + + if (rec.deleted === true) { + if (existing.deleted_at === null) await markRemoved(tx, st, { ...existing, merged_into_key: mergedInto }); + return; + } + st.counts.changed++; + if (existing.deleted_at !== null) await reappeared(tx, existing); + + if (rec.mergedIntoKey && rec.mergedIntoKey !== (prev?.mergedIntoKey ?? null)) await fileMerged(tx, st, existing, rec.mergedIntoKey); + + 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.db, st.userId, target, patch, st.actor); + for (const [field, to] of Object.entries(patch)) { + await writeEvent(tx, st, { + recordId: existing.id, + type: "changed", + status: "auto_applied", + targetEntity: target.entity, + targetId: target.id, + payload: { field, from: from[field], to }, + }); + } + // The source's description of the work changed: refresh the (shared) work it maps to; never re-match. + if ((rec.kind === "entry" || rec.kind === "engagement") && prev?.work !== undefined && !sameValue(prev.work, next.work)) { + const workId = await targetWorkId(tx, target); + if (workId) await updateWorkFromHintIn(tx, workId, rec.work, st.actor); + } + }); +} + +async function targetWorkId(k: Executor, t: Target): Promise { + if (t.entity === "list") return null; + const entryId = + t.entity === "entry" + ? t.id + : (await k.selectFrom("engagements").select("entry_id").where("id", "=", t.id).executeTakeFirst())?.entry_id; + if (!entryId) return null; + return (await k.selectFrom("entries").select("work_id").where("id", "=", entryId).executeTakeFirst())?.work_id ?? null; +} + +/** The record is back at the source: clear `deleted_at` and retire still-pending removal events for it. */ +async function reappeared(k: Executor, rec: SourceRecordRow): Promise { + await k.updateTable("source_records").set({ deleted_at: null }).where("id", "=", rec.id).execute(); + await k + .updateTable("change_events") + .set({ status: "ignored", resolved_at: nowIso() }) + .where("record_id", "=", rec.id) + .where("type", "=", "removed") + .where("status", "=", "pending") + .execute(); +} + +/** + * 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 { + 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, rec.target_entity, rec.target_id); + if (!target) return; + const already = await k + .selectFrom("change_events") + .select("id") + .where("record_id", "=", rec.id) + .where("type", "=", "removed") + .where("status", "=", "pending") + .executeTakeFirst(); + if (already) return; + await writeEvent(k, st, { + recordId: rec.id, + type: "removed", + status: "pending", + targetEntity: target.entity, + targetId: target.id, + payload: { key: rec.key, kind: rec.kind }, + }); +} + +/** 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 { + st.counts.merged++; + 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 + : (await k.selectFrom("engagements").select("entry_id").where("id", "=", target.id).executeTakeFirstOrThrow()).entry_id; + await writeEvent(k, st, { + recordId: rec.id, + type: "merged", + status: "pending", + targetEntity: "entry", + targetId: entryId, + payload: { key: rec.key, intoKey }, + }); +} diff --git a/packages/core/src/sync/changes.ts b/packages/core/src/sync/changes.ts new file mode 100644 index 0000000..03ecf6b --- /dev/null +++ b/packages/core/src/sync/changes.ts @@ -0,0 +1,257 @@ +import type { Db } from "@sttracker/db/db"; +import { nowIso } from "@sttracker/db/ids"; +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 { resolveAliasIn } from "../works.ts"; +import { applyPatch, engagementValues, entryValues } from "./apply.ts"; +import { decodeVersion } from "./snapshot.ts"; + +export interface ChangeEvent { + id: string; + type: ChangeType; + status: ChangeStatus; + connectionId: string | null; + runId: string | null; + recordId: string | null; + targetEntity: string; + targetId: string | null; + payload: Record; + createdAt: string; + resolvedAt: string | null; +} + +function toChange(r: Database["change_events"]): ChangeEvent { + return { + id: r.id, + type: r.type, + status: r.status, + connectionId: r.connection_id, + runId: r.run_id, + recordId: r.record_id, + targetEntity: r.target_entity, + targetId: r.target_id, + payload: fromJson>(r.payload, {}), + createdAt: r.created_at, + resolvedAt: r.resolved_at, + }; +} + +/** + * The user's change events, newest first. `status` defaults to "pending" (the inbox); "all" includes + * auto-applied and resolved ones. `cursor` is the `nextCursor` of the previous page. + */ +export async function listChanges( + db: Db, + userId: string, + opts: { status?: "pending" | "all"; connectionId?: string; limit?: number; cursor?: string } = {}, +): Promise<{ items: ChangeEvent[]; nextCursor: string | null }> { + const limit = Math.min(Math.max(opts.limit ?? 50, 1), 200); + let q = db.k.selectFrom("change_events").selectAll().where("user_id", "=", userId); + if ((opts.status ?? "pending") === "pending") q = q.where("status", "=", "pending"); + if (opts.connectionId) q = q.where("connection_id", "=", opts.connectionId); + if (opts.cursor) q = q.where("id", "<", opts.cursor); + const rows = await q.orderBy("id", "desc").limit(limit + 1).execute(); + const items = rows.slice(0, limit).map(toChange); + return { items, nextCursor: rows.length > limit ? items[items.length - 1]!.id : null }; +} + +const STATUS_FOR = { accept: "accepted", ignore: "ignored", keep_mine: "kept_mine" } as const; + +/** + * 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 (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`. 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, + userId: string, + changeId: string, + action: "accept" | "ignore" | "keep_mine", +): Promise { + 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"); + const ev = toChange(row); + const actor = await actorFor(db, ev.connectionId); + + // matchWork runs its own transaction; resolve the merge target before opening ours. + const mergeTarget = action === "accept" && ev.type === "merged" ? await mergeTargetWork(db, userId, ev, actor) : null; + + await db.k.transaction().execute(async (tx) => { + // Claim the event first so a concurrent resolve of the same event becomes a no-op failure. + const claimed = await tx + .updateTable("change_events") + .set({ status: STATUS_FOR[action], resolved_at: nowIso() }) + .where("id", "=", ev.id) + .where("status", "=", "pending") + .executeTakeFirst(); + if (Number(claimed.numUpdatedRows) === 0) throw new ValidationError("status", "not_pending"); + if (action === "ignore") return; + 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 (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); + }); +} + +async function actorFor(db: Db, connectionId: string | null): Promise { + if (!connectionId) return "user"; + const c = await db.k.selectFrom("source_connections").select(["id", "plugin_id"]).where("id", "=", connectionId).executeTakeFirst(); + return c ? `plugin:${c.plugin_id}:${c.id}` : "user"; +} + +/** + * 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 { + if (!ev.targetId) return null; + if (ev.targetEntity === "entry") { + 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") + .select("id") + .where("user_id", "=", userId) + .where("work_id", "=", to) + .where("deleted_at", "is", null) + .executeTakeFirst(); + 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") + .select("g.id") + .where("g.id", "=", ev.targetId) + .where("e.user_id", "=", userId) + .where("g.deleted_at", "is", null) + .where("e.deleted_at", "is", null) + .executeTakeFirst(); + 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 { + 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, 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, 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 { + 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); + } +} + +/** + * The work a merged event moves its entry to. An admin merge carries `toWorkId`. A source merge names + * the target record (`intoKey`): its mapped entry's work when it has one, else the work matched for + * the target record's latest hint. + */ +async function mergeTargetWork(db: Db, userId: string, ev: ChangeEvent, actor: Actor): Promise { + if (typeof ev.payload.toWorkId === "string") return ev.payload.toWorkId; + const intoKey = ev.payload.intoKey; + if (!ev.connectionId || typeof intoKey !== "string") throw new ValidationError("payload", "no_merge_target"); + const target = await db.k + .selectFrom("source_records") + .selectAll() + .where("connection_id", "=", ev.connectionId) + .where("key", "=", intoKey) + .executeTakeFirst(); + if (!target) throw new ValidationError("intoKey", "unknown_record"); + // 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 + : (await db.k.selectFrom("engagements").select("entry_id").where("id", "=", target.target_id).executeTakeFirst())?.entry_id; + const entry = entryId + ? await db.k.selectFrom("entries").select("work_id").where("id", "=", entryId).where("user_id", "=", userId).executeTakeFirst() + : undefined; + if (entry) return entry.work_id; + } + const version = await db.k + .selectFrom("source_record_versions") + .select("raw_gz") + .where("record_id", "=", target.id) + .where("hash", "=", target.current_hash) + .orderBy("seen_at", "desc") + .orderBy("id", "desc") + .executeTakeFirst(); + const hint = version ? (decodeVersion(version.raw_gz).normalized.work as WorkHint | undefined) : undefined; + if (!hint) throw new ValidationError("intoKey", "no_work_hint"); + return (await matchWork(db, userId, hint, actor)).workId; +} + +/** 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 { + const entry = await k.selectFrom("entries").select("work_id").where("id", "=", entryId).executeTakeFirstOrThrow(); + if (entry.work_id === workId) return; + await moveEntryToWork(db, userId, entryId, workId, actor, k); +} diff --git a/packages/core/src/sync/listens.ts b/packages/core/src/sync/listens.ts new file mode 100644 index 0000000..a1ccd71 --- /dev/null +++ b/packages/core/src/sync/listens.ts @@ -0,0 +1,95 @@ +import type { Db } from "@sttracker/db/db"; +import { newId, nowIso } from "@sttracker/db/ids"; +import { toJson } from "@sttracker/db/json"; +import type { ListenFields, SourceRecord } from "@sttracker/plugin-sdk/types"; +import type { Actor } from "../history.ts"; +import { isUniqueViolation } from "../matching.ts"; +import { createWork, findWorkByExternalIds } from "../works.ts"; + +/** The one external-ID namespace the engine names itself (spec §6.4). */ +export const RECORDING_NAMESPACE = "musicbrainz_recording"; + +/** + * The shared track work for a MusicBrainz recording: an existing work carrying the id, else a new + * shared `track` work. Never renames an existing work, and never creates drafts. + */ +async function trackWork(db: Db, recording: string, l: ListenFields, actor: Actor): Promise { + const ids = { [RECORDING_NAMESPACE]: recording }; + for (let attempt = 1; ; attempt++) { + const found = await findWorkByExternalIds(db, ids); + if (found) return found; + try { + return await createWork( + db, + { hint: { title: l.track.trim() || "Untitled track", mediaType: "track", externalIds: ids, creators: l.artist ? [{ name: l.artist }] : [] } }, + actor, + ); + } catch (err) { + // A concurrent ingest created it first: look again. + if (attempt >= 3 || !isUniqueViolation(err)) throw err; + } + } +} + +/** 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. 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, + userId: string, + source: string, + records: SourceRecord[], + actor: Actor = `plugin:${source}:ingest`, +): Promise<{ inserted: number }> { + const listens = records.filter((r): r is Extract => r.kind === "listen"); + let inserted = 0; + for (let i = 0; i < listens.length; i += BATCH) { + const batch = new Map(); + 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("source_key") + .where("user_id", "=", userId) + .where("source", "=", source) + .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(); + 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 }; +} diff --git a/packages/core/src/sync/run.ts b/packages/core/src/sync/run.ts new file mode 100644 index 0000000..150fcac --- /dev/null +++ b/packages/core/src/sync/run.ts @@ -0,0 +1,164 @@ +import type { Db } from "@sttracker/db/db"; +import { newId, nowIso } from "@sttracker/db/ids"; +import { fromJson, toJson } from "@sttracker/db/json"; +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, 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. */ +export interface Connection { + id: string; + userId: string; + pluginId: string; + name: string; + config: Record; + credentialsEnc: string | null; + scheduleMinutes: number | null; + cursor: string | null; + status: ConnectionStatus; + lastRunAt: string | null; + createdAt: string; +} + +export interface SyncDeps { + db: Db; + /** Builds the plugin context (http client, credentials, cursor, logging). Injected by the caller. */ + makeContext: (conn: Connection, runId: string) => PluginContext; +} + +export type SyncMode = "full" | "changes" | { file: { name: string; bytes: Uint8Array } }; + +export interface RunSummary { + runId: string; + status: "ok" | "failed"; + counts: RunCounts; + error?: string; +} + +const LISTEN_BATCH = 500; + +function toConnection(r: Database["source_connections"]): Connection { + return { + id: r.id, + userId: r.user_id, + pluginId: r.plugin_id, + name: r.name, + config: fromJson>(r.config, {}), + credentialsEnc: r.credentials_enc, + scheduleMinutes: r.schedule_minutes, + cursor: r.cursor, + status: r.status, + lastRunAt: r.last_run_at, + createdAt: r.created_at, + }; +} + +function iterate(plugin: SourcePlugin, ctx: PluginContext, mode: SyncMode): AsyncIterable { + const id = plugin.manifest.id; + if (mode === "full") { + if (!plugin.pull) throw new Error(`plugin "${id}" cannot pull`); + return plugin.pull(ctx); + } + if (mode === "changes") { + if (!plugin.pullChanges) throw new Error(`plugin "${id}" cannot pull changes`); + return plugin.pullChanges(ctx); + } + if (!plugin.importFile) throw new Error(`plugin "${id}" cannot import files`); + return plugin.importFile(ctx, mode.file); +} + +const message = (e: unknown) => (e instanceof Error ? e.message : String(e)); + +/** + * One sync run (spec §8.3): pull records, snapshot and diff each against its previous version, apply + * or file change events, and append listens. Absent records are marked removed only after a `full` + * pull whose iteration finished and whose records all applied. Any failure marks the run `failed` and + * the connection `error`; the summary is returned, not thrown. + */ +export async function runSync(deps: SyncDeps, plugin: SourcePlugin, connectionId: string, mode: SyncMode): Promise { + const { db } = deps; + const row = await db.k.selectFrom("source_connections").selectAll().where("id", "=", connectionId).executeTakeFirst(); + if (!row) throw new NotFoundError("connection"); + if (row.plugin_id !== plugin.manifest.id) throw new ValidationError("plugin", "connection_plugin_mismatch"); + const conn = toConnection(row); + const runId = newId(); + const runMode: RunMode = typeof mode === "string" ? mode : "file"; + await db.k + .insertInto("source_runs") + .values({ id: runId, connection_id: conn.id, mode: runMode, started_at: nowIso(), finished_at: null, status: "running", counts: toJson(emptyCounts()), log: "", error: null }) + .execute(); + + const actor: Actor = `plugin:${conn.pluginId}:${conn.id}`; + const st: SyncState = { db, userId: conn.userId, connectionId: conn.id, pluginId: conn.pluginId, runId, actor, counts: emptyCounts(), log: [] }; + const seen = new Set(); + const errors: string[] = []; + let pullError: string | null = null; + let listens: SourceRecord[] = []; + const flushListens = async () => { + if (listens.length === 0) return; + const batch = listens; + listens = []; + try { + st.counts.listens += (await ingestListens(db, conn.userId, conn.pluginId, batch, actor)).inserted; + } catch (e) { + errors.push(`listens: ${message(e)}`); + } + }; + + try { + const ctx = deps.makeContext(conn, runId); + for await (const rec of iterate(plugin, ctx, mode)) { + st.counts.seen++; + if (rec.kind === "listen") { + listens.push(rec); + if (listens.length >= LISTEN_BATCH) await flushListens(); + continue; + } + seen.add(rec.key); + try { + await applyRecord(st, rec); + } catch (e) { + errors.push(`record ${rec.key}: ${message(e)}`); + } + } + } catch (e) { + pullError = message(e); + } + await flushListens(); + + // Removal by absence: only a completed full pull with every record applied can prove a record is gone. + if (mode === "full" && pullError === null && errors.length === 0) { + try { + const live = await db.k + .selectFrom("source_records") + .selectAll() + .where("connection_id", "=", conn.id) + .where("deleted_at", "is", null) + .execute(); + for (const r of live) { + if (seen.has(r.key)) continue; + await committed(st, (local) => db.k.transaction().execute((tx) => markRemoved(tx, local, r))); + } + } catch (e) { + errors.push(`removal: ${message(e)}`); + } + } + + const failed = pullError !== null || errors.length > 0; + const error = pullError ?? (errors.length > 0 ? `${errors.length} error(s); first: ${errors[0]}` : undefined); + const log = [...st.log, ...errors]; + if (pullError !== null) log.push(`pull failed: ${pullError}`); + const now = nowIso(); + await db.k + .updateTable("source_runs") + .set({ finished_at: now, status: failed ? "failed" : "ok", counts: toJson(st.counts), log: log.join("\n"), error: error ?? null }) + .where("id", "=", runId) + .execute(); + const nextStatus: ConnectionStatus = failed ? "error" : conn.status === "error" ? "active" : conn.status; + await db.k.updateTable("source_connections").set({ last_run_at: now, status: nextStatus }).where("id", "=", conn.id).execute(); + + return { runId, status: failed ? "failed" : "ok", counts: st.counts, ...(error !== undefined ? { error } : {}) }; +} diff --git a/packages/core/src/sync/snapshot.ts b/packages/core/src/sync/snapshot.ts new file mode 100644 index 0000000..59b2362 --- /dev/null +++ b/packages/core/src/sync/snapshot.ts @@ -0,0 +1,75 @@ +import type { SourceRecord } from "@sttracker/plugin-sdk/types"; + +/** + * The part of a `SourceRecord` that the change engine compares: everything except `key` and `raw`. + * `deleted` is always a boolean and `mergedIntoKey` always present (null when unset), so an absent + * flag and an explicit `false` hash the same. + */ +export interface NormalizedRecord { + kind: SourceRecord["kind"]; + work?: unknown; + entry?: Record; + engagement?: Record; + listen?: Record; + list?: Record; + deleted: boolean; + mergedIntoKey: string | null; +} + +/** JSON with object keys sorted at every depth; `undefined` members are dropped, as in `JSON.stringify`. */ +export function stableStringify(v: unknown): string { + return JSON.stringify(sortKeys(v)) ?? "null"; +} + +function sortKeys(v: unknown): unknown { + if (Array.isArray(v)) return v.map((x) => (x === undefined ? null : sortKeys(x))); + if (v === null || typeof v !== "object") return v; + const out: Record = {}; + for (const k of Object.keys(v).sort()) { + const x = (v as Record)[k]; + if (x !== undefined) out[k] = sortKeys(x); + } + return out; +} + +/** True when two values serialise identically (`undefined` counts as `null`). */ +export function sameValue(a: unknown, b: unknown): boolean { + return stableStringify(a ?? null) === stableStringify(b ?? null); +} + +export function normalizeRecord(r: SourceRecord): NormalizedRecord { + const base = { kind: r.kind, deleted: r.deleted === true, mergedIntoKey: r.mergedIntoKey ?? null }; + const plain = (o: unknown) => JSON.parse(stableStringify(o)) as Record; + switch (r.kind) { + case "entry": + return { ...base, work: plain(r.work), entry: plain(r.entry) }; + case "engagement": + return { ...base, work: plain(r.work), engagement: plain(r.engagement) }; + case "listen": + return { ...base, listen: plain(r.listen) }; + case "list": + return { ...base, list: plain(r.list) }; + } +} + +/** sha256 (hex) of the stable JSON of the normalised record. The raw payload is not part of it. */ +export function hashRecord(r: SourceRecord): string { + return new Bun.CryptoHasher("sha256").update(stableStringify(normalizeRecord(r))).digest("hex"); +} + +/** What a `source_record_versions.raw_gz` holds: the normalised record (for diffing) and the raw payload. */ +export interface StoredVersion { + normalized: NormalizedRecord; + raw: unknown; +} + +/** gzip (`Bun.gzipSync`) + base64 of `{ normalized, raw }`. */ +export function encodeVersion(r: SourceRecord): string { + const json = JSON.stringify({ normalized: normalizeRecord(r), raw: r.raw ?? null }); + return Buffer.from(Bun.gzipSync(new TextEncoder().encode(json))).toString("base64"); +} + +export function decodeVersion(s: string): StoredVersion { + const bytes = Bun.gunzipSync(Buffer.from(s, "base64")); + return JSON.parse(new TextDecoder().decode(bytes)) as StoredVersion; +} diff --git a/packages/core/test/sync/helpers.ts b/packages/core/test/sync/helpers.ts new file mode 100644 index 0000000..0c8ddeb --- /dev/null +++ b/packages/core/test/sync/helpers.ts @@ -0,0 +1,114 @@ +import type { Db } from "@sttracker/db/db"; +import { newId, nowIso } from "@sttracker/db/ids"; +import type { Database } from "@sttracker/db/schema"; +import { createTestContext } from "@sttracker/plugin-sdk/testing"; +import { definePlugin, type SourcePlugin, type SourceRecord, type WorkHint } from "@sttracker/plugin-sdk/types"; +import type { SyncDeps } from "../../src/sync/run.ts"; + +export { makeUser, makeWork } from "../user-layer-fixtures.ts"; + +/** Insert a `source_connections` row directly and return its id. */ +export async function makeConnection(db: Db, userId: string, pluginId: string): Promise { + const id = newId(); + await db.k + .insertInto("source_connections") + .values({ + id, + user_id: userId, + plugin_id: pluginId, + name: pluginId, + config: "{}", + credentials_enc: null, + schedule_minutes: null, + cursor: null, + status: "active", + last_run_at: null, + created_at: nowIso(), + }) + .execute(); + return id; +} + +/** A source whose records the test mutates between runs. `failAfter` makes `pull` throw after that many records. */ +export interface FakeSource { + records: SourceRecord[]; + failAfter?: number; +} + +export function fakePlugin(id: string, src: FakeSource): SourcePlugin { + async function* gen(): AsyncIterable { + let i = 0; + for (const r of src.records) { + if (src.failAfter !== undefined && i >= src.failAfter) throw new Error("source exploded"); + yield r; + i++; + } + if (src.failAfter !== undefined && i >= src.failAfter) throw new Error("source exploded"); + } + return definePlugin({ + manifest: { id, name: id, version: "0.0.0", sdkVersion: 1, mediaTypes: ["book", "movie", "show"], capabilities: ["pull", "pullChanges"], config: [] }, + pull: () => gen(), + pullChanges: () => gen(), + }); +} + +export function deps(db: Db): SyncDeps { + return { db, makeContext: () => createTestContext() }; +} + +export function book(isbn: string, title = `Book ${isbn}`): WorkHint { + return { title, mediaType: "book", externalIds: { isbn13: isbn } }; +} + +export function entryRec(key: string, work: WorkHint, entry: Extract["entry"], extra: Partial = {}): SourceRecord { + return { key, kind: "entry", work, entry, raw: { key, entry }, ...extra } as SourceRecord; +} + +export function engagementRec( + key: string, + work: WorkHint, + engagement: Extract["engagement"], +): SourceRecord { + return { key, kind: "engagement", work, engagement, raw: { key } }; +} + +export function listenRec(key: string, listen: Extract["listen"]): SourceRecord { + return { key, kind: "listen", listen, raw: { key } }; +} + +export async function count(db: Db, table: keyof Database): Promise { + const r = await db.k + .selectFrom(table) + .select((eb) => eb.fn.countAll().as("n")) + .executeTakeFirstOrThrow(); + return Number(r.n); +} + +export async function events(db: Db, where: { type?: string; status?: string } = {}) { + let q = db.k.selectFrom("change_events").selectAll().orderBy("id"); + if (where.type) q = q.where("type", "=", where.type as never); + if (where.status) q = q.where("status", "=", where.status as never); + const rows = await q.execute(); + return rows.map((r) => ({ ...r, payload: JSON.parse(r.payload) as Record })); +} + +export async function entryFor(db: Db, userId: string, isbn: string) { + const ext = await db.k + .selectFrom("external_ids") + .select("work_id") + .where("namespace", "=", "isbn13") + .where("value", "=", isbn) + .executeTakeFirst(); + if (!ext) return undefined; + const row = await db.k + .selectFrom("entries") + .selectAll() + .where("user_id", "=", userId) + .where("work_id", "=", ext.work_id) + .executeTakeFirst(); + return row ? { ...row, fieldSources: JSON.parse(row.field_sources) as Record } : undefined; +} + +export function listRec(key: string, name: string, items: { work: WorkHint; note?: string }[], description?: string): SourceRecord { + return { key, kind: "list", list: { name, items, ...(description !== undefined ? { description } : {}) }, raw: { key } }; +} diff --git a/packages/core/test/sync/listens.test.ts b/packages/core/test/sync/listens.test.ts new file mode 100644 index 0000000..9422875 --- /dev/null +++ b/packages/core/test/sync/listens.test.ts @@ -0,0 +1,94 @@ +import { expect, test } from "bun:test"; +import { forEachDialect } from "@sttracker/db/testing"; +import { ingestListens } from "../../src/sync/listens.ts"; +import { runSync } from "../../src/sync/run.ts"; +import { count, deps, entryRec, book, fakePlugin, listenRec, makeConnection, makeUser, makeWork } from "./helpers.ts"; + +forEachDialect("listens", (getDb) => { + test("listens insert-or-ignore and mbid matching", async () => { + const db = getDb(); + const userId = await makeUser(db); + const known = await makeWork(db, { title: "Known Song", mediaType: "track", externalIds: { musicbrainz_recording: "rec-1" } }); + const records = [ + listenRec("lb:1", { listenedAt: "2024-01-01T10:00:00Z", artist: "A", track: "Known song", mbids: { recording: "rec-1" } }), + listenRec("lb:2", { listenedAt: "2024-01-01T10:05:00Z", artist: "B", track: "New Song", album: "LP", durationMs: 1000, mbids: { recording: "rec-2" } }), + listenRec("lb:3", { listenedAt: "2024-01-01T10:10:00Z", artist: "C", track: "No Ids" }), + ]; + const works0 = await count(db, "works"); + expect(await ingestListens(db, userId, "listenbrainz", records)).toEqual({ inserted: 3 }); + + const rows = await db.k.selectFrom("listens").selectAll().orderBy("source_key").execute(); + expect(rows.map((r) => r.source_key)).toEqual(["lb:1", "lb:2", "lb:3"]); + expect(rows[0]!.work_id).toBe(known); + expect(rows[2]!.work_id).toBeNull(); + expect(rows[1]!.album).toBe("LP"); + expect(rows[1]!.duration_ms).toBe(1000); + expect(JSON.parse(rows[1]!.mbids)).toEqual({ recording: "rec-2" }); + const created = await db.k.selectFrom("works").selectAll().where("id", "=", rows[1]!.work_id!).executeTakeFirstOrThrow(); + expect(created.media_type).toBe("track"); + expect(created.title).toBe("New Song"); + expect(created.owner_user_id).toBeNull(); + const ext = await db.k.selectFrom("external_ids").selectAll().where("work_id", "=", created.id).execute(); + expect(ext.map((e) => [e.namespace, e.value])).toEqual([["musicbrainz_recording", "rec-2"]]); + expect(await count(db, "works")).toBe(works0 + 1); + // A listen never renames the shared work. + expect((await db.k.selectFrom("works").select("title").where("id", "=", known).executeTakeFirstOrThrow()).title).toBe("Known Song"); + + expect(await ingestListens(db, userId, "listenbrainz", records)).toEqual({ inserted: 0 }); + expect(await count(db, "listens")).toBe(3); + expect(await count(db, "works")).toBe(works0 + 1); + expect(await count(db, "change_events")).toBe(0); + + // Same key under another source, or for another user, is a separate listen. + expect(await ingestListens(db, userId, "lastfm", records.slice(0, 1))).toEqual({ inserted: 1 }); + const bob = await makeUser(db, "bob"); + expect(await ingestListens(db, bob, "listenbrainz", records.slice(0, 1))).toEqual({ inserted: 1 }); + }); + + test("1200 listens ingest correctly and idempotently", async () => { + const db = getDb(); + const userId = await makeUser(db); + const records = Array.from({ length: 1200 }, (_, i) => + listenRec(`k${i}`, { + listenedAt: `2024-01-01T00:00:${String(i % 60).padStart(2, "0")}Z`, + artist: "A", + track: `T${i % 7}`, + ...(i % 3 === 0 ? { mbids: { recording: `rec-${i % 7}` } } : {}), + }), + ); + // A duplicate key inside the same call is inserted once. + records.push(records[5]!); + expect(await ingestListens(db, userId, "lb", records)).toEqual({ inserted: 1200 }); + expect(await count(db, "listens")).toBe(1200); + expect(await count(db, "works")).toBe(7); + const linked = await db.k.selectFrom("listens").select("work_id").where("work_id", "is not", null).execute(); + expect(linked).toHaveLength(400); + expect(await ingestListens(db, userId, "lb", records)).toEqual({ inserted: 0 }); + expect(await count(db, "listens")).toBe(1200); + expect(await count(db, "works")).toBe(7); + }); + + test("runSync routes listens to ingestListens and counts them", async () => { + const db = getDb(); + const userId = await makeUser(db); + const connectionId = await makeConnection(db, userId, "lb"); + const src = { + records: [ + listenRec("1", { listenedAt: "2024-01-01T10:00:00Z", artist: "A", track: "T" }), + listenRec("2", { listenedAt: "2024-01-01T10:01:00Z", artist: "A", track: "U" }), + entryRec("e", book("1"), { rating: 50 }), + ], + }; + const s = await runSync(deps(db), fakePlugin("lb", src), connectionId, "full"); + expect(s.counts.listens).toBe(2); + expect(s.counts.seen).toBe(3); + const rows = await db.k.selectFrom("listens").select(["source", "source_key"]).orderBy("source_key").execute(); + expect(rows).toEqual([{ source: "lb", source_key: "1" }, { source: "lb", source_key: "2" }]); + // Listens are not snapshotted as source records and never removed. + expect(await count(db, "source_records")).toBe(1); + src.records = []; + const s2 = await runSync(deps(db), fakePlugin("lb", src), connectionId, "full"); + expect(s2.counts.removed).toBe(1); + expect(await count(db, "listens")).toBe(2); + }); +}); diff --git a/packages/core/test/sync/run.test.ts b/packages/core/test/sync/run.test.ts new file mode 100644 index 0000000..62cadb7 --- /dev/null +++ b/packages/core/test/sync/run.test.ts @@ -0,0 +1,611 @@ +import { expect, test } from "bun:test"; +import { forEachDialect } from "@sttracker/db/testing"; +import { moveEntryToWork, softDeleteEntry, upsertEntry } from "../../src/entries.ts"; +import { updateList } from "../../src/lists.ts"; +import { listChanges, resolveChange } from "../../src/sync/changes.ts"; +import { runSync } from "../../src/sync/run.ts"; +import { NotFoundError, ValidationError } from "../../src/errors.ts"; +import { + book, + count, + deps, + engagementRec, + entryFor, + entryRec, + events, + fakePlugin, + type FakeSource, + listenRec, + listRec, + makeConnection, + makeUser, + makeWork, +} from "./helpers.ts"; + +forEachDialect("sync engine", (getDb) => { + async function setup(pluginId = "fake") { + const db = getDb(); + const userId = await makeUser(db); + const connectionId = await makeConnection(db, userId, pluginId); + const src: FakeSource = { records: [] }; + const plugin = fakePlugin(pluginId, src); + const actor = `plugin:${pluginId}:${connectionId}`; + return { db, userId, connectionId, src, plugin, actor }; + } + + test("first full pull creates entries and auto_applied events", async () => { + const { db, userId, connectionId, src, plugin, actor } = await setup(); + src.records = [ + entryRec("e1", book("111"), { status: "completed", rating: 80, notes: "loved it" }), + engagementRec("g1", book("111"), { finishedAt: { value: "2024-01-05", precision: "day" } }), + ]; + const s = await runSync(deps(db), plugin, connectionId, "full"); + expect(s.status).toBe("ok"); + expect(s.counts).toEqual({ seen: 2, added: 2, changed: 0, removed: 0, merged: 0, conflicts: 0, autoApplied: 2, pending: 0, listens: 0 }); + + const e = await entryFor(db, userId, "111"); + expect(e?.status).toBe("completed"); + expect(e?.rating).toBe(80); + expect(e?.private_notes).toBe("loved it"); + expect(e?.fieldSources).toEqual({ status: actor, rating: actor, privateNotes: actor }); + + const eg = await db.k.selectFrom("engagements").selectAll().execute(); + expect(eg).toHaveLength(1); + expect(eg[0]!.entry_id).toBe(e!.id); + expect(eg[0]!.finished_at).toBe("2024-01-05"); + expect(await count(db, "engagement_sources")).toBe(1); + + const ev = await events(db); + expect(ev.map((x) => [x.type, x.status])).toEqual([["added", "auto_applied"], ["added", "auto_applied"]]); + expect(ev[0]!.run_id).toBe(s.runId); + expect(ev[0]!.target_entity).toBe("entry"); + expect(ev[0]!.target_id).toBe(e!.id); + expect(ev[1]!.target_entity).toBe("engagement"); + + expect(await count(db, "source_records")).toBe(2); + expect(await count(db, "source_record_versions")).toBe(2); + const rec = await db.k.selectFrom("source_records").selectAll().where("key", "=", "e1").executeTakeFirstOrThrow(); + expect(rec.target_entity).toBe("entry"); + expect(rec.target_id).toBe(e!.id); + + const run = await db.k.selectFrom("source_runs").selectAll().where("id", "=", s.runId).executeTakeFirstOrThrow(); + expect(run.status).toBe("ok"); + expect(run.mode).toBe("full"); + expect(run.finished_at).not.toBeNull(); + expect(JSON.parse(run.counts)).toEqual(s.counts); + const conn = await db.k.selectFrom("source_connections").selectAll().where("id", "=", connectionId).executeTakeFirstOrThrow(); + expect(conn.last_run_at).not.toBeNull(); + expect(conn.status).toBe("active"); + }); + + test("an entry on a child work also gets a status-none entry on the root ancestor", async () => { + const { db, userId, connectionId, src, plugin, actor } = await setup(); + const show = { title: "Show", mediaType: "show" as const, externalIds: { tmdb_show: "9" } }; + const season = { title: "Season 1", mediaType: "season" as const, externalIds: {}, parent: show, position: { season: 1 } }; + const episode = { title: "Pilot", mediaType: "episode" as const, externalIds: { tmdb_episode: "91" }, parent: season, position: { episode: 1 } }; + src.records = [engagementRec("ep1", episode, { finishedAt: { value: "2024-02-01T21:00:00Z", precision: "day" } })]; + await runSync(deps(db), plugin, connectionId, "full"); + + const rows = await db.k + .selectFrom("entries as e") + .innerJoin("works as w", "w.id", "e.work_id") + .select(["w.media_type", "e.status", "e.field_sources"]) + .where("e.user_id", "=", userId) + .orderBy("w.media_type") + .execute(); + expect(rows.map((r) => [r.media_type, r.status])).toEqual([["episode", "none"], ["show", "none"]]); + const hist = await db.k.selectFrom("user_history").select("actor").execute(); + expect(new Set(hist.map((h) => h.actor))).toEqual(new Set([actor])); + + // An existing (even deleted) root entry is never touched. + const showEntry = await db.k + .selectFrom("entries as e") + .innerJoin("works as w", "w.id", "e.work_id") + .select("e.id") + .where("w.media_type", "=", "show") + .executeTakeFirstOrThrow(); + await softDeleteEntry(db, userId, showEntry.id, "user"); + src.records.push(engagementRec("ep2", { ...episode, title: "Two", externalIds: { tmdb_episode: "92" }, position: { episode: 2 } }, {})); + await runSync(deps(db), plugin, connectionId, "full"); + const show2 = await db.k.selectFrom("entries").selectAll().where("id", "=", showEntry.id).executeTakeFirstOrThrow(); + expect(show2.deleted_at).not.toBeNull(); + }); + + test("second identical pull creates nothing new", async () => { + const { db, connectionId, src, plugin } = await setup(); + src.records = [ + entryRec("e1", book("111"), { status: "completed", rating: 80 }), + engagementRec("g1", book("111"), { finishedAt: { value: "2024-01-05", precision: "day" } }), + listenRec("l1", { listenedAt: "2024-01-01T10:00:00Z", artist: "A", track: "T" }), + ]; + const first = await runSync(deps(db), plugin, connectionId, "full"); + expect(first.counts.listens).toBe(1); + const tables = ["source_records", "source_record_versions", "change_events", "entries", "engagements", "engagement_sources", "listens", "user_history", "works"] as const; + const before = await Promise.all(tables.map((t) => count(db, t))); + const second = await runSync(deps(db), plugin, connectionId, "full"); + const after = await Promise.all(tables.map((t) => count(db, t))); + expect(after).toEqual(before); + expect(second.status).toBe("ok"); + expect(second.counts).toEqual({ seen: 3, added: 0, changed: 0, removed: 0, merged: 0, conflicts: 0, autoApplied: 0, pending: 0, listens: 0 }); + expect(await count(db, "source_runs")).toBe(2); + }); + + test("failed pull marks nothing removed", async () => { + const { db, connectionId, src, plugin } = await setup(); + src.records = [entryRec("a", book("1"), { status: "completed" }), entryRec("b", book("2"), { status: "planned" })]; + await runSync(deps(db), plugin, connectionId, "full"); + + src.failAfter = 1; + const s = await runSync(deps(db), plugin, connectionId, "full"); + expect(s.status).toBe("failed"); + expect(s.error).toContain("source exploded"); + expect(s.counts.removed).toBe(0); + const deleted = await db.k.selectFrom("source_records").select("key").where("deleted_at", "is not", null).execute(); + expect(deleted).toEqual([]); + expect(await events(db, { type: "removed" })).toEqual([]); + const run = await db.k.selectFrom("source_runs").selectAll().where("id", "=", s.runId).executeTakeFirstOrThrow(); + expect(run.status).toBe("failed"); + expect(run.error).toContain("source exploded"); + const conn = await db.k.selectFrom("source_connections").selectAll().where("id", "=", connectionId).executeTakeFirstOrThrow(); + expect(conn.status).toBe("error"); + + // A later successful run puts the connection back to active. + src.failAfter = undefined; + expect((await runSync(deps(db), plugin, connectionId, "full")).status).toBe("ok"); + const conn2 = await db.k.selectFrom("source_connections").selectAll().where("id", "=", connectionId).executeTakeFirstOrThrow(); + expect(conn2.status).toBe("active"); + }); + + test("record missing from completed full pull → pending removed; accept soft-deletes entry", async () => { + const { db, userId, connectionId, src, plugin } = await setup(); + src.records = [entryRec("a", book("1"), { status: "completed" }), entryRec("b", book("2"), { status: "planned" })]; + await runSync(deps(db), plugin, connectionId, "full"); + + src.records = [src.records[0]!]; + // A "changes" pull never infers removal from absence. + expect((await runSync(deps(db), plugin, connectionId, "changes")).counts.removed).toBe(0); + expect(await events(db, { type: "removed" })).toEqual([]); + + const s = await runSync(deps(db), plugin, connectionId, "full"); + expect(s.counts.removed).toBe(1); + expect(s.counts.pending).toBe(1); + const rec = await db.k.selectFrom("source_records").selectAll().where("key", "=", "b").executeTakeFirstOrThrow(); + expect(rec.deleted_at).not.toBeNull(); + const removed = await events(db, { type: "removed" }); + expect(removed).toHaveLength(1); + expect(removed[0]!.status).toBe("pending"); + const entryB = await entryFor(db, userId, "2"); + expect(removed[0]!.target_id).toBe(entryB!.id); + expect(entryB!.deleted_at).toBeNull(); // nothing deleted without the user's say + + // Rerun: not re-filed. + await runSync(deps(db), plugin, connectionId, "full"); + expect(await events(db, { type: "removed" })).toHaveLength(1); + + await resolveChange(db, userId, removed[0]!.id, "accept"); + expect((await entryFor(db, userId, "2"))!.deleted_at).not.toBeNull(); + const ev = (await events(db, { type: "removed" }))[0]!; + expect(ev.status).toBe("accepted"); + expect(ev.resolved_at).not.toBeNull(); + }); + + test("deleted: true in a changes pull files a pending removed event", async () => { + const { db, userId, connectionId, src, plugin } = await setup(); + src.records = [entryRec("a", book("1"), { status: "completed" })]; + await runSync(deps(db), plugin, connectionId, "full"); + src.records = [entryRec("a", book("1"), { status: "completed" }, { deleted: true })]; + const s = await runSync(deps(db), plugin, connectionId, "changes"); + expect(s.counts.removed).toBe(1); + const removed = await events(db, { type: "removed", status: "pending" }); + expect(removed).toHaveLength(1); + await resolveChange(db, userId, removed[0]!.id, "ignore"); + expect((await entryFor(db, userId, "1"))!.deleted_at).toBeNull(); + expect((await events(db, { type: "removed" }))[0]!.status).toBe("ignored"); + }); + + test("user-edited rating not overwritten; pending changed event", async () => { + const { db, userId, connectionId, src, plugin, actor } = await setup(); + src.records = [entryRec("a", book("1"), { status: "in_progress", rating: 80 })]; + await runSync(deps(db), plugin, connectionId, "full"); + const e0 = (await entryFor(db, userId, "1"))!; + await upsertEntry(db, userId, e0.work_id, { rating: 90 }, "user"); + + src.records = [entryRec("a", book("1"), { status: "completed", rating: 70 })]; + const s = await runSync(deps(db), plugin, connectionId, "full"); + expect(s.counts.changed).toBe(1); + expect(s.counts.autoApplied).toBe(1); + expect(s.counts.pending).toBe(1); + const e1 = (await entryFor(db, userId, "1"))!; + expect(e1.rating).toBe(90); + expect(e1.status).toBe("completed"); // never edited by the user: applied + expect(e1.fieldSources).toEqual({ status: actor, rating: "user" }); + const pending = await events(db, { type: "changed", status: "pending" }); + expect(pending).toHaveLength(1); + expect(pending[0]!.payload).toMatchObject({ field: "rating", from: 90, to: 70 }); + const applied = await events(db, { type: "changed", status: "auto_applied" }); + expect(applied.map((x) => x.payload)).toEqual([expect.objectContaining({ field: "status", from: "in_progress", to: "completed" })]); + + // Rerun: nothing new. + await runSync(deps(db), plugin, connectionId, "full"); + expect(await events(db, { type: "changed", status: "pending" })).toHaveLength(1); + + const listed = await listChanges(db, userId, {}); + expect(listed.items.map((c) => c.id)).toEqual([pending[0]!.id]); + + await resolveChange(db, userId, pending[0]!.id, "accept"); + const e2 = (await entryFor(db, userId, "1"))!; + expect(e2.rating).toBe(70); + expect(e2.fieldSources.rating).toBe(actor); + expect((await listChanges(db, userId, {})).items).toEqual([]); + await expect(resolveChange(db, userId, pending[0]!.id, "accept")).rejects.toBeInstanceOf(ValidationError); + }); + + test("a user-deleted entry is never revived by a sync", async () => { + const { db, userId, connectionId, src, plugin } = await setup(); + src.records = [entryRec("a", book("1"), { rating: 80 })]; + await runSync(deps(db), plugin, connectionId, "full"); + const e = (await entryFor(db, userId, "1"))!; + await softDeleteEntry(db, userId, e.id, "user"); + const before = await count(db, "change_events"); + // A changed record, a brand-new entry record and a new engagement record, all on the deleted entry's work. + src.records = [ + entryRec("a", book("1"), { rating: 50 }), + entryRec("a2", book("1"), { status: "completed" }), + engagementRec("g", book("1"), { finishedAt: { value: "2024-01-01", precision: "day" } }), + ]; + const s = await runSync(deps(db), plugin, connectionId, "full"); + expect(s.status).toBe("ok"); + const after = (await entryFor(db, userId, "1"))!; + expect(after.deleted_at).not.toBeNull(); + expect(after.rating).toBe(80); + expect(after.status).toBe("none"); + expect(await count(db, "engagements")).toBe(0); + expect(await count(db, "change_events")).toBe(before); + }); + + test("an id-less record is updated through its mapping, not re-matched", async () => { + const { db, userId, connectionId, src, plugin } = await setup(); + const hint = { title: "Obscure Zine", mediaType: "book" as const, externalIds: {} }; + src.records = [entryRec("csv-1", hint, { rating: 40 })]; + await runSync(deps(db), plugin, connectionId, "full"); + const works = await count(db, "works"); + src.records = [entryRec("csv-1", hint, { rating: 60 })]; + await runSync(deps(db), plugin, connectionId, "full"); + expect(await count(db, "works")).toBe(works); + const rows = await db.k.selectFrom("entries").selectAll().where("user_id", "=", userId).execute(); + expect(rows).toHaveLength(1); + expect(rows[0]!.rating).toBe(60); + }); + + test("two connections disagree → one pending conflict, not duplicated on rerun; keep_mine stops further conflicts for that value", async () => { + const { db, userId, connectionId: connA, src: srcA, plugin: pluginA, actor: actorA } = await setup("alpha"); + const connB = await makeConnection(db, userId, "beta"); + const srcB: FakeSource = { records: [] }; + const pluginB = fakePlugin("beta", srcB); + const actorB = `plugin:beta:${connB}`; + + srcA.records = [entryRec("a", book("1"), { rating: 80 })]; + await runSync(deps(db), pluginA, connA, "full"); + srcB.records = [entryRec("b", book("1"), { rating: 60, status: "completed" })]; + const s = await runSync(deps(db), pluginB, connB, "full"); + expect(s.counts.conflicts).toBe(1); + + const e = (await entryFor(db, userId, "1"))!; + expect(e.rating).toBe(80); + expect(e.status).toBe("completed"); // field nobody had set: applied + expect(e.fieldSources).toEqual({ rating: actorA, status: actorB }); + let conflicts = await events(db, { type: "conflict", status: "pending" }); + expect(conflicts).toHaveLength(1); + expect(conflicts[0]!.payload).toMatchObject({ field: "rating", current: 80, currentSource: actorA, incoming: 60, incomingSource: actorB }); + + // Rerun both: nothing new. + await runSync(deps(db), pluginA, connA, "full"); + await runSync(deps(db), pluginB, connB, "full"); + expect(await events(db, { type: "conflict" })).toHaveLength(1); + + // B moves 60 → 70: the stale pending conflict is superseded, not left beside the new one. + srcB.records = [entryRec("b", book("1"), { rating: 70, status: "completed" })]; + await runSync(deps(db), pluginB, connB, "full"); + let all = await events(db, { type: "conflict" }); + expect(all.map((c) => [c.payload.incoming, c.status])).toEqual([[60, "ignored"], [70, "pending"]]); + expect(all[0]!.payload.supersededBy).toBe(all[1]!.id); + expect(all[0]!.resolved_at).not.toBeNull(); + + // And back 70 → 60: still exactly one pending conflict for the field. + srcB.records = [entryRec("b", book("1"), { rating: 60, status: "completed" })]; + await runSync(deps(db), pluginB, connB, "full"); + conflicts = await events(db, { type: "conflict", status: "pending" }); + expect(conflicts.map((c) => c.payload.incoming)).toEqual([60]); + + await resolveChange(db, userId, conflicts[0]!.id, "keep_mine"); + const kept = (await entryFor(db, userId, "1"))!; + expect(kept.rating).toBe(80); + expect(kept.fieldSources.rating).toBe("user"); + all = await events(db, { type: "conflict" }); + expect(all.find((c) => c.id === conflicts[0]!.id)!.status).toBe("kept_mine"); + + // Reruns file nothing; a NEW value from B is now a user-decided pending change, not a conflict. + await runSync(deps(db), pluginA, connA, "full"); + await runSync(deps(db), pluginB, connB, "full"); + expect(await events(db, { type: "conflict" })).toHaveLength(3); + srcB.records = [entryRec("b", book("1"), { rating: 65, status: "completed" })]; + await runSync(deps(db), pluginB, connB, "full"); + expect(await events(db, { type: "conflict" })).toHaveLength(3); + const changed = await events(db, { type: "changed", status: "pending" }); + expect(changed).toHaveLength(1); + expect(changed[0]!.payload).toMatchObject({ field: "rating", from: 80, to: 65 }); + expect((await entryFor(db, userId, "1"))!.rating).toBe(80); + }); + + test("same play from two sources dedupes into one engagement with two engagement_sources", async () => { + const { db, userId, connectionId: connA, src: srcA, plugin: pluginA } = await setup("alpha"); + const connB = await makeConnection(db, userId, "beta"); + const srcB: FakeSource = { records: [] }; + const pluginB = fakePlugin("beta", srcB); + + srcA.records = [engagementRec("play-1", book("1"), { finishedAt: { value: "2024-03-01T20:00:00Z", precision: "day" } })]; + await runSync(deps(db), pluginA, connA, "full"); + srcB.records = [ + engagementRec("row-7", book("1"), { finishedAt: { value: "2024-03-02", precision: "day" }, rating: 90 }), + engagementRec("row-8", book("1"), { finishedAt: { value: "2024-03-10", precision: "day" } }), + ]; + await runSync(deps(db), pluginB, connB, "full"); + + const egs = await db.k.selectFrom("engagements").selectAll().orderBy("finished_at").execute(); + expect(egs).toHaveLength(2); + expect(egs[0]!.finished_at).toBe("2024-03-01T20:00:00Z"); // first source's value kept + expect(egs[0]!.rating).toBe(90); // empty field filled from the second source + const srcs = await db.k.selectFrom("engagement_sources").selectAll().where("engagement_id", "=", egs[0]!.id).execute(); + expect(srcs.map((x) => x.record_key).sort()).toEqual(["play-1", "row-7"]); + expect(await count(db, "engagement_sources")).toBe(3); + + // Equal start datetime also dedupes. + srcA.records.push(engagementRec("play-2", book("1"), { startedAt: { value: "2024-05-01T08:00:00Z", precision: "day" } })); + srcB.records.push(engagementRec("row-9", book("1"), { startedAt: { value: "2024-05-01T08:00:00Z", precision: "day" } })); + await runSync(deps(db), pluginA, connA, "full"); + await runSync(deps(db), pluginB, connB, "full"); + expect(await count(db, "engagements")).toBe(3); + expect(await count(db, "engagement_sources")).toBe(5); + }); + + test("merged record → pending merged; accept moves entry", async () => { + const { db, userId, connectionId, src, plugin } = await setup(); + src.records = [entryRec("x", book("1"), { status: "completed" })]; + await runSync(deps(db), plugin, connectionId, "full"); + const ex = (await entryFor(db, userId, "1"))!; + + // The source merged x into y. Record y is new and creates its own (empty) entry on book 2. + src.records = [entryRec("x", book("1"), { status: "completed" }, { mergedIntoKey: "y" }), entryRec("y", book("2"), {})]; + const s = await runSync(deps(db), plugin, connectionId, "full"); + expect(s.counts.merged).toBe(1); + const merged = await events(db, { type: "merged", status: "pending" }); + expect(merged).toHaveLength(1); + expect(merged[0]!.target_id).toBe(ex.id); + expect(merged[0]!.payload).toMatchObject({ intoKey: "y" }); + expect((await entryFor(db, userId, "1"))!.deleted_at).toBeNull(); + + await resolveChange(db, userId, merged[0]!.id, "accept"); + const ey = (await entryFor(db, userId, "2"))!; + expect(ey.deleted_at).toBeNull(); + expect(ey.status).toBe("completed"); // filled from the moved entry + expect((await entryFor(db, userId, "1"))!.deleted_at).not.toBeNull(); + expect((await events(db, { type: "merged" }))[0]!.status).toBe("accepted"); + }); + + test("a deduped play with a disagreeing value files a conflict instead of dropping it", async () => { + const { db, userId, connectionId: connA, src: srcA, plugin: pluginA, actor: actorA } = await setup("alpha"); + const connB = await makeConnection(db, userId, "beta"); + const pluginB = fakePlugin("beta", { records: [engagementRec("r", book("1"), { finishedAt: { value: "2024-03-01", precision: "day" }, rating: 90 })] }); + srcA.records = [engagementRec("p", book("1"), { finishedAt: { value: "2024-03-01T20:00:00Z", precision: "day" }, rating: 80 })]; + await runSync(deps(db), pluginA, connA, "full"); + const s = await runSync(deps(db), pluginB, connB, "full"); + expect(s.counts.conflicts).toBe(1); + const egs = await db.k.selectFrom("engagements").selectAll().execute(); + expect(egs).toHaveLength(1); + expect(egs[0]!.rating).toBe(80); + expect(egs[0]!.finished_at).toBe("2024-03-01T20:00:00Z"); // the matching date is not a conflict + const c = await events(db, { type: "conflict", status: "pending" }); + expect(c).toHaveLength(1); + expect(c[0]!.target_entity).toBe("engagement"); + expect(c[0]!.payload).toMatchObject({ field: "rating", current: 80, currentSource: actorA, incoming: 90 }); + }); + + test("a merged-away record that vanishes files no removal; the survivor is untouched", async () => { + const { db, userId, connectionId, src, plugin } = await setup(); + src.records = [entryRec("x", book("1"), { status: "completed" })]; + await runSync(deps(db), plugin, connectionId, "full"); + src.records = [entryRec("x", book("1"), { status: "completed" }, { mergedIntoKey: "y" }), entryRec("y", book("2"), {})]; + await runSync(deps(db), plugin, connectionId, "full"); + // Before the user decides, x vanishing files nothing either. + src.records = [entryRec("y", book("2"), {})]; + expect((await runSync(deps(db), plugin, connectionId, "full")).counts.removed).toBe(0); + expect(await events(db, { type: "removed" })).toEqual([]); + + const [merged] = await events(db, { type: "merged", status: "pending" }); + await resolveChange(db, userId, merged!.id, "accept"); + const survivor = (await entryFor(db, userId, "2"))!; + const s = await runSync(deps(db), plugin, connectionId, "full"); + expect(s.counts.removed).toBe(0); + expect(await events(db, { type: "removed" })).toEqual([]); + const after = (await entryFor(db, userId, "2"))!; + expect(after.deleted_at).toBeNull(); + expect(after.status).toBe(survivor.status); + }); + + test("an id-less record whose apply fails is retried against the same draft", async () => { + const { db, userId, connectionId, src, plugin } = await setup(); + const hint = { title: "Obscure Zine", mediaType: "book" as const, externalIds: {} }; + src.records = [entryRec("csv-1", hint, { rating: 150 })]; + const failed = await runSync(deps(db), plugin, connectionId, "full"); + expect(failed.status).toBe("failed"); + expect(failed.counts.added).toBe(0); + src.records = [entryRec("csv-1", hint, { rating: 60 })]; + const ok = await runSync(deps(db), plugin, connectionId, "full"); + expect(ok.status).toBe("ok"); + expect(ok.counts.added).toBe(1); + const drafts = await db.k.selectFrom("works").select("id").where("owner_user_id", "=", userId).execute(); + expect(drafts).toHaveLength(1); + const e = await db.k.selectFrom("entries").selectAll().where("user_id", "=", userId).execute(); + expect(e.map((x) => [x.work_id, x.rating])).toEqual([[drafts[0]!.id, 60]]); + }); + + test("a record-level error fails the run and skips removal", async () => { + const { db, connectionId, src, plugin } = await setup(); + src.records = [entryRec("a", book("1"), { status: "completed" }), entryRec("b", book("2"), { status: "planned" })]; + await runSync(deps(db), plugin, connectionId, "full"); + src.records = [entryRec("a", book("1"), { status: "completed", rating: 999 })]; + const s = await runSync(deps(db), plugin, connectionId, "full"); + expect(s.status).toBe("failed"); + expect(s.error).toContain("record a"); + expect(s.counts.removed).toBe(0); + expect(s.counts.changed).toBe(0); // the failed record's work rolled back, and is not counted + expect(await events(db, { type: "removed" })).toEqual([]); + }); + + test("after the user move-merges an entry, the source keeps syncing into the survivor", async () => { + const { db, userId, connectionId, src, plugin, actor } = await setup(); + src.records = [entryRec("a", book("1"), { rating: 50 })]; + await runSync(deps(db), plugin, connectionId, "full"); + const e1 = (await entryFor(db, userId, "1"))!; + const w2 = await makeWork(db, { title: "Other" }); + const e2 = await upsertEntry(db, userId, w2, { status: "planned" }, "user"); + await moveEntryToWork(db, userId, e1.id, w2, "user"); + const rec = await db.k.selectFrom("source_records").selectAll().where("key", "=", "a").executeTakeFirstOrThrow(); + expect(rec.target_id).toBe(e2.id); + + src.records = [entryRec("a", book("1"), { rating: 60 })]; + const s = await runSync(deps(db), plugin, connectionId, "full"); + expect(s.counts.autoApplied).toBe(1); + const survivor = await db.k.selectFrom("entries").selectAll().where("id", "=", e2.id).executeTakeFirstOrThrow(); + expect(survivor.rating).toBe(60); + expect(JSON.parse(survivor.field_sources).rating).toBe(actor); + }); + + test("accepting removal of a play another source still reports only detaches this source", async () => { + const { db, userId, connectionId: connA, src: srcA, plugin: pluginA } = await setup("alpha"); + const connB = await makeConnection(db, userId, "beta"); + const srcB: FakeSource = { records: [engagementRec("r", book("1"), { finishedAt: { value: "2024-03-01", precision: "day" } })] }; + const pluginB = fakePlugin("beta", srcB); + srcA.records = [engagementRec("p", book("1"), { finishedAt: { value: "2024-03-01", precision: "day" } })]; + await runSync(deps(db), pluginA, connA, "full"); + await runSync(deps(db), pluginB, connB, "full"); + const [eg] = await db.k.selectFrom("engagements").selectAll().execute(); + + srcA.records = []; + await runSync(deps(db), pluginA, connA, "full"); + const [removedA] = await events(db, { type: "removed", status: "pending" }); + expect(removedA!.target_id).toBe(eg!.id); + await resolveChange(db, userId, removedA!.id, "accept"); + let row = await db.k.selectFrom("engagements").selectAll().where("id", "=", eg!.id).executeTakeFirstOrThrow(); + expect(row.deleted_at).toBeNull(); + const left = await db.k.selectFrom("engagement_sources").select("connection_id").where("engagement_id", "=", eg!.id).execute(); + expect(left.map((x) => x.connection_id)).toEqual([connB]); + + srcB.records = []; + await runSync(deps(db), pluginB, connB, "full"); + const [removedB] = await events(db, { type: "removed", status: "pending" }); + await resolveChange(db, userId, removedB!.id, "accept"); + row = await db.k.selectFrom("engagements").selectAll().where("id", "=", eg!.id).executeTakeFirstOrThrow(); + expect(row.deleted_at).not.toBeNull(); + }); + + test("accepting a change whose target is gone resolves it as ignored (target_gone)", async () => { + const { db, userId, connectionId, src, plugin } = await setup(); + src.records = [entryRec("a", book("1"), { rating: 80 })]; + await runSync(deps(db), plugin, connectionId, "full"); + const e = (await entryFor(db, userId, "1"))!; + await upsertEntry(db, userId, e.work_id, { rating: 90 }, "user"); + src.records = [entryRec("a", book("1"), { rating: 70 })]; + await runSync(deps(db), plugin, connectionId, "full"); + const [pending] = await events(db, { type: "changed", status: "pending" }); + await softDeleteEntry(db, userId, e.id, "user"); + await resolveChange(db, userId, pending!.id, "accept"); + const [ev] = await events(db, { type: "changed" }); + expect(ev!.status).toBe("ignored"); + expect(ev!.payload.reason).toBe("target_gone"); + expect(ev!.resolved_at).not.toBeNull(); + }); + + test("list records: create, update, user-edited → pending, removed", async () => { + const { db, userId, connectionId, src, plugin, actor } = await setup(); + const draftHint = { title: "Zine", mediaType: "book" as const, externalIds: {} }; + src.records = [listRec("l1", "Favourites", [{ work: book("1") }, { work: draftHint, note: "rare" }])]; + const s1 = await runSync(deps(db), plugin, connectionId, "full"); + expect(s1.counts.added).toBe(1); + const list = await db.k.selectFrom("lists").selectAll().where("user_id", "=", userId).executeTakeFirstOrThrow(); + expect([list.name, list.description, list.kind]).toEqual(["Favourites", "Imported from fake", "list"]); + const itemsOf = async () => + (await db.k.selectFrom("list_items as i").innerJoin("works as w", "w.id", "i.work_id").select(["w.title", "i.note"]).where("i.list_id", "=", list.id).orderBy("i.position").execute()).map((r) => [r.title, r.note]); + expect(await itemsOf()).toEqual([["Book 1", null], ["Zine", "rare"]]); + const rec = await db.k.selectFrom("source_records").selectAll().where("key", "=", "l1").executeTakeFirstOrThrow(); + expect([rec.target_entity, rec.target_id]).toEqual(["list", list.id]); + + // Source update, list never touched by the user: name and items applied. The ID-less item reuses its draft. + src.records = [listRec("l1", "Top", [{ work: draftHint, note: "rare" }, { work: book("2") }])]; + await runSync(deps(db), plugin, connectionId, "full"); + expect((await db.k.selectFrom("lists").select("name").where("id", "=", list.id).executeTakeFirstOrThrow()).name).toBe("Top"); + expect(await itemsOf()).toEqual([["Zine", "rare"], ["Book 2", null]]); + expect(await db.k.selectFrom("works").select("id").where("owner_user_id", "=", userId).execute()).toHaveLength(1); + const applied = await events(db, { type: "changed", status: "auto_applied" }); + expect(applied.map((e) => e.payload.field).sort()).toEqual(["items", "name"]); + + // The user edits the list: further source changes become pending. + await updateList(db, userId, list.id, { name: "Mine" }, "user"); + src.records = [listRec("l1", "Top 2", [{ work: book("3") }])]; + const s3 = await runSync(deps(db), plugin, connectionId, "full"); + expect(s3.counts.pending).toBe(2); + expect((await db.k.selectFrom("lists").select("name").where("id", "=", list.id).executeTakeFirstOrThrow()).name).toBe("Mine"); + expect(await itemsOf()).toEqual([["Zine", "rare"], ["Book 2", null]]); + const pending = await events(db, { type: "changed", status: "pending" }); + expect(pending.map((e) => e.payload.field).sort()).toEqual(["items", "name"]); + const itemsEv = pending.find((e) => e.payload.field === "items")!; + expect((itemsEv.payload.to as unknown[]).length).toBe(1); + await resolveChange(db, userId, itemsEv.id, "accept"); + expect(await itemsOf()).toEqual([["Book 3", null]]); + const hist = await db.k.selectFrom("user_history").select("actor").where("entity_id", "=", list.id).orderBy("created_at", "desc").orderBy("id", "desc").executeTakeFirstOrThrow(); + expect(hist.actor).toBe(actor); + + // Removed → pending; accept soft-deletes the list. + src.records = []; + await runSync(deps(db), plugin, connectionId, "full"); + const [removed] = await events(db, { type: "removed", status: "pending" }); + expect(removed!.target_entity).toBe("list"); + expect((await db.k.selectFrom("lists").select("deleted_at").where("id", "=", list.id).executeTakeFirstOrThrow()).deleted_at).toBeNull(); + await resolveChange(db, userId, removed!.id, "accept"); + expect((await db.k.selectFrom("lists").select("deleted_at").where("id", "=", list.id).executeTakeFirstOrThrow()).deleted_at).not.toBeNull(); + }); + + test("resolveChange is scoped to the user", async () => { + const { db, userId, connectionId, src, plugin } = await setup(); + src.records = [entryRec("a", book("1"), { status: "completed" })]; + await runSync(deps(db), plugin, connectionId, "full"); + src.records = []; + await runSync(deps(db), plugin, connectionId, "full"); + const [ev] = await events(db, { type: "removed" }); + const other = await makeUser(db, "bob"); + await expect(resolveChange(db, other, ev!.id, "accept")).rejects.toBeInstanceOf(NotFoundError); + expect((await listChanges(db, other, { status: "all" })).items).toEqual([]); + expect((await listChanges(db, userId, { status: "all" })).items).toHaveLength(2); + }); + + test("listChanges pages newest first and filters by connection", async () => { + const { db, userId, connectionId, src, plugin } = await setup(); + src.records = [1, 2, 3].map((i) => entryRec(`k${i}`, book(String(i)), { status: "planned" })); + await runSync(deps(db), plugin, connectionId, "full"); + const p1 = await listChanges(db, userId, { status: "all", limit: 2 }); + expect(p1.items).toHaveLength(2); + expect(p1.nextCursor).not.toBeNull(); + const p2 = await listChanges(db, userId, { status: "all", limit: 2, cursor: p1.nextCursor! }); + expect(p2.items).toHaveLength(1); + expect(p2.nextCursor).toBeNull(); + const ids = [...p1.items, ...p2.items].map((c) => c.id); + expect(ids).toEqual([...ids].sort().reverse()); + expect(p1.items[0]!.payload).toMatchObject({ key: "k3" }); + expect((await listChanges(db, userId, { status: "all", connectionId: "nope" })).items).toEqual([]); + }); + + test("a plugin that does not match the connection is refused", async () => { + const { db, connectionId } = await setup("alpha"); + await expect(runSync(deps(db), fakePlugin("beta", { records: [] }), connectionId, "full")).rejects.toBeInstanceOf(ValidationError); + }); +}); diff --git a/packages/core/test/sync/snapshot.test.ts b/packages/core/test/sync/snapshot.test.ts new file mode 100644 index 0000000..8e14eba --- /dev/null +++ b/packages/core/test/sync/snapshot.test.ts @@ -0,0 +1,31 @@ +import { expect, test } from "bun:test"; +import type { SourceRecord } from "@sttracker/plugin-sdk/types"; +import { decodeVersion, encodeVersion, hashRecord, normalizeRecord } from "../../src/sync/snapshot.ts"; + +const rec = (raw: unknown, entry: Record): SourceRecord => + ({ key: "k", kind: "entry", work: { title: "T", mediaType: "book", externalIds: { a: "1", b: "2" } }, entry, raw }) as SourceRecord; + +test("hash ignores raw payload and key order", async () => { + const a = rec({ x: 1 }, { rating: 1, status: "planned" }); + const b = rec({ y: 2 }, { status: "planned", rating: 1 }); + expect(await hashRecord(a)).toBe(await hashRecord(b)); + expect(await hashRecord(a)).toMatch(/^[0-9a-f]{64}$/); +}); + +test("hash covers normalised fields, deleted and mergedIntoKey", async () => { + const base = rec({}, { rating: 1 }); + const h = await hashRecord(base); + expect(await hashRecord(rec({}, { rating: 2 }))).not.toBe(h); + expect(await hashRecord({ ...base, deleted: true })).not.toBe(h); + expect(await hashRecord({ ...base, mergedIntoKey: "z" })).not.toBe(h); + expect(await hashRecord({ ...base, deleted: false })).toBe(h); +}); + +test("versions round-trip the normalised record and the raw payload", () => { + const r = rec({ deep: { list: [1, 2] } }, { rating: 5 }); + const encoded = encodeVersion(r); + expect(typeof encoded).toBe("string"); + const v = decodeVersion(encoded); + expect(v.raw).toEqual({ deep: { list: [1, 2] } }); + expect(v.normalized).toEqual(normalizeRecord(r)); +});