messenzy-widget/src/transport/ws-client.ts

import type { WidgetConfig } from '../config.js';

export type WsMessage =
  | { type: 'welcome'; visitor_id: string }
  | {
      type: 'history';
      messages: Array<{
        id: string;
        from: 'bot' | 'user';
        text: string;
        at: string;
      }>;
    }
  | { type: 'bot_message'; id: string; text: string; created_at: string }
  | { type: 'typing'; state: 'on' | 'off' }
  | { type: 'handoff'; message: string }
  | { type: 'error'; code: string; retry_after_s?: number }
  | { type: 'pong' };

export type ClientMessage =
  | { type: 'user_message'; id: string; text: string }
  | { type: 'ping' };

export type Transport = {
  send: (msg: ClientMessage) => void;
  close: () => void;
  isConnected: () => boolean;
};

type TransportOpts = {
  botId: string;
  apiKey: string;
  visitorId: string;
  serverUrl: string;
  onMessage: (msg: WsMessage) => void;
  onConnectionChange?: (connected: boolean) => void;
};

const MAX_WS_RETRIES = 5;
const POLL_INTERVAL_MS = 5000;
const PING_INTERVAL_MS = 30000;

export function createTransport(opts: TransportOpts): Transport {
  let ws: WebSocket | null = null;
  let pollTimer: ReturnType<typeof setInterval> | null = null;
  let pingTimer: ReturnType<typeof setInterval> | null = null;
  let reconnectAttempts = 0;
  let pendingMessages: ClientMessage[] = [];
  let lastSeenAt: string | null = null;
  let usingPolling = false;
  let closed = false;

  function connectWs() {
    if (closed) return;
    // The browser WebSocket constructor does NOT accept custom headers, so
    // we encode credentials into the Sec-WebSocket-Protocol list (the
    // standard workaround). The bot reads `req.headers['sec-websocket-
    // protocol']` and parses these tokens. See bot's `webchat/subprotocol.ts`.
    const wsUrl = opts.serverUrl.replace(/^http/, 'ws') + '/webchat/ws';
    const subprotocols = [
      'messenzy.v1',
      `messenzy-bot.${opts.botId}`,
      `messenzy-visitor.${opts.visitorId}`,
      `messenzy-key.${opts.apiKey}`,
    ];
    try {
      ws = new WebSocket(wsUrl, subprotocols);
    } catch {
      onWsClose();
      return;
    }

    ws.onopen = () => {
      reconnectAttempts = 0;
      opts.onConnectionChange?.(true);
      // Flush pending messages accumulated while disconnected
      for (const m of pendingMessages) ws!.send(JSON.stringify(m));
      pendingMessages = [];
      pingTimer = setInterval(() => {
        try {
          ws?.send(JSON.stringify({ type: 'ping' }));
        } catch {
          /* ignore */
        }
      }, PING_INTERVAL_MS);
    };

    ws.onmessage = (ev) => {
      try {
        const msg = JSON.parse(ev.data as string) as WsMessage;
        opts.onMessage(msg);
      } catch {
        /* ignore malformed frames */
      }
    };

    ws.onclose = onWsClose;
    ws.onerror = () => {
      /* close handler runs after error */
    };
  }

  function onWsClose() {
    if (closed) return;
    if (pingTimer) {
      clearInterval(pingTimer);
      pingTimer = null;
    }
    opts.onConnectionChange?.(false);
    reconnectAttempts++;
    if (reconnectAttempts <= MAX_WS_RETRIES) {
      const delay = Math.min(
        1000 * Math.pow(2, reconnectAttempts - 1),
        30_000,
      );
      setTimeout(connectWs, delay);
    } else {
      // Exceeded retry budget — fall back to HTTP polling
      usingPolling = true;
      void startPolling();
    }
  }

  async function startPolling() {
    pollTimer = setInterval(async () => {
      if (closed) return;
      // HTTP fallback uses Authorization: Bearer (browser fetch DOES allow
      // custom headers, unlike the WebSocket constructor). botId/visitorId
      // are public identifiers and stay in the query string.
      const url =
        `${opts.serverUrl}/webchat/history?` +
        `botId=${encodeURIComponent(opts.botId)}` +
        `&visitorId=${encodeURIComponent(opts.visitorId)}` +
        (lastSeenAt ? `&since=${encodeURIComponent(lastSeenAt)}` : '');
      try {
        const res = await fetch(url, {
          credentials: 'omit',
          headers: { authorization: `Bearer ${opts.apiKey}` },
        });
        if (!res.ok) return;
        const data = (await res.json()) as {
          messages?: Array<{ id: string; text: string; at: string }>;
        };
        if (data.messages) {
          for (const m of data.messages) {
            opts.onMessage({
              type: 'bot_message',
              id: m.id,
              text: m.text,
              created_at: m.at,
            });
            lastSeenAt = m.at;
          }
        }
      } catch {
        /* network failure — will retry on next interval */
      }
    }, POLL_INTERVAL_MS);
    opts.onConnectionChange?.(false); // polling = not "connected" in strict sense
  }

  async function sendViaHttp(msg: ClientMessage) {
    if (msg.type !== 'user_message') return;
    try {
      await fetch(`${opts.serverUrl}/webchat/msg`, {
        method: 'POST',
        headers: {
          'content-type': 'application/json',
          authorization: `Bearer ${opts.apiKey}`,
        },
        body: JSON.stringify({
          botId: opts.botId,
          visitorId: opts.visitorId,
          text: msg.text,
          idempotency_key: msg.id,
        }),
        credentials: 'omit',
      });
    } catch {
      /* ignore — user will see message unacknowledged */
    }
  }

  connectWs();

  return {
    send(msg) {
      if (usingPolling) {
        void sendViaHttp(msg);
        return;
      }
      if (ws && ws.readyState === WebSocket.OPEN) {
        ws.send(JSON.stringify(msg));
      } else {
        pendingMessages.push(msg);
      }
    },
    close() {
      closed = true;
      if (pingTimer) clearInterval(pingTimer);
      if (pollTimer) clearInterval(pollTimer);
      if (ws) ws.close();
    },
    isConnected() {
      return !!(ws && ws.readyState === WebSocket.OPEN);
    },
  };
}

// Re-export WidgetConfig so callers can import from one place if needed
export type { WidgetConfig };