JavaScript

/*
 * A WebSocket client for the RUAL cluster, in plain JavaScript.
 *
 * Works unmodified in a browser and in Node 22+ (which has a global
 * WebSocket). No dependencies.
 *
 * The socket carries a request/response protocol on top of plain WebSocket
 * frames. You send {method, uri, requestid} and the reply comes back carrying
 * the same requestid, so several requests can be in flight at once. The server
 * also pushes broadcasts that were never requested.
 */

const DEFAULT_OPTIONS = {
  // The socket lives in rual-core itself, on the cluster domain.
  //
  // Note the path is /ws. An older generation of this example used a separate
  // rual-ws host with a /primus path; that service is gone. If you still have
  // that URL somewhere, it will fail in a way that is easy to misread: the
  // hostname usually has no record of its own, resolves through the wildcard
  // to a load balancer, and gets a 301 redirect. A WebSocket handshake cannot
  // follow a redirect, so the socket never opens and nothing reports an error.
  'url': 'wss://{{cluster-domain}}/ws',

  // Reconnect backoff. Doubles from min to max.
  'reconnectMinMs': 500,
  'reconnectMaxMs': 30000,

  // How long a connection must stay up before its backoff is forgiven. See
  // scheduleReconnect for why this is not simply reset when the socket opens.
  'stableAfterMs': 10000,

  // How long to wait for a reply before rejecting the request.
  'requestTimeoutMs': 15000
};

export default class RualWebSocket {
  constructor (options = {}) {
    this.options = Object.assign({}, DEFAULT_OPTIONS, options);
    this.accessToken = options.accessToken || null;

    this.connection = null;
    this.socketId = null;
    this.requestId = 0;
    this.openRequests = new Map();

    this.reconnecting = false;
    this.reconnectAttempts = 0;
    this.stableTimer = null;
    this.closedByUs = false;

    // Broadcast handlers, keyed by the broadcast payload type.
    this.listeners = new Map();
  }

  /* ---------------------------------------------------------------- connect */

  connect () {
    if (this.connection) {
      const state = this.connection.readyState;
      if (state === WebSocket.CONNECTING || state === WebSocket.OPEN) {
        return;
      }
    }

    this.closedByUs = false;
    this.requestId = 0;
    this.connection = new WebSocket(this.options.url);

    this.connection.onopen = () => {
      console.log('[ws] connected');

      /*
       * Forgive the backoff only once this connection has STAYED up.
       *
       * A backend that is draining will accept you, assign an id, and then
       * immediately announce that it is going down. If the backoff resets
       * whenever a socket opens, it never grows, and you reconnect in a tight
       * loop against a server that is trying to shut down.
       */
      clearTimeout(this.stableTimer);
      this.stableTimer = setTimeout(() => {
        this.reconnectAttempts = 0;
      }, this.options.stableAfterMs);

      this.onOpen();
    };

    this.connection.onmessage = (event) => {
      this.handleMessage(event);
    };

    this.connection.onerror = () => {
      // Do not merely drop the socket. Without a scheduled reconnect nothing
      // reopens it, and because broadcasts are pushed rather than polled, the
      // client goes quiet instead of going wrong. That is much harder to spot.
      this.scheduleReconnect();
    };

    this.connection.onclose = () => {
      this.onClose();
      this.scheduleReconnect();
    };
  }

  /*
   * Tear the socket down for good. Call this when you are finished with the
   * client, for example when unmounting the view that owns it.
   */
  close () {
    this.closedByUs = true;
    this.teardown();
  }

  teardown () {
    clearTimeout(this.stableTimer);
    this.stableTimer = null;

    if (this.connection) {
      /*
       * Detach the handlers BEFORE closing.
       *
       * Otherwise close() fires our own onclose, which schedules a reconnect,
       * and a socket we deliberately tore down comes back from the dead. Worse,
       * the old instance keeps its own reconnect loop running alongside the new
       * one and the duplicates multiply.
       */
      const socket = this.connection;
      socket.onopen = null;
      socket.onmessage = null;
      socket.onerror = null;
      socket.onclose = null;
      try {
        socket.close();
      } catch (err) {
        // Already closing or closed. Nothing to do.
      }
    }

    this.connection = null;
    this.socketId = null;
  }

  /*
   * One reconnect cycle at a time, backed off exponentially and capped.
   *
   * The guard matters: close and error can both fire for the same failure, and
   * without it each would schedule its own timer, each timer would open its own
   * socket, and you would end up with several live connections all reconnecting.
   */
  scheduleReconnect () {
    if (this.closedByUs || this.reconnecting) {
      return;
    }
    this.reconnecting = true;

    this.teardown();

    this.reconnectAttempts += 1;
    const delay = Math.min(
      this.options.reconnectMinMs * Math.pow(2, (this.reconnectAttempts - 1)),
      this.options.reconnectMaxMs
    );
    console.log('[ws] reconnecting in ' + delay + 'ms (attempt ' + this.reconnectAttempts + ')');

    setTimeout(() => {
      this.reconnecting = false;
      if (!this.closedByUs) {
        this.connect();
      }
    }, delay);
  }

  async waitForConnected () {
    if (this.connection && this.connection.readyState === WebSocket.OPEN) {
      return;
    }

    if (!this.connection && !this.reconnecting) {
      this.connect();
    }

    // Poll rather than recurse: an unbounded recursive wait keeps a stack
    // frame per attempt and never gives up.
    const deadline = Date.now() + this.options.requestTimeoutMs;
    while (Date.now() < deadline) {
      if (this.connection && this.connection.readyState === WebSocket.OPEN) {
        return;
      }
      await new Promise((resolve) => {
        setTimeout(resolve, 100);
      });
    }

    throw new Error('WS_NOT_CONNECTED');
  }

  /* ---------------------------------------------------------------- request */

  /*
   * Send a request and resolve with its result.
   *
   *   const me = await ws.request('GET', 'v1/users/me');
   *   await ws.request('POST', 'v1/some/endpoint', { body: { name: 'x' } });
   */
  async request (method, uri, options = {}) {
    await this.waitForConnected();

    this.requestId += 1;
    const requestid = 'RN' + String(this.requestId).padStart(10, '0');

    const payload = {
      'method': String(method).toUpperCase(),
      'requestid': requestid,
      'uri': '/api/' + uri
    };

    const token = options.accessToken || this.accessToken;
    if (token) {
      payload.access_token = token;
    }
    if (options.body) {
      payload.body = options.body;
    }
    if (options.query) {
      payload.query = options.query;
    }

    this.connection.send(JSON.stringify(payload));

    return new Promise((resolve, reject) => {
      // Always time out. Without this, a reply that never arrives leaks a
      // pending promise and its entry in openRequests, forever.
      const timer = setTimeout(() => {
        this.openRequests.delete(requestid);
        reject(new Error('WS_REQUEST_TIMEOUT'));
      }, this.options.requestTimeoutMs);

      this.openRequests.set(requestid, {
        'resolve': resolve,
        'reject': reject,
        'method': method,
        'uri': uri,
        'timer': timer
      });
    });
  }

  /* -------------------------------------------------------------- broadcast */

  /*
   * Register a handler for a broadcast type. Returns a function that removes
   * it again.
   */
  on (type, handler) {
    if (!this.listeners.has(type)) {
      this.listeners.set(type, new Set());
    }
    this.listeners.get(type).add(handler);

    return () => {
      const handlers = this.listeners.get(type);
      if (handlers) {
        handlers.delete(handler);
      }
    };
  }

  emit (type, payload) {
    const handlers = this.listeners.get(type);
    if (!handlers) {
      return;
    }
    handlers.forEach((handler) => {
      try {
        handler(payload);
      } catch (err) {
        console.error('[ws] broadcast handler for ' + type + ' threw', err);
      }
    });
  }

  /* ---------------------------------------------------------------- inbound */

  handleMessage (message) {
    if (!message.data) {
      return;
    }

    let data = null;
    try {
      data = JSON.parse(message.data);
    } catch (err) {
      console.error('[ws] received a frame that is not JSON');
      return;
    }

    // Heartbeat. The server sends a bare string and expects it echoed back
    // with "ping" swapped for "pong".
    if (typeof data === 'string') {
      this.replyPing(message.data);
      return;
    }

    // The server greets every new socket with the id it assigned. Keep it:
    // when broadcasts go missing, being able to say which socket you are on is
    // the difference between a five minute diagnosis and an hour of guessing.
    if (data.type === '_id') {
      this.socketId = data.id;
      console.log('[ws] socket id ' + data.id);
      return;
    }

    // The backend announces its own shutdown before it drops you. Acting on it
    // means reconnecting to whichever node comes up next, rather than waiting
    // for a TCP close that can take considerably longer to notice.
    if (data.server === 'down') {
      console.log('[ws] server announced it is going down');
      this.scheduleReconnect();
      return;
    }

    if (data.requestid) {
      this.resolveRequest(data);
      return;
    }

    if (data.broadcast) {
      const items = Array.isArray(data.broadcast) ? data.broadcast : [data.broadcast];
      items.forEach((item) => {
        const payload = item.message || item || {};
        this.emit(payload.type, payload);
      });
    }
  }

  resolveRequest (data) {
    const pending = this.openRequests.get(data.requestid);
    if (!pending) {
      // Usually a reply to a request that already timed out.
      console.log('[ws] reply for unknown request ' + data.requestid);
      return;
    }

    clearTimeout(pending.timer);
    this.openRequests.delete(data.requestid);

    if (data.code === 200 || data.code === 201) {
      pending.resolve(data.result);
      return;
    }

    // Reject with a real Error so a stack trace survives, and hang the server
    // detail off it rather than throwing a bare object.
    const result = data.result || {};
    const err = new Error(result.error || ('HTTP_' + data.code));
    err.code = data.code;
    err.result = result;
    err.uri = pending.uri;
    err.method = pending.method;
    pending.reject(err);
  }

  replyPing (raw) {
    if (!this.connection || this.connection.readyState !== WebSocket.OPEN) {
      return;
    }
    this.connection.send(raw.replace('ping', 'pong'));
  }

  /* ------------------------------------------------------------- overridable */

  // Called once the socket is open. Subscribe to what you need here: it runs
  // again after every reconnect, which is exactly when subscriptions have to
  // be re-established.
  onOpen () {}

  // Called whenever the socket closes, for any reason.
  onClose () {
    console.log('[ws] disconnected');
  }
}

/*
 * Usage
 *
 *   const ws = new RualWebSocket({ accessToken: myToken });
 *
 *   ws.onOpen = async () => {
 *     // Re-run on every reconnect, not only the first connect.
 *     await ws.request('GET', 'v1/ws/subscribe');
 *   };
 *
 *   ws.on('some-broadcast-type', (payload) => {
 *     console.log('received', payload);
 *   });
 *
 *   ws.connect();
 *
 *   const me = await ws.request('GET', 'v1/users/me');
 *
 *   // When you are finished:
 *   ws.close();
 */

const ws = new RualWebSocket({ accessToken: myToken });

ws.onOpen = async () => {
  // Runs on every reconnect, not only the first connect.
  await ws.request('GET', 'v1/ws/subscribe');
};

ws.on('some-broadcast-type', (payload) => {
  console.log('received', payload);
});

ws.connect();

const me = await ws.request('GET', 'v1/users/me');

// When you are finished with the client:
ws.close();