realtime-systems

What this skill does

# Real-time Systems

## Overview

Building real-time applications with WebSocket, Server-Sent Events, and event-driven architectures.

---

## WebSocket

### Server Implementation (Node.js)

```typescript
import { WebSocketServer, WebSocket } from 'ws';
import { createServer } from 'http';
import { v4 as uuid } from 'uuid';

const server = createServer();
const wss = new WebSocketServer({ server });

interface Client {
  id: string;
  ws: WebSocket;
  userId?: string;
  rooms: Set<string>;
}

const clients = new Map<string, Client>();
const rooms = new Map<string, Set<string>>();

wss.on('connection', (ws, req) => {
  const clientId = uuid();
  const client: Client = {
    id: clientId,
    ws,
    rooms: new Set(),
  };

  clients.set(clientId, client);
  console.log(`Client connected: ${clientId}`);

  // Handle messages
  ws.on('message', (data) => {
    try {
      const message = JSON.parse(data.toString());
      handleMessage(client, message);
    } catch (error) {
      console.error('Invalid message:', error);
    }
  });

  // Handle disconnection
  ws.on('close', () => {
    // Leave all rooms
    client.rooms.forEach(room => leaveRoom(client, room));
    clients.delete(clientId);
    console.log(`Client disconnected: ${clientId}`);
  });

  // Send connection confirmation
  send(ws, { type: 'connected', clientId });
});

function handleMessage(client: Client, message: any) {
  switch (message.type) {
    case 'authenticate':
      client.userId = message.userId;
      break;

    case 'join':
      joinRoom(client, message.room);
      break;

    case 'leave':
      leaveRoom(client, message.room);
      break;

    case 'message':
      broadcastToRoom(message.room, {
        type: 'message',
        from: client.userId,
        content: message.content,
        timestamp: Date.now(),
      }, client.id);
      break;

    case 'ping':
      send(client.ws, { type: 'pong' });
      break;
  }
}

function joinRoom(client: Client, room: string) {
  if (!rooms.has(room)) {
    rooms.set(room, new Set());
  }
  rooms.get(room)!.add(client.id);
  client.rooms.add(room);

  // Notify room members
  broadcastToRoom(room, {
    type: 'user_joined',
    userId: client.userId,
    room,
  }, client.id);
}

function leaveRoom(client: Client, room: string) {
  rooms.get(room)?.delete(client.id);
  client.rooms.delete(room);

  // Notify room members
  broadcastToRoom(room, {
    type: 'user_left',
    userId: client.userId,
    room,
  });
}

function broadcastToRoom(room: string, message: any, excludeClientId?: string) {
  const roomClients = rooms.get(room);
  if (!roomClients) return;

  roomClients.forEach(clientId => {
    if (clientId !== excludeClientId) {
      const client = clients.get(clientId);
      if (client?.ws.readyState === WebSocket.OPEN) {
        send(client.ws, message);
      }
    }
  });
}

function send(ws: WebSocket, message: any) {
  ws.send(JSON.stringify(message));
}

server.listen(8080);
```

### Client Implementation

```typescript
class WebSocketClient {
  private ws: WebSocket | null = null;
  private reconnectAttempts = 0;
  private maxReconnectAttempts = 5;
  private reconnectDelay = 1000;
  private messageHandlers = new Map<string, Set<Function>>();
  private messageQueue: any[] = [];

  constructor(private url: string) {}

  connect(): Promise<void> {
    return new Promise((resolve, reject) => {
      this.ws = new WebSocket(this.url);

      this.ws.onopen = () => {
        console.log('WebSocket connected');
        this.reconnectAttempts = 0;
        this.flushMessageQueue();
        resolve();
      };

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

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

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

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

    this.reconnectAttempts++;
    const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1);

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

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

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

  private flushMessageQueue() {
    while (this.messageQueue.length > 0) {
      const message = this.messageQueue.shift();
      this.send(message);
    }
  }

  private handleMessage(message: any) {
    const handlers = this.messageHandlers.get(message.type);
    handlers?.forEach(handler => handler(message));

    // Also emit to wildcard handlers
    const wildcardHandlers = this.messageHandlers.get('*');
    wildcardHandlers?.forEach(handler => handler(message));
  }

  on(type: string, handler: Function) {
    if (!this.messageHandlers.has(type)) {
      this.messageHandlers.set(type, new Set());
    }
    this.messageHandlers.get(type)!.add(handler);

    // Return unsubscribe function
    return () => {
      this.messageHandlers.get(type)?.delete(handler);
    };
  }

  // Convenience methods
  joinRoom(room: string) {
    this.send({ type: 'join', room });
  }

  leaveRoom(room: string) {
    this.send({ type: 'leave', room });
  }

  sendMessage(room: string, content: string) {
    this.send({ type: 'message', room, content });
  }

  disconnect() {
    this.ws?.close();
    this.ws = null;
  }
}

// Usage
const ws = new WebSocketClient('wss://api.example.com/ws');

ws.on('connected', (msg) => {
  console.log('Connected with ID:', msg.clientId);
  ws.joinRoom('general');
});

ws.on('message', (msg) => {
  console.log(`[${msg.from}]: ${msg.content}`);
});

await ws.connect();
```

---

## Socket.IO

### Server

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

const httpServer = createServer();
const io = new Server(httpServer, {
  cors: {
    origin: process.env.ALLOWED_ORIGINS?.split(',') || '*',
    credentials: true,
  },
});

// Redis adapter for horizontal scaling
const pubClient = createClient({ url: process.env.REDIS_URL });
const subClient = pubClient.duplicate();

Promise.all([pubClient.connect(), subClient.connect()]).then(() => {
  io.adapter(createAdapter(pubClient, subClient));
});

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

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

// Namespace for chat
const chatNamespace = io.of('/chat');

chatNamespace.on('connection', (socket) => {
  const user = socket.data.user;
  console.log(`User connected: ${user.name}`);

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

  // Join a chat room
  socket.on('join_room', async (roomId: string) => {
    // Verify access
    const hasAccess = await checkRoomAccess(user.id, roomId);
    if (!hasAccess) {
      socket.emit('error', { message: 'Access denied' });
      return;
    }

    socket.join(roomId);

    // Notify room members
    socket.to(roomId).emit('user_joined', {
      userId: user.id,
      userName: user.name,
    });

    // Send recent messages
    const messages = await getRecentMessages(roomId, 50);
    socket.emit('room_history', { roomId, messages });
  });

  // Leave room
  socket.on('leave_room', (roomId: string) => {
    socket.leave(roomId);
    socket.to(roomId).emit('user