@jamx-framework/realtime

Descripción

Módulo de comunicación en tiempo real para JAMX Framework. Proporciona capacidades de WebSocket, Server-Sent Events (SSE) y presencia (presence) para construir aplicaciones con actualizaciones en tiempo real, chat, notificaciones push, y colaboración en vivo. Soporta tanto servidor como cliente, con manejo automático de reconexión, canales (rooms) y broadcast.

Cómo funciona

El módulo implementa un sistema de eventos en tiempo real con múltiples transportes:

1. WebSocketServer/Client: Comunicación bidireccional full-duplex
2. SSEServer: Server-Sent Events para push unidireccional desde servidor
3. EventEmitter: Emisión y suscripción a eventos con canales
4. Presence: Sistema de presencia para rastrear usuarios conectados y su estado

Componentes principales

• src/websocket-server.ts: Servidor WebSocket que maneja conexiones y broadcast
• src/websocket-client.ts: Cliente WebSocket para navegadores/Node
• src/sse-server.ts: Servidor SSE para eventos server → client
• src/event-emitter.ts: Emisor de eventos con canales (rooms)
• src/presence.ts: Gestor de presencia (usuarios en línea, estados)
• src/types.ts: Tipos compartidos (Message, Channel, Presence, etc.)
• src/index.ts: Punto de exportación

Uso básico

import { WebSocketServer, EventEmitter } from '@jamx-framework/realtime';

// Crear servidor WebSocket
const wss = new WebSocketServer({ port: 8080 });

// Emisor de eventos
const emitter = new EventEmitter();

// Suscribirse a canal
emitter.on('chat:general', (message) => {
  console.log('Mensaje en canal general:', message);
});

// Publicar en canal
emitter.emit('chat:general', { user: 'Juan', text: '¡Hola!' });

// Con WebSocket, broadcast a todos los conectados
wss.onConnection((ws) => {
  ws.onMessage((data) => {
    const msg = JSON.parse(data);
    // Broadcast a todos los clientes
    wss.broadcast(JSON.stringify(msg));
  });
});

Ejemplos

Servidor WebSocket completo

import { WebSocketServer, EventEmitter } from '@jamx-framework/realtime';

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

// Manejar conexiones
wss.onConnection((ws, req) => {
  const userId = req.headers['x-user-id'];
  
  // Unir a canal de chat
  ws.join('chat:general');
  
  // Escuchar mensajes
  ws.onMessage((data) => {
    const message = JSON.parse(data);
    
    // Validar y procesar
    if (message.type === 'chat') {
      // Broadcast a todos en el canal
      wss.broadcastTo('chat:general', JSON.stringify({
        type: 'chat',
        user: userId,
        text: message.text,
        timestamp: Date.now(),
      }));
    }
  });
  
  // Manejar disconnect
  ws.onClose(() => {
    wss.leave('chat:general', ws);
  });
});

console.log('WebSocket server listening on ws://localhost:8080');

Cliente WebSocket

import { WebSocketClient } from '@jamx-framework/realtime';

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

// Conectar
await client.connect();

// Unirse a canal
client.subscribe('chat:general');

// Escuchar mensajes
client.onMessage((data) => {
  const msg = JSON.parse(data);
  console.log('Mensaje recibido:', msg);
});

// Enviar mensaje
client.send(JSON.stringify({
  type: 'chat',
  text: '¡Hola a todos!',
}));

// Desconectar
client.disconnect();

Server-Sent Events (SSE)

import { SSEServer } from '@jamx-framework/realtime';

const sse = new SSEServer();

// Cliente se conecta a /events
sse.onConnection((req, res) => {
  const userId = req.headers['x-user-id'];
  
  // Unir a canal de notificaciones
  sse.join(`user:${userId}:notifications`);
  
  // Enviar evento inicial
  sse.send(`user:${userId}:notifications`, {
    type: 'connected',
    message: 'Conectado a notificaciones',
  });
  
  // Mantener conexión viva
  const interval = setInterval(() => {
    sse.send(`user:${userId}:notifications`, { type: 'ping' });
  }, 30000);
  
  // Limpiar al desconectar
  req.on('close', () => {
    clearInterval(interval);
    sse.leave(`user:${userId}:notifications`);
  });
});

// Desde cualquier parte, enviar notificación a usuario
sse.send(`user:123:notifications`, {
  type: 'notification',
  title: 'Nuevo mensaje',
  body: 'Tienes un mensaje nuevo',
});

Sistema de presencia

import { Presence } from '@jamx-framework/realtime';

const presence = new Presence();

// Usuario se conecta
presence.userConnected('user-123', {
  status: 'online',
  lastSeen: Date.now(),
  metadata: { device: 'mobile', location: 'US' },
});

// Usuario se desconecta
presence.userDisconnected('user-123');

// Obtener usuarios en línea
const onlineUsers = await presence.getOnlineUsers();
console.log('Usuarios en línea:', onlineUsers);

// Obtener estado de usuario específico
const status = await presence.getUserStatus('user-123');
console.log('Estado:', status); // 'online' | 'offline' | 'away'

// Escuchar cambios de presencia
presence.onPresenceChange((userId, status) => {
  console.log(`Usuario ${userId} cambió a ${status}`);
  // Actualizar UI, notificar a otros, etc.
});

// Establecer estado personalizado
await presence.setStatus('user-123', {
  status: 'away',
  lastSeen: Date.now(),
  metadata: { reason: 'inactive' },
});

Canales (rooms) y broadcast selectivo

import { EventEmitter } from '@jamx-framework/realtime';

const emitter = new EventEmitter();

// Crear canales
const channel1 = 'room:123';
const channel2 = 'room:456';

// Suscribirse a múltiples canales
emitter.subscribe(['room:123', 'room:456']);

// Escuchar eventos en canal específico
emitter.on('room:123', (data) => {
  console.log('Mensaje para room 123:', data);
});

// Broadcast a un canal específico
emitter.broadcastTo('room:123', { message: 'Hello room 123!' });

// Broadcast a todos los canales
emitter.broadcastAll({ message: 'Global announcement' });

// Unirse/irse de canales dinámicamente
emitter.subscribe('dynamic-channel');
emitter.leave('dynamic-channel');

Integración con JAMX Server

import { Server } from '@jamx-framework/server';
import { WebSocketServer } from '@jamx-framework/realtime';

const server = new Server();
const wss = new WebSocketServer();

// Middleware que inyecta WebSocket en request
server.use(async (req, res, next) => {
  const ws = wss.getConnectionForRequest(req);
  if (ws) {
    req.locals.ws = ws;
  }
  next();
});

// Handler que envía notificación via WebSocket
server.post('/notify', async (req, res) => {
  const { userId, message } = req.body;
  
  // Enviar notificación en tiempo real
  wss.sendToUser(userId, JSON.stringify({
    type: 'notification',
    message,
  }));
  
  res.ok({ sent: true });
});

Presencia con WebSocket

import { WebSocketServer, Presence } from '@jamx-framework/realtime';

const wss = new WebSocketServer();
const presence = new Presence();

wss.onConnection((ws, req) => {
  const userId = req.headers['x-user-id'];
  
  // Marcar usuario como en línea
  presence.userConnected(userId, {
    connectedAt: Date.now(),
    ip: req.ip,
    userAgent: req.headers['user-agent'],
  });
  
  // Enviar lista de usuarios en línea
  const online = await presence.getOnlineUsers();
  ws.send(JSON.stringify({ type: 'presence', users: online }));
  
  ws.onClose(() => {
    presence.userDisconnected(userId);
  });
});

Flujo interno

1. Conexión: Cliente se conecta via WebSocket o SSE
2. Handshake: Servidor acepta y asigna ID de conexión
3. Suscripción: Cliente se suscribe a canales (rooms)
4. Emisión: Cualquier parte emite eventos a canales
5. Broadcast: Servidor envía mensajes a todos los suscriptores del canal
6. Presencia: Sistema rastrea quién está conectado y a qué canales
7. Reconexión: Clientes pueden reconectarse automáticamente
8. Cleanup: Al desconectar, se limpian suscripciones y presencia

API Reference (Resumen)

WebSocketServer

• constructor(options: WebSocketServerOptions)
• onConnection(handler: (ws: WebSocket, req: Request) => void): void
• broadcast(message: string, exclude?: WebSocket[]): void
• broadcastTo(channel: string, message: string): void
• sendToUser(userId: string, message: string): void
• getConnections(): WebSocket[]
• close(): Promise<void>

WebSocketClient

• constructor(url: string, options?: ClientOptions)
• connect(): Promise<void>
• disconnect(): void
• subscribe(channel: string): void
• unsubscribe(channel: string): void
• send(data: string | object): void
• onMessage(handler: (data: any) => void): void
• onClose(handler: () => void): void
• onError(handler: (err: Error) => void): void

SSEServer

• constructor()
• onConnection(handler: (req: Request, res: Response) => void): void
• join(channel: string, res: Response): void
• leave(channel: string, res: Response): void
• send(channel: string, data: any): void
• broadcast(channel: string, data: any): void

EventEmitter

• subscribe(channels: string | string[]): void
• unsubscribe(channels: string | string[]): void
• on(channel: string, handler: (data: any) => void): void
• off(channel: string, handler?: Function): void
• emit(channel: string, data: any): void
• broadcastTo(channel: string, data: any): void
• broadcastAll(data: any): void

Presence

• userConnected(userId: string, metadata?: any): void
• userDisconnected(userId: string): void
• setStatus(userId: string, status: PresenceStatus): void
• getUserStatus(userId: string): Promise<PresenceStatus>
• getOnlineUsers(): Promise<string[]>
• onPresenceChange(handler: (userId: string, status: string) => void): void

Performance Considerations

• Connection pooling: Los servidores manejan miles de conexiones concurrentes
• Broadcast optimization: Los mensajes a canales se envían solo a suscriptores
• Memory usage: Cada conexión consume memoria; limitar canales por conexión
• Heartbeat: Ping/pong automático para detectar conexiones muertas
• Compression: Soporte para permessage-deflate en WebSocket

Configuration Options

// WebSocketServer
const wss = new WebSocketServer({
  port: 8080,
  maxConnections: 10000,
  pingInterval: 30000,    // ms entre pings
  pingTimeout: 5000,      // timeout de pong
  verifyClient: (info) => Promise.resolve(true), // auth custom
  channels: ['chat', 'notifications', 'presence'],
});

// WebSocketClient
const client = new WebSocketClient('ws://localhost:8080', {
  reconnect: true,
  reconnectInterval: 3000,
  maxReconnectAttempts: 10,
  protocols: ['json', 'binary'],
});

Testing

Tests en packages/realtime/tests/unit/:

pnpm test

Cubre:

• Conexión/desconexión WebSocket
• Suscripción a canales
• Broadcast de mensajes
• Sistema de presencia
• SSE server
• Reconexión automática

Compatibility

• Compatible con Node.js 18+
• Navegadores modernos (WebSocket API, EventSource)
• Funciona con cualquier WebSocket client (ws, socket.io, etc.)
• SSE funciona en IE con polyfill

CLI Integration

• jamx realtime:server: Inicia servidor WebSocket/SSE
• jamx realtime:clients: Muestra clientes conectados
• jamx realtime:channels: Lista canales activos y suscriptores
• jamx realtime:presence: Muestra usuarios en línea
• jamx realtime:broadcast <channel> <message>: Envía mensaje a canal

Security Considerations

1. Authentication: Verificar identidad en handshake (JWT, session)
2. Authorization: Validar permisos para unirse a canales
3. Input validation: Sanitizar mensajes entrantes
4. Rate limiting: Limitar mensajes por conexión
5. Origin check: Validar Origin header para evitar CSRF
6. TLS: Usar wss:// en producción

Best Practices

1. Canales con nombres jerárquicos: room:123, user:456:notifications
2. Mensajes pequeños: No enviar payloads grandes; usar referencias
3. Heartbeat: Configurar ping/pong para detectar conexiones caídas
4. Graceful shutdown: Cerrar conexiones limpiamente al apagar servidor
5. Monitor connections: Rastrear número de conexiones por canal
6. Backpressure: Manejar cuando el cliente no puede seguir el ritmo
7. Message ordering: Los mensajes en un canal mantienen orden

Troubleshooting

Conexiones se caen

Verificar firewall, proxies, y que el puerto esté abierto.

Mensajes no llegan

Verificar que el cliente esté suscrito al canal correcto.

Alta latencia

Reducir tamaño de mensajes, usar compresión, optimizar broadcast.

Memory leak

Asegurarse de limpiar suscripciones al desconectar.

This realtime module provides a powerful, scalable solution for real-time communication in JAMX applications, enabling features like live chat, notifications, presence, and collaborative editing with minimal code.