@guestcraft/jobs

GuestCraft’s background jobs, on PostgreSQL with pg-boss (ADR 0019):

• the typed catalog of jobs;
• enqueueing in the unit of work;
• the resident worker and the budgeted drain;
• the schedules;
• the dead letter queue.

Nothing here reads the environment: the composition root passes the configuration in. Nothing connects at import (ADR 0010).

Export	For
defineJob, field, defineCatalog	the catalog: names, payloads of identifiers, timeouts and retries
createJobQueue	a runtime that only enqueues, such as the API with the app role
createJobRuntime	a runtime that runs jobs with the worker role: start/stop, drain, deadLetters, redrive, enqueue
registerJobs	the migration step: creates the queues and stores the schedules
channexJobs, privacyJobs, allJobs	the production catalogs: Channex alone, guest data rights alone, and every catalog for the worker and the migration step

import { createJobQueue, createJobRuntime, defineCatalog, defineJob, field } from "@guestcraft/jobs";

const pushAri = defineJob({
  name: "channex.push_ari",
  scope: "workspace", // the payload carries workspaceId, and the handler's unit of work is scoped to it
  payload: { propertyId: field.id("prop"), from: field.date() },
  timeoutSeconds: 30,
  retry: { limit: 5, delaySeconds: 10, maxDelaySeconds: 600 },
});
const catalog = defineCatalog([pushAri]);

// The API: a job is written in the transaction of the use case.
const queue = createJobQueue({ connectionString: appUrl, catalog, onError: log });
await database.unitOfWork.run(scope, async (capabilities) => {
  await capabilities.properties.insert(property);
  await queue.enqueue(capabilities, { name: "channex.push_ari", payload: { workspaceId, propertyId, from } });
});

// The worker: Docker runs start(); a Vercel cron route runs drain().
const runtime = createJobRuntime({
  connectionString: workerUrl,
  catalog,
  unitOfWork: workerDatabase.unitOfWork,
  onError: log,
  handlers: {
    "channex.push_ari": async (payload, context) => {
      const property = await context.run(({ properties }) => properties.get(payload.propertyId));
      // … network calls happen here, outside the unit of work
    },
  },
  schedules: [{ job: "channex.push_ari", key: "hotel-a", cron: "0 3 * * *", timeZone: "Europe/Rome", payload }],
});
await runtime.start(); // and on SIGTERM: await runtime.stop({ timeoutMs: 30_000 })
const report = await runtime.drain({ budgetMs: 50_000 });

What a job may carry

A payload carries identifiers, never personal data. Its fields are built with field only:

• field.id(prefix): a TypeID with that prefix;
• field.oneOf([...]): one value of a closed set;
• field.integer({ min, max }): an integer;
• field.date(): a calendar date.

There is no text field, and unknown keys are refused. The handler reads everything else from the database, in context.run.

How jobs run

• At least once. A job can run twice, for example after a crash, so handlers must be idempotent.
• Outcomes follow the error model (ADR 0008):
  • an InfrastructureError that is retryable, or whose outcome is unknown, is retried with exponential backoff, up to the retry limit its queue had when the job was enqueued (a later catalog change does not alter it), then sent to the dead_letter queue;
  • an attempt that outlives its timeout, or that a stopping worker gives up on, is retried in the same way, even if its handler returns afterwards (jobs.interrupted): stop when context.signal aborts;
  • anything else goes there at once.
• Dead jobs. deadLetters() lists them, with stable safe failure codes (such as jobs.unexpected_failure, jobs.infrastructure_failure and jobs.invalid_payload; see the failure diagnostics under Tests) and the retry flag; arbitrary names, messages and causes are discarded; redrive() sends them back.
• Timeouts. A job still active after its timeout, as when its worker crashed, is failed by the recovery pass and retried. The pass runs every recoveryIntervalSeconds on the resident worker, and once at the start of a drain.
• The drain runs only the jobs whose timeout fits in its budget. It stops taking jobs in time to return within the budget, and returns early when nothing is due.
• A request pump passes idleWaitMs to keep polling through delayed coalescing and short continuations, while retaining the same overall budget and job timeout checks. The default remains zero for callers that want to return as soon as due work is exhausted.
• Poolers. Everything works through a transaction pooler: workers poll, and nothing uses LISTEN/NOTIFY or session state.

Tests

pnpm test runs the unit tests of the catalog, the payloads, the outcomes and the registration. pnpm test:integration starts PostgreSQL 17 and PgBouncer in transaction mode (@guestcraft/testing), migrates a template as a deploy does, and covers:

• enqueue with rollback and commit;
• the tenant context of handlers;
• retries with backoff, then the dead letter queue;
• a worker killed with SIGKILL in the middle of a job;
• a cron schedule that fires;
• the budget of a drain;
• everything again through the pooler;
• the privileges of the app role.

The production channexJobs catalog defines feed-drain and safety jobs. JobQueue.ready() prepares queue metadata before taking a transaction connection; this avoids cold-start deadlocks through a one-connection transaction pooler. Enqueueing still joins the unit of work.

The Channex catalog registers the workspace jobs channex.push_ari (one property per job), channex.drain (the booking feed) and channex.messages_drain, and the system jobs channex.ari_safety (every minute) and channex.safety (every five minutes). Transactional writes and durable continuations wake the same publisher in resident and pump execution; PostgreSQL property leases/pacing coordinate replicas.

Failure diagnostics use a closed set determined by error classes, never error.name, messages or causes. jobs.infrastructure_failure identifies InfrastructureError and retains its retry/outcome-unknown policy. jobs.invariant_failure, jobs.type_failure, jobs.range_failure, jobs.syntax_failure and jobs.reference_failure identify respectively InvariantError, TypeError, RangeError, SyntaxError and ReferenceError; these go to dead letters immediately. jobs.unexpected_failure covers every other thrown value and jobs.invalid_payload means the payload failed its catalog schema without running a handler. Handler-failure callbacks contain job identifiers and these fixed codes, never handler messages, names or causes. An aborted handler reports jobs.interrupted to that callback. The deployed worker logs a fixed generic failure message.

Migration scheduler_safe_output normalizes both output and source_output before every job insert or update, including pg-boss’s own atomic retry/dead-letter writes. It also scrubs retained outputs. The operator’s stored report identifies timeout and crash recovery as jobs.timed_out, forced shutdown as jobs.interrupted, heartbeat expiry as jobs.heartbeat_lost, and a conflicting redrive as jobs.redrive_conflict. Unrecognized queue output becomes jobs.queue_failure. Known GuestCraft reports keep their code, class and retry flag; extra fields are discarded. No library message, stack or cause is retained. Completed jobs store no output; if a future completion result writes one, the trigger rewrites it to jobs.queue_failure because these columns are reserved for fixed failure reports. The retry flag describes the failure’s policy; the job’s stored retry count/limit and state still decide whether another attempt remains. deadLetters() returns the normalized report; there is currently no public API or UI for dead letters. Adding a handler failure code or upgrading pg-boss requires reviewing the SQL normalizer and its PostgreSQL regressions.

The internal webhookJobs catalog (ADR 0040), exported only through allJobs, adds the system job webhooks.dispatch, scheduled every minute, and the workspace job webhooks.deliver, which fans the webhook outbox out and sends due deliveries for about a minute. allJobs combines both catalogs for the worker and the migration step.

The privacyJobs catalog (ADR 0052) adds the system job privacy.dispatch, scheduled every hour, which claims each workspace’s due retention run; the workspace job privacy.retention, one bounded retention pass that queues the next while a batch is full; and the workspace job privacy.export, which prepares one access export named by its request (requestId). Like every payload, theirs carry identifiers only: a test proves that no field of any catalog job accepts an e-mail address or a name.

Audit automation and cash drawers

Night audit settings, configured no-shows and expired tentative options share the durable property-date audit workflow. Cash drawer shifts provide currency-aware opening float, server-linked cash activity and versioned counted closure. See the night audit contract and cash shift contract for public operations, permissions, recovery, persistence and the Vercel daily-trigger cadence limit.