Googstr/public/relay.js

// Minimal nostr relay pool — fast-start, fault-tolerant.
//
// Design:
//   * Open all relays in parallel, each with a 4s connect timeout.
//   * connectAll() resolves as soon as the FIRST relay opens (or after the
//     overall timeout). The rest keep opening in the background.
//   * Late-arriving relays automatically replay anything we've already sent
//     (active subscriptions + recent publishes), so the user never waits.
//   * Dropped relays reconnect with backoff (2s → 60s), and immediately when
//     the browser comes back online. The NIP-46 signer relies on this.
//   * Every incoming event's signature is verified before handlers see it.
//   * Per-relay status is exposed via onStatus() so the UI can show progress.

import { verifyEvent } from "./vendor/nostr-tools.js";
import { RELAYS } from "./config.js";

const PER_RELAY_CONNECT_TIMEOUT_MS = 4000;
const OVERALL_CONNECT_TIMEOUT_MS = 4500; // hard ceiling for connectAll()
const PUBLISH_REPLAY_WINDOW_MS = 60_000; // re-send recent publishes to late relays
const EOSE_TIMEOUT_MS = 5000;            // onEose fires by then even if a relay is silent
const MAX_RETRY_DELAY_MS = 60_000;

class RelayPool {
  constructor(urls = RELAYS) {
    this.urls = [...urls];
    this.sockets = new Map();    // url -> WebSocket (only OPEN ones)
    this.status = new Map();     // url -> "connecting"|"open"|"closed"|"timeout"
    this.retries = new Map();    // url -> consecutive failures
    this.retryTimers = new Map();
    this.subs = new Map();       // subId -> { req, handler }
    this.recentPublishes = [];   // { msg, t }
    this.subId = 0;
    this._listeners = new Set(); // status change listeners
    this._started = false;

    addEventListener("online", () => this._reconnectNow());
    document.addEventListener("visibilitychange", () => {
      if (document.visibilityState === "visible") this._reconnectNow();
    });
  }

  onStatus(fn) {
    this._listeners.add(fn);
    fn(this.statusSnapshot());
    return () => this._listeners.delete(fn);
  }

  statusSnapshot() {
    const out = {};
    for (const u of this.urls) out[u] = this.status.get(u) || "idle";
    return out;
  }

  _notify() {
    const snap = this.statusSnapshot();
    for (const fn of this._listeners) fn(snap);
  }

  _scheduleRetry(url) {
    if (!this._started || this.retryTimers.has(url)) return;
    const n = (this.retries.get(url) || 0) + 1;
    this.retries.set(url, n);
    const delay = Math.min(MAX_RETRY_DELAY_MS, 2000 * 2 ** (n - 1));
    this.retryTimers.set(url, setTimeout(() => {
      this.retryTimers.delete(url);
      this._open(url);
    }, delay));
  }

  _reconnectNow() {
    if (!this._started) return;
    for (const url of this.urls) {
      if (this.status.get(url) === "open" || this.status.get(url) === "connecting") continue;
      clearTimeout(this.retryTimers.get(url));
      this.retryTimers.delete(url);
      this.retries.set(url, 0);
      this._open(url);
    }
  }

  _open(url) {
    return new Promise((resolve) => {
      let settled = false;
      const finish = (ok) => {
        if (settled) return;
        settled = true;
        resolve(ok);
      };
      const failed = (state) => {
        this.sockets.delete(url);
        this.status.set(url, state);
        this._notify();
        this._scheduleRetry(url);
        finish(false);
      };

      let ws;
      try {
        ws = new WebSocket(url);
      } catch {
        return failed("closed");
      }

      this.status.set(url, "connecting");
      this._notify();

      const timer = setTimeout(() => {
        if (ws.readyState === WebSocket.CONNECTING) {
          try { ws.close(); } catch {}
          failed("timeout");
        }
      }, PER_RELAY_CONNECT_TIMEOUT_MS);

      ws.addEventListener("open", () => {
        clearTimeout(timer);
        this.sockets.set(url, ws);
        this.status.set(url, "open");
        this.retries.set(url, 0);
        this._notify();

        // Replay active subscriptions and recent publishes to this new relay.
        for (const { req } of this.subs.values()) {
          try { ws.send(req); } catch {}
        }
        const cutoff = Date.now() - PUBLISH_REPLAY_WINDOW_MS;
        for (const { msg, t } of this.recentPublishes) {
          if (t >= cutoff) {
            try { ws.send(msg); } catch {}
          }
        }
        finish(true);
      });

      ws.addEventListener("close", () => {
        clearTimeout(timer);
        if (this.sockets.get(url) === ws || !settled) {
          failed(this.status.get(url) === "timeout" ? "timeout" : "closed");
        }
      });

      // Wire messages to active subscription handlers.
      ws.addEventListener("message", (ev) => {
        let msg;
        try { msg = JSON.parse(ev.data); } catch { return; }
        const sub = this.subs.get(msg[1]);
        if (sub) sub.handler(msg, url);
      });
    });
  }

  // Resolves as soon as we have at least `minRelays` open, or after the
  // overall timeout. Always returns the number of relays that opened.
  async connectAll({ minRelays = 1 } = {}) {
    this._started = true;
    const attempts = this.urls.map((u) => this._open(u));
    let opened = 0;

    return new Promise((resolve) => {
      let done = false;
      const finish = () => {
        if (done) return;
        done = true;
        clearTimeout(ceiling);
        resolve(this.sockets.size);
      };
      const ceiling = setTimeout(finish, OVERALL_CONNECT_TIMEOUT_MS);
      attempts.forEach((p) => p.then((ok) => { if (ok && ++opened >= minRelays) finish(); }));
      Promise.allSettled(attempts).then(finish);
    });
  }

  // Add a relay at runtime (e.g. one named in a nostrconnect:// link).
  addRelay(url) {
    if (!/^wss?:\/\//.test(url) || this.urls.includes(url)) return;
    this.urls.push(url);
    this._open(url);
  }

  publish(event) {
    const msg = JSON.stringify(["EVENT", event]);
    this.recentPublishes.push({ msg, t: Date.now() });
    if (this.recentPublishes.length > 200) this.recentPublishes.shift();
    for (const ws of this.sockets.values()) {
      if (ws.readyState === WebSocket.OPEN) {
        try { ws.send(msg); } catch {}
      }
    }
  }

  // filters: array of REQ filters. onEvent(event) per verified match. onEose(n)
  // fires once: when every relay open at that moment has sent EOSE, or after
  // EOSE_TIMEOUT_MS, whichever is first.
  subscribe(filters, onEvent, onEose) {
    const id = `s${++this.subId}`;
    const req = JSON.stringify(["REQ", id, ...filters]);
    const seen = new Set();
    const eoseFrom = new Set();
    let eoseFired = false;
    const fireEose = () => {
      if (eoseFired) return;
      eoseFired = true;
      clearTimeout(eoseTimer);
      onEose?.(eoseFrom.size); // how many relays actually answered
    };
    const eoseTimer = setTimeout(fireEose, EOSE_TIMEOUT_MS);

    const handler = (msg, url) => {
      if (msg[0] === "EVENT") {
        const e = msg[2];
        if (!e || seen.has(e.id) || !verifyEvent(e)) return;
        seen.add(e.id);
        onEvent(e);
      } else if (msg[0] === "EOSE") {
        eoseFrom.add(url);
        if ([...this.sockets.keys()].every((u) => eoseFrom.has(u))) fireEose();
      }
    };

    this.subs.set(id, { req, handler });

    for (const ws of this.sockets.values()) {
      if (ws.readyState === WebSocket.OPEN) {
        try { ws.send(req); } catch {}
      }
    }

    return () => {
      clearTimeout(eoseTimer);
      const close = JSON.stringify(["CLOSE", id]);
      for (const ws of this.sockets.values()) {
        if (ws.readyState === WebSocket.OPEN) {
          try { ws.send(close); } catch {}
        }
      }
      this.subs.delete(id);
    };
  }
}

export const pool = new RelayPool();