real-time-systems

name	real-time-systems
description	WebSocket, Server-Sent Events, and real-time communication patterns for live features Use when this capability is needed.
metadata	{"author":"doanchienthangdev"}

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();
  });

  
  req.(,  {
    (pingInterval);
    subscription.();
    .();
  });
});


router.(, authenticate,  (req, res) => {
   { channel } = req.;

  res.(, );
  res.(, );
  res.(, );

  
   initialData =  (channel);
  res.();

  
   unsubscribe = (channel,  {
    res.();
  });

  
  res.();

  req.(,  {
    ();
  });
});


  = () => {
  ( {
     eventSource =  (, {
      : ,
    });

    eventSource. =  {
      .();
    };

    eventSource.(,  {
       notification = .(event.);
      (notification);
    });

    eventSource. =  {
      .(, error);
      
    };

      {
      eventSource.();
    };
  }, []);

   ;
};

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(,  {
       handlers = ..(channel);
       (handlers) {
         parsed = .(message);
        handlers.( (parsed));
      }
    });
  }

   (: , : ): <> {
     ..(channel, .(message));
  }

  (: , :  ):   {
     (!..(channel)) {
      ..(channel,  ());
      ..(channel);
    }

    ..(channel)!.(handler);

    
      {
       handlers = ..(channel);
       (handlers) {
        handlers.(handler);
         (handlers. === ) {
          ..(channel);
          ..(channel);
        }
      }
    };
  }

  
   (: , :  ): < > {
     ..(pattern,  {
      (channel, .(message));
    });

      {
      ..(pattern);
    };
  }
}

 pubsub =  ();


  {
   (: , : ): <> {
    
     db..({ : { ...notification, userId } });

    
     pubsub.(, notification);
  }

   (: , : , : ): <> {
     pubsub.(, { 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)
      .();

    
     pubsub.(, {
      userId,
      status,
      :  (),
    });
  }

   (: , : ): <> {
     key = ;

     ..(, socketId);
     remaining =  ..();

     (remaining === ) {
       ..(key, , );
       pubsub.(, {
        userId,
        : ,
        :  (),
      });
    }
  }

   (: ): < | > {
     key = ;
     data =  ..(key);

     (!data.)  ;

     {
      : data.  [],
      :  ((data.)),
      :  ..(),
    };
  }

   (: []): <<, >> {
     pipeline = ..();

    userIds.( {
      pipeline.();
    });

     results =  pipeline.();
     presenceMap =  <, >();

    userIds.( {
       data = results[index]  <, >;
       (data?.) {
        presenceMap.(id, {
          : data.  [],
          :  ((data.)),
          : [],
        });
      }
    });

     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.();

      
       (.. > ) {
         msg = ..();
        .(msg);
      }
    };

    .. =  {
       data = .(event.);
      ..(data);
    };

    .. =  {
      .(, event., event.);
      ..();
      .();
    };

    .. =  {
      .(, error);
    };
  }

   ():  {
     (. >= .) {
      .();
      ;
    }

     delay = .(
      . * .(, .),
       
    );

    .();

    ( {
      .++;
      .();
    }, delay);
  }

  (: ):  {
     (.?. === .) {
      ..(.(data));
    }  {
      
      ..(data);
    }
  }

  ():  {
    . = ; 
    .?.();
  }
}


  {
   (: , : ): <[]> {
    
     db..({
      : {
        roomId,
        : { : lastMessageId },
      },
      : { :  },
      : , 
    });
  }

   (: , : ): <{
    : [];
    : [];
    : [];
  }> {
     since =  (lastSyncTimestamp);

     {
      :  .(userId, since),
      :  .(userId, since),
      :  .(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(: ):  {
    pubsub.(,  {
       [, , id] = channel.();

       ( === ) {
        io.().(message., message.);
      }   ( === ) {
        io.().(message., message.);
      }
    });
  }
}

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.