@azsiaz/jobs-kit (0.2.0)
Installation
@azsiaz:registry=npm install @azsiaz/jobs-kit@0.2.0"@azsiaz/jobs-kit": "0.2.0"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:
dispatchdispatchManyregisterWorkerrunFlow- 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:
-
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
import { z } from "zod";
import {
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({
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: [
limiters.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",
});
runtime.registerWorker("sendEmail", async ({ input }) => {
return {
messageId: `msg:${input.to}`,
};
});
const handle = await runtime.dispatch("sendEmail", {
tenantId: "t_123",
to: "hello@example.com",
subject: "Hi",
body: "Hello",
});
console.log(handle.status);
console.log(handle.job.id);
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 enqueueddeduped: 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({
key: "sendEmail",
queue: "outbound",
input: z.object({
tenantId: z.string(),
mailboxId: z.string().optional(),
to: z.string().email(),
}),
limits: [
limiters.tenant((input) => ({
tenantId: input.tenantId,
})),
limiters.mailbox.when(
(input) => input.mailboxId !== undefined,
(input) => ({
mailboxId: input.mailboxId,
}),
),
],
});
Rate limiting is enforced at execution time, not dispatch time.
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
Two useful details:
- Dedup selectors infer their input directly from the job
inputschema. - Limiter handles enforce the limiter payload shape, but TypeScript cannot infer the limiter selector input from a sibling
inputschema field inside the samedefineJob({...})object. If you want stronger local checking there, annotate the selector input explicitly.
Example:
limiters.tenant((input: { tenantId: string }) => ({
tenantId: input.tenantId,
}));
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 |