feat(scale): swappable pub/sub layer + fix chat.js missed awaits (Phase 6)

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 <noreply@anthropic.com>

.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

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

server/chat.js

@@ -1,14 +1,28 @@
-// Chat presence + real-time delivery. A logged-in user opens a WebSocket and sends
+// Chat presence + real-time delivery. A logged-in user opens a WebSocket and sends `chat-hello`;
-// `chat-hello`; signaling.js registers the socket here. Messages are persisted over HTTP
+// signaling.js registers the socket here. Messages are persisted over HTTP (routes.js) and pushed live to
-// (routes.js) and pushed live to the recipient's sockets via pushToUser().
+// 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);
+  deliverLocal(userId, obj);            // sockets on this instance
-  if (!s) return;
+  pubsub.publish('u:' + userId, obj);   // other instances (no-op on the memory backend)
-  const data = JSON.stringify(obj);
-  for (const ws of s) { if (ws.readyState === 1) { try { ws.send(data); } catch (_) {} } }
 }
 
 // --- Live presence -------------------------------------------------------------------------
-// A user's effective status (online / in-a-call / away / …) changes with no HTTP round-trip:
+// A user's effective status (online / in-a-call / away / …) changes with no HTTP round-trip: they connect
-// they connect or disconnect a socket, or join/leave a call. Without pushing that change, OTHER
+// or disconnect a socket, or join/leave a call. Without pushing that change, OTHER users only see it after
-// users only see it after a full page reload — impossible in the desktop/mobile apps. So whenever
+// a full page reload — impossible in the desktop/mobile apps. So whenever it changes we broadcast the
-// it changes we broadcast the user's fresh status to everyone else's sockets, and the client
+// user's fresh status to everyone else's sockets, and the client updates that contact's dot/subtitle.
-// updates that contact's dot/subtitle in place.
 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 (_) {}
+  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: effectiveStatus(userId), lastSeen });
+  const payload = JSON.stringify({ type: 'presence', userId, online: isOnline(userId), status: await effectiveStatus(userId), lastSeen });
-  for (const [uid, set] of chatClients) {
+  deliverPresenceLocal(userId, payload);                    // this instance
-    if (uid === userId) continue;                  // no need to tell someone about their own status
+  pubsub.publish('presence', { userId, payload });          // other instances (no-op on memory)
-    for (const ws of set) { if (ws.readyState === 1) { try { ws.send(payload); } catch (_) {} } }
-  }
 }
 
+// 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 };

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"
   },

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);

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
+};

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 };

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 };