Redis Pub/Sub vs WebSockets: What I Learned Building Real-Time SaaS

Redis Pub/Sub and WebSockets are not competing technologies. They solve different problems in the same pipeline. Confusing them is how you end up with an architecture that doesn't scale.

This is the horizontal scaling trap: you ship a real-time feature with WebSockets, it works perfectly in development (one server), then breaks in production (multiple servers) because messages only reach clients connected to the same instance that produced the event.

Gmail/Outlook API → Webhook Handler → Email Processor → Redis Pub/Sub
                                                               ↓
                                                      All WebSocket Servers
                                                               ↓
                                                      Connected Browsers

import { createClient, type RedisClientType } from "redis";

// Pub/Sub requires separate clients for publishing and subscribing.
// A subscribing client enters a mode where it can only subscribe/unsubscribe.
let publisher: RedisClientType;
let subscriber: RedisClientType;

export async function initRedis() {
  const url = process.env.REDIS_URL;

  publisher = createClient({ url });
  subscriber = createClient({ url });

  await publisher.connect();
  await subscriber.connect();
}

// Channel naming convention: user-specific channels
// Scoped by userId so each user only receives their own events
function getUserChannel(userId: string): string {
  return `user:${userId}:newsletters`;
}

// Publisher: called when a new newsletter is processed
export async function publishNewsletter(userId: string, newsletter: Newsletter) {
  const channel = getUserChannel(userId);
  const payload = JSON.stringify({
    type: "new_newsletter",
    newsletter,
    timestamp: Date.now(),
  });

  await publisher.publish(channel, payload);
}

// Subscriber: called once per WebSocket server at startup
// The callback receives every message published to matching channels
export async function subscribeToUser(
  userId: string,
  onMessage: (payload: unknown) => void
) {
  const channel = getUserChannel(userId);

  await subscriber.subscribe(channel, (message) => {
    try {
      const parsed = JSON.parse(message);
      onMessage(parsed);
    } catch {
      console.error("Failed to parse Redis message:", message);
    }
  });

  return () => subscriber.unsubscribe(channel);
}

import { WebSocketServer, WebSocket } from "ws";
import { subscribeToUser } from "./redis-pubsub";
import { verifyAuthToken } from "./auth";

// Map: userId → Set of connected WebSocket clients
// A user might have multiple tabs open — broadcast to all
const userConnections = new Map<string, Set<WebSocket>>();

export function createWebSocketServer(httpServer: HTTPServer) {
  const wss = new WebSocketServer({ server: httpServer, path: "/ws" });

  wss.on("connection", async (ws, req) => {
    // Authenticate via token in query string
    const url = new URL(req.url!, `http://${req.headers.host}`);
    const token = url.searchParams.get("token");

    let userId: string;
    try {
      userId = await verifyAuthToken(token);
    } catch {
      ws.close(4001, "Unauthorized");
      return;
    }

    // Register this connection
    if (!userConnections.has(userId)) {
      userConnections.set(userId, new Set());
    }
    userConnections.get(userId)!.add(ws);

    // Subscribe to Redis for this user's events
    // Only subscribe once per userId (not once per connection)
    const isFirstConnection = userConnections.get(userId)!.size === 1;
    let unsubscribe: (() => Promise<void>) | undefined;

    if (isFirstConnection) {
      unsubscribe = await subscribeToUser(userId, (payload) => {
        broadcast(userId, payload);
      });
    }

    ws.on("close", async () => {
      userConnections.get(userId)?.delete(ws);

      // Unsubscribe from Redis when user has no more connections
      if (userConnections.get(userId)?.size === 0) {
        userConnections.delete(userId);
        await unsubscribe?.();
      }
    });

    // Send initial connection confirmation
    ws.send(JSON.stringify({ type: "connected", userId }));
  });

  return wss;
}

function broadcast(userId: string, payload: unknown) {
  const connections = userConnections.get(userId);
  if (!connections || connections.size === 0) return;

  const message = JSON.stringify(payload);
  for (const ws of connections) {
    if (ws.readyState === WebSocket.OPEN) {
      ws.send(message);
    }
  }
}

import { createClient } from "redis";
import { db } from "./database";

const BUFFER_KEY = "newsletter:write_buffer";
const BATCH_SIZE = 100;
const FLUSH_INTERVAL_MS = 500;

const redis = createClient({ url: process.env.REDIS_URL });

// Queue a newsletter for batch insertion
export async function queueNewsletter(newsletter: ProcessedNewsletter) {
  await redis.rPush(BUFFER_KEY, JSON.stringify(newsletter));
}

// Flush buffered newsletters to PostgreSQL
async function flush() {
  // Atomically get and clear up to BATCH_SIZE items
  // LMPOP is atomic — safe for multiple server instances
  const result = await redis.lmPop(1, BUFFER_KEY, "LEFT", { COUNT: BATCH_SIZE });
  if (!result || result.elements.length === 0) return;

  const newsletters = result.elements.map((el) =>
    JSON.parse(el.element) as ProcessedNewsletter
  );

  if (newsletters.length === 0) return;

  // Batch INSERT with ON CONFLICT DO NOTHING (idempotent)
  await db.query(`
    INSERT INTO newsletters (id, user_id, subject, sender, category, received_at, preview)
    SELECT * FROM UNNEST($1::uuid[], $2::uuid[], $3::text[], $4::text[], $5::text[], $6::timestamptz[], $7::text[])
    ON CONFLICT (id) DO NOTHING
  `, [
    newsletters.map(n => n.id),
    newsletters.map(n => n.userId),
    newsletters.map(n => n.subject),
    newsletters.map(n => n.sender),
    newsletters.map(n => n.category),
    newsletters.map(n => n.receivedAt),
    newsletters.map(n => n.preview),
  ]);

  console.log(`Flushed ${newsletters.length} newsletters to DB`);
}

// Start the flush loop
export function startBatchWriter() {
  setInterval(flush, FLUSH_INTERVAL_MS);
  redis.connect();
}

The 40% throughput gain came entirely from the batched DB writes — not from Redis Pub/Sub itself. Redis Pub/Sub enables the architecture to scale horizontally. The batch writer makes each server instance more efficient. Both matter, but for different reasons.