Jobs

k.jobs declares background work beside your routes, each job carrying the handler that runs it.

import { z } from 'zod';
import { cron } from 'kizunajs';
import { k } from './k';

export const jobs = k.jobs('scheduler', {
    sendDigests: k
        .job({
            schedule: cron.daily('05:00'),
            summary: 'Send the daily digest to every user',
            result: z.object({
                sent: z.int(),
            }),
        })
        .handler(async () => ({
            status: 200,
            body: {
                sent: await sendPendingDigests(),
            },
        })),
    indexUser: k
        .job({
            retry: 3,
            input: z.object({
                userId: z.string(),
            }),
        })
        .handler(async ({ input, throwError }) => {
            const user = await db.users.findById(input.userId);
            if (!user) {
                throwError({
                    status: 422,
                    body: {
                        detail: `No user with id ${input.userId}`,
                    },
                });
            }
            await search.index(user);
        }),
});

export const jobs = k.jobs('scheduler', {
    billing: {
        reconcileInvoices: k
            .job({
                schedule: cron.every('15m'),
            })
            .handler(async () => ({
                status: 200,
                body: {
                    reconciled: await reconcile(),
                },
            })),
    },
});

import { jobs } from './src/jobs';
import { scheduler } from './src/identities';

export default defineConfig({
    adapter: expressAdapter(),
    routes,
    jobs,
    auth: {
        identities: {
            scheduler,
        },
    },
});

Schedules

A five-field cron expression, read as UTC, or an object to read it in a time zone:

schedule: '0 5 * * *';

schedule: {
    cron: '0 3 * * *',
    timezone: 'Europe/Oslo',
}

Helpers return plain cron strings:

An invalid expression throws when the config is assembled, naming the field.

Return status and retries

Running and queueing

Every handler receives a jobs runner shaped like the declaration, with two methods per job:

createUser: k
    .route({
        method: 'POST',
        path: '/users',
        body: CreateUserSchema,
        responses: {
            201: UserSchema,
        },
    })
    .handler(async ({ body, jobs }) => {
        const user = await db.users.create(body);

        await jobs.indexUser.queue({
            input: {
                userId: user.id,
            },
        });

        return {
            status: 201,
            body: user,
        };
    }),

// Run it now, block, get the result.
const result = await jobs.billing.reconcileInvoices.run({ since });

// Put it in line and answer the request.
await jobs.search.reindex.queue();

queue takes a message:

await jobs.chargeInvoice.queue({
    input: {
        invoiceId,
    },
    dedupeKey: `charge:${invoiceId}`,
    runAt: threeDaysFromNow,
});

Input is validated against the job's input schema either way.

A job's handler receives the same runner, so a scheduled job can queue one job per unit of work rather than doing all of it inline.

Setting it up

pnpm add kizunajs@beta

bun add kizunajs@beta

npm install kizunajs@beta

export default defineConfig({
    adapter: expressAdapter(),
    routes,
    jobs,
    jobRunner: {
        onError: (job, error) => Sentry.captureException(error, { tags: { job } }),
    },
});

import { startJobs } from 'kizunajs/jobs';
import kizuna from '../kizuna.config';

kizuna.api.mount(app);
startJobs(kizuna.api);

export default defineConfig({
    adapter: expressAdapter(),
    routes,
    jobs,
    jobRunner: {
        transport: bullmq(connection),
    },
});

kizuna.api.mount(app);
startJobs(kizuna.api);

import { startJobWorker } from 'kizunajs/jobs';
import kizuna from '../kizuna.config';

const worker = await startJobWorker(kizuna.api);
process.on('SIGTERM', () => void worker?.stop());

export default defineConfig({
    adapter: nextAdapter(),
    routes,
    jobs,
    auth: {
        identities: {
            scheduler,
        },
    },
    jobRunner: {
        method: 'GET',
    },
});

{
    "crons": [
        {
            "path": "/jobs/dispatch",
            "schedule": "* * * * *"
        }
    ]
}

export default defineConfig({
    adapter: nextAdapter(),
    routes,
    jobs,
    jobRunner: {
        transport: qstash({
            token: process.env.QSTASH_TOKEN,
            baseUrl: 'https://api.example.com',
            schedulerSecret: process.env.CRON_SECRET,
        }),
    },
});

Transports

A transport carries a queued job out of this process to whatever runs it. Name one under jobRunner and your existing queue calls become as durable as it is:

export default defineConfig({
    adapter: expressAdapter(),
    routes,
    jobs,
    jobRunner: {
        transport: myQueue,
    },
});

Kizuna ships no transports. QStash, BullMQ, pg-boss, Cloud Tasks, SQS, or a table in your own database all fit the same interface: see Create a job transport.

A job is addressed by its dotted key, so a transport carries the name and the input. Queues come in two shapes, and the shape decides whether you run anything extra:

startJobWorker returns undefined against a queue that needs no worker. Run it in the API's process or its own.

retry: 3 on a job says how many attempts a failure deserves; your transport does the retrying. Kizuna warns when a job asks for something its transport will drop, at startup for retry and at the queue call for runAt and dedupeKey.

The two job endpoints

api.mount serves two endpoints for every job. Both mount as ordinary routes, so the jobs' identity guards them and failures render as Problem Details.

POST /jobs/dispatch runs whichever jobs the elapsed window belonged to. Point platform cron at it.

POST /jobs/run runs the one job the body names, validating input against that job's schema and answering 422 when it does not fit. It answers with the job's own status, which a queue reads as the retry signal.

Configure both under jobRunner on defineConfig, beside the transport and onError above:

method moves /jobs/dispatch only: /jobs/run takes a body, so it is always POST. Nothing is served on /jobs itself.

A tick runs each due job inline, in the request the scheduler made, so the function's timeout is the budget for all of them. Keep the tick short by having a scheduled job queue one job per unit of work:

export const jobs = k.jobs('scheduler', {
    sendDigests: k
        .job({
            schedule: cron.daily('05:00'),
            result: z.object({
                queued: z.int(),
            }),
        })
        .handler(async ({ jobs }) => {
            const due = await db.users.needingDigest();
            for (const user of due) {
                await jobs.sendOneDigest.queue({
                    input: {
                        userId: user.id,
                    },
                    dedupeKey: `digest:${user.id}:${today()}`,
                });
            }
            return {
                status: 200,
                body: {
                    queued: due.length,
                },
            };
        }),
    sendOneDigest: k
        .job({
            input: z.object({
                userId: z.string(),
            }),
        })
        .handler(async ({ input }) => {
            await mailer.sendDigest(input.userId);
        }),
});

Local development

When you deploy to platform cron and want pnpm dev to exercise the real HTTP path, startJobsDevRunner ticks the dispatch endpoint over the network on an interval, the way your platform cron will:

import { startJobsDevRunner } from 'kizunajs/jobs';

if (process.env.NODE_ENV !== 'production') {
    startJobsDevRunner(kizuna.api, {
        baseUrl: `http://localhost:${port}`,
        secret: process.env.CRON_SECRET,
    });
}

What Kizuna does not do

Kizuna owns no database and no queue, so it does not:

• Store your jobs. Durability, dead-lettering, and a dashboard come from your transport.
• Retry. Your transport does. Without one, nothing retries.
• Know whether a job already ran. It only names the run so a deduplicating transport can tell.
• Prevent overlapping runs. Take a lock in your own database.
• Keep run history or backfill missed ticks.