Adding a scheduled job

Adding a scheduled (cron-style) job touches three places. The contract is intentionally split so the wire-name catalogue stays small, the descriptors live with the domain that owns them, and processors stay close to the worker that runs them.

The three places

• packages/business/jobs/src/job-names.ts — add the string name to the JOB_NAMES map. This is the wire-contract identifier BullMQ uses; renaming it requires a coordinated rolling deploy, so pick well.
• packages/business/jobs/src/scheduled-jobs/my-new-job.ts — the descriptor: the cron expression and, when the job needs it, the overlap lock. It is a plain object literal typed as ScheduledJobDescriptor, not a call to a factory helper. Imports the job name from job-names.ts.
• apps/backend/worker/src/processors/my-new-job.ts — the processor: a @Service() class extending ScheduledJobProcessor that names its descriptor and implements handle.

The worker reads the descriptor registry at boot and reconciles it with BullMQ via JobScheduler.upsertAll, which also removes schedules whose descriptor has since been deleted. There is no separate cron container.

The step-by-step

1. Pick a wire-name. Kebab-case, descriptive, stable.

   packages/business/jobs/src/job-names.ts

   export const JOB_NAMES = {
     // ...existing...
     myNewJob: 'my-new-job',
   } as const;

2. Write the descriptor.

   packages/business/jobs/src/scheduled-jobs/my-new-job.ts

   import type { ScheduledJobStepDescriptor } from '@scani/queue';
   import { JOB_NAMES } from '../job-names';
   
   // A step of the nightly group: no cron, the group owns the schedule.
   export const MY_NEW_JOB_SCHEDULE: ScheduledJobStepDescriptor = {
     name: JOB_NAMES.myNewJob,
     lockName: JOB_NAMES.myNewJob,  // opt into overlap protection
   };

   ScheduledJobStepDescriptor (packages/infra/queue/src/core/job-descriptor.ts) has name plus the optional lockName and jitterMs. A job that needs a schedule of its own uses ScheduledJobDescriptor, which adds cron, timezone and defaultOpts; since SC-1688 only active-pricing and weekly-digest do. It has no retryPolicy and no payloadSchema — those belong to user-initiated jobs, and Retry policies below says where.

3. Put it in a group. Add a { name: JOB_NAMES.myNewJob } line to its group in packages/business/jobs/src/scheduled-jobs/groups.ts, at the place its dependencies put it, and add the descriptor to SCHEDULED_JOB_STEPS in that directory’s index.ts. A step can take runOn (a weekly step in the nightly group) and timeoutMs (default 30 minutes).

4. Write the processor. Extend ScheduledJobProcessor. The base class applies jitterMs, takes the lock when the descriptor names one, and records the run’s heartbeat — so handle only holds the work.

   apps/backend/worker/src/processors/my-new-job.ts

   import { MyService } from '@scani/domain';
   import { MY_NEW_JOB_SCHEDULE } from '@scani/jobs';
   import { ScheduledJobProcessor } from '@scani/queue';
   import { Container, Service } from 'typedi';
   
   @Service()
   export class MyNewJobProcessor extends ScheduledJobProcessor {
     readonly descriptor = MY_NEW_JOB_SCHEDULE;
   
     protected async handle(): Promise<void> {
       await Container.get(MyService).run();
     }
   }

5. Register the processor in the worker’s single declarative list. There is no processors/index.ts barrel to re-export from:

   apps/backend/worker/src/index.ts

   import { MyNewJobProcessor } from './processors/my-new-job';
   
   function resolveProcessors() {
     return [
       // ...existing...
       Container.get(MyNewJobProcessor),
     ];
   }

6. Write tests for the underlying service. Don’t test the base class itself — it is shared infra and is tested in packages/infra/queue/tests/consumer/scheduled-job-processor.test.ts.

7. Document it in Job catalogue with the schedule, what it does, and what it depends on.

Overlap protection is opt-in, and there are two mechanisms

Set lockName and ScheduledJobProcessor.process wraps handle() in a Postgres advisory lock (PostgresJobLock, packages/business/jobs/src/infrastructure/postgres-job-lock.ts). A second fire is skipped and logged rather than raced, so two worker pods on the same cron tick will not both run it. With no lock implementation bound to the container the job runs unlocked — that fall-through is deliberate, and it is the dev and self-host path.

Leave lockName unset and overlap is tolerated by design, not prevented. The reconcile-* sweepers do this on purpose: they are idempotent re-scans, BullMQ’s deterministic job ID dedupes the duplicate enqueue, and jitterMs spreads replicas that would otherwise fire on the same wallclock second.

Read the descriptor, don't assume the lock

These are different guarantees with different failure modes, and neither is the default. A non-idempotent processor with no lockName gets no protection at all, and it surfaces as a duplicated side effect under load — far from the line that chose it. Decide which guarantee your job needs, then say so in the descriptor.

apps/backend/worker/src/lib/cron-lock.ts is a third thing and is not this path: a worker-local logging wrapper over the same primitive, with a single caller, and that caller is a user-initiated job rather than a scheduled one. Don’t reach for it when adding a scheduled job.

Full rationale in Why BullMQ + Postgres advisory locks.

Retry policies

A scheduled job does not name one. ScheduledJobDescriptor has no retryPolicy field. The named retry shapes are for user-initiated jobs, whose UserJobDescriptor carries the defaultOpts BullMQ reads.

They live in packages/business/jobs/src/retry-policies.ts and are JobsOptions fragments rather than strings — RETRY_NONE, RETRY_FAST, RETRY_EXTERNAL, RETRY_HEAVY. A user job spreads one:

export const WALLET_IMPORT: UserJobDescriptor<WalletImportJob> = {
  name: JOB_NAMES.walletImport,
  schema: walletImportSchema,
  defaultOpts: {
    ...RETRY_EXTERNAL,
    removeOnComplete: 100,
    removeOnFail: 500,
  },
  // computeJobId / summarizePayload …
};

A scheduled descriptor may set defaultOpts if it genuinely needs BullMQ options, but none does today. A failing step is retried inside its group’s run (3 attempts, 5s then 10s apart); after that it waits for the group’s next run, and the steps after it run regardless.

Schedule conventions

Frequency	Convention
Every 15 minutes	A step of housekeeping (2-59/15 * * * *) — probes and reconcilers.
Hourly	A step of hourly (0 * * * *) — pricing, balance syncs.
Nightly	A step of nightly (0 0 * * *), placed after what it reads.
Weekly	A nightly step with runOn for its day.
Morning mail	A step of morning (0 9 * * *).

The current schedule allocations are documented in Job catalogue. A heavy nightly job goes in the nightly group at the point its dependencies allow; the step order, not the clock, is what orders the chain.

User-initiated jobs (not the same thing)

A user-initiated job (screenshot-parse, exchange-import, wallet-import, …) lives under packages/business/jobs/src/user-jobs/ and is enqueued by the api in response to a user action. It carries a payload schema and a retry policy — neither of which a scheduled descriptor has — and each enqueue computes its own job ID. Different contract, similar shape.

See also

• Engineering conventions
• Why BullMQ + Postgres advisory locks
• Job catalogue