RedisSocketPubSub

RedisSocketPubSub is the Socket broker for multiple instances: each channel/room pair is a Redis channel (<prefix>socket:<canal>:<sala>, with channel and room encoded using encodeURIComponent), to which each instance subscribes once while it has local peers in that room. A broadcast in any instance reaches peers in all instances.

Delivery is at most once: Redis pub/sub does not store messages, so anything published while an instance is disconnected from Redis is lost for that instance.

Usage#

// 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(); // the subscriber needs its own connection
  await subscriber.connect();
  return {
    socket: { pubsub: new RedisSocketPubSub({ publisher: redis, subscriber, prefix: "chat:prod:" }) },
    close: async () => { await subscriber.quit(); await redis.quit(); },
  };
});

Type#

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>
}

Parameters#

new RedisSocketPubSub(options):

node-redis 4 or later clients implement both interfaces as-is.

Return value#

A SocketPubSub:

Examples#

Redis Cluster#

// publisher provides sPublish; subscriber provides sSubscribe and sUnsubscribe (Redis 7 or later).
const pubsub = new RedisSocketPubSub({ publisher, subscriber, prefix: "chat:prod:", sharded: true });

Observe discarded messages#

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

A message from another format version or malformed (another process publishing to the same channel) is discarded and reported here, without closing any peer.

Deploy with multiple replicas#

# docker-compose.yml (excerpt): two replicas behind the proxy, one shared Redis
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

Sessions without configuration already use that Redis; declare the Socket broker in server/runtime.ts as in “Usage”.

Common mistakes#

A prefix without a trailing :#

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

TypeError: Socket Redis prefix must be nonempty, contain no whitespace, and end with ':', when creating the runtime.

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

sharded with a client lacking sharded pub/sub#

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

TypeError: Redis sharded pub/sub requires sPublish, sSubscribe and sUnsubscribe. Requires node-redis with these methods and Redis 7 or later.

new RedisSocketPubSub({ publisher: redis, subscriber, prefix: "chat:prod:" }); // without sharded outside a cluster

Use the normal connection as subscriber#

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

In subscriber mode, a node-redis connection does not accept other commands: publish (and the session store, if shared) fails on the first broadcast.

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

Notes#

• Validation in each instance. The receiving instance validates the payload against the channel schema again and runs authorize and authorizeSend before sending to each local peer.
• Size. The hub rejects messages larger than its maxMessageBytes (64 KiB by default) before publishing; the broker limit (1 MiB) protects Redis from messages that bypass the hub.
• Security. Anyone with PUBLISH access to Redis can inject messages into a room; the receiver validates them, but restrict access with ACLs and use a prefix per application and environment.
• Workers. Not supported: the Worker rejects any broker other than DurableObjectSocketPubSub.
• Shutdown. The application opens and closes publisher and subscriber.

Related#

• LocalSocketPubSubSocket broadcast broker within a single process, used by the hub by default in Node, Bun and Deno; does not reach other instances.
• RedisSessionStorePersistent Redis session store, with Redis expiration, atomic rotation and revocation, cross-instance revocation and shared rateLimit counters.
• DurableObjectSocketPubSubThe Socket broker for Cloudflare Workers, where each channel and room is a Durable Object that receives publications over RPC and distributes them to its peers.
• defineSocketChannelDeclares a WebSocket channel in server/socket with schemas for each direction, per-room and per-event guards, and handlers for client messages.
• Broadcast and multiple instancesMeeko as a full-stack framework
• WebSocketMeeko as a full-stack framework