From 63f2c588dafd708d94eece681061ab229dec2833 Mon Sep 17 00:00:00 2001 From: sravan Date: Fri, 24 Jul 2026 23:20:02 +0530 Subject: [PATCH] feat(scale): swappable pub/sub layer + fix chat.js missed awaits (Phase 6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two things: 1. FIX a live regression the async conversion missed: chat.js calls repos via the lazy repos() helper (not the R. prefix), so my sweep skipped it — effectiveStatus / broadcastPresence read `repos().users.byId(userId)` synchronously, but that's a Promise now, so presence broadcasts always reported status 'active' and dropped last_seen. Now awaited (effectiveStatus/broadcastPresence async); touchSeen is a fire-and-forget UPDATE with .catch. Audited all non-R. repo calls — only chat.js was affected (media.js backfill was already awaited). 2. Swappable pub/sub for multi-instance real-time fan-out (the actual blocker to running >1 instance — not the DB). server/pubsub.js picks a backend by PUBSUB_BACKEND (default 'memory'). Local socket delivery is UNCHANGED; publish is additive — memory = no-op (zero hot-path cost, identical single-instance behaviour), redis = fan-out to other instances with a self-echo guard. chat.js pushToUser/broadcastPresence now also publish; each instance subscribes to deliver remote events to its local sockets. Interface is tiny so Redis is one swappable file (Postgres LISTEN/NOTIFY or NATS could drop in the same way — never hardwired, as requested). Dormant redis service added to compose behind the 'scale' profile; redis dep added; PUBSUB_BACKEND/REDIS_URL documented. Validated: smoke 22/22 (memory), e2e chat delivery green. NOTE: full multi-instance also needs distributed presence (isOnline is per-process) + meeting-signaling sharing — chat/presence fan out via this layer; those are follow-ups. Co-Authored-By: Claude Opus 4.8 --- .env.example | 6 ++++ docker-compose.yml | 14 +++++++++ server/chat.js | 67 +++++++++++++++++++++++++++-------------- server/package.json | 1 + server/pubsub.js | 10 ++++++ server/pubsub/memory.js | 9 ++++++ server/pubsub/redis.js | 46 ++++++++++++++++++++++++++++ server/server.js | 8 ++++- 8 files changed, 137 insertions(+), 24 deletions(-) create mode 100644 server/pubsub.js create mode 100644 server/pubsub/memory.js create mode 100644 server/pubsub/redis.js diff --git a/.env.example b/.env.example index 796db04..dbbadfe 100644 --- a/.env.example +++ b/.env.example @@ -36,3 +36,9 @@ TURN_CREDENTIAL= # POSTGRES_PASSWORD= # DATABASE_URL=postgres://bizgaze:PASSWORD@bizgaze-postgres:5432/bizgaze # DB_BACKEND=pg + +# Optional: cross-instance real-time (chat/presence) fan-out via Redis, for running MULTIPLE app instances. +# Leave unset for single-instance (in-memory pub/sub, the default). To scale out: start the redis service +# (`docker compose --profile scale up -d`), then set both below + a sticky load balancer for /ws. +# PUBSUB_BACKEND=redis +# REDIS_URL=redis://bizgaze-redis:6379 diff --git a/docker-compose.yml b/docker-compose.yml index 7b66f86..2dd4094 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -63,6 +63,20 @@ services: timeout: 3s retries: 12 + # Redis — cross-instance real-time fan-out (chat/presence), used only when PUBSUB_BACKEND=redis. Dormant + # by default (behind the 'scale' profile, like livekit), so a normal deploy never starts it and the app + # stays single-instance on the in-memory pubsub. To run multiple app instances: start this + # (`docker compose --profile scale up -d`), set PUBSUB_BACKEND=redis + REDIS_URL in .env, and put the app + # behind a load balancer with sticky sessions for the /ws WebSocket. (Meeting SIGNALING state is still + # per-process — cross-instance meetings need sticky routing or further work; chat/presence fan out here.) + bizgazeredis: + image: redis:7-alpine + container_name: bizgaze-redis + restart: unless-stopped + profiles: ["scale"] + networks: + - npm + # LiveKit SFU — meeting media server. Optional: only started/used when the app's .env has # LIVEKIT_URL/API_KEY/API_SECRET set (otherwise meetings use the built-in P2P mesh). NPM proxies # wss://livekit.bizgaze.com -> livekit:7880 (signaling); media flows over the published UDP/TCP diff --git a/server/chat.js b/server/chat.js index 70e5999..f2191fb 100644 --- a/server/chat.js +++ b/server/chat.js @@ -1,14 +1,28 @@ -// Chat presence + real-time delivery. A logged-in user opens a WebSocket and sends -// `chat-hello`; signaling.js registers the socket here. Messages are persisted over HTTP -// (routes.js) and pushed live to the recipient's sockets via pushToUser(). +// Chat presence + real-time delivery. A logged-in user opens a WebSocket and sends `chat-hello`; +// signaling.js registers the socket here. Messages are persisted over HTTP (routes.js) and pushed live to +// the recipient's sockets via pushToUser(). +// +// MULTI-INSTANCE: local delivery (to sockets on THIS process) is unchanged. Every push is ALSO published +// via the swappable pubsub layer so, when >1 instance runs, a recipient connected to another instance +// still gets it. With the default in-memory pubsub (single instance) publish is a no-op, so this is +// behaviourally identical to before — no hot-path cost. const { chatClients, meetingRooms } = require('./presence'); +const pubsub = require('./pubsub'); let _repos = null; const repos = () => (_repos || (_repos = require('./repos'))); // lazy: avoid a require cycle +// Deliver an already-built or plain object to a user's sockets on THIS instance. +function deliverLocal(userId, obj) { + const s = chatClients.get(userId); + if (!s) return; + const data = typeof obj === 'string' ? obj : JSON.stringify(obj); + for (const ws of s) { if (ws.readyState === 1) { try { ws.send(data); } catch (_) {} } } +} + function register(userId, ws) { if (!chatClients.has(userId)) chatClients.set(userId, new Set()); chatClients.get(userId).add(ws); ws._chatUserId = userId; - try { repos().users.touchSeen(userId); } catch (_) {} // "last seen" (#2) + repos().users.touchSeen(userId).catch(() => {}); // "last seen" (#2) — fire-and-forget UPDATE } function unregister(ws) { @@ -18,7 +32,7 @@ function unregister(ws) { if (set) { set.delete(ws); // Their LAST socket went away → stamp when they were last here, so contacts can show "last seen …". - if (!set.size) { chatClients.delete(id); try { repos().users.touchSeen(id); } catch (_) {} } + if (!set.size) { chatClients.delete(id); repos().users.touchSeen(id).catch(() => {}); } } } @@ -28,37 +42,44 @@ function isOnline(userId) { } function pushToUser(userId, obj) { - const s = chatClients.get(userId); - if (!s) return; - const data = JSON.stringify(obj); - for (const ws of s) { if (ws.readyState === 1) { try { ws.send(data); } catch (_) {} } } + deliverLocal(userId, obj); // sockets on this instance + pubsub.publish('u:' + userId, obj); // other instances (no-op on the memory backend) } // --- Live presence ------------------------------------------------------------------------- -// A user's effective status (online / in-a-call / away / …) changes with no HTTP round-trip: -// they connect or disconnect a socket, or join/leave a call. Without pushing that change, OTHER -// users only see it after a full page reload — impossible in the desktop/mobile apps. So whenever -// it changes we broadcast the user's fresh status to everyone else's sockets, and the client -// updates that contact's dot/subtitle in place. +// A user's effective status (online / in-a-call / away / …) changes with no HTTP round-trip: they connect +// or disconnect a socket, or join/leave a call. Without pushing that change, OTHER users only see it after +// a full page reload — impossible in the desktop/mobile apps. So whenever it changes we broadcast the +// user's fresh status to everyone else's sockets, and the client updates that contact's dot/subtitle. function isInCall(userId) { for (const [, peers] of meetingRooms) { for (const [, p] of peers) { if (p.ws && p.ws._meetingUserId === userId) return true; } } return false; } -function effectiveStatus(userId) { +async function effectiveStatus(userId) { if (isInCall(userId)) return 'incall'; // derived (overrides the stored status) - try { const u = repos().users.byId(userId); return (u && u.status) || 'active'; } catch (_) { return 'active'; } + try { const u = await repos().users.byId(userId); return (u && u.status) || 'active'; } catch (_) { return 'active'; } } -function broadcastPresence(userId) { +// Send a presence payload to every local socket EXCEPT the subject's own. +function deliverPresenceLocal(subjectId, payload) { + for (const [uid, set] of chatClients) { + if (uid === subjectId) continue; + for (const ws of set) { if (ws.readyState === 1) { try { ws.send(payload); } catch (_) {} } } + } +} +async function broadcastPresence(userId) { if (!userId) return; // Carry last_seen too, so a contact going offline immediately reads "Last seen just now" instead of a // bare "Offline" until the next sidebar reload (#2). let lastSeen = null; - try { const u = repos().users.byId(userId); lastSeen = (u && u.last_seen) || null; } catch (_) {} - const payload = JSON.stringify({ type: 'presence', userId, online: isOnline(userId), status: effectiveStatus(userId), lastSeen }); - for (const [uid, set] of chatClients) { - if (uid === userId) continue; // no need to tell someone about their own status - for (const ws of set) { if (ws.readyState === 1) { try { ws.send(payload); } catch (_) {} } } - } + try { const u = await repos().users.byId(userId); lastSeen = (u && u.last_seen) || null; } catch (_) {} + const payload = JSON.stringify({ type: 'presence', userId, online: isOnline(userId), status: await effectiveStatus(userId), lastSeen }); + deliverPresenceLocal(userId, payload); // this instance + pubsub.publish('presence', { userId, payload }); // other instances (no-op on memory) } +// Cross-instance inbound: deliver events published by OTHER instances to our local sockets. On the memory +// backend these never fire; on redis they carry the fan-out. (Buffered until pubsub.init() connects.) +pubsub.subscribe('u:*', (channel, obj) => deliverLocal(channel.slice(2), obj)); +pubsub.subscribe('presence', (channel, msg) => { if (msg && msg.payload) deliverPresenceLocal(msg.userId, msg.payload); }); + module.exports = { register, unregister, isOnline, pushToUser, broadcastPresence }; diff --git a/server/package.json b/server/package.json index 5436820..bba369b 100644 --- a/server/package.json +++ b/server/package.json @@ -11,6 +11,7 @@ }, "dependencies": { "pg": "^8.13.1", + "redis": "^4.7.0", "web-push": "^3.6.7", "ws": "^8.18.0" }, diff --git a/server/pubsub.js b/server/pubsub.js new file mode 100644 index 0000000..d4df52f --- /dev/null +++ b/server/pubsub.js @@ -0,0 +1,10 @@ +// Swappable pub/sub for cross-instance real-time fan-out. The app delivers to its OWN WebSocket clients +// locally (chat.js) exactly as before; this layer only carries a copy to OTHER app instances so a message +// POSTed on instance A reaches a recipient whose socket lives on instance B. +// +// Backend chosen by PUBSUB_BACKEND (default 'memory'). 'memory' = single instance: publish is a no-op and +// subscriptions never fire, so behaviour is identical to before this layer existed — zero hot-path cost. +// 'redis' fans out via Redis. The interface (publish/subscribe/init) is deliberately tiny so Redis is one +// swappable file — a Postgres LISTEN/NOTIFY or NATS backend could drop in the same way. Never hardwired. +const name = process.env.PUBSUB_BACKEND || 'memory'; +module.exports = require('./pubsub/' + name); diff --git a/server/pubsub/memory.js b/server/pubsub/memory.js new file mode 100644 index 0000000..00335eb --- /dev/null +++ b/server/pubsub/memory.js @@ -0,0 +1,9 @@ +// Single-instance pub/sub backend. There are no OTHER app instances, so a cross-instance publish has +// nowhere to go (no-op) and remote subscriptions never fire. All real-time delivery happens locally in +// chat.js — this is exactly the pre-pubsub behaviour, at zero cost. The default backend. +module.exports = { + name: 'memory', + init: () => Promise.resolve(), + publish: () => {}, // no other instance to reach + subscribe: () => {}, // nothing remote will ever arrive +}; diff --git a/server/pubsub/redis.js b/server/pubsub/redis.js new file mode 100644 index 0000000..3bef8e9 --- /dev/null +++ b/server/pubsub/redis.js @@ -0,0 +1,46 @@ +// Redis pub/sub backend — enables running MULTIPLE app instances. Each instance publishes every local +// real-time event; every other instance receives it and delivers to ITS local sockets. Selected by +// PUBSUB_BACKEND=redis; connection from REDIS_URL (default redis://bizgaze-redis:6379). +// +// Self-echo guard: Redis delivers a publish to ALL subscribers including the publisher, but the publishing +// instance ALREADY delivered locally — so every message is tagged with this instance's id and ignored on +// the way back in. Subscriptions made before connect are buffered and flushed in init(). +const crypto = require('crypto'); +const INSTANCE = crypto.randomBytes(8).toString('hex'); + +let pub = null, sub = null; +const pending = []; // [pattern, handler] queued before connect + +async function doSubscribe(pattern, handler) { + const onMessage = (message, channel) => { + let m; try { m = JSON.parse(message); } catch { return; } + if (m.i === INSTANCE) return; // our own publish — already delivered locally + try { handler(channel, m.d); } catch (_) {} + }; + if (pattern.includes('*')) await sub.pSubscribe(pattern, onMessage); + else await sub.subscribe(pattern, onMessage); +} + +async function init() { + const { createClient } = require('redis'); + const url = process.env.REDIS_URL || 'redis://bizgaze-redis:6379'; + pub = createClient({ url }); + sub = pub.duplicate(); + pub.on('error', () => {}); sub.on('error', () => {}); // never let a redis blip crash the app + await pub.connect(); + await sub.connect(); + for (const [pattern, handler] of pending) { try { await doSubscribe(pattern, handler); } catch (_) {} } + pending.length = 0; +} + +function publish(channel, data) { + if (!pub) return; + pub.publish(channel, JSON.stringify({ i: INSTANCE, d: data })).catch(() => {}); +} + +function subscribe(pattern, handler) { + if (!sub) { pending.push([pattern, handler]); return; } // buffer until init() connects + doSubscribe(pattern, handler).catch(() => {}); +} + +module.exports = { name: 'redis', init, publish, subscribe }; diff --git a/server/server.js b/server/server.js index 187aa67..9bebbcd 100644 --- a/server/server.js +++ b/server/server.js @@ -40,6 +40,7 @@ wss.on('connection', onConnection); // no-op (the schema is already applied synchronously when db.js is required). Serving only starts once the // store is ready, so the first request can never hit a missing table. const db = require('./dbx'); +const pubsub = require('./pubsub'); function startListening() { server.listen(PORT, () => { @@ -72,6 +73,11 @@ function startListening() { } } -db.init().then(startListening).catch((e) => { console.error('DB init failed:', (e && e.message) || e); process.exit(1); }); +// DB schema first, then the pub/sub layer (redis connects + flushes buffered subscriptions; memory is a +// no-op), then serve. A pubsub failure must NOT block booting — degrade to local-only delivery. +db.init() + .then(() => pubsub.init().catch((e) => console.error('pubsub init failed (local-only delivery):', (e && e.message) || e))) + .then(startListening) + .catch((e) => { console.error('DB init failed:', (e && e.message) || e); process.exit(1); }); module.exports = { server };