argus/src/lib/bns-source-chain.js

// The source chain: how a reader (Ariadne's resolver, and Theseus from
// migration step 3) gets BNS names without depending on the indexer.
// Spec: TheseusNavigator/DESIGN-bns-indexer-service.md, "The source chain".
//
//   1. memory              whatever is loaded (normally all of it)
//   2. the live indexer    over the pipe, server-checked: snapshot, then pushes
//   3. local copies        the indexer's two files + this reader's own copy
//   4. public snapshots    one download, rebuilt locally; background only,
//                          when the local data is older than an hour
//   5. public BNS indexer  GET <gateway>/api/name/<n>, single names, only when
//                          there is no data yet or it is stale; PROVISIONAL:
//                          served at once, replaced when verified data
//                          arrives, never good enough for publisher checks
//
// A lookup never waits on 2-4; they refresh memory in the background.
import fs from "node:fs";
import path from "node:path";
import { EventEmitter } from "node:events";
import { readSnapshotFile, diffIndex } from "./bns-index-core.js";

const PROVISIONAL_TTL_MS = 30_000;
const NEGATIVE_TTL_MS = 30_000;

// Index-format file: what a reader keeps as its own copy, since over the pipe
// it receives the index, not the raw evidence.
function readIndexFile(p) {
  try {
    if (!p || !fs.existsSync(p)) return null;
    const j = JSON.parse(fs.readFileSync(p, "utf8"));
    if (j && j.format === "bns-index" && Array.isArray(j.names)) return j;
  } catch {}
  return null;
}

// opts:
//   resolver          resolver-web.js module
//   connectPipe       async () => client (bns-pipe.js connectIndexerPipe), or null
//   localPaths        the indexer's copies (current, previous), in order
//   statePath         the indexer's small state file ({builtAt, ...}), optional
//   ownCopyPath       this reader's last-known-good copy (index format)
//   publicSnapshotUrls, publicIndexerUrl
//   staleMs           local data older than this counts as stale (default 5 min)
//   log
// Events: "change" {changed, source}, "mismatch" {name, provisional, verified},
//         "pipe" {connected, error?}
export function createSourceChain(opts) {
  const R = opts.resolver;
  const log = opts.log || (() => {});
  const ev = new EventEmitter();
  const staleMs = opts.staleMs ?? 5 * 60_000;
  const publicRefreshMs = opts.publicRefreshMs ?? 60 * 60_000;
  const publicSnapshotUrls = (opts.publicSnapshotUrls || [R.NETWORK?.nameListUrl, R.NETWORK?.nameListMirrorUrl]).filter(Boolean);
  const publicIndexerUrl = opts.publicIndexerUrl ?? "https://navigate.st";

  const st = {
    index: new Map(), source: null, builtAt: 0, gen: 0, tlds: [], historyLen: 0,
    fileMtimes: {}, pipe: null, pipeError: null, pipeSince: 0, lastPush: 0,
    lastPublic: 0, lastOwnWrite: 0,
  };
  const provisional = new Map(); // key -> { at, entry|null }

  function adopt(next, source, { builtAt, gen, tlds, historyLen } = {}) {
    const changed = diffIndex(st.index, next);
    st.index = next;
    st.source = source;
    if (builtAt != null) st.builtAt = builtAt;
    if (gen != null) st.gen = gen;
    st.tlds = tlds || (typeof R.getBcnrTlds === "function" ? R.getBcnrTlds(next) : st.tlds);
    if (historyLen != null) st.historyLen = historyLen;
    checkProvisional(changed);
    if (changed.length) ev.emit("change", { changed, source });
    return changed;
  }

  // Verified data arrived: drop provisional answers it covers, say so when
  // they differ (the reader replaces tabs / cached answers).
  function checkProvisional(changed) {
    for (const [k, p] of provisional) {
      const v = st.index.get(k);
      if (!v) continue;
      provisional.delete(k);
      if (p.entry && JSON.stringify(strip(p.entry)) !== JSON.stringify(strip(v))) ev.emit("mismatch", { name: k, provisional: p.entry, verified: v });
    }
    void changed;
  }
  const strip = (e) => ({ records: e.records || {}, owner: e.owner ?? null, category: e.category ?? null });

  // ---- 3: local copies ----
  function readState() {
    try { return opts.statePath ? JSON.parse(fs.readFileSync(opts.statePath, "utf8")) : null; } catch { return null; }
  }
  function loadLocal(reason = "start") {
    for (const p of opts.localPaths || []) {
      const snap = readSnapshotFile(p);
      if (!snap) continue;
      try {
        const mtime = fs.statSync(p).mtimeMs;
        st.fileMtimes[p] = mtime;
        if (snap.history.length < st.historyLen && st.source !== null && reason !== "start") return false;
        const idx = R.buildIndexFromSnapshot({ snapshot: snap });
        const state = readState();
        const saved = Date.parse(snap.savedAt || "") || mtime;
        const builtAt = Math.max(saved, Number(state?.builtAt) || 0);
        adopt(idx, `local:${path.basename(p)}`, { builtAt, historyLen: snap.history.length, gen: Number(state?.gen) || st.gen });
        return true;
      } catch (e) { log(`source chain: ${p} unusable: ${e.message}`); }
    }
    const own = readIndexFile(opts.ownCopyPath);
    if (own && own.names.length >= st.index.size) {
      adopt(new Map(own.names), "own-copy", { builtAt: own.builtAt || 0, gen: own.gen, tlds: own.tlds });
      return true;
    }
    return false;
  }
  // Only reload when a file actually changed (one stat per file per tick).
  function localChanged() {
    for (const p of opts.localPaths || []) {
      try { const m = fs.statSync(p).mtimeMs; if (m !== st.fileMtimes[p]) return true; } catch {}
    }
    return false;
  }

  function writeOwnCopy() {
    if (!opts.ownCopyPath || Date.now() - st.lastOwnWrite < 60_000) return;
    st.lastOwnWrite = Date.now();
    try {
      fs.mkdirSync(path.dirname(opts.ownCopyPath), { recursive: true });
      const tmp = opts.ownCopyPath + ".tmp";
      fs.writeFileSync(tmp, JSON.stringify({ format: "bns-index", network: R.NETWORK?.id, gen: st.gen, builtAt: st.builtAt, tlds: st.tlds, names: [...st.index] }));
      fs.renameSync(tmp, opts.ownCopyPath);
    } catch (e) { log("source chain: own copy not written:", e.message); }
  }

  // ---- 2: the live indexer ----
  let backoff = 5_000, reconnectTimer = null, stopped = false;
  async function connect() {
    reconnectTimer = null;
    if (stopped || !opts.connectPipe || st.pipe) return;
    let c;
    try {
      c = await opts.connectPipe();
      const hello = await c.request("hello");
      if (hello.network && R.NETWORK?.id && hello.network !== R.NETWORK.id) throw new Error(`indexer is on ${hello.network}`);
      const snap = await c.request("snapshot", {}, { timeoutMs: 60_000 });
      st.pipe = c; st.pipeError = null; st.pipeSince = st.lastPush = Date.now(); backoff = 5_000;
      adopt(new Map(snap.names), "pipe", { builtAt: snap.builtAt, gen: snap.gen, tlds: snap.tlds });
      writeOwnCopy();
      c.on("push", onPush);
      c.once("close", () => {
        if (st.pipe !== c) return;
        st.pipe = null;
        st.pipeError = "closed";
        log("source chain: indexer pipe closed, reading local copies");
        ev.emit("pipe", { connected: false, error: "closed" });
        loadLocal("pipe-closed");
        schedule();
      });
      await c.request("subscribe");
      log(`source chain: live indexer (gen ${snap.gen}, ${snap.names.length} names)`);
      ev.emit("pipe", { connected: true, verdict: c.verdict });
    } catch (e) {
      st.pipeError = e?.message || String(e);
      try { c?.close(); } catch {}
      if (st.pipe === c) st.pipe = null;
      ev.emit("pipe", { connected: false, error: st.pipeError });
      schedule();
    }
  }
  function schedule() {
    if (stopped || reconnectTimer || !opts.connectPipe) return;
    reconnectTimer = setTimeout(connect, backoff);
    backoff = Math.min(backoff * 2, 60_000);
  }
  async function onPush(m) {
    st.lastPush = Date.now();
    if (m.event === "fresh") { st.builtAt = m.builtAt || st.builtAt; return; }
    if (m.event !== "change") return;
    if (m.gen <= st.gen) return;
    if (m.entries && m.gen === st.gen + 1) {
      const next = new Map(st.index);
      for (const n of m.changed) { const e = m.entries[n]; e ? next.set(n, e) : next.delete(n); }
      adopt(next, "pipe", { gen: m.gen, builtAt: Date.now() });
    } else {
      // missed a generation, or too many changes for a push: take it all
      try {
        const snap = await st.pipe.request("snapshot", {}, { timeoutMs: 60_000 });
        adopt(new Map(snap.names), "pipe", { builtAt: snap.builtAt, gen: snap.gen, tlds: snap.tlds });
      } catch {}
    }
    writeOwnCopy();
  }

  // ---- 4: public snapshots ----
  async function refreshPublic() {
    st.lastPublic = Date.now();
    for (const url of publicSnapshotUrls) {
      try {
        const res = await fetch(url, { redirect: "follow", signal: AbortSignal.timeout(20_000) });
        if (!res.ok) continue;
        const snap = await res.json();
        if (!snap || !Array.isArray(snap.history) || (R.BEACON_SCRIPTHASH && snap.beacon && snap.beacon !== R.BEACON_SCRIPTHASH)) continue;
        if (st.pipe) return false; // the live indexer took over meanwhile
        if (snap.history.length < st.historyLen) return false;
        adopt(R.buildIndexFromSnapshot({ snapshot: snap }), `public:${new URL(url).host}`, { builtAt: Date.now(), historyLen: snap.history.length });
        writeOwnCopy();
        log(`source chain: public snapshot from ${new URL(url).host} (${st.index.size} names)`);
        return true;
      } catch (e) { log(`source chain: public snapshot ${url}: ${e?.cause?.code || e?.message}`); }
    }
    return false;
  }

  // ---- 5: public BNS indexer (provisional) ----
  function stale() {
    if (st.pipe) return false;
    return !st.index.size || Date.now() - st.builtAt > staleMs;
  }
  async function fetchProvisional(key) {
    const c = provisional.get(key);
    if (c?.pending) return c.pending;
    if (c && Date.now() - c.at < (c.entry ? PROVISIONAL_TTL_MS : NEGATIVE_TTL_MS)) return c.entry;
    const pending = (async () => {
      try {
        const r = await fetch(`${publicIndexerUrl}/api/name/${encodeURIComponent(key)}`, { signal: AbortSignal.timeout(5_000), headers: { "user-agent": "ariadne-source-chain/1" } });
        if (!r.ok) return null;
        const j = await r.json();
        if (!j || j.registered === false || !j.records) return null;
        return { name: j.name || key, records: j.records, category: j.category, txid: j.txid, height: j.height, owner: j.owner ?? null };
      } catch { return null; }
    })();
    provisional.set(key, { at: Date.now(), entry: null, pending });
    const entry = await pending;
    provisional.set(key, { at: Date.now(), entry, pending: null });
    return entry;
  }

  // Memory only, synchronous: for hot paths (DNS, SNI).
  function lookup(key) {
    const e = st.index.get(key);
    if (e) return { entry: e, provisional: false, source: st.source };
    const p = provisional.get(key);
    if (p?.entry && Date.now() - p.at < PROVISIONAL_TTL_MS) return { entry: p.entry, provisional: true, source: "public-indexer" };
    return null;
  }
  // Memory, then the public indexer when the data is missing or stale.
  // { verifiedOnly: true } never answers provisionally (publisher checks).
  async function resolve(key, { verifiedOnly = false } = {}) {
    const hit = lookup(key);
    if (hit && (!hit.provisional || !verifiedOnly)) return hit;
    if (verifiedOnly || !stale()) return null;
    const entry = await fetchProvisional(key);
    return entry ? { entry, provisional: true, source: "public-indexer" } : null;
  }

  let tick = null, beat = null;
  function start() {
    loadLocal("start");
    connect();
    tick = setInterval(() => {
      if (!st.pipe && localChanged()) loadLocal("file-changed");
      if (!st.pipe && Date.now() - st.builtAt > publicRefreshMs && Date.now() - st.lastPublic > publicRefreshMs) refreshPublic();
    }, 10_000);
    // A silent pipe is a wedged indexer: the indexer pushes "fresh" after
    // every 30 s poll, so minutes of silence means ask, and drop it if mute.
    beat = setInterval(async () => {
      const c = st.pipe;
      if (!c || Date.now() - st.lastPush < 3 * 60_000) return;
      try { const h = await c.request("hello", {}, { timeoutMs: 10_000 }); st.lastPush = Date.now(); if (h.gen > st.gen) onPush({ event: "change", gen: h.gen, changed: [] }); }
      catch { c.close(); }
    }, 60_000);
    // No data at all yet: the public snapshot now, in the background.
    if (!st.index.size) setTimeout(() => { if (!st.pipe) refreshPublic(); }, opts.connectPipe ? 3_000 : 0);
    tick.unref?.(); beat.unref?.();
  }
  function stop() {
    stopped = true;
    clearInterval(tick); clearInterval(beat); clearTimeout(reconnectTimer);
    try { st.pipe?.close(); } catch {}
  }

  // defineProperties, not assign: assign would freeze the getters.
  return Object.defineProperties(ev, Object.getOwnPropertyDescriptors({
    start, stop, lookup, resolve, refreshPublic, loadLocal,
    get index() { return st.index; },
    get tlds() { return st.tlds; },
    status() {
      return {
        source: st.source, size: st.index.size, builtAt: st.builtAt, gen: st.gen, stale: stale(),
        pipe: { connected: !!st.pipe, since: st.pipe ? st.pipeSince : null, error: st.pipeError, verdict: st.pipe?.verdict || null },
        localPaths: opts.localPaths || [], ownCopyPath: opts.ownCopyPath || null,
        provisional: [...provisional.values()].filter((p) => p.entry).length,
        lastPublic: st.lastPublic,
      };
    },
  }));
}