Construir notificaciones en tiempo real con WebSockets

Visión General

Hacer polling a un endpoint /notifications cada pocos segundos desperdicia peticiones y aun así entrega datos desactualizados. Y en cuanto levantas una segunda instancia, el enfoque ingenuo —guardar una lista de sockets abiertos y recorrerla— se rompe: la instancia A no puede alcanzar a un cliente conectado a la instancia B.

Esta receta conecta las conexiones WebSocket con Redis pub/sub para que cualquier instancia pueda enviar a cualquier cliente. Cubre entrega por rooms, un cliente de navegador con reconexión automática y los detalles operativos que mantienen el sistema vivo en producción.

Cuándo Usar

Elige este enfoque cuando:

• Los usuarios necesitan actualizaciones instantáneas (chat, alertas, dashboards en vivo).
• El polling genera demasiada carga en tu infraestructura.
• Ejecutas varias instancias de API detrás de un balanceador de carga y aun así necesitas alcanzar a cada cliente conectado.
• Un evento producido por un servicio tiene que llegar a usuarios conectados a otra instancia.

Solución

Python (FastAPI)

import asyncio
import json
from contextlib import asynccontextmanager

import redis.asyncio as redis
from fastapi import FastAPI, WebSocket
from fastapi.websockets import WebSocketDisconnect

redis_client = redis.from_url("redis://localhost:6379")


class ConnectionManager:
    def __init__(self):
        self.rooms: dict[str, set[WebSocket]] = {}

    async def join(self, room: str, websocket: WebSocket):
        await websocket.accept()
        self.rooms.setdefault(room, set()).add(websocket)

    def leave(self, room: str, websocket: WebSocket):
        self.rooms.get(room, set()).discard(websocket)

    async def broadcast(self, room: str, message: str):
        for connection in self.rooms.get(room, ()):
            await connection.send_text(message)


manager = ConnectionManager()


async def redis_listener():
    pubsub = redis_client.pubsub()
    await pubsub.subscribe("notifications")
    async for message in pubsub.listen():
        if message["type"] == "message":
            payload = json.loads(message["data"])
            await manager.broadcast(payload["room"], payload["text"])


@asynccontextmanager
async def lifespan(app: FastAPI):
    listener = asyncio.create_task(redis_listener())
    yield
    listener.cancel()


app = FastAPI(lifespan=lifespan)


@app.websocket("/ws/{room}")
async def websocket_endpoint(websocket: WebSocket, room: str):
    await manager.join(room, websocket)
    try:
        while True:
            await websocket.receive_text()  # mantiene viva la conexión
    except WebSocketDisconnect:
        manager.leave(room, websocket)


@app.post("/notify/{room}")
async def notify(room: str, text: str):
    payload = json.dumps({"room": room, "text": text})
    await redis_client.publish("notifications", payload)

JavaScript (Node.js)

const WebSocket = require('ws');
const Redis = require('ioredis');
const redis = new Redis();
const subscriber = new Redis();

const wss = new WebSocket.Server({ port: 8080 });
const rooms = new Map(); // nombre de room -> Set de clientes conectados

wss.on('connection', (ws, req) => {
  const room = new URL(req.url, 'http://localhost').searchParams.get('room') || 'default';
  if (!rooms.has(room)) rooms.set(room, new Set());
  rooms.get(room).add(ws);

  ws.on('close', () => {
    const members = rooms.get(room);
    members.delete(ws);
    if (members.size === 0) rooms.delete(room);
  });
});

subscriber.subscribe('notifications');
subscriber.on('message', (channel, message) => {
  const { room, text } = JSON.parse(message);
  for (const client of rooms.get(room) ?? []) {
    if (client.readyState === WebSocket.OPEN) {
      client.send(text);
    }
  }
});

// Cualquier parte de la API puede publicar una notificación:
// redis.publish('notifications', JSON.stringify({ room: 'orders', text: 'New order!' }));

Java (Spring)

@Component
public class NotificationHandler extends TextWebSocketHandler {
    private final Map<String, Set<WebSocketSession>> rooms = new ConcurrentHashMap<>();

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        rooms.computeIfAbsent(roomOf(session), k -> ConcurrentHashMap.newKeySet())
             .add(session);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        Set<WebSocketSession> members = rooms.get(roomOf(session));
        if (members != null) {
            members.remove(session);
        }
    }

    public void broadcast(String room, String message) throws IOException {
        for (WebSocketSession session : rooms.getOrDefault(room, Set.of())) {
            if (session.isOpen()) {
                session.sendMessage(new TextMessage(message));
            }
        }
    }

    private String roomOf(WebSocketSession session) {
        String query = session.getUri() != null ? session.getUri().getQuery() : "";
        return query != null && query.startsWith("room=") ? query.substring(5) : "default";
    }
}

El handler por sí solo no basta — también necesitas un bean que se suscriba a Redis y enrute los mensajes a broadcast():

@Configuration
public class RedisListenerConfig {
    @Bean
    RedisMessageListenerContainer redisContainer(RedisConnectionFactory factory,
                                                 RedisNotificationListener listener) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        container.addMessageListener(listener, new PatternTopic("notifications"));
        return container;
    }
}

@Component
public class RedisNotificationListener implements MessageListener {
    private final NotificationHandler handler;
    private final ObjectMapper mapper = new ObjectMapper();

    RedisNotificationListener(NotificationHandler handler) {
        this.handler = handler;
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        try {
            JsonNode payload = mapper.readTree(message.getBody());
            handler.broadcast(payload.get("room").asText(), payload.get("text").asText());
        } catch (IOException e) {
            // registra y descarta payloads malformados
        }
    }
}

Cliente de navegador

let retries = 0;

function connect() {
  const ws = new WebSocket('wss://api.example.com/ws?room=orders');

  ws.onopen = () => { retries = 0; };
  ws.onmessage = (event) => showNotification(event.data);
  ws.onclose = () => {
    const delay = Math.min(1000 * 2 ** retries++, 30_000);
    setTimeout(connect, delay);
  };
}

connect();

Archivos complementarios: el repo companion de real-time-notifications tiene un servidor Node.js ejecutable, el cliente de navegador de arriba y un docker-compose con Redis para probar todo el flujo en local.

Explicación

Tres piezas hacen el trabajo:

• Capa WebSocket: mantiene una conexión persistente por cliente, agrupada en rooms.
• Redis pub/sub: es el bus de distribución — cada instancia se suscribe al mismo canal y entrega los mensajes a sus clientes locales.
• Capa de aplicación: publica {room, text} cada vez que ocurre algo que vale la pena notificar — un pedido nuevo, un trabajo terminado, un despliegue.

Redis pub/sub dispara y olvida — un cliente desconectado cuando sale el mensaje nunca lo ve. Cuando la entrega importa, escribe las notificaciones en Redis Streams o en una tabla y repásalas al reconectar.

Variantes

Tecnología	Transporte	Escalado	Caso de uso
WebSockets	TCP full-duplex	Redis pub/sub	Chat, actualizaciones en vivo
Server-Sent Events	HTTP unidireccional	Redis pub/sub	Precios de acciones, logs
Long Polling	Fallback HTTP	No necesario	Navegadores legacy
MQTT	TCP ligero	Clúster de brokers	Dispositivos IoT

Notificaciones dirigidas a usuarios

Los rooms cubren el caso de “todos los que siguen este pedido”. Para notificar a un solo usuario —“tu exportación terminó”— indexa el registro por ID de usuario en lugar de room:

const userConnections = new Map(); // userId -> Set de sockets

// Después de validar el token durante el handshake:
const userId = ws.userId;
if (!userConnections.has(userId)) userConnections.set(userId, new Set());
userConnections.get(userId).add(ws);

// Alcanza todos los dispositivos que el usuario tiene conectados:
function notifyUser(userId, payload) {
  for (const ws of userConnections.get(userId) ?? []) {
    if (ws.readyState === WebSocket.OPEN) {
      ws.send(JSON.stringify(payload));
    }
  }
}

Publica {user, text} a través de Redis cuando el destinatario pueda estar en otra instancia, y expón un endpoint REST como POST /notify/user/:id para que los servicios internos disparen notificaciones.

Lo que funciona

• Heartbeat cada 30 segundos: detecta conexiones muertas y libera memoria antes de que se acumulen.
• Segmentación por rooms o usuarios: entrega al conjunto más pequeño posible de clientes.
• Autenticación en el handshake: valida el JWT durante el upgrade de WebSocket, no en un mensaje posterior.
• Ruta de fallback: sirve SSE o polling cuando el upgrade falla — los proxies corporativos siguen rompiendo WebSockets.
• Límite en broadcasts: aplica rate limit a quién puede publicar, o un productor con bug inunda a todos los clientes conectados.

Errores Comunes

• No manejar reconexiones: las redes tiran sockets constantemente; sin backoff exponencial los clientes quedan muertos.
• Esperar que pub/sub persista: Redis pub/sub descarta mensajes cuando nadie está suscrito — usa Redis Streams para entrega durable.
• Difundir a todos: acota los mensajes por room o usuario; un canal global se convierte en ruido rápido.
• Ignorar límites de conexión: cada socket cuesta memoria y un file descriptor; impón topes por IP y globales.
• Omitir auth en el handshake: un WebSocket sin validación JWT es un canal abierto hacia tu aplicación.

Cuándo No Usar Este Enfoque

• APIs puramente request-response: si nada necesita push, el polling REST es más simple de construir y depurar.
• Equipos sin experiencia en tiempo real: el ciclo de vida de conexiones, el backpressure y la lógica de reconexión son trabajo especializado — el polling perdona más.
• Actualizaciones tolerantes a latencia: si retrasos de 5–10 segundos son aceptables, el polling gana en simplicidad.
• Redes cerradas: los firewalls corporativos siguen matando los upgrades de WebSocket — verifica la ruta hasta producción antes de elegir esto.
• Prototipos de un solo servidor: una instancia no necesita Redis pub/sub — difunde al registro local hasta que escales.

Benchmarks de Rendimiento

Métrica	WebSocket	SSE	REST Polling (5s)
Latencia (entrega de mensaje)	2ms	5ms	2500ms promedio
Conexiones por servidor	10,000	8,000	N/A
Memoria por conexión	4KB	6KB	N/A
Ancho de banda (1000 msg/min)	50KB/min	80KB/min	2.4MB/min
Tiempo de reconexión	100ms	300ms	N/A
CPU por 1000 conexiones	2%	3%	0.5%

Medido en Node.js 20, single core, mensajes de 1KB. Los resultados reales cambian con el tamaño del payload, la tasa de fan-out y las condiciones de red — tómalos como órdenes de magnitud, no como garantías.

Estrategia de Testing

• Ciclo de vida de la conexión: conectar, autenticar, intercambiar mensajes, desconectar limpiamente.
• Reconexión: corta el socket a mitad de stream y confirma que el cliente hace backoff y se reincorpora a su room.
• Fan-out: levanta dos instancias del servidor y verifica que un publish en una llega a los clientes de la otra.
• Backpressure: publica más rápido de lo que el cliente consume y confirma que el servidor descarta carga en lugar de bufferizar sin límite.
• Fallo de autenticación: verifica que los upgrades no autenticados se rechazan antes de que fluya cualquier mensaje.
• Orden: envía 100 mensajes rápido y confirma que llegan en orden por una misma conexión.

Estimación de Costos

• Estado de conexión: 4–6KB por conexión viva — presupuesta unos 40–60MB de RAM para 10K clientes concurrentes, además de la memoria de la aplicación.
• Balanceador de carga: AWS ALB maneja WebSocket nativamente sin costo extra; un NLB con sticky sessions ronda los $15–25/mes en setups pequeños.
• Redis: una instancia gestionada pequeña (clase cache.t4g.micro, unos $10–15/mes) cubre decenas de miles de mensajes pub/sub por segundo.
• Monitorización: el conteo de conexiones, la tasa de fan-out y la latencia de entrega requieren métricas propias; la monitorización alojada ronda los $30–50/mes a esta escala.
• Tiempo de ingeniería: espera alrededor de un 30% más de esfuerzo que una feature REST equivalente — el ciclo de vida, la reconexión y las pruebas de carga cuestan trabajo real.

Monitorización y Observabilidad

• Conexiones concurrentes por instancia: alerta caídas de más del 20% en cinco minutos — ese patrón suele indicar problemas de red o de despliegue.
• Tasa de fan-out: mensajes por segundo por canal; un pico repentino desde un room apunta a un productor desbocado.
• Tasa de reconexión: más de una reconexión por cliente por minuto sugiere sockets inestables o timeouts agresivos del servidor.
• Latencia de entrega: un tiempo de publish-a-recepción por encima de 100ms señala cuello de botella entre Redis y el cliente.
• Fallos de autenticación por IP: los picos sugieren credential stuffing o intentos de replay de tokens.

Checklist de Despliegue

• Cerrar conexiones inactivas tras un timeout razonable (alrededor de 5 minutos).
• Limitar conexiones por instancia para prevenir agotamiento de recursos.
• Enviar frames heartbeat/ping-pong para detectar sockets muertos.
• Habilitar sticky sessions (o fan-out por pub/sub) en el balanceador de carga.
• Ejecutar Redis pub/sub para broadcasting entre instancias.
• Servir solo wss:// en producción — nada de ws:// plano.
• Entregar reconexión de cliente con backoff exponencial.
• Monitorizar conteo de conexiones, tasa de fan-out y latencia de entrega.
• Ensayar failover: mata una instancia y observa cómo los clientes se reconectan a otra.
• Documentar el formato de mensaje en la documentación de la API.

Consideraciones de Seguridad

Autenticación y autorización

• Valida el header Origin: los WebSockets saltan la política same-origin — comprueba el origen contra una allowlist para frenar el cross-site WebSocket hijacking (CSWSH).
• Mantén los tokens fuera de la URL: wss://api.example.com/ws?token=abc filtra credenciales a los access logs; envía la autenticación en un header o en el primer mensaje.
• Revalida los tokens en conexiones abiertas: un socket que sobrevive a su token es acceso no autorizado — desconéctalo cuando el JWT expire.
• Autoriza cada suscripción: unirse a un room como user-123 debería exigir prueba de que el cliente es su dueño.

Denegación de servicio

• Limita las conexiones por IP: los floods de sockets no autenticados agotan la memoria rápido.
• Pon tope al tamaño de mensaje: un límite de 1MB por frame evita que payloads gigantes quemen RAM.
• Acota el buffer de envío: desconecta a los consumidores lentos en lugar de encolar sin límite.
• Aplica rate limit a los frames ping/pong: los floods de pongs no solicitados son un vector de DoS real.

Integridad de mensajes

• Escapa el contenido de usuario en mensajes SSE: un \n\n sin escapar permite falsificar eventos en el cliente.
• Firma los mensajes sensibles con HMAC: un cliente comprometido puede falsificar mensajes que parecen legítimos.
• Añade timestamp y secuencia a los mensajes: sin metadatos de orden, los replays son invisibles.

Transporte

• Sirve solo wss://: ofrecer ws:// invita a ataques de downgrade — redirige o recházalo.
• No confíes ciegamente en X-Forwarded-For: los headers falsificados derrotan los rate limits por IP detrás de un proxy mal configurado.
• Drena las conexiones al apagar: envía un close frame con código de razón para que los clientes se reconecten a una instancia sana en lugar de machacar una muerta.

Resolución de problemas en producción

• Las conexiones caen cada ~60 segundos: un timeout de inactividad del proxy está matando sockets silenciosos — envía heartbeats cada 30s o sube el timeout de nginx/ALB.
• “Connection reset” bajo carga: revisa los límites de file descriptors (ulimit -n) y los topes por IP antes de culpar al código.
• Los mensajes llegan a unos clientes pero no a otros: casi siempre es un bug de membresía del room — un publish disparado antes del join, o un nombre de room distinto entre publicador y suscriptor.
• La memoria crece tras un pico de tráfico: los sockets muertos nunca se quitaron del registro; audita tus handlers de close y disconnect.
• Funciona en local, falla en staging: los proxies eliminan los headers Upgrade — confirma que la ruta permite WebSockets o entrega un fallback a SSE.

Puntos Clave

• Los WebSockets poseen la conexión; Redis pub/sub posee la distribución entre instancias — resuelven problemas distintos.
• Enruta cada mensaje a un room o a un usuario; el broadcast global no escala.
• Pub/sub olvida todo lo publicado mientras un cliente estaba offline — persiste con Redis Streams o una tabla cuando la entrega importa.
• Los heartbeats, la reconexión y la auth en el handshake no son extras opcionales en producción.

Lectura Recomendada

• RFC 6455 — The WebSocket Protocol
• Documentación de Redis Pub/Sub
• Guía de WebSockets de FastAPI
• Referencia de WebSocket de Spring