Library
Step Queue
Define BullMQ jobs as a set of fully type-safe steps. Job data accumulates as it moves through the pipeline, and a retried job resumes on the exact step it left off at.
npm install @michaelrwalker/step-queue 1 fetchUser
2 chargeCard
3 sendReceipt
4 archive
job.data
{} status queued
// reference
One job, hover to learn it
Every highlighted token is real Step Queue API — hover or tab to it to watch job.data accumulate step by step.
import { z } from "zod";
import { QueueSystem, NameFactory } from "@michaelrwalker/step-queue";
const checkoutInput = z.object({ accountId: z.string(), amount: z.number() });
const checkout = new QueueSystem(NameFactory({ name: "checkout", env: "prod" }))
.defineInputSchema(checkoutInput)
.defineConnection({ host: "localhost", port: 6379 })
.defineProcessor((processor) =>
processor
.addStep({
name: "reserveFunds",
handler: async (job) => {
const reservationId = await reserve(job.data.accountId, job.data.amount);
return { reservationId };
},
rollbackKeys: ["accountId", "reservationId"],
rollback: async (job) => {
await releaseReservation(job.data.accountId, job.data.reservationId);
},
})
.addStep({
name: "chargeCard",
handler: async (job) => {
const chargeId = await charge(job.data.accountId, job.data.amount);
return { chargeId };
},
clean: ["amount"],
})
.addFinalStep({
handler: async (job) => ({
receiptId: job.data.chargeId,
reservation: job.data.reservationId,
}),
}),
);
const job = await checkout.addJob({ accountId: "acct_1", amount: 4200 });
const result = await job.waitUntilFinished(checkout.QueueEvents, 10_000);