@wolfstar/plugin-broker

npm install @wolfstar/plugin-broker ioredis

pnpm add @wolfstar/plugin-broker ioredis

yarn add @wolfstar/plugin-broker ioredis

bun add @wolfstar/plugin-broker ioredis

npm

Source

See the Plugins guide for how plugins from wolfstar-project/plugins fit alongside the core framework.

Description

@wolfstar/plugin-broker publishes and consumes messages over Redis Streams with consumer-group semantics. One gateway process, or several sharded ones with @wolfstar/plugin-sharder, publishes dispatches. Any number of worker processes consume them and share the load, so a worker can restart, crash, or deploy without dropping the gateway connection. Unlike Redis Pub/Sub, entries survive a worker being down (XADD and XREADGROUP instead of PUBLISH). Unlike a plain XREAD, workers share the load through a consumer group instead of each receiving every message.

Producer

import { createBroker } from '@wolfstar/plugin-broker';
import { Redis } from 'ioredis';

const broker = createBroker({
	redis: new Redis(process.env.REDIS_URL!),
	stream: 'wolfstar:events',
	maxLength: 100_000 // XADD MAXLEN ~, omit for no trimming
});

await broker.publish('messageCreate', payload);

Consumer

Import the register entry point before you create your Client, then configure broker on it. The consumer starts automatically once the client starts listening:

import '@wolfstar/plugin-broker/register';
import { Client } from '@wolfstar/http-framework';
import { Redis } from 'ioredis';

const client = new Client({
	broker: {
		redis: new Redis(process.env.REDIS_URL!),
		stream: 'wolfstar:events',
		group: 'workers',
		consumer: `worker-${process.pid}`,
		batchSize: 10,
		block: 5_000,
		claimIdle: 60_000,
		maxDeliveries: 5,
		deadLetterStream: 'wolfstar:events:dead',
		shutdownSignals: ['SIGTERM', 'SIGINT']
	}
});

Activating with `modules`

On framework 6.1 and later, list the ./module entry in modules in stars.config instead of importing register. It needs the optional @wolfstar/kit peer dependency, at ^0.2.0 || ^0.3.0.

// stars.config.ts
import { defineConfig } from '@wolfstar/http-framework/config';
import '@wolfstar/plugin-broker/module';

export default defineConfig({
	modules: ['@wolfstar/plugin-broker/module'],
	broker: { stream: 'wolfstar:events', group: 'workers' }
});

The module reads its options from the broker key of stars.config. The [specifier, options] tuple of modules still works. Its options are merged over the broker key, so they win. The optional @wolfstar/schema peer dependency types the broker key in defineConfig. TypeScript only loads that type once the module is imported, so import @wolfstar/plugin-broker/module in stars.config.ts. Module options must be JSON-serialisable. redis is a client instance and consumer is per process, so neither goes under the broker key of stars.config. Set both through ClientOptions.broker, which is shallowly merged over the module options. ClientOptions.broker is currently typed as the full BrokerConsumerOptions, so it must still include stream and group. Without the stars CLI, pass the ./plugin definePlugin factory to the plugins client option:

import brokerPlugin from '@wolfstar/plugin-broker/plugin';

const client = new Client({
	plugins: [brokerPlugin({ redis, stream: 'wolfstar:events', group: 'workers', consumer: 'w1' })]
});

Use only one of the module, the ./plugin factory, and register. Combining them installs the plugin’s hooks twice.

Delivery guarantees

An entry is acknowledged once every listener for its event has resolved. A listener that throws leaves the entry pending. A restarted consumer of the same name receives it again. With claimIdle set, whichever consumer claims the entry after it has been idle that long receives it instead.

Idle time counts from an entry’s last delivery. Set claimIdle well above your slowest listener’s run time, or another consumer can claim and handle an entry that is still running.

With maxDeliveries set, an entry delivered more than that many times moves to the dead-letter stream. The dead-letter entry keeps its event and payload fields plus the original id, stream, group, consumer, and deliveries. Without it, a failing entry is retried forever.

Graceful shutdown

container.broker.stop() stops reading and waits for the in-flight entry. Anything left unacknowledged stays pending for the next consumer. With shutdownSignals, the consumer does this on its own when the process receives one of those signals.

Forwarding gateway dispatches

forwardGatewayDispatches publishes every dispatch a @wolfstar/plugin-gateway GatewayClient receives, once it is written to the client’s cache. Workers that share a @wolfstar/plugin-cache Redis cache see the state each dispatch left.

import { createBroker, forwardGatewayDispatches } from '@wolfstar/plugin-broker';
import { GatewayDispatchEvents } from 'discord-api-types/v10';

const broker = createBroker({ redis, stream: 'wolfstar:events', maxLength: 100_000 });
const stopForwarding = forwardGatewayDispatches(gatewayClient, broker, {
	events: [GatewayDispatchEvents.MessageCreate], // omit to forward every dispatch
	onError: (error, payload) => console.error(`Failed to forward ${payload.t}`, error)
});

For a GatewayClient, each entry also carries the previous state of the dispatch’s entity (for example, the cached message a MESSAGE_UPDATE replaces), the shard that received the dispatch, and its gateway sequence number. Forwarding includes the GUILD_DELETE the client emits on dispatch for a cached guild that a READY no longer lists, such as one the bot left while offline. Workers clean up those guilds too.

Run listeners on the workers only

A bot that loads the same pieces on the gateway process and on the workers would handle each dispatch twice. forwardGatewayDispatches marks a GatewayClient as forwarding (client.forwardsDispatches returns true) until you call its stop function. RegisterAsDispatchListener from @wolfstar/plugin-gateway reads that flag on every emit. It runs the listener on the workers while the gateway forwards, and on the gateway client when nothing is forwarded, with no environment variable to read:

import { EventGatewayListener, RegisterAsDispatchListener, type Message } from '@wolfstar/plugin-gateway';

@RegisterAsDispatchListener('messageCreate')
export class ModerationListener extends EventGatewayListener<'messageCreate'> {
	public override run(message: Message) {
		// Runs once per dispatch, on whichever process handles it
	}
}

The decorator throws for the events a worker never receives, such as clientReady and shardReady. With events set, a listener of an event built from a dispatch that is not forwarded runs nowhere. Delivery is at-least-once, so keep these listeners idempotent.

Replaying gateway dispatches on workers

replayGatewayDispatches(consumer, client) replays the dispatches forwardGatewayDispatches publishes on a worker’s own GatewayClient, which never connects to Discord. The worker’s EventGatewayListener pieces receive the same events, with the same structures and old arguments, as if the dispatch happened in-process. Use it when you want to run gateway listeners on workers instead of on the process connected to Discord. Both clients must use the same @wolfstar/plugin-cache Redis cache.

import { BrokerConsumer, createBroker, forwardGatewayDispatches, replayGatewayDispatches } from '@wolfstar/plugin-broker';
import { GatewayClient } from '@wolfstar/plugin-gateway';
import { GatewayDispatchEvents } from 'discord-api-types/v10';

// Gateway process
const gatewayClient = new GatewayClient({ ...options, cache: redisCache });
const broker = createBroker({ redis, stream: 'wolfstar:events' });
forwardGatewayDispatches(gatewayClient, broker);
await gatewayClient.start();

// Worker: same cache, never connected
const worker = new GatewayClient({ ...options, cache: redisCache });
const consumer = new BrokerConsumer({
	redis,
	stream: 'wolfstar:events',
	group: 'workers',
	consumer: 'worker-1'
});
const stopReplaying = replayGatewayDispatches(consumer, worker, {
	events: [GatewayDispatchEvents.MessageUpdate] // omit to replay every type the client handles
});
await consumer.start();

A worker’s messageUpdate listener receives the message as it was before the edit. The gateway process ships that previous state with the dispatch as raw API data, and the worker rebuilds it. Relations of the previous state (author, guild, and so on) resolve from the cache when the worker handles the entry, so they can be newer than the dispatch. raw listeners receive the same payload (op, s, t, d) on both sides. replayGatewayDispatches does not replay READY, INTERACTION_CREATE, or shard lifecycle events (shardReady, shardClose, and so on). A worker’s client.user stays null.

Acknowledgement and ordering

An entry is acknowledged once its listeners have resolved, async ones included. A worker listener that throws or rejects leaves the entry pending for redelivery. A failure applies to the whole entry, not to a single listener. When one of several listeners of an event throws, the redelivery runs the listeners that already succeeded again. A connected GatewayClient would not do this.

Keep worker listeners idempotent, and enqueue slow work instead of awaiting it. A consumer handles one entry at a time, and another worker can claim an entry pending for longer than claimIdle while it still runs.

Ordering holds per consumer, not across the consumer group. Workers in one group take different entries and run them concurrently, so two events of the same guild can replay out of order across workers. The cached relations a listener resolves can also differ from the gateway process’s. The previous state shipped with each entry is a snapshot, so old stays correct.

Entry fields

Stream entries can carry optional state, shard, and sequence fields next to event and payload. Pass them to broker.publish as a third options argument. state is encoded with the configured codec, like the payload. Listeners read them from BrokerMessage.state, BrokerMessage.shard, and BrokerMessage.sequence. Entries without these fields, and consumers that ignore them, work as before.

await broker.publish('MESSAGE_UPDATE', payload, { state: previousMessage, shard: 0, sequence: 42 });

BrokerListener piece

A BrokerListener piece, typed by event name, is loaded from the listeners directory. Augment BrokerEvents in your application to type the events you publish and consume:

import { BrokerListener, RegisterAsBrokerListener, type BrokerMessage } from '@wolfstar/plugin-broker';

declare module '@wolfstar/plugin-broker' {
	interface BrokerEvents {
		messageCreate: { content: string };
	}
}

@RegisterAsBrokerListener('messageCreate')
export class LogMessagesListener extends BrokerListener<'messageCreate'> {
	public override run(payload: { content: string }, message: BrokerMessage) {
		console.log(`[${message.id}] ${payload.content}`);
	}
}

Codec

Payloads use a CacheCodec from @wolfstar/plugin-cache: jsonCodec() by default, or msgpackCodec() from @wolfstar/plugin-cache/msgpack for a denser format.

Every producer and consumer of a stream must use the same codec.

Infrastructure

Shared pieces

Platform helpers

Plugins

npm

Source

Description

Producer

Consumer

Activating with `modules`

Delivery guarantees

Graceful shutdown

Forwarding gateway dispatches

Run listeners on the workers only

Replaying gateway dispatches on workers

Acknowledgement and ordering

Entry fields

BrokerListener piece

Codec

npm

Source

​Description

​Producer

​Consumer

​Activating with modules

​Delivery guarantees

​Graceful shutdown

​Forwarding gateway dispatches

​Run listeners on the workers only

​Replaying gateway dispatches on workers

​Acknowledgement and ordering

​Entry fields

​BrokerListener piece

​Codec

Description

Producer

Consumer

Activating with `modules`

Delivery guarantees

Graceful shutdown

Forwarding gateway dispatches

Run listeners on the workers only

Replaying gateway dispatches on workers

Acknowledgement and ordering

Entry fields

BrokerListener piece

Codec