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.
//   * Per-relay status is exposed via onStatus() so the UI can show progress.

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

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.subs = new Map();       // subId -> { req, handler }
    this.recentPublishes = [];   // { msg, t }
    this.subId = 0;
    this._listeners = new Set(); // status change listeners
  }

  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);
  }

  _open(url) {
    return new Promise((resolve) => {
      let settled = false;
      const finish = (ok) => {
        if (settled) return;
        settled = true;
        resolve(ok);
      };

      let ws;
      try {
        ws = new WebSocket(url);
      } catch {
        this.status.set(url, "closed");
        this._notify();
        return finish(false);
      }

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

      const timer = setTimeout(() => {
        if (ws.readyState === WebSocket.CONNECTING) {
          this.status.set(url, "timeout");
          this._notify();
          try { ws.close(); } catch {}
          finish(false);
        }
      }, PER_RELAY_CONNECT_TIMEOUT_MS);

      ws.addEventListener("open", () => {
        clearTimeout(timer);
        const t0 = performance.now();
        this.sockets.set(url, ws);
        this.status.set(url, "open");
        this._notify();
        console.log(`[relay] open ${url} (${Math.round(t0)}ms since page)`);

        // 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("error", () => {
        clearTimeout(timer);
        if (this.status.get(url) !== "open") this.status.set(url, "closed");
        this._notify();
        finish(false);
      });

      ws.addEventListener("close", () => {
        clearTimeout(timer);
        this.sockets.delete(url);
        if (this.status.get(url) !== "timeout") this.status.set(url, "closed");
        this._notify();
        finish(false);
      });

      // Wire messages to active subscription handlers.
      ws.addEventListener("message", (ev) => {
        let msg;
        try { msg = JSON.parse(ev.data); } catch { return; }
        const subId = msg[1];
        const sub = this.subs.get(subId);
        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 } = {}) {
    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;
        resolve(this.sockets.size);
      };

      const ceiling = setTimeout(finish, OVERALL_CONNECT_TIMEOUT_MS);

      attempts.forEach((p) =>
        p.then((ok) => {
          if (ok) opened++;
          if (opened >= minRelays) {
            clearTimeout(ceiling);
            // Resolve quickly once we hit the threshold; other relays keep
            // opening in the background and self-register via _open().
            finish();
          }
          if (opened === 0 && attempts.every((_) => false)) {
            /* unreachable; here for clarity */
          }
        })
      );

      // If every attempt fully settled with zero opens, resolve too.
      Promise.allSettled(attempts).then(() => finish());
    });
  }

  publish(event) {
    const msg = JSON.stringify(["EVENT", event]);
    this.recentPublishes.push({ msg, t: Date.now() });
    // trim
    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 match. onEose() once
  // every currently-open relay has signaled EOSE.
  subscribe(filters, onEvent, onEose) {
    const id = `s${++this.subId}`;
    const req = JSON.stringify(["REQ", id, ...filters]);
    const seen = new Set();
    const eoseSet = new Set();

    const handler = (msg, url) => {
      if (msg[0] === "EVENT" && msg[1] === id) {
        const e = msg[2];
        if (seen.has(e.id)) return;
        seen.add(e.id);
        onEvent(e);
      } else if (msg[0] === "EOSE" && msg[1] === id) {
        eoseSet.add(url);
        if (eoseSet.size >= this.sockets.size && onEose) onEose();
      }
    };

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

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

    return () => {
      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();