diff --git a/server/index.js b/server/index.js new file mode 100644 index 0000000..7d8bf4f --- /dev/null +++ b/server/index.js @@ -0,0 +1,190 @@ +/* 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=. + */ + +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 } +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}`);