Queues

Typed background job queues for Seyfert — produce jobs anywhere, process them with decorated classes, in memory or on BullMQ/Redis.

@slipher/queues gives your bot typed background job queues: producers enqueue work from anywhere, processors handle it with decorated classes, and a driver decides where jobs live. Run everything in-memory in the current process, or swap to a BullMQ/Redis driver so jobs survive restarts and spread across workers — the same API either way.

Installation

pnpm add @slipher/queues

Requires Seyfert v5. Install BullMQ only when you want Redis-backed persistent queues:

pnpm add bullmq

Usage

You declare each queue's shape in RegisteredQueues, then work with it from two sides. Producers enqueue jobs with ctx.queues.get(name).add(...), fully typed against the queue's payload. Processors are @Processor() classes with a single @Process() handler, plus optional @OnQueueEvent / @OnWorkerEvent listeners.

Everything flows through one registry, exposed as ctx.queues, client.queues, and the plugin's registry — the same object — so you can produce jobs from a command, an event, or a plain service.

Install the plugin into the client via new Client({ plugins }) and declare your queues:

import { Client, definePlugins } from 'seyfert';
import {
	OnQueueEvent,
	OnWorkerEvent,
	Process,
	Processor,
	type QueueJobOf,
	type QueueRegistration,
	memory,
	queues,
} from '@slipher/queues';

// Discriminated union: the `audio` processor handles several named jobs,
// each with its own payload, switched on the `job` field.
type AudioJob =
	| { job: 'transcode'; fileId: string; format: 'mp3' | 'ogg' }
	| { job: 'concatenate'; fileIds: string[] };

// register the queue's shape so producers and processors are typed
declare module '@slipher/queues' {
	interface RegisteredQueues {
		audio: QueueRegistration<AudioJob, string>;
	}
}

// a processor class bound to the 'audio' queue
@Processor('audio')
class AudioProcessor {
	// the single handler for the queue; dispatch on job.name yourself
	@Process()
	async handle(job: QueueJobOf<'audio'>) {
		switch (job.name) {
			case 'transcode':
				return transcode(job.data.fileId, job.data.format);
			case 'concatenate':
				return concatenate(job.data.fileIds);
		}
	}

	// fires locally when this worker picks up a job
	@OnWorkerEvent('active')
	onActive({ job }: { job: QueueJobOf<'audio'> }) {
		job.snapshot();
	}

	// fires on the queue-wide channel when a job completes
	@OnQueueEvent('completed')
	onCompleted({ job, result }: { job: QueueJobOf<'audio'>; result: string }) {
		job.snapshot();
		void result;
	}
}

// build the plugin with an in-process driver and the processors
const queuesPlugin = queues({
	driver: memory({
		attempts: 3,
		retryDelay: '5s',
		concurrency: 4,
	}),
	processors: [AudioProcessor],
});
const plugins = definePlugins(queuesPlugin);

declare module 'seyfert' {
	interface SeyfertRegistry { plugins: typeof plugins }
}

// export the registry for ctx-less producers
export const registry = queuesPlugin.registry;
// install the plugin into the client
export const client = new Client({
	plugins,
});

Each @Processor() class has exactly one @Process() handler — switch on job.name for named-job queues. There is no framework-level per-name dispatch, so typos are caught by the typed producer surface instead of becoming "Queue process not found" retries at runtime.

Producing Jobs

For named-job queues, produce with the queue name, job name, and payload. The final argument carries per-job options:

// enqueue a named job onto the audio queue
await ctx.queues.get('audio').add('concatenate', {
	fileIds: ['a', 'b'],
});

// enqueue a named job with per-job options as the final argument
await ctx.queues.get('audio').add(
	'transcode',
	{ fileId: 'file-2', format: 'ogg' },
	{ attempts: 5, retryDelay: '10s' },
);

Simple queues do not declare a job discriminant and use add(data, options):

// a simple queue declares no job discriminant
declare module '@slipher/queues' {
	interface RegisteredQueues {
		welcome: QueueRegistration<{ userId: string }>;
	}
}

// enqueue with add(data, options) — no job name needed
await ctx.queues.get('welcome').add({ userId: ctx.author.id });

queue.add(name, payload, options) uses the third argument to disambiguate named jobs. A call like queue.add('send', { delay: '5s' }) is ambiguous — it can mean a string payload plus job options, or a named job whose payload happens to look like job options. Slipher throws a descriptive TypeError instead of guessing; pass queue.add('send', { delay: '5s' }, {}) to force name = 'send', or pass non-string data to add(data, options).

Accessing The Registry Without `ctx`

The registry is the same object everywhere it's exposed: ctx.queues === client.queues === queuesPlugin.registry.

Where	How
Command, component, modal	`ctx.queues`
Event	`entity.client.queues`
Anywhere with the client at hand	`client.queues`
Code with no client	capture the plugin `registry`

Capturing the registry at composition time keeps services independent and avoids circular imports:

// services/media.ts
import { registry } from '../index';

export function scheduleTranscode(fileId: string) {
	// produce a job from a plain service using the captured registry
	return registry.get('audio').add('transcode', { fileId, format: 'mp3' });
}

Avoid importing the exported client from processors or services that are loaded by index.ts; that creates a circular import. Capturing the registry instead keeps the service independent of the client.

Events

@OnWorkerEvent(event) is local to the process that ran the job — use it for local cache updates or process-local instrumentation. @OnQueueEvent(event) is the queue/global channel. With memory() both decorators observe the same single-process queue; with persistent(), queue-level events are prepared for BullMQ QueueEvents.

All listeners receive one object payload: { job }, { job, result }, { job, error }, { job, error, delay }, or {} for idle. Direct queue instances also support on() and once().

Listener errors are isolated — throwing inside a handler is reported through reportListenerError and never changes job state, retry counts, or whether later listeners run:

queues({
	driver: memory({
		// listener errors are isolated and reported here, never changing job state
		reportListenerError(event, error) {
			logger.error('queue listener failed', { event, error });
		},
	}),
});

Drivers

The driver decides where jobs live — your queue and processor code stays the same, so you can swap drivers without touching anything else.

memory() runs jobs in the current process and supports delay, attempts, retryDelay, priority, concurrency, and the lifecycle events added, active, completed, failed, retrying, and idle.

persistent() runs jobs on BullMQ/Redis so they survive restarts and spread across workers:

import { persistent, queues } from '@slipher/queues';

const queuesPlugin = queues({
	// run jobs on BullMQ/Redis so they survive restarts and spread across workers
	driver: persistent({
		connection: { host: '127.0.0.1', port: 6379 },
		prefix: 'slipher',
		defaultJobOptions: { removeOnComplete: true, attempts: 3 },
	}),
	processors: [AudioProcessor],
});

Per-job options are the final add() argument:

// per-job options ride in the final add() argument
await ctx.queues
	.get('audio')
	.add('transcode', { fileId: 'file-1', format: 'mp3' }, { attempts: 3, retryDelay: '5s' });

retryDelay only matters when attempts > 1. If a queue or job sets retryDelay while attempts resolves to 1, Slipher emits a SLIPHER_QUEUE_RETRY_DELAY_NO_RETRIES warning because no retry will be scheduled. Invalid duration strings throw InvalidDurationError, exported from @slipher/queues for instanceof checks.

Shutdown

client.close() tears the queues down — wire it to your process signals:

process.on('SIGTERM', () => {
	// tear the queues down cleanly on shutdown
	void client.close().then(() => process.exit(0));
});