WebSocket connection and runtime

npm ci
export API_BASE_URL="https://gateway.example.com/v1"
export API_KEY="YOUR_API_KEY"
export API_MODEL="YOUR_MODEL_ID"
export API_WS_TIMEOUT_MS="60000"

import WebSocket from 'ws';

// Shared by both server-side examples. No reconnect or automatic retry.
export async function openWebSocket(endpoint, query = {}) {
  const key = process.env.API_KEY;
  const model = process.env.API_MODEL;
  const timeout = Number(process.env.API_WS_TIMEOUT_MS ?? 60000);
  if (!key?.trim() || !model?.trim() || !Number.isInteger(timeout) || timeout < 100 || timeout > 120000) {
    throw new Error('Set API_KEY, API_MODEL, and a 100..120000 ms timeout.');
  }
  let url;
  try { url = new URL(process.env.API_BASE_URL); }
  catch { throw new Error('Set a valid HTTP(S) API_BASE_URL including /v1.'); }
  if (!['http:', 'https:'].includes(url.protocol) || url.username || url.password || url.search || url.hash) {
    throw new Error('API_BASE_URL must use HTTP(S) without credentials, query, or fragment.');
  }
  url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:';
  url.pathname = `${url.pathname.replace(/\/+$/, '')}/${endpoint}`;
  for (const [name, value] of Object.entries(query)) url.searchParams.set(name, value);
  const ws = new WebSocket(url, {
    headers: { Authorization: `Bearer ${key}` },
    followRedirects: false, perMessageDeflate: false,
    handshakeTimeout: timeout, maxPayload: 1024 * 1024,
  });
  const queue = [];
  let queuedBytes = 0;
  let waiting;
  let fault;
  let closed = false;
  let rejectOpen;
  const opened = new Promise((resolve, reject) => {
    rejectOpen = reject;
    ws.once('open', resolve);
  });
  function fail(message) {
    if (fault || closed) return;
    fault = new Error(message);
    rejectOpen(fault);
    waiting?.reject(fault);
    waiting = undefined;
    ws.terminate();
  }
  const deadline = setTimeout(() => fail('WebSocket workflow deadline exceeded.'), timeout);
  ws.on('unexpected-response', (_request, response) => {
    response.resume();
    fail(`WebSocket handshake failed (HTTP ${response.statusCode}).`);
  });
  ws.on('error', () => fail('WebSocket transport failed.'));
  ws.on('close', () => fail('WebSocket closed before workflow completion.'));
  ws.on('message', (data, isBinary) => {
    if (closed || fault) return;
    if (isBinary) { fail('Expected a JSON text message, not a binary frame.'); return; }
    let event;
    try { event = JSON.parse(data.toString()); }
    catch { fail('Invalid JSON event.'); return; }
    if (!event || typeof event !== 'object' || Array.isArray(event) || typeof event.type !== 'string') {
      fail('Invalid event envelope.'); return;
    }
    if (event.type === 'error') { fail('Server returned a WebSocket error event.'); return; }
    if (waiting) {
      const pending = waiting;
      waiting = undefined;
      pending.resolve(event);
      return;
    }
    queuedBytes += data.length;
    if (queue.length >= 256 || queuedBytes > 4 * 1024 * 1024) {
      fail('WebSocket receive queue limit exceeded.'); return;
    }
    queue.push({ event, bytes: data.length });
  });
  try { await opened; }
  catch (error) { clearTimeout(deadline); throw error; }
  return {
    model,
    async send(event) {
      if (fault) throw fault;
      if (closed || ws.readyState !== WebSocket.OPEN) throw new Error('WebSocket is not open.');
      await new Promise((resolve, reject) => ws.send(JSON.stringify(event), (error) => {
        if (error) reject(new Error('WebSocket send failed.'));
        else resolve();
      }));
    },
    async receive() {
      if (fault) throw fault;
      if (closed) throw new Error('WebSocket is closed.');
      if (queue.length) {
        const next = queue.shift();
        queuedBytes -= next.bytes;
        return next.event;
      }
      if (waiting) throw new Error('Only one event reader is allowed.');
      return await new Promise((resolve, reject) => { waiting = { resolve, reject }; });
    },
    close() {
      closed = true;
      clearTimeout(deadline);
      waiting?.reject(new Error('WebSocket reader closed.'));
      waiting = undefined;
      ws.close(1000);
      // Bound shutdown even if the peer never acknowledges the close frame.
      const cleanup = setTimeout(() => ws.terminate(), 1000);
      cleanup.unref();
      ws.once('close', () => clearTimeout(cleanup));
    },
  };
}