diff --git a/src/lib/bns-index-core.js b/src/lib/bns-index-core.js new file mode 100644 index 0000000..16be928 --- /dev/null +++ b/src/lib/bns-index-core.js @@ -0,0 +1,340 @@ +// BNS index core: the one indexer implementation shared by Ariadne's Thread's +// indexer process and (from migration step 3) Theseus's fallback indexer. +// Spec: TheseusNavigator/DESIGN-bns-indexer-service.md, "Ariadne's Thread: the +// indexer". Lifted from TheseusNavigator/bns-indexer.js (Theseus 0.3.70): +// +// warm start local snapshot copies (first that parses), no network +// catch-up the published raw snapshot (dl.silentmode.st, then the Sia +// mirror behind navigate.st): one download, merged, only adds +// delta electrum get_history on the name beacon every 30 s, fetching +// only the transaction bodies not already held +// index rebuilt locally from the beacon transactions with the +// resolver's own reduction (records + owner); TLDs from the +// tlds.bch record +// +// No process or transport lives here: the host decides where the snapshot +// copies go, how the index is published (Electron port, named pipe) and when +// network work is paused (Economy). Node only (fs). +import fs from "node:fs"; +import path from "node:path"; +import { EventEmitter } from "node:events"; + +export const POLL_INTERVAL_MS = 30_000; +// Neither the electrum connect nor its WebSocket handshake has a timeout of +// its own; a server that accepts TCP and then stalls would wedge the loop. +export const POLL_DEADLINE_MS = 45_000; +const ELECTRUM_REFRESH_MS = 30 * 60_000; +const PUBLISHED_REFRESH_MS = 60 * 60_000; + +// ---- snapshot files ---------------------------------------------------------- +// Two copies so a deleted, moved or half-written file still leaves one. +export function snapshotPaths(dir, base = "bns-name-snapshot") { + return { current: path.join(dir, `${base}.json`), previous: path.join(dir, `${base}.prev.json`) }; +} + +export function readSnapshotFile(p) { + try { + if (!p || !fs.existsSync(p)) return null; + const parsed = JSON.parse(fs.readFileSync(p, "utf8").replace(/^/, "")); + if (!parsed || Array.isArray(parsed) || !Array.isArray(parsed.history) || typeof parsed.txs !== "object") return null; + return parsed; + } catch { return null; } +} + +// Atomic as far as readers are concerned: the new state is written to a temp +// file and renamed over the current copy only once complete; the current copy +// becomes the previous one first. Between the two renames the current name is +// briefly absent and the previous copy answers. +export function writeSnapshotCopies(dir, snap, base = "bns-name-snapshot") { + const { current, previous } = snapshotPaths(dir, base); + fs.mkdirSync(dir, { recursive: true }); + const tmp = `${current}.${process.pid}.tmp`; + const fd = fs.openSync(tmp, "w"); + try { fs.writeSync(fd, JSON.stringify(snap)); fs.fsyncSync(fd); } finally { fs.closeSync(fd); } + try { if (fs.existsSync(current)) fs.renameSync(current, previous); } catch { /* previous copy is best-effort */ } + fs.renameSync(tmp, current); + return current; +} + +// Cheap fingerprint: an index rebuilt from the same history at the same +// heights is the same index. +export const sigOf = (snap) => `${snap.history.length}:${snap.history.reduce((a, h) => a + (Number(h.height) || 0), 0)}`; + +// Merge a published snapshot into a base state. Only adds evidence; keeps a +// confirmed height over a mempool one. Returns null when nothing changed or +// the snapshot is for another beacon. +export function mergeSnapshot(base, pub, beacon) { + if (!pub || !Array.isArray(pub.history) || !pub.txs) return null; + if (beacon && pub.beacon && pub.beacon !== beacon) return null; + base = base || { beacon, history: [], txs: {} }; + const merged = new Map(base.history.map((h) => [h.tx_hash, h])); + const txs = { ...(base.txs || {}) }; + let added = 0; + for (const h of pub.history) { + const mine = merged.get(h.tx_hash); + if (!pub.txs[h.tx_hash] && !txs[h.tx_hash]) continue; // no evidence, skip + if (!txs[h.tx_hash]) { txs[h.tx_hash] = pub.txs[h.tx_hash]; added++; } + if (!mine || (!(Number(mine.height) > 0) && Number(h.height) > 0)) merged.set(h.tx_hash, { tx_hash: h.tx_hash, height: h.height }); + } + const next = { ...base, beacon: base.beacon || pub.beacon, history: [...merged.values()], txs }; + if (sigOf(next) === sigOf(base) && added === 0) return null; + return { next, added }; +} + +// Names whose entry differs between two indexes (added, changed or removed). +export function diffIndex(prev, next) { + const changed = []; + if (!prev) return [...next.keys()]; + for (const [k, e] of next) { + const o = prev.get(k); + if (!o || JSON.stringify(o) !== JSON.stringify(e)) changed.push(k); + } + for (const k of prev.keys()) if (!next.has(k)) changed.push(k); + return changed; +} + +const serverKey = (s) => (typeof s === "string" ? s : s && s.url); +function mergeServers(preferred, rest) { + const seen = new Set(), out = []; + for (const s of [...(preferred || []), ...(rest || [])]) { + const k = serverKey(s); + if (k && !seen.has(k)) { seen.add(k); out.push(s); } + } + return out; +} + +// ---- the indexer --------------------------------------------------------------- +// opts: +// resolver the resolver-web.js module (already imported) +// WebSocket WebSocket class for electrum (swap with setWebSocket, e.g. Tor) +// snapshotDir where the two local copies are written (null: never write) +// warmPaths extra read-only snapshots to warm from, in order, after +// the local copies (Ariadne's file for Theseus, a bundled one) +// electrumCacheFile discovered electrum pool (must be as protected as the +// snapshot: a writable pool file chooses the evidence) +// publishedUrls override for NETWORK.nameListUrl / nameListMirrorUrl +// log (...args) => void +// +// Events: "change" {gen, changed, reason, builtAt}, "fresh" {gen, builtAt}, +// "paused" boolean. +export function createIndexer(opts) { + const R = opts.resolver; + const log = opts.log || (() => {}); + const ev = new EventEmitter(); + let WS = opts.WebSocket; + const network = R.NETWORK?.id || "chipnet"; + const beacon = R.BEACON_SCRIPTHASH; + const paths = opts.snapshotDir ? snapshotPaths(opts.snapshotDir) : null; + const publishedUrls = (opts.publishedUrls || [R.NETWORK?.nameListUrl, R.NETWORK?.nameListMirrorUrl]).filter(Boolean); + + const st = { + index: null, // Map name -> entry + snapshot: null, // raw { beacon, history, txs } behind the index + gen: 0, // bumps on every change of the index + builtAt: 0, // last time the state was confirmed against the network + tlds: [], + lastSig: "", + paused: false, + started: false, + lastPollOk: 0, + lastPollError: null, + lastPublished: 0, + warmedFrom: null, + }; + + // ---- electrum pool ---- + let pool = null, lastPoolRefresh = 0; + function initPool() { + let saved = []; + try { if (opts.electrumCacheFile && fs.existsSync(opts.electrumCacheFile)) saved = JSON.parse(fs.readFileSync(opts.electrumCacheFile, "utf8")); } catch {} + pool = mergeServers(Array.isArray(saved) ? saved : [], R.ELECTRUM || R.CHIPNET_ELECTRUM); + } + async function refreshPool() { + if (st.paused) return; + lastPoolRefresh = Date.now(); + try { + const found = await R.fetchElectrumServers({ WebSocket: WS, directIP: true, electrum: pool }); + if (found && found.length) { + pool = mergeServers(found, pool); + if (opts.electrumCacheFile) { try { fs.writeFileSync(opts.electrumCacheFile, JSON.stringify(found, null, 2)); } catch {} } + } + } catch { /* list unpublished or unreachable: keep the pool */ } + } + + // ---- publishing ---- + function adopt(idx, snap, reason, confirmed) { + if (confirmed) st.builtAt = Date.now(); + const sig = snap ? sigOf(snap) : `full:${Date.now()}`; + st.snapshot = snap || st.snapshot; + if (sig === st.lastSig && st.index) { + ev.emit("fresh", { gen: st.gen, builtAt: st.builtAt }); + return false; + } + const changed = diffIndex(st.index, idx); + st.lastSig = sig; + st.index = idx; + st.tlds = typeof R.getBcnrTlds === "function" ? R.getBcnrTlds(idx) : []; + if (snap && paths && reason !== "warm") { + try { writeSnapshotCopies(opts.snapshotDir, { ...snap, savedAt: new Date().toISOString() }); } + catch (e) { log("snapshot write failed:", e.message); } + } + if (!changed.length && reason !== "warm") { ev.emit("fresh", { gen: st.gen, builtAt: st.builtAt }); return false; } + st.gen++; + ev.emit("change", { gen: st.gen, changed, reason, builtAt: st.builtAt }); + return true; + } + + // ---- phase 1: warm ---- + function warm() { + const candidates = []; + if (paths) candidates.push(paths.current, paths.previous); + for (const p of opts.warmPaths || []) if (p) candidates.push(p); + for (const p of candidates) { + const snap = readSnapshotFile(p); + if (!snap || (beacon && snap.beacon && snap.beacon !== beacon)) continue; + try { + const t0 = Date.now(); + const idx = R.buildIndexFromSnapshot({ snapshot: snap }); + st.warmedFrom = p; + adopt(idx, snap, "warm", false); + // A local copy written by this indexer records when it was confirmed. + const saved = Date.parse(snap.savedAt || ""); + if (Number.isFinite(saved)) st.builtAt = saved; + log(`warm-started from ${p}: ${idx.size} names in ${Date.now() - t0} ms`); + return true; + } catch (e) { log(`warm-start from ${p} failed: ${e.message}`); } + } + return false; + } + + // ---- delta poll ---- + let pollInFlight = null; + function poll({ force = false } = {}) { + if (st.paused && !force) return Promise.resolve(false); + if (pollInFlight) return pollInFlight; + let conn = null, deadline; + const work = (async () => { + if (!pool) initPool(); + const snap = st.snapshot || { beacon, history: [], txs: {} }; + const el = conn = await R.connectElectrum({ electrum: pool, WebSocket: WS, directIP: true }); + try { + const fresh = await el.call("blockchain.scripthash.get_history", [beacon]); + const known = new Set(snap.history.map((h) => h.tx_hash)); + const merged = new Map(snap.history.map((h) => [h.tx_hash, h])); + const txs = { ...(snap.txs || {}) }; + let added = 0; + for (const h of fresh) { + if (!known.has(h.tx_hash)) { + try { txs[h.tx_hash] = await el.call("blockchain.transaction.get", [h.tx_hash, true]); added++; } + catch { /* unreadable: the reduction ignores missing txs */ } + } + // keep the fresh height: a mempool tx at snapshot time now has one + merged.set(h.tx_hash, { tx_hash: h.tx_hash, height: h.height }); + } + const next = { ...snap, beacon, history: [...merged.values()], txs }; + adopt(R.buildIndexFromSnapshot({ snapshot: next }), next, "poll", true); + st.lastPollOk = Date.now(); st.lastPollError = null; + if (added) log(`delta: +${added} tx (total ${next.history.length}, ${st.index.size} names)`); + return true; + } finally { try { el.close(); } catch {} } + })().catch((e) => { st.lastPollError = e?.message || String(e); return false; }); + const timeout = new Promise((resolve) => { + deadline = setTimeout(() => { st.lastPollError = "deadline"; try { conn?.close(); } catch {} resolve(false); }, POLL_DEADLINE_MS); + }); + const p = Promise.race([work, timeout]).finally(() => { + clearTimeout(deadline); + if (pollInFlight === p) pollInFlight = null; + if (st.started && !st.paused && Date.now() - lastPoolRefresh > ELECTRUM_REFRESH_MS) refreshPool(); + }); + pollInFlight = p; + return p; + } + + // ---- published snapshot ---- + async function refreshPublished() { + if (st.paused) return false; + st.lastPublished = Date.now(); + let pub = null; + for (const url of publishedUrls) { + try { + const res = await fetch(url, { redirect: "follow", signal: AbortSignal.timeout(15_000) }); + if (!res.ok) { log(`published snapshot ${new URL(url).host}: HTTP ${res.status}`); continue; } + pub = await res.json(); + break; + } catch (e) { log(`published snapshot ${new URL(url).host}: ${e?.cause?.code || e?.message || e}`); } + } + const m = mergeSnapshot(st.snapshot, pub, beacon); + if (!m) return false; + adopt(R.buildIndexFromSnapshot({ snapshot: m.next }), m.next, "published", false); + log(`published snapshot merged: +${m.added} tx (total ${m.next.history.length}, ${st.index.size} names)`); + return true; + } + + // ---- phase 2 ---- + let pollTimer = null, pubTimer = null; + function start() { + if (st.started) return; + st.started = true; + (async () => { + if (!pool) initPool(); + await refreshPublished(); + refreshPool(); + poll(); + pollTimer = setInterval(() => poll(), POLL_INTERVAL_MS); + pubTimer = setInterval(() => { if (Date.now() - st.builtAt > PUBLISHED_REFRESH_MS) refreshPublished(); }, 10 * 60_000); + })().catch((e) => log("start failed:", e?.message)); + } + function stop() { + st.started = false; + if (pollTimer) clearInterval(pollTimer); + if (pubTimer) clearInterval(pubTimer); + pollTimer = pubTimer = null; + } + + // Economy: pause only the network work. The index stays served; resuming + // does one immediate catch-up poll. + function setPaused(on) { + on = !!on; + if (on === st.paused) return; + st.paused = on; + log(on ? "network sync paused (economy)" : "network sync resumed"); + ev.emit("paused", on); + if (!on && st.started) poll(); + } + + async function ready() { + if (st.index) return true; + await refreshPublished(); + if (st.index) return true; + return poll({ force: true }); + } + + // Full electrum walk: only an explicit rebuild request. + async function rebuild() { + if (!pool) initPool(); + const idx = await R.buildIndex({ WebSocket: WS, directIP: true, electrum: pool }); + adopt(idx, null, "rebuild", true); + return true; + } + + // defineProperties, not assign: assign would freeze the getters. + return Object.defineProperties(ev, Object.getOwnPropertyDescriptors({ + warm, start, stop, poll, ready, rebuild, setPaused, refreshPublished, + setWebSocket(W) { WS = W; }, + get index() { return st.index; }, + get gen() { return st.gen; }, + get builtAt() { return st.builtAt; }, + get tlds() { return st.tlds; }, + get paused() { return st.paused; }, + get network() { return network; }, + get snapshotPath() { return paths ? paths.current : null; }, + get(name) { return st.index ? st.index.get(name) ?? null : null; }, + status() { + return { + network, gen: st.gen, builtAt: st.builtAt, size: st.index ? st.index.size : 0, + paused: st.paused, started: st.started, lastPollOk: st.lastPollOk, lastPollError: st.lastPollError, + warmedFrom: st.warmedFrom, history: st.snapshot ? st.snapshot.history.length : 0, tlds: st.tlds, + }; + }, + })); +} diff --git a/src/lib/bns-pipe.js b/src/lib/bns-pipe.js new file mode 100644 index 0000000..9dd0e12 --- /dev/null +++ b/src/lib/bns-pipe.js @@ -0,0 +1,210 @@ +// The indexer's local pipe: \\.\pipe\ariadne-bns (All users) or +// \\.\pipe\ariadne-bns- (Just me). Newline-delimited JSON both ways. +// Spec: TheseusNavigator/DESIGN-bns-indexer-service.md, "The pipe". +// +// Requests {id, op, ...} replies {id, ok, ...} | {id, ok:false, error} +// hello {version, network, gen, builtAt, snapshotPath, paused, pid} +// snapshot {network, gen, builtAt, tlds, names: [[key, entry], ...]} +// get {name} {gen, entry | null} +// subscribe {gen}; then pushes (below) until the client goes +// poll {gen, polled}: one delta poll on demand +// Pushes {event:"change", gen, changed:[names], entries:{name: entry|null}} +// {event:"fresh", gen, builtAt} a poll confirmed the state, nothing changed +// {event:"paused", paused} +// +// Never a TCP port: 127.0.0.1 cannot say who is listening. Even a pipe name +// can be taken while the indexer is down, and Node grants everyone the right +// to add instances (readableAll/writableAll), so clients only trust a +// connection whose server process they have checked (connectVerified). +import net from "node:net"; +import { spawn } from "node:child_process"; +import { EventEmitter } from "node:events"; + +const MAX_REQUEST = 64 * 1024; +const MAX_CLIENTS = 64; +const MAX_PUSH_ENTRIES = 500; +const POLL_MIN_INTERVAL_MS = 5_000; + +// ---- framing ----------------------------------------------------------------- +export function lineReader(onLine, { max = Infinity, onOverflow } = {}) { + let buf = ""; + return (chunk) => { + buf += chunk; + let i; + while ((i = buf.indexOf("\n")) >= 0) { + const line = buf.slice(0, i).trim(); + buf = buf.slice(i + 1); + if (line) onLine(line); + } + if (buf.length > max) { buf = ""; onOverflow?.(); } + }; +} + +// ---- server ------------------------------------------------------------------ +export function createPipeServer({ indexer, pipeName, version, log = () => {} }) { + const subscribers = new Set(); + let lastOnDemandPoll = 0; + const send = (sock, obj) => { try { if (!sock.destroyed) sock.write(JSON.stringify(obj) + "\n"); } catch {} }; + + indexer.on("change", ({ gen, changed }) => { + const push = { event: "change", gen, changed }; + if (changed.length <= MAX_PUSH_ENTRIES) { + push.entries = {}; + for (const n of changed) push.entries[n] = indexer.get(n); + } + for (const s of subscribers) send(s, push); + }); + indexer.on("fresh", ({ gen, builtAt }) => { for (const s of subscribers) send(s, { event: "fresh", gen, builtAt }); }); + indexer.on("paused", (paused) => { for (const s of subscribers) send(s, { event: "paused", paused }); }); + + async function handle(sock, req) { + const reply = (o) => send(sock, { id: req.id, ok: true, ...o }); + switch (req.op) { + case "hello": + return reply({ version, network: indexer.network, gen: indexer.gen, builtAt: indexer.builtAt, snapshotPath: indexer.snapshotPath, paused: indexer.paused, pid: process.pid }); + case "snapshot": + return reply({ network: indexer.network, gen: indexer.gen, builtAt: indexer.builtAt, tlds: indexer.tlds, names: indexer.index ? [...indexer.index] : [] }); + case "get": + return reply({ gen: indexer.gen, entry: indexer.get(String(req.name || "").toLowerCase()) }); + case "subscribe": + subscribers.add(sock); + return reply({ gen: indexer.gen }); + case "poll": { + // Any local process may ask; electrum is not theirs to hammer. + if (indexer.paused) return send(sock, { id: req.id, ok: false, error: "paused" }); + const now = Date.now(); + let polled = false; + if (now - lastOnDemandPoll >= POLL_MIN_INTERVAL_MS) { lastOnDemandPoll = now; polled = await indexer.poll(); } + return reply({ gen: indexer.gen, polled }); + } + default: + return send(sock, { id: req.id, ok: false, error: "unknown op" }); + } + } + + const server = net.createServer((sock) => { + if (server._clients >= MAX_CLIENTS) { sock.destroy(); return; } + server._clients++; + sock.setEncoding("utf8"); + sock.on("data", lineReader((line) => { + let req; + try { req = JSON.parse(line); } catch { return send(sock, { ok: false, error: "bad json" }); } + handle(sock, req || {}).catch((e) => send(sock, { id: req?.id, ok: false, error: e?.message || String(e) })); + }, { max: MAX_REQUEST, onOverflow: () => sock.destroy() })); + sock.once("close", () => { server._clients--; subscribers.delete(sock); }); + sock.on("error", () => {}); + }); + server._clients = 0; + return { + server, + listen() { + return new Promise((resolve, reject) => { + server.once("error", reject); + // Clients run as other accounts (All users) and need write access to + // send requests. See the header: trust comes from the client's check. + server.listen({ path: pipeName, readableAll: true, writableAll: true }, () => { + server.off("error", reject); + log(`pipe: listening on ${pipeName}`); + resolve(); + }); + }); + }, + close() { for (const s of subscribers) s.destroy(); server.close(); }, + }; +} + +// ---- client ------------------------------------------------------------------ +// The server check needs GetNamedPipeServerProcessId on the connected handle, +// which Node cannot reach (a pipe socket has no fd on Windows). The helper +// (Ariadne's tools\ariadne-helper.exe, or the same C# source run through +// PowerShell) connects, checks the server process, prints one verdict line +// and then relays the pipe over its stdin/stdout. So the connection that was +// checked is the connection that is used. +// +// helper: { exe } | { csSource } expect: { scope: "machine"|"user", imagePrefix } +export function spawnPipeHelper({ helper, pipeName, expect }) { + const args = ["pipe", pipeName, expect.scope, expect.imagePrefix || ""]; + if (helper?.exe) return spawn(helper.exe, args, { windowsHide: true, stdio: ["pipe", "pipe", "pipe"] }); + if (helper?.csSource) { + const lit = (s) => "'" + String(s).replace(/'/g, "''") + "'"; + const ps = `$ErrorActionPreference='Stop'; Add-Type -TypeDefinition ([IO.File]::ReadAllText(${lit(helper.csSource)})); exit [AriadneHelper]::Main(@(${args.map(lit).join(",")}))`; + return spawn("powershell.exe", ["-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-Command", ps], { windowsHide: true, stdio: ["pipe", "pipe", "pipe"] }); + } + throw new Error("no pipe helper"); +} + +// Resolves with a connected client, or rejects with the reason the server +// was not trusted (a failed check simply skips to the next source). +// insecureNoCheck: tests and non-Windows only. +export function connectIndexerPipe({ pipeName, helper, expect, insecureNoCheck = false, timeoutMs = 15_000 }) { + return new Promise((resolve, reject) => { + let io, child = null; + if (insecureNoCheck) { + io = net.connect(pipeName); + io.once("connect", () => done(null, { verified: false, insecure: true })); + io.once("error", (e) => done(e)); + } else { + try { child = spawnPipeHelper({ helper, pipeName, expect }); } catch (e) { return reject(e); } + let err = ""; + child.stderr.on("data", (d) => { err += d; }); + child.once("error", (e) => done(e)); + child.once("exit", (code) => done(new Error(`pipe check: helper exited ${code}${err ? ": " + err.trim().split(/\r?\n/).pop() : ""}`))); + let first = ""; + const onFirst = (d) => { + first += d.toString("utf8"); + const i = first.indexOf("\n"); + if (i < 0) return; + child.stdout.off("data", onFirst); + let verdict; + try { verdict = JSON.parse(first.slice(0, i)); } catch { return done(new Error("pipe check: bad verdict")); } + if (!verdict.verified) return done(new Error(`pipe check failed: ${verdict.reason || "unverified server"}`)); + const rest = first.slice(i + 1); + if (rest) setImmediate(() => client._feed(rest)); + done(null, verdict); + }; + child.stdout.on("data", onFirst); + } + const timer = setTimeout(() => done(new Error("pipe connect timeout")), timeoutMs); + let settled = false; + const client = new EventEmitter(); + function done(e, verdict) { + if (settled) return; + settled = true; + clearTimeout(timer); + if (e) { try { child?.kill(); } catch {} try { io?.destroy(); } catch {} return reject(e); } + wire(verdict); + resolve(client); + } + function wire(verdict) { + const out = child ? child.stdin : io; + const inp = child ? child.stdout : io; + let nextId = 1; + const pending = new Map(); + client.verdict = verdict; + client._feed = lineReader((line) => { + let m; try { m = JSON.parse(line); } catch { return; } + if (m.event) return client.emit("push", m); + const p = pending.get(m.id); + if (p) { pending.delete(m.id); m.ok ? p.resolve(m) : p.reject(new Error(m.error || "error")); } + }); + inp.on("data", (d) => client._feed(d.toString("utf8"))); + const closed = () => { + if (client.closed) return; + client.closed = true; + for (const p of pending.values()) p.reject(new Error("pipe closed")); + pending.clear(); + client.emit("close"); + }; + if (child) { child.removeAllListeners("exit"); child.once("exit", closed); child.stdin.on("error", closed); } + else { io.removeAllListeners("error"); io.on("error", closed); io.once("close", closed); } + client.request = (op, extra = {}, { timeoutMs: t = 30_000 } = {}) => new Promise((res, rej) => { + if (client.closed) return rej(new Error("pipe closed")); + const id = nextId++; + const timer = setTimeout(() => { pending.delete(id); rej(new Error(`${op}: timeout`)); }, t); + pending.set(id, { resolve: (m) => { clearTimeout(timer); res(m); }, reject: (e) => { clearTimeout(timer); rej(e); } }); + try { out.write(JSON.stringify({ id, op, ...extra }) + "\n"); } catch (e) { pending.delete(id); clearTimeout(timer); rej(e); } + }); + client.close = () => { try { child ? child.kill() : io.destroy(); } catch {} closed(); }; + } + }); +} diff --git a/src/lib/bns-source-chain.js b/src/lib/bns-source-chain.js new file mode 100644 index 0000000..9d2544f --- /dev/null +++ b/src/lib/bns-source-chain.js @@ -0,0 +1,296 @@ +// 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 /api/name/, 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, + }; + }, + })); +}