siftlode/frontend/src/lib/messagesSocket.ts

// Live message delivery over a WebSocket. The server pushes a {type:"message", message} frame
// to all of a user's open tabs when a message is stored; subscribers react (e.g. refetch the
// thread + conversations). Auto-reconnects with backoff; a low-frequency poll elsewhere is the
// safety net if the socket is down.
import type { Message } from "./api";

type Handler = (msg: Message) => void;

const handlers = new Set<Handler>();
let ws: WebSocket | null = null;
let started = false;
let backoff = 1000;
let reconnectTimer: ReturnType<typeof setTimeout> | null = null;

function url(): string {
  const proto = location.protocol === "https:" ? "wss" : "ws";
  return `${proto}://${location.host}/api/messages/ws`;
}

function open() {
  try {
    ws = new WebSocket(url());
  } catch {
    schedule();
    return;
  }
  ws.onopen = () => {
    backoff = 1000;
  };
  ws.onmessage = (ev) => {
    try {
      const data = JSON.parse(ev.data);
      if (data?.type === "message" && data.message) {
        for (const h of handlers) h(data.message as Message);
      }
    } catch {
      /* ignore malformed frames */
    }
  };
  ws.onclose = () => {
    ws = null;
    if (started) schedule();
  };
  ws.onerror = () => {
    ws?.close();
  };
}

function schedule() {
  if (reconnectTimer) return;
  reconnectTimer = setTimeout(() => {
    reconnectTimer = null;
    if (started) open();
  }, backoff);
  backoff = Math.min(backoff * 2, 30000);
}

// Subscribe to live messages. Opens the socket on the first subscriber and closes it when the
// last one leaves.
export function onMessage(h: Handler): () => void {
  handlers.add(h);
  if (!started) {
    started = true;
    open();
  }
  return () => {
    handlers.delete(h);
    if (handlers.size === 0) {
      started = false;
      if (reconnectTimer) {
        clearTimeout(reconnectTimer);
        reconnectTimer = null;
      }
      ws?.close();
      ws = null;
    }
  };
}