cube.js

// cube.js – the local SQL copy ("cube") of the shadowed FileMaker data.
// Bones version: pull each included base table over OData, land it in a DuckDB
// file. Query via the duckdb CLI (no native module to fight during early dev;
// swap for @duckdb/node-api later if we want in-process). SELECT-only guard on
// the query path. Everything here is deliberately loose – we expect churn.

import "./env.js";
import fs from "fs";
import path from "path";
import os from "os";
import { execFile } from "child_process";
import { fileURLToPath } from "url";
import { fetchAllRows, fetchAllRowsDataApi, bestLayoutFor, pythiaLayoutFor, fetchCounts, fetchMaxModTs, takeEncodingFixNote } from "./fm.js";

// FM_ROWS_VIA=dataapi forces row pulls through Data API layouts even when the
// schema came from OData – for servers whose OData engine dies on row reads.
const ROWS_VIA = process.env.FM_ROWS_VIA || "odata";

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const DATA_DIR = process.env.DATA_DIR || path.join(__dirname, "data");
const CUBE_DIR = path.join(DATA_DIR, "cube");
const DB_PATH = process.env.DUCKDB_PATH || path.join(DATA_DIR, "pythia.duckdb");
// Prefer a bundled binary (postinstall fetches one into ./bin so `npm start`
// works without `brew install duckdb`); else DUCKDB_BIN; else PATH.
const LOCAL_DUCKDB = path.join(__dirname, "bin", process.platform === "win32" ? "duckdb.exe" : "duckdb");
const DUCKDB = process.env.DUCKDB_BIN || (fs.existsSync(LOCAL_DUCKDB) ? LOCAL_DUCKDB : "duckdb");
fs.mkdirSync(CUBE_DIR, { recursive: true });

const q = (id) => `"${String(id).replace(/"/g, '""')}"`; // quote a SQL identifier

// Each sql() shells out to a fresh duckdb process. A write process holds an
// EXCLUSIVE lock on the file that blocks any concurrent reader ("Conflicting
// lock" error) – e.g. a status/overview read landing during a multi-table sync.
// Serialize all invocations through one in-process queue so, on a single
// instance, two duckdb processes never touch the file at once; plus a short
// retry to cover the brief post-exit window and any out-of-band writer.
let dbQueue = Promise.resolve();
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// DuckDB defaults its memory limit to ~80% of SYSTEM ram and knows nothing
// about Node already holding a chunk of it. On a 512MB Fly machine that meant
// the kernel killed duckdb (four times in one evening) instead of duckdb
// spilling to disk. Give it a real ceiling and somewhere to spill.
const VM_MB = Number(process.env.FLY_VM_MEMORY_MB || 0);
const DUCK_MB = Math.max(128, Math.floor((VM_MB || 1024) * 0.45));
const SPILL_DIR = path.join(DATA_DIR, "duck-spill");
fs.mkdirSync(SPILL_DIR, { recursive: true });
const PRAGMAS = `SET memory_limit='${DUCK_MB}MB'; SET temp_directory='${SPILL_DIR.replace(/'/g, "''")}'; SET preserve_insertion_order=false;`;

function runDuckDB(query, allowWrite) {
  return new Promise((resolve, reject) => {
    // Read path runs in -safe mode: no filesystem reads, no getenv, no extension
    // installs – model-written SQL can't touch anything but the cube itself.
    // (The write path needs read_json for sync loads, so it stays unrestricted.)
    // -safe mode LOCKS configuration, so the pragmas can only go on the write
    // path. That is where the memory actually goes: sync loads, not SELECTs.
    const args = allowWrite ? [DB_PATH, "-json", "-c", PRAGMAS + query] : [DB_PATH, "-readonly", "-safe", "-json", "-c", query];
    execFile(DUCKDB, args, { maxBuffer: 256 * 1024 * 1024 }, (err, stdout, stderr) => {
      if (err) {
        // A SIGKILL with nothing on stderr is the kernel, not DuckDB: the
        // process was killed before it could complain. That used to surface as
        // a truncated echo of the command, which told nobody anything and got
        // an innocent field blamed for it.
        if (err.signal === "SIGKILL" || (!stderr && /killed/i.test(String(err.message)))) {
          return reject(new Error(`RAN OUT OF MEMORY loading this table. The server has ${VM_MB || "?"}MB and DuckDB was capped at ${DUCK_MB}MB. Sync fewer tables at once, or give the machine more memory.`));
        }
        // Otherwise keep the REASON, not the command that produced it.
        const reason = (stderr || "").trim() || String(err.message).replace(/^Command failed:[\s\S]*?\n/, "").trim() || String(err.message);
        return reject(new Error(reason.slice(0, 600)));
      }
      try { resolve(stdout.trim() ? JSON.parse(stdout) : []); }
      catch (e) { reject(new Error("Bad DuckDB output: " + e.message)); }
    });
  });
}

// Run SQL against the cube, JSON rows back. Read-only unless allowWrite.
export function sql(query, { allowWrite = false } = {}) {
  if (!allowWrite && !/^\s*(select|with|pragma|describe|summarize)\b/i.test(query)) {
    return Promise.reject(new Error("Only SELECT/WITH queries are allowed here."));
  }
  // Nothing synced yet is a SETUP STATE, not a database failure. DuckDB opened
  // read-only on a missing file throws "IO Error: Cannot open database ... in
  // read-only mode", which leaked to a user in chat as the answer to their
  // question (field report 2026-08-10). Every read path gets one typed error
  // so callers can say the useful thing instead of relaying plumbing.
  if (!allowWrite && !cubeExists()) {
    const e = new Error("No optimized copy of your data exists yet - nothing has been synced.");
    e.noCube = true;
    return Promise.reject(e);
  }
  const attempt = async () => {
    for (let i = 0; ; i++) {
      try { return await runDuckDB(query, allowWrite); }
      catch (e) {
        if (i < 40 && /conflicting lock|set lock/i.test(String(e.message))) { await sleep(250); continue; }
        throw e;
      }
    }
  };
  const result = dbQueue.then(attempt, attempt); // serialize regardless of prior outcome
  dbQueue = result.then(() => {}, () => {});      // keep the queue alive
  return result;
}

// Sync one base table. Incremental when possible: if the table has a primary
// key (UUID) and a modification-timestamp field and we already hold a
// watermark, pull only rows changed since the watermark and upsert by PK.
// Otherwise a full pull. Returns the new watermark for next time.
async function syncTable(table, opts0 = {}) {
  const { onProgress, watermark, onEvent = () => {}, shouldStop } = opts0;
  const prior0 = opts0.prior || null; // this table's manifest entry from last sync
  const t0 = Date.now();
  // Sync stored DATA fields only: no containers, and (where the server flags
  // them) no calculations, summaries, or globals – see fm.js parseEntityType.
  // Also skip x-prefixed fields (the FileMaker convention for deprecated
  // fields, e.g. xxMarginModPlus "not in use") – EXCEPT anything that looks
  // like a key (ID/UUID), because legacy joins often live on x-named fields.
  let cols = table.fields.filter((f) =>
    f.type !== "Binary" && !f.summary && !f.global &&
    // Calcs stay out UNLESS the SaXML proves this one is STORED (data at
    // rest, zero serve cost) - the blunt calc flag was dropping cheap fields.
    (!f.calc || f.storedCalc === true) &&
    !(/^x/i.test(f.name) && !/id|uuid/i.test(f.name)));

  // Resolve the row transport. FM_ROWS_VIA=dataapi = hybrid mode: schema and
  // naming come from OData but rows travel through a Data API layout, because
  // this server's OData engine dies evaluating unstored calcs on row reads.
  let via = table.via || "odata";
  let layout = table.layout;
  let layoutFields = null;
  // RUNG 0 of the ladder: a pythia_<table> layout is an explicit instruction.
  // It wins before anything else is tried - no strikes, no guessing.
  if (via !== "dataapi") {
    const pl = await pythiaLayoutFor(table.db, table.occurrences).catch(() => null);
    if (pl) {
      via = "dataapi"; layout = pl.layout; layoutFields = new Set(pl.fields);
      onEvent({ type: "table-note", name: table.name, note: `using your layout "${pl.layout}" (the pythia_ convention): ${pl.fields.length} fields, exactly what you put on it` });
    }
  }
  // ONE strike now (was two): a whole-table timeout costs minutes and is
  // expensive evidence on its own. The learned store remembers the switch.
  if (via !== "dataapi" && (opts0.slowTables || new Set()).has(table.name)) {
    const bl = await bestLayoutFor(table.db, table.occurrences);
    if (bl) {
      via = "dataapi"; layout = bl.layout;
      const onLayout = new Set(bl.fields);
      cols = cols.filter((f) => onLayout.has(f.name));
      onEvent({ type: "table-note", name: table.name, note: `this table timed out twice on OData - switched to Data API rows via layout "${bl.layout}" (remembered)` });
    }
  }
  // User-directed skips from the translated instructions: absolute.
  const userSkips = new Set((opts0.skipFields || {})[table.name] || []);
  if (userSkips.size) {
    cols = cols.filter((f) => !userSkips.has(f.name));
    onEvent({ type: "table-note", name: table.name, note: `your instructions skip: ${[...userSkips].join(", ")}` });
  }
  if (layoutFields) {
    // The pythia_ layout is authoritative: what is on it is what syncs,
    // including calcs the user placed deliberately.
    cols = table.fields.filter((f) => layoutFields.has(f.name) && f.type !== "Binary" && !f.summary && !f.global);
  }
  if (via !== "dataapi" && ROWS_VIA === "dataapi") {
    const bl = await bestLayoutFor(table.db, table.occurrences);
    if (bl) {
      via = "dataapi";
      layout = bl.layout;
      const onLayout = new Set(bl.fields);
      const before = cols.length;
      cols = cols.filter((f) => onLayout.has(f.name));
      onEvent({ type: "table-note", name: table.name, note: `hybrid transport: rows via Data API layout "${layout}" – ${cols.length} of ${before} stored fields are on it` });
    } else {
      onEvent({ type: "table-note", name: table.name, note: `no usable layout found – falling back to OData rows (may be slow on this server)` });
    }
  }

  const names = cols.map((f) => f.name);
  // DECIMAL(18,4), not DOUBLE: FileMaker numbers are decimal and money must not
  // drift. TimeOfDay had no entry at all, so time fields silently became text.
  const typeMap = { Decimal: "DECIMAL(18,4)", String: "VARCHAR", Date: "DATE", DateTimeOffset: "TIMESTAMP", TimeOfDay: "TIME", Boolean: "BOOLEAN" };
  // Data API rows carry FM-formatted date/timestamp STRINGS
  // ("07/17/2026 12:00:00"), which DuckDB's DATE/TIMESTAMP casts reject –
  // land those as VARCHAR there; model-written SQL can strptime when needed.
  const ddlType = (n) => {
    const ty = typeMap[cols.find((c) => c.name === n).type] || "VARCHAR";
    return via === "dataapi" && (ty === "DATE" || ty === "TIMESTAMP") ? "VARCHAR" : ty;
  };
  const name = table.name, esc = (s) => String(s).replace(/'/g, "''");

  const pk = table.keys?.[0] || names.find((n) => /^id$/i.test(n)) || null;
  const modField = cols.find((f) => f.type === "DateTimeOffset" && /mod/i.test(f.name))?.name || null;

  // THE FRESHNESS LADDER (Matt, 2026-08-19). Before any rows move:
  //   Rung 1 - STATIC SKIP: newest mod timestamp equals the stored watermark
  //            AND the server's count equals ours -> nothing changed, nothing
  //            deleted, the table is skipped whole. Static tables cost two
  //            tiny requests instead of a re-pull.
  //   Rung 2 - keyed incremental (PK + timestamp): changed records only.
  //   Rung 3 - TS-KEYED incremental: no PK, but the timestamp is verified
  //            unique in our copy -> the timestamp IS the key. Merge by it,
  //            verify the count afterward, fall back to full on any doubt.
  //   Rung 4 - full pull. No key and no timestamp is the honest minimum.
  let fmCount = null, probedMax = null;
  if (watermark && modField && via !== "dataapi" && prior0) {
    try {
      probedMax = await fetchMaxModTs(table.db, table.occurrences[0], modField);
      const counts = await fetchCounts([table]);
      fmCount = counts?.[table.name];
      if (probedMax != null && String(probedMax) === String(watermark) && Number.isFinite(fmCount) && fmCount === prior0.rows) {
        onEvent({ type: "table-mode", name: table.name, mode: "skip", db: table.db, occurrence: table.occurrences[0], fields: 0, reason: "unchanged since last sync - skipped" });
        onEvent({ type: "table-done", name: table.name, rows: prior0.rows, changed: 0, mode: "unchanged", ms: Date.now() - t0 });
        return { ...prior0, changed: 0, mode: "unchanged", elapsedMs: Date.now() - t0 };
      }
    } catch { /* probe is best-effort; the ladder continues */ }
  }
  // Incremental only if we have a watermark, the two fields, and the existing
  // table's columns still match (schema change forces a full rebuild).
  // Data API transport always does full pulls (no $filter watermark there yet).
  let incremental = Boolean(watermark && pk && modField) && via !== "dataapi";
  // Rung 3: the timestamp as a pseudo-key, only when PROVEN unique locally.
  let tsKeyed = false;
  if (!incremental && !pk && modField && watermark && via !== "dataapi" && prior0) {
    try {
      const [u] = await sql(`SELECT count(*) AS n, count(DISTINCT ${q(modField)}) AS d FROM ${q(name)}`);
      if (u && u.n > 0 && u.n === u.d) { tsKeyed = true; incremental = true; }
    } catch { /* table missing locally: full pull */ }
  }
  if (incremental) {
    const mainCols = (await sql(`SELECT column_name FROM information_schema.columns WHERE table_name='${esc(name)}' ORDER BY ordinal_position`)).map((r) => r.column_name);
    if (mainCols.length !== names.length || !names.every((n, i) => mainCols[i] === n)) incremental = false;
  }

  // Watermark comparison: `ge` (>=) only while the watermark second is recent
  // enough that new edits could still land in it (clock-skew guard). Once it's
  // safely in the past, strict `gt` – otherwise a bulk import that stamped
  // thousands of rows in one second gets re-pulled on every sync forever.
  const wmOp = watermark && Date.now() - Date.parse(watermark) > 5 * 60 * 1000 ? "gt" : "ge";
  onEvent({
    type: "table-mode", name, mode: incremental ? (tsKeyed ? "ts-incremental" : "incremental") : "full",
    db: table.db, occurrence: via === "dataapi" ? `layout "${layout}" (Data API)` : table.occurrences[0], fields: names.length,
    reason: via === "dataapi" ? (table.via === "dataapi" ? "OData down – pulling through the Data API, full pull" : "hybrid – OData schema, Data API rows, full pull")
      : incremental ? (tsKeyed ? `changed records only, keyed by the timestamp (verified unique)` : `changes since ${watermark}`) : (watermark ? "schema changed – full rebuild" : (pk && modField ? "first pull" : `no ${pk ? "modification timestamp" : "primary key"} – always full`)),
  });
  const onNote = (note) => onEvent({ type: "table-note", name, note });
  // Keyset candidate: a NUMERIC key we are already pulling. The PK when it is
  // numeric; otherwise a conventional serial (RecID and friends). Text UUIDs
  // stay on $skip – "gt" on strings is lexicographic roulette. fetchAllRows
  // tries keyset once per table and falls back on its own.
  const numericSerial = (n) => /^(rec_?id|record_?id|serial(_?number)?)$/i.test(n);
  const keysetField =
    (pk && names.includes(pk) && cols.find((c) => c.name === pk)?.type === "Decimal" && pk) ||
    cols.find((c) => c.type === "Decimal" && numericSerial(c.name))?.name || undefined;
  const rows = via === "dataapi"
    ? await fetchAllRowsDataApi(table.db, layout, names, { onProgress, onNote })
    : await fetchAllRows(table.occurrences[0], names, {
        db: table.db, onProgress, filter: incremental ? `${modField} ${wmOp} ${watermark}` : undefined, onNote,
        shouldStop, keysetField,
        startPageSize: opts0.pageSizes?.[table.name],
        onPageSettled: (size) => opts0.onPageSize?.(table.name, size),
        onProbe: (p) => onEvent({ type: "table-probe", name: table.name, msPerRecord: Math.round(p.msPerRecord), sampled: p.sampled }),
      });
  const newWatermark = modField ? rows.reduce((mx, r) => (r[modField] && r[modField] > mx ? r[modField] : mx), watermark || "") : null;

  // Data API rows carry FM's loose number typing: a number field can hold
  // "00" or "1,200" as a STRING, which DuckDB's strict DOUBLE cast rejects.
  // Coerce cleanly-numeric strings; anything unparseable becomes null.
  if (via === "dataapi") {
    const numCols = cols.filter((f) => f.type === "Decimal").map((f) => f.name);
    for (const r of rows) for (const n of numCols) {
      const v = r[n];
      if (v != null && typeof v !== "number") {
        const num = Number(String(v).replace(/,/g, "").trim());
        r[n] = Number.isFinite(num) ? num : null;
      }
    }
  }
  // MONSTER COLUMNS: a plain text field holding base64 images passes every
  // type filter and averages hundreds of KB per row (IL_Product's photo field
  // OOM-killed a 512MB machine). Measure what actually arrived; a column
  // averaging over 8KB per row is skipped WHOLE, never truncated - a silently
  // shortened contract is worse than an excluded photo column. Keys are exempt.
  if (rows.length >= 20) {
    const sample = rows.slice(0, 200);
    for (const n of [...names]) {
      if (n === pk || n === modField) continue;
      const avg = sample.reduce((a, r) => a + (typeof r[n] === "string" ? r[n].length : 0), 0) / sample.length;
      if (avg > 8192) {
        names.splice(names.indexOf(n), 1);
        for (const r of rows) delete r[n];
        const note = `column "${n}" skipped - averages ${Math.round(avg / 1024)}KB per record (likely embedded files); it stays in FileMaker`;
        onEvent({ type: "table-coercions", name, coercions: [{ field: n, kept: "skipped", bad: 0, nonNull: sample.length }] });
        onEvent({ type: "table-note", name, note });
      }
    }
  }

  // INGEST NEVER FAILS ON CONTENT OR SIZE.
  //
  // FileMaker's field type is a preference; DuckDB's columns={} is a contract.
  // A single "-10" in a date field used to abort the whole read and lose all
  // 9,000 rows. And one giant JSON per table meant peak memory scaled with the
  // table, which OOM-killed DuckDB four times on a 512MB machine.
  //
  // So: land every column as TEXT, in batches, into a staging table (cannot
  // fail), then build the real typed table from it with TRY_CAST (returns NULL
  // instead of throwing). Values that refuse to cast are recorded with their
  // record key and raw text, so nothing is silently lost.
  const stage = `_stage_${name}`;
  const stageStruct = names.map((n) => `${q(n)}: 'VARCHAR'`).join(", ");
  await sql(`DROP TABLE IF EXISTS ${q(stage)}; CREATE TABLE ${q(stage)} (${names.map((n) => `${q(n)} VARCHAR`).join(", ")});`, { allowWrite: true });
  const BATCH = Number(process.env.SYNC_BATCH_ROWS || 5000);
  for (let i = 0; i < rows.length; i += BATCH) {
    if (shouldStop?.()) { const err = new Error(`cancelled while loading ${name}`); err.cancelled = true; throw err; }
    const chunk = rows.slice(i, i + BATCH);
    const file = path.join(CUBE_DIR, `${name}.part.json`);
    fs.writeFileSync(file, JSON.stringify(chunk));
    try {
      await sql(`INSERT INTO ${q(stage)} SELECT * FROM read_json('${file.replace(/'/g, "''")}', columns={${stageStruct}}, format='array', maximum_object_size=16777216);`, { allowWrite: true });
    } catch (e) {
      if (/no space left|enospc|disk.*full/i.test(String(e.message))) {
        const err = new Error("This server's data volume is FULL - the sync cannot store more. Free space (remove unneeded tables) or give the volume more room, then sync again. Your FileMaker data is untouched.");
        err.diskFull = true; throw err;
      }
      throw e;
    }
    onEvent({ type: "table-load", name, rows: Math.min(i + chunk.length, rows.length), of: rows.length, incremental });
    try { fs.unlinkSync(file); } catch { /* best effort */ }
  }

  // Type the columns. A column whose values mostly refuse to cast was never
  // really that type - FileMaker's declaration was aspirational - so it stays
  // text rather than being emptied out.
  const coercions = [];
  const typed = [];
  for (const n of names) {
    const t = ddlType(n);
    if (t === "VARCHAR") { typed.push(`${q(n)} AS ${q(n)}`); continue; }
    const [{ bad = 0, nonNull = 0 } = {}] = await sql(
      `SELECT count(*) FILTER (WHERE ${q(n)} IS NOT NULL AND trim(${q(n)}) <> '' AND TRY_CAST(${q(n)} AS ${t}) IS NULL) AS bad,
              count(*) FILTER (WHERE ${q(n)} IS NOT NULL AND trim(${q(n)}) <> '') AS nonNull
       FROM ${q(stage)}`);
    const rate = nonNull ? bad / nonNull : 0;
    if (rate > 0.05) { typed.push(`${q(n)} AS ${q(n)}`); if (bad) coercions.push({ field: n, kept: "text", bad, nonNull }); continue; }
    typed.push(`TRY_CAST(${q(n)} AS ${t}) AS ${q(n)}`);
    if (bad) {
      coercions.push({ field: n, type: t, bad, nonNull });
      const keyCol = pk && names.includes(pk) ? q(pk) : `NULL`;
      await sql(
        `CREATE TABLE IF NOT EXISTS "_pythia_coercions" ("table" VARCHAR, "field" VARCHAR, "record" VARCHAR, "value" VARCHAR, "expected" VARCHAR, "at" TIMESTAMP);
         DELETE FROM "_pythia_coercions" WHERE "table" = '${esc(name)}' AND "field" = '${esc(n)}';
         INSERT INTO "_pythia_coercions" SELECT '${esc(name)}', '${esc(n)}', CAST(${keyCol} AS VARCHAR), ${q(n)}, '${esc(t)}', now()
           FROM ${q(stage)} WHERE ${q(n)} IS NOT NULL AND trim(${q(n)}) <> '' AND TRY_CAST(${q(n)} AS ${t}) IS NULL LIMIT 500;`,
        { allowWrite: true });
    }
  }
  if (coercions.length) onEvent({ type: "table-coercions", name, coercions });

  if (incremental && rows.length) {
    const mergeKey = tsKeyed ? modField : pk; // rung 3 merges by the verified-unique timestamp
    await sql(
      `CREATE OR REPLACE TABLE "_delta_${esc(name)}" AS SELECT ${typed.join(", ")} FROM ${q(stage)};` +
      `DELETE FROM ${q(name)} WHERE ${q(mergeKey)} IN (SELECT ${q(mergeKey)} FROM "_delta_${esc(name)}");` +
      `INSERT INTO ${q(name)} SELECT * FROM "_delta_${esc(name)}";` +
      `DROP TABLE "_delta_${esc(name)}";`,
      { allowWrite: true });
    if (tsKeyed) {
      // Trust, then verify: our count must equal FileMaker's. Any drift means
      // the timestamp lied as a key (a batch stamped duplicates since the
      // uniqueness check) - rebuild whole rather than serve a maybe.
      const [{ c: ourCount }] = await sql(`SELECT count(*) c FROM ${q(name)}`);
      let liveCount = fmCount;
      if (!Number.isFinite(liveCount)) { try { liveCount = (await fetchCounts([table]))?.[table.name]; } catch { liveCount = null; } }
      if (Number.isFinite(liveCount) && Number(ourCount) !== Number(liveCount)) {
        onEvent({ type: "table-note", name, note: `timestamp-keyed merge failed its count check (ours ${ourCount}, FileMaker ${liveCount}) - rebuilding whole` });
        const all = await fetchAllRows(table.occurrences[0], names, { db: table.db, onProgress, onNote, shouldStop, keysetField });
        rows.length = 0; rows.push(...all);
        await sql(`DROP TABLE IF EXISTS ${q(stage)}; CREATE TABLE ${q(stage)} (${names.map((n) => `${q(n)} VARCHAR`).join(", ")});`, { allowWrite: true });
        for (let i = 0; i < rows.length; i += BATCH) {
          const chunk = rows.slice(i, i + BATCH);
          const file = path.join(CUBE_DIR, `${name}.part.json`);
          fs.writeFileSync(file, JSON.stringify(chunk));
          await sql(`INSERT INTO ${q(stage)} SELECT * FROM read_json('${file.replace(/'/g, "''")}', columns={${names.map((n) => `${q(n)}: 'VARCHAR'`).join(", ")}}, format='array', maximum_object_size=16777216);`, { allowWrite: true });
          try { fs.unlinkSync(file); } catch {}
        }
        await sql(`CREATE OR REPLACE TABLE ${q(name)} AS SELECT ${typed.join(", ")} FROM ${q(stage)};`, { allowWrite: true });
      }
    }
  } else if (!incremental) {
    await sql(`CREATE OR REPLACE TABLE ${q(name)} AS SELECT ${typed.join(", ")} FROM ${q(stage)};`, { allowWrite: true });
  }
  await sql(`DROP TABLE IF EXISTS ${q(stage)};`, { allowWrite: true });

  const total = (await sql(`SELECT count(*) c FROM ${q(name)}`))[0]?.c ?? rows.length;
  return { name, db: table.db, rows: total, changed: rows.length, mode: incremental ? "incremental" : "full", watermark: newWatermark, columns: names, coercions, pk, elapsedMs: Date.now() - t0 };
}

// Sync a specific set of base tables into the existing cube (create-or-replace
// each), merging their entries into the manifest. Used by both full builds and
// the smart-live path (refresh only the tables a query touches).
export async function syncTables(tables, names, log = () => {}, onEvent = () => {}, opts = {}) {
  // Sync in the ORDER GIVEN (the caller passes display order), not raw schema
  // order, so the screen and the work agree about what happens next.
  const wanted = new Map(tables.map((t) => [t.name, t]));
  const chosen = names.map((n) => wanted.get(n)).filter(Boolean);
  // A pick that no longer matches the schema must FAIL AUDIBLY. The silent
  // filter above once gutted a 3-table plan to 1 and declared success
  // (2026-08-18): renamed-apart files orphaned the saved picks and nothing
  // said a word. Identity is frozen now, but this guard stays: belt, braces.
  const failed = [];
  for (const n of names) {
    if (wanted.has(n)) continue;
    const error = "this table's name no longer matches the current schema - re-pick it in Settings > Tables and sync again";
    log(`  ${n}: FAILED – ${error}`);
    failed.push({ name: n, error });
    onEvent({ type: "table-error", name: n, error });
  }
  const prior = Object.fromEntries((cubeManifest().tables || []).map((t) => [t.name, t]));
  const results = [];
  let cancelled = false;
  for (let ti = 0; ti < chosen.length; ti++) {
    const t = chosen[ti];
    // Cancel lands between tables, never inside one: every table in the cube
    // is whole. What already synced stays synced.
    if (opts.shouldStop?.()) {
      cancelled = true;
      const remaining = chosen.slice(ti).map((x) => x.name);
      log(`sync cancelled; ${remaining.length} table(s) untouched: ${remaining.join(", ")}`);
      onEvent({ type: "cancelled", finished: results.map((r) => r.name), remaining });
      break;
    }
    log(`syncing ${t.name}…`);
    onEvent({ type: "table-start", name: t.name });
    // One bad table must never kill the run: catch, report, move on. Any
    // prior manifest entry survives, so old data for that table stays usable.
    try {
      const r = await syncTable(t, {
        watermark: prior[t.name]?.watermark || null,
        prior: prior[t.name] || null,
        onEvent,
        shouldStop: opts.shouldStop,
        slowTables: opts.slowTables,
        pageSizes: opts.pageSizes, skipFields: opts.skipFields, onPageSize: opts.onPageSize,
        onProgress: (n, pg) => { log(`  ${t.name}: ${n} rows`); onEvent({ type: "table-rows", name: t.name, rows: n, page: pg?.page, pageRows: pg?.pageRows }); },
      });
      log(`  ${t.name}: ${r.mode} · ${r.changed} changed · ${r.rows} total`);
      { const fx = takeEncodingFixNote(); if (fx) { log(`  ${fx}`); onEvent({ type: "table-fix", name: t.name, note: fx }); } }
      // DELETE RECONCILIATION. Mod-date sync cannot see a deleted record (it is
      // not there to answer the $filter), so ghosts accumulate - a production
      // system reported on 6 records that no longer existed. When FileMaker's count is
      // LOWER than ours after an incremental sync, pull just the key column and
      // delete what FileMaker no longer has. Full pulls need none of this.
      if (r.mode === "incremental" && r.pk && r.rows <= 200000) {
        try {
          const counts = await fetchCounts([t]);
          const fmCount = counts?.[t.name];
          if (Number.isFinite(fmCount) && fmCount < r.rows) {
            const keyRows = await fetchAllRows(t.occurrences[0], [r.pk], { db: t.db });
            const liveKeys = new Set(keyRows.map((k) => String(k[r.pk])));
            const ours = await sql(`SELECT ${q(r.pk)} AS k FROM ${q(t.name)}`);
            const ghosts = ours.map((x) => String(x.k)).filter((k) => !liveKeys.has(k));
            if (ghosts.length && ghosts.length < r.rows * 0.5) {
              const listSql = ghosts.map((g) => `'${String(g).replace(/'/g, "''")}'`).join(",");
              await sql(`DELETE FROM ${q(t.name)} WHERE ${q(r.pk)} IN (${listSql})`, { allowWrite: true });
              r.rows -= ghosts.length;
              log(`  ${t.name}: removed ${ghosts.length} record(s) deleted in FileMaker`);
              onEvent({ type: "table-note", name: t.name, note: `${ghosts.length} record(s) deleted in FileMaker were removed from the local copy` });
            }
          }
        } catch (e) { log(`  ${t.name}: delete check skipped (${String(e.message).slice(0, 60)})`); }
      }
      onEvent({ type: "table-done", name: t.name, rows: r.rows, changed: r.changed, mode: r.mode, ms: r.elapsedMs });
      results.push(r);
    } catch (e) {
      if (e && e.cancelled) {
        // Stop NOW: the in-flight table is abandoned whole – its staging table
        // is dropped, the cube keeps whatever copy it had before. Tables after
        // it are untouched.
        cancelled = true;
        try { await sql(`DROP TABLE IF EXISTS ${q(`_stage_${t.name}`)}`, { allowWrite: true }); } catch { /* best effort */ }
        const remaining = chosen.slice(ti + 1).map((x) => x.name);
        log(`sync stopped during ${t.name}; nothing kept for it. ${remaining.length} table(s) untouched.`);
        onEvent({ type: "cancelled", current: t.name, finished: results.map((r) => r.name), remaining });
        break;
      }
      const error = String(e.message || e);
      log(`  ${t.name}: FAILED – ${error}`);
      failed.push({ name: t.name, error });
      if (/page limit|timeout/i.test(error) && opts.onSlow) opts.onSlow(t.name); // candidate for the Data API road
      onEvent({ type: "table-error", name: t.name, error });
    }
  }
  const manifest = cubeManifest();
  const byName = Object.fromEntries((manifest.tables || []).map((t) => [t.name, t]));
  for (const r of results) byName[r.name] = r;
  const merged = Object.values(byName);
  const out = { db: DB_PATH, syncedAt: new Date().toISOString(), host: os.hostname(), tables: merged, totalRows: merged.reduce((a, b) => a + b.rows, 0) };
  fs.writeFileSync(path.join(DATA_DIR, "cube-manifest.json"), JSON.stringify(out, null, 2));
  const extra = {};
  if (failed.length) extra.failed = failed;
  if (cancelled) extra.cancelled = true;
  return { ...out, ...extra };
}

// Full rebuild for the chosen tables.
export async function buildCube(tables, includeNames, log = () => {}) {
  return syncTables(tables, includeNames, log);
}

// Drop tables from the local copy (delete data) and prune them from the
// manifest. The table name still exists upstream and in /api/schema, so it can
// be turned back on later – we only remove the local copy.
export async function dropTables(names, log = () => {}, onEvent = () => {}) {
  for (const n of names) {
    try { await sql(`DROP TABLE IF EXISTS ${q(n)}`, { allowWrite: true }); log(`dropped ${n}`); onEvent({ type: "drop", name: n }); } catch (e) { log(`drop ${n} failed: ${e.message}`); }
  }
  const m = cubeManifest();
  const kept = (m.tables || []).filter((t) => !names.includes(t.name));
  const out = { ...m, tables: kept, totalRows: kept.reduce((a, b) => a + b.rows, 0), syncedAt: new Date().toISOString() };
  fs.writeFileSync(path.join(DATA_DIR, "cube-manifest.json"), JSON.stringify(out, null, 2));
  return out;
}

// Full wipe for "reset to fresh install": remove the local database, its
// per-table watermark files, and the manifest. The next sync rebuilds from
// scratch, and a fresh schema scan sees the server as if for the first time.
export function resetCube(log = () => {}) {
  for (const p of [DB_PATH, DB_PATH + ".wal", path.join(DATA_DIR, "cube-manifest.json")]) {
    try { fs.unlinkSync(p); log("removed " + path.basename(p)); } catch { /* not there is fine */ }
  }
  try {
    for (const f of fs.readdirSync(CUBE_DIR)) {
      if (f.endsWith(".json")) { try { fs.unlinkSync(path.join(CUBE_DIR, f)); } catch {} }
    }
  } catch { /* no cube dir yet is fine */ }
}

// Reconcile the local copy to exactly `includeNames`: sync those, drop the rest.
export async function reconcileCube(tables, includeNames, log = () => {}, onEvent = () => {}, opts = {}) {
  const manifest = cubeManifest();
  const have = (manifest.tables || []).map((t) => t.name);
  const toDrop = have.filter((n) => !includeNames.includes(n));
  onEvent({ type: "plan", tables: includeNames, toDrop });
  if (toDrop.length) await dropTables(toDrop, log, onEvent);
  const result = await syncTables(tables, includeNames, log, onEvent, opts);
  const after = cubeManifest();
  onEvent({ type: "done", totalRows: after.totalRows, tables: (after.tables || []).map((t) => t.name),
            cancelled: Boolean(result?.cancelled) });
  return after;
}

export function cubeManifest() {
  try { return JSON.parse(fs.readFileSync(path.join(DATA_DIR, "cube-manifest.json"), "utf8")); }
  catch { return { syncedAt: null, tables: [], totalRows: 0 }; }
}

export function cubeExists() {
  return fs.existsSync(DB_PATH);
}