aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/api.js21
-rw-r--r--src/wallet.js180
-rw-r--r--tools/zmq-watcher.js257
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);
generated by cgit v1.2.3 (git 2.51.0) at 2026-10-15 09:00:11 +0000