Jobs

Every API grows work that should not happen inside a request: nightly digests, invoice reconciliation, re-indexing a document someone just saved. k.jobs puts that work in the contract.

Declaring jobs

Jobs sit alongside routes on the contract. The first argument is the identity every job requires.

import { z } from 'zod';
import { cron } from '@ts-kizuna/core';
import { k } from './k';

export const jobs = k.jobs('scheduler', {
    sendDigests: {
        schedule: cron.daily('05:00'),
        result: z.object({
            sent: z.int(),
        }),
    },
    indexUser: {
        retry: 3,
        input: z.object({
            userId: z.string(),
        }),
    },
});

sendDigests runs on a clock. indexUser has no schedule; it gets queued when a user changes. A job can have both.

They nest like routes:

export const jobs = k.jobs('scheduler', {
    billing: {
        reconcileInvoices: {
            schedule: cron.every('15m'),
        },
    },
});

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 build the expression, and return plain cron strings:

An invalid expression throws when the contract is built, naming the field.

Writing handlers

server.jobs mirrors the declaration, nesting included:

export const jobHandlers = server.jobs({
    sendDigests: async () => ({
        status: 200,
        body: {
            sent: await sendPendingDigests(),
        },
    }),

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

A handler receives input, throwError, and jobs, and no request, response, or auth. A job with no declared result can return nothing.

Pass the handlers to server.api next to the router:

export const api = server.api({
    router,
    jobs: jobHandlers,
    guards: {
        scheduler: requireScheduler,
    },
});

The status you return is the retry contract

Return 422 for input that will never become valid, so it is not retried.

Running and queueing

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

export const router = server.router({
    createUser: 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.

Outside a request, build a runner yourself:

import { createJobRunner } from '@ts-kizuna/core';

const jobs = createJobRunner(contract, jobHandlers);
await jobs.sendDigests.run();

Setting it up

pnpm add @ts-kizuna/core

bun add @ts-kizuna/core

npm install @ts-kizuna/core

export const server = new KizunaServer(contract, {
    onJobError: (job, error) => Sentry.captureException(error, { tags: { job } }),
});

import { startJobs } from '@ts-kizuna/core/jobs';

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

export const server = new KizunaServer(contract, {
    jobTransport: bullmq(connection),
});

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

import { startJobWorker } from '@ts-kizuna/core/jobs';

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

export const k = new Kizuna({
    identities: {
        scheduler,
    },
    jobs: {
        method: 'GET',
    },
});

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

export const server = new KizunaServer(contract, {
    jobTransport: 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 and the same queue calls become as durable as it is, with no handler change:

export const server = new KizunaServer(contract, {
    jobTransport: 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, never a URL per job. Queues come in two shapes, and it decides whether you run anything extra:

startJobWorker returns undefined against a queue that needs no worker, so it is safe either way. Run it in the same process as the API 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, and no job has one of its own. 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 and there is nothing else to wire.

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. The status it answers with is the job's own, so a queue reads the retry contract straight off it.

Configure both on new Kizuna() under jobs:

method moves /jobs/dispatch only: /jobs/run takes a body, so it is always POST. Nothing is served on /jobs itself, and every adapter serves both the same way.

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 jobHandlers = server.jobs({
    sendDigests: 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: async ({ input }) => {
        await mailer.sendDigest(input.userId);
    },
});

sendOneDigest declares no schedule, so nothing ticks it: it exists only to be queued.

Local development

startJobs is the same line locally as in production. 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 '@ts-kizuna/core/jobs';

if (process.env.NODE_ENV !== 'production') {
    startJobsDevRunner(contract, {
        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.