Hankan/app/src/sync/client.ts

/* The sync loop, protocol 2: pull everything, then push what changed here.

   Offline-first is not negotiable. Sync runs only when a server is
   configured, and every failure is non-fatal — the app must stay completely
   usable with the Pi unreachable. Nothing in here may throw into the UI.

   The order is the first gate. A device may not push until it has pulled
   every page the server holds for this epoch: the week-old laptop that
   pushed first is exactly how the artifact lost a week of the phone's work.
   See shared/sync-protocol.mjs for the other two gates. */

import type { Db } from "../db/types.js";
import { deriveCustomLemma } from "../db/writes.js";
import {
  PAGE_SIZE,
  PROTOCOL,
  PROTOCOL_HEADER,
  RESET_SCOPES,
  SYNC_TABLES,
  SYNC_TABLE_NAMES,
  decodePk,
  encodePk,
  isSyncableMetaKey,
  resetCovers,
  type PullResponse,
  type PushResponse,
  type ServerRow,
  type WireRow,
} from "@shared/sync-protocol.mjs";
import { lemmaId } from "@shared/lemma-id.mjs";
import { resolve, type Data, type ResolveContext } from "./resolve.js";

export interface SyncConfig {
  baseUrl: string;
  token: string;
}

export interface SyncResult {
  pushed: number;
  pulled: number;
  conflicts: number;
  cursor: number;
  /** The server was new to this device, or had been reset: everything was
      pulled from the start and every stamped row offered back. */
  rehydrated: boolean;
}

const EPOCH = "sync.epoch";
const CURSOR = "sync.cursor";
const HYDRATED = "sync.hydrated";

/** Rounds of push-and-settle before giving up until the next sync. */
const MAX_PUSH_ROUNDS = 5;

/* ── local bookkeeping ────────────────────────────────────────────────
   Device state, not user data: written unstamped and never dirty, and
   `sync.*` is outside the syncable-meta allowlist besides. */

async function readMeta(db: Db, k: string): Promise<string | undefined> {
  return (await db.get<{ v: string }>("SELECT v FROM meta WHERE k = ?", [k]))?.v;
}

async function writeMeta(db: Db, k: string, v: string | number): Promise<void> {
  await db.run("INSERT OR REPLACE INTO meta (k, v, updated_at) VALUES (?, ?, 0)", [k, String(v)]);
}

/** Forget this device's place with the server, so the next sync starts over. */
export async function forgetServerPosition(db: Db): Promise<void> {
  await db.run("DELETE FROM meta WHERE k IN (?, ?, ?)", [EPOCH, CURSOR, HYDRATED]);
}

/** Has this device pulled everything its server holds? */
export async function isHydrated(db: Db): Promise<boolean> {
  return (await readMeta(db, HYDRATED)) === "1";
}

/* ── rows ─────────────────────────────────────────────────────────────── */

type LocalRow = Data & { base_seq: number; dirty: number; rev: number; updated_at: number };

interface Local {
  live: LocalRow | undefined;
  tomb: { base_seq: number; dirty: number; rev: number } | undefined;
}

const whereOf = (tbl: string) => SYNC_TABLES[tbl]!.pk.map((c) => `${c} = ?`).join(" AND ");

async function readLocal(db: Db, tbl: string, pk: string): Promise<Local> {
  const live = await db.get<LocalRow>(`SELECT * FROM ${tbl} WHERE ${whereOf(tbl)}`, decodePk(pk));
  const tomb = await db.get<{ base_seq: number; dirty: number; rev: number }>(
    "SELECT base_seq, dirty, rev FROM tombstone WHERE tbl = ? AND pk = ?",
    [tbl, pk],
  );
  return { live, tomb };
}

const pick = (tbl: string, row: Data): Data =>
  Object.fromEntries(SYNC_TABLES[tbl]!.cols.map((c) => [c, row[c] ?? null]));

/** Write a row as the server holds it: clean, and agreed at `seq`. */
async function adopt(db: Db, row: ServerRow): Promise<void> {
  const spec = SYNC_TABLES[row.tbl]!;
  const key = decodePk(row.pk);

  if (row.deleted) {
    const before = await db.get<Data>(`SELECT * FROM ${row.tbl} WHERE ${whereOf(row.tbl)}`, key);
    await db.run(`DELETE FROM ${row.tbl} WHERE ${whereOf(row.tbl)}`, key);
    // A clean tombstone remembers the server's seq for this key, so a later
    // re-creation here pushes against the right base.
    await db.run(
      `INSERT INTO tombstone (tbl, pk, updated_at, base_seq, dirty, rev) VALUES (?, ?, ?, ?, 0, 0)
       ON CONFLICT(tbl, pk) DO UPDATE SET base_seq = excluded.base_seq, dirty = 0`,
      [row.tbl, row.pk, row.updated_at ?? 0, row.seq],
    );
    if (row.tbl === "custom_word" && before) {
      const id = lemmaId(String(before.headword), String(before.pos));
      await db.run("DELETE FROM surface WHERE lemma_id = ? AND lemma_id IN (SELECT id FROM lemma WHERE source = 'custom')", [id]);
      await db.run("DELETE FROM lemma WHERE id = ? AND source = 'custom'", [id]);
    }
    return;
  }

  const data = pick(row.tbl, row.data ?? {});
  const cols = spec.cols;
  await db.run(
    `INSERT INTO ${row.tbl} (${cols.join(", ")}, updated_at, base_seq, dirty)
     VALUES (${cols.map(() => "?").join(", ")}, ?, ?, 0)
     ON CONFLICT(${spec.pk.join(", ")}) DO UPDATE SET
       ${cols.map((c) => `${c} = excluded.${c}`).join(", ")},
       updated_at = excluded.updated_at, base_seq = excluded.base_seq, dirty = 0`,
    [...cols.map((c) => data[c] ?? null), row.updated_at ?? 0, row.seq],
  );
  await db.run("DELETE FROM tombstone WHERE tbl = ? AND pk = ?", [row.tbl, row.pk]);

  if (row.tbl === "custom_word") {
    await deriveCustomLemma(db, {
      headword: String(data.headword),
      pos: String(data.pos),
      gloss: String(data.gloss ?? ""),
    });
  }
  if (row.tbl === "meta" && String(data.k).startsWith("reset.")) {
    await obeyMarker(db, String(data.k), String(data.v));
  }
}

/** Ours stands: agree with the server's seq, so the next push overwrites it. */
async function keep(db: Db, tbl: string, pk: string, local: Local, seq: number): Promise<void> {
  if (local.live) {
    await db.run(`UPDATE ${tbl} SET base_seq = ?, dirty = 1 WHERE ${whereOf(tbl)}`, [seq, ...decodePk(pk)]);
  } else {
    await db.run("UPDATE tombstone SET base_seq = ?, dirty = 1 WHERE tbl = ? AND pk = ?", [seq, tbl, pk]);
  }
}

/** A combination of both: written here, agreed at `seq`, and pushed. */
async function merge(db: Db, tbl: string, pk: string, seq: number, data: Data): Promise<void> {
  const cols = SYNC_TABLES[tbl]!.cols;
  await db.run(
    `UPDATE ${tbl} SET ${cols.map((c) => `${c} = ?`).join(", ")},
       base_seq = ?, dirty = 1, rev = rev + 1 WHERE ${whereOf(tbl)}`,
    [...cols.map((c) => data[c] ?? null), seq, ...decodePk(pk)],
  );
}

/** Settle one disagreement between this device and the server. */
async function settle(db: Db, remote: ServerRow, local: Local, ctx: ResolveContext): Promise<void> {
  // The server has no row at all: nothing to disagree with.
  if (remote.seq === 0) {
    if (local.live) await keep(db, remote.tbl, remote.pk, local, 0);
    else await db.run("DELETE FROM tombstone WHERE tbl = ? AND pk = ?", [remote.tbl, remote.pk]);
    return;
  }

  const localSide = local.live
    ? { deleted: false, data: local.live as Data }
    : { deleted: true, data: null };
  const outcome = resolve(remote.tbl, localSide, { deleted: remote.deleted, data: remote.data }, ctx);
  if (outcome.kind === "adopt") await adopt(db, remote);
  else if (outcome.kind === "keep") await keep(db, remote.tbl, remote.pk, local, remote.seq);
  else await merge(db, remote.tbl, remote.pk, remote.seq, outcome.data);
}

/* ── deliberate shrinks ───────────────────────────────────────────────── */

/**
 * Empty what a reset marker covers, once per rise of its counter.
 *
 * Unsynced edits here go too: a reset this device had not heard of still
 * wins over work done on the data it reset. Rows written after the reset
 * arrive later in the pull — higher change_seq — and survive it.
 */
async function obeyMarker(db: Db, key: string, value: string): Promise<void> {
  const scope = key.slice("reset.".length) as keyof typeof RESET_SCOPES;
  const spec = RESET_SCOPES[scope];
  if (!spec) return;
  const seen = Number((await readMeta(db, `sync.applied.${key}`)) ?? 0);
  const n = Number(value);
  if (!(n > seen)) return;

  for (const tbl of spec.tables) {
    await db.run(`DELETE FROM ${tbl}`);
    await db.run("DELETE FROM tombstone WHERE tbl = ?", [tbl]);
  }
  for (const prefix of spec.meta) {
    await db.run("DELETE FROM meta WHERE k LIKE ?", [`${prefix}%`]);
  }
  if (spec.tables.includes("custom_word")) {
    await db.run("DELETE FROM surface WHERE lemma_id IN (SELECT id FROM lemma WHERE source = 'custom')");
    await db.run("DELETE FROM lemma WHERE source = 'custom'");
  }
  await writeMeta(db, `sync.applied.${key}`, n);
}

/** Resets raised here that the server has not heard of yet. */
async function pendingResets(db: Db): Promise<string[]> {
  const rows = await db.all<{ k: string }>("SELECT k FROM meta WHERE k LIKE 'reset.%' AND dirty = 1");
  return rows.map((r) => r.k.slice("reset.".length));
}

/* ── pull ─────────────────────────────────────────────────────────────── */

async function applyIncoming(
  db: Db,
  row: ServerRow,
  pending: string[],
  ctx: ResolveContext,
): Promise<"applied" | "settled" | "skipped"> {
  const spec = SYNC_TABLES[row.tbl];
  if (!spec) return "skipped"; // a newer server; ignore rather than fail
  const key = decodePk(row.pk);
  const probe: Data = row.data ?? Object.fromEntries(spec.pk.map((c, i) => [c, key[i] ?? null]));
  if (row.tbl === "meta" && !isSyncableMetaKey(String(probe.k ?? key[0]))) return "skipped";

  // A reset raised here, not yet pushed, covers this row: the server's copy
  // predates it and is about to be emptied everywhere anyway.
  if (pending.some((scope) => resetCovers(scope, row.tbl, probe))) return "skipped";

  const local = await readLocal(db, row.tbl, row.pk);
  const mine = local.live ?? local.tomb;
  if (mine && mine.base_seq === row.seq) return "skipped"; // our own write, echoed back

  if (!mine || !mine.dirty) {
    await adopt(db, row);
    return "applied";
  }
  await settle(db, row, local, ctx);
  return "settled";
}

/* ── push ─────────────────────────────────────────────────────────────── */

interface Outgoing {
  wire: WireRow;
  rev: number;
  /** A live row re-created over a tombstone this device had applied. */
  overTomb: boolean;
}

const keyOf = (tbl: string, pk: string) => JSON.stringify([tbl, pk]);

/** Up to one page of dirty rows: deletes and reset markers first, so the
    server orders a reset before anything written after it. */
async function collectDirty(db: Db): Promise<Outgoing[]> {
  const out: Outgoing[] = [];

  const graves = await db.all<{ tbl: string; pk: string; base_seq: number; rev: number; updated_at: number }>(
    "SELECT tbl, pk, base_seq, rev, updated_at FROM tombstone WHERE dirty = 1 LIMIT ?",
    [PAGE_SIZE],
  );
  for (const g of graves) {
    if (!SYNC_TABLES[g.tbl]) continue;
    out.push({
      wire: { tbl: g.tbl, pk: g.pk, base: g.base_seq, deleted: true, data: null, updated_at: g.updated_at },
      rev: g.rev,
      overTomb: false,
    });
  }

  const ordered = ["meta", ...SYNC_TABLE_NAMES.filter((t) => t !== "meta")];
  for (const tbl of ordered) {
    if (out.length >= PAGE_SIZE) break;
    const rows = await db.all<LocalRow>(
      `SELECT * FROM ${tbl} WHERE dirty = 1 ${tbl === "meta" ? "ORDER BY k LIKE 'reset.%' DESC" : ""} LIMIT ?`,
      [PAGE_SIZE - out.length],
    );
    for (const row of rows) {
      if (tbl === "meta" && !isSyncableMetaKey(String(row.k))) {
        // Device bookkeeping that was edited: it never travels, so it is
        // never really dirty.
        await db.run("UPDATE meta SET dirty = 0 WHERE k = ?", [String(row.k)]);
        continue;
      }
      const pk = encodePk(tbl, row);
      let base = row.base_seq;
      let overTomb = false;
      if (base === 0) {
        const tomb = await db.get<{ base_seq: number }>(
          "SELECT base_seq FROM tombstone WHERE tbl = ? AND pk = ? AND dirty = 0",
          [tbl, pk],
        );
        if (tomb) {
          base = tomb.base_seq;
          overTomb = true;
        }
      }
      out.push({
        wire: { tbl, pk, base, deleted: false, data: pick(tbl, row), updated_at: row.updated_at },
        rev: row.rev,
        overTomb,
      });
    }
  }
  return out;
}

async function acknowledge(db: Db, sent: Outgoing[], applied: PushResponse["applied"]): Promise<void> {
  const byKey = new Map(sent.map((s) => [keyOf(s.wire.tbl, s.wire.pk), s]));
  for (const a of applied) {
    const s = byKey.get(keyOf(a.tbl, a.pk));
    if (!s) continue;
    if (s.wire.deleted) {
      await db.run(
        `UPDATE tombstone SET base_seq = ?, dirty = CASE WHEN rev = ? THEN 0 ELSE dirty END
          WHERE tbl = ? AND pk = ?`,
        [a.seq, s.rev, a.tbl, a.pk],
      );
    } else {
      // Clean only if nothing changed the row while the push was in flight.
      await db.run(
        `UPDATE ${a.tbl} SET base_seq = ?, dirty = CASE WHEN rev = ? THEN 0 ELSE dirty END
          WHERE ${whereOf(a.tbl)}`,
        [a.seq, s.rev, ...decodePk(a.pk)],
      );
      if (s.overTomb) await db.run("DELETE FROM tombstone WHERE tbl = ? AND pk = ?", [a.tbl, a.pk]);
    }
  }
}

/* ── transport ────────────────────────────────────────────────────────── */

async function call<T>(cfg: SyncConfig, path: string, init: RequestInit, signal?: AbortSignal): Promise<T> {
  const res = await fetch(`${cfg.baseUrl.replace(/\/$/, "")}${path}`, {
    ...init,
    signal,
    headers: {
      "content-type": "application/json",
      authorization: `Bearer ${cfg.token}`,
      [PROTOCOL_HEADER]: String(PROTOCOL),
      ...(init.headers ?? {}),
    },
  });
  if (res.status === 426) throw new Error("the server speaks a different sync protocol — update it");
  if (!res.ok) throw new Error(`sync ${path}: HTTP ${res.status}`);
  return (await res.json()) as T;
}

/* ── the loop ─────────────────────────────────────────────────────────── */

/** Re-agree with a server from nothing: every stamped row is offered again. */
async function rehydrate(db: Db, epoch: string): Promise<void> {
  await db.tx(async (tx) => {
    for (const tbl of [...SYNC_TABLE_NAMES, "tombstone"]) {
      await tx.run(
        `UPDATE ${tbl} SET base_seq = 0, dirty = CASE WHEN updated_at > 0 THEN 1 ELSE 0 END, rev = rev + 1`,
      );
    }
    await writeMeta(tx, EPOCH, epoch);
    await writeMeta(tx, CURSOR, 0);
    await writeMeta(tx, HYDRATED, 0);
  });
}

/**
 * One full exchange: pull every page, then push until nothing is dirty.
 *
 * `ctx` lets a conflict ask the roadmap which of two units is further on.
 */
export async function syncOnce(
  db: Db,
  cfg: SyncConfig,
  ctx: ResolveContext,
  signal?: AbortSignal,
): Promise<SyncResult> {
  let epoch = (await readMeta(db, EPOCH)) ?? "";
  let cursor = Number((await readMeta(db, CURSOR)) ?? 0) || 0;
  let rehydrated = false;
  let pulled = 0;
  let conflicts = 0;

  /* Gate 1: hydrate. Nothing leaves this device until every page is in. */
  for (;;) {
    const page = await call<PullResponse>(cfg, `/api/sync?cursor=${cursor}`, { method: "GET" }, signal);
    if (page.epoch !== epoch) {
      if (rehydrated) throw new Error("the server was reset during this sync");
      await rehydrate(db, page.epoch);
      epoch = page.epoch;
      cursor = 0;
      rehydrated = true;
      continue;
    }
    const pending = await pendingResets(db);
    for (const row of page.rows) {
      const r = await db.tx((tx) => applyIncoming(tx, row, pending, ctx));
      if (r === "applied") pulled++;
      if (r === "settled") conflicts++;
    }
    cursor = page.cursor;
    await writeMeta(db, CURSOR, cursor);
    if (!page.more) break;
  }
  await writeMeta(db, HYDRATED, 1);

  /* Then push. A conflict is settled locally and, if ours stands, pushed
     again against the server's current seq. */
  let pushed = 0;
  for (let round = 0; round < MAX_PUSH_ROUNDS; round++) {
    const batch = await collectDirty(db);
    if (!batch.length) break;
    const res = await call<PushResponse>(
      cfg,
      "/api/sync",
      { method: "POST", body: JSON.stringify({ rows: batch.map((b) => b.wire) }) },
      signal,
    );
    if (res.epoch !== epoch) {
      // Reset between our pull and our push: start over next time.
      await forgetServerPosition(db);
      break;
    }
    await acknowledge(db, batch, res.applied);
    pushed += res.applied.length;
    for (const c of res.conflicts) {
      await db.tx(async (tx) => settle(tx, c, await readLocal(tx, c.tbl, c.pk), ctx));
      conflicts++;
    }
  }

  return { pushed, pulled, conflicts, cursor, rehydrated };
}

/**
 * Sync if configured, and swallow anything that goes wrong.
 *
 * Callers are UI code on a timer; a sync failure is a diagnostic, never an
 * interruption. Returns null when there is nothing configured or the
 * exchange failed.
 */
export async function trySync(
  db: Db,
  cfg: SyncConfig | null,
  ctx: ResolveContext,
  signal?: AbortSignal,
): Promise<SyncResult | null> {
  if (!cfg?.baseUrl || !cfg.token) return null;
  try {
    return await syncOnce(db, cfg, ctx, signal);
  } catch (err) {
    console.warn("[sync] skipped:", err instanceof Error ? err.message : err);
    return null;
  }
}