Push silent payments over WebSocket instead of polling

The indexer now serves a /ws endpoint and publishes each newly-indexed block's
SP items to subscribers. The wallet subscribes (_spConnect) and scans pushed
items on arrival (_scanSpItems) — folding receipts into the balance instantly —
catching up via /scan on connect/reconnect. Polling for SP is now only a fallback
for when the socket isn't live.

Verified on regtest: connect, mine a block, wallet receives the push.

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

2 files changed, 83 insertions, 2 deletions

diff --git a/src/wallet.js b/src/wallet.js
index bff8910..212c471 100644
--- a/src/wallet.js
+++ b/src/wallet.js
@@ -369,6 +369,66 @@ export class Wallet {
     this.spUtxos = next;
   }
 
+  // Live silent-payment push: subscribe to the indexer's WebSocket so each new
+  // block's SP items are scanned on arrival instead of polling. Catches up via a
+  // /scan on connect (covers anything missed while disconnected) and reconnects.
+  _spConnect() {
+    if (this.offline) return;
+    const base = spIndexerUrl(this.netName);
+    if (!base || !this.silentPaymentKeys()) return;
+    const url = base.replace(/^http/, 'ws') + '/ws';
+    if (this._spWs && this._spWsUrl === url) return;
+    this._spDisconnect();
+    this._spWsUrl = url;
+    let ws;
+    try { ws = new WebSocket(url); } catch { return; }
+    this._spWs = ws;
+    ws.onopen = () => { this.spLive = true; this.scanSilentPayments({ silent: true }).catch(() => {}); };
+    ws.onmessage = (ev) => {
+      let msg;
+      try { msg = JSON.parse(typeof ev.data === 'string' ? ev.data : new TextDecoder().decode(ev.data)); } catch { return; }
+      if (msg && msg.type === 'block' && Array.isArray(msg.items)) this._scanSpItems(msg.items, msg.height).catch(() => {});
+    };
+    ws.onerror = () => { try { ws.close(); } catch {} };
+    ws.onclose = () => {
+      if (this._spWs !== ws) return;
+      this.spLive = false; this._spWs = null;
+      if (!this.offline) this._spReconnect = setTimeout(() => this._spConnect(), 3000);
+    };
+  }
+
+  _spDisconnect() {
+    clearTimeout(this._spReconnect);
+    this.spLive = false;
+    const ws = this._spWs;
+    this._spWs = null; this._spWsUrl = null;
+    if (ws) { try { ws.onclose = null; ws.close(); } catch {} }
+  }
+
+  // Scan a pushed block's items (no fetch) and fold any of ours into spUtxos. The
+  // block is confirmed (just mined), so matches are recorded confirmed; the
+  // backstop _refreshSpUtxos reconciles later spends.
+  async _scanSpItems(items, height) {
+    const keys = this.silentPaymentKeys();
+    if (!keys) return;
+    const have = new Set(this.spUtxos.map((u) => `${u.txid}:${u.vout}`));
+    let added = false;
+    for (const it of items) {
+      const matches = silentPaymentScan({ scanPriv: keys.scanPriv, spendPub: keys.spendPub, tweak: hex.decode(it.tweak), outputs: it.outputs.map((o) => o.xonly) });
+      for (const m of matches) {
+        const o = it.outputs.find((o) => o.xonly === m.output);
+        if (have.has(`${it.txid}:${o.vout}`)) continue;
+        have.add(`${it.txid}:${o.vout}`);
+        this.spUtxos.push({ txid: it.txid, vout: o.vout, value: o.value, xonly: m.output, tweak: m.tweak, address: this._spAddress(m.output), confirmed: true });
+        added = true;
+      }
+    }
+    if (height && height > this.lastSpScan) this.lastSpScan = height;
+    if (added) this._recomputeBalanceFromChains();
+    this.saveCache();
+    if (added) this.emit();
+  }
+
   node(chain, index) {
     // Relative derivation from the account node (chain/index are non-hardened).
     return this.account().deriveChild(chain).deriveChild(index);
@@ -1635,8 +1695,11 @@ export class Wallet {
       if (Date.now() - (this._lastPoll || 0) < due) return;
       this._lastPoll = Date.now();
       this.refreshLive();
-      this.scanSilentPayments({ silent: true }).catch(() => {}); // pick up new SP receipts (incremental, cheap)
+      // SP receipts arrive via the indexer WebSocket (_spConnect); only poll for
+      // them as a fallback when that socket isn't live.
+      if (!this.spLive) this.scanSilentPayments({ silent: true }).catch(() => {});
     }, 1000);
+    this._spConnect(); // live push of silent-payment receipts
     // No periodic full re-scan: the WS (and the frontier poll above) keep the
     // balance current, and refreshLive's _reconcileHeld catches spends of held
     // coins. The only thing neither covers is a deposit to a reused old address,
@@ -1899,6 +1962,7 @@ export class Wallet {
   stopRealtime() {
     this._wsWant = false;
     this.live = false;
+    this._spDisconnect();
     this._rejectPending('realtime stopped'); // fail any in-flight Electrum data calls
     this._callQueue = [];
     clearTimeout(this._wsRetry);
diff --git a/tools/sp-indexer.js b/tools/sp-indexer.js
index d573ab9..bae1e15 100644
--- a/tools/sp-indexer.js
+++ b/tools/sp-indexer.js
@@ -76,13 +76,20 @@ async function indexBlock(height) {
   indexed = height;
 }
 
+let live = false; // true once caught up to the tip — then new blocks are pushed over WS
+
 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
+      // 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) { live = true; console.log('caught up — pushing new blocks over websocket'); }
   } catch (e) {
     console.error('sync error:', e.message);
   }
@@ -91,12 +98,17 @@ async function sync() {
 const CORS = { 'access-control-allow-origin': '*', 'access-control-allow-headers': '*' };
 const json = (obj, status = 200) => new Response(JSON.stringify(obj), { status, headers: { 'content-type': 'application/json', ...CORS } });
 
-Bun.serve({
+const server = Bun.serve({
   port: PORT,
   async fetch(req) {
     const url = new URL(req.url);
     const p = url.pathname;
     if (req.method === 'OPTIONS') return new Response(null, { headers: CORS });
+    // WebSocket: subscribers get a {type:'block', height, items} push per new block.
+    if (p === '/ws') {
+      if (server.upgrade(req)) return undefined;
+      return new Response('expected websocket', { status: 400 });
+    }
     if (p === '/height') return json({ height: indexed });
     let m;
     if ((m = p.match(/^\/block\/(\d+)$/))) {
@@ -123,6 +135,11 @@ Bun.serve({
     }
     return json({ error: 'not found' }, 404);
   },
+  websocket: {
+    open(ws) { ws.subscribe('blocks'); ws.send(JSON.stringify({ type: 'height', height: indexed })); },
+    message() {}, // clients don't send anything
+    close(ws) { ws.unsubscribe('blocks'); },
+  },
 });
 
 console.log(`sp-indexer: core=${RPC} start=${START} serving :${PORT}`);