@azsiaz/jobs-kit (1.0.0)

Published 2026-05-17 16:18:17 +02:00 by AzSiAz in AzSiAz/jobs-kit

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@1.0.0

or add it to the package.json file:

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

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

@azsiaz/jobs-kit

Typed BullMQ job runtime for:

Standard Schema-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:

npm install @azsiaz/jobs-kit

Job, result, and limiter schemas use the Standard Schema contract. Install whichever compatible validation library you want to author schemas with, for example:

npm install 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
upsertScheduler, getScheduler, listSchedulers, removeScheduler
dashboard queue and job reads

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

Testing Without Redis

Use the fake runtime from the testing subpath for unit tests that only need to assert dispatches or drive worker handlers deterministically:

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

const runtime = createFakeRuntime({ registry });

await runtime.dispatch("sendEmail", {
	tenantId: "t_123",
	to: "hello@example.com",
	subject: "Hi",
	body: "Hello",
});

expect(runtime.testing.dispatched("sendEmail")).toMatchObject([{
	input: {
		tenantId: "t_123",
		to: "hello@example.com",
	},
	status: "created",
}]);

Registered workers run only when the test drains the fake queue:

runtime.registerWorker("sendEmail", async ({ input }) => ({
	messageId: `msg:${input.to}`,
}));

await runtime.testing.drain({ key: "sendEmail" });

expect(runtime.testing.queueState("outbound").completed).toHaveLength(1);

0.4 Migration Notes

0.4.0 removes BullMQ worker and job option types from the default public runtime surface. createBullmqRuntime() still uses BullMQ internally, but application code should type against Runtime<TRegistry>, WorkerRuntimeOptions, and WorkerHandle.

Worker options moved from BullMQ's WorkerOptions shape to a runtime-agnostic shape:

runtime.registerWorker("sendEmail", handler, {
	concurrency: 4,
	limiter: { max: 10, durationMs: 1000 },
	metrics: { maxDataPoints: MetricsRetention.WEEKS(1) },
	bullmq: {
		maxStalledCount: 2,
	},
});

limiter.duration is now limiter.durationMs. BullMQ metrics constants such as MetricsTime.ONE_WEEK can be replaced with MetricsRetention.WEEKS(1), which returns a plain minute count.

registerWorker() now returns WorkerHandle instead of BullMQ's Worker. Use pause(), resume(), isPaused(), close(), and on("completed" | "failed" | "stalled" | "progress", handler) for runtime-level worker control. BullmqRuntime<TRegistry> and RegisterWorkerOptions remain as deprecated aliases for one release.

Environment Variables

jobs-kit supports these environment variables:

Variable	Scope	Values	Default	Description
`JOBS_KIT_SCHEDULER_ALLOW_FOREIGN_OVERWRITE`	Runtime	`true`, `1`, `yes`, `on`, `false`, `0`, `no`, `off`	`false`	Allows `upsertScheduler` to replace an existing BullMQ scheduler that lacks jobs-kit ownership metadata. The `createBullmqRuntime({ scheduler: { allowForeignSchedulerOverwrite } })` option takes priority over this environment variable.
`JOB_KIT_REDIS_URL`	Local tests	Redis URL, for example `redis://127.0.0.1:6379`	unset	Enables Redis-backed integration tests in this repository. When unset, those tests are skipped.

Core Concepts

There are three separate concepts in the API:

key
- stable job type identifier
dedup
- dispatch-time duplicate suppression for a single job type
limits
- execution-time throttling rules that may be reused across jobs

key is not a dedup key.

Quick Start

This example uses Zod, but any Standard Schema-compatible library can be used.

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

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

const sendEmail = defineJob({
	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(),
	}),
	defaults: {
		attempts: 3,
		backoff: { type: "fixed", delay: 1000 },
		removeOnComplete: { count: 1000 },
		removeOnFail: { count: 1000 },
		keepLogs: 100,
	},
	dedup: {
		mode: "unique",
		select: (input) => ({
			tenantId: input.tenantId,
			to: input.to,
			subject: input.subject,
		}),
	},
	limits: ({ use }) => [
		use(tenant).select((input) => ({
			tenantId: input.tenantId,
		})),
	],
});

const registry = createRegistry({
	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",
}, {
	priority: 2,
});

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

For batches with per-job names or options, pass { input, options } items:

await runtime.dispatchMany("sendEmail", [
	{ input: verificationEmail, options: { name: "email.verification" } },
	{ input: resetEmail, options: { name: "email.reset" } },
]);

Dispatch Defaults

Jobs can declare default dispatch options next to their schema and queue. These defaults apply to dispatch(), dispatchMany(), runFlow(), and scheduler template jobs created with upsertScheduler().

const sendEmail = defineJob({
	key: "sendEmail",
	queue: "outbound",
	input: z.object({
		to: z.string().email(),
	}),
	defaults: {
		attempts: 3,
		backoff: { type: "exponential", delay: 5000 },
		removeOnComplete: { age: 60 * 60 * 24 },
		removeOnFail: { age: 60 * 60 * 24 * 7 },
	},
});

Precedence is:

per-call dispatch, flow step, or scheduler template options
job.defaults
createBullmqRuntime({ defaultJobOptions })

Job defaults and runtime-wide defaultJobOptions support the dispatch fields that are safe to share across every enqueue path: attempts, priority, backoff, removeOnComplete, removeOnFail, and keepLogs. Per-call options still own fields like name, jobId, delay, and dashboard metadata.

Dedup

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

const syncTenant = defineJob({
	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

A keyed limiter throttles jobs by a value derived from the parsed job input. Use it when the limit is not just "N jobs per worker", but "N jobs per tenant", "N jobs per mailbox", "N jobs per source", or another domain key.

There are three pieces in a limiter binding:

use(limiter) chooses the limiter policy.
.select(input => key) maps the job input to the limiter key.
.when(input => condition) optionally skips that limiter for this job execution.

Rate limiting is enforced at execution time, not dispatch time. Persisted job data may contain stale limiter metadata, but workers recompute limits from the current job definition before enforcing them.

Shared Limiters

Use defineLimiter() when multiple jobs should share the same rate-limit bucket. Jobs using the same limiter key and selected value are throttled together.

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

const mailbox = defineLimiter({
	key: "mailbox",
	windowMs: 10_000,
	max: 5,
	schema: z.object({ mailboxId: z.string() }),
});

Bind a shared limiter with use(limiter).select(...). Use this form when the limiter should always apply:

const sendEmail = defineJob({
	key: "sendEmail",
	queue: "outbound",
	input: z.object({
		tenantId: z.string(),
		mailboxId: z.string().optional(),
		to: z.string().email(),
	}),
	limits: ({ use }) => [
		use(tenant).select((input) => ({
			tenantId: input.tenantId,
		})),
	],
});

In this example, all sendEmail jobs for the same tenantId share the tenant bucket. If another job also uses tenant and selects the same { tenantId }, it shares that same bucket too.

Conditional Limiters

Use .when(...).select(...) when a limiter only applies for some inputs. The when callback decides whether the limiter is active; select still defines the key to throttle when it is active.

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

Here, the tenant limiter always applies. The mailbox limiter applies only when mailboxId is present.

Inline Limiters

Use an inline limiter spec when the policy is local to one job and there is no reason to name or import a shared limiter handle.

const syncSource = defineJob({
	key: "syncSource",
	queue: "sync",
	input: z.object({
		sourceId: z.string(),
	}),
	limits: ({ use }) => [
		use({
			key: "source",
			windowMs: 60_000,
			max: 10,
			schema: z.object({ sourceId: z.string() }),
		}).select((input) => ({
			sourceId: input.sourceId,
		})),
	],
});

Inline limiters still need a stable key, because the key becomes part of the Redis bucket and dashboard metadata. Prefer a shared defineLimiter() when another job must intentionally share that exact bucket.

Worker and Dispatch Options

registerWorker() accepts runtime-agnostic worker options. BullMQ-specific worker settings live under the explicit bullmq extension bag, while connection and prefix are always owned by the runtime.

Use worker options for worker process behavior:

const worker = runtime.registerWorker("sendEmail", async ({ input }) => {
	return {
		messageId: `msg:${input.to}`,
	};
}, {
	concurrency: 8,
	limiter: { max: 10, durationMs: 1000 },
	metrics: { maxDataPoints: 60 * 24 * 7 },
	bullmq: {
		maxStalledCount: 2,
	},
});

worker.on("failed", (event) => {
	console.error(event.jobId, event.reason);
});

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 },
});

Pass name when one job key should be enqueued under multiple BullMQ job names in the same queue. Treat it as an operational label for BullMQ tooling, not as the worker's type-safe business discriminator.

const sendEmail = defineJob({
	key: "sendEmail",
	queue: "outbound",
	input: z.discriminatedUnion("kind", [
		z.object({
			kind: z.literal("password-reset"),
			tenantId: z.string(),
			to: z.string().email(),
		}),
		z.object({
			kind: z.literal("email-verification"),
			tenantId: z.string(),
			to: z.string().email(),
		}),
	]),
});

await runtime.dispatch("sendEmail", {
	kind: "password-reset",
	tenantId: "t_123",
	to: "user@example.com",
}, {
	name: "email.password-reset",
});

runtime.registerWorker("sendEmail", async ({ input }) => {
	switch (input.kind) {
		case "password-reset":
			return sendPasswordReset(input);
		case "email-verification":
			return sendVerification(input);
	}
});

Workers also receive a narrow execution control API for the common BullMQ operations that handlers need at runtime:

runtime.registerWorker("syncSource", async ({ input, job, progress, control, log }) => {
	await log.info("Starting sync", {
		attempt: job.attemptsMade + 1,
		priority: job.priority,
	});

	await progress.set({ stage: "fetching", percent: 10 });

	try {
		await fetchRemote(input.sourceId);
	} catch (error) {
		if (error instanceof RemoteRateLimitError) {
			await control.updateInput((current) => ({
				...current,
				rateLimitRetryAttempt: (current.rateLimitRetryAttempt ?? 0) + 1,
			}));

			await control.delayFor(error.retryAfterMs);
		}

		throw error;
	}

	await progress.set(100);
});

control.delayFor(ms) and control.delayUntil(timestampMs) end the current execution attempt after moving the job back to BullMQ's delayed set. They do not expose BullMQ tokens or require throwing BullMQ errors from application code.

Schedulers

Use runtime schedulers to create or update recurring jobs without importing BullMQ queues in application code.

await runtime.upsertScheduler("fetch-latest", "sendEmail", {
	tenantId: "t_123",
	to: "ops@example.com",
	subject: "Digest",
	body: "Latest activity",
}, {
	pattern: "*/15 * * * *",
	name: "digest-email",
	tz: "Europe/Paris",
	attempts: 3,
	keepLogs: 100,
});

Fixed intervals use milliseconds:

await runtime.upsertScheduler("refresh-cache", "sendEmail", {
	tenantId: "t_123",
	to: "ops@example.com",
	subject: "Refresh",
	body: "Cache refresh",
}, {
	every: 60_000,
	removeOnComplete: { count: 100 },
});

Schedulers use the target job definition for queue and input validation. Their BullMQ job name defaults to the job definition name or key, and can be overridden with name in the scheduler options. The scheduler id is stable within the resolved BullMQ queue; keep ids unique for jobs that share a queue. upsertScheduler replaces an existing scheduler that has the same id and job key, and throws if the id is already bound to a different job key on the same queue. It also throws when a scheduler with the same id exists without jobs-kit ownership metadata. To deliberately replace a scheduler created directly through BullMQ or another library, opt in at runtime:

const runtime = createBullmqRuntime({
	registry,
	connection,
	scheduler: { allowForeignSchedulerOverwrite: true },
});

If the runtime option is omitted, JOBS_KIT_SCHEDULER_ALLOW_FOREIGN_OVERWRITE=true enables the same behavior. The JavaScript option takes priority over the environment variable, and the default is false.

upsertScheduler and removeScheduler serialize through a short-lived Redis lock keyed on (queue, id), so concurrent writers to the same scheduler fail fast rather than racing; a writer that encounters the lock throws a RegistryError. To opt into bounded automatic retries instead, pass schedulerLock when creating the runtime:

const runtime = createBullmqRuntime({
	registry,
	connection,
	schedulerLock: { retries: 5, delayMs: 100 },
});

retries is the number of additional attempts after the first (default 0 — fail fast); delayMs is the fixed delay between attempts (default 50); ttlMs is the Redis lock TTL (default 5000). Raise ttlMs if a single upsertScheduler or removeScheduler might take longer than the default under Redis pressure or slow connection setup — a lock that expires mid-operation allows overlapping writes.

Only schedulers created through upsertScheduler are visible to getScheduler, listSchedulers, and removeScheduler. Schedulers created directly against a BullMQ queue (or left over from another library) lack the jobs-kit template marker and will appear as not-found; clean them up with BullMQ's own API unless you deliberately opt into replacing them.

const current = await runtime.getScheduler("fetch-latest", "sendEmail");
const schedulers = await runtime.listSchedulers({ key: "sendEmail" });

await runtime.removeScheduler("refresh-cache", "sendEmail");

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,
	createBullmqRuntime,
} from "@azsiaz/jobs-kit/bullmq";

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";
import {
	createBullmqJobLogger,
	createBullmqRuntime,
} from "@azsiaz/jobs-kit/bullmq";

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.

Workers that dispatch follow-up jobs or inspect runtime state can close over the constructed runtime. Define the runtime first, then register handlers:

const runtime = createBullmqRuntime({ registry, connection });

runtime.registerWorker("sendEmail", async ({ input }) => {
	if (input.sendReceipt) {
		await runtime.dispatch("sendReceipt", { messageId: input.messageId });
	}
});

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",
		}, undefined, { name: "projection-finished-email" }),
	],
	{ name: "projection-rebuild" },
);

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 latestJobs = await dashboard.listJobs({ queue: "outbound", state: "latest", start: 0, end: 49 });
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",
});
const schedulers = await dashboard.listSchedulers("outbound");
const scheduler = await dashboard.getScheduler({
	queue: "outbound",
	id: "daily-summary",
});

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

Queue summaries expose separate counts for waiting, prioritized, waitingChildren, active, completed, failed, delayed, and paused.

Use state: "latest" to get a timestamp-sorted page merged across the dashboard-visible states.

Queue metrics expose BullMQ's per-minute completed/failed throughput without requiring application dashboards to hold raw queue or Redis references:

const completed = await dashboard.getQueueMetrics("outbound", {
	type: "completed",
	start: 0,
	end: 59,
	asc: true,
});

renderChart({
	series: completed.data,
	pending: completed.pending,
	total: completed.count,
});

data is newest-first by default, matching BullMQ. Pass asc: true for chronological order. pending is the number of jobs counted since BullMQ's previous metrics flush, clamped to zero, and prevTimestamp is the Unix millisecond timestamp for that flush or null when metrics have not started. Workers must opt in with registerWorker(..., { metrics: { maxDataPoints } }); queues without enabled metrics return an empty zero-valued series.

The normalized dashboard job shape includes:

queue/name/state
input data
return value
BullMQ job options, delay, computed scheduled timestamp, and stack trace
dedup metadata
parent job reference
scheduler back-reference for jobs produced by a scheduler

TypeScript Notes

Five useful details:

Dispatch, schedulers, flow steps, and persisted envelopes use the schema input type.
Workers, dedup selectors, and limiter selectors use the parsed schema output type.
Limiter callbacks inside limits: ({ use }) => [...] infer input from the job schema output type.
Limiter selector output is checked against the limiter key schema input at dispatch time and recomputed from parsed job input at worker execution time.
Job input schemas must validate without transforming the persisted representation. Use refinements, enums, unions, and discriminated unions freely, but keep transforms/coercions/preprocess logic inside the worker.

Example:

const sendEmail = defineJob({
	key: "sendEmail",
	queue: "outbound",
	input: z.object({
		tenantId: z.string(),
		mailboxId: z.string().optional(),
	}),
	limits: ({ use }) => [
		use(tenant).select((input) => ({
			tenantId: input.tenantId,
		})),
		use(mailbox).when((input) => input.mailboxId !== undefined).select((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

ID	Version
@standard-schema/spec	^1.1.0
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

Details

npm

AzSiAz/jobs-kit

2026-05-17 16:18:17 +02:00

MIT

59 KiB

Assets (1)

jobs-kit-1.0.0.tgz 59 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

@azsiaz/jobs-kit (1.0.0)

Installation

About this package

@azsiaz/jobs-kit

Installation

Redis Requirement

Testing Without Redis

0.4 Migration Notes

Environment Variables

Core Concepts

Quick Start

Dispatch Defaults

Dedup

Keyed Rate Limiting

Shared Limiters

Conditional Limiters

Inline Limiters

Worker and Dispatch Options

Schedulers

Worker Logging

Flows

Dashboard Client

TypeScript Notes

Minimal Runtime Smoke Test Without Redis

Verification Status

Local Development

Dependencies