Jobs and the worker

Anything slow, retryable or heavy runs as a job: PNG renders, exports, mails, webhooks, rebuilds, nightly rollups. A request handler or server action only enqueues; a worker process does the work with retries and backoff. The queue lives in Postgres (pg-boss), so local development and a small production need no extra infrastructure, and enqueue() is the only API callers use, so the backend can be swapped later without touching call sites.

The package: @kwisple/jobs

packages/jobs/src/index.ts exports everything:

pg-boss keeps its tables in their own jobs schema in the same database; you never migrate it yourself. Queues are created on first use (createQueue before every send, work and schedule).

Every job

From jobSchemas, with who sends it and what the handler does.

Enqueueing

import { enqueue } from "@kwisple/jobs";

await enqueue("workspace.export", { workspaceId: ctx.workspace.id, userId: ctx.user.id });

The payload is parsed with the job's schema first, so a wrong payload throws in the caller, not in the worker. Options:

How the codebase uses them:

Treat enqueue as fallible where the user's action must still succeed: call sites that only start background work end with .catch(...) and log, so a queue hiccup never fails a save.

Working: work() and concurrency

await work("webhook.deliver", async (job) => { await deliverWebhook(job); }, { concurrency: 5 });

work fetches one job at a time (batchSize: 1). concurrency (default 1) becomes pg-boss localConcurrency: that many jobs of the queue run side by side in this process. Only webhook.deliver uses it today (5), so one slow receiver with a 10 s timeout does not hold up everyone else's deliveries.

A handler that throws marks the attempt failed; pg-boss retries it according to the options the job was sent with. Handlers must be idempotent: a retry, a duplicate delivery or a manual re-run must not double anything. The nightly jobs are written that way (a rollup deletes and rebuilds its days; the purge only removes what is past retention).

Schedules: schedule()

await schedule("analytics.rollup", "15 2 * * *", {});

A cron expression evaluated in tz (default Europe/Brussels). It is idempotent: the worker calls it on every boot and pg-boss keeps one schedule per name. pg-boss turns each tick into an ordinary job of that name, so the handler is registered with work like any other job.

The worker process

apps/dashboard/src/worker/:

Handlers reuse the same service functions as the server actions (lib/services/*, lib/email/*), open tenant data with forWorkspace(db, workspaceId) from the payload, and log one line per job with the label ([worker] or [jobs:inline]).

Run it:

pnpm --filter @kwisple/dashboard worker   # tsx --env-file-if-exists=../../.env src/worker/index.ts

In production the worker target of the Dockerfile runs the same entry with tsx and JOBS_INLINE=0 (Deploying the platform).

The schedule table

Registered at the end of startWorkers, all in Europe/Brussels time:

JOBS_INLINE: jobs inside the dashboard

inlineWorkersEnabled() decides whether the dashboard server registers the handlers in its own process:

src/instrumentation.ts calls ensureInlineWorkers() at server start (after checking the secrets), so in development pnpm dev processes its own jobs and you never start a worker. Call sites that enqueue work a person waits for (asset.process, social.render, cache.purgeAll, cms.scheduled) call ensureInlineWorkers() first, so a dev server that was already running before a handler existed, or after hot reload, still picks it up. ensureInlineJob("site.deploy") does the same for one job that was added later; start.ts marks it with markInlineJob. Both are no-ops when inline workers are off.

In production compose sets JOBS_INLINE: "0" on the dashboard and runs the worker container instead, sized for renders and exports.

Retention and housekeeping

Other finished jobs follow pg-boss's own retention defaults.

Adding a job

1. Declare it in jobSchemas (packages/jobs/src/index.ts): a dotted name (<module>.<verb>) and a Zod object. Put ids in the payload, not data: the handler loads what it needs, so a retry works on fresh rows and no personal data sits in the queue. Cap strings with .max(). A scheduled job takes z.object({}).default({}). Add a one-line comment above it.
2. Write the handler. A short one inline in startWorkers, otherwise apps/dashboard/src/worker/handlers/<name>.ts exporting a JobHandler<"<name>">. Open tenant data with forWorkspace(db, payload.workspaceId); never trust that the row still exists (it may be trashed or deleted since).
3. Register it in startWorkers (apps/dashboard/src/worker/start.ts) with await work("<name>", handler), add { concurrency } only if one slow job must not block others, and add it to the "listening for jobs" log line.
4. Schedule it if recurring: await schedule("<name>", "<cron>", {}) next to the others, with a comment saying when and why.
5. Enqueue it from the service layer, not from a component: await enqueue("<name>", {...}, { retryLimit, singletonKey, … }). If a person waits for the result in development, call ensureInlineWorkers() before it.
6. Report back. Long work that a person asked for ends with a notification (as workspace.export and brand.guide do), not with a spinner that waits on the job.
7. Test the service function the handler calls; apps/dashboard/test/jobs-assets.test.ts is an example of a job-related test (Testing and checks).
8. Document it: the job table on this page, and docs/handleiding if people see its effect.