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