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, fetchHasNewerThan, takeEncodingFixNote, tsToEpoch, tsToFmFind } 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.
//
// Two corrections (2026-08-24, after a 256MB machine OOMed repeatedly):
//   1. Budget from what the machine ACTUALLY has. A "256MB" Fly VM reports
//      207MB of usable memory; the kernel and init take the rest.
//   2. The old floor of 128MB was applied UPWARD. On a small box the 45%
//      share came to 115MB and the floor raised it to 128, so DuckDB was
//      promised 128MB while Node held 113MB on a 207MB machine. The guard
//      meant to protect DuckDB is what killed the process.
// Node gets its reserve first, DuckDB gets what is genuinely left.
const usableMB = (() => {
  let mem = 0;
  try {
    const m = /MemTotal:\s+(\d+)\s*kB/.exec(fs.readFileSync("/proc/meminfo", "utf8"));
    if (m) mem = Math.floor(Number(m[1]) / 1024);
  } catch { /* not Linux, or no procfs */ }
  // In a container, /proc/meminfo reports the HOST's memory. The cgroup
  // limit is the real ceiling (Docker with -m; Fly reports the VM either
  // way): budgeting from a 16GB host inside a 512MB container is a
  // guaranteed OOM kill. cgroup v2 first, v1 as fallback.
  for (const p of ["/sys/fs/cgroup/memory.max", "/sys/fs/cgroup/memory/memory.limit_in_bytes"]) {
    try {
      const raw = fs.readFileSync(p, "utf8").trim();
      const v = Number(raw);
      if (raw !== "max" && Number.isFinite(v) && v > 0 && v < 2 ** 50)
        mem = mem ? Math.min(mem, Math.floor(v / 1048576)) : Math.floor(v / 1048576);
      break;
    } catch { /* try the next flavor */ }
  }
  if (mem) return mem;
  return Number(process.env.FLY_VM_MEMORY_MB || 0) || 1024;
})();
// The reserve scales with the machine but is clamped: a small box cannot
// afford 96MB of reserve (it starved DuckDB below the level that synced
// fine before), and a huge box does not need 35% held back for Node.
const NODE_RESERVE_MB = Math.min(Math.max(80, Math.floor(usableMB * 0.35)), 512);
const DUCK_MB = Math.max(48, usableMB - NODE_RESERVE_MB - 24); // 24MB for the OS
const VM_MB = usableMB;
const SPILL_DIR = path.join(DATA_DIR, "duck-spill");
fs.mkdirSync(SPILL_DIR, { recursive: true });
console.log(`[duckdb] machine has ${usableMB}MB usable; node reserve ${NODE_RESERVE_MB}MB, duckdb limit ${DUCK_MB}MB`);
const PRAGMAS = `SET memory_limit='${DUCK_MB}MB'; SET temp_directory='${SPILL_DIR.replace(/'/g, "''")}'; SET preserve_insertion_order=false;`;

// What this machine should be, given the data. Evidence, not theory: at a
// 602MB DuckDB share, a 485k-row table builds fine; at an 87MB share, a
// 134k-row table dies. The ladder below has a rung of headroom on both ends.
export function recommendMemoryMB(largestTableRows) {
  const r = Number(largestTableRows) || 0;
  if (r <= 50000) return 256;
  if (r <= 250000) return 512;
  if (r <= 1200000) return 1024;
  return 2048;
}
export function machineMemory() { return { usableMB: usableMB, nodeReserveMB: NODE_RESERVE_MB, duckMB: DUCK_MB }; }

// The kernel gives no warning before it kills. Node CAN see it coming:
// MemAvailable between batches. Under this floor we stop the sync ourselves,
// cleanly, with a message - instead of the kernel choosing a victim and the
// whole app rebooting mid-click (the pre-2026-08-24 experience).
const MEM_FLOOR_MB = 25;
export function memoryPressure() {
  try {
    const m = /MemAvailable:\s+(\d+)\s*kB/.exec(fs.readFileSync("/proc/meminfo", "utf8"));
    if (!m) return null;
    const availMB = Math.floor(Number(m[1]) / 1024);
    return { availMB, critical: availMB < MEM_FLOOR_MB };
  } catch { return null; } // not Linux: no guard, no harm
}

// Cancel must reach a RUNNING write. A long typed-build statement on a large
// table ran for a minute after the user pressed Cancel, because aborting
// fetches and checking flags between pages cannot interrupt a statement the
// database engine is already executing (Matt, 2026-08-26). So write children
// are tracked, and cancelWrites() terminates them; the resulting failure is
// reported as a CANCEL, never as an error or an OOM.
const activeWrites = new Set();
let writesCancelled = false;
export function cancelWrites() {
  writesCancelled = true;
  for (const child of activeWrites) { try { child.kill("SIGTERM"); } catch { /* already gone */ } }
}
export function resetCancelWrites() { writesCancelled = false; }

function runDuckDB(query, allowWrite, cleanup) {
  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];
    // `cleanup` writes (dropping an abandoned staging table) are how cancel
    // tidies up after itself; the cancel gate must not refuse its own broom.
    if (allowWrite && writesCancelled && !cleanup) { const e = new Error("cancelled"); e.cancelled = true; return reject(e); }
    const child = execFile(DUCKDB, args, { maxBuffer: 256 * 1024 * 1024 }, (err, stdout, stderr) => {
      if (allowWrite) activeWrites.delete(child);
      if (err) {
        if (allowWrite && writesCancelled && !cleanup) { const e2 = new Error("cancelled during a database write"); e2.cancelled = true; return reject(e2); }
        // 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)))) {
          const e2 = new Error(`RAN OUT OF MEMORY. This machine has ${VM_MB || "?"}MB; increase the Fly machine's memory and sync again. Tables already synced are kept.`);
          e2.outOfMemory = true;
          return reject(e2);
        }
        // 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)); }
    });
    if (allowWrite) activeWrites.add(child);
  });
}

// Run SQL against the cube, JSON rows back. Read-only unless allowWrite.
export function sql(query, { allowWrite = false, cleanup = 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, cleanup); }
      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;
  // The probe rides OData even when ROWS ride the Data API (the pythia_
  // layout road): the OData engine answers $filter/$count fine there. The
  // one exception is via forced to dataapi because OData itself is DOWN
  // (table.via === "dataapi") - no probe possible, ladder continues to full.
  // Watermarks are compared as epochs: OData speaks ISO, Data API rows speak
  // FileMaker format, and a string comparison across dialects never matches
  // (which silently disabled this rung - the 2026-08-22 "10s no-change" bug).
  if (watermark && modField && table.via !== "dataapi" && prior0) {
    try {
      // "Is there anything newer?" instead of "what is the newest?". The old
      // $orderby max-probe cost 53-60s server-side against a 20s timeout, so
      // it always failed and this rung never fired on the tables it was built
      // for. Both run together: the count still catches DELETIONS, which a
      // newer-row check cannot see.
      onEvent({ type: "table-note", name: table.name, note: "checking for changes…" });
      const [hasNewer, counts] = await Promise.all([
        fetchHasNewerThan(table.db, table.occurrences[0], modField, watermark, shouldStop),
        fetchCounts([table], undefined, shouldStop),
      ]);
      fmCount = counts?.[table.name];
      if (hasNewer === false && Number.isFinite(fmCount) && fmCount === prior0.rows) {
        onEvent({ type: "table-mode", name: table.name, mode: "skip", via, layout: null, pythiaLayout: false, db: table.db, occurrence: table.occurrences[0], fields: 0, reason: `nothing newer than ${watermark} and the count still matches (${prior0.rows} rows) - skipped in ${((Date.now() - t0) / 1000).toFixed(1)}s` });
        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 (e) {
      if (e && e.cancelled) throw e; // a cancel is never "best effort"
      /* 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).
  // The Data API road does incrementals through a _find on the mod field -
  // measured at 0.18s for 281 changed rows, so the fastest first-sync lane
  // is no longer the slowest steady-state lane.
  // On the Data API road, `names` holds only what the LAYOUT carries, but pk
  // comes from OData schema keys - a merge whose pk field is missing from the
  // layout would fail on every sync forever. Both fields must be present.
  let incremental = Boolean(watermark && pk && modField) && (via !== "dataapi" || (names.includes(modField) && names.includes(pk)));
  // Rung 3: the timestamp as a pseudo-key, only when PROVEN unique locally.
  let tsKeyed = false;
  if (!incremental && !pk && modField && watermark && (via !== "dataapi" || names.includes(modField)) && 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.
  // tsToEpoch, not Date.parse: a Data-API watermark is UTC wall time in FM
  // format, which Date.parse reads as LOCAL time and shifts the skew guard
  // by the host's whole UTC offset.
  const wmEpoch = watermark ? tsToEpoch(watermark) : null;
  const wmOp = wmEpoch != null && Date.now() - wmEpoch > 5 * 60 * 1000 ? "gt" : "ge";
  if (process.env.PYTHIA_DEBUG_SYNC) console.log(`[sync-mode] ${name}: ${incremental ? (tsKeyed ? "ts-incremental" : "incremental") : "full"} | wm=${Boolean(watermark)} pk=${pk || "-"} modField=${modField || "-"}`);
  onEvent({
    type: "table-mode", name, mode: incremental ? (tsKeyed ? "ts-incremental" : "incremental") : "full",
    via, layout: via === "dataapi" ? layout : null, pythiaLayout: Boolean(layoutFields),
    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;
  // STREAMING INGEST. Rows are consumed page by page and released; nothing
  // holds the whole table. The write side was already batched, but the fetch
  // built the entire table in memory first, so the peak had already happened
  // before the first batch was written - a half-applied fix that read as a
  // complete one. A 100k-row table OOM-killed a 256MB machine (2026-08-24).
  //
  // Two decisions must be made from the FIRST page, before any row lands:
  // which columns to drop as monsters, and therefore the staging table's
  // shape. Both were previously made from a sample of the full array.
  const stage = `_stage_${name}`;
  const BATCH = Number(process.env.SYNC_BATCH_ROWS || 5000);
  const numCols = via === "dataapi" ? cols.filter((f) => f.type === "Decimal").map((f) => f.name) : [];
  let rowCount = 0, staged = false, buf = [], newWatermark = modField ? (watermark || "") : null;
  const dropped = new Set();

  const flush = async () => {
    if (!buf.length) return;
    const chunk = buf; buf = [];
    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={${names.map((n) => `${q(n)}: 'VARCHAR'`).join(", ")}}, format='array', maximum_object_size=16777216);`, { allowWrite: true });
    } catch (e) {
      try { fs.unlinkSync(file); } catch { /* the chunk failed to load; don't leak it */ }
      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: rowCount, of: null, incremental });
    try { fs.unlinkSync(file); } catch { /* best effort */ }
  };

  const onPage = async (page) => {
    if (shouldStop?.()) { const err = new Error(`cancelled while loading ${name}`); err.cancelled = true; throw err; }
    // Stop OURSELVES before the kernel stops us. An OOM kill takes the whole
    // app down mid-click; this stops one sync with a sentence instead.
    const mp = memoryPressure();
    if (mp?.critical) {
      const err = new Error(`the machine is nearly out of memory (${mp.availMB}MB left). Increase the Fly machine's memory and sync again. Tables already synced are kept.`);
      err.outOfMemory = true; throw err;
    }
    // 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.
    for (const r of page) 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;
      }
    }
    if (!staged) {
      // MONSTER COLUMNS: a plain text field holding base64 images passes every
      // type filter and averages hundreds of KB per row. A column averaging
      // over 8KB per row is skipped WHOLE, never truncated - a silently
      // shortened contract is worse than an excluded photo column.
      // Any non-empty first page can vote: a shrunken or learned page size
      // under 20 rows used to skip the check entirely and ship every base64
      // photo column into the cube (the class of OOM that took a machine down).
      if (page.length >= 1) {
        const sample = page.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);
            dropped.add(n);
            onEvent({ type: "table-coercions", name, coercions: [{ field: n, kept: "skipped", bad: 0, nonNull: sample.length }] });
            onEvent({ type: "table-note", name, note: `column "${n}" skipped - averages ${Math.round(avg / 1024)}KB per record (likely embedded files); it stays in FileMaker` });
          }
        }
      }
      await sql(`DROP TABLE IF EXISTS ${q(stage)}; CREATE TABLE ${q(stage)} (${names.map((n) => `${q(n)} VARCHAR`).join(", ")});`, { allowWrite: true });
      staged = true;
    }
    if (dropped.size) for (const r of page) for (const n of dropped) delete r[n];
    // Walk the watermark as we go. Compare as epochs, never as strings:
    // FileMaker-format dates sort wrong lexically (12/31/2024 > 01/05/2026),
    // which would walk it BACKWARD and re-pull or miss changed records.
    if (modField) for (const r of page) {
      const v = r[modField]; if (!v) continue;
      const e = tsToEpoch(v), me = tsToEpoch(newWatermark);
      if (e !== null && (me === null || e > me)) newWatermark = v;
    }
    rowCount += page.length;
    buf.push(...page);
    if (buf.length >= BATCH) await flush();
  };

  if (via === "dataapi") {
    await fetchAllRowsDataApi(table.db, layout, names, { onProgress, onNote, shouldStop, onPage,
      find: incremental ? { field: modField, op: wmOp === "gt" ? ">" : ">=", value: tsToFmFind(watermark) } : undefined });
  } else {
    await fetchAllRows(table.occurrences[0], names, {
      db: table.db, onProgress, filter: incremental ? { field: modField, op: wmOp, value: watermark } : undefined, onNote,
      shouldStop, keysetField, onPage,
      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 }),
    });
  }
  await flush();
  // An empty pull still needs its staging table, or the typed build below
  // has nothing to read from.
  if (!staged) {
    await sql(`DROP TABLE IF EXISTS ${q(stage)}; CREATE TABLE ${q(stage)} (${names.map((n) => `${q(n)} VARCHAR`).join(", ")});`, { allowWrite: true });
    staged = true;
  }

  // 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 && rowCount) {
    const mergeKey = tsKeyed ? modField : pk; // rung 3 merges by the verified-unique timestamp
    // One transaction: a cancel's SIGTERM landing between the DELETE and the
    // INSERT must roll back, not leave the live table missing its changed
    // rows until the next sync ("tables land whole or not at all").
    await sql(
      `CREATE OR REPLACE TABLE "_delta_${esc(name)}" AS SELECT ${typed.join(", ")} FROM ${q(stage)};` +
      `BEGIN TRANSACTION;` +
      `DELETE FROM ${q(name)} WHERE ${q(mergeKey)} IN (SELECT ${q(mergeKey)} FROM "_delta_${esc(name)}");` +
      `INSERT INTO ${q(name)} SELECT * FROM "_delta_${esc(name)}";` +
      `COMMIT;` +
      `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` });
        // Stream the rebuild too. This path re-pulls a WHOLE table, so it is
        // the last place that should hold one in memory.
        await sql(`DROP TABLE IF EXISTS ${q(stage)}; CREATE TABLE ${q(stage)} (${names.map((n) => `${q(n)} VARCHAR`).join(", ")});`, { allowWrite: true });
        let rebuf = [];
        const reflush = async () => {
          if (!rebuf.length) return;
          const chunk = rebuf; rebuf = [];
          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 fetchAllRows(table.occurrences[0], names, { db: table.db, onProgress, onNote, shouldStop, keysetField,
          onPage: async (page) => { rebuf.push(...page); if (rebuf.length >= BATCH) await reflush(); } });
        await reflush();
        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 ?? rowCount;
  return { name, db: table.db, rows: total, changed: rowCount, 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;
  let oomCount = 0; // one big table OOMing must not doom the small ones after it
  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, cleanup: 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;
      }
      // A FAILED table leaves its staging table behind. Four of them were
      // still sitting in a cube hours later, holding disk and showing up in
      // table listings as _stage_-prefixed ghosts
      // (2026-08-24). Cancel already cleaned up after itself; failure did not.
      try { await sql(`DROP TABLE IF EXISTS ${q(`_stage_${t.name}`)}`, { allowWrite: true }); } catch { /* best effort */ }
      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 });
      if (e && e.outOfMemory) {
        oomCount++;
        // The recommendation must include the table that just DIED, not only
        // the ones already in the manifest (advising 256MB right after a
        // 500k-row table killed the sync helped nobody).
        const known = (cubeManifest().tables || []).map((x) => x.rows || 0);
        const rec = recommendMemoryMB(Math.max(0, Number(t.rowCount) || 0, ...known));
        if (oomCount === 1) {
          // One monster table is not the machine: keep going, the small
          // tables after it deserve their sync ("one bad table must never
          // kill the run"). A SECOND out-of-memory means the machine itself
          // is too small - then stop with one clear message.
          const note = `${t.name} is too big for this machine's memory (${machineMemory().usableMB}MB usable; ${rec}MB or more recommended). Skipping it and continuing with the rest.`;
          log(`  ${note}`);
          onEvent({ type: "table-note", name: t.name, note });
        } else {
          const remaining = chosen.slice(ti + 1).map((x) => x.name);
          const advice = `Sync stopped: out of memory on ${t.name}. This machine has ${machineMemory().usableMB}MB usable; ${rec}MB or more is recommended for this database. Run: fly scale memory ${rec}. Tables already synced are kept; ${remaining.length} table(s) were not attempted.`;
          log(advice);
          onEvent({ type: "fatal", error: advice });
          break;
        }
      }
    }
  }
  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);
  // THE GUARD (2026-08-23 wreck): "not in the include list" is only a
  // deselection when the table's file is VISIBLE to the current login. After
  // a login change, whole files vanish from the schema - their tables fell
  // out of the include list through no choice of the user's, and dropping
  // them destroyed every watermark. A table from an invisible file is left
  // alone: data kept, priors kept, and it rejoins the plan when the file
  // comes back. Only an explicit drop (file removed in Settings, or table
  // deselected while its file is present) removes anything.
  const visible = new Set((tables || []).map((t) => t.name));
  const toDrop = have.filter((n) => !includeNames.includes(n) && visible.has(n));
  const kept = have.filter((n) => !includeNames.includes(n) && !visible.has(n));
  if (kept.length) log(`keeping ${kept.length} table(s) from files this login can't see right now: ${kept.join(", ")}`);
  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);
}