RedisSocketPubSub

RedisSocketPubSub es el broker de Socket para más de una instancia: cada par canal/sala es un canal Redis (<prefix>socket:<canal>:<sala>, con canal y sala codificados con encodeURIComponent), al que cada instancia se suscribe una vez mientras tenga peers locales en esa sala. Un broadcast en cualquier instancia llega a los peers de todas.

La entrega es a lo sumo una vez: Redis pub/sub no guarda mensajes, así que lo publicado mientras una instancia está desconectada de Redis se pierde para ella.

Uso#

// server/runtime.ts
import { createClient } from "redis";
import { defineServerRuntime, RedisSocketPubSub } from "@meekojs/meeko/server";

export default defineServerRuntime(async ({ env }) => {
  const redis = createClient({ url: env.MEEKO_REDIS_URL });
  await redis.connect();
  const subscriber = redis.duplicate(); // el suscriptor necesita su propia conexión
  await subscriber.connect();
  return {
    socket: { pubsub: new RedisSocketPubSub({ publisher: redis, subscriber, prefix: "chat:prod:" }) },
    close: async () => { await subscriber.quit(); await redis.quit(); },
  };
});

Tipo#

class RedisSocketPubSub implements SocketPubSub {
  constructor(options: RedisSocketPubSubOptions);
  channelName(channel: string, room: string): string;
  publish(channel: string, room: string, message: SocketMessage): Promise<void>;
  subscribe(
    channel: string,
    room: string,
    listener: (message: SocketMessage) => void | Promise<void>,
  ): Promise<Unsubscribe>;
}

interface RedisSocketPubSubOptions {
  readonly publisher: RedisPublisherClient;
  readonly subscriber: RedisSubscriberClient;
  readonly prefix: string;
  readonly maxMessageBytes?: number;
  readonly sharded?: boolean;
  readonly onError?: (
    error: unknown,
    context: { readonly channel: string; readonly room: string },
  ) => void;
}

interface RedisPublisherClient {
	publish(channel: string, message: string): Promise<unknown>
	sPublish?(channel: string, message: string): Promise<unknown>
}

interface RedisSubscriberClient {
	subscribe(channel: string, listener: RedisMessageListener): Promise<unknown>
	unsubscribe(channel: string, listener: RedisMessageListener): Promise<unknown>
	sSubscribe?(channel: string, listener: RedisMessageListener): Promise<unknown>
	sUnsubscribe?(channel: string, listener: RedisMessageListener): Promise<unknown>
}

Parámetros#

new RedisSocketPubSub(options):

Los clientes de node-redis 4 o posterior cumplen ambas interfaces tal cual.

Valor devuelto#

Un SocketPubSub:

Ejemplos#

Redis Cluster#

// publisher ofrece sPublish; subscriber, sSubscribe y sUnsubscribe (Redis 7 o posterior).
const pubsub = new RedisSocketPubSub({ publisher, subscriber, prefix: "chat:prod:", sharded: true });

Observar los mensajes descartados#

new RedisSocketPubSub({
  publisher: redis,
  subscriber,
  prefix: "chat:prod:",
  onError: (error, { channel, room }) => logger.warn({ channel, room, error }, "socket message dropped"),
});

Un mensaje de otra versión del formato o mal formado (otro proceso que publica en el mismo canal) se descarta y se informa aquí, sin cerrar a ningún peer.

Despliegue con varias réplicas#

# docker-compose.yml (fragmento): dos réplicas detrás del proxy, un Redis compartido
services:
  app:
    image: tienda:1.7.0
    deploy: { replicas: 2 }
    environment:
      MEEKO_PUBLIC_ORIGIN: https://tienda.example.com
      MEEKO_REDIS_URL: rediss://default:${REDIS_PASSWORD}@redis.internal:6380
      MEEKO_TRUST_PROXY: "1"
      MEEKO_HOST: 0.0.0.0

Las sesiones sin configuración ya usan ese Redis; el broker de Socket se declara en server/runtime.ts como en «Uso».

Errores comunes#

Un prefijo sin : final#

new RedisSocketPubSub({ publisher: redis, subscriber, prefix: "chat:prod" });

TypeError: Socket Redis prefix must be nonempty, contain no whitespace, and end with ':', al crear el runtime.

new RedisSocketPubSub({ publisher: redis, subscriber, prefix: "chat:prod:" });

sharded con un cliente sin pub/sub fragmentado#

new RedisSocketPubSub({ publisher: oldClient, subscriber: oldSubscriber, prefix: "chat:prod:", sharded: true });

TypeError: Redis sharded pub/sub requires sPublish, sSubscribe and sUnsubscribe. Necesita node-redis con esos métodos y Redis 7 o posterior.

new RedisSocketPubSub({ publisher: redis, subscriber, prefix: "chat:prod:" }); // sin sharded fuera de un clúster

La conexión normal como suscriptor#

new RedisSocketPubSub({ publisher: redis, subscriber: redis, prefix: "chat:prod:" });

En modo suscriptor, una conexión de node-redis no admite otros comandos: publish (y, si la comparte, el store de sesiones) falla al primer broadcast.

const subscriber = redis.duplicate();
await subscriber.connect();
new RedisSocketPubSub({ publisher: redis, subscriber, prefix: "chat:prod:" });

Notas#

• Validación en cada instancia. La instancia receptora vuelve a validar el payload con el schema del canal y ejecuta authorize y authorizeSend antes de enviar a cada peer local.
• Tamaño. El hub rechaza antes de publicar un mensaje mayor que su maxMessageBytes (64 KiB por defecto); el límite del broker (1 MiB) protege a Redis de mensajes que no pasan por el hub.
• Seguridad. Cualquiera con PUBLISH sobre el Redis puede inyectar mensajes en una sala; el receptor los valida, pero restrinja el acceso con ACL y use un prefijo por aplicación y entorno.
• Workers. No se admite: el Worker rechaza cualquier broker que no sea DurableObjectSocketPubSub.
• Cierre. La aplicación abre y cierra publisher y subscriber.

Relacionado#

• LocalSocketPubSubBroker de broadcasts de Socket dentro de un solo proceso, el que usa el hub por defecto en Node, Bun y Deno; no alcanza a otras instancias.
• RedisSessionStoreStore de sesiones persistente en Redis, con caducidad de Redis, rotación y revocación atómicas, revocación entre instancias y contadores de rateLimit compartidos.
• DurableObjectSocketPubSubEl broker de Socket de Cloudflare Workers, donde cada canal y sala es un Durable Object que recibe las publicaciones por RPC y las reparte a sus peers.
• defineSocketChannelDeclara un canal WebSocket de server/socket con los schemas de cada dirección, sus guardas por sala y evento y los handlers de los mensajes del cliente.
• Broadcast y varias instanciasMeeko como framework full-stack
• WebSocketMeeko como framework full-stack