Jobs

A job is work the app does after a request, or that a request should not wait for: mail an invoice, push a change to another system, recompute a report. Jobs live in the app’s own Postgres database, in a schema named jobs, so a job can commit in the same transaction as the write that asks for it. The queue underneath is pg-boss, pinned to one version; Protobase’s own API in front of it is what the app uses.

config/jobs.ts

import { defineJob, FatalJobError } from '@protobase/server'
import { z } from 'zod'

export const sendInvoice = defineJob({
  name: 'send-invoice',
  payload: z.object({ invoiceId: z.number() }),
  maxAttempts: 10, // then kept as failed, with its last error
  retryDelay: '30 seconds', // doubling with each retry, up to maxRetryDelay
  run: async ({ invoiceId }, { db }) => {
    const invoice = await db.selectFrom('sales.invoices').select(['id', 'status']).where('id', '=', invoiceId).executeTakeFirst()
    // Retrying will not bring it back: fail for good.
    if (!invoice) throw new FatalJobError(`Invoice ${invoiceId} is gone`)
    // ... render the PDF and mail it
  },
})

// One at a time per customer, in the order the changes were made.
export const syncCustomer = defineJob({
  name: 'sync-customer',
  ordered: true,
  payload: z.object({ customerId: z.number(), fields: z.array(z.string()) }),
  run: async ({ customerId, fields }) => {
    // ... push the changed fields to the CRM
  },
})

// Every morning at 7 in Amsterdam: one run per time, however many workers run.
export const remindOverdue = defineJob({
  name: 'remind-overdue',
  schedule: '0 7 * * *',
  timezone: 'Europe/Amsterdam',
  run: async (_, { db }) => {
    // ... mail a reminder for every invoice past its due date
  },
})

protobase.config.ts

import { defineConfig } from '@protobase/server'
import * as jobs from './jobs'

export default defineConfig({
  config: {},
  jobs, // run by the worker in `protobase dev` and `protobase serve`
  options: {
    writeHooks: [
      // In the write's own transaction: the invoice and its job commit together or not at all.
      async (event, trx) => {
        if (event.resource.name !== 'invoices' || event.after?.status !== 'sent' || event.before?.status === 'sent') return
        const invoiceId = Number(event.after.id)
        await jobs.sendInvoice.enqueue(trx, { invoiceId }, { key: `invoice:${invoiceId}` })
      },
    ],
  },
})

defineJob option
name	Letters, digits, _, -, . and /. Two jobs may not share a name.
payload	A Standard Schema such as a zod schema, checked when the job is enqueued and again before it runs. A payload it refuses at run time fails the job for good.
run(payload, { db, job, signal, mail })	The work. db is the app’s pool: the enqueuer’s transaction has committed by then. job holds its id, name, attempt and maxAttempts. signal aborts when the try times out or the worker shuts down. mail sends mail, see Mail.
maxAttempts	Tries before the job is kept as failed; default 10.
retryDelay, maxRetryDelay	The wait before the first retry, doubling with each one (with some jitter), and its ceiling; default 2s and 1 hour. Milliseconds or text such as 30 seconds.
timeout	How long one try may run before it counts as failed; default 15 minutes.
concurrency	Jobs of this name running at once in each worker process; default 1.
ordered	Jobs run one at a time per group, in the order they were enqueued; see Ordered groups.
schedule, timezone	Runs the job on a cron schedule, in a time zone (default UTC); see Recurring jobs.

A job runs at least once: a worker that dies mid-job, or a try cut off by a shutdown, means it runs again. Write run so that running twice does no harm, for example by checking the state it is about to change.

Enqueueing

await sendInvoice.enqueue(db, { invoiceId: 7 }, { delay: '5 minutes', key: 'invoice:7', priority: 10 })

db is the app’s database, or the transaction of a write hook or an API function: there, the job commits with the caller’s writes, or not at all. It returns the job’s id.

Option
runAt or delay	When the job may run first: a Date, or milliseconds or text from now. Default now.
priority	Higher runs first; default 0.
key and mode	One waiting job per key: with mode: 'dedupe' (the default) an enqueue while one waits returns that one’s id and leaves it as it is; with mode: 'replace' the waiting job takes this payload and these options. A job that is running does not count, so a change made while it runs gets a job of its own.
group	The group of an ordered job: required there, refused elsewhere.

A job enqueued outside a transaction starts at once when this process runs its jobs. One enqueued in a transaction, or by another process, is picked up within the poll interval, 2 seconds.

From SQL

A trigger or a migration can enqueue in its own transaction:

select jobs.enqueue('send-invoice', '{"invoiceId": 7}');
select jobs.enqueue('send-invoice', '{"invoiceId": 7}', run_at => now() + interval '1 hour', key => 'invoice:7');
select jobs.enqueue('sync-customer', '{"customerId": 3, "fields": ["email"]}', group_key => 'customer:3');

jobs.enqueue(job_name, payload, run_at, key, group_key, priority) returns the job’s id. A key dedupes as mode: 'dedupe' does. The job takes its retry settings from its definition, as the workers last saw it; a job no worker has started yet gets pg-boss’s defaults until one does.

Mail

mail.send({ to, subject, text, html }) sends one message, with an optional HTML part beside the plain text, through the app’s SMTP server, the one that also sends password resets and sign-in codes: PROTOBASE_SMTP_URL and PROTOBASE_MAIL_FROM (see Email). An email template sends through it with template.send(to, data, mail). It rejects when the server refuses the message, so the try fails and is retried. Without those variables mail is undefined: a job that must send fails for good with FatalJobError, and one that may skip mail leaves it out.

run: async ({ invoiceId }, { db, mail }) => {
  if (!mail) throw new FatalJobError('Mail is off: set PROTOBASE_SMTP_URL and PROTOBASE_MAIL_FROM')
  await mail.send({ to: '[email protected]', subject: `Invoice ${invoiceId}`, text: 'Your invoice is attached in the portal.' })
}

A message is sent once per try, so a try that sends and then fails sends again: record what was sent in the same run, and check it first.

Failures

A try that throws is retried after its backoff, until maxAttempts. Then the job is kept as failed with its last error, for 7 days. throw new FatalJobError('410 Gone') fails it at once, without the attempts it has left.

A job that has waited 14 days past its start time without running is deleted.

While the database is read-only, as during a deploy’s write freeze, a write in run fails with SQLSTATE 25006: the try waits 5 seconds and runs run again, without using an attempt, until the database takes writes or the try times out.

Ordered groups

With ordered: true, every enqueue names a group, and the jobs of one group run one at a time, in the order they were enqueued; different groups run side by side, up to concurrency. A job that is waiting for a retry, or has failed, holds back the rest of its group until it succeeds or is retried or deleted: that is what keeps the order. An ordered job takes no key: its group is how pg-boss finds its jobs.

Turning ordered on or off for a job whose queue exists stops startup: let its jobs finish or delete them, then drop the queue with select jobs.delete_queue('<name>').

Recurring jobs

schedule: '0 7 * * *' runs a job on that cron schedule, with no payload; a seconds field may come first (*/30 * * * * *). timezone reads it in a zone such as Europe/Amsterdam, so 07:00 stays 07:00 across daylight saving time.

Each run is a job of its own, enqueued ahead for its time by the workers: its id is made from the job’s name and the time, so however many workers enqueue a run, it is one job and runs once. As a run starts, the next one is enqueued. A run that came due while no worker ran, as when the app slept or was deployed, runs once as soon as one does; the times missed after it are skipped. Changing the schedule drops the waiting run that is no longer one of its times, and a job that loses its schedule, or is removed, loses its waiting run.

Because the next run waits in the database like any delayed job, jobs.next_due() sees it, and a host that wakes the app for due jobs wakes it for its recurring jobs too. That is why Protobase does not use pg-boss’s own scheduler, which keeps no next run in the database and works the time out in a running process.

Failed jobs page

With jobs, the admin app has a Failed jobs page, for admins only: each job that used up its attempts or failed for good, with its payload and last error. Retry gives a job one more attempt, from now, and Delete removes it. It reads the jobs.failed view, which a project’s SQL can query too.

Where jobs run

• protobase dev runs them in its process. An edited job runs its new code from the next job on.
• protobase serve runs them in its process, beside the API, unless PROTOBASE_JOBS=false; it enqueues them either way. On SIGTERM it stops taking jobs and gives running ones 30 seconds to finish. A job still running then is failed, to be retried.
• protobase worker <bundle> runs the same jobs from the same bundle in a process of its own, with the same shutdown, for a serve with PROTOBASE_JOBS=false.
• Each worker checks every job name for due jobs every 2 seconds: one indexed query per job name and interval. Once a minute, one of the workers also counts the queues and retries the jobs whose worker died (their timeout passed); once a day it deletes completed and failed jobs older than 7 days.

Hosts that sleep

A host that stops an idle app needs to know when a job is due, and to have it run:

• select jobs.next_due() answers when the next job is due, or null when none waits: a waiting job’s start time, a recurring job’s next run, or the timeout of a job that was running when the app stopped. It is one read-only statement, through any pooler.
• POST /api/jobs/drain?seconds=25 runs the due jobs until none is due or running, or the seconds have passed, and answers { "drained": true, "remaining": 0, "nextDue": "2026-10-12T05:00:00.000Z" }. It takes the operator’s token, PROTOBASE_JOBS_DRAIN_TOKEN, as a bearer token; without that variable there is no such endpoint. A process with PROTOBASE_JOBS=false answers 409.

The jobs schema

The schema is pg-boss’s tables at its pinned schema version, plus Protobase’s jobs.enqueue, jobs.next_due and jobs.failed. It is created and upgraded like the auth schema, by the project’s migrations:

bun run protobase migration   # writes db/migrations/NNN_jobs.sql when the jobs schema is missing or behind
bun run db:migrate

protobase migration writes what the database lacks, taken from pg-boss’s own construction and migration plans for the pinned version, without their BEGIN and COMMIT and with any CONCURRENTLY index build made a plain one, so the file runs in the project’s migration transaction. The server never creates or migrates it: protobase dev and protobase serve read its version at startup and refuse to start while it is missing or behind, naming the command. A database whose jobs schema is newer than the running Protobase is refused too.

pg-boss runs here without what needs a session of its own: no LISTEN/NOTIFY (a job enqueued in the same process wakes its worker directly, the rest is polled), only transaction-scoped advisory locks, unnamed prepared statements, no index rebuilds and no instance registry. It works through PgBouncer in transaction mode and on PGlite, over the app’s own connection pool.