llink/js/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.
 */

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) => {
      console.warn("[pusher] websocket error", event);
      // onclose will fire after onerror, so reconnection is handled there
    };

    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 {
    if (!this.listeners.has(channelId)) {
      this.listeners.set(channelId, new Map());
    }
    const channelListeners = this.listeners.get(channelId)!;
    if (!channelListeners.has(event)) {
      channelListeners.set(event, new Set());
    }
    channelListeners.get(event)!.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 {
      console.warn("[pusher] failed to parse message", data);
      return;
    }

    if (msg.type === "error") {
      console.warn("[pusher] server error:", msg.message);
      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) {
        console.error("[pusher] listener error", err);
      }
    }
  }

  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);
    }
  }
}