@sdxc/jobs

Declared background jobs dispatched over a pluggable queue backend.

A job map declares every job by name: the payload it carries, the cron it runs on and any meta its callers read. A dispatcher maps a lazily loaded handler onto each one and runs it inside a middleware chain, a timeout and one log record per run. A cron trigger only enqueues, so a scheduled job gets the same middleware, retries and dead-letter handling as any other.

Installation

npm add @sdxc/jobs

Payload schemas are any Standard Schema, such as remix/data-schema from remix. The optional peers are @cloudflare/workers-types for ./cloudflare, remix for the context keys and ./router, @sdxc/cli (with remix) for ./cli, and vitest for ./conformance.

Usage

Declare the jobs

import { job, jobs } from "@sdxc/jobs";
import * as s from "remix/data-schema";

export default jobs({
	cleanup: job({ cron: "0 0 * * *" }),
	sendReceipt: job({ input: s.object({ orderId: s.string() }) }),
	digests: {
		daily: job({ cron: "0 8 * * *" }),
		weekly: job({ cron: "0 9 * * 1" }),
	},
});

A job's name is its key, dot-joined when nested (digests.daily), and messages are addressed by it, so renaming a key strands the messages already in flight. The map holds no handlers, so importing it costs only its schemas.

Write a handler

import { createJobHandler } from "@sdxc/jobs";

import jobs from "./jobs.js";

export default createJobHandler(jobs.sendReceipt, async (ctx) => {
	let order = await ctx.log.time("db", () => findOrder(ctx.input.orderId));
	ctx.log.set({ order: { id: order.id, total: order.total } });
	await sendReceipt(order, { signal: ctx.signal });
});

ctx.input is typed by the job's schema. ctx.log is the run's record, emitted once when it ends: set() queryable fields, note() breadcrumbs, time() durations.

Map handlers onto a dispatcher

import { createJobDispatcher } from "@sdxc/jobs";
import * as cloudflare from "@sdxc/jobs/cloudflare";
import { createLogger } from "@sdxc/logger";
import { env } from "cloudflare:workers";

import jobs from "./jobs.js";

export const dispatcher = createJobDispatcher({
	logger: createLogger({ service: "worker" }),
	queue: cloudflare.queue(() => env.QUEUE),
	timeout: "5 minutes",
});

dispatcher.map(jobs.cleanup, () => import("./jobs/cleanup.js"));
dispatcher.map(jobs.sendReceipt, () => import("./jobs/send-receipt.js"));

export default { ...cloudflare.worker(dispatcher) } satisfies ExportedHandler<Env>;

A handler module loads once per isolate, only after a message for it has arrived and parsed. worker() turns each cron trigger into dispatcher.tick() and each queue batch into dispatcher.deliverBatch(). Spell every cron exactly as its trigger in the Worker's config.

Enqueue a job

await dispatcher.enqueue(jobs.sendReceipt, { orderId: order.id });
await dispatcher.enqueue(jobs.cleanup);
await dispatcher.enqueueMany(
	jobs.sendReceipt,
	orders.map((order) => ({ orderId: order.id })),
);

Each call takes exactly what the job's schema accepts, and no argument for a job without one. A route handler enqueues through ctx.jobs from ./router, which brings only the queue into the request path:

import * as cloudflare from "@sdxc/jobs/cloudflare";
import { jobEnqueuer } from "@sdxc/jobs/router";
import { createRouter } from "remix/router";

let router = createRouter({
	middleware: [jobEnqueuer(cloudflare.queue(() => env.QUEUE))],
});

router.post(routes.orders.receipt, async (ctx) => {
	await ctx.jobs.enqueue(jobs.sendReceipt, { orderId: ctx.params.id });
	return new Response(null, { status: 202 });
});

End a run

export default createJobHandler(jobs.sendReceipt, async (ctx) => {
	let order = await findOrder(ctx.input.orderId);
	if (order === null) return ctx.exit("Order no longer exists");

	let response = await fetch(mailerUrl, { method: "POST", signal: ctx.signal });
	if (response.status === 429) return ctx.retry({ delay: "5 minutes", reason: "Rate limited" });
});

Returning acks the message. ctx.ack(), ctx.retry(), ctx.exit() and ctx.timeout() throw, so they end the run from anywhere in the call stack; write return ctx.retry(…) so TypeScript reads the lines below as unreachable. Reach for exit() when a redelivery would reach the same result, such as bad input.

API

jobs(tree: JobTree): JobMap

The job map: the same shape as tree, with every leaf a JobDefinition carrying its name, cron, input schema and meta.

job(options?: JobOptions): JobLeaf

Declares one job. Every option is optional:

• input: the schema the payload is parsed against before the handler runs. An object or a bare value both work. Absent means no payload.

• cron: the five-field expression the job is enqueued on. Fewer fields is a type error, and an expression the platform would reject throws InvalidCronExpression from @sdxc/cron at declaration.

• meta: anything the app reads back through ctx.of(job), typed as written.

createJobHandler(job, handler): JobHandler

Pairs a handler with its job, typing ctx.input. The dispatcher refuses a handler mapped onto a different job.

createJobContext(job, init: JobContextInit): JobHandlerContext

A context typed the way the handler receives it, for calling a handler directly in a test. It skips the middleware chain, so the test publishes what the chain would with ctx.set().

messageBody(job, input?): JSONValue

The body one message carries, { job, body?, traceparent?, tracestate? }, for sending through a queue client of your own.

await env.QUEUE.send(messageBody(jobs.sendReceipt, { orderId }), { contentType: "json" });

createJobDispatcher(options?: JobDispatcherOptions): JobDispatcher

• logger: a Logger from @sdxc/logger whose service every record carries.

• queue: the JobQueue to enqueue through. Without one the dispatcher still runs deliveries and throws when asked to enqueue.

• middleware: the chain every job runs inside, in order. Write it inline so handlers see what it installs.

• timeout: a duration after which ctx.signal aborts. The handler then has five seconds to end the run before the dispatcher records a timeout and retries the message.

• onEnd(ctx, status: JobStatus): runs once the ending is decided and before the message is settled. A throw is recorded as job.hook_failed and running past five seconds as job.hook_overran; the ending settles as decided either way. It runs after the middleware chain has unwound.

• maxAttempts: deliveries a message gets before it is dead-lettered, on a queue whose retries is "core".

tick with only enqueues the jobs declaring that exact expression; without it, the jobs whose schedule fires in now's UTC minute. deliverBatch with deadLettered: true records and acks every delivery instead of running it. A message naming no mapped job or failing its schema is refused and dead-lettered without loading a handler.

JobContext

What every middleware and handler receives for one delivery. new JobContext(job, init) builds the untyped one the dispatcher uses.

JobContextInit takes id, attempts, and optionally input, batchSize, log, signal and trace. Inside a middleware only keys are visible, so read an earlier middleware's value with ctx.require(Key).

Endings

The four verbs throw subclasses of Ending, grouped as Job (Job.Ack, Job.Retry, Job.NonRetriable, Job.Timeout, Job.Ending) and exported one by one from ./errors. Retry carries the delay it asked for.

Once the timeout has aborted the signal, any ending is recorded as timeout; ctx.ack() still acks the message.

Records

Each invocation emits one record through the dispatcher's logger, with flat dotted fields and the W3C Trace Context it ran under (trace_id, span_id, parent_span_id).

A refused message ends refused with job.refusal (unknown-job or invalid-input), and one read off the dead-letter queue ends dead_letter; both keep the body as job.body. A message enqueued during a request or another job continues that trace, so the two share a trace_id.

jobEnqueuer(queue: JobQueue): Middleware from ./router

Publishes ctx.jobs (a JobEnqueuer with enqueue and enqueueMany) and the Jobs context key. Importing it types ctx.jobs on every RequestContext, and a failed write throws the queue's JobQueueError.

jobEnqueuer(queue: JobQueue, options?: { dispatcher }): CommandMiddleware from ./cli

Publishes ctx.jobs to an @sdxc/cli command under the same Jobs key, writing through queue; each message carries the run's trace. With options.dispatcher, queue is a memory queue whose jobs run through that dispatcher in the command's process once the handler settles, round after round until nothing is due. A job that throws settles as a retry, and messages still delayed are noted as jobs.pending on the command's log.

queue(binding) and worker(dispatcher, options?) from ./cloudflare

queue(() => env.QUEUE) writes through a Cloudflare Queue binding, splitting large enqueueMany calls into batches the platform accepts; the platform counts retries. worker(dispatcher, options) answers { queue, scheduled } handlers. options.deadLetterQueue names the dead-letter queue this Worker also consumes, whose batches are recorded and acked, and options.deadLetter: () => env.DLQ is where a refused message is written. A retry delay past 12 hours is held for 12 hours, and send fails a JobMessage.delay past 12 hours with unsupported_delay.

queue(options?) from ./memory

A MemoryQueue holding serialized messages in an array, with its own attempt count (retries: "core") and dead-letter list. options.now supplies the clock. drain(deliver) runs and settles everything claimable, messages and deadLettered read what it holds, and reset() empties both.

JobQueue from ./queue

The backend interface. Every call answers a Result from @sdxc/result failing with a JobQueueError whose code is unavailable, invalid_message or unsupported_delay.

interface JobQueue {
	readonly retries: "backend" | "core";
	send(messages: JobMessage[]): Promise<Result<void, JobQueueError>>;
	claim?(options: ClaimOptions): Promise<Result<JobDelivery[], JobQueueError>>;
	settle?(delivery: JobDelivery, settlement: Settlement): Promise<Result<void, JobQueueError>>;
}

A pushed backend implements send; a pulled one adds claim and settle together. retries says whether the backend or the dispatcher's maxAttempts decides when to dead-letter. invalidMessage(body) wraps a refused body for a backend that writes to its dead-letter queue, and readDeadLetter(body) answers { reason, body } for a dead-lettered one.

createUptimeReporter({ token, url? }) from ./uptime

A ping (monitorId) => Promise<Result<void, UptimeError>> for a cron monitor, meant for onEnd. UptimeError.code is refused, unreachable, or unconfigured when token() answers undefined.

conformance(options) from ./conformance

Registers the Vitest suite a JobQueue adapter passes. options takes name, create, messages(queue), and optionally advance(ms) to run the delay tests and unsupportedDelay to run the ceiling tests.

Types

JobDefinition, JobMap, JobTree, JobLeaf and JobOptions describe a map, and AnyJobDefinition any job in it. JobInput<typeof job> and JobArgs<typeof job> name what enqueuing a job takes, for a helper of your own. JobHandler, JobHandlerFunction and JobHandlerContext<Schema> describe handlers. JobMiddleware<Effect> is a middleware whose Effect ({ key, value, property }) names what it installs. JobTypes is augmented with context: JobDispatcherContext<typeof dispatcher> to type handlers with what the chain installs. JobStatus is what onEnd receives; JobDelivery, JobMessage and Settlement are what a backend translates.

Pattern: Middleware That Provides A Database

A middleware publishes a value under a context key and installs it as a property; augmenting JobTypes types that property on every handler.

import type { JobDispatcherContext, JobMiddleware } from "@sdxc/jobs";

import { createJobDispatcher, createJobHandler } from "@sdxc/jobs";
import { createContextKey } from "remix/router";

import jobs from "./jobs.js";

export const Database = createContextKey<Database>();

function database(): JobMiddleware<{
	key: typeof Database;
	value: Database;
	property: "database";
}> {
	return async (ctx, next) => {
		let db = await connect();
		ctx.set(Database, db, { property: "database" });
		try {
			await next();
		} finally {
			await db.close();
		}
	};
}

export const dispatcher = createJobDispatcher({ middleware: [database()] });

declare module "@sdxc/jobs" {
	interface JobTypes {
		context: JobDispatcherContext<typeof dispatcher>;
	}
}

export default createJobHandler(jobs.cleanup, async (ctx) => {
	let removed = await ctx.database.deleteExpiredSessions();
	ctx.log.set({ sessions: { removed } });
});

A later middleware reads it with ctx.require(Database).

Pattern: Scheduled Jobs Watched By A Cron Monitor

Two schedules share their work through a plain function, each keeps its own handler, and each declares the monitor that watches it in meta. onEnd pings only for a done run, so a monitor alerting means the job really stopped completing.

import type { JobHandlerContext } from "@sdxc/jobs";

import { createJobDispatcher, createJobHandler, job, jobs } from "@sdxc/jobs";
import { createUptimeReporter } from "@sdxc/jobs/uptime";
import { env } from "cloudflare:workers";

interface Monitored {
	monitorId: string;
}

export const JOBS = jobs({
	dailyDigest: job({ cron: "0 8 * * *", meta: { monitorId: "mon_daily" } satisfies Monitored }),
	weeklyDigest: job({ cron: "0 9 * * 1", meta: { monitorId: "mon_weekly" } satisfies Monitored }),
});

async function sendDigests(ctx: JobHandlerContext<undefined>, period: "day" | "week") {
	ctx.log.set({ digest: { period } });
	await mailDigests(period);
}

export const daily = createJobHandler(JOBS.dailyDigest, (ctx) => sendDigests(ctx, "day"));
export const weekly = createJobHandler(JOBS.weeklyDigest, (ctx) => sendDigests(ctx, "week"));

let uptime = createUptimeReporter({ token: () => env.UPTIME_TOKEN });

export const dispatcher = createJobDispatcher({
	async onEnd(ctx, status) {
		if (status.type !== "done") return;
		for (let watched of [JOBS.dailyDigest, JOBS.weeklyDigest]) {
			let run = ctx.of(watched);
			if (run !== null) await uptime(run.meta.monitorId);
		}
	},
});

dispatcher.map(JOBS.dailyDigest, daily);
dispatcher.map(JOBS.weeklyDigest, weekly);

An awaited ping finishes before the message is acked. Its Result is discarded here on purpose: a failed ping never redelivers work that succeeded.

Pattern: Retries, Timeouts And Cancellation

A timeout aborts ctx.signal but cannot stop a handler, so pass the signal to every fetch and check it between steps. A broad catch rethrows any Ending so the handler's own decision survives.

import { createJobHandler, Job } from "@sdxc/jobs";

import jobs from "./jobs.js";

export default createJobHandler(jobs.syncInvoices, async (ctx) => {
	for (let invoice of await pendingInvoices()) {
		if (ctx.signal.aborted) return ctx.ack(); // what already synced stays synced

		try {
			let response = await fetch(invoice.url, { signal: ctx.signal });
			if (response.status === 429) {
				return ctx.retry({ delay: "1 minute", reason: "Rate limited" });
			}
			if (response.status === 404) return ctx.exit(`Invoice ${invoice.id} is gone`);
			await markSynced(invoice, await response.json());
		} catch (error) {
			if (error instanceof Job.Ending) throw error;
			return ctx.retry({ reason: "Sync failed", delay: "1 minute", cause: error });
		}
	}
});

Acking under an aborted signal settles the message but records a timeout, so onEnd sees timeout rather than done.

Pattern: A Job That Fans Out Work

A cron job whose handler enqueues keeps the fan-out on the queue, where each check gets its own run, retries and record.

import { createJobHandler } from "@sdxc/jobs";

import { dispatcher } from "./dispatcher.js";
import jobs from "./jobs.js";

export default createJobHandler(jobs.scheduleChecks, async (ctx) => {
	let due = await sitesDueForCheck(new Date());
	await dispatcher.enqueueMany(
		jobs.checkSite,
		due.map((site) => ({ siteId: site.id })),
	);
	ctx.log.set({ checks: { enqueued: due.length } });
});

Every message carries this run's trace, so each checkSite record names it as parent.

Pattern: Testing A Handler

A handler is a function over a context. Build one with createJobContext, publish what the middleware would, and assert on the ending it throws or the record it writes.

import { createJobContext, Job } from "@sdxc/jobs";
import { Log } from "@sdxc/logger";
import { expect, test } from "vitest";

import handler from "./jobs/send-receipt.js";
import jobs from "./jobs.js";
import { Database } from "./middleware.js";

test("asks for a retry while the mailer rate limits", async () => {
	let ctx = createJobContext(jobs.sendReceipt, { id: "m1", attempts: 1, input: { orderId: "o1" } });
	ctx.set(Database, await testDatabase(), { property: "database" });

	await expect(handler(ctx)).rejects.toBeInstanceOf(Job.Retry);
});

test("records the order it sent", async () => {
	let records: Record<string, unknown>[] = [];
	let log = new Log({ kind: "job", sink: (record) => void records.push(record) });
	let ctx = createJobContext(jobs.sendReceipt, {
		id: "m1",
		attempts: 1,
		input: { orderId: "o1" },
		log,
	});
	ctx.set(Database, await testDatabase(), { property: "database" });

	await log.run(() => handler(ctx));

	expect(records[0]).toMatchObject({ "order.id": "o1" });
});

Pattern: Running Jobs Without A Platform

The memory queue runs the whole loop in one process: tick() without only evaluates each schedule against now, and drain() delivers and settles what is queued, counting attempts against maxAttempts. Asserting dispatcher.crons against the Worker's configured triggers catches a schedule that would never fire.

import { createJobDispatcher } from "@sdxc/jobs";
import * as memory from "@sdxc/jobs/memory";
import { expect, test } from "vitest";

import jobs from "./jobs.js";

test("the midnight tick runs cleanup once", async () => {
	let queue = memory.queue();
	let dispatcher = createJobDispatcher({ queue, maxAttempts: 3 });
	dispatcher.map(jobs.cleanup, () => import("./jobs/cleanup.js"));

	await dispatcher.tick({ now: new Date("2026-01-01T00:00:00Z") });
	expect(queue.messages).toMatchObject([{ job: "cleanup" }]);

	let settlements = await queue.drain((delivery) => dispatcher.deliver(delivery));
	expect(settlements).toEqual([{ type: "ack" }]);
	expect(dispatcher.crons).toEqual(["0 0 * * *"]);
});

Pattern: Running Enqueued Jobs From A Command

A command line's jobEnqueuer() takes the queue the run should write to. Locally that is a memory queue drained through the app's own dispatcher, so a job a command enqueues runs in the terminal the developer is watching, against the same database; against production it is the deployed queue, and the deployed consumer does the work.

import { command, commands, createCLI, success } from "@sdxc/cli";
import { createJobDispatcher } from "@sdxc/jobs";
import { jobEnqueuer } from "@sdxc/jobs/cli";
import * as memory from "@sdxc/jobs/memory";
import { createLogger } from "@sdxc/logger";
import { log } from "@sdxc/logger/middleware/cli";
import { trace } from "@sdxc/trace-context/middleware/cli";

import jobs from "./jobs.js";

let logger = createLogger({ service: "app-cli" });
let queue = memory.queue();
let dispatcher = createJobDispatcher({ queue, logger, maxAttempts: 3 });
dispatcher.map(jobs.sendReceipt, () => import("./jobs/send-receipt.js"));

let contract = commands({
	receipts: commands({ resend: command({ description: "Resend receipts" }) }),
});

let cli = createCLI({
	name: "app",
	commands: contract,
	middleware: [log(logger), trace(), jobEnqueuer(queue, { dispatcher })],
});

cli.map(contract.receipts.resend, async (ctx) => {
	let orders = await findUnsentOrders();
	await ctx.jobs.enqueueMany(
		jobs.sendReceipt,
		orders.map((order) => ({ orderId: order.id })),
	);
	return success();
});

process.exitCode = await cli.run(process.argv.slice(2));

Each job writes its own job record under the command's log and continues its trace. Building the dispatcher over the same memory queue means the jobs those jobs enqueue run as well.

Pattern: Writing A Queue Adapter

An adapter implements JobQueue and passes the conformance suite, which is what makes it a substitute for the shipped ones.

import type { JobMessage, JobQueue } from "@sdxc/jobs/queue";

import { conformance } from "@sdxc/jobs/conformance";
import { JobQueueError } from "@sdxc/jobs/queue";
import { failure, success } from "@sdxc/result";

interface Topic {
	publish(bodies: unknown[]): Promise<void>;
	readonly published: unknown[];
}

export function topicQueue(topic: Topic): JobQueue {
	return {
		retries: "backend",
		async send(messages: JobMessage[]) {
			try {
				await topic.publish(
					messages.map(({ job, body }) => (body === undefined ? { job } : { job, body })),
				);
				return success(undefined);
			} catch (error) {
				return failure(
					new JobQueueError("The topic refused the write", { code: "unavailable", cause: error }),
				);
			}
		},
	};
}

let topic = createTestTopic();

conformance({
	name: "topicQueue",
	create() {
		topic = createTestTopic();
		return topicQueue(topic);
	},
	messages: () => topic.published,
});

A backend whose platform delivers by push calls dispatcher.deliverBatch() from its consumer and applies each Settlement; one that is polled adds claim and settle, and advance lets the suite check delays.