Hankan/app/src/sync/client.ts

/* The sync loop: push what changed here, pull what changed there, apply.

   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, which is the acceptance criterion the
   whole port was built around. Nothing in here may throw into the UI. */

import type { Db } from "../db/types.js";
import {
  PAGE_SIZE,
  SYNC_TABLES,
  SYNC_TABLE_NAMES,
  incomingWins,
  isSyncableMetaKey,
} from "@shared/sync-protocol.mjs";

/** A row on the wire: its table, its columns, and when it last changed. */
export interface SyncRow {
  tbl: string;
  data: Record<string, string | number | null>;
  updated_at: number;
  /** Present and true when this row is a delete. */
  deleted?: boolean;
}

export interface PushRequest {
  rows: SyncRow[];
}
export interface PushResponse {
  cursor: number;
}
export interface PullResponse {
  rows: SyncRow[];
  cursor: number;
  more: boolean;
}

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

export interface SyncResult {
  pushed: number;
  pulled: number;
  cursor: number;
}

const CURSOR_KEY = "sync.cursor";
const PUSHED_KEY = "sync.pushedAt";

/* ── local bookkeeping ────────────────────────────────────────────────
   These are device state, not user data. They are written with
   updated_at = 0 and excluded from the syncable-meta allowlist, so they
   never travel. */

async function readNum(db: Db, key: string): Promise<number> {
  const row = await db.get<{ v: string }>("SELECT v FROM meta WHERE k = ?", [key]);
  const n = row ? Number(row.v) : 0;
  return Number.isFinite(n) ? n : 0;
}

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

/* ── collect ─────────────────────────────────────────────────────────── */

/** Everything edited on this device since the last successful push. */
export async function collectDirty(db: Db, pushedAt: number): Promise<SyncRow[]> {
  const out: SyncRow[] = [];

  for (const tbl of SYNC_TABLE_NAMES) {
    const spec = SYNC_TABLES[tbl]!;
    const cols = [...spec.cols, "updated_at"].join(", ");
    const rows = await db.all<Record<string, string | number>>(
      `SELECT ${cols} FROM ${tbl} WHERE updated_at > ? ORDER BY updated_at LIMIT ?`,
      [pushedAt, PAGE_SIZE],
    );

    for (const row of rows) {
      // The allowlist is applied at the point of collection, so a
      // device-local meta key cannot leave this machine even by accident.
      if (tbl === "meta" && !isSyncableMetaKey(String(row.k))) continue;
      const { updated_at, ...data } = row;
      out.push({ tbl, data, updated_at: Number(updated_at) });
    }
  }

  const graves = await db.all<{ tbl: string; pk: string; updated_at: number }>(
    "SELECT tbl, pk, updated_at FROM tombstone WHERE updated_at > ? ORDER BY updated_at LIMIT ?",
    [pushedAt, PAGE_SIZE],
  );
  for (const g of graves) {
    out.push({ tbl: g.tbl, data: { pk: g.pk }, updated_at: Number(g.updated_at), deleted: true });
  }

  return out;
}

/* ── apply ───────────────────────────────────────────────────────────── */

/** Apply one incoming row, last-write-wins against what is already here. */
async function applyRow(db: Db, row: SyncRow): Promise<boolean> {
  const spec = SYNC_TABLES[row.tbl];
  if (!spec) return false; // unknown table — a newer server; ignore rather than fail

  if (row.deleted) {
    const pk = String(row.data.pk);
    const where = spec.pk.map((c) => `${c} = ?`).join(" AND ");
    const parts = pk.split(" ");
    const local = await db.get<{ updated_at: number }>(
      `SELECT updated_at FROM ${row.tbl} WHERE ${where}`,
      parts,
    );
    if (local && !incomingWins(row.updated_at, local.updated_at)) return false;

    await db.tx(async (tx) => {
      await tx.run(`DELETE FROM ${row.tbl} WHERE ${where}`, parts);
      await tx.run(
        `INSERT INTO tombstone (tbl, pk, updated_at) VALUES (?, ?, ?)
         ON CONFLICT(tbl, pk) DO UPDATE SET updated_at = excluded.updated_at`,
        [row.tbl, pk, row.updated_at],
      );
    });
    return true;
  }

  if (row.tbl === "meta" && !isSyncableMetaKey(String(row.data.k))) return false;

  const where = spec.pk.map((c) => `${c} = ?`).join(" AND ");
  const keys = spec.pk.map((c) => row.data[c] as string | number);
  const local = await db.get<{ updated_at: number }>(
    `SELECT updated_at FROM ${row.tbl} WHERE ${where}`,
    keys,
  );
  if (local && !incomingWins(row.updated_at, local.updated_at)) return false;

  const cols = [...spec.cols, "updated_at"];
  const holes = cols.map(() => "?").join(", ");
  const values = [...spec.cols.map((c) => row.data[c] ?? null), row.updated_at];
  await db.run(`INSERT OR REPLACE INTO ${row.tbl} (${cols.join(", ")}) VALUES (${holes})`, values);
  return true;
}

/* ── 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}`,
      ...(init.headers ?? {}),
    },
  });
  if (!res.ok) throw new Error(`sync ${path}: HTTP ${res.status}`);
  return (await res.json()) as T;
}

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

/**
 * One full exchange. Push first so local edits are never lost to an
 * incoming row that would have overwritten them; then pull, paging until
 * the server has nothing newer.
 */
export async function syncOnce(
  db: Db,
  cfg: SyncConfig,
  signal?: AbortSignal,
): Promise<SyncResult> {
  // Anything edited after this instant stays dirty, whatever the pull
  // brings back. That is what keeps an edit made *during* a sync from being
  // marked clean by a remote row that happens to carry a later timestamp.
  const startedAt = Date.now();

  const pushedAt = await readNum(db, PUSHED_KEY);
  const dirty = await collectDirty(db, pushedAt);
  let watermark = pushedAt;

  if (dirty.length) {
    const body: PushRequest = { rows: dirty };
    await call<PushResponse>(
      cfg,
      "/api/sync",
      { method: "POST", body: JSON.stringify(body) },
      signal,
    );
    // Only advance once the server has them. A failed push throws before
    // this line and the rows stay dirty, which is the safe direction.
    watermark = dirty.reduce((m, r) => Math.max(m, r.updated_at), watermark);
    await writeNum(db, PUSHED_KEY, watermark);
  }

  /* The pull cursor advances ONLY as rows are applied.
     It must never be taken from the push response: the server's newest
     change_seq includes rows this device has not seen, and adopting it
     would skip them permanently — a silent, unrecoverable data loss, since
     nothing would ever ask for that range again. */
  let cursor = await readNum(db, CURSOR_KEY);
  let pulled = 0;

  for (;;) {
    const page = await call<PullResponse>(
      cfg,
      `/api/sync?cursor=${cursor}`,
      { method: "GET" },
      signal,
    );

    for (const row of page.rows) {
      if (await applyRow(db, row)) pulled++;
      // A row we just received is not a local edit. Marking it clean stops
      // the next sync pushing it straight back. Bounded by startedAt so a
      // concurrent local edit is never swept up with it.
      if (row.updated_at <= startedAt) watermark = Math.max(watermark, row.updated_at);
    }

    cursor = page.cursor;
    await writeNum(db, CURSOR_KEY, cursor);
    if (!page.more) break;
  }

  if (watermark > pushedAt) await writeNum(db, PUSHED_KEY, watermark);

  return { pushed: dirty.length, pulled, cursor };
}

/**
 * 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,
  signal?: AbortSignal,
): Promise<SyncResult | null> {
  if (!cfg?.baseUrl || !cfg.token) return null;
  try {
    return await syncOnce(db, cfg, signal);
  } catch (err) {
    console.warn("[sync] skipped:", err instanceof Error ? err.message : err);
    return null;
  }
}