zmq logic

5 files changed, 873 insertions, 10 deletions

diff --git a/lib/esplora.ts b/lib/esplora.ts
new file mode 100644
index 00000000..4fd81b32
--- /dev/null
+++ b/lib/esplora.ts
@@ -0,0 +1,153 @@
+import config from "$config";
+import { HDKey } from "@scure/bip32";
+import { p2wpkh, NETWORK, TEST_NETWORK } from "@scure/btc-signer";
+
+const { esploraUrl } = config.bitcoin;
+
+const REGTEST_NETWORK = {
+  bech32: "bcrt",
+  pubKeyHash: 0x6f,
+  scriptHash: 0xc4,
+  wif: 0xef,
+};
+
+const REGTEST_VERSIONS = { private: 0x04358394, public: 0x043587cf };
+
+const btcNetwork =
+  config.bitcoin.network === "regtest"
+    ? REGTEST_NETWORK
+    : config.bitcoin.network === "testnet"
+      ? TEST_NETWORK
+      : NETWORK;
+
+const hdVersions =
+  config.bitcoin.network === "regtest" || config.bitcoin.network === "testnet"
+    ? REGTEST_VERSIONS
+    : undefined;
+
+// Esplora API
+
+export const getUtxos = async (address: string) => {
+  const r = await fetch(`${esploraUrl}/address/${address}/utxo`);
+  if (!r.ok) throw new Error(`esplora getUtxos: ${r.status}`);
+  return r.json();
+};
+
+export const getAddressUtxos = async (addresses: string[]) => {
+  const results = [];
+  for (const address of addresses) {
+    const utxos = await getUtxos(address);
+    for (const u of utxos) {
+      u.address = address;
+      results.push(u);
+    }
+  }
+  return results;
+};
+
+export const broadcastTx = async (txHex: string) => {
+  const r = await fetch(`${esploraUrl}/tx`, {
+    method: "POST",
+    body: txHex,
+    headers: { "Content-Type": "text/plain" },
+  });
+  if (!r.ok) {
+    const body = await r.text();
+    throw new Error(`broadcast failed: ${body}`);
+  }
+  return r.text();
+};
+
+export const getTxStatus = async (txid: string) => {
+  const r = await fetch(`${esploraUrl}/tx/${txid}/status`);
+  if (!r.ok) throw new Error(`esplora getTxStatus: ${r.status}`);
+  return r.json();
+};
+
+export const getFeeEstimates = async () => {
+  const r = await fetch(`${esploraUrl}/fee-estimates`);
+  if (!r.ok) throw new Error(`esplora getFeeEstimates: ${r.status}`);
+  return r.json();
+};
+
+export const getTx = async (txid: string) => {
+  const r = await fetch(`${esploraUrl}/tx/${txid}`);
+  if (!r.ok) throw new Error(`esplora getTx: ${r.status}`);
+  return r.json();
+};
+
+export const getAddressTxs = async (address: string) => {
+  const r = await fetch(`${esploraUrl}/address/${address}/txs`);
+  if (!r.ok) throw new Error(`esplora getAddressTxs: ${r.status}`);
+  return r.json();
+};
+
+export const getTxHex = async (txid: string) => {
+  const r = await fetch(`${esploraUrl}/tx/${txid}/hex`);
+  if (!r.ok) throw new Error(`esplora getTxHex: ${r.status}`);
+  return r.text();
+};
+
+// Address derivation
+
+const hdVersionsForKey = (pubkey: string) => {
+  if (pubkey.startsWith("tpub") || pubkey.startsWith("tprv"))
+    return REGTEST_VERSIONS;
+  // xpub/xprv use default BITCOIN_VERSIONS (mainnet) — pass undefined
+  return undefined;
+};
+
+export const deriveAddress = (
+  pubkey: string,
+  fingerprint: string,
+  index: number,
+  internal = false,
+) => {
+  const accountKey = HDKey.fromExtendedKey(pubkey, hdVersionsForKey(pubkey));
+  const chain = internal ? 1 : 0;
+  const child = accountKey.deriveChild(chain).deriveChild(index);
+
+  const { address } = p2wpkh(child.publicKey, btcNetwork);
+  const path = `m/${chain}/${index}`;
+
+  return { address, path };
+};
+
+export const deriveAddresses = (
+  pubkey: string,
+  fingerprint: string,
+  count: number,
+  internal = false,
+) => {
+  const addresses = [];
+  for (let i = 0; i < count; i++) {
+    const { address } = deriveAddress(pubkey, fingerprint, i, internal);
+    addresses.push(address);
+  }
+  return addresses;
+};
+
+// Migration helpers
+
+export const parseDescriptor = (desc: string) => {
+  // Parse wpkh([fingerprint]pubkey/0/*)#checksum
+  const match = desc.match(/wpkh\(\[([a-f0-9]+)\]([^/]+)\/\d+\/\*\)/);
+  if (!match) return null;
+  return { fingerprint: match[1], pubkey: match[2] };
+};
+
+export const findLastUsedIndex = async (
+  pubkey: string,
+  fingerprint: string,
+  maxScan = 100,
+) => {
+  let lastUsed = -1;
+  for (let i = 0; i < maxScan; i++) {
+    const { address } = deriveAddress(pubkey, fingerprint, i, false);
+    const txs = await getAddressTxs(address);
+    if (txs.length > 0) lastUsed = i;
+  }
+  return lastUsed + 1;
+};
+
+export { btcNetwork, hdVersions };
diff --git a/lib/invoices.ts b/lib/invoices.ts
index 50cddd9a..4a9d5c89 100644
--- a/lib/invoices.ts
+++ b/lib/invoices.ts
@@ -1,6 +1,7 @@
 import config from "$config";
 import { db, g, s } from "$lib/db";
 import { request } from "$lib/ecash";
+import { deriveAddress } from "$lib/esplora";
 import ln from "$lib/ln";
 import { emit } from "$lib/sockets";
 import { SATS, bip21, fail, getInvoice, getUser } from "$lib/utils";
@@ -68,11 +69,16 @@ export const generate = async ({ invoice, user }) => {
     text = account.arkAddress;
   } else if (account.seed) {
     type = PaymentType.bitcoin;
-    const node = rpc({ ...config[type], wallet: aid });
-    hash = await node.getNewAddress({ address_type });
+    const nextIndex = account.nextIndex || 0;
+    const { address, path: hdpath } = deriveAddress(account.pubkey, account.fingerprint, nextIndex);
+    hash = address;
+    path = hdpath;
     text = bip21(hash, invoice);
 
-    ({ hdkeypath: path } = await node.getAddressInfo(hash));
+    account.nextIndex = nextIndex + 1;
+    await s(`account:${aid}`, account);
+    await s(`invoice:${hash}`, id);
+    await db.sAdd("watching", hash);
   } else if (type === PaymentType.lightning) {
     let r;
     if (bolt11) {
diff --git a/lib/migrate.ts b/lib/migrate.ts
index 58623ec1..8fd4ec1a 100644
--- a/lib/migrate.ts
+++ b/lib/migrate.ts
@@ -1,4 +1,6 @@
-import { g } from "$lib/db";
+import { db, g, s } from "$lib/db";
+import { findLastUsedIndex, parseDescriptor } from "$lib/esplora";
+import { warn } from "$lib/logging";
 
 async function migrate(id) {
   const k = id?.replace(/\s/g, "").toLowerCase();
@@ -7,4 +9,44 @@ async function migrate(id) {
   return user;
 }
 
+export const migrateAccounts = async () => {
+  const keys = await db.keys("account:*");
+  let migrated = 0;
+
+  for (const key of keys) {
+    try {
+      if (key.endsWith(":invoices")) continue;
+      const keyType = await db.type(key);
+      if (keyType !== "string") continue;
+
+      const account = await g(key);
+      if (!account || !account.id) continue;
+      if (account.pubkey || account.type === "ark" || !account.seed) continue;
+      if (!account.descriptors?.length) continue;
+
+      const parsed = parseDescriptor(account.descriptors[0].desc);
+      if (!parsed) continue;
+
+      account.pubkey = parsed.pubkey;
+      account.fingerprint = parsed.fingerprint;
+      try {
+        account.nextIndex = await findLastUsedIndex(
+          parsed.pubkey,
+          parsed.fingerprint,
+        );
+      } catch (e) {
+        warn("account migration index scan skipped", key, e.message);
+        account.nextIndex = account.nextIndex || 0;
+      }
+
+      await s(`account:${account.id}`, account);
+      migrated++;
+    } catch (e) {
+      warn("failed to migrate account", key, e.message);
+    }
+  }
+
+  return migrated;
+};
+
 export default migrate;
diff --git a/lib/payments.ts b/lib/payments.ts
index b0140f52..7dae4a91 100644
--- a/lib/payments.ts
+++ b/lib/payments.ts
@@ -1,11 +1,24 @@
 import config from "$config";
 import api from "$lib/api";
 import { db, g, ga, gf, s } from "$lib/db";
+import {
+  broadcastTx,
+  btcNetwork,
+  deriveAddress,
+  deriveAddresses,
+  getAddressTxs,
+  getAddressUtxos,
+  getTxHex,
+  getTxStatus,
+  hdVersions,
+} from "$lib/esplora";
+import { HDKey } from "@scure/bip32";
 import { generate } from "$lib/invoices";
 import ln from "$lib/ln";
 import { err, l, warn } from "$lib/logging";
 import { handleZap } from "$lib/nostr";
 import { notify, nwcNotify } from "$lib/notifications";
+import { emit } from "$lib/sockets";
 import { squarePayment } from "$lib/square";
 import {
   SATS,
@@ -23,6 +36,7 @@ import {
 } from "$lib/utils";
 import { callWebhook } from "$lib/webhooks";
 import rpc from "@coinos/rpc";
+import { selectUTXO, p2wpkh } from "@scure/btc-signer";
 import { bech32 } from "bech32";
 import got from "got";
 import { v4 } from "uuid";
@@ -325,6 +339,60 @@ export const completePayment = async (inv, p, user) => {
   callWebhook(inv, p);
 };
 
+const confirmWatchedIncoming = async (address, existing) => {
+  existing.confirmed = true;
+  const inv = await getInvoice(address);
+  if (inv) {
+    inv.received += Number.parseInt(inv.pending);
+    inv.pending = 0;
+    await s(`invoice:${inv.id}`, inv);
+  }
+  await s(`payment:${existing.id}`, existing);
+  await db.decrBy(`pending:${existing.aid || existing.uid}`, existing.amount);
+  await db.incrBy(`balance:${existing.aid || existing.uid}`, existing.amount);
+  const user = await getUser(existing.uid);
+  if (inv) await completePayment(inv, existing, user);
+  await db.sRem("watching", address);
+};
+
+export const processWatchedTx = async (tx) => {
+  const txid = tx.txid;
+
+  for (let vout = 0; vout < tx.vout.length; vout++) {
+    const output = tx.vout[vout];
+    const address = output.scriptpubkey_address;
+    if (!address) continue;
+    if (!(await db.sIsMember("watching", address))) continue;
+
+    const invoice = await getInvoice(address);
+    if (!invoice) continue;
+
+    const existing = await getPayment(`${txid}:${vout}`);
+    if (existing) {
+      if (!existing.confirmed && tx.status?.confirmed) {
+        await confirmWatchedIncoming(address, existing);
+      }
+      continue;
+    }
+
+    if (output.value < 300) continue;
+    await credit({
+      hash: address,
+      amount: output.value,
+      ref: `${txid}:${vout}`,
+      type: PaymentType.bitcoin,
+      aid: invoice.aid,
+    });
+
+    if (tx.status?.confirmed) {
+      const created = await getPayment(`${txid}:${vout}`);
+      if (created && !created.confirmed) {
+        await confirmWatchedIncoming(address, created);
+      }
+    }
+  }
+};
+
 const pay = async ({ aid = undefined, amount, to, user }) => {
   if (!aid) aid = user.id;
   amount = Number.parseInt(amount) || 0;
@@ -383,14 +451,94 @@ export const decode = async (hex) => {
 };
 
 const inflight = {};
+
+const sendNonCustodial = async (params) => {
+  let { aid, hex, rate, user } = params;
+  if (!hex) ({ hex } = await build(params));
+
+  const { tx } = await decode(hex);
+  let { txid } = tx;
+
+  try {
+    if (inflight[txid]) fail("payment in flight");
+    inflight[txid] = true;
+
+    const account = await g(`account:${aid}`);
+    const nextIndex = account.nextIndex || 0;
+
+    // Build set of own addresses for change detection
+    const ownAddresses = new Set<string>();
+    for (let i = 0; i <= nextIndex; i++) {
+      ownAddresses.add(
+        deriveAddress(account.pubkey, account.fingerprint, i, false).address,
+      );
+      ownAddresses.add(
+        deriveAddress(account.pubkey, account.fingerprint, i, true).address,
+      );
+    }
+
+    let totalIn = 0;
+    for (const { txid: inputTxid, vout } of tx.vin) {
+      const inputHex = await getTxHex(inputTxid);
+      const inputTx = await bc.decodeRawTransaction(inputHex);
+      totalIn += sats(inputTx.vout[vout].value);
+    }
+
+    let total = 0;
+    let change = 0;
+    for (const {
+      scriptPubKey: { address },
+      value,
+    } of tx.vout) {
+      total += sats(value);
+      const invoice = await getInvoice(address);
+      if (invoice?.aid === aid) fail("Cannot send to internal address");
+
+      if (ownAddresses.has(address)) {
+        change += sats(value);
+      }
+    }
+
+    const fee = totalIn - total;
+    const amount = total - change;
+
+    const p = await debit({
+      aid,
+      hash: txid,
+      amount,
+      fee,
+      rate,
+      user,
+      type: PaymentType.bitcoin,
+    });
+
+    p.hex = hex;
+    await s(`payment:${p.id}`, p);
+
+    await broadcastTx(hex);
+    await db.sAdd(`inflight:${aid}`, p.id);
+
+    delete inflight[txid];
+    return p;
+  } catch (e) {
+    delete inflight[txid];
+    throw e;
+  }
+};
+
 export const sendOnchain = async (params) => {
   let { aid, hex, rate, user, signed } = params;
   if (!aid) aid = user.id;
+
+  // Non-custodial bitcoin account — use esplora
+  if (aid !== user.id) {
+    return sendNonCustodial(params);
+  }
+
   if (!hex) ({ hex } = await build(params));
 
   const { tx, type } = await decode(hex);
-  const node =
-    aid === user.id ? rpc(config[type]) : rpc({ ...config[type], wallet: aid });
+  const node = rpc(config[type]);
   let { txid } = tx;
 
   try {
@@ -635,6 +783,158 @@ const getAddressType = async (a) => {
   }
 };
 
+const buildNonCustodial = async ({
+  aid,
+  amount,
+  address,
+  feeRate,
+  subtract,
+  user,
+}) => {
+  const account = await g(`account:${aid}`);
+  if (!account?.pubkey) fail("account missing pubkey");
+
+  amount = Number.parseInt(amount);
+  if (amount < 0) fail("invalid amount");
+
+  const fees: any = await fetch(
+    `${api[PaymentType.bitcoin]}/fees/recommended`,
+  ).then((r) => r.json());
+
+  fees.hourFee = fees.halfHourFee;
+  fees.halfHourFee = fees.fastestFee;
+  fees.fastestFee = Math.ceil(fees.fastestFee * 1.5);
+
+  if (!feeRate) feeRate = fees.halfHourFee;
+  if (feeRate < fees.hourFee) fail("fee rate too low");
+
+  const nextIndex = account.nextIndex || 0;
+
+  // Derive all used external + internal addresses and fetch UTXOs
+  const externalAddrs = deriveAddresses(
+    account.pubkey,
+    account.fingerprint,
+    nextIndex + 1,
+    false,
+  );
+  const internalAddrs = deriveAddresses(
+    account.pubkey,
+    account.fingerprint,
+    nextIndex + 1,
+    true,
+  );
+  const allAddrs = [...externalAddrs, ...internalAddrs];
+
+  const rawUtxos = await getAddressUtxos(allAddrs);
+  if (!rawUtxos.length) fail("no UTXOs available");
+
+  // Build address-to-path lookup
+  const addrToPath = {};
+  for (let i = 0; i <= nextIndex; i++) {
+    const { address: extAddr } = deriveAddress(
+      account.pubkey,
+      account.fingerprint,
+      i,
+      false,
+    );
+    addrToPath[extAddr] = `m/0/${i}`;
+    const { address: intAddr } = deriveAddress(
+      account.pubkey,
+      account.fingerprint,
+      i,
+      true,
+    );
+    addrToPath[intAddr] = `m/1/${i}`;
+  }
+
+  // Convert esplora UTXOs to selectUTXO input format
+  const keyVersions = account.pubkey.startsWith("tpub")
+    ? hdVersions
+    : undefined;
+  const accountKey = HDKey.fromExtendedKey(account.pubkey, keyVersions);
+
+  const utxoInputs = rawUtxos.map((u) => {
+    const path = addrToPath[u.address];
+    const parts = path.split("/").slice(-2);
+    const child = accountKey
+      .deriveChild(Number.parseInt(parts[0]))
+      .deriveChild(Number.parseInt(parts[1]));
+    const payment = p2wpkh(child.publicKey, btcNetwork);
+
+    return {
+      txid: u.txid,
+      index: u.vout,
+      witnessUtxo: {
+        amount: BigInt(u.value),
+        script: payment.script,
+      },
+    };
+  });
+
+  const balance = await g(`balance:${aid}`);
+  let ourfee = 0; // Non-custodial accounts don't pay platform fee
+
+  const outputs = [{ address, amount: BigInt(amount) }];
+
+  // Derive a change address (next internal address)
+  const { address: changeAddress } = deriveAddress(
+    account.pubkey,
+    account.fingerprint,
+    nextIndex,
+    true,
+  );
+
+  let selected = selectUTXO(utxoInputs, outputs, "default", {
+    changeAddress,
+    feePerByte: BigInt(feeRate),
+    network: btcNetwork,
+    createTx: true,
+  });
+
+  if (!selected) {
+    subtract = true;
+    if (amount <= dust) {
+      fail(`insufficient funds ⚡️${balance} of ⚡️${amount + dust}`);
+    }
+
+    // Try with subtracted fee — send max
+    const maxOutputs = [{ address, amount: BigInt(amount) }];
+    selected = selectUTXO(utxoInputs, maxOutputs, "all", {
+      changeAddress,
+      feePerByte: BigInt(feeRate),
+      network: btcNetwork,
+      createTx: true,
+    });
+
+    if (!selected) fail("insufficient funds");
+  }
+
+  const fee = Number(selected.fee);
+
+  // Build input metadata for client signing
+  const inputs = selected.inputs.map((input) => {
+    const inputTxid =
+      typeof input.txid === "string"
+        ? input.txid
+        : Buffer.from(input.txid).toString("hex");
+    const utxo = rawUtxos.find(
+      (u) => u.txid === inputTxid && u.vout === input.index,
+    );
+    const path = utxo ? addrToPath[utxo.address] : undefined;
+    return {
+      witnessUtxo: {
+        amount: Number(input.witnessUtxo.amount),
+        script: Buffer.from(input.witnessUtxo.script).toString("hex"),
+      },
+      path,
+    };
+  });
+
+  const hex = Buffer.from(selected.tx.toPSBT()).toString("hex");
+
+  return { feeRate, ourfee, fee, fees, hex, inputs, subtract };
+};
+
 export const build = async ({
   aid,
   amount,
@@ -645,8 +945,13 @@ export const build = async ({
 }) => {
   const type = await getAddressType(address);
   if (!aid) aid = user.id;
-  const node =
-    aid === user.id ? rpc(config[type]) : rpc({ ...config[type], wallet: aid });
+
+  // Non-custodial bitcoin account — use esplora
+  if (aid !== user.id && type === PaymentType.bitcoin) {
+    return buildNonCustodial({ aid, amount, address, feeRate, subtract, user });
+  }
+
+  const node = rpc(config[type]);
 
   amount = Number.parseInt(amount);
   if (amount < 0) fail("invalid amount");
@@ -699,7 +1004,6 @@ export const build = async ({
   const covered = Math.min(credit, ourfee);
   ourfee -= covered;
 
-  if (aid && aid !== user.id) ourfee = 0;
   if (subtract || amount + fee + ourfee > balance) {
     subtract = true;
     if (amount <= fee + ourfee + dust) {
@@ -741,6 +1045,7 @@ export const build = async ({
 
 export const catchUp = async () => {
   try {
+    // Platform wallets (custodial bitcoin + liquid)
     const txns = [];
     for (const [type, n] of Object.entries({ bitcoin: bc, liquid: lq })) {
       txns.push(
@@ -765,13 +1070,95 @@ export const catchUp = async () => {
         err("problem confirming", e.message);
       }
     }
+
+    // Non-custodial accounts: check pending outgoing payments
+    const inflightAccounts = await db.keys("inflight:*");
+    for (const key of inflightAccounts) {
+      const aid = key.replace("inflight:", "");
+      const keyType = await db.type(key);
+      if (keyType !== "set") continue;
+      const paymentIds = await db.sMembers(key);
+      for (const pid of paymentIds) {
+        try {
+          const p = await gf(`payment:${pid}`);
+          if (!p || p.confirmed) {
+            await db.sRem(key, pid);
+            continue;
+          }
+          const status = await getTxStatus(p.hash);
+          if (status.confirmed) {
+            p.confirmed = true;
+            await s(`payment:${p.id}`, p);
+            await db.sRem(key, pid);
+            emit(p.uid, "payment", p);
+          }
+        } catch (e) {
+          err("problem checking inflight payment", e.message);
+        }
+      }
+    }
+
   } catch (e) {
     err("problem syncing", e.message);
   }
+};
+
+export const reconcile = async (account, initial = false) => {
+  try {
+    const { id, uid, pubkey, fingerprint, nextIndex } = account;
+    const user = await getUser(uid);
+
+    // Derive all addresses and sum confirmed UTXOs from esplora
+    const count = (nextIndex || 0) + 1;
+    const externalAddrs = deriveAddresses(pubkey, fingerprint, count, false);
+    const internalAddrs = deriveAddresses(pubkey, fingerprint, count, true);
+    const allAddrs = [...externalAddrs, ...internalAddrs];
 
-  setTimeout(catchUp, 10000);
+    const utxos = await getAddressUtxos(allAddrs);
+    const total = utxos
+      .filter((u) => u.status.confirmed)
+      .reduce((sum, u) => sum + u.value, 0);
+
+    const { balanceAdjustment: memo } = t(user);
+
+    const balance = await g(`balance:${id}`);
+
+    const amount = Math.abs(total - balance);
+    const hash = v4();
+
+    if (total > balance) {
+      const inv = {
+        memo,
+        type: PaymentType.reconcile,
+        hash,
+        amount,
+        uid,
+        aid: id,
+      };
+      await s(`invoice:${hash}`, inv);
+      await credit({
+        hash,
+        amount,
+        type: PaymentType.reconcile,
+        aid: id,
+      });
+    } else if (total < balance) {
+      await debit({
+        aid: id,
+        amount,
+        hash: v4(),
+        memo,
+        user,
+        type: PaymentType.reconcile,
+      });
+    }
+  } catch (e) {
+    console.log(e);
+    warn("problem reconciling", e.message, account);
+  }
 };
 
+
 export const check = async () => {
   if (process.env.URL.includes("dev")) return;
   try {
diff --git a/lib/zmq.ts b/lib/zmq.ts
new file mode 100644
index 00000000..f4d22fce
--- /dev/null
+++ b/lib/zmq.ts
@@ -0,0 +1,275 @@
+import net from "node:net";
+import config from "$config";
+import { btcNetwork } from "$lib/esplora";
+import { err, l, warn } from "$lib/logging";
+import { processWatchedTx } from "$lib/payments";
+import { Transaction } from "@scure/btc-signer";
+import { concatBytes, sha256x2 } from "@scure/btc-signer/utils.js";
+import { bytesToHex } from "@noble/hashes/utils";
+
+const zmqHost = config.bitcoin.host || "127.0.0.1";
+const zmqRawTxPort = config.bitcoin.zmq?.rawtx || 18507;
+const zmqRawBlockPort = config.bitcoin.zmq?.rawblock || 18506;
+const RAWTX_URL = `tcp://${zmqHost}:${zmqRawTxPort}`;
+const RAWBLOCK_URL = `tcp://${zmqHost}:${zmqRawBlockPort}`;
+
+const TX_OPTS = {
+  allowUnknownOutputs: true,
+  allowUnknownInputs: true,
+  disableScriptCheck: true,
+};
+
+const enc = new TextEncoder();
+const dec = new TextDecoder();
+
+const reverseHex = (buf: Uint8Array) =>
+  bytesToHex(Uint8Array.from(buf).reverse());
+
+const readVarInt = (data: Uint8Array, offset: number) => {
+  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: Uint8Array, start: number) => {
+  let offset = start;
+  offset += 4; // version
+
+  const marker = buf[offset];
+  const flag = buf[offset + 1];
+  const segwit = marker === 0x00 && flag !== 0x00;
+  if (segwit) offset += 2;
+
+  const vinVar = readVarInt(buf, offset);
+  offset += vinVar.size;
+  for (let i = 0; i < vinVar.value; i++) {
+    offset += 36; // prev_hash(32) + prev_index(4)
+    const scriptLen = readVarInt(buf, offset);
+    offset += scriptLen.size + scriptLen.value + 4; // script + sequence
+  }
+
+  const voutVar = readVarInt(buf, offset);
+  offset += voutVar.size;
+  for (let i = 0; i < voutVar.value; i++) {
+    offset += 8; // value
+    const scriptLen = readVarInt(buf, offset);
+    offset += scriptLen.size + scriptLen.value;
+  }
+
+  if (segwit) {
+    for (let i = 0; i < vinVar.value; i++) {
+      const itemCount = readVarInt(buf, offset);
+      offset += itemCount.size;
+      for (let j = 0; j < itemCount.value; j++) {
+        const itemLen = readVarInt(buf, offset);
+        offset += itemLen.size + itemLen.value;
+      }
+    }
+  }
+
+  offset += 4; // locktime
+  return offset - start;
+};
+
+const decodeTx = (raw: Uint8Array, confirmed: boolean) => {
+  const tx = Transaction.fromRaw(raw, TX_OPTS);
+  const vout = [];
+  for (let i = 0; i < tx.outputsLength; i++) {
+    const { amount } = tx.getOutput(i);
+    vout.push({
+      scriptpubkey_address: tx.getOutputAddress(i, btcNetwork),
+      value: Number(amount),
+    });
+  }
+  return { txid: tx.id, vout, status: { confirmed } };
+};
+
+const handleRawTx = async (raw: Uint8Array) => {
+  await processWatchedTx(decodeTx(raw, false));
+};
+
+const handleRawBlock = async (raw: Uint8Array) => {
+  const blockHash = reverseHex(sha256x2(raw.subarray(0, 80)));
+  l("zmq block", blockHash);
+
+  let offset = 80;
+  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 {
+      await processWatchedTx(decodeTx(txRaw, true));
+    } catch (e) {
+      warn("zmq block tx decode failed", i, e.message);
+    }
+  }
+};
+
+// ZMTP 3.0 NULL protocol
+
+const greeting = () => {
+  const buf = new Uint8Array(64);
+  buf[0] = 0xff;
+  buf[9] = 0x7f;
+  buf[10] = 3; // major
+  buf[11] = 0; // minor
+  buf.set(enc.encode("NULL"), 12);
+  return buf;
+};
+
+const frame = (flags: number, body: Uint8Array) => {
+  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: string, props: Record<string, string> = {}) => {
+  const parts: Uint8Array[] = [
+    Uint8Array.of(name.length),
+    enc.encode(name),
+  ];
+  for (const [k, v] of Object.entries(props)) {
+    const kb = enc.encode(k);
+    const 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: string) => {
+  const t = enc.encode(topic);
+  return frame(
+    0x04,
+    concatBytes(
+      Uint8Array.of(9),
+      enc.encode("SUBSCRIBE"),
+      Uint8Array.of(t.length),
+      t,
+    ),
+  );
+};
+
+const legacySubscribe = (topic: string) =>
+  frame(0x00, concatBytes(Uint8Array.of(1), enc.encode(topic)));
+
+const startSub = (
+  url: string,
+  topic: string,
+  onMessage: (b: Uint8Array) => void,
+) =>
+  new Promise<void>((resolve) => {
+    const [host, port] = url.replace("tcp://", "").split(":");
+    const socket = net.connect({ host, port: Number(port) });
+
+    let buffer = new Uint8Array(0);
+    let handshakeDone = false;
+    let frames: Uint8Array[] = [];
+    let closed = false;
+
+    const processBuffer = async () => {
+      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));
+        l("zmq subscribed", topic, url);
+      }
+
+      while (buffer.length >= 2) {
+        const flags = buffer[0];
+        let size: number;
+        let header: number;
+        if (flags & 0x02) {
+          if (buffer.length < 9) return;
+          const view = new DataView(
+            buffer.buffer,
+            buffer.byteOffset,
+            buffer.byteLength,
+          );
+          size = Number(view.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;
+        frames.push(body);
+        if (flags & 0x01) continue;
+
+        if (frames.length >= 2 && dec.decode(frames[0]) === topic) {
+          onMessage(frames[1]);
+        }
+        frames = [];
+      }
+    };
+
+    socket.on("connect", () => socket.write(greeting()));
+    socket.on("data", (d: Uint8Array) => {
+      buffer = concatBytes(buffer, d);
+      processBuffer().catch((e) => err("zmq parse error", e.message));
+    });
+    socket.on("error", (e) => {
+      if (closed) return;
+      closed = true;
+      warn("zmq socket error", e.message);
+      resolve();
+    });
+    socket.on("close", () => {
+      if (closed) return;
+      closed = true;
+      warn("zmq socket closed", url);
+      resolve();
+    });
+  });
+
+export const startZmq = async () => {
+  if (process.env.DISABLE_ZMQ === "1") return;
+  l("zmq connecting", RAWTX_URL, RAWBLOCK_URL);
+
+  const retry = async (
+    url: string,
+    topic: string,
+    handler: (b: Uint8Array) => Promise<void>,
+  ) => {
+    let delay = 1000;
+    const maxDelay = 30000;
+    while (true) {
+      await startSub(url, topic, (raw) =>
+        handler(raw).catch((e) => warn(`${topic} error`, e.message)),
+      );
+      await new Promise((r) => setTimeout(r, delay));
+      delay = Math.min(maxDelay, Math.floor(delay * 1.5));
+    }
+  };
+
+  retry(RAWTX_URL, "rawtx", handleRawTx).catch(() => {});
+  retry(RAWBLOCK_URL, "rawblock", handleRawBlock).catch(() => {});
+};