> ## Documentation Index
> Fetch the complete documentation index at: https://stars-components.js.org/llms.txt
> Use this file to discover all available pages before exploring further.

# @wolfstar/plugin-broker

> Distribute gateway events across processes over Redis Streams with at-least-once delivery.

```bash npm theme={"system"}
npm install @wolfstar/plugin-broker ioredis
```

```bash pnpm theme={"system"}
pnpm add @wolfstar/plugin-broker ioredis
```

```bash yarn theme={"system"}
yarn add @wolfstar/plugin-broker ioredis
```

```bash bun theme={"system"}
bun add @wolfstar/plugin-broker ioredis
```

> [card] [npm](https://npmx.dev/package/@wolfstar/plugin-broker)

> [card] [Source](https://github.com/wolfstar-project/plugins/tree/main/packages/plugin-broker)

> [!TIP]
>
> See the [Plugins guide](https://stars-components.js.org/documentation/guide/plugins.md) for how plugins from wolfstar-project/plugins fit alongside the core framework.

## Description

`@wolfstar/plugin-broker` publishes and consumes messages over [Redis Streams](https://redis.io/docs/latest/develop/data-types/streams/)
with consumer-group semantics. One gateway process, or several sharded ones with
[`@wolfstar/plugin-sharder`](https://stars-components.js.org/documentation/packages/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

```ts theme={"system"}
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:

```ts theme={"system"}
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`](https://stars-components.js.org/documentation/packages/kit) peer dependency, at `^0.2.0 || ^0.3.0`.

```ts theme={"system"}
// 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`](https://stars-components.js.org/documentation/packages/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`](https://stars-components.js.org/documentation/packages/http-framework#plugins) factory to the
`plugins` client option:

```ts theme={"system"}
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.

> [!WARNING]
>
> 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`](https://stars-components.js.org/documentation/packages/plugin-gateway)
`GatewayClient` receives, once it is written to the client's cache. Workers that share a
[`@wolfstar/plugin-cache`](https://stars-components.js.org/documentation/packages/plugin-cache) Redis cache see the state each dispatch left.

```ts theme={"system"}
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`](https://stars-components.js.org/documentation/packages/plugin-gateway#run-a-listener-on-one-process)
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:

```ts theme={"system"}
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`](https://stars-components.js.org/documentation/packages/plugin-cache) Redis cache.

```ts theme={"system"}
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.

> [!WARNING]
>
> 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.

```ts theme={"system"}
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:

```ts theme={"system"}
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.

> [!WARNING]
>
> Every producer and consumer of a stream must use the same codec.

[![version](https://npmx.dev/api/registry/badge/version/@wolfstar/plugin-broker)](https://npmx.dev/package/@wolfstar/plugin-broker)
[![downloads](https://npmx.dev/api/registry/badge/downloads/@wolfstar/plugin-broker)](https://npmx.dev/package/@wolfstar/plugin-broker)
[![license](https://img.shields.io/github/license/wolfstar-project/plugins?style=flat-square\&color=informational)](https://github.com/wolfstar-project/plugins/blob/main/LICENSE)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.