From 37a0b0ba94a0a2c485a52ee02f65dc90a4b2e86b Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Fri, 30 Oct 2026 20:26:03 +0100 Subject: [PATCH 1/2] feat(core): source record snapshots with stable hash and gzipped versions Co-Authored-By: Claude Opus 5.5 --- packages/core/src/sync/snapshot.ts | 75 ++++++++++++++++++++++++ packages/core/test/sync/snapshot.test.ts | 31 ++++++++++ 2 files changed, 106 insertions(+) create mode 100644 packages/core/src/sync/snapshot.ts create mode 100644 packages/core/test/sync/snapshot.test.ts 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/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)); +}); From 1975e971f98790d1868c7fa0ccb332ec132c0221 Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Fri, 30 Oct 2026 20:26:03 +0100 Subject: [PATCH 2/2] feat(core): sync engine and changes inbox runSync snapshots and diffs source records, applies new records and never-edited fields, files pending changed/conflict/removed/merged events, de-duplicates plays across sources and ingests listens. listChanges/resolveChange drive the Changes inbox. Co-Authored-By: Claude Opus 5.5 --- docs/ARCHITECTURE.md | 27 +- docs/DECISIONS.md | 48 ++ docs/TODO.md | 9 + packages/core/src/sync/apply.ts | 605 ++++++++++++++++++++++++ packages/core/src/sync/changes.ts | 238 ++++++++++ packages/core/src/sync/listens.ts | 81 ++++ packages/core/src/sync/run.ts | 164 +++++++ packages/core/test/sync/helpers.ts | 110 +++++ packages/core/test/sync/listens.test.ts | 71 +++ packages/core/test/sync/run.test.ts | 409 ++++++++++++++++ 10 files changed, 1755 insertions(+), 7 deletions(-) create mode 100644 packages/core/src/sync/apply.ts create mode 100644 packages/core/src/sync/changes.ts create mode 100644 packages/core/src/sync/listens.ts create mode 100644 packages/core/src/sync/run.ts create mode 100644 packages/core/test/sync/helpers.ts create mode 100644 packages/core/test/sync/listens.test.ts create mode 100644 packages/core/test/sync/run.test.ts diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 16699ad..e249abd 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -16,20 +16,33 @@ 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`. Change engine and export 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`, and `sync/` (the change engine: `snapshot` hashing and versions, `apply` per-record diff and apply, `run` = `runSync`, `changes` = Changes inbox `listChanges`/`resolveChange`, `listens` = `ingestListens`). Export planned | 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), then in one apply transaction create/patch the entry or + engagement (de-duplicating plays across sources) 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 4c69ff3..994c3d0 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -4,6 +4,54 @@ 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.** A pending `changed`/`conflict` for the same (record, field, incoming value) is never + filed twice. `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`. A changed work hint + refreshes the mapped work via `updateWorkFromHint` (shared works only). +- **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 only fields that + are empty and that no source has set; the first source's values otherwise stand. +- **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`. +- **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. +- **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`. + +**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:** `list` records are snapshotted but not applied, and removal of a de-duplicated engagement +deletes it even if another source still reports it (both in `docs/TODO.md`). + ## 2026-10-30: User layer: per-user moves merge entries, keep aliases, and roll the library up by hierarchy **Status:** accepted **Context:** Task 6 built the user layer (`packages/core/src/{entries,engagements,history,lists,copies}.ts`). diff --git a/docs/TODO.md b/docs/TODO.md index 75fa908..d0e8e04 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -24,6 +24,15 @@ Small, specific, independently actionable. Remove only when actually done. file, and compare steps and token use. ## Known issues +- [ ] Sync: `list` records are snapshotted but not applied to `lists` (Trakt/Hardcover emit them). Decide + the mapping (one list per record, items by matched work) and apply it through the change engine. +- [ ] Sync: accepting a `removed` event on an engagement de-duplicated across sources soft-deletes it even if + another connection still reports it; consider detaching that connection's `engagement_sources` row instead. +- [ ] Sync: pending `changed`/`conflict` events for a value the source has since moved past stay pending; + consider superseding them when a newer value for the same field arrives. +- [ ] 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. +- [ ] Sync: `ingestListens` does one lookup and one insert per listen; batch it for large backfills. - [ ] Admin merge does not merge colliding children recursively; colliding children stay under the tombstoned parent. - [ ] Matching: a shared parent resolved without external IDs or position (title/year reuse in `packages/core/src/matching.ts`) has no unique key, so concurrent matches can create it twice. diff --git a/packages/core/src/sync/apply.ts b/packages/core/src/sync/apply.ts new file mode 100644 index 0000000..1e36bdc --- /dev/null +++ b/packages/core/src/sync/apply.ts @@ -0,0 +1,605 @@ +import type { Db } from "@sttracker/db/db"; +import { newId, nowIso } from "@sttracker/db/ids"; +import { fromJson, toJson } from "@sttracker/db/json"; +import type { ChangeStatus, ChangeType, Database } from "@sttracker/db/schema"; +import type { EngagementFields, EntryFields, PartialDate, SourceRecord } from "@sttracker/plugin-sdk/types"; +import { addEngagement, type EngagementPatch, updateEngagement } from "../engagements.ts"; +import { type EntryPatch, fromStoredDate, resolveVisibleWork, upsertEntry } from "../entries.ts"; +import type { Actor, Executor } from "../history.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; + runId: string; + actor: Actor; + counts: RunCounts; + log: string[]; +} + +type SourceRecordRow = Database["source_records"]; +type Fields = Record; + +// --------------------------------------------------------------------------------------------- +// 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; +} + +/** 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, + }; +} + +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); + return {}; +} + +// --------------------------------------------------------------------------------------------- +// Change events +// --------------------------------------------------------------------------------------------- + +async function writeEvent( + k: Executor, + st: SyncState, + e: { recordId: string; type: ChangeType; status: ChangeStatus; targetEntity: string; targetId: string | null; payload: Fields }, +): Promise { + await k + .insertInto("change_events") + .values({ + 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++; +} + +/** + * Anti-flip-flop: a pending changed/conflict event for the same (record, field, incoming value) is + * never filed twice. + */ +async function hasPendingFieldEvent(k: Executor, recordId: string, field: string, incoming: unknown): Promise { + const rows = await k + .selectFrom("change_events") + .select("payload") + .where("record_id", "=", recordId) + .where("status", "=", "pending") + .where("type", "in", ["changed", "conflict"]) + .execute(); + return rows.some((r) => { + const p = fromJson(r.payload, {}); + return p.field === field && sameValue("incoming" in p ? p.incoming : p.to, incoming); + }); +} + +interface Target { + entity: "entry" | "engagement"; + id: string; + current: Fields; + fieldSources: Record; +} + +/** + * Decide each incoming field against the target's `field_sources` (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. + * 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]; + if (sameValue(current, value)) continue; + const source = t.fieldSources[field]; + if (source === undefined || source === st.actor) { + patch[field] = value; + from[field] = current ?? null; + continue; + } + if (await hasPendingFieldEvent(k, recordId, field, value)) continue; + if (source === "user") { + await writeEvent(k, st, { + recordId, + type: "changed", + status: "pending", + targetEntity: t.entity, + targetId: t.id, + payload: { field, from: current ?? null, to: value }, + }); + } else { + await writeEvent(k, st, { + 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 }; +} + +async function applyPatch(k: Executor, st: SyncState, t: Target, patch: Fields): 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(st.db, st.userId, row.work_id, patch as Partial, st.actor, k); + } else { + await updateEngagement(st.db, st.userId, t.id, patch as Partial, st.actor, k); + } +} + +/** The live target of a mapped record, or null (unmapped, gone, or soft-deleted: never revived). */ +async function loadTarget(k: Executor, userId: string, entity: string | null, id: string | null): Promise { + if (!id) return null; + if (entity === "entry") { + const r = await k + .selectFrom("entries") + .selectAll() + .where("id", "=", id) + .where("user_id", "=", 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", "=", 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; + } + 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. */ +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 r; + if (start && s && sameStartInstant(start, s)) return r; + } + return null; +} + +const isEmptyValue = (v: unknown) => + v === null || v === undefined || (typeof v === "object" && !Array.isArray(v) && Object.keys(v as object).length === 0); + +// --------------------------------------------------------------------------------------------- +// Records +// --------------------------------------------------------------------------------------------- + +/** Snapshot one non-listen record and apply it. Call outside any transaction (matchWork has its own). */ +export async function applyRecord(st: SyncState, rec: SourceRecord): Promise { + const hash = hashRecord(rec); + const existing = await st.db.k + .selectFrom("source_records") + .selectAll() + .where("connection_id", "=", st.connectionId) + .where("key", "=", rec.key) + .executeTakeFirst(); + if (!existing) return addRecord(st, rec, hash); + if (existing.current_hash === hash) { + await st.db.k.transaction().execute(async (tx) => { + await tx.updateTable("source_records").set({ last_seen: nowIso() }).where("id", "=", existing.id).execute(); + if (existing.deleted_at !== null && rec.deleted !== true) await reappeared(tx, existing); + }); + return; + } + return changeRecord(st, existing, rec, hash); +} + +async function addRecord(st: SyncState, rec: SourceRecord, hash: string): Promise { + const applicable = (rec.kind === "entry" || rec.kind === "engagement") && rec.deleted !== true && !rec.mergedIntoKey; + // Matching runs in its own transaction (it retries on unique violations), before the apply transaction. + const workId = + applicable && (rec.kind === "entry" || rec.kind === "engagement") ? (await matchWork(st.db, st.userId, rec.work, st.actor)).workId : null; + await st.db.k.transaction().execute(async (tx) => { + const now = nowIso(); + const recordId = newId(); + await tx + .insertInto("source_records") + .values({ + id: recordId, + connection_id: st.connectionId, + key: rec.key, + kind: rec.kind, + current_hash: hash, + first_seen: now, + last_seen: now, + deleted_at: rec.deleted === true ? now : null, + merged_into_key: rec.mergedIntoKey ?? null, + target_entity: rec.kind === "entry" || rec.kind === "engagement" ? rec.kind : null, + target_id: null, + }) + .execute(); + await tx.insertInto("source_record_versions").values({ id: newId(), record_id: recordId, hash, raw_gz: encodeVersion(rec), seen_at: now }).execute(); + if (!applicable || workId === null) { + if (rec.kind === "list") st.log.push(`record ${rec.key}: list records are stored but not applied`); + return; + } + const targetId = + rec.kind === "entry" + ? await addEntryRecord(tx, st, recordId, rec.key, workId, entryPatchFrom(rec.entry)) + : rec.kind === "engagement" + ? await addEngagementRecord(tx, st, recordId, rec.key, workId, engagementPatchFrom(rec.engagement)) + : null; + if (targetId) { + await tx.updateTable("source_records").set({ target_id: targetId }).where("id", "=", recordId).execute(); + st.counts.added++; + } + }); +} + +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, t, patch); + 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 applied: string[]; + let payload: Fields; + if (dup) { + // Same play from another source: attach, and only fill fields nobody has set yet. + const current = engagementValues(dup); + const sources = fromJson>(dup.field_sources, {}); + const fill: Fields = {}; + for (const [f, v] of Object.entries(incoming)) { + if (sources[f] === undefined && isEmptyValue(current[f]) && !isEmptyValue(v)) fill[f] = v; + } + if (Object.keys(fill).length > 0) await updateEngagement(st.db, st.userId, dup.id, fill as Partial, st.actor, k); + engagementId = dup.id; + applied = Object.keys(fill); + payload = { key, kind: "engagement", fields: applied, dedupedInto: dup.id }; + } else { + const eg = await addEngagement(st.db, st.userId, live.row.id, incoming as Partial, st.actor, k); + engagementId = eg.id; + applied = Object.keys(incoming); + payload = { key, kind: "engagement", fields: applied }; + } + 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 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); + + await st.db.k.transaction().execute(async (tx) => { + const now = nowIso(); + await tx.insertInto("source_record_versions").values({ id: newId(), record_id: existing.id, hash, raw_gz: encodeVersion(rec), seen_at: now }).execute(); + await tx + .updateTable("source_records") + .set({ current_hash: hash, last_seen: now, merged_into_key: rec.mergedIntoKey ?? null }) + .where("id", "=", existing.id) + .execute(); + + if (rec.deleted === true) { + if (existing.deleted_at === null) await markRemoved(tx, st, existing); + 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.userId, 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; + const { patch, from } = await decideFields(tx, st, existing.id, target, incoming); + await applyPatch(tx, st, target, patch); + 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 { + 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. */ +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(); + st.counts.removed++; + const target = await loadTarget(k, st.userId, 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.userId, rec.target_entity, rec.target_id); + if (!target) 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..7ee7f3f --- /dev/null +++ b/packages/core/src/sync/changes.ts @@ -0,0 +1,238 @@ +import type { Db } from "@sttracker/db/db"; +import { nowIso } from "@sttracker/db/ids"; +import { fromJson } 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 { matchWork } from "../matching.ts"; +import { 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; 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`. + */ +export async function resolveChange( + db: Db, + userId: string, + changeId: string, + action: "accept" | "ignore" | "keep_mine", +): Promise { + if (!(action in STATUS_FOR)) 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; + if (ev.type === "changed" || ev.type === "conflict") { + const field = String(ev.payload.field); + if (action === "accept") { + const value = ev.type === "conflict" ? ev.payload.incoming : ev.payload.to; + await setField(db, tx, userId, ev, field, value, actor); + } else { + await setField(db, tx, userId, ev, field, undefined, "user"); + } + return; + } + if (action !== "accept") return; + if (ev.type === "removed") return removeTarget(db, tx, userId, ev, actor); + if (ev.type === "merged" && mergeTarget) return moveTo(db, tx, userId, ev, mergeTarget, actor); + }); +} + +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"; +} + +/** Write `value` to the target's field as `actor`; with `value` undefined, re-stamp the current value. */ +async function setField(db: Db, k: Executor, userId: string, ev: ChangeEvent, field: string, value: unknown, actor: Actor): Promise { + if (!ev.targetId) throw new NotFoundError(ev.targetEntity); + if (ev.targetEntity === "entry") { + const r = await k + .selectFrom("entries") + .selectAll() + .where("id", "=", ev.targetId) + .where("user_id", "=", userId) + .where("deleted_at", "is", null) + .executeTakeFirst(); + if (!r) throw new NotFoundError("entry"); + const v = value === undefined ? entryValues(r)[field] : value; + await upsertEntry(db, userId, r.work_id, { [field]: v } as Partial, actor, k); + return; + } + if (ev.targetEntity === "engagement") { + const r = await k + .selectFrom("engagements as g") + .innerJoin("entries as e", "e.id", "g.entry_id") + .selectAll("g") + .where("g.id", "=", ev.targetId) + .where("e.user_id", "=", userId) + .executeTakeFirst(); + if (!r) throw new NotFoundError("engagement"); + const v = value === undefined ? engagementValues(r)[field] : value; + await updateEngagement(db, userId, r.id, { [field]: v } as Partial, actor, k); + return; + } + throw new ValidationError("targetEntity", "unsupported"); +} + +async function removeTarget(db: Db, k: Executor, userId: string, ev: ChangeEvent, actor: Actor): Promise { + if (!ev.targetId) return; + if (ev.targetEntity === "entry") { + const live = await k + .selectFrom("entries") + .select("id") + .where("id", "=", ev.targetId) + .where("user_id", "=", userId) + .where("deleted_at", "is", null) + .executeTakeFirst(); + if (live) await softDeleteEntry(db, userId, live.id, actor, k); + } else if (ev.targetEntity === "engagement") { + const live = await k + .selectFrom("engagements as g") + .innerJoin("entries as e", "e.id", "g.entry_id") + .select("g.id") + .where("g.id", "=", ev.targetId) + .where("e.user_id", "=", userId) + .where("g.deleted_at", "is", null) + .where("e.deleted_at", "is", null) + .executeTakeFirst(); + if (live) await softDeleteEngagement(db, userId, live.id, 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"); + if (target.target_id) { + 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; +} + +async function moveTo(db: Db, k: Executor, userId: string, ev: ChangeEvent, workId: string, actor: Actor): Promise { + if (!ev.targetId) return; + const entry = await k + .selectFrom("entries") + .select(["id", "work_id"]) + .where("id", "=", ev.targetId) + .where("user_id", "=", userId) + .where("deleted_at", "is", null) + .executeTakeFirst(); + if (!entry) throw new NotFoundError("entry"); + if (entry.work_id === workId) return; + const result = await moveEntryToWork(db, userId, entry.id, workId, actor, k); + if (result.id !== entry.id) { + // The entry merged into an existing one: records that mapped to it now map to the survivor. + await k + .updateTable("source_records") + .set({ target_id: result.id }) + .where("target_entity", "=", "entry") + .where("target_id", "=", entry.id) + .execute(); + } +} diff --git a/packages/core/src/sync/listens.ts b/packages/core/src/sync/listens.ts new file mode 100644 index 0000000..2eabc61 --- /dev/null +++ b/packages/core/src/sync/listens.ts @@ -0,0 +1,81 @@ +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; + } + } +} + +/** + * 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. + */ +export async function ingestListens( + db: Db, + userId: string, + source: string, + records: SourceRecord[], + actor: Actor = `plugin:${source}:ingest`, +): Promise<{ inserted: number }> { + let inserted = 0; + for (const r of records) { + if (r.kind !== "listen") continue; + const exists = await db.k + .selectFrom("listens") + .select("id") + .where("user_id", "=", userId) + .where("source", "=", source) + .where("source_key", "=", r.key) + .executeTakeFirst(); + if (exists) continue; + const l = r.listen; + const recording = l.mbids?.recording; + const workId = recording ? await trackWork(db, recording, l, actor) : null; + const res = await db.k + .insertInto("listens") + .values({ + id: newId(), + user_id: userId, + listened_at: l.listenedAt, + artist: l.artist, + track: l.track, + album: l.album ?? null, + duration_ms: l.durationMs ?? null, + mbids: toJson(l.mbids ?? {}), + work_id: workId, + source, + source_key: r.key, + created_at: nowIso(), + }) + .onConflict((oc) => oc.columns(["user_id", "source", "source_key"]).doNothing()) + .executeTakeFirst(); + 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..ee321df --- /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, 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, 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 db.k.transaction().execute((tx) => markRemoved(tx, st, 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/test/sync/helpers.ts b/packages/core/test/sync/helpers.ts new file mode 100644 index 0000000..1563304 --- /dev/null +++ b/packages/core/test/sync/helpers.ts @@ -0,0 +1,110 @@ +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; +} diff --git a/packages/core/test/sync/listens.test.ts b/packages/core/test/sync/listens.test.ts new file mode 100644 index 0000000..55bf8f5 --- /dev/null +++ b/packages/core/test/sync/listens.test.ts @@ -0,0 +1,71 @@ +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("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..c82260c --- /dev/null +++ b/packages/core/test/sync/run.test.ts @@ -0,0 +1,409 @@ +import { expect, test } from "bun:test"; +import { forEachDialect } from "@sttracker/db/testing"; +import { softDeleteEntry, upsertEntry } from "../../src/entries.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, + makeConnection, + makeUser, +} 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. Then B flips 60 → 70 → 60: the conflict for 60 is not filed twice. + await runSync(deps(db), pluginA, connA, "full"); + await runSync(deps(db), pluginB, connB, "full"); + expect(await events(db, { type: "conflict" })).toHaveLength(1); + srcB.records = [entryRec("b", book("1"), { rating: 70, status: "completed" })]; + await runSync(deps(db), pluginB, connB, "full"); + 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, 70]); + + 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"); + expect((await events(db, { type: "conflict" }))[0]!.status).toBe("kept_mine"); + + await runSync(deps(db), pluginA, connA, "full"); + await runSync(deps(db), pluginB, connB, "full"); + expect(await events(db, { type: "conflict" })).toHaveLength(2); + 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("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); + }); +});