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 { getActiveAccount, type Message, type MessageUser } from "./api";
import { log } from "./log";

// What the client log records of the socket: its lifecycle, and at DEBUG each frame's TYPE. Never a
// frame's content — chat plaintext stays out of every log structurally, not by redaction. The path
// is fixed, without the `?account=` query.
const WS_PATH = "/api/messages/ws";

// A pushed message plus both parties, so the dock can open/flash the right window.
interface IncomingMessage {
  message: Message;
  from?: MessageUser;
  to?: MessageUser;
}
type Handler = (e: IncomingMessage) => void;
type UnreadHandler = () => void;

const handlers = new Set<Handler>();
const unreadHandlers = new Set<UnreadHandler>();
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";
  // A WebSocket can't carry the X-Siftlode-Account header, so the per-tab account rides in the
  // query string (validated against the browser wallet server-side). The account is read once
  // here: switching accounts always goes through location.reload() (NavSidebar), which rebuilds
  // this module and reopens the socket, so it's intentionally fixed for the socket's lifetime.
  const account = getActiveAccount();
  const qs = account != null ? `?account=${account}` : "";
  return `${proto}://${location.host}${WS_PATH}${qs}`;
}

function open() {
  let sock: WebSocket;
  try {
    sock = new WebSocket(url());
  } catch (e) {
    log.warn("ws.failed", "could not open the socket", { path: WS_PATH, message: String(e) });
    schedule();
    return;
  }
  ws = sock;
  // Every handler checks it still belongs to the CURRENT socket. A socket closed by the last
  // subscriber leaving closes asynchronously: when a subscriber came back in between (a remount,
  // StrictMode's double effect) a newer socket is already open, and the old one's close used to
  // clear that newer reference and schedule a THIRD — two live sockets, every message delivered twice.
  sock.onopen = () => {
    if (ws !== sock) return;
    log.info("ws.open", "socket open", { path: WS_PATH });
    backoff = 1000;
  };
  sock.onmessage = (ev) => {
    if (ws !== sock) return;
    try {
      const data = JSON.parse(ev.data);
      log.debug("ws.frame", "socket frame", { path: WS_PATH, type: String(data?.type) });
      if (data?.type === "message" && data.message) {
        const e: IncomingMessage = {
          message: data.message as Message,
          from: data.from,
          to: data.to,
        };
        for (const h of handlers) h(e);
      } else if (data?.type === "unread") {
        // Sent to a user's OTHER tabs when they read a thread elsewhere, so the badge updates live.
        for (const h of unreadHandlers) h();
      }
    } catch {
      /* ignore malformed frames */
    }
  };
  sock.onclose = (ev) => {
    const current = ws === sock;
    // A socket we let go of (the last subscriber left) closing is routine; the current one closing
    // is the server or the network going away, and the reconnect that follows is worth seeing.
    (current ? log.warn : log.info)("ws.close", "socket closed", {
      path: WS_PATH,
      code: ev.code,
      clean: ev.wasClean,
      reconnect_ms: current && started ? backoff : null,
    });
    if (!current) return;
    ws = null;
    if (started) schedule();
  };
  sock.onerror = () => {
    sock.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;
    }
  };
}

// Subscribe to "your unread changed elsewhere" pings (a read on another tab/device). The socket is
// already opened/closed by onMessage subscribers (the app-wide ChatDock keeps one), so this only
// registers a callback — it does not open the socket on its own.
export function onUnread(h: UnreadHandler): () => void {
  unreadHandlers.add(h);
  return () => {
    unreadHandlers.delete(h);
  };
}