Realtime via self-hosted ZMQ watcher (Electrum-over-WS)

Replace the unreliable public WebSocket push (mempool/blockchain.info address
subscriptions never reliably fire) with a private notification service:

- tools/zmq-watcher.js: a standalone Bun service (reusing coinos lib/zmq.ts's
  ZMTP subscriber) that reads Bitcoin Core's ZMQ rawtx/rawblock, computes the
  scripthash of each output, and pushes a blockchain.scripthash.subscribe
  notification to subscribed clients. No historical index (~24MB RAM) — just the
  live tx stream. Runs on the coinos box reading its Core's ZMQ.
- wallet.js: realtime client now speaks the Electrum protocol (server.version,
  scripthash.subscribe, server.ping) over the WS; any scripthash notification
  triggers the existing refreshLive() (REST stays the source of truth).
- api.js: wsUrl -> wss://coinos.io/electrum.

Verified end-to-end: live mempool txs to subscribed addresses notify instantly.

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

3 files changed, 350 insertions, 108 deletions

diff --git a/src/api.js b/src/api.js
index 08f03b2..20041e2 100644
--- a/src/api.js
+++ b/src/api.js
@@ -45,11 +45,24 @@ function resolveHosts(net) {
   return [host('https://mempool.space', 'mempool'), host('https://blockstream.info', 'esplora')];
 }
 
-// WebSocket URL for live updates — only mempool.space exposes one, so other
-// explorers fall back to polling (returns null → no socket).
+// Realtime is a private Fulcrum (Electrum protocol over WebSocket-Secure). It
+// reliably pushes a scripthash-status notification the moment a tx touches a
+// watched address — unlike the public Esplora/blockchain.info sockets, whose
+// address push doesn't fire. Independent of the REST explorer choice. Mainnet
+// only; testnet falls back to polling (null). Overridable for staging/testing.
+const ELECTRUM_KEY = 'btc-wallet-electrum';
+export function getElectrumWsUrl() {
+  try {
+    const u = localStorage.getItem(ELECTRUM_KEY);
+    if (u) return u;
+  } catch {}
+  // Served as a path under coinos.io (a dedicated subdomain hit a Cloudflare
+  // WebSocket quirk; the main hostname proxies WS cleanly).
+  return 'wss://coinos.io/electrum';
+}
 export function wsUrl(net) {
-  if (getExplorerConfig().server !== 'mempool') return null;
-  return 'wss://mempool.space' + (net === 'testnet' ? '/testnet/api/v1/ws' : '/api/v1/ws');
+  if (net === 'testnet') return null;
+  return getElectrumWsUrl();
 }
 
 const REQUEST_TIMEOUT_MS = 10000;
diff --git a/src/wallet.js b/src/wallet.js
index 569c0bd..4cfa36b 100644
--- a/src/wallet.js
+++ b/src/wallet.js
@@ -1161,19 +1161,37 @@ export class Wallet {
 
   // What realtime watches: the fresh frontier (next receive + next change), plus
   // any address holding an unconfirmed coin — so a pending deposit's confirmation
-  // pushes instantly instead of waiting for a poll. Capped at the socket's 10-
-  // address-per-connection limit (frontier first, then pending coins).
-  watchedAddresses() {
-    const set = new Set([
-      this.derive(0, this.nextReceiveIndex).address,
-      this.derive(1, this.nextChangeIndex).address,
-    ]);
+  // pushes instantly instead of waiting for a poll. Returns derivation targets
+  // ({chain,index}); Electrum/Fulcrum handles many subscriptions, but our watched
+  // set is naturally small so we cap generously.
+  _watchedTargets() {
+    const seen = new Set();
+    const out = [];
+    const add = (chain, index) => {
+      const k = chain + '/' + index;
+      if (!seen.has(k)) { seen.add(k); out.push({ chain, index }); }
+    };
+    add(0, this.nextReceiveIndex);
+    add(1, this.nextChangeIndex);
     const stuck = this._stuckTxids();
     for (const u of this.utxos) {
-      if (set.size >= 10) break;
-      if (!u.confirmed && !stuck.has(u.txid)) set.add(u.address);
+      if (out.length >= 25) break;
+      if (!u.confirmed && !stuck.has(u.txid)) add(u.chain, u.index);
     }
-    return [...set];
+    return out;
+  }
+  watchedAddresses() {
+    return this._watchedTargets().map((t) => this.derive(t.chain, t.index).address);
+  }
+  // Electrum scripthash for one of our addresses: sha256(scriptPubKey), reversed,
+  // hex. This is what Fulcrum subscribes on.
+  _scripthash(chain, index) {
+    const h = sha256(this.derive(chain, index).script);
+    h.reverse();
+    return hex.encode(h);
+  }
+  watchedScripthashes() {
+    return this._watchedTargets().map((t) => this._scripthash(t.chain, t.index));
   }
 
   startRealtime() {
@@ -1220,9 +1238,7 @@ export class Wallet {
       if (Date.now() - this._lastMsg > 45000) {
         this._reconnectNow();
       } else {
-        try {
-          this._ws.send(JSON.stringify({ action: 'ping' }));
-        } catch {}
+        this._rpcSend('server.ping', []);
       }
     }, 15000);
 
@@ -1266,34 +1282,36 @@ export class Wallet {
       return;
     }
     this._ws = ws;
+    this._wsBuf = '';            // partial-line accumulator (newline-delimited JSON-RPC)
+    this._subscribed = new Set(); // scripthashes subscribed on this connection
+    this._rpcId = 0;
     ws.onopen = () => {
       this.live = true;
       this._wsBackoff = 0; // healthy again — reset the reconnect backoff
       this._lastMsg = Date.now();
+      // Electrum handshake, then subscribe our watched scripthashes.
+      this._rpcSend('server.version', ['halwallet', '1.4']);
       this.retrack();
       this.emit();
-      // Catch any deposit that landed while we were disconnected — mempool only
-      // pushes new activity after tracking starts, not what's already there.
+      // The server only pushes activity AFTER we subscribe, so reconcile once on
+      // open to catch anything that landed while we were disconnected.
       this.refreshLive().catch(() => {});
     };
     ws.onmessage = (ev) => {
       this._lastMsg = Date.now();
-      let msg;
-      try {
-        msg = JSON.parse(typeof ev.data === 'string' ? ev.data : '');
-      } catch {
-        return;
+      this._wsBuf += (typeof ev.data === 'string' ? ev.data : '');
+      // Electrum frames are newline-delimited JSON-RPC; parse each complete line.
+      let nl;
+      while ((nl = this._wsBuf.indexOf('\n')) >= 0) {
+        const line = this._wsBuf.slice(0, nl).trim();
+        this._wsBuf = this._wsBuf.slice(nl + 1);
+        if (line) this._handleRpc(line);
       }
-      // React only to address/block events — never the global tx firehose.
-      if (
-        msg['multi-address-transactions'] ||
-        msg['address-transactions'] ||
-        msg['block-transactions'] ||
-        msg['txConfirmed']
-      ) {
-        // Update instantly from the pushed tx data, then reconcile via a scan.
-        this._ingestWs(msg);
-        this._scheduleRefresh();
+      // Fallback: a server that sends one complete JSON per frame without a
+      // trailing newline. Only consume the buffer if it parses cleanly.
+      const rest = this._wsBuf.trim();
+      if (rest) {
+        try { JSON.parse(rest); this._handleRpc(rest); this._wsBuf = ''; } catch {}
       }
     };
     ws.onerror = () => {
@@ -1309,85 +1327,39 @@ export class Wallet {
     };
   }
 
-  retrack() {
-    if (this._ws && this._ws.readyState === 1) {
-      try {
-        this._ws.send(JSON.stringify({ 'track-addresses': this.watchedAddresses() }));
-      } catch {}
-    }
+  // Send an Electrum JSON-RPC request (fire-and-forget; we react to notifications,
+  // not responses). Newline-terminated per the protocol.
+  _rpcSend(method, params) {
+    if (!this._ws || this._ws.readyState !== 1) return;
+    const id = ++this._rpcId;
+    try {
+      this._ws.send(JSON.stringify({ id, method, params }) + '\n');
+    } catch {}
+    return id;
   }
 
-  // Optimistically apply transactions from a WS push so an incoming deposit
-  // shows the instant mempool.space pushes it — no extra round-trip. The
-  // debounced scan that follows reconciles everything (spends, confirmations).
-  _ingestWs(msg) {
-    const txs = [];
-    const multi = msg['multi-address-transactions'];
-    if (multi && typeof multi === 'object') {
-      for (const addr of Object.keys(multi)) {
-        const g = multi[addr] || {};
-        for (const t of g.mempool || []) txs.push(t);
-        for (const t of g.confirmed || []) txs.push(t);
-      }
+  // A scripthash-status notification means a tx touched one of our addresses (a
+  // deposit, or a confirmation). We don't trust the pushed status payload — it's
+  // just the trigger; reconcile authoritative state over REST.
+  _handleRpc(line) {
+    let msg;
+    try { msg = JSON.parse(line); } catch { return; }
+    if (msg && msg.method === 'blockchain.scripthash.subscribe') {
+      this._scheduleRefresh();
     }
-    const single = msg['address-transactions'];
-    if (Array.isArray(single)) txs.push(...single);
+  }
 
-    let changed = false;
-    const have = new Set(this.utxos.map(utxoId));
-    const mine = new Set(this.addrMap.keys());
-    for (const tx of txs) {
-      if (!tx || !Array.isArray(tx.vout)) continue;
-      let relevant = false;
-      tx.vout.forEach((vo, i) => {
-        const a = vo && vo.scriptpubkey_address;
-        if (!a || !this.addrMap.has(a)) return;
-        relevant = true;
-        const id = `${tx.txid}:${i}`;
-        if (have.has(id)) return;
-        const p = this.addrMap.get(a);
-        const confirmed = !!(tx.status && tx.status.confirmed);
-        this.utxos.push({ txid: tx.txid, vout: i, value: vo.value, address: a, chain: p.chain, index: p.index, confirmed });
-        have.add(id);
-        // Mark the receiving address used + record its balance, so the fresh
-        // index advances immediately (next address/QR shows right away).
-        const arr = p.chain === 0 ? this.receive : this.change;
-        let entry = arr.find((e) => e.index === p.index);
-        if (!entry) {
-          entry = { chain: p.chain, index: p.index, address: a, used: false, confirmed: 0, pending: 0 };
-          arr.push(entry);
-        }
-        entry.used = true;
-        if (confirmed) entry.confirmed = (entry.confirmed || 0) + vo.value;
-        else entry.pending = (entry.pending || 0) + vo.value;
-        changed = true;
-      });
-      for (const vin of tx.vin || []) {
-        const a = vin.prevout && vin.prevout.scriptpubkey_address;
-        if (a && mine.has(a)) relevant = true;
-      }
-      // Record it in history — or refresh it, so a pending tx flips to confirmed.
-      if (relevant) {
-        const idx = this.txs.findIndex((t) => t.txid === tx.txid);
-        const summary = this._txSummary(tx);
-        if (idx < 0) {
-          this.txs.push(summary);
-          changed = true;
-        } else if (this.txs[idx].confirmed !== summary.confirmed) {
-          this.txs[idx] = summary;
-          changed = true;
-        }
-      }
-    }
-    if (changed) {
-      this.nextReceiveIndex = firstUnused(this.receive);
-      this.nextChangeIndex = firstUnused(this.change);
-      this._recomputeBalanceFromChains();
-      this._sortTxs();
-      this.retrack(); // start watching the new fresh address
-      this.emit();
+  // Subscribe any watched scripthashes not yet subscribed on this connection.
+  // Reset on reconnect (_subscribed is cleared in _connectWs), so a fresh socket
+  // re-subscribes the full set.
+  retrack() {
+    if (!this._ws || this._ws.readyState !== 1) return;
+    if (!this._subscribed) this._subscribed = new Set();
+    for (const sh of this.watchedScripthashes()) {
+      if (this._subscribed.has(sh)) continue;
+      this._subscribed.add(sh);
+      this._rpcSend('blockchain.scripthash.subscribe', [sh]);
     }
-    return changed;
   }
 
   _scheduleReconnect() {
diff --git a/tools/zmq-watcher.js b/tools/zmq-watcher.js
new file mode 100644
index 0000000..55cb399
--- /dev/null
+++ b/tools/zmq-watcher.js
@@ -0,0 +1,257 @@
+// Standalone ZMQ-backed address-notification service for halwallet.
+//
+// Subscribes to Bitcoin Core's ZMQ rawtx/rawblock streams and pushes a
+// `blockchain.scripthash.subscribe` notification to any WebSocket client that
+// subscribed to a scripthash a new tx (mempool or block) pays. No historical
+// index — just the live tx firehose plus an in-memory map of subscriptions. It
+// speaks just enough of the Electrum protocol that the halwallet realtime client
+// works against it unchanged (it treats any scripthash notification as "refresh
+// authoritative state over REST"). A drop-in stand-in for Fulcrum's push.
+//
+// The ZMTP 3.0 (NULL) subscriber is adapted from coinos-server lib/zmq.ts, which
+// implements the wire protocol directly over a TCP socket (no native zeromq dep).
+
+import net from 'node:net';
+import { Transaction } from '@scure/btc-signer';
+import { concatBytes } from '@scure/btc-signer/utils.js';
+import { sha256 } from '@noble/hashes/sha256';
+import { bytesToHex } from '@noble/hashes/utils.js';
+
+const BC_HOST = process.env.BC_HOST || 'bc';
+const RAWTX_PORT = Number(process.env.RAWTX_PORT || 18503);
+const RAWBLOCK_PORT = Number(process.env.RAWBLOCK_PORT || 18502);
+const WS_PORT = Number(process.env.WS_PORT || 50010);
+
+const enc = new TextEncoder();
+const dec = new TextDecoder();
+
+// scripthash (hex) -> Set<ws> of clients watching it.
+const subs = new Map();
+
+// Electrum scripthash: sha256(scriptPubKey), reversed, hex.
+const scripthashOf = (script) => {
+  const h = sha256(script);
+  h.reverse();
+  return bytesToHex(h);
+};
+
+// ---- ZMTP 3.0 NULL SUB (adapted from coinos lib/zmq.ts) -------------------
+const greeting = () => {
+  const b = new Uint8Array(64);
+  b[0] = 0xff; b[9] = 0x7f; b[10] = 3; b[11] = 0;
+  b.set(enc.encode('NULL'), 12);
+  return b;
+};
+const frame = (flags, body) => {
+  if (body.length <= 0xff) {
+    const b = new Uint8Array(2 + body.length);
+    b[0] = flags; b[1] = body.length; b.set(body, 2);
+    return b;
+  }
+  const b = new Uint8Array(9 + body.length);
+  b[0] = flags | 0x02;
+  new DataView(b.buffer).setBigUint64(1, BigInt(body.length));
+  b.set(body, 9);
+  return b;
+};
+const command = (name, props = {}) => {
+  const parts = [Uint8Array.of(name.length), enc.encode(name)];
+  for (const [k, v] of Object.entries(props)) {
+    const kb = enc.encode(k), vb = enc.encode(v);
+    parts.push(Uint8Array.of(kb.length), kb);
+    const len = new Uint8Array(4);
+    new DataView(len.buffer).setUint32(0, vb.length);
+    parts.push(len, vb);
+  }
+  return frame(0x04, concatBytes(...parts));
+};
+const subscribeCommand = (topic) => {
+  const t = enc.encode(topic);
+  return frame(0x04, concatBytes(Uint8Array.of(9), enc.encode('SUBSCRIBE'), Uint8Array.of(t.length), t));
+};
+const legacySubscribe = (topic) => frame(0x00, concatBytes(Uint8Array.of(1), enc.encode(topic)));
+
+const startSub = (host, port, topic, onMessage) => new Promise((resolve) => {
+  const socket = net.connect({ host, port });
+  let buffer = new Uint8Array(0);
+  let handshakeDone = false;
+  let frames = [];
+  let closed = false;
+  const done = () => { if (!closed) { closed = true; resolve(); } };
+
+  const processBuffer = () => {
+    if (!handshakeDone) {
+      if (buffer.length < 64) return;
+      buffer = buffer.slice(64);
+      handshakeDone = true;
+      socket.write(command('READY', { 'Socket-Type': 'SUB' }));
+      socket.write(subscribeCommand(topic));
+      socket.write(legacySubscribe(topic));
+      console.log('zmq subscribed', topic, host + ':' + port);
+    }
+    while (buffer.length >= 2) {
+      const flags = buffer[0];
+      let size, header;
+      if (flags & 0x02) {
+        if (buffer.length < 9) return;
+        const v = new DataView(buffer.buffer, buffer.byteOffset, buffer.byteLength);
+        size = Number(v.getBigUint64(1)); header = 9;
+      } else {
+        size = buffer[1]; header = 2;
+      }
+      if (buffer.length < header + size) return;
+      const body = buffer.slice(header, header + size);
+      buffer = buffer.slice(header + size);
+      if (flags & 0x04) continue; // command frame (e.g. READY)
+      frames.push(body);
+      if (flags & 0x01) continue; // more frames follow
+      if (frames.length >= 2 && dec.decode(frames[0]) === topic) onMessage(frames[1]);
+      frames = [];
+    }
+  };
+
+  socket.on('connect', () => socket.write(greeting()));
+  socket.on('data', (d) => {
+    buffer = concatBytes(buffer, new Uint8Array(d));
+    try { processBuffer(); } catch (e) { console.error('zmq parse error', e.message); }
+  });
+  socket.on('error', done);
+  socket.on('close', done);
+});
+
+const TX_OPTS = { allowUnknownOutputs: true, allowUnknownInputs: true, disableScriptCheck: true };
+
+// Push a notification to every client watching a scripthash this tx pays to.
+const notifyForTx = (raw) => {
+  let tx;
+  try { tx = Transaction.fromRaw(raw, TX_OPTS); } catch { return; }
+  const txid = tx.id;
+  const hit = new Set();
+  for (let i = 0; i < tx.outputsLength; i++) {
+    let script;
+    try { script = tx.getOutput(i).script; } catch { continue; }
+    if (!script) continue;
+    const sh = scripthashOf(script);
+    if (subs.has(sh)) hit.add(sh);
+  }
+  for (const sh of hit) {
+    const set = subs.get(sh);
+    if (!set) continue;
+    // Status param is the txid — the client ignores its value and just refreshes.
+    const note = JSON.stringify({ jsonrpc: '2.0', method: 'blockchain.scripthash.subscribe', params: [sh, txid] }) + '\n';
+    for (const ws of set) { try { ws.send(note); } catch {} }
+  }
+};
+
+// Block parsing (varint + tx-size walk) — also from coinos lib/zmq.ts.
+const readVarInt = (data, offset) => {
+  const first = data[offset];
+  if (first < 0xfd) return { value: first, size: 1 };
+  const view = new DataView(data.buffer, data.byteOffset, data.byteLength);
+  if (first === 0xfd) return { value: view.getUint16(offset + 1, true), size: 3 };
+  if (first === 0xfe) return { value: view.getUint32(offset + 1, true), size: 5 };
+  return { value: Number(view.getBigUint64(offset + 1, true)), size: 9 };
+};
+const rawTxSize = (buf, start) => {
+  let o = start;
+  o += 4; // version
+  const marker = buf[o], flag = buf[o + 1];
+  const segwit = marker === 0x00 && flag !== 0x00;
+  if (segwit) o += 2;
+  const vin = readVarInt(buf, o); o += vin.size;
+  for (let i = 0; i < vin.value; i++) { o += 36; const sl = readVarInt(buf, o); o += sl.size + sl.value + 4; }
+  const vout = readVarInt(buf, o); o += vout.size;
+  for (let i = 0; i < vout.value; i++) { o += 8; const sl = readVarInt(buf, o); o += sl.size + sl.value; }
+  if (segwit) {
+    for (let i = 0; i < vin.value; i++) {
+      const ic = readVarInt(buf, o); o += ic.size;
+      for (let j = 0; j < ic.value; j++) { const il = readVarInt(buf, o); o += il.size + il.value; }
+    }
+  }
+  o += 4; // locktime
+  return o - start;
+};
+const handleRawBlock = (raw) => {
+  // A new block confirms watched txs: walk its txs and notify matching scripthashes.
+  let offset = 80; // skip header
+  const txCount = readVarInt(raw, offset);
+  offset += txCount.size;
+  for (let i = 0; i < txCount.value; i++) {
+    const size = rawTxSize(raw, offset);
+    const txRaw = raw.subarray(offset, offset + size);
+    offset += size;
+    try { notifyForTx(txRaw); } catch {}
+  }
+};
+
+const retry = async (host, port, topic, handler) => {
+  let delay = 2000;
+  for (;;) {
+    await startSub(host, port, topic, handler);
+    console.warn('zmq disconnected, retrying', topic);
+    await new Promise((r) => setTimeout(r, delay));
+    delay = Math.min(30000, Math.floor(delay * 1.5));
+  }
+};
+
+retry(BC_HOST, RAWTX_PORT, 'rawtx', notifyForTx);
+retry(BC_HOST, RAWBLOCK_PORT, 'rawblock', handleRawBlock);
+
+// ---- minimal Electrum WS server -------------------------------------------
+const reply = (ws, id, result) => {
+  try { ws.send(JSON.stringify({ jsonrpc: '2.0', id, result }) + '\n'); } catch {}
+};
+const handleReq = (ws, line) => {
+  let msg;
+  try { msg = JSON.parse(line); } catch { return; }
+  const { id, method, params } = msg || {};
+  if (method === 'server.version') return reply(ws, id, ['halwallet-zmq 1.0', '1.4']);
+  if (method === 'server.ping') return reply(ws, id, null);
+  if (method === 'blockchain.scripthash.subscribe') {
+    const sh = params && params[0];
+    if (typeof sh === 'string') {
+      ws.data.shes.add(sh);
+      let set = subs.get(sh);
+      if (!set) { set = new Set(); subs.set(sh, set); }
+      set.add(ws);
+    }
+    return reply(ws, id, null); // no real status; notifications drive the refresh
+  }
+  if (method === 'blockchain.scripthash.unsubscribe') {
+    const sh = params && params[0];
+    if (sh) {
+      ws.data.shes.delete(sh);
+      const set = subs.get(sh);
+      if (set) { set.delete(ws); if (!set.size) subs.delete(sh); }
+    }
+    return reply(ws, id, true);
+  }
+  return reply(ws, id, null);
+};
+
+Bun.serve({
+  port: WS_PORT,
+  fetch(req, server) {
+    if (server.upgrade(req)) return;
+    return new Response('halwallet zmq-electrum watcher\n');
+  },
+  websocket: {
+    open(ws) { ws.data = { buf: '', shes: new Set() }; },
+    message(ws, message) {
+      ws.data.buf += (typeof message === 'string' ? message : new TextDecoder().decode(message));
+      let nl;
+      while ((nl = ws.data.buf.indexOf('\n')) >= 0) {
+        const line = ws.data.buf.slice(0, nl).trim();
+        ws.data.buf = ws.data.buf.slice(nl + 1);
+        if (line) handleReq(ws, line);
+      }
+    },
+    close(ws) {
+      for (const sh of ws.data.shes) {
+        const set = subs.get(sh);
+        if (set) { set.delete(ws); if (!set.size) subs.delete(sh); }
+      }
+    },
+  },
+});
+console.log('halwallet zmq-electrum watcher: ws listening on', WS_PORT, '| bc', BC_HOST + ':' + RAWTX_PORT + '/' + RAWBLOCK_PORT);