spb/hfmarketdata

1/**2 * Bounded consumer for the filings websocket (`wss://…/v1/stream`).3 *4 * Opens the socket, sends one `subscribe` message, then collects messages until either5 * `maxMessages` are received or `timeoutMs` elapses, and closes. Uses the global `WebSocket`6 * (Node >= 22, or Node 20 with `--experimental-websocket`).7 */89export interface SubscribeOptions {10  url: string;11  channel?: string;12  tickers?: string[] | 'all';13  forms?: string[];14  apiKey?: string;15  maxMessages?: number;16  timeoutMs?: number;17  /** Injected for tests. */18  WebSocketImpl?: typeof WebSocket;19}2021export interface SubscribeResult {22  url: string;23  subscription: Record<string, unknown>;24  messages: unknown[];25  stopped_because: 'max_messages' | 'timeout' | 'closed' | 'error';26  elapsed_ms: number;27  error?: string;28}2930export const MAX_TIMEOUT_MS = 120_000;31export const MAX_MESSAGES = 100;3233export async function subscribeFilings(opts: SubscribeOptions): Promise<SubscribeResult> {34  const WS = opts.WebSocketImpl ?? (globalThis as any).WebSocket as typeof WebSocket | undefined;35  if (!WS) {36    throw new Error('WebSocket is not available in this Node.js runtime. Use Node >= 22, or run Node 20 with --experimental-websocket.');37  }38  const timeoutMs = Math.min(Math.max(opts.timeoutMs ?? 30_000, 1_000), MAX_TIMEOUT_MS);39  const maxMessages = Math.min(Math.max(opts.maxMessages ?? 10, 1), MAX_MESSAGES);40  const subscription: Record<string, unknown> = { action: 'subscribe', channel: opts.channel ?? 'filings', tickers: opts.tickers ?? 'all' };41  if (opts.forms?.length) subscription.forms = opts.forms;42  if (opts.apiKey) subscription.api_key = opts.apiKey;4344  const url = new URL(opts.url);45  if (opts.apiKey) url.searchParams.set('api_key', opts.apiKey); // servers commonly accept the key as a query param on ws46  const t0 = Date.now();47  const messages: unknown[] = [];4849  return new Promise<SubscribeResult>((resolve) => {50    let done = false;51    let ws: WebSocket;52    const finish = (why: SubscribeResult['stopped_because'], error?: string) => {53      if (done) return;54      done = true;55      clearTimeout(timer);56      try { ws?.close(); } catch { /* ignore */ }57      const publicSub = { ...subscription };58      delete publicSub.api_key;59      resolve({ url: opts.url, subscription: publicSub, messages, stopped_because: why, elapsed_ms: Date.now() - t0, error });60    };61    const timer = setTimeout(() => finish('timeout'), timeoutMs);62    try {63      ws = new WS(url.toString());64    } catch (e) {65      finish('error', e instanceof Error ? e.message : String(e));66      return;67    }68    ws.addEventListener('open', () => ws.send(JSON.stringify(subscription)));69    ws.addEventListener('message', (ev: MessageEvent) => {70      const raw = typeof ev.data === 'string' ? ev.data : String(ev.data);71      let parsed: unknown = raw;72      try { parsed = JSON.parse(raw); } catch { /* keep raw text */ }73      messages.push(parsed);74      if (messages.length >= maxMessages) finish('max_messages');75    });76    ws.addEventListener('error', () => finish('error', 'websocket error (is the stream endpoint deployed? check hfmarketdata://status)'));77    ws.addEventListener('close', () => finish('closed'));78  });79}80