lego-builder/server/index.js

/* eslint-env node */
/**
 * Sovereign self-hosted real-time WebSocket relay for the lego-builder rooms.
 *
 * Replaces the Liveblocks SaaS dependency. Pure room broadcast relay:
 *   - one WebSocket connection == one client in exactly one room
 *   - the room keeps the current shared `storage` snapshot
 *     ({ bricks, cursorColors }) plus a Map of connected clients
 *     ({ connId, presence }).
 *
 * Wire protocol (all JSON):
 *   server -> client on connect:
 *     { t: 'init', connectionId, storage, others: [{ connectionId, presence }] }
 *   server -> client on peer join / leave:
 *     { t: 'join', connectionId, presence }
 *     { t: 'leave', connectionId }
 *   server -> client on peer presence / storage update:
 *     { t: 'presence', connectionId, presence }
 *     { t: 'storage', storage }
 *   server -> client on relayed ephemeral event:
 *     { t: 'event', from: connectionId, event }
 *
 *   client -> server:
 *     { t: 'presence', presence }   // updates this client's presence, fan-out to others
 *     { t: 'storage', storage }     // replaces the shared snapshot, fan-out to others
 *     { t: 'event', event }         // ephemeral (cursors), relayed to OTHERS only, NOT persisted
 *
 * Runs standalone: `node server/index.js` (no build step). Listens on
 * process.env.PORT || 8090, accepts ws(s)://HOST/?room=<id>.
 */

import { WebSocketServer } from "ws";
import { parse as parseUrl } from "node:url";

const PORT = process.env.PORT || 8090;
const HEARTBEAT_INTERVAL_MS = 30000;

// roomId -> { storage: { bricks, cursorColors }, clients: Map<connId, ws> }
const rooms = new Map();

let connCounter = 0;
function nextConnectionId() {
  connCounter += 1;
  return connCounter;
}

function getRoom(roomId) {
  let room = rooms.get(roomId);
  if (!room) {
    room = { storage: { bricks: [], cursorColors: {} }, clients: new Map() };
    rooms.set(roomId, room);
  }
  return room;
}

function send(ws, payload) {
  if (ws.readyState === ws.OPEN) {
    ws.send(JSON.stringify(payload));
  }
}

function broadcastToOthers(room, exceptWs, payload) {
  const data = JSON.stringify(payload);
  for (const peer of room.clients.values()) {
    if (peer !== exceptWs && peer.readyState === peer.OPEN) {
      peer.send(data);
    }
  }
}

function othersList(room, exceptWs) {
  const others = [];
  for (const peer of room.clients.values()) {
    if (peer !== exceptWs) {
      others.push({ connectionId: peer.connectionId, presence: peer.presence });
    }
  }
  return others;
}

const wss = new WebSocketServer({ port: PORT });

wss.on("connection", (ws, req) => {
  const { query } = parseUrl(req.url, true);
  const roomId = (query && query.room) || "default";

  const room = getRoom(roomId);
  const connectionId = nextConnectionId();

  ws.isAlive = true;
  ws.connectionId = connectionId;
  ws.roomId = roomId;
  ws.presence = null;

  room.clients.set(connectionId, ws);

  // Send the current snapshot + roster to the newcomer.
  send(ws, {
    t: "init",
    connectionId,
    storage: room.storage,
    others: othersList(room, ws),
  });

  ws.on("pong", () => {
    ws.isAlive = true;
  });

  ws.on("message", (raw) => {
    let msg;
    try {
      msg = JSON.parse(raw.toString());
    } catch {
      return; // ignore malformed frames, never crash
    }
    if (!msg || typeof msg !== "object") return;

    switch (msg.t) {
      case "presence": {
        const first = ws.presence === null;
        ws.presence = msg.presence;
        // First presence == this client announcing itself => `join`.
        // Subsequent updates => `presence`.
        broadcastToOthers(room, ws, {
          t: first ? "join" : "presence",
          connectionId,
          presence: msg.presence,
        });
        break;
      }
      case "storage": {
        // Last-writer-wins replace of the shared snapshot.
        if (msg.storage && typeof msg.storage === "object") {
          room.storage = msg.storage;
          broadcastToOthers(room, ws, { t: "storage", storage: room.storage });
        }
        break;
      }
      case "event": {
        // Ephemeral, relayed to others only, never persisted.
        broadcastToOthers(room, ws, {
          t: "event",
          from: connectionId,
          event: msg.event,
        });
        break;
      }
      default:
        break;
    }
  });

  const cleanup = () => {
    if (!room.clients.has(connectionId)) return;
    room.clients.delete(connectionId);
    broadcastToOthers(room, ws, { t: "leave", connectionId });
    if (room.clients.size === 0) {
      rooms.delete(roomId);
    }
  };

  ws.on("close", cleanup);
  ws.on("error", cleanup);
});

// Heartbeat: drop dead sockets so rooms get cleaned up reliably.
const heartbeat = setInterval(() => {
  for (const ws of wss.clients) {
    if (ws.isAlive === false) {
      ws.terminate();
      continue;
    }
    ws.isAlive = false;
    try {
      ws.ping();
    } catch {
      // ignore
    }
  }
}, HEARTBEAT_INTERVAL_MS);

wss.on("close", () => clearInterval(heartbeat));

// Join/leave semantics: a newcomer learns existing peers via `init.others`,
// then announces itself by sending its first `presence` frame right after
// `init` — the relay forwards that first frame as `join` (with presence) and
// any later presence changes as `presence`. Departures are broadcast as
// `leave` on socket close.

console.log(`[realtime] lego-builder relay listening on :${PORT}`);