/**
 * Bounded consumer for the filings websocket (`wss://…/v1/stream`).
 *
 * Opens the socket, sends one `subscribe` message, then collects messages until either
 * `maxMessages` are received or `timeoutMs` elapses, and closes. Uses the global `WebSocket`
 * (Node >= 22, or Node 20 with `--experimental-websocket`).
 */

export interface SubscribeOptions {
  url: string;
  channel?: string;
  tickers?: string[] | 'all';
  forms?: string[];
  apiKey?: string;
  maxMessages?: number;
  timeoutMs?: number;
  /** Injected for tests. */
  WebSocketImpl?: typeof WebSocket;
}

export interface SubscribeResult {
  url: string;
  subscription: Record<string, unknown>;
  messages: unknown[];
  stopped_because: 'max_messages' | 'timeout' | 'closed' | 'error';
  elapsed_ms: number;
  error?: string;
}

export const MAX_TIMEOUT_MS = 120_000;
export const MAX_MESSAGES = 100;

export async function subscribeFilings(opts: SubscribeOptions): Promise<SubscribeResult> {
  const WS = opts.WebSocketImpl ?? (globalThis as any).WebSocket as typeof WebSocket | undefined;
  if (!WS) {
    throw new Error('WebSocket is not available in this Node.js runtime. Use Node >= 22, or run Node 20 with --experimental-websocket.');
  }
  const timeoutMs = Math.min(Math.max(opts.timeoutMs ?? 30_000, 1_000), MAX_TIMEOUT_MS);
  const maxMessages = Math.min(Math.max(opts.maxMessages ?? 10, 1), MAX_MESSAGES);
  const subscription: Record<string, unknown> = { action: 'subscribe', channel: opts.channel ?? 'filings', tickers: opts.tickers ?? 'all' };
  if (opts.forms?.length) subscription.forms = opts.forms;
  if (opts.apiKey) subscription.api_key = opts.apiKey;

  const url = new URL(opts.url);
  if (opts.apiKey) url.searchParams.set('api_key', opts.apiKey); // servers commonly accept the key as a query param on ws
  const t0 = Date.now();
  const messages: unknown[] = [];

  return new Promise<SubscribeResult>((resolve) => {
    let done = false;
    let ws: WebSocket;
    const finish = (why: SubscribeResult['stopped_because'], error?: string) => {
      if (done) return;
      done = true;
      clearTimeout(timer);
      try { ws?.close(); } catch { /* ignore */ }
      const publicSub = { ...subscription };
      delete publicSub.api_key;
      resolve({ url: opts.url, subscription: publicSub, messages, stopped_because: why, elapsed_ms: Date.now() - t0, error });
    };
    const timer = setTimeout(() => finish('timeout'), timeoutMs);
    try {
      ws = new WS(url.toString());
    } catch (e) {
      finish('error', e instanceof Error ? e.message : String(e));
      return;
    }
    ws.addEventListener('open', () => ws.send(JSON.stringify(subscription)));
    ws.addEventListener('message', (ev: MessageEvent) => {
      const raw = typeof ev.data === 'string' ? ev.data : String(ev.data);
      let parsed: unknown = raw;
      try { parsed = JSON.parse(raw); } catch { /* keep raw text */ }
      messages.push(parsed);
      if (messages.length >= maxMessages) finish('max_messages');
    });
    ws.addEventListener('error', () => finish('error', 'websocket error (is the stream endpoint deployed? check hfmarketdata://status)'));
    ws.addEventListener('close', () => finish('closed'));
  });
}