Scaling Live Streaming Announcements with Redis Pub/Sub at Enterprise Scale

// Channel naming convention
// stream:{streamId}:announce      → general announcements
// stream:{streamId}:system        → system-level alerts
// stream:{streamId}:moderation    → mod actions

const CHANNEL_PATTERNS = {
  announce:   `stream:${streamId}:announce`,
  system:     `stream:${streamId}:system`,
  moderation: `stream:${streamId}:moderation`,
};

// Message payload schema (JSON)
{
  type:      "STREAM_START" | "HOST_JOIN" | "VIEWER_MILESTONE" | "ALERT",
  streamId: "uuid-v4",
  payload:  { /* typed per announcement type */ },
  ts:       1712345678901,   // epoch ms
  ttl:      30000,            // client-side display TTL in ms
  priority: "high" | "normal" | "low"
}

import { createClient } from 'redis';

const publisher = await createClient({ url: process.env.REDIS_URL })
  .on('error', (err) => logger.error('Redis publisher error', err))
  .connect();

async function publishAnnouncement(streamId, announcement) {
  const channel = `stream:${streamId}:announce`;
  const message = JSON.stringify({
    ...announcement,
    ts: Date.now(),
    streamId,
  });

  const receiverCount = await publisher.publish(channel, message);
  logger.info({ channel, receiverCount }, 'Announcement published');
  return receiverCount; // # of subscribed WS pods that received it
}

import { createClient } from 'redis';
import { WebSocketServer } from 'ws';

const subscriber = await createClient({ url: process.env.REDIS_URL }).connect();
const roomRegistry = new Map(); // streamId → Set<WebSocket>

// Subscribe to all stream announcement channels via pattern
await subscriber.pSubscribe('stream:*:announce', (message, channel) => {
  const streamId = channel.split(':')[1];
  const room = roomRegistry.get(streamId);
  if (!room?.size) return; // no local viewers for this stream

  room.forEach((ws) => {
    if (ws.readyState === WebSocket.OPEN) {
      ws.send(message); // raw JSON — no re-serialization
    }
  });
});

// Register viewer on connection
wss.on('connection', (ws, req) => {
  const { streamId } = parseQuery(req.url);
  if (!roomRegistry.has(streamId)) roomRegistry.set(streamId, new Set());
  roomRegistry.get(streamId).add(ws);
  ws.on('close', () => roomRegistry.get(streamId)?.delete(ws));
});