import { sql } from "kysely";
import type { Db } from "@sttracker/db/db";
import { newId, nowIso } from "@sttracker/db/ids";
import { fromJson, toJson } from "@sttracker/db/json";
import type { ChangeStatus, ChangeType, Database } from "@sttracker/db/schema";
import type { EngagementFields, EntryFields, PartialDate, SourceRecord, WorkHint } from "@sttracker/plugin-sdk/types";
import { addEngagement, type EngagementPatch, updateEngagement } from "../engagements.ts";
import { type EntryPatch, fromStoredDate, resolveVisibleWork, upsertEntry } from "../entries.ts";
import type { Actor, Executor } from "../history.ts";
import { createList, replaceListItems, updateList } from "../lists.ts";
import { matchWork } from "../matching.ts";
import { resolveAliasIn, updateWorkFromHintIn } from "../works.ts";
import { decodeVersion, encodeVersion, hashRecord, type NormalizedRecord, normalizeRecord, sameValue } from "./snapshot.ts";

export interface RunCounts {
  seen: number;
  added: number;
  changed: number;
  removed: number;
  merged: number;
  conflicts: number;
  autoApplied: number;
  pending: number;
  listens: number;
}

export function emptyCounts(): RunCounts {
  return { seen: 0, added: 0, changed: 0, removed: 0, merged: 0, conflicts: 0, autoApplied: 0, pending: 0, listens: 0 };
}

/** Everything one run needs while applying records. */
export interface SyncState {
  db: Db;
  userId: string;
  connectionId: string;
  pluginId: string;
  runId: string;
  actor: Actor;
  counts: RunCounts;
  log: string[];
}

/**
 * Run `fn` against scratch counts and log, and fold them into the run's only when `fn` resolves, so
 * the run summary reflects committed work only.
 */
export async function committed(st: SyncState, fn: (local: SyncState) => Promise<void>): Promise<void> {
  const local: SyncState = { ...st, counts: emptyCounts(), log: [] };
  await fn(local);
  for (const key of Object.keys(local.counts) as (keyof RunCounts)[]) st.counts[key] += local.counts[key];
  st.log.push(...local.log);
}

type SourceRecordRow = Database["source_records"];
type Fields = Record<string, unknown>;
type ListItemValue = { workId: string; note: string | null };

/**
 * `source_records.current_hash` of a record that has been matched (its work id stored as
 * `target_entity "work"`) but whose apply has not committed yet. A retry reuses the stored work.
 */
const PENDING_HASH = "";

// ---------------------------------------------------------------------------------------------
// Field mapping between SourceRecord fields and the user layer's patch keys
// ---------------------------------------------------------------------------------------------

/** 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;
}

/** A list record's own fields (items are matched separately). */
function listPatchFrom(l: Record<string, unknown> | undefined): Fields {
  const out: Fields = {};
  if (typeof l?.name === "string") out.name = l.name;
  if (typeof l?.description === "string") out.description = l.description;
  return out;
}

/** An entry row's current values under the sync-managed patch keys. */
export function entryValues(r: Database["entries"]): Fields {
  return {
    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,
  };
}

/** A list's current values: name, description and ordered items. */
export async function listValues(k: Executor, r: Database["lists"]): Promise<Fields> {
  const items = await k.selectFrom("list_items").select(["work_id", "note"]).where("list_id", "=", r.id).orderBy("position").orderBy("id").execute();
  return { name: r.name, description: r.description, items: items.map((i): ListItemValue => ({ workId: i.work_id, note: i.note })) };
}

/**
 * Lists carry no `field_sources`, so ownership is read from their history: `name`/`description` belong
 * to whoever last changed them, and `items` to the user once any user edit exists, else to the source.
 */
export async function listFieldSources(k: Executor, listId: string, sourceActor: Actor): Promise<Record<string, string>> {
  const rows = await k
    .selectFrom("user_history")
    .select(["after", "actor"])
    .where("entity", "=", "list")
    .where("entity_id", "=", listId)
    .orderBy("created_at")
    .orderBy("id")
    .execute();
  const fs: Record<string, string> = { items: sourceActor };
  for (const r of rows) {
    const after = fromJson<Fields>(r.after, {});
    for (const f of ["name", "description"]) if (f in after) fs[f] = r.actor;
    if (r.actor === "user") fs.items = "user";
  }
  return fs;
}

function patchOf(n: NormalizedRecord | undefined): Fields {
  if (!n) return {};
  if (n.kind === "entry") return entryPatchFrom(n.entry);
  if (n.kind === "engagement") return engagementPatchFrom(n.engagement);
  if (n.kind === "list") return listPatchFrom(n.list);
  return {};
}

// ---------------------------------------------------------------------------------------------
// Change events
// ---------------------------------------------------------------------------------------------

async function writeEvent(
  k: Executor,
  st: SyncState,
  e: { id?: string; recordId: string; type: ChangeType; status: ChangeStatus; targetEntity: string; targetId: string | null; payload: Fields },
): Promise<void> {
  await k
    .insertInto("change_events")
    .values({
      id: e.id ?? newId(),
      user_id: st.userId,
      connection_id: st.connectionId,
      run_id: st.runId,
      record_id: e.recordId,
      type: e.type,
      target_entity: e.targetEntity,
      target_id: e.targetId,
      payload: toJson(e.payload),
      status: e.status,
      created_at: nowIso(),
      resolved_at: null,
    })
    .execute();
  if (e.status === "auto_applied") st.counts.autoApplied++;
  if (e.status === "pending") st.counts.pending++;
  if (e.status === "pending" && e.type === "conflict") st.counts.conflicts++;
}

const incomingOf = (p: Fields) => ("incoming" in p ? p.incoming : p.to);

/** Pending changed/conflict events of a record for one field. */
async function pendingFieldEvents(k: Executor, recordId: string, field: string) {
  const rows = await k
    .selectFrom("change_events")
    .select(["id", "payload"])
    .where("record_id", "=", recordId)
    .where("status", "=", "pending")
    .where("type", "in", ["changed", "conflict"])
    .execute();
  return rows.map((r) => ({ id: r.id, payload: fromJson<Fields>(r.payload, {}) })).filter((r) => r.payload.field === field);
}

/** Resolve stale pending events as `ignored`, recording why in their payload. */
async function retire(k: Executor, stale: { id: string; payload: Fields }[], extra: Fields): Promise<void> {
  for (const s of stale) {
    await k
      .updateTable("change_events")
      .set({ status: "ignored", resolved_at: nowIso(), payload: toJson({ ...s.payload, ...extra }) })
      .where("id", "=", s.id)
      .execute();
  }
}

interface Target {
  entity: "entry" | "engagement" | "list";
  id: string;
  current: Fields;
  fieldSources: Record<string, string>;
}

/**
 * Decide each incoming field against the target's field ownership (spec §8.3):
 * unset or this connection → applied; `"user"` → pending `changed`; another source → pending `conflict`.
 * Values equal to the current one are skipped, so the plugin never takes ownership of a field it did not change.
 * Anti-flip-flop: a pending event for the same (record, field, incoming value) is never filed twice, and a
 * newer incoming value supersedes an older pending one (`ignored`, `payload.supersededBy`).
 * Returns the patch that may be applied, with each applied field's previous value.
 */
async function decideFields(
  k: Executor,
  st: SyncState,
  recordId: string,
  t: Target,
  incoming: Fields,
): Promise<{ patch: Fields; from: Fields }> {
  const patch: Fields = {};
  const from: Fields = {};
  for (const [field, value] of Object.entries(incoming)) {
    const current = t.current[field];
    const pending = await pendingFieldEvents(k, recordId, field);
    if (sameValue(current, value)) {
      // The source now agrees with what STTracker holds: its older proposals are moot.
      await retire(k, pending, { reason: "source_agrees" });
      continue;
    }
    const source = t.fieldSources[field];
    if (source === undefined || source === st.actor) {
      patch[field] = value;
      from[field] = current ?? null;
      continue;
    }
    if (pending.some((p) => sameValue(incomingOf(p.payload), value))) continue;
    const id = newId();
    await retire(k, pending, { supersededBy: id });
    if (source === "user") {
      await writeEvent(k, st, {
        id,
        recordId,
        type: "changed",
        status: "pending",
        targetEntity: t.entity,
        targetId: t.id,
        payload: { field, from: current ?? null, to: value },
      });
    } else {
      await writeEvent(k, st, {
        id,
        recordId,
        type: "conflict",
        status: "pending",
        targetEntity: t.entity,
        targetId: t.id,
        payload: { field, current: current ?? null, currentSource: source, incoming: value, incomingSource: st.actor },
      });
    }
  }
  return { patch, from };
}

/** Write an allowed patch to the target as this connection's actor. */
export async function applyPatch(k: Executor, db: Db, userId: string, t: Pick<Target, "entity" | "id">, patch: Fields, actor: Actor): Promise<void> {
  if (Object.keys(patch).length === 0) return;
  if (t.entity === "entry") {
    const row = await k.selectFrom("entries").select("work_id").where("id", "=", t.id).executeTakeFirstOrThrow();
    await upsertEntry(db, userId, row.work_id, patch as Partial<EntryPatch>, actor, k);
  } else if (t.entity === "engagement") {
    await updateEngagement(db, userId, t.id, patch as Partial<EngagementPatch>, actor, k);
  } else {
    const { items, ...own } = patch;
    if (Object.keys(own).length > 0) await updateList(db, userId, t.id, own, actor, k);
    if (items !== undefined) await replaceListItems(db, userId, t.id, items as ListItemValue[], actor, k);
  }
}

/** The live target of a mapped record, or null (unmapped, gone, or soft-deleted: never revived). */
async function loadTarget(k: Executor, st: SyncState, entity: string | null, id: string | null): Promise<Target | null> {
  if (!id) return null;
  if (entity === "entry") {
    const r = await k
      .selectFrom("entries")
      .selectAll()
      .where("id", "=", id)
      .where("user_id", "=", st.userId)
      .where("deleted_at", "is", null)
      .executeTakeFirst();
    return r ? { entity, id, current: entryValues(r), fieldSources: fromJson(r.field_sources, {}) } : null;
  }
  if (entity === "engagement") {
    const r = await k
      .selectFrom("engagements as g")
      .innerJoin("entries as e", "e.id", "g.entry_id")
      .selectAll("g")
      .where("g.id", "=", id)
      .where("e.user_id", "=", st.userId)
      .where("g.deleted_at", "is", null)
      .where("e.deleted_at", "is", null)
      .executeTakeFirst();
    return r ? { entity, id, current: engagementValues(r), fieldSources: fromJson(r.field_sources, {}) } : null;
  }
  if (entity === "list") {
    const r = await k
      .selectFrom("lists")
      .selectAll()
      .where("id", "=", id)
      .where("user_id", "=", st.userId)
      .where("deleted_at", "is", null)
      .executeTakeFirst();
    return r ? { entity, id, current: await listValues(k, r), fieldSources: await listFieldSources(k, id, st.actor) } : null;
  }
  return null;
}

// ---------------------------------------------------------------------------------------------
// Entry creation helpers
// ---------------------------------------------------------------------------------------------

async function findEntryRow(k: Executor, userId: string, workId: string) {
  return k.selectFrom("entries").selectAll().where("user_id", "=", userId).where("work_id", "=", workId).executeTakeFirst();
}

/**
 * Hierarchy ruling: when the engine creates an entry on a child work, the user also gets an entry on
 * the root ancestor (status "none", plugin actor) unless one exists. An existing root entry, live or
 * deleted, is never touched.
 */
async function ensureRootEntry(k: Executor, st: SyncState, workId: string): Promise<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.
 * Also returns the date field that matched (its values differ only within the dedupe tolerance).
 */
async function findDuplicateEngagement(k: Executor, st: SyncState, entryId: string, incoming: Fields) {
  const fin = incoming.finishedAt as PartialDate | null | undefined;
  const start = incoming.startedAt as PartialDate | null | undefined;
  if (!fin && !start) return null;
  const rows = await k
    .selectFrom("engagements")
    .selectAll()
    .where("entry_id", "=", entryId)
    .where("deleted_at", "is", null)
    .where((eb) =>
      eb.not(
        eb.exists(
          eb
            .selectFrom("engagement_sources as s")
            .select("s.id")
            .whereRef("s.engagement_id", "=", "engagements.id")
            .where("s.connection_id", "=", st.connectionId),
        ),
      ),
    )
    .orderBy("id")
    .execute();
  for (const r of rows) {
    const v = engagementValues(r);
    const f = v.finishedAt as PartialDate | null;
    const s = v.startedAt as PartialDate | null;
    if (fin && f && finishedWithin24h(fin, f)) return { row: r, matchedOn: "finishedAt" };
    if (start && s && sameStartInstant(start, s)) return { row: r, matchedOn: "startedAt" };
  }
  return null;
}

const isEmptyValue = (v: unknown) =>
  v === null || v === undefined || (typeof v === "object" && !Array.isArray(v) && Object.keys(v as object).length === 0);

// ---------------------------------------------------------------------------------------------
// List items
// ---------------------------------------------------------------------------------------------

/**
 * Match list item hints to works, outside any transaction. An ID-less hint first reuses the user's own
 * live private work with the same media type and title, so re-matching never piles up drafts.
 */
async function matchListItems(st: SyncState, items: { work: WorkHint; note?: string }[]): Promise<ListItemValue[]> {
  const out: ListItemValue[] = [];
  const seen = new Set<string>();
  for (const it of items) {
    let workId: string | undefined;
    if (Object.keys(it.work.externalIds).length === 0) {
      workId = (
        await st.db.k
          .selectFrom("works")
          .select("id")
          .where("owner_user_id", "=", st.userId)
          .where("merged_into_id", "is", null)
          .where("media_type", "=", it.work.mediaType)
          .where(sql<boolean>`lower(title) = ${it.work.title.toLowerCase()}`)
          .orderBy("id")
          .executeTakeFirst()
      )?.id;
    }
    workId ??= (await matchWork(st.db, st.userId, it.work, st.actor)).workId;
    if (seen.has(workId)) continue;
    seen.add(workId);
    out.push({ workId, note: it.note ?? null });
  }
  return out;
}

// ---------------------------------------------------------------------------------------------
// Records
// ---------------------------------------------------------------------------------------------

/**
 * Snapshot one non-listen record and apply it. Call outside any transaction (matchWork has its own).
 * Counts land in `st` only once the record's work has committed.
 */
export function applyRecord(st: SyncState, rec: SourceRecord): Promise<void> {
  return committed(st, async (local) => {
    const hash = hashRecord(rec);
    const existing = await local.db.k
      .selectFrom("source_records")
      .selectAll()
      .where("connection_id", "=", local.connectionId)
      .where("key", "=", rec.key)
      .executeTakeFirst();
    if (!existing || existing.current_hash === PENDING_HASH) return addRecord(local, rec, hash, existing ?? null);
    if (existing.current_hash === hash) {
      await local.db.k.transaction().execute(async (tx) => {
        await tx.updateTable("source_records").set({ last_seen: nowIso() }).where("id", "=", existing.id).execute();
        if (existing.deleted_at !== null && rec.deleted !== true) await reappeared(tx, existing);
      });
      return;
    }
    return changeRecord(local, existing, rec, hash);
  });
}

async function addRecord(st: SyncState, rec: SourceRecord, hash: string, pendingRow: SourceRecordRow | null): Promise<void> {
  const applicable = rec.deleted !== true && !rec.mergedIntoKey && rec.kind !== "listen";
  let recordId = pendingRow?.id ?? null;
  let workId: string | null = null;
  let items: ListItemValue[] = [];
  if (applicable && (rec.kind === "entry" || rec.kind === "engagement")) {
    if (pendingRow?.target_entity === "work" && pendingRow.target_id) {
      workId = pendingRow.target_id;
    } else {
      // Matching runs in its own transaction (it retries on unique violations). Its result is committed
      // right away, so a failed apply is retried against the same work instead of a fresh draft.
      workId = (await matchWork(st.db, st.userId, rec.work, st.actor)).workId;
      recordId = await persistMatched(st, rec, workId, recordId);
    }
  } else if (applicable && rec.kind === "list") {
    items = await matchListItems(st, rec.list.items);
  }

  await st.db.k.transaction().execute(async (tx) => {
    const now = nowIso();
    const row = {
      current_hash: hash,
      last_seen: now,
      deleted_at: rec.deleted === true ? now : null,
      merged_into_key: rec.mergedIntoKey ?? null,
      target_entity: rec.kind === "listen" ? null : rec.kind,
      target_id: null,
    };
    if (recordId) {
      await tx.updateTable("source_records").set(row).where("id", "=", recordId).execute();
    } else {
      recordId = newId();
      await tx
        .insertInto("source_records")
        .values({ id: recordId, connection_id: st.connectionId, key: rec.key, kind: rec.kind, first_seen: now, ...row })
        .execute();
    }
    const id = recordId;
    await tx.insertInto("source_record_versions").values({ id: newId(), record_id: id, hash, raw_gz: encodeVersion(rec), seen_at: now }).execute();
    if (!applicable) return;
    let targetId: string | null = null;
    if (rec.kind === "entry" && workId) targetId = await addEntryRecord(tx, st, id, rec.key, workId, entryPatchFrom(rec.entry));
    else if (rec.kind === "engagement" && workId) targetId = await addEngagementRecord(tx, st, id, rec.key, workId, engagementPatchFrom(rec.engagement));
    else if (rec.kind === "list") targetId = await addListRecord(tx, st, id, rec.key, rec.list.name, items);
    if (targetId) {
      await tx.updateTable("source_records").set({ target_id: targetId }).where("id", "=", id).execute();
      st.counts.added++;
    }
  });
}

/** Commit the record row with its matched work (`target_entity "work"`, pending hash) before applying. */
async function persistMatched(st: SyncState, rec: SourceRecord, workId: string, recordId: string | null): Promise<string> {
  const now = nowIso();
  if (recordId) {
    await st.db.k.updateTable("source_records").set({ target_entity: "work", target_id: workId, last_seen: now }).where("id", "=", recordId).execute();
    return recordId;
  }
  const id = newId();
  await st.db.k
    .insertInto("source_records")
    .values({
      id,
      connection_id: st.connectionId,
      key: rec.key,
      kind: rec.kind,
      current_hash: PENDING_HASH,
      first_seen: now,
      last_seen: now,
      deleted_at: null,
      merged_into_key: null,
      target_entity: "work",
      target_id: workId,
    })
    .execute();
  return id;
}

async function addEntryRecord(k: Executor, st: SyncState, recordId: string, key: string, workId: string, incoming: Fields): Promise<string | null> {
  const live = await liveEntryFor(k, st, workId, incoming);
  if (!live) {
    st.log.push(`record ${key}: the user deleted this entry; not revived`);
    return null;
  }
  let applied = Object.keys(incoming);
  if (!live.created) {
    // The user (or another source) already has this entry: decide field by field.
    const t: Target = { entity: "entry", id: live.row.id, current: entryValues(live.row), fieldSources: fromJson(live.row.field_sources, {}) };
    const { patch } = await decideFields(k, st, recordId, t, incoming);
    await applyPatch(k, st.db, st.userId, t, patch, st.actor);
    applied = Object.keys(patch);
  }
  await writeEvent(k, st, {
    recordId,
    type: "added",
    status: "auto_applied",
    targetEntity: "entry",
    targetId: live.row.id,
    payload: { key, kind: "entry", fields: applied },
  });
  return live.row.id;
}

async function addEngagementRecord(k: Executor, st: SyncState, recordId: string, key: string, workId: string, incoming: Fields): Promise<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 payload: Fields;
  if (dup) {
    // Same play from another source: attach. Fields nobody has set yet are filled; any other differing
    // field goes through the per-field rules (conflict / pending changed). The date that identified the
    // play differs only within the dedupe tolerance and is left alone.
    const t: Target = {
      entity: "engagement",
      id: dup.row.id,
      current: engagementValues(dup.row),
      fieldSources: fromJson(dup.row.field_sources, {}),
    };
    const fill: Fields = {};
    const rest: Fields = {};
    for (const [f, v] of Object.entries(incoming)) {
      if (f === dup.matchedOn) continue;
      if (t.fieldSources[f] === undefined && isEmptyValue(t.current[f]) && !isEmptyValue(v)) fill[f] = v;
      else rest[f] = v;
    }
    const { patch } = await decideFields(k, st, recordId, t, rest);
    await applyPatch(k, st.db, st.userId, t, { ...fill, ...patch }, st.actor);
    engagementId = dup.row.id;
    payload = { key, kind: "engagement", fields: Object.keys({ ...fill, ...patch }), dedupedInto: dup.row.id };
  } else {
    const eg = await addEngagement(st.db, st.userId, live.row.id, incoming as Partial<EngagementPatch>, st.actor, k);
    engagementId = eg.id;
    payload = { key, kind: "engagement", fields: Object.keys(incoming) };
  }
  await k
    .insertInto("engagement_sources")
    .values({ id: newId(), engagement_id: engagementId, connection_id: st.connectionId, record_key: key })
    .onConflict((oc) => oc.columns(["connection_id", "record_key"]).doNothing())
    .execute();
  await writeEvent(k, st, { recordId, type: "added", status: "auto_applied", targetEntity: "engagement", targetId: engagementId, payload });
  return engagementId;
}

async function addListRecord(k: Executor, st: SyncState, recordId: string, key: string, name: string, items: ListItemValue[]): Promise<string> {
  const list = await createList(st.db, st.userId, { name, description: `Imported from ${st.pluginId}`, kind: "list" }, st.actor, k);
  if (items.length > 0) await replaceListItems(st.db, st.userId, list.id, items, st.actor, k);
  await writeEvent(k, st, {
    recordId,
    type: "added",
    status: "auto_applied",
    targetEntity: "list",
    targetId: list.id,
    payload: { key, kind: "list", fields: ["name", "description", "items"] },
  });
  return list.id;
}

async function changeRecord(st: SyncState, existing: SourceRecordRow, rec: SourceRecord, hash: string): Promise<void> {
  const prevRow = await st.db.k
    .selectFrom("source_record_versions")
    .select("raw_gz")
    .where("record_id", "=", existing.id)
    .where("hash", "=", existing.current_hash)
    .orderBy("seen_at", "desc")
    .orderBy("id", "desc")
    .executeTakeFirst();
  const prev = prevRow ? decodeVersion(prevRow.raw_gz).normalized : undefined;
  const next = normalizeRecord(rec);
  // List items are matched before the transaction (matchWork has its own), and only when they changed.
  const itemsChanged = rec.kind === "list" && !rec.deleted && !sameValue(prev?.list?.items, next.list?.items);
  const items = itemsChanged && rec.kind === "list" ? await matchListItems(st, rec.list.items) : null;

  await st.db.k.transaction().execute(async (tx) => {
    const now = nowIso();
    // A merged record that is later reported deleted stays "merged": its disappearance is expected.
    const mergedInto = rec.mergedIntoKey ?? (rec.deleted === true ? existing.merged_into_key : null);
    await tx.insertInto("source_record_versions").values({ id: newId(), record_id: existing.id, hash, raw_gz: encodeVersion(rec), seen_at: now }).execute();
    await tx
      .updateTable("source_records")
      .set({ current_hash: hash, last_seen: now, merged_into_key: mergedInto })
      .where("id", "=", existing.id)
      .execute();

    if (rec.deleted === true) {
      if (existing.deleted_at === null) await markRemoved(tx, st, { ...existing, merged_into_key: mergedInto });
      return;
    }
    st.counts.changed++;
    if (existing.deleted_at !== null) await reappeared(tx, existing);

    if (rec.mergedIntoKey && rec.mergedIntoKey !== (prev?.mergedIntoKey ?? null)) await fileMerged(tx, st, existing, rec.mergedIntoKey);

    const target = await loadTarget(tx, st, existing.target_entity, existing.target_id);
    if (!target) return; // unmapped, or the user deleted it: never revived
    const before = patchOf(prev);
    const after = patchOf(next);
    const incoming: Fields = {};
    for (const [f, v] of Object.entries(after)) if (!sameValue(before[f], v)) incoming[f] = v;
    if (items) incoming.items = items;
    const { patch, from } = await decideFields(tx, st, existing.id, target, incoming);
    await applyPatch(tx, st.db, st.userId, target, patch, st.actor);
    for (const [field, to] of Object.entries(patch)) {
      await writeEvent(tx, st, {
        recordId: existing.id,
        type: "changed",
        status: "auto_applied",
        targetEntity: target.entity,
        targetId: target.id,
        payload: { field, from: from[field], to },
      });
    }
    // The source's description of the work changed: refresh the (shared) work it maps to; never re-match.
    if ((rec.kind === "entry" || rec.kind === "engagement") && prev?.work !== undefined && !sameValue(prev.work, next.work)) {
      const workId = await targetWorkId(tx, target);
      if (workId) await updateWorkFromHintIn(tx, workId, rec.work, st.actor);
    }
  });
}

async function targetWorkId(k: Executor, t: Target): Promise<string | null> {
  if (t.entity === "list") return null;
  const entryId =
    t.entity === "entry"
      ? t.id
      : (await k.selectFrom("engagements").select("entry_id").where("id", "=", t.id).executeTakeFirst())?.entry_id;
  if (!entryId) return null;
  return (await k.selectFrom("entries").select("work_id").where("id", "=", entryId).executeTakeFirst())?.work_id ?? null;
}

/** The record is back at the source: clear `deleted_at` and retire still-pending removal events for it. */
async function reappeared(k: Executor, rec: SourceRecordRow): Promise<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.
 * A record merged into another (its `merged_into_key` set, or a pending `merged` event) is expected
 * to vanish: no removal is counted or filed for it, ever.
 */
export async function markRemoved(k: Executor, st: SyncState, rec: SourceRecordRow): Promise<void> {
  await k.updateTable("source_records").set({ deleted_at: nowIso() }).where("id", "=", rec.id).execute();
  if (rec.merged_into_key) return;
  const pendingMerge = await k
    .selectFrom("change_events")
    .select("id")
    .where("record_id", "=", rec.id)
    .where("type", "=", "merged")
    .where("status", "=", "pending")
    .executeTakeFirst();
  if (pendingMerge) return;
  st.counts.removed++;
  const target = await loadTarget(k, st, 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, rec.target_entity, rec.target_id);
  if (!target || target.entity === "list") return;
  const entryId =
    target.entity === "entry"
      ? target.id
      : (await k.selectFrom("engagements").select("entry_id").where("id", "=", target.id).executeTakeFirstOrThrow()).entry_id;
  await writeEvent(k, st, {
    recordId: rec.id,
    type: "merged",
    status: "pending",
    targetEntity: "entry",
    targetId: entryId,
    payload: { key: rec.key, intoKey },
  });
}