Diseño de Sistemas: Aplicación de Chat

// Gestión de conexiones WebSocket
import { WebSocketServer, WebSocket } from "ws";

interface ConnectedUser {
  userId: string;
  socket: WebSocket;
  serverId: string; // A qué servidor de chat está conectado este usuario
}

class ChatWebSocketServer {
  private connections = new Map<string, WebSocket>();
  private wss: WebSocketServer;

  constructor(port: number) {
    this.wss = new WebSocketServer({ port });
    this.wss.on("connection", this.handleConnection.bind(this));
  }

  private handleConnection(socket: WebSocket, request: any) {
    const userId = this.authenticateUser(request);
    if (!userId) {
      socket.close(4001, "Unauthorized");
      return;
    }

    // Registrar conexión
    this.connections.set(userId, socket);
    this.updatePresence(userId, "online");

    socket.on("message", (data) => this.handleMessage(userId, data));

    socket.on("close", () => {
      this.connections.delete(userId);
      this.updatePresence(userId, "offline");
    });

    // Heartbeat para detectar conexiones obsoletas
    socket.on("pong", () => { /* la conexión está viva */ });
  }

  async sendToUser(targetUserId: string, message: any): Promise<boolean> {
    const socket = this.connections.get(targetUserId);
    if (socket && socket.readyState === WebSocket.OPEN) {
      socket.send(JSON.stringify(message));
      return true; // Entregado
    }
    return false; // Usuario no está en este servidor
  }
}

// Flujo de mensajes: Usuario A envía un mensaje al Usuario B

interface ChatMessage {
  id: string;            // ID globalmente único (Snowflake)
  conversationId: string;
  senderId: string;
  content: string;
  contentType: "text" | "image" | "file";
  timestamp: number;
  status: "sent" | "delivered" | "read";
}

class MessageRouter {
  private redis: RedisClient;
  private kafka: KafkaProducer;
  private pushService: PushNotificationService;

  async routeMessage(message: ChatMessage, recipientId: string): Promise<void> {
    // Paso 1: Persistir el mensaje
    await this.kafka.publish("messages", {
      key: message.conversationId, // Particionar por conversación para ordenamiento
      value: message,
    });

    // Paso 2: Encontrar en qué servidor está conectado el destinatario
    const serverInfo = await this.redis.get(`conn:${recipientId}`);

    if (serverInfo) {
      // Usuario en línea - enrutar a su servidor de chat
      const { serverId } = JSON.parse(serverInfo);
      await this.forwardToServer(serverId, recipientId, message);
    } else {
      // Usuario desconectado - enviar notificación push
      await this.pushService.send(recipientId, {
        title: `Nuevo mensaje de ${message.senderId}`,
        body: message.content.substring(0, 100),
        data: { conversationId: message.conversationId },
      });
    }
  }

  private async forwardToServer(
    serverId: string,
    recipientId: string,
    message: ChatMessage
  ): Promise<void> {
    // Usar RPC interno o pub/sub para alcanzar el servidor correcto
    await this.redis.publish(`server:${serverId}`, JSON.stringify({
      type: "deliver",
      recipientId,
      message,
    }));
  }
}

// Esquema de almacenamiento de mensajes (Cassandra)
// Clave de partición: conversation_id
// Clave de clustering: message_id (Snowflake, así está ordenado por tiempo)

// CREATE TABLE messages (
//   conversation_id TEXT,
//   message_id BIGINT,
//   sender_id TEXT,
//   content TEXT,
//   content_type TEXT,
//   created_at TIMESTAMP,
//   PRIMARY KEY (conversation_id, message_id)
// ) WITH CLUSTERING ORDER BY (message_id DESC);

class MessageStore {
  // Obtener historial de mensajes para una conversación (paginado)
  async getMessages(
    conversationId: string,
    beforeMessageId?: string,
    limit = 50
  ): Promise<ChatMessage[]> {
    let query = "SELECT * FROM messages WHERE conversation_id = ?";
    const params: any[] = [conversationId];

    if (beforeMessageId) {
      query += " AND message_id < ?";
      params.push(beforeMessageId);
    }

    query += " ORDER BY message_id DESC LIMIT ?";
    params.push(limit);

    return this.cassandra.execute(query, params);
  }

  // Almacenar un nuevo mensaje
  async saveMessage(message: ChatMessage): Promise<void> {
    await this.cassandra.execute(
      "INSERT INTO messages (conversation_id, message_id, sender_id, content, content_type, created_at) VALUES (?, ?, ?, ?, ?, ?)",
      [message.conversationId, message.id, message.senderId, message.content, message.contentType, message.timestamp]
    );
  }
}

interface GroupChat {
  id: string;
  name: string;
  memberIds: string[];
  adminIds: string[];
  createdAt: Date;
}

class GroupMessageHandler {
  async sendGroupMessage(
    groupId: string,
    message: ChatMessage
  ): Promise<void> {
    // Persistir el mensaje una vez (no por miembro)
    await this.messageStore.saveMessage(message);

    // Obtener miembros del grupo
    const members = await this.getGroupMembers(groupId);

    // Entregar a cada miembro (excepto el remitente)
    const deliveryPromises = members
      .filter((memberId) => memberId !== message.senderId)
      .map((memberId) => this.router.routeMessage(message, memberId));

    // Fan-out de entrega en paralelo
    await Promise.allSettled(deliveryPromises);
  }
}

// Para grupos grandes (>100 miembros), considerar:
// 1. Entrega por lotes para reducir overhead por mensaje
// 2. Usar un canal pub/sub por grupo en lugar de entrega individual
// 3. Limitar la tasa de mensajes para prevenir spam

class PresenceService {
  private redis: RedisClient;

  // Llamado cuando el usuario se conecta
  async setOnline(userId: string): Promise<void> {
    await this.redis.set(`presence:${userId}`, "online");
    // Notificar a amigos/contactos sobre cambio de estado
    await this.broadcastStatusChange(userId, "online");
  }

  // Llamado cuando el usuario se desconecta
  async setOffline(userId: string): Promise<void> {
    // No marcar desconectado inmediatamente (maneja desconexiones breves)
    await this.redis.set(`presence:${userId}`, "offline");
    await this.redis.set(`last_seen:${userId}`, Date.now().toString());
    await this.broadcastStatusChange(userId, "offline");
  }

  // Enfoque heartbeat: los clientes envían heartbeats cada 30 segundos
  // Si no se recibe heartbeat por 60 segundos, marcar como desconectado
  async heartbeat(userId: string): Promise<void> {
    await this.redis.setex(`heartbeat:${userId}`, 60, "alive");
    await this.redis.set(`presence:${userId}`, "online");
  }

  async isOnline(userId: string): Promise<boolean> {
    return (await this.redis.get(`presence:${userId}`)) === "online";
  }

  // Solo transmitir a usuarios que tienen conversación con este usuario
  // No transmitir a TODOS los usuarios (demasiado costoso)
  private async broadcastStatusChange(
    userId: string,
    status: string
  ): Promise<void> {
    const recentContacts = await this.getRecentContacts(userId);
    for (const contactId of recentContacts) {
      await this.router.sendToUser(contactId, {
        type: "presence_update",
        userId,
        status,
        lastSeen: status === "offline" ? Date.now() : undefined,
      });
    }
  }
}