Indexer: SQLite persistence + instant ZMQ hashblock detection

- Persist blocks/items to SQLite (SP_DB; defaults to in-memory) so the indexer
  resumes instead of re-walking the chain on restart — verified resume-from the
  stored height.
- Subscribe to Core's hashblock ZMQ (SP_ZMQ) via a tiny pure-JS ZMTP/3.0 SUB
  client (no native dep, bundle stays self-contained) to trigger a sync the
  instant a block lands — measured ~200ms end-to-end vs the poll interval. Poll
  remains a backstop.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

1 files changed, 92 insertions, 26 deletions

diff --git a/tools/sp-indexer.js b/tools/sp-indexer.js
index bae1e15..9bbedd6 100644
--- a/tools/sp-indexer.js
+++ b/tools/sp-indexer.js
@@ -1,4 +1,4 @@
-// Minimal BIP-352 silent-payment tweak indexer (an "oracle") for regtest dev.
+// BIP-352 silent-payment tweak indexer (an "oracle").
 //
 // It walks the chain from a Bitcoin Core node and, per block, emits one entry per
 // silent-payment-eligible transaction: the tweak (input_hash·A over the tx's
@@ -9,9 +9,12 @@
 // are tracked by the normal Electrum backend; this server only does discovery.
 //
 // Shares the SP crypto with the wallet (../src/silentpay.js) so the tweak math is
-// the same code validated against the BIP-352 vectors. In-memory, regtest-sized;
-// re-indexes on restart. Run with bun.
+// the same code validated against the BIP-352 vectors. Persists to SQLite (set
+// SP_DB to a file to survive restarts; defaults to in-memory). New blocks are
+// pushed to subscribed wallets over WebSocket (/ws). Run with bun.
 
+import { Database } from 'bun:sqlite';
+import net from 'node:net';
 import { inputPubKey, silentPaymentTweak } from '../src/silentpay.js';
 
 const RPC = process.env.CORE_RPC || 'http://127.0.0.1:18443';
@@ -20,6 +23,7 @@ const RPC_PASS = process.env.RPC_PASS || '123';
 const PORT = Number(process.env.SP_PORT || 8888);
 const START = Number(process.env.START_HEIGHT || 0);
 const POLL_MS = Number(process.env.POLL_MS || 2000);
+const DB_PATH = process.env.SP_DB || ':memory:';
 
 const auth = 'Basic ' + Buffer.from(`${RPC_USER}:${RPC_PASS}`).toString('base64');
 let rpcId = 0;
@@ -35,9 +39,19 @@ async function rpc(method, params = []) {
   return j.result;
 }
 
-// height -> { hash, items: [{ txid, tweak(hex), outputs: [{ vout, value(sat), xonly(hex) }] }] }
-const blocks = new Map();
-let indexed = -1; // highest indexed height
+// ---- persistent store (SQLite) --------------------------------------------
+// blocks(height, hash, items-json); meta(indexed). Survives restarts so we don't
+// re-walk the chain — important on mainnet where the initial index is slow.
+const db = new Database(DB_PATH);
+db.exec('PRAGMA journal_mode = WAL');
+db.run('CREATE TABLE IF NOT EXISTS blocks (height INTEGER PRIMARY KEY, hash TEXT, items TEXT)');
+db.run('CREATE TABLE IF NOT EXISTS meta (k TEXT PRIMARY KEY, v INTEGER)');
+const qGetMeta = db.query('SELECT v FROM meta WHERE k = ?');
+const qSetMeta = db.query('INSERT OR REPLACE INTO meta (k, v) VALUES (?, ?)');
+const qPutBlock = db.query('INSERT OR REPLACE INTO blocks (height, hash, items) VALUES (?, ?, ?)');
+const qGetBlock = db.query('SELECT hash, items FROM blocks WHERE height = ?');
+const qScanFrom = db.query('SELECT height, items FROM blocks WHERE height >= ? ORDER BY height');
+let indexed = qGetMeta.get('indexed')?.v ?? (START - 1);
 
 // Is a scriptPubKey (hex) a P2TR output? OP_1 PUSH32 <x-only> = 5120<64 hex>.
 const taprootXonly = (hex) => (hex && hex.length === 68 && hex.startsWith('5120') ? hex.slice(4) : null);
@@ -72,8 +86,10 @@ async function indexBlock(height) {
   const block = await rpc('getblock', [hash, 3]); // verbosity 3: full txs with prevouts
   const items = [];
   for (const tx of block.tx) { const it = scanTx(tx); if (it) items.push(it); }
-  blocks.set(height, { hash, items });
+  qPutBlock.run(height, hash, JSON.stringify(items));
+  qSetMeta.run('indexed', height);
   indexed = height;
+  return items;
 }
 
 let live = false; // true once caught up to the tip — then new blocks are pushed over WS
@@ -82,12 +98,12 @@ async function sync() {
   try {
     const tip = await rpc('getblockcount', []);
     for (let h = Math.max(START, indexed + 1); h <= tip; h++) {
-      await indexBlock(h);
-      // Push each newly-indexed block (once we're live) to subscribed wallets, so
-      // they scan on arrival instead of polling. The payload carries the items so
+      const items = await indexBlock(h);
+      // Push each newly-indexed block (once live) to subscribed wallets, so they
+      // scan on arrival instead of polling. The payload carries the items so
       // there's no follow-up fetch.
-      if (live) server.publish('blocks', JSON.stringify({ type: 'block', height: h, items: blocks.get(h).items }));
-      if (h % 100 === 0 || h === tip) console.log(`indexed height ${h}/${tip} (${blocks.get(h).items.length} sp-tx)`);
+      if (live) server.publish('blocks', JSON.stringify({ type: 'block', height: h, items }));
+      if (h % 100 === 0 || h === tip) console.log(`indexed height ${h}/${tip} (${items.length} sp-tx)`);
     }
     if (!live) { live = true; console.log('caught up — pushing new blocks over websocket'); }
   } catch (e) {
@@ -112,25 +128,20 @@ const server = Bun.serve({
     if (p === '/height') return json({ height: indexed });
     let m;
     if ((m = p.match(/^\/block\/(\d+)$/))) {
-      const h = Number(m[1]);
-      const b = blocks.get(h);
-      if (!b) return json({ error: 'not indexed', height: h, tip: indexed }, 404);
-      return json({ height: h, hash: b.hash, items: b.items });
+      const b = qGetBlock.get(Number(m[1]));
+      if (!b) return json({ error: 'not indexed', height: Number(m[1]), tip: indexed }, 404);
+      return json({ height: Number(m[1]), hash: b.hash, items: JSON.parse(b.items) });
     }
     if ((m = p.match(/^\/tweaks\/(\d+)$/))) {
-      const b = blocks.get(Number(m[1]));
-      return json(b ? b.items.map((i) => i.tweak) : []);
+      const b = qGetBlock.get(Number(m[1]));
+      return json(b ? JSON.parse(b.items).map((i) => i.tweak) : []);
     }
     // All SP-eligible tx items from height `from` to the tip, in one response —
     // each tagged with its height. Lets a client scan a range without one fetch
-    // per block. (Regtest-scale; mainnet would page or use BIP-158 filters.)
+    // per block. (A future BIP-158/Bloom filter will cut this for large ranges.)
     if ((m = p.match(/^\/scan\/(\d+)$/))) {
-      const from = Number(m[1]);
       const items = [];
-      for (let h = from; h <= indexed; h++) {
-        const b = blocks.get(h);
-        if (b) for (const it of b.items) items.push({ height: h, ...it });
-      }
+      for (const row of qScanFrom.all(Number(m[1]))) for (const it of JSON.parse(row.items)) items.push({ height: row.height, ...it });
       return json({ tip: indexed, items });
     }
     return json({ error: 'not found' }, 404);
@@ -142,6 +153,61 @@ const server = Bun.serve({
   },
 });
 
-console.log(`sp-indexer: core=${RPC} start=${START} serving :${PORT}`);
+// Instant block detection via Core's `hashblock` ZMQ — trigger a sync the moment
+// a block arrives, instead of waiting for the poll. A minimal ZMTP/3.0 SUB client
+// (no native dep, so the bundle stays self-contained). The poll below remains a
+// backstop if the socket drops.
+function subscribeHashblock(endpoint, onBlock) {
+  const [host, port] = endpoint.replace(/^tcp:\/\//, '').split(':');
+  const sock = net.connect(Number(port), host);
+  let buf = Buffer.alloc(0);
+  let greeted = false;
+  const sendFrame = (flags, body) => {
+    const head = body.length < 256
+      ? Buffer.from([flags, body.length])
+      : Buffer.concat([Buffer.from([flags | 0x02]), (() => { const b = Buffer.alloc(8); b.writeBigUInt64BE(BigInt(body.length)); return b; })()]);
+    sock.write(Buffer.concat([head, body]));
+  };
+  sock.on('connect', () => {
+    const g = Buffer.alloc(64);
+    g[0] = 0xff; g[9] = 0x7f; g[10] = 3; g[11] = 0; // signature + ZMTP version 3.0
+    g.write('NULL', 12); // security mechanism, null-padded to 20 bytes
+    sock.write(g);
+  });
+  sock.on('data', (d) => {
+    buf = Buffer.concat([buf, d]);
+    if (!greeted) {
+      if (buf.length < 64) return;
+      buf = buf.subarray(64);
+      greeted = true;
+      // READY (command) advertising Socket-Type=SUB, then subscribe to "hashblock".
+      sendFrame(0x04, Buffer.concat([
+        Buffer.from([5]), Buffer.from('READY'),
+        Buffer.from([11]), Buffer.from('Socket-Type'), Buffer.from([0, 0, 0, 3]), Buffer.from('SUB'),
+      ]));
+      sendFrame(0x00, Buffer.concat([Buffer.from([0x01]), Buffer.from('hashblock')]));
+    }
+    while (buf.length >= 2) {
+      const flags = buf[0];
+      const long = flags & 0x02;
+      const headerLen = long ? 9 : 2;
+      if (buf.length < headerLen) break;
+      const size = long ? Number(buf.readBigUInt64BE(1)) : buf[1];
+      if (buf.length < headerLen + size) break;
+      const body = buf.subarray(headerLen, headerLen + size);
+      buf = buf.subarray(headerLen + size);
+      if (!(flags & 0x04) && body.length >= 9 && body.subarray(0, 9).toString() === 'hashblock') onBlock();
+    }
+  });
+  sock.on('error', (e) => console.error('zmq error:', e.message));
+  sock.on('close', () => setTimeout(() => subscribeHashblock(endpoint, onBlock), 3000)); // reconnect
+}
+
+console.log(`sp-indexer: core=${RPC} db=${DB_PATH} start=${START} resume-from=${indexed + 1} serving :${PORT}`);
 await sync();
-setInterval(sync, POLL_MS);
+if (process.env.SP_ZMQ) {
+  let t;
+  subscribeHashblock(process.env.SP_ZMQ, () => { clearTimeout(t); t = setTimeout(sync, 50); }); // debounce a burst
+  console.log(`zmq: subscribed to hashblock at ${process.env.SP_ZMQ}`);
+}
+setInterval(sync, POLL_MS); // backstop (and the only trigger when SP_ZMQ is unset)