Redis pub/sub

The bus is in-memory by default, which is exactly right for a single instance. To fan messages out across instances, or to share channels with non-JS producers, swap the engine for Redis once at server start.

On Bun there is nothing to install: it ships a Redis client (docs; pub/sub needs Bun ≥ 1.2.23). On Node any client works, and the ioredis variant is below. Either way you need Redis ≥ 7.2, and the connection URL comes from REDIS_URL (falling back to redis://localhost:6379).

The engine

A subscribed Redis connection can only subscribe, so publishing needs a second connection; hence the two clients. Locally the engine reuses the in-memory bus for fan-out, which keeps iterator semantics (per-subscriber FIFO, cleanup on return) identical to the default engine.

// src/lib/server/redis-pubsub.ts
import type { PubSubEngine, SubscribeOptions } from 'svelte-ws/pubsub';

import { RedisClient } from 'bun';
import { createInMemoryPubSub } from 'svelte-ws/pubsub';

export function createRedisPubSub(url?: string): PubSubEngine {
	const publisher = new RedisClient(url);
	const subscriber = new RedisClient(url); // subscribing takes over this connection
	const local = createInMemoryPubSub();
	const subscribed = new Map<string, number>(); // channel → local subscriber count

	return {
		async publish(channel, payload) {
			await publisher.publish(channel, JSON.stringify(payload));
		},

		subscribe<T>(channel: string, options?: SubscribeOptions) {
			const count = subscribed.get(channel) ?? 0;
			subscribed.set(channel, count + 1);
			if (count === 0) {
				// first local subscriber → subscribe on Redis and relay into the local bus
				subscriber
					.subscribe(channel, (message) => {
						// anything else sharing this Redis may publish non-JSON on the channel
						try {
							local.publish(channel, JSON.parse(message));
						} catch {
							console.warn(`[redis-pubsub] dropped a non-JSON message on ${channel}`);
						}
					})
					.catch((err) => {
						console.error(`[redis-pubsub] subscribe ${channel} failed`, err);
						subscribed.delete(channel); // the next subscriber tries again
					});
			}

			// pass `options` on: it carries the `signal` that ends an idle subscription promptly
			const iterator = local.subscribe<T>(channel, options);
			let released = false;
			const release = () => {
				// `return()` runs twice when a signal ends the subscription: count it once
				if (released) return;
				released = true;
				const left = (subscribed.get(channel) ?? 1) - 1;
				if (left <= 0) {
					subscribed.delete(channel);
					subscriber
						.unsubscribe(channel)
						.catch((err) => console.error(`[redis-pubsub] unsubscribe ${channel} failed`, err));
				} else {
					subscribed.set(channel, left);
				}
			};

			// run the Redis cleanup when the consumer stops iterating
			const originalReturn = iterator.return?.bind(iterator);
			iterator.return = (value?: unknown) => {
				release();
				return originalReturn ? originalReturn(value) : Promise.resolve({ value, done: true });
			};
			return iterator;
		}
	};
}

On the Node adapter

The engine never mentions Bun; only the two client calls do. With ioredis the differences from the code above are mechanical:

import Redis from 'ioredis';

const publisher = new Redis(url ?? process.env.REDIS_URL ?? 'redis://localhost:6379');
const subscriber = new Redis(url ?? process.env.REDIS_URL ?? 'redis://localhost:6379');

// delivery arrives through one listener instead of a per-channel callback — register once
subscriber.on('message', (channel, message) => {
	try {
		local.publish(channel, JSON.parse(message));
	} catch {
		console.warn(`[redis-pubsub] dropped a non-JSON message on ${channel}`);
	}
});

…and the per-channel calls become subscriber.subscribe(channel) (no callback argument) and subscriber.unsubscribe(channel), each with the same .catch: an unhandled rejection there (Redis restarting, say) is fatal to a Node process. Everything else stays the same file: the local fan-out, the reference counting, the cleanup on return.

Wiring it up in socket.config.ts

import { defineSocketServer } from 'svelte-ws/config';
import { setPubSub } from 'svelte-ws/pubsub';
import { createRedisPubSub } from '#lib/server/redis-pubsub.js';

export default defineSocketServer({
	init() {
		setPubSub(createRedisPubSub());
	}
});

Nothing else changes: every stream/mutate, native command/query.live and +server.ts that uses publish/subscribe now talks through Redis. Payloads must be JSON-serialisable (they already had to be to cross the WebSocket).

Notes

• Bun’s Redis pub/sub is channel-based (subscribe). Pattern subscriptions (PSUBSCRIBE) aren’t in the typed API; send raw commands via client.send() if you need them.
• RedisClient auto-reconnects (up to 20 retries) and queues commands while offline.
• Messages published by this instance also come back through Redis, so publishers hear their own messages exactly once, same as with the in-memory engine.