Background jobs

Anything slow, anything that calls another service, and anything on a schedule is a job. Jobs run from BullMQ on Redis, and the API runs the worker.

A job is written to the jobs table in Postgres first, as part of whatever change it belongs to. Once that is saved, a relay hands it to BullMQ, under the row’s id, so handing it over twice runs it once. That is what lets a job exist only if its transaction commits, and a key keep a job from being queued twice, with the queue itself on Redis.

Add a job

Define it: a name and the shape of its payload. Put ids in the payload, not whole records, so the job reads what is true when it runs.

import { z } from 'zod';
import { defineJob } from '../jobs/job.js';

export const buildReport = defineJob(
  'projects.build-report',
  z.object({ projectId: z.uuid(), sendTo: z.email() }),
);

Handle it in a method of any provider in a module. It is found when the API starts; a job has exactly one handler.

@Injectable()
export class ProjectReports {
  @OnJob(buildReport)
  async build({ projectId, sendTo }: PayloadOf<typeof buildReport>): Promise<void> {
    // ...
  }
}

Queue it from anywhere, by injecting JobsService. The payload is checked against the schema here and again when the job runs.

await this.jobs.enqueue(buildReport, { projectId, sendTo: user.email });

Only if the change is saved, only once, or later

// Inside database.transaction(), it exists only if the transaction commits
await this.database.transaction(async () => {
  await saveTheReport();
  await this.jobs.enqueue(buildReport, payload);
});

// A second job with this key is dropped; enqueue answers false
await this.jobs.enqueue(buildReport, payload, { key: `report:${projectId}:2026-10` });

// Not before this time
await this.jobs.enqueue(buildReport, payload, { runAt: tomorrow });

A key is remembered for 7 days after its job was handed over: the cleanup removes its row then.

Write handlers that can run twice

• Throw to have it tried again. Do not catch an error just to log it: the job then counts as done.
• Make a second run harmless. Check before you act, or use a key. A worker can stop after the work and before saying so, and BullMQ then runs the job again.
• Never use setTimeout, setInterval or an in-memory list for work that has to happen.

A job is tried 5 times by default. The waits start at 15 seconds and double (15 s, 30 s, 1 min, 2 min), so it gives up about 4 minutes after it first failed. For work that should outlast a longer outage, pass { maxAttempts: 10 } to defineJob: about 2 hours. BullMQ keeps a job that gave up for 30 days, with the reason, and a finished one for 7.

Run something on a schedule

@Scheduled({ name: 'projects.weekly-report', cron: '0 9 * * 1' })
async sendWeeklyReports(): Promise<void> {
  // Every Monday at 09:00 UTC
}

Times are in UTC and names must be unique. Each occurrence runs once, however many copies of the API are running.

Test it

Nothing runs in the background in tests. Run what is queued with drain():

const t = await createTestApp({ teams: false });
const queue = t.app.get(JobsService);

await queue.enqueue(buildReport, { projectId, sendTo: 'ada@example.com' });
await queue.drain();

The handler is already registered through its module. Pass providers: [...] to createTestApp only for a handler that is in no module, such as one written for the test.

Pass a date to run jobs due by then: queue.drain(new Date(Date.now() + 60 * 60 * 1000)). Use a date in the future: BullMQ keeps a job until its time by the real clock.

A scheduled method has nothing to enqueue: ask the worker to queue the next occurrence, then run the queue at that time:

import { JobsWorker } from '../src/jobs/jobs.worker.js';

const worker = t.app.get(JobsWorker);
// nextMonday(time): a Monday to come, at that time in UTC
await worker.queueSchedules(nextMonday('08:00'));
await queue.drain(nextMonday('09:00'));

Put a scheduled method on a provider of the feature’s module, as apps/api/src/jobs/jobs.cleanup.ts does. A clean-up that deletes rows by age across every workspace is fine there: it reads nobody’s data. Anything that reads across workspaces belongs in AdminService.

Run workers on their own

By default the API runs the worker, and several copies of the API share the queue safely. To keep jobs from competing with requests, set JOBS_WORKER=false on the API and run the worker separately: bun run worker locally, bun src/worker.ts in the production image. On Railway, RAILWAY_WORKER=true does both: see Railway.

To look at the queue, or retry a job that gave up, use BullMQ through JobsService.queue: (await jobs.queue.getJob(id))?.retry(). Its keys in Redis start with app:queue.

Upgrading from the queue in Postgres

Nothing to do. Jobs that were waiting in the jobs table when the new version started are handed to BullMQ like new ones, keeping the attempts they had left; one a worker of the old version was running is taken over after 10 minutes, as before.

With traces on, a job’s span continues the trace of the request that queued it, and jobs.duration, jobs.wait and jobs.queue.depth say how the queue keeps up.

Errors you may see: Troubleshooting.