Real Time Systems

Real-Time Systems

Build real-time communication systems with WebSocket, SSE, and pub/sub patterns. This skill covers connection management, scaling, and production deployment.

Purpose

Implement live features that users expect:

• Real-time messaging and chat
• Live notifications and updates
• Collaborative editing
• Presence detection
• Live dashboards and metrics
• Gaming and interactive experiences

Features

1. WebSocket Server with Socket.io

import { Server } from 'socket.io';
import { createAdapter } from '@socket.io/redis-adapter';
import { createClient } from 'redis';

// Initialize Socket.io with Redis adapter for scaling
async function createSocketServer(httpServer: http.Server) {
  const io = new Server(httpServer, {
    cors: {
      origin: process.env.CLIENT_URL,
      credentials: true,
    },
    pingTimeout: 60000,
    pingInterval: 25000,
  });

  // Redis adapter for multi-server deployment
  const pubClient = createClient({ url: process.env.REDIS_URL });
  const subClient = pubClient.duplicate();
  await Promise.all([pubClient.connect(), subClient.connect()]);
  io.adapter(createAdapter(pubClient, subClient));

  // Authentication middleware
  io.use(async (socket, next) => {
    const token = socket.handshake.auth.token;

    if (!token) {
      return next(new Error('Authentication required'));
    }

    try {
      const user = await verifyToken(token);
      socket.data.user = user;
      next();
    } catch (error) {
      next(new Error('Invalid token'));
    }
  });

  // Connection handling
  io.on('connection', (socket) => {
    const userId = socket.data.user.id;
    console.log(`User connected: ${userId}`);

    // Join user's personal room
    socket.join(`user:${userId}`);

    // Handle joining rooms
    socket.on('join:room', async (roomId: string) => {
      // Verify access
      const hasAccess = await checkRoomAccess(userId, roomId);
      if (!hasAccess) {
        socket.emit('error', { message: 'Access denied' });
        return;
      }

      socket.join(`room:${roomId}`);
      socket.to(`room:${roomId}`).emit('user:joined', {
        userId,
        username: socket.data.user.name,
      });
    });

    // Handle messages
    socket.on('message:send', async (data: { roomId: string; content: string }) => {
      const message = await saveMessage({
        roomId: data.roomId,
        userId,
        content: data.content,
      });

      io.to(`room:${data.roomId}`).emit('message:new', message);
    });

    // Typing indicators
    socket.on('typing:start', (roomId: string) => {
      socket.to(`room:${roomId}`).emit('typing:user', {
        userId,
        username: socket.data.user.name,
        typing: true,
      });
    });

    socket.on('typing:stop', (roomId: string) => {
      socket.to(`room:${roomId}`).emit('typing:user', {
        userId,
        typing: false,
      });
    });

    // Presence
    socket.on('presence:update', async (status: 'online' | 'away' | 'busy') => {
      await updatePresence(userId, status);
      io.emit('presence:changed', { userId, status });
    });

    // Disconnect handling
    socket.on('disconnect', async (reason) => {
      console.log(`User disconnected: ${userId}, reason: ${reason}`);
      await updatePresence(userId, 'offline');
      io.emit('presence:changed', { userId, status: 'offline' });
    });
  });

  return io;
}

2. Server-Sent Events (SSE)

import { Router } from 'express';

const router = Router();

// SSE endpoint for notifications
router.get('/events/notifications', authenticate, (req, res) => {
  const userId = req.user.id;

  // Set SSE headers
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  res.setHeader('X-Accel-Buffering', 'no'); // Disable nginx buffering

  // Send initial connection event
  res.write(`event: connected\ndata: ${JSON.stringify({ userId })}\n\n`);

  // Keep-alive ping
  const pingInterval = setInterval(() => {
    res.write(`: ping\n\n`);
  }, 30000);

  // Subscribe to user's notifications
  const subscription = pubsub.subscribe(`notifications:${userId}`, (message) => {
    res.write(`event: notification\ndata: ${JSON.stringify(message)}\n\n`);
  });

  // Cleanup on disconnect
  req.on('close', () => {
    clearInterval(pingInterval);
    subscription.unsubscribe();
    console.log(`SSE connection closed for user ${userId}`);
  });
});

// SSE for live updates (e.g., stock prices, metrics)
router.get('/events/stream/:channel', authenticate, async (req, res) => {
  const { channel } = req.params;

  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');

  // Send initial data
  const initialData = await getChannelData(channel);
  res.write(`event: initial\ndata: ${JSON.stringify(initialData)}\n\n`);

  // Stream updates
  const unsubscribe = subscribeToChannel(channel, (update) => {
    res.write(`event: update\ndata: ${JSON.stringify(update)}\n\n`);
  });

  // Handle retry on reconnection
  res.write(`retry: 3000\n\n`);

  req.on('close', () => {
    unsubscribe();
  });
});

// Client-side SSE handling
const EventSourceComponent = () => {
  useEffect(() => {
    const eventSource = new EventSource('/api/events/notifications', {
      withCredentials: true,
    });

    eventSource.onopen = () => {
      console.log('SSE connected');
    };

    eventSource.addEventListener('notification', (event) => {
      const notification = JSON.parse(event.data);
      showNotification(notification);
    });

    eventSource.onerror = (error) => {
      console.error('SSE error:', error);
      // EventSource auto-reconnects
    };

    return () => {
      eventSource.close();
    };
  }, []);

  return null;
};

3. Pub/Sub with Redis

import { createClient } from 'redis';

class PubSubService {
  private publisher: ReturnType<typeof createClient>;
  private subscriber: ReturnType<typeof createClient>;
  private handlers: Map<string, Set<(message: any) => void>> = new Map();

  async connect() {
    this.publisher = createClient({ url: process.env.REDIS_URL });
    this.subscriber = this.publisher.duplicate();

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

    // Handle incoming messages
    this.subscriber.on('message', (channel, message) => {
      const handlers = this.handlers.get(channel);
      if (handlers) {
        const parsed = JSON.parse(message);
        handlers.forEach(handler => handler(parsed));
      }
    });
  }

  async publish(channel: string, message: any): Promise<void> {
    await this.publisher.publish(channel, JSON.stringify(message));
  }

  subscribe(channel: string, handler: (message: any) => void): () => void {
    if (!this.handlers.has(channel)) {
      this.handlers.set(channel, new Set());
      this.subscriber.subscribe(channel);
    }

    this.handlers.get(channel)!.add(handler);

    // Return unsubscribe function
    return () => {
      const handlers = this.handlers.get(channel);
      if (handlers) {
        handlers.delete(handler);
        if (handlers.size === 0) {
          this.handlers.delete(channel);
          this.subscriber.unsubscribe(channel);
        }
      }
    };
  }

  // Pattern subscription
  async psubscribe(pattern: string, handler: (channel: string, message: any) => void): Promise<() => void> {
    await this.subscriber.pSubscribe(pattern, (message, channel) => {
      handler(channel, JSON.parse(message));
    });

    return () => {
      this.subscriber.pUnsubscribe(pattern);
    };
  }
}

const pubsub = new PubSubService();

// Usage in services
class NotificationService {
  async sendNotification(userId: string, notification: Notification): Promise<void> {
    // Save to database
    await db.notification.create({ data: { ...notification, userId } });

    // Publish to real-time channel
    await pubsub.publish(`notifications:${userId}`, notification);
  }

  async broadcastToRoom(roomId: string, event: string, data: any): Promise<void> {
    await pubsub.publish(`room:${roomId}`, { event, data });
  }
}

4. Presence System

interface PresenceData {
  status: 'online' | 'away' | 'busy' | 'offline';
  lastSeen: Date;
  socketIds: string[];
}

class PresenceService {
  private redis: ReturnType<typeof createClient>;
  private readonly PRESENCE_TTL = 300; // 5 minutes

  async setPresence(userId: string, socketId: string, status: string): Promise<void> {
    const key = `presence:${userId}`;

    // Use MULTI for atomic operations
    await this.redis.multi()
      .hSet(key, {
        status,
        lastSeen: Date.now().toString(),
      })
      .sAdd(`${key}:sockets`, socketId)
      .expire(key, this.PRESENCE_TTL)
      .exec();

    // Publish presence change
    await pubsub.publish('presence:updates', {
      userId,
      status,
      lastSeen: new Date(),
    });
  }

  async removeSocket(userId: string, socketId: string): Promise<void> {
    const key = `presence:${userId}`;

    await this.redis.sRem(`${key}:sockets`, socketId);
    const remaining = await this.redis.sCard(`${key}:sockets`);

    if (remaining === 0) {
      await this.redis.hSet(key, 'status', 'offline');
      await pubsub.publish('presence:updates', {
        userId,
        status: 'offline',
        lastSeen: new Date(),
      });
    }
  }

  async getPresence(userId: string): Promise<PresenceData | null> {
    const key = `presence:${userId}`;
    const data = await this.redis.hGetAll(key);

    if (!data.status) return null;

    return {
      status: data.status as PresenceData['status'],
      lastSeen: new Date(parseInt(data.lastSeen)),
      socketIds: await this.redis.sMembers(`${key}:sockets`),
    };
  }

  async getMultiplePresence(userIds: string[]): Promise<Map<string, PresenceData>> {
    const pipeline = this.redis.multi();

    userIds.forEach(id => {
      pipeline.hGetAll(`presence:${id}`);
    });

    const results = await pipeline.exec();
    const presenceMap = new Map<string, PresenceData>();

    userIds.forEach((id, index) => {
      const data = results[index] as Record<string, string>;
      if (data?.status) {
        presenceMap.set(id, {
          status: data.status as PresenceData['status'],
          lastSeen: new Date(parseInt(data.lastSeen)),
          socketIds: [],
        });
      }
    });

    return presenceMap;
  }
}

5. Connection Recovery

// Client-side reconnection logic
class ReconnectingWebSocket {
  private ws: WebSocket | null = null;
  private reconnectAttempts = 0;
  private maxReconnectAttempts = 10;
  private reconnectInterval = 1000;
  private messageQueue: any[] = [];

  constructor(
    private url: string,
    private options: {
      onMessage: (data: any) => void;
      onConnect: () => void;
      onDisconnect: () => void;
    }
  ) {
    this.connect();
  }

  private connect(): void {
    this.ws = new WebSocket(this.url);

    this.ws.onopen = () => {
      console.log('WebSocket connected');
      this.reconnectAttempts = 0;
      this.options.onConnect();

      // Flush queued messages
      while (this.messageQueue.length > 0) {
        const msg = this.messageQueue.shift();
        this.send(msg);
      }
    };

    this.ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      this.options.onMessage(data);
    };

    this.ws.onclose = (event) => {
      console.log('WebSocket closed:', event.code, event.reason);
      this.options.onDisconnect();
      this.scheduleReconnect();
    };

    this.ws.onerror = (error) => {
      console.error('WebSocket error:', error);
    };
  }

  private scheduleReconnect(): void {
    if (this.reconnectAttempts >= this.maxReconnectAttempts) {
      console.error('Max reconnection attempts reached');
      return;
    }

    const delay = Math.min(
      this.reconnectInterval * Math.pow(2, this.reconnectAttempts),
      30000 // Max 30 seconds
    );

    console.log(`Reconnecting in ${delay}ms...`);

    setTimeout(() => {
      this.reconnectAttempts++;
      this.connect();
    }, delay);
  }

  send(data: any): void {
    if (this.ws?.readyState === WebSocket.OPEN) {
      this.ws.send(JSON.stringify(data));
    } else {
      // Queue message for when connection is restored
      this.messageQueue.push(data);
    }
  }

  close(): void {
    this.maxReconnectAttempts = 0; // Prevent reconnection
    this.ws?.close();
  }
}

// Server-side missed message recovery
class MessageRecovery {
  async getMessagesSince(roomId: string, lastMessageId: string): Promise<Message[]> {
    // Fetch messages after the last seen message
    return db.message.findMany({
      where: {
        roomId,
        id: { gt: lastMessageId },
      },
      orderBy: { createdAt: 'asc' },
      take: 100, // Limit recovery batch
    });
  }

  async recoverClientState(userId: string, lastSyncTimestamp: number): Promise<{
    messages: Message[];
    notifications: Notification[];
    presenceUpdates: PresenceUpdate[];
  }> {
    const since = new Date(lastSyncTimestamp);

    return {
      messages: await this.getUnreadMessages(userId, since),
      notifications: await this.getUnreadNotifications(userId, since),
      presenceUpdates: await this.getPresenceChanges(since),
    };
  }
}

6. Scaling WebSockets

// Horizontal scaling with sticky sessions
// nginx.conf
upstream websocket_servers {
    ip_hash; // Sticky sessions
    server ws1.example.com:3000;
    server ws2.example.com:3000;
    server ws3.example.com:3000;
}

server {
    location /socket.io/ {
        proxy_pass http://websocket_servers;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_read_timeout 86400;
    }
}

// Broadcasting across servers
class ScaledBroadcaster {
  async broadcastToRoom(roomId: string, event: string, data: any): Promise<void> {
    // Publish to Redis - all servers will receive
    await pubsub.publish(`broadcast:room:${roomId}`, {
      event,
      data,
      timestamp: Date.now(),
    });
  }

  // Each server subscribes and emits locally
  setupBroadcastListener(io: Server): void {
    pubsub.psubscribe('broadcast:*', (channel, message) => {
      const [, type, id] = channel.split(':');

      if (type === 'room') {
        io.to(`room:${id}`).emit(message.event, message.data);
      } else if (type === 'user') {
        io.to(`user:${id}`).emit(message.event, message.data);
      }
    });
  }
}

Use Cases

1. Chat Application

// Real-time chat with typing indicators and read receipts
socket.on('chat:message', async (data) => {
  const message = await createMessage(data);
  io.to(`room:${data.roomId}`).emit('chat:message', message);
});

socket.on('chat:read', async ({ roomId, messageId }) => {
  await markAsRead(socket.data.user.id, roomId, messageId);
  socket.to(`room:${roomId}`).emit('chat:read', {
    userId: socket.data.user.id,
    messageId,
  });
});

2. Live Dashboard

// Real-time metrics with SSE
setInterval(async () => {
  const metrics = await gatherMetrics();
  await pubsub.publish('dashboard:metrics', metrics);
}, 5000);

Best Practices

Do's

• Implement heartbeat/ping - Detect dead connections
• Handle reconnection gracefully - Queue messages, recover state
• Use rooms for scaling - Don't broadcast to all
• Implement backpressure - Handle slow clients
• Plan for offline scenarios - Message queuing
• Monitor connection metrics - Track active connections

Don'ts

• Don't trust client data without validation
• Don't skip authentication
• Don't broadcast sensitive data
• Don't ignore connection limits
• Don't forget cleanup on disconnect
• Don't use WebSocket for everything

Related Skills

• redis - Pub/sub and state management
• backend-development - Server architecture
• api-architecture - REST fallbacks

Reference Resources

• Socket.io Documentation
• WebSocket MDN
• SSE MDN

Converted and distributed by TomeVault — claim your Tome and manage your conversions.