Real-time APIs with WebSockets: Implementation Guide

// Node.js WebSocket server
import { WebSocketServer, WebSocket } from 'ws';

const wss = new WebSocketServer({ port: 8080 });

interface Client {
  ws: WebSocket;
  userId?: string;
}

const clients = new Map<WebSocket, Client>();

wss.on('connection', (ws, req) => {
  // Store client
  clients.set(ws, { ws });
  
  ws.on('message', (message) => {
    const data = JSON.parse(message.toString());
    
    switch (data.type) {
      case 'auth':
        // Authenticate user
        const client = clients.get(ws);
        if (client) client.userId = data.userId;
        break;
        
      case 'message':
        // Broadcast to room
        broadcastToRoom(data.room, {
          type: 'message',
          content: data.content,
          from: clients.get(ws)?.userId,
        });
        break;
        
      case 'join':
        // Join room
        ws.send(JSON.stringify({ type: 'joined', room: data.room }));
        break;
    }
  });
  
  ws.on('close', () => {
    clients.delete(ws);
  });
});

function broadcastToRoom(room: string, message: object) {
  const msg = JSON.stringify(message);
  clients.forEach((client) => {
    client.ws.send(msg);
  });
}

// Client WebSocket
class RealTimeClient {
  private ws: WebSocket;
  private handlers = new Map<string, (data: any) => void>();
  
  constructor(url: string) {
    this.ws = new WebSocket(url);
    
    this.ws.onopen = () => {
      console.log('Connected');
    };
    
    this.ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      const handler = this.handlers.get(data.type);
      if (handler) handler(data);
    };
  }
  
  on(type: string, handler: (data: any) => void) {
    this.handlers.set(type, handler);
  }
  
  send(type: string, data: any) {
    this.ws.send(JSON.stringify({ type, ...data }));
  }
}

// Usage
const client = new RealTimeClient('ws://localhost:8080');

client.on('message', (data) => {
  console.log('New message:', data.content);
});

client.send('auth', { token: 'user-token' });

// Socket.io server
import { Server } from 'socket.io';
import { createServer } from 'http';

const httpServer = createServer(app);
const io = new Server(httpServer, {
  cors: {
    origin: '*',
  },
});

io.on('connection', (socket) => {
  console.log('Client connected:', socket.id);
  
  // Join room
  socket.on('join-room', (roomId: string) => {
    socket.join(roomId);
    socket.to(roomId).emit('user-joined', socket.id);
  });
  
  // Leave room
  socket.on('leave-room', (roomId: string) => {
    socket.leave(roomId);
  });
  
  // Send message to room
  socket.on('chat-message', (data: { room: string; message: string }) => {
    io.to(data.room).emit('chat-message', {
      message: data.message,
      sender: socket.id,
      timestamp: Date.now(),
    });
  });
  
  // Private message
  socket.on('private-message', (data: { to: string; message: string }) => {
    io.to(data.to).emit('private-message', {
      message: data.message,
      from: socket.id,
    });
  });
  
  socket.on('disconnect', () => {
    console.log('Client disconnected:', socket.id);
  });
});

httpServer.listen(3000);

// Socket.io client
import { io, Socket } from 'socket.io-client';

const socket: Socket = io('http://localhost:3000', {
  auth: { token: 'user-token' },
  transports: ['websocket'], // Force WebSocket
});

socket.on('connect', () => {
  console.log('Connected:', socket.id);
});

// Join room
socket.emit('join-room', 'room-123');

// Listen for messages
socket.on('chat-message', (data) => {
  console.log('New message:', data.message);
});

// Disconnect
socket.on('disconnect', () => {
  console.log('Disconnected');
});

// Express SSE endpoint
app.get('/api/events', (req, res) => {
  // Set headers for SSE
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  res.flushHeaders();
  
  // Send initial message
  res.write('data: {"status": "connected"}\n\n');
  
  // Send updates periodically
  const interval = setInterval(() => {
    const data = JSON.stringify({ 
      time: new Date().toISOString(),
      users: getOnlineUsers() 
    });
    res.write(`data: ${data}\n\n`);
  }, 5000);
  
  // Clean up on close
  req.on('close', () => {
    clearInterval(interval);
    res.end();
  });
});

// SSE client
const eventSource = new EventSource('/api/events');

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log('Update:', data);
};

eventSource.onerror = (error) => {
  console.error('SSE error:', error);
  eventSource.close();
};

// Clean up
// eventSource.close();

// Scale with Redis (Socket.io)
import { createAdapter } from '@socket.io/redis-adapter';
import { createClient } from 'redis';

const pubClient = createClient({ url: process.env.REDIS_URL });
const subClient = pubClient.duplicate();

const io = new Server({
  adapter: createAdapter(pubClient, subClient),
});

// Now handles multiple server instances

# Production considerations
production:
  load_balancing:
    - "Sticky sessions required"
    - "Redis for state"
    
  monitoring:
    - "Track connected clients"
    - "Alert on disconnections"
    - "Measure message throughput"
    
  security:
    - "WSS (WebSocket Secure)"
    - "Authentication on connect"
    - "Rate limiting"

// Supabase Realtime
import { createClient } from '@supabase/supabase-js';

const supabase = createClient(
  process.env.SUPABASE_URL!,
  process.env.SUPABASE_KEY!
);

// Subscribe to database changes
const channel = supabase
  .channel('posts')
  .on('postgres_changes', {
    event: '*',
    schema: 'public',
    table: 'posts',
  }, (payload) => {
    console.log('Change received!', payload);
  })
  .subscribe();

// Unsubscribe
channel.unsubscribe();