# Background jobs

**Status:** Accepted (2026-09-14) · **Amends:** ADR-0005, ADR-0019, ADR-0023, ADR-0025, ADR-0026

## Context

Full apps send email, purge deleted accounts, enforce retention and run scheduled work. Earlier ADRs chose River on PostgreSQL (no Redis), `jobs.AsyncSender` for email (ADR-0019, ADR-0025), job metadata carrying request ID, trace, actor and org (ADR-0030), and runners with graceful shutdown (ADR-0017).

Operators also need to manage jobs the way they manage serverless functions such as AWS Lambda: the code is written and deployed by developers, but each job's configuration (whether it runs, its schedule, timeout and retries) is visible and editable in the admin panel without a redeploy, together with its recent runs and controls to run, retry, cancel or pause work.

Building `modules/jobs` surfaced details the earlier ADRs did not settle:

- River's migrations use a schema placeholder, create indexes `CONCURRENTLY` inside `DO` blocks (not allowed in a transaction), and keep their own `river_migration` history that later River releases read. They can't be copied into the app's goose history as ADR-0005 assumed.
- A worker can't be registered after a River client is built, so `jobs.AsyncSender(provider)` can't both register the worker and enqueue.
- Job arguments are stored in PostgreSQL; email jobs contain recipients and bodies, including verification codes.

## Options for job configuration

1. Configuration only in code; change it with a deploy.
2. Job configuration as runtime settings (ADR-0031).
3. Job definitions declared in code, with operator overrides in a dedicated jobs table and API.

Option 3: job configuration has its own shape (schedule, next run, queue, last run status) and screens, so it gets a tailored table and API instead of generic settings.

## Decision

`apistock.dev/modules/jobs` wraps River (v0.47). River's own types (`river.JobArgs`, `river.Worker`, `river.InsertOpts`) are used directly, as pgx types are in `modules/postgres`.

### Client and lifecycle

| Topic | Decision |
|---|---|
| Construction | `jobs.New(pool, workers, opts...)`; options for queues, logger, tracer provider, retention and stop timeout |
| Runner | `Client.Run(ctx)` starts River, blocks until ctx is done, then stops gracefully: running jobs get the stop timeout (default 20 s, inside the app's 25 s shutdown) before their contexts are cancelled |
| Insert-only | A client with no queues only enqueues; `cmd/api` can enqueue while `cmd/worker` works jobs (ADR-0022) |
| Enqueue | `Client.Insert` and `Client.InsertTx(ctx, tx, …)`; `InsertTx` commits the job with the caller's transaction |

### Job definitions (Lambda-style configuration)

A **job definition** is a named, documented kind of job declared in code. Its code-level defaults can be overridden by operators at runtime.

```go
// Shape only: internal/app/job_cleanup_sessions.go, generated by
//   aps gen job CleanupSessions --cron "0 3 * * *" --timeout 5m --max-attempts 5
jobs.Define(defs, jobs.Definition[cleanupsessions.Args]{
	Name:        "cleanup_sessions",
	Description: "Deletes expired sessions.",
	Worker:      cleanupsessions.NewWorker(sessionStore),
	Schedule:    "0 3 * * *",       // empty: runs only when enqueued or run manually
	Enabled:     true,
	Timeout:     5 * time.Minute,
	MaxAttempts: 5,
	Queue:       river.QueueDefault,
	NewArgs:     func() cleanupsessions.Args { return cleanupsessions.Args{} },
})
```

| Field | Editable at runtime | Effect of a change |
|---|---|---|
| Name, description, worker code, args type | No: code, changed by deploy | — |
| `enabled` | Yes | Disabled: the schedule stops and "run now" is refused. Jobs already queued, or enqueued by application code (for example emails), still run |
| `schedule` (5-field cron, `@every 15m`, or empty) | Yes, validated; minimum interval 1 minute | The leader replaces the River periodic job within seconds |
| `timeout` | Yes, 1 s – 24 h | Applies to attempts that start after the change |
| `max_attempts` | Yes, 1 – 100 | Applies to jobs enqueued after the change |
| `queue`, `priority` | Yes; the queue must be one the workers run | Applies to jobs enqueued after the change |

| Topic | Decision |
|---|---|
| Storage | `jobs_definitions` holds only overrides (NULL column = code default), with `version`, `updated_at`, `updated_by` and a nullable `org_id` reserved; `jobs_definition_history` records every change with old and new values, reason, actor and request ID. Both ship as a goose migration from `modules/jobs` |
| Writes | One transaction: optimistic version check, override row, history row, `pg_notify('apistock_jobs', name)`; an audit event `jobs.definition.changed`. Resetting sets columns back to NULL |
| Propagation | Every instance listens and resyncs periodically (as ADR-0031); the elected leader reconciles River periodic jobs with the effective schedules |
| Unknown rows | Overrides for definitions no longer in code are kept, ignored and reported by `aps doctor` |
| Validation | Reason required for disabling a job or changing its schedule; changes need an authenticated actor |

### Admin APIs (generated `internal/modules/ops`, Full preset, v0.2)

| Menu | Endpoints |
|---|---|
| Job definitions | `GET /ops/jobs/definitions` (effective config, defaults, `modified`, next run, last run state); `GET/PUT/DELETE /ops/jobs/definitions/{name}` (update with `version` and `reason`; delete resets to code defaults); `GET /ops/jobs/definitions/{name}/history`; `POST /ops/jobs/definitions/{name}/run` (run now) |
| Scheduled jobs | `GET /ops/jobs/scheduled`: enabled definitions with a schedule and their next run times |
| Runs | `GET /ops/jobs/runs` (filter by definition, state, queue; cursor pagination); `GET /ops/jobs/runs/{id}` (attempts, errors, timings; arguments omitted); `POST /ops/jobs/runs/{id}/retry`; `POST /ops/jobs/runs/{id}/cancel`. Runs live under `/runs` because `/ops/jobs/{id}/retry` would conflict with `/ops/jobs/definitions/{name}` in Go's router |
| Queues | `GET /ops/queues`; `POST /ops/queues/{name}/pause`; `POST /ops/queues/{name}/resume` |

Permissions: `ops.jobs.read`, `ops.jobs.write` (config, pause/resume), `ops.jobs.run` (run now, retry, cancel). Every write is audited. Job arguments never appear in ops responses, because they can contain personal data.

### CLI

`aps gen job <Name> [--cron SPEC | --every DURATION] [--queue NAME] [--timeout DURATION] [--max-attempts N]` generates, one-shot (ADR-0021):

- `internal/jobs/<name>/args.go` (args type and kind), `worker.go` (worker with dependencies as constructor parameters), `worker_test.go`
- `internal/app/job_<name>.go` (the `jobs.Define` call) and one line at `//aps:anchor jobs`

### Migrations

River's tables are migrated by River's migrator: `jobs.Migrate(ctx, pool)`, called by the app's `cmd/migrate` after goose. `jobs.MigrationsPending(ctx, pool)` reports unapplied versions for readiness reports and `aps doctor`. This is the one exception to ADR-0005's single goose history; the jobs definition tables use goose normally.

### Context propagation (ADR-0030)

| Step | Behaviour |
|---|---|
| Enqueue | Middleware stores an `apistock` object in job metadata: request ID, W3C `traceparent`/`tracestate`, actor kind, ID and label, org ID. Permissions are never stored. A job enqueued by another job keeps the original actor |
| Work | Middleware restores the request ID, starts a consumer span (`job <kind>`) as a child of the enqueuing trace, and sets the context actor to `actor.System("jobs")` with the original org ID. `jobs.OnBehalfOf(ctx)` returns the original actor for audit metadata |
| Errors | Workers return errors; the client's error handler logs each failure once with job ID, kind and attempt (warning while retries remain, error on the last attempt); panics are logged with their stack |

A job never runs with the enqueuing user's permissions; anything a job does on a user's behalf is authorised when the job is enqueued.

### Email

```go
// Shape only.
workers := river.NewWorkers()
jobs.AddMailWorker(workers, resendSender) // works "apistock.mail.send" jobs
client, err := jobs.New(pool, workers, jobs.WithQueues(jobs.DefaultQueues()))
mailer := jobs.AsyncSender(client)         // mail.Sender that validates and enqueues
```

The worker sets `mail.Message.IdempotencyKey` to `job-<id>` when empty, so provider retries of the same job never send twice (ADR-0025). Invalid messages fail at enqueue. Mail delivery is an internal job, not an editable definition.

### Retention defaults

Completed jobs 1 hour, cancelled 24 hours, discarded 7 days. Job arguments can contain email bodies, so completed jobs are kept only long enough to debug. Run history beyond that lives in logs and traces.

### Tenancy

River's tables have no `org_id` column; a job's org ID lives in its metadata. The jobs definition tables have a nullable `org_id` (ADR-0023).

## Why

- Operators get Lambda-style control (enable, schedule, timeout, retries, run now) without redeploys, while code stays reviewed and versioned.
- A dedicated table fits job-specific screens (next run, last run, queues) better than generic settings.
- River's migrator is the only supported way to apply River's schema changes safely across versions.
- Propagating correlation but not permissions keeps jobs traceable without widening what a leaked job row could do.

## Trade-offs

- A second runtime-configuration mechanism next to ADR-0031, with its own tables, history and listener.
- Two migration histories (goose and `river_migration`) run by one command.
- River is pre-1.0; its API may change between minor versions (ADR-0029 threat 22).
- `robfig/cron/v3` parses schedules; it is stable but unmaintained, and River's `PeriodicSchedule` interface is the swap path.

## Consequences

- ADR-0026's jobs overview moves from v0.5 to v0.2 and grows into the admin APIs above.
- The Full preset's `cmd/migrate` runs goose migrations, then `jobs.Migrate`; `aps doctor` and `/ops/system` report both.
- Job definition names are public API (ADR-0015): renaming one orphans its overrides and history.
- Threat model row 24 covers abuse of job controls.

## v0.2 implementation notes (2026-09-14)

- Library API: `jobs.New(pool, workers, WithQueues, WithDefinitions, WithLogger, WithTracerProvider, WithPropagator, WithStopTimeout, WithRetention, …)`; `jobs.NewDefinitions`, `jobs.Define`; `jobs.NewManager(ctx, pool, client, recorder)` with `Definitions`, `Scheduled`, `Definition`, `Update(ConfigPatch, Change)`, `Reset`, `History`, `RunNow`, `Jobs(JobFilter)`, `Job`, `Retry`, `Cancel`, `Queues`, `PauseQueue`, `ResumeQueue`; `Manager.Run` is the listener runner.
- A reason is required when a change disables a job or changes the schedule of an enabled job (including a reset that does so).
- Moving a definition to another queue is accepted only if that queue is active (some worker runs it), so jobs can't be sent to a queue nobody works.
- The effective queue, priority and max attempts are applied by insert middleware to every job of a defined kind, whoever enqueues it; the timeout is read when each attempt starts.
- Setting a field to its code default removes that override; a change that alters nothing creates no version, history row or audit event.
- `NextRunAt` is computed from the schedule and is approximate: River's leader keeps the real timer in memory and restarts it on leader change.
- Audit actions: `jobs.definition.changed`, `jobs.definition.run_requested`, `jobs.run.retried`, `jobs.run.cancelled`, `jobs.queue.paused`, `jobs.queue.resumed`.
- The `/ops/jobs/*` and `/ops/queues` endpoints are implemented in `examples/full-single` (`internal/modules/ops`), protected by the interim ops token (ADR-0034). Error codes: `job_definition_not_found` (404), `job_definition_version_conflict` (409), `job_reason_required` (422), `invalid_job_config` (422), `job_definition_disabled` (409), `job_not_found` (404), `queue_not_active` (422), `invalid_cursor` (400), `invalid_job_state` (422).
- `aps gen job` is implemented (ADR-0035): interactive prompts or flags, generating the `internal/jobs/<name>/` and `internal/app/job_<name>.go` layout; a golden test checks it reproduces `examples/full-single`'s heartbeat job exactly. See [the background jobs guide](/technical/background-jobs/) and [the CLI guide](/cli/).