theseus/bundled-addons/aegis/lib/electrum.js

// Electrum (Fulcrum) JSON-RPC over WebSocket for the wallet. One live
// connection at a time, chosen by walking the server list in order; the
// caller gets a stable `call()` that reconnects transparently on the next
// request after a drop. Notifications (headers / scripthash subscriptions)
// fan out to `onNotify`.
module.exports = function makeElectrum({ WebSocket, log = () => {} }) {
  const CALL_TIMEOUT_MS = 20000;

  class Connection {
    constructor(url) {
      this.url = url;
      this.id = 0;
      this.pending = new Map();
      this.buf = "";
      this.closed = false;
      this.onNotify = null;
      this.onClose = null;
    }
    connect() {
      return new Promise((resolve, reject) => {
        const ws = new WebSocket(this.url);
        this.ws = ws;
        const fail = (e) => { if (!this.closed) { this.closed = true; reject(e instanceof Error ? e : new Error("electrum ws error: " + this.url)); } };
        ws.on("open", async () => {
          try { await this.call("server.version", ["theseus-bchwallet", "1.4"]); resolve(this); }
          catch (e) { fail(e); this.close(); }
        });
        ws.on("error", fail);
        ws.on("message", (d) => this._onData(String(d)));
        ws.on("close", () => {
          this.closed = true;
          for (const p of this.pending.values()) p.reject(new Error("electrum connection closed"));
          this.pending.clear();
          if (this.onClose) this.onClose();
        });
      });
    }
    _onData(chunk) {
      this.buf += chunk;
      let nl;
      while ((nl = this.buf.indexOf("\n")) >= 0) {
        const line = this.buf.slice(0, nl).trim();
        this.buf = this.buf.slice(nl + 1);
        if (line) this._handleLine(line);
      }
      const rest = this.buf.trim();
      if (rest) { try { JSON.parse(rest); this._handleLine(rest); this.buf = ""; } catch {} }
    }
    _handleLine(line) {
      let msg;
      try { msg = JSON.parse(line); } catch { return; }
      if (msg.id != null && this.pending.has(msg.id)) {
        const p = this.pending.get(msg.id);
        this.pending.delete(msg.id);
        clearTimeout(p.timer);
        if (msg.error) p.reject(new Error(typeof msg.error === "object" ? (msg.error.message || JSON.stringify(msg.error)) : String(msg.error)));
        else p.resolve(msg.result);
      } else if (msg.method && this.onNotify) {
        this.onNotify(msg.method, msg.params || []);
      }
    }
    call(method, params = []) {
      if (this.closed) return Promise.reject(new Error("electrum connection closed"));
      const id = ++this.id;
      return new Promise((resolve, reject) => {
        const timer = setTimeout(() => {
          if (this.pending.has(id)) { this.pending.delete(id); reject(new Error(`electrum timeout: ${method}`)); }
        }, CALL_TIMEOUT_MS);
        this.pending.set(id, { resolve, reject, timer });
        try { this.ws.send(JSON.stringify({ id, method, params }) + "\n"); }
        catch (e) { clearTimeout(timer); this.pending.delete(id); reject(e); }
      });
    }
    close() { this.closed = true; try { this.ws.close(); } catch {} }
  }

  class Client {
    constructor(servers) {
      this.servers = servers.slice();
      this.conn = null;
      this.connecting = null;
      this.subscriptions = new Map();   // method+key -> params (replayed on reconnect)
      this.onNotify = null;
      this.onServer = null;             // (url|null) connection state for the UI
    }
    setServers(servers) {
      this.servers = servers.slice();
      this.disconnect();
    }
    get url() { return this.conn && !this.conn.closed ? this.conn.url : null; }
    async _ensure() {
      if (this.conn && !this.conn.closed) return this.conn;
      if (this.connecting) return this.connecting;
      this.connecting = (async () => {
        let lastErr;
        for (const url of this.servers) {
          try {
            const c = await new Connection(url).connect();
            c.onNotify = (m, p) => { if (this.onNotify) this.onNotify(m, p); };
            c.onClose = () => { if (this.conn === c) { this.conn = null; if (this.onServer) this.onServer(null); } };
            this.conn = c;
            log("connected", url);
            if (this.onServer) this.onServer(url);
            // Re-arm subscriptions so a reconnect keeps the live feed.
            for (const params of this.subscriptions.values()) c.call(params[0], params[1]).catch(() => {});
            return c;
          } catch (e) { lastErr = e; log("failed", url, e?.message); }
        }
        throw lastErr || new Error("no electrum server reachable");
      })();
      try { return await this.connecting; }
      finally { this.connecting = null; }
    }
    async call(method, params = []) {
      const c = await this._ensure();
      return c.call(method, params);
    }
    // Remember a subscription so it survives reconnects.
    async subscribe(method, params = []) {
      this.subscriptions.set(method + ":" + JSON.stringify(params), [method, params]);
      return this.call(method, params);
    }
    clearSubscriptions() { this.subscriptions.clear(); }
    disconnect() {
      if (this.conn) { const c = this.conn; this.conn = null; c.close(); }
      if (this.onServer) this.onServer(null);
    }
  }

  return { Client };
};