Node 1 generates event → publish to Redis channel "tenant:T1"
Node 1 (as subscriber) → receives from Redis → delivers to local T1 sockets
Node 2 (as subscriber) → receives from Redis → delivers to local T1 sockets
Node 3 (as subscriber) → receives from Redis → delivers to local T1 sockets

# /etc/nginx/sites-available/fleettracker.conf

upstream fleet_ws {
    # Round-robin load balancing across Node.js instances
    server 127.0.0.1:3001;
    server 127.0.0.1:3002;
    server 127.0.0.1:3003;
    keepalive 256; # Persistent upstream connections
}

upstream fleet_api {
    server 127.0.0.1:3001;
    server 127.0.0.1:3002;
    server 127.0.0.1:3003;
    keepalive 64;
}

server {
    listen 443 ssl http2;
    server_name api.fleettracker.io;

    # WebSocket endpoint
    location /ws/ {
        proxy_pass http://fleet_ws;

        # Critical: Must use HTTP/1.1 — WebSocket upgrades require HTTP/1.1
        proxy_http_version 1.1;

        # Critical: Pass Upgrade and Connection headers to upstream
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";

        # Standard proxy headers
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # Extend timeout for persistent WebSocket connections
        # Default is 60s — WebSocket connections are long-lived
        proxy_read_timeout 3600s;
        proxy_send_timeout 3600s;

        # Disable response buffering for WebSocket frames
        proxy_buffering off;
    }

    # REST API endpoint (standard HTTP)
    location /api/ {
        proxy_pass http://fleet_api;
        proxy_http_version 1.1;
        proxy_set_header Connection '';
        proxy_set_header Host $host;
    }
}

1. Client opens WebSocket to Nginx
2. Nginx routes the HTTP/1.1 Upgrade request to Node 2 (round-robin)
3. Upgrade completes — TCP connection is now pinned to Node 2 at the transport layer
4. All subsequent frames from this client flow through the same TCP connection
5. Nginx forwards TCP packets from this established connection to Node 2
6. No additional HTTP requests are made — there is no session to correlate

1. Client sends: POST /socket.io/?EIO=4&transport=polling  → Nginx routes to Node 2
2. Node 2 assigns session ID: sid=ABC
3. Client sends: GET /socket.io/?EIO=4&transport=polling&sid=ABC
   → Nginx round-robins to Node 1
   → Node 1 has NO record of sid=ABC
   → 400 Session ID Unknown ← FAILURE

upstream fleet_socketio {
    # ip_hash: routes clients to the same backend based on client IP
    ip_hash;
    server 127.0.0.1:3001;
    server 127.0.0.1:3002;
    server 127.0.0.1:3003;
}

// server/lib/RedisPubSubBackplane.ts
import { createClient, type RedisClientType } from 'redis'
import { connectionManager } from './ConnectionManager'

class RedisPubSubBackplane {
  private publisher!: RedisClientType
  private subscriber!: RedisClientType

  async connect(url: string): Promise<void> {
    // Redis requires separate clients for pub and sub
    this.publisher = createClient({ url })
    this.subscriber = createClient({ url })

    await Promise.all([this.publisher.connect(), this.subscriber.connect()])
  }

  /** Publish a tenant event to the Redis channel (called by ChangeStreamWorker) */
  async publish(tenantId: string, event: unknown): Promise<void> {
    await this.publisher.publish(
      `tenant:${tenantId}`,
      JSON.stringify(event),
    )
  }

  /** Subscribe this node to all tenant channels it has active connections for */
  async subscribeTenant(tenantId: string): Promise<void> {
    await this.subscriber.subscribe(`tenant:${tenantId}`, (message) => {
      const event = JSON.parse(message)
      // Deliver to all local sockets for this tenant
      connectionManager.broadcastToTenant(tenantId, event)
    })
  }

  async unsubscribeTenant(tenantId: string): Promise<void> {
    await this.subscriber.unsubscribe(`tenant:${tenantId}`)
  }
}

export const backplane = new RedisPubSubBackplane()

// Integration: ChangeStreamWorker publishes to Redis instead of directly to ConnectionManager
private async processChange(change: ChangeStreamDocument): Promise<void> {
  const doc = (change as any).fullDocument
  if (!doc?.tenantId) return

  const event = {
    type: `fleet:${change.operationType}`,
    id: ++this.eventCounter,
    data: doc,
  }

  // Publish to Redis — all nodes (including this one) will receive and deliver locally
  await backplane.publish(doc.tenantId, event)
}

Timeline:
t=0  → Node 2 starts restarting (rolling deployment)
t=1  → Redis: PUBLISH tenant:T1 (event #1042)
       → Node 1 subscriber receives ✅
       → Node 2 subscriber: DISCONNECTED — message dropped ❌
       → Node 3 subscriber receives ✅

t=5  → Node 2 restarts, reconnects to Redis
       → Node 2 missed event #1042 — no replay mechanism exists in Pub/Sub
       → Clients on Node 2 have a gap in their event stream

# Redis config: redis.conf
# Default client output buffer for pubsub subscribers
client-output-buffer-limit pubsub 32mb 8mb 60

# This means:
# - Hard limit: 32MB → subscriber disconnected immediately if buffer exceeds 32MB
# - Soft limit: 8MB for 60 consecutive seconds → subscriber disconnected after 60s at 8MB

// Mitigation: rate-limit broadcasts and batch multi-tenant publishes
async function batchPublish(events: Array<{ tenantId: string; event: unknown }>) {
  const pipeline = backplane.publisher.multi()
  for (const { tenantId, event } of events) {
    pipeline.publish(`tenant:${tenantId}`, JSON.stringify(event))
  }
  await pipeline.exec() // Atomic pipeline — reduces round-trips
}

Redis Pub/Sub:
PUBLISH channel message → 0 or more subscribers receive it → gone forever

Redis Streams:
XADD stream * key value → persisted to stream → consumers read at their own pace
XREADGROUP → reads from consumer group offset → explicit XACK required

// server/lib/RedisStreamsBackplane.ts
import { createClient } from 'redis'

const STREAM_KEY = (tenantId: string) => `events:${tenantId}`
const GROUP_NAME = 'fleet-nodes'
const MAX_STREAM_LENGTH = 10_000 // Capped stream — prevents OOM

class RedisStreamsBackplane {
  private client = createClient({ url: process.env.REDIS_URL })
  private consumerId = `node-${process.env.NODE_ID ?? randomUUID()}`

  async connect(): Promise<void> {
    await this.client.connect()
  }

  /** Publish event to the stream (called by ChangeStreamWorker) */
  async publish(tenantId: string, event: unknown): Promise<void> {
    await this.client.xAdd(
      STREAM_KEY(tenantId),
      '*', // Auto-generated ID (timestamp-based)
      { payload: JSON.stringify(event) },
      {
        // Critical: cap the stream to prevent unbounded memory growth
        TRIM: { strategy: 'MAXLEN', threshold: MAX_STREAM_LENGTH, strategyModifier: '~' },
      },
    )
  }

  /** Consumer group reader — called by each Node.js instance */
  async startConsuming(tenantId: string): Promise<void> {
    const streamKey = STREAM_KEY(tenantId)

    // Create consumer group if it doesn't exist
    try {
      await this.client.xGroupCreate(streamKey, GROUP_NAME, '0', { MKSTREAM: true })
    } catch (err) {
      if (!(err as Error).message.includes('BUSYGROUP')) throw err
    }

    // Continuous consumer loop
    while (true) {
      const results = await this.client.xReadGroup(
        GROUP_NAME,
        this.consumerId,
        { key: streamKey, id: '>' }, // '>' = only undelivered messages
        { COUNT: 100, BLOCK: 2000 }, // Block 2s waiting for new messages
      )

      for (const stream of results ?? []) {
        for (const message of stream.messages) {
          const event = JSON.parse(message.message.payload)
          connectionManager.broadcastToTenant(tenantId, event)

          // Acknowledge: removes from pending list — won't be redelivered
          await this.client.xAck(streamKey, GROUP_NAME, message.id)
        }
      }
    }
  }

  /** On restart: claim and reprocess any unacknowledged messages */
  async recoverPending(tenantId: string): Promise<void> {
    const streamKey = STREAM_KEY(tenantId)
    const CLAIM_AFTER_MS = 30_000 // Claim messages unacked for >30s

    const pending = await this.client.xAutoClaim(
      streamKey,
      GROUP_NAME,
      this.consumerId,
      CLAIM_AFTER_MS,
      '0-0', // Start from the beginning of pending entries
    )

    for (const message of pending.messages) {
      const event = JSON.parse(message.message.payload)
      connectionManager.broadcastToTenant(tenantId, event)
      await this.client.xAck(streamKey, GROUP_NAME, message.id)
    }
  }
}

// ❌ Uncapped stream — production OOM crash waiting to happen
await redis.xAdd('events:T1', '*', { payload: data })

// At 1,000 events/second × 200 bytes/event × 86,400 seconds/day:
// = 17.28 GB/day of stream data per active tenant
// Redis is an in-memory database — this kills the server

// ✅ Capped stream with MAXLEN
await redis.xAdd('events:T1', '*', { payload: data }, {
  TRIM: {
    strategy: 'MAXLEN',
    strategyModifier: '~', // Approximate trim (faster than exact)
    threshold: 10_000,     // Keep last 10,000 events per tenant
  },
})

// ❌ Global broadcast — every tenant receives every event
redis.publish('global:events', payload)

// ✅ Tenant-scoped channels — each node only processes events for its connected tenants
redis.publish(`tenant:${tenantId}`, payload)

// ✅ Room-scoped channels for even finer granularity
redis.publish(`room:${vehicleId}`, payload)
// Only nodes with clients subscribed to vehicleId receive the message