llink/js/desktop/src/lib/pusher-client.ts

/**
 * PusherClient manages a WebSocket connection to the pusher service.
 * Handles authentication, reconnection with exponential backoff,
 * channel subscriptions, and event dispatching.
 */

import { logError, reportError } from '@/lib/errors';

export type ConnectionState =
  | 'disconnected'
  | 'connecting'
  | 'connected'
  | 'reconnecting';

export interface ChannelMessage {
  humanId: string;
  payload: unknown;
}

// Server → Client message shape
interface ServerMessage {
  type: 'subscribed' | 'join' | 'leave' | 'message' | 'error';
  channel?: string;
  humanId?: string;
  presence?: string[];
  payload?: unknown;
  message?: string;
}

type ChannelEventType = 'subscribed' | 'join' | 'leave' | 'message';
type ChannelEventCallback = (msg: ServerMessage) => void;

interface PusherClientConfig {
  url: string;
  getToken: () => string | null;
}

const INITIAL_RECONNECT_DELAY = 1000;
const MAX_RECONNECT_DELAY = 30000;
const PING_INTERVAL = 20000; // 20s — keeps alive through GKE gateway timeout

export class PusherClient {
  private config: PusherClientConfig;
  private ws: WebSocket | null = null;
  private state: ConnectionState = 'disconnected';
  private stateListeners = new Set<(state: ConnectionState) => void>();

  // Channel event listeners: channelId → eventType → callbacks
  private listeners = new Map<
    string,
    Map<ChannelEventType, Set<ChannelEventCallback>>
  >();

  // Active subscriptions for re-subscribe on reconnect
  private activeSubscriptions = new Set<string>();

  // Reconnection state
  private reconnectDelay = INITIAL_RECONNECT_DELAY;
  private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
  private shouldReconnect = false;

  // Keep-alive ping
  private pingTimer: ReturnType<typeof setInterval> | null = null;

  constructor(config: PusherClientConfig) {
    this.config = config;
  }

  get connectionState(): ConnectionState {
    return this.state;
  }

  connect(): void {
    if (this.ws) return;

    const token = this.config.getToken();
    if (!token) {
      console.warn('[pusher] no token available, cannot connect');
      return;
    }

    this.shouldReconnect = true;
    this.setState(
      this.state === 'reconnecting' ? 'reconnecting' : 'connecting',
    );

    const url = `${this.config.url}?token=${encodeURIComponent(token)}`;
    this.ws = new WebSocket(url);

    this.ws.onopen = () => {
      this.setState('connected');
      this.reconnectDelay = INITIAL_RECONNECT_DELAY;
      this.startPing();
      this.resubscribeAll();
    };

    this.ws.onclose = () => {
      this.cleanup();
      if (this.shouldReconnect) {
        this.scheduleReconnect();
      }
    };

    this.ws.onerror = (event) => {
      // onclose fires after onerror — reconnection is handled there.
      logError(event, { scope: 'pusher.ws' });
    };

    this.ws.onmessage = (event) => {
      this.handleMessage(event.data as string);
    };
  }

  disconnect(): void {
    this.shouldReconnect = false;
    this.clearReconnectTimer();
    this.cleanup();
    this.activeSubscriptions.clear();
    this.setState('disconnected');
  }

  subscribe(channelId: string): void {
    this.activeSubscriptions.add(channelId);
    this.send({ type: 'subscribe', channel: channelId });
  }

  unsubscribe(channelId: string): void {
    this.activeSubscriptions.delete(channelId);
    this.send({ type: 'unsubscribe', channel: channelId });
  }

  sendMessage(channelId: string, payload: unknown): void {
    this.send({ type: 'message', channel: channelId, payload });
  }

  on(
    channelId: string,
    event: ChannelEventType,
    callback: ChannelEventCallback,
  ): void {
    let channelListeners = this.listeners.get(channelId);
    if (!channelListeners) {
      channelListeners = new Map();
      this.listeners.set(channelId, channelListeners);
    }
    let eventListeners = channelListeners.get(event);
    if (!eventListeners) {
      eventListeners = new Set();
      channelListeners.set(event, eventListeners);
    }
    eventListeners.add(callback);
  }

  off(
    channelId: string,
    event: ChannelEventType,
    callback: ChannelEventCallback,
  ): void {
    const channelListeners = this.listeners.get(channelId);
    if (!channelListeners) return;
    const eventListeners = channelListeners.get(event);
    if (!eventListeners) return;
    eventListeners.delete(callback);

    // Cleanup empty maps
    if (eventListeners.size === 0) channelListeners.delete(event);
    if (channelListeners.size === 0) this.listeners.delete(channelId);
  }

  onStateChange(callback: (state: ConnectionState) => void): () => void {
    this.stateListeners.add(callback);
    return () => this.stateListeners.delete(callback);
  }

  // --- Private ---

  private send(msg: {
    type: string;
    channel?: string;
    payload?: unknown;
  }): void {
    if (this.ws?.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(msg));
    }
  }

  private handleMessage(data: string): void {
    // Ignore keep-alive pong responses
    if (data === 'pong') return;

    let msg: ServerMessage;
    try {
      msg = JSON.parse(data);
    } catch (err) {
      logError(err, { scope: 'pusher.parse', data });
      return;
    }

    if (msg.type === 'error') {
      logError(new Error(msg.message ?? 'pusher server error'), {
        scope: 'pusher.server',
      });
      return;
    }

    if (!msg.channel) return;

    const channelListeners = this.listeners.get(msg.channel);
    if (!channelListeners) return;

    const eventListeners = channelListeners.get(msg.type as ChannelEventType);
    if (!eventListeners) return;

    for (const cb of eventListeners) {
      try {
        cb(msg);
      } catch (err) {
        // Listener bugs silently break user flows — escalate to reportError.
        reportError(err, { scope: 'pusher.listener', channel: msg.channel });
      }
    }
  }

  private resubscribeAll(): void {
    for (const channelId of this.activeSubscriptions) {
      this.send({ type: 'subscribe', channel: channelId });
    }
  }

  private scheduleReconnect(): void {
    this.setState('reconnecting');

    // Exponential backoff with jitter
    const jitter = Math.random() * 0.5 + 0.75; // 0.75 - 1.25x
    const delay = Math.min(this.reconnectDelay * jitter, MAX_RECONNECT_DELAY);

    this.reconnectTimer = setTimeout(() => {
      this.reconnectDelay = Math.min(
        this.reconnectDelay * 2,
        MAX_RECONNECT_DELAY,
      );
      this.connect();
    }, delay);
  }

  private cleanup(): void {
    this.stopPing();
    if (this.ws) {
      this.ws.onopen = null;
      this.ws.onclose = null;
      this.ws.onerror = null;
      this.ws.onmessage = null;
      if (
        this.ws.readyState === WebSocket.OPEN ||
        this.ws.readyState === WebSocket.CONNECTING
      ) {
        this.ws.close();
      }
      this.ws = null;
    }
  }

  private clearReconnectTimer(): void {
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer);
      this.reconnectTimer = null;
    }
  }

  private startPing(): void {
    this.stopPing();
    this.pingTimer = setInterval(() => {
      // Send an empty message as a keep-alive
      if (this.ws?.readyState === WebSocket.OPEN) {
        this.ws.send('ping');
      }
    }, PING_INTERVAL);
  }

  private stopPing(): void {
    if (this.pingTimer) {
      clearInterval(this.pingTimer);
      this.pingTimer = null;
    }
  }

  private setState(state: ConnectionState): void {
    if (this.state === state) return;
    this.state = state;
    for (const cb of this.stateListeners) {
      cb(state);
    }
  }
}