Stephan Deumier

Repositories

7

Projects Packages

25

Public activity Starred repositories

@azsiaz/jobs-kit (0.3.0)

Published 2026-04-20 11:24:32 +02:00 by AzSiAz in AzSiAz/jobs-kit

Installation

Setup this registry in your project .npmrc file:

@azsiaz:registry=

To install the package using npm, run the following command:

npm install @azsiaz/jobs-kit@0.3.0

or add it to the package.json file:

"@azsiaz/jobs-kit": "0.3.0"

For more information on the npm registry, see the documentation.

About this package

@azsiaz/jobs-kit

Typed BullMQ job runtime for:

• Zod-validated job definitions
• typed dispatch and worker registration
• job-local deduplication
• keyed Redis-backed rate limiting
• BullMQ flows with typed step builders
• normalized dashboard queries over queues, jobs, and flow trees

Installation

Configure the scoped registry in your project .npmrc:

@azsiaz:registry=https://forge.netserv.fr/api/packages/AzSiAz/npm/

Install the package and its peer dependency:

npm install @azsiaz/jobs-kit zod

If you use TypeScript in a Node project, you will usually also want:

npm install -D typescript @types/node

Redis Requirement

createBullmqRuntime() can be instantiated without an active Redis server if you use a lazy connection for smoke testing, but real queue operations require Redis:

• dispatch
• dispatchMany
• registerWorker
• runFlow
• dashboard queue and job reads

For actual usage, point the runtime at a reachable Redis instance.

Core Concepts

There are three separate concepts in the API:

1. key

   • stable job type identifier

2. dedup

   • dispatch-time duplicate suppression for a single job type

3. limits

   • execution-time throttling rules that may be reused across jobs

key is not a dedup key.

Quick Start

import { z } from "zod";
import {
	createBullmqJobLogger,
	createBullmqRuntime,
	createRegistry,
	defineJob,
	defineLimiter,
	defineLimiters,
} from "@azsiaz/jobs-kit";

const limiters = defineLimiters({
	tenant: defineLimiter({
		windowMs: 60_000,
		max: 30,
		key: z.object({
			tenantId: z.string(),
		}),
	}),
});

const sendEmail = defineJob(
	{ limiters },
	{
		key: "sendEmail",
		queue: "outbound",
		input: z.object({
			tenantId: z.string(),
			to: z.string().email(),
			subject: z.string(),
			body: z.string(),
		}),
		result: z.object({
			messageId: z.string(),
		}),
		dedup: {
			mode: "unique",
			select: (input) => ({
				tenantId: input.tenantId,
				to: input.to,
				subject: input.subject,
			}),
		},
		limits: ({ L }) => [
			L.tenant((input) => ({
				tenantId: input.tenantId,
			})),
		],
	},
);

const registry = createRegistry({
	limiters,
	jobs: {
		sendEmail,
	},
});

const runtime = createBullmqRuntime({
	registry,
	connection: { host: "127.0.0.1", port: 6379, maxRetriesPerRequest: null },
	prefix: "app",
	loggers: [
		createBullmqJobLogger(),
	],
});

runtime.registerWorker("sendEmail", async ({ input, log }) => {
	await log.info("Sending email", {
		tenantId: input.tenantId,
		to: input.to,
	});

	return {
		messageId: `msg:${input.to}`,
	};
}, {
	concurrency: 4,
});

const handle = await runtime.dispatch("sendEmail", {
	tenantId: "t_123",
	to: "hello@example.com",
	subject: "Hi",
	body: "Hello",
}, {
	attempts: 3,
	backoff: { type: "fixed", delay: 1000 },
	keepLogs: 100,
});

console.log(handle.status);
console.log(handle.job.id);

Dedup

Dedup is defined per job and is evaluated at dispatch time.

const syncTenant = defineJob(
	{ limiters },
	{
		key: "syncTenant",
		queue: "sync",
		input: z.object({
			tenantId: z.string(),
			full: z.boolean(),
		}),
		dedup: {
			mode: "unique",
			select: (input) => ({
				tenantId: input.tenantId,
				full: input.full,
			}),
		},
	},
);

For mode: "unique", repeated dispatches of the same work return the same job identity.

dispatch() always returns a normalized handle:

type DispatchHandle = {
	job: {
		id: string;
		key: string;
		name: string;
		queue: string;
	};
	status: "created" | "deduped";
};

Meaning:

• created: a new job was enqueued
• deduped: an equivalent existing job was reused

Keyed Rate Limiting

Limiters are shared policy handles created with defineLimiters().

const limiters = defineLimiters({
	tenant: defineLimiter({
		windowMs: 60_000,
		max: 30,
		key: z.object({ tenantId: z.string() }),
	}),
	mailbox: defineLimiter({
		windowMs: 10_000,
		max: 5,
		key: z.object({ mailboxId: z.string() }),
	}),
});

Jobs can bind one or more limiter rules:

const sendEmail = defineJob(
	{ limiters },
	{
		key: "sendEmail",
		queue: "outbound",
		input: z.object({
			tenantId: z.string(),
			mailboxId: z.string().optional(),
			to: z.string().email(),
		}),
		limits: ({ L }) => [
			L.tenant((input) => ({
				tenantId: input.tenantId,
			})),
			L.mailbox.when(
				(input) => input.mailboxId !== undefined,
				(input) => ({
					mailboxId: input.mailboxId,
				}),
			),
		],
	},
);

Rate limiting is enforced at execution time, not dispatch time.

Worker and Dispatch Options

registerWorker() accepts BullMQ worker options except for connection and prefix, which are owned by the runtime.

Use worker options for worker process behavior:

runtime.registerWorker("sendEmail", async ({ input }) => {
	return {
		messageId: `msg:${input.to}`,
	};
}, {
	concurrency: 8,
	maxStalledCount: 2,
	maxStartedAttempts: 3,
});

Use dispatch or flow step options for job retry behavior and enqueue-time options:

await runtime.dispatch("sendEmail", {
	tenantId: "t_123",
	to: "hello@example.com",
	subject: "Hi",
	body: "Hello",
}, {
	attempts: 3,
	backoff: { type: "fixed", delay: 1000 },
	keepLogs: 100,
	removeOnComplete: { count: 1000 },
	removeOnFail: { count: 1000 },
});

Worker Logging

Workers receive a log helper. Logging is adapter-based, so you can choose where log events go.

Use BullMQ Redis job logs:

import { createBullmqJobLogger } from "@azsiaz/jobs-kit";

const runtime = createBullmqRuntime({
	registry,
	connection,
	loggers: [
		createBullmqJobLogger({
			format: "text",
		}),
	],
});

Add structured logging for Pino, Winston, Loki, or any other sink:

import { createJobLogger } from "@azsiaz/jobs-kit";

const runtime = createBullmqRuntime({
	registry,
	connection,
	loggers: [
		createBullmqJobLogger({ format: "json" }),
		createJobLogger(event => {
			pinoLogger[event.level](event, event.message);
		}, { name: "pino" }),
	],
});

Use the logger in workers:

runtime.registerWorker("sendEmail", async ({ input, log }) => {
	await log.info("Sending email", {
		tenantId: input.tenantId,
		to: input.to,
	});

	return { messageId: "msg_123" };
});

BullMQ Redis log retention is controlled by the job option keepLogs:

await runtime.dispatch("sendEmail", input, {
	keepLogs: 100,
});

Read Redis job logs through the dashboard client:

const logs = await runtime.dashboard().getJobLogs({
	queue: "outbound",
	id: handle.job.id,
});

Logger adapter failures are ignored by default. Use loggerFailureMode: "fail-job" if logging failures should fail the job.

Flows

Use runtime.flow.step() to build typed BullMQ flows without writing BullMQ flow objects directly.

const flow = runtime.flow.step(
	"rebuildProjection",
	{
		projection: "users",
	},
	[
		runtime.flow.step("sendEmail", {
			tenantId: "t_123",
			to: "ops@example.com",
			subject: "Projection finished",
			body: "Done",
		}),
	],
);

const flowHandle = await runtime.runFlow(flow);
console.log(flowHandle.root.id);

You can also use:

• runtime.flow.chain(...)
• runtime.flow.parallel(...)

Dashboard Client

The runtime exposes a normalized dashboard client:

const dashboard = runtime.dashboard();

const queues = await dashboard.listQueues();
const summary = await dashboard.getQueueSummary("outbound");
const jobs = await dashboard.listJobs({ queue: "outbound", state: "waiting" });
const job = await dashboard.getJob({ queue: "outbound", id: "123" });
const state = await dashboard.getJobState({ queue: "outbound", id: "123" });
const flowTree = await dashboard.getFlowTree({
	queue: "flow-parent",
	id: "123",
});

await dashboard.pauseQueue("outbound");
await dashboard.resumeQueue("outbound");
await dashboard.retryJob({ queue: "outbound", id: "123" });
await dashboard.removeJob({ queue: "outbound", id: "123" });
await dashboard.promoteJob({ queue: "outbound", id: "123" });

The normalized dashboard job shape includes:

• queue/name/state
• input data
• return value
• dedup metadata
• limiter metadata
• parent job reference

TypeScript Notes

Three useful details:

1. Dedup selectors infer their input directly from the job input schema.
2. Limiter selectors infer their input from the job input schema when you use defineJob({ limiters }, spec).
3. Limiter selector output is checked against the limiter key schema.

Example:

const sendEmail = defineJob(
	{ limiters },
	{
		key: "sendEmail",
		queue: "outbound",
		input: z.object({
			tenantId: z.string(),
			mailboxId: z.string().optional(),
		}),
		limits: ({ L }) => [
			L.tenant((input) => ({
				tenantId: input.tenantId,
			})),
			L.mailbox.when(
				(input) => input.mailboxId !== undefined,
				(input) => ({
					mailboxId: input.mailboxId,
				}),
			),
		],
	},
);

Minimal Runtime Smoke Test Without Redis

For a construction-only smoke test, you can instantiate the runtime with a lazy connection and close it immediately:

const runtime = createBullmqRuntime({
	registry,
	connection: {
		host: "127.0.0.1",
		port: 6379,
		lazyConnect: true,
		maxRetriesPerRequest: null,
	},
});

await runtime.close();

This does not verify real queue behavior. It only verifies that the published package can be imported and instantiated.

Verification Status

The published package has been verified from a fresh consumer project with:

• install from the private registry
• import and strict TypeScript typecheck
• real Redis-backed runtime execution
• dedup behavior
• flow execution
• dashboard reads
• stored Redis envelope metadata inspection

Local Development

Run the full verification suite:

npm run check

Redis integration tests are present and can be enabled by setting:

JOB_KIT_REDIS_URL=redis://127.0.0.1:6379

Dependencies

Dependencies

ID	Version
bullmq	^5.74.1
ioredis	^5.8.2

Development dependencies

ID	Version
@types/node	^24.10.1
tsd	^0.33.0
typescript	^5.9.3
vitest	^4.0.15
zod	^4.3.6

Peer dependencies

ID	Version
zod	^4.3.6

Details

npm

AzSiAz/jobs-kit

2026-04-20 11:24:32 +02:00

3

MIT

28 KiB

Assets (1)

jobs-kit-0.3.0.tgz 28 KiB

Versions (7) View all

3.0.0 2026-05-28

2.0.1 2026-05-27

2.0.0 2026-05-17

1.0.0 2026-05-17

0.3.0 2026-04-20

Issues