merge: task 8 sync engine
This commit is contained in:
Lucas Winther
@@ -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<string, unknown>;
|
||||
|
||||
// ---------------------------------------------------------------------------------------------
|
||||
// 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<string, unknown> | 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<string, unknown> | 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<string[]>(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<Record<string, unknown>>(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<void> {
|
||||
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<boolean> {
|
||||
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<Fields>(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<string, string>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<void> {
|
||||
if (Object.keys(patch).length === 0) return;
|
||||
if (t.entity === "entry") {
|
||||
const row = await k.selectFrom("entries").select("work_id").where("id", "=", t.id).executeTakeFirstOrThrow();
|
||||
await upsertEntry(st.db, st.userId, row.work_id, patch as Partial<EntryPatch>, st.actor, k);
|
||||
} else {
|
||||
await updateEngagement(st.db, st.userId, t.id, patch as Partial<EngagementPatch>, 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<Target | null> {
|
||||
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<void> {
|
||||
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<EntryPatch>, 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<void> {
|
||||
const hash = hashRecord(rec);
|
||||
const existing = await st.db.k
|
||||
.selectFrom("source_records")
|
||||
.selectAll()
|
||||
.where("connection_id", "=", st.connectionId)
|
||||
.where("key", "=", rec.key)
|
||||
.executeTakeFirst();
|
||||
if (!existing) return addRecord(st, rec, hash);
|
||||
if (existing.current_hash === hash) {
|
||||
await st.db.k.transaction().execute(async (tx) => {
|
||||
await tx.updateTable("source_records").set({ last_seen: nowIso() }).where("id", "=", existing.id).execute();
|
||||
if (existing.deleted_at !== null && rec.deleted !== true) await reappeared(tx, existing);
|
||||
});
|
||||
return;
|
||||
}
|
||||
return changeRecord(st, existing, rec, hash);
|
||||
}
|
||||
|
||||
async function addRecord(st: SyncState, rec: SourceRecord, hash: string): Promise<void> {
|
||||
const applicable = (rec.kind === "entry" || rec.kind === "engagement") && rec.deleted !== true && !rec.mergedIntoKey;
|
||||
// Matching runs in its own transaction (it retries on unique violations), before the apply transaction.
|
||||
const workId =
|
||||
applicable && (rec.kind === "entry" || rec.kind === "engagement") ? (await matchWork(st.db, st.userId, rec.work, st.actor)).workId : null;
|
||||
await st.db.k.transaction().execute(async (tx) => {
|
||||
const now = nowIso();
|
||||
const recordId = newId();
|
||||
await tx
|
||||
.insertInto("source_records")
|
||||
.values({
|
||||
id: recordId,
|
||||
connection_id: st.connectionId,
|
||||
key: rec.key,
|
||||
kind: rec.kind,
|
||||
current_hash: hash,
|
||||
first_seen: now,
|
||||
last_seen: now,
|
||||
deleted_at: rec.deleted === true ? now : null,
|
||||
merged_into_key: rec.mergedIntoKey ?? null,
|
||||
target_entity: rec.kind === "entry" || rec.kind === "engagement" ? rec.kind : null,
|
||||
target_id: null,
|
||||
})
|
||||
.execute();
|
||||
await tx.insertInto("source_record_versions").values({ id: newId(), record_id: recordId, hash, raw_gz: encodeVersion(rec), seen_at: now }).execute();
|
||||
if (!applicable || workId === null) {
|
||||
if (rec.kind === "list") st.log.push(`record ${rec.key}: list records are stored but not applied`);
|
||||
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<string | null> {
|
||||
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<string | null> {
|
||||
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<Record<string, string>>(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<EngagementPatch>, 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<EngagementPatch>, 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<void> {
|
||||
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<string | 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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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 },
|
||||
});
|
||||
}
|
||||
@@ -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<string, unknown>;
|
||||
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<Record<string, unknown>>(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<void> {
|
||||
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<Actor> {
|
||||
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<void> {
|
||||
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<EntryPatch>, 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<EngagementPatch>, actor, k);
|
||||
return;
|
||||
}
|
||||
throw new ValidationError("targetEntity", "unsupported");
|
||||
}
|
||||
|
||||
async function removeTarget(db: Db, k: Executor, userId: string, ev: ChangeEvent, actor: Actor): Promise<void> {
|
||||
if (!ev.targetId) return;
|
||||
if (ev.targetEntity === "entry") {
|
||||
const live = await k
|
||||
.selectFrom("entries")
|
||||
.select("id")
|
||||
.where("id", "=", ev.targetId)
|
||||
.where("user_id", "=", userId)
|
||||
.where("deleted_at", "is", null)
|
||||
.executeTakeFirst();
|
||||
if (live) await softDeleteEntry(db, userId, live.id, actor, k);
|
||||
} else if (ev.targetEntity === "engagement") {
|
||||
const live = await k
|
||||
.selectFrom("engagements as g")
|
||||
.innerJoin("entries as e", "e.id", "g.entry_id")
|
||||
.select("g.id")
|
||||
.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<string> {
|
||||
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<void> {
|
||||
if (!ev.targetId) return;
|
||||
const entry = await k
|
||||
.selectFrom("entries")
|
||||
.select(["id", "work_id"])
|
||||
.where("id", "=", ev.targetId)
|
||||
.where("user_id", "=", userId)
|
||||
.where("deleted_at", "is", null)
|
||||
.executeTakeFirst();
|
||||
if (!entry) throw new NotFoundError("entry");
|
||||
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();
|
||||
}
|
||||
}
|
||||
@@ -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<string> {
|
||||
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 };
|
||||
}
|
||||
@@ -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<string, unknown>;
|
||||
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<Record<string, unknown>>(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<SourceRecord> {
|
||||
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<RunSummary> {
|
||||
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<string>();
|
||||
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 } : {}) };
|
||||
}
|
||||
@@ -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<string, unknown>;
|
||||
engagement?: Record<string, unknown>;
|
||||
listen?: Record<string, unknown>;
|
||||
list?: Record<string, unknown>;
|
||||
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<string, unknown> = {};
|
||||
for (const k of Object.keys(v).sort()) {
|
||||
const x = (v as Record<string, unknown>)[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<string, unknown>;
|
||||
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;
|
||||
}
|
||||
Reference in New Issue
Block a user