# modules/jobs

```go
import "apistock.dev/modules/jobs"
```

Package jobs runs background jobs on PostgreSQL with River (ADR-0033).

Jobs are declared as definitions: code written and deployed by developers, with configuration (enabled, schedule, timeout, retries, queue) that operators change at runtime through a [Manager](#Manager), like serverless functions:

	defs := jobs.NewDefinitions()
	jobs.Define(defs, jobs.Definition[cleanupsessions.Args]{
		Name:     "cleanup_sessions",
		Worker:   cleanupsessions.NewWorker(store),
		NewArgs:  func() cleanupsessions.Args { return cleanupsessions.Args{} },
		Enabled:  true,
		Schedule: "0 3 * * *",
	})
	workers := river.NewWorkers()
	_ = jobs.AddMailWorker(workers, resendSender)
	client, err := jobs.New(pool, workers,
		jobs.WithQueues(jobs.DefaultQueues()), jobs.WithDefinitions(defs))

The client is an app.Runner. Jobs enqueued from a request carry its request ID, trace and actor (but never its permissions); workers see them in their context (ADR-0030). River's tables are created by [Migrate](#Migrate).

Stability: pre-1.0 (ADR-0015).

## Constants

### const MinScheduleInterval, …

```go
const (
	MinScheduleInterval = time.Minute
	MinTimeout          = time.Second
	MaxTimeout          = 24 * time.Hour
	MaxAttemptsLimit    = 100
)
```

Bounds for job definition configuration (ADR-0033, threat 24).

### const DefaultDefinitionTimeout, …

```go
const (
	DefaultDefinitionTimeout     = time.Minute
	DefaultDefinitionMaxAttempts = 25
)
```

Defaults for definition fields left zero.

### const DefaultStopTimeout, …

```go
const (
	DefaultStopTimeout        = 20 * time.Second
	DefaultCompletedRetention = time.Hour
	DefaultCancelledRetention = 24 * time.Hour
	DefaultDiscardedRetention = 7 * 24 * time.Hour
)
```

Defaults applied by [New](#New).

### const DefaultResyncInterval

```go
const DefaultResyncInterval = 5 * time.Minute
```

DefaultResyncInterval is how often a running [Manager](#Manager) reloads every override, covering notifications lost while disconnected.

### const MailKind

```go
const MailKind = "apistock.mail.send"
```

MailKind is the job kind that delivers queued email.

## Variables

### var ErrUnknownDefinition, …

```go
var (
	// ErrUnknownDefinition reports a name that no definition declares.
	ErrUnknownDefinition = errors.New("jobs: unknown job definition")

	// ErrVersionConflict reports that the definition changed after the
	// caller read it. Read it again and retry.
	ErrVersionConflict = errors.New("jobs: job definition changed since it was read")

	// ErrReasonRequired reports disabling or rescheduling a job without a
	// reason.
	ErrReasonRequired = errors.New("jobs: a reason is required to disable or reschedule a job")

	// ErrActorRequired reports a change without an authenticated actor in
	// the context.
	ErrActorRequired = errors.New("jobs: changes require an authenticated actor")

	// ErrDefinitionDisabled reports running a disabled job on demand.
	ErrDefinitionDisabled = errors.New("jobs: job definition is disabled")

	// ErrJobNotFound reports a job ID that doesn't exist, for example
	// because retention removed it.
	ErrJobNotFound = errors.New("jobs: job not found")

	// ErrUnknownQueue reports a queue no worker runs.
	ErrUnknownQueue = errors.New("jobs: queue is not active")

	// ErrInvalidConfig reports configuration outside the allowed bounds.
	// The error is an [*InvalidConfigError].
	ErrInvalidConfig = errors.New("jobs: invalid job configuration")

	// ErrInvalidCursor reports a malformed pagination cursor.
	ErrInvalidCursor = errors.New("jobs: invalid cursor")
)
```

Errors returned by [Manager](#Manager) methods. Check them with [errors.Is](https://pkg.go.dev/errors#Is).

### var Migrations

```go
var Migrations fs.FS = mustSub(migrationFiles, "migrations")
```

Migrations holds the goose migrations for job definition overrides and their history. Apps copy them into db/migrations; River's own tables come from [Migrate](#Migrate).

## Functions

### func AddMailWorker

```go
func AddMailWorker(workers *river.Workers, sender mail.Sender) error
```

AddMailWorker registers the worker that delivers queued email through sender, such as a Resend or SMTP provider. Register it on the workers of the client that works jobs.

### func AsyncSender

```go
func AsyncSender(client *Client) mail.Sender
```

AsyncSender returns a [mail.Sender](/reference/mail/#Sender) that validates each message and queues it for the mail worker, returning once the job is stored. Delivery is retried up to 8 times, and each job's ID becomes the provider idempotency key, so a retry never sends twice (ADR-0025).

### func DefaultQueues

```go
func DefaultQueues() map[string]river.QueueConfig
```

DefaultQueues returns the default queue with 10 workers per instance.

### func Define

```go
func Define[T river.JobArgs](defs *Definitions, d Definition[T])
```

Define adds a job definition. Invalid definitions are programming errors found at startup, so Define panics.

### func Migrate

```go
func Migrate(ctx context.Context, pool *pgxpool.Pool) ([]int, error)
```

Migrate applies River's pending schema migrations with River's own migrator and returns the versions applied. Run it from the app's migrate command after goose migrations (ADR-0033); never at startup (ADR-0017).

### func MigrationsPending

```go
func MigrationsPending(ctx context.Context, pool *pgxpool.Pool) ([]string, error)
```

MigrationsPending describes River migrations not yet applied, empty when the schema is current. It changes nothing.

### func OnBehalfOf

```go
func OnBehalfOf(ctx context.Context) (actor.Actor, bool)
```

OnBehalfOf returns the actor who enqueued the job running with ctx. Inside a job the context actor is actor.System("jobs"); record this actor in audit metadata. It has no permissions: authorise work when enqueuing.

## Types

### type AttemptError

```go
type AttemptError struct {
	At      time.Time
	Attempt int
	Message string
}
```

AttemptError is a failed attempt. Stack traces are omitted; they are in the logs.

### type Change

```go
type Change struct {
	// Version is the DefinitionView.Version the caller last read.
	Version int64
	// Reason explains the change; required to disable or reschedule a job.
	Reason string
}
```

Change carries what a caller supplies with a change. The actor comes from the context.

### type Client

```go
type Client struct {
	// contains filtered or unexported fields
}
```

Client enqueues and works jobs. It is safe for concurrent use.

#### func New

```go
func New(pool *pgxpool.Pool, workers *river.Workers, opts ...Option) (*Client, error)
```

New builds a client on pool. workers may be nil for an insert-only client or when every job comes from definitions.

#### func (*Client) Insert

```go
func (c *Client) Insert(ctx context.Context, args river.JobArgs, opts *river.InsertOpts) (*rivertype.JobInsertResult, error)
```

Insert enqueues a job. The job carries ctx's request ID, trace and actor.

#### func (*Client) InsertTx

```go
func (c *Client) InsertTx(ctx context.Context, tx pgx.Tx, args river.JobArgs, opts *river.InsertOpts) (*rivertype.JobInsertResult, error)
```

InsertTx enqueues a job in tx: it becomes visible to workers only if tx commits.

#### func (*Client) River

```go
func (c *Client) River() *river.Client[pgx.Tx]
```

River returns the underlying River client for advanced use.

#### func (*Client) Run

```go
func (c *Client) Run(ctx context.Context) error
```

Run works jobs until ctx is done, then stops: no new jobs are fetched, and running jobs get the stop timeout before their contexts are cancelled. It returns nil after a clean stop. An insert-only client just waits for ctx.

### type Config

```go
type Config struct {
	// Enabled jobs run on their schedule and can be run on demand. Jobs
	// enqueued by application code run either way.
	Enabled bool
	// Schedule is a 5-field cron expression evaluated in UTC ("0 3 * * *"),
	// a descriptor ("@daily", "@every 15m"), or empty for on-demand jobs.
	Schedule string
	// Timeout bounds each attempt.
	Timeout time.Duration
	// MaxAttempts is the number of attempts before the job is discarded.
	MaxAttempts int
	// Queue is the queue new jobs go to; workers must run it.
	Queue string
	// Priority orders jobs within a queue, 1 (highest) to 4.
	Priority int
}
```

Config is a job definition's operator-editable configuration.

#### func (Config) Validate

```go
func (c Config) Validate() error
```

Validate reports whether c is within the allowed bounds.

### type ConfigPatch

```go
type ConfigPatch struct {
	Enabled     *bool
	Schedule    *string
	Timeout     *time.Duration
	MaxAttempts *int
	Queue       *string
	Priority    *int
}
```

ConfigPatch changes selected fields; nil fields keep their current value. Setting a field to its code default removes the override for that field.

### type Definition

```go
type Definition[T river.JobArgs] struct {
	// Name identifies the job; T's Kind() must return it. Names are public
	// API: renaming one orphans its overrides and history.
	Name        string
	Description string
	// Worker runs the job. Its Timeout method is ignored: the definition's
	// timeout applies.
	Worker river.Worker[T]
	// NewArgs builds the arguments for scheduled and on-demand runs.
	NewArgs func() T

	Enabled     bool
	Schedule    string
	Timeout     time.Duration
	MaxAttempts int
	Queue       string
	Priority    int
}
```

A Definition declares a named job whose configuration operators can change at runtime, like a serverless function: the code is deployed, the settings are edited in the admin panel. Zero Timeout, MaxAttempts, Queue and Priority take the defaults.

### type DefinitionChange

```go
type DefinitionChange struct {
	ID        int64
	Name      string
	Action    string
	OldConfig json.RawMessage
	NewConfig json.RawMessage
	Version   int64
	Reason    string
	ActorKind string
	ActorID   string
	RequestID string
	ChangedAt time.Time
}
```

DefinitionChange is one change to a job definition. OldConfig and NewConfig are the overridden fields as JSON objects; {} means the code defaults.

### type DefinitionView

```go
type DefinitionView struct {
	Name        string
	Description string
	// Config is the effective configuration; Defaults is the code's.
	Config   Config
	Defaults Config
	// Modified reports a valid override; InvalidOverride one that no longer
	// passes validation, so the defaults apply.
	Modified        bool
	InvalidOverride bool
	// Version increases with every change; pass it back to change the
	// definition. It is 0 for a definition never changed.
	Version   int64
	UpdatedAt time.Time
	UpdatedBy string
	// NextRunAt is the next scheduled time, approximately; zero when the job
	// is disabled or has no schedule.
	NextRunAt time.Time
	// LastRun is the most recent job of this definition, if any.
	LastRun *JobRun
}
```

DefinitionView is a job definition's configuration and status, for the admin panel.

### type Definitions

```go
type Definitions struct {
	// contains filtered or unexported fields
}
```

Definitions holds job definitions and the live overrides applied to them. Declare every definition before building the client. It is safe for concurrent use.

#### func NewDefinitions

```go
func NewDefinitions() *Definitions
```

NewDefinitions returns an empty set of definitions.

### type InvalidConfigError

```go
type InvalidConfigError struct {
	Name   string
	Reason string
}
```

InvalidConfigError describes why a configuration change was rejected.

#### func (*InvalidConfigError) Error

```go
func (e *InvalidConfigError) Error() string
```


#### func (*InvalidConfigError) Unwrap

```go
func (e *InvalidConfigError) Unwrap() error
```

Unwrap returns [ErrInvalidConfig](#ErrInvalidConfig).

### type JobFilter

```go
type JobFilter struct {
	Kind   string
	Queue  string
	States []rivertype.JobState
	// Limit is clamped to 1–100; 0 means 50.
	Limit int
	// Cursor is JobPage.NextCursor from the previous page.
	Cursor string
}
```

JobFilter selects jobs to list. Empty fields match everything.

### type JobPage

```go
type JobPage struct {
	Jobs []JobRun
	// NextCursor fetches the next page; empty on the last page.
	NextCursor string
}
```

JobPage is a page of jobs, newest first.

### type JobRun

```go
type JobRun struct {
	ID          int64
	Kind        string
	Queue       string
	State       rivertype.JobState
	Attempt     int
	MaxAttempts int
	Priority    int
	CreatedAt   time.Time
	ScheduledAt time.Time
	AttemptedAt *time.Time
	FinalizedAt *time.Time
	Errors      []AttemptError
	// RequestID and the enqueuing actor come from the job's context metadata.
	RequestID string
	ActorKind string
	ActorID   string
}
```

JobRun is one job and its attempts, without its arguments: arguments can contain personal data and are never shown in the admin panel.

### type Manager

```go
type Manager struct {
	// contains filtered or unexported fields
}
```

Manager is the admin-panel backend for jobs (ADR-0033): it stores operator overrides of job definitions, keeps every instance's definitions and schedules current, and inspects and controls runs and queues. It is an app.Runner. It is safe for concurrent use.

#### func NewManager

```go
func NewManager(ctx context.Context, pool *pgxpool.Pool, client *Client, recorder audit.Recorder, opts ...ManagerOption) (*Manager, error)
```

NewManager loads stored overrides for client's definitions and returns the manager. Changes are recorded as audit events through recorder. The tables come from [Migrations](#Migrations).

#### func (*Manager) Cancel

```go
func (m *Manager) Cancel(ctx context.Context, id int64) (JobRun, error)
```

Cancel cancels a job: queued jobs never run, and a running job's context is cancelled. It returns [ErrJobNotFound](#ErrJobNotFound) or [ErrActorRequired](#ErrActorRequired).

#### func (*Manager) Definition

```go
func (m *Manager) Definition(ctx context.Context, name string) (DefinitionView, error)
```

Definition returns one definition, or [ErrUnknownDefinition](#ErrUnknownDefinition).

#### func (*Manager) Definitions

```go
func (m *Manager) Definitions(ctx context.Context) ([]DefinitionView, error)
```

Definitions returns every definition in declaration order.

#### func (*Manager) DeleteHistoryBefore

```go
func (m *Manager) DeleteHistoryBefore(ctx context.Context, before time.Time, limit int) (int64, error)
```

DeleteHistoryBefore deletes up to limit job configuration changes made before before, oldest first, and returns how many it deleted. Retention calls it until it deletes fewer than limit (ADR-0051).

#### func (*Manager) History

```go
func (m *Manager) History(ctx context.Context, name string, before int64, limit int) ([]DefinitionChange, error)
```

History returns changes to name, newest first. For the next page, pass the last entry's ID as before (0 starts from the newest). limit is clamped to 1–100.

#### func (*Manager) Job

```go
func (m *Manager) Job(ctx context.Context, id int64) (JobRun, error)
```

Job returns one job, or [ErrJobNotFound](#ErrJobNotFound).

#### func (*Manager) Jobs

```go
func (m *Manager) Jobs(ctx context.Context, f JobFilter) (JobPage, error)
```

Jobs lists jobs, newest first. It returns [ErrInvalidCursor](#ErrInvalidCursor) for a malformed cursor.

#### func (*Manager) OldestHistory

```go
func (m *Manager) OldestHistory(ctx context.Context) (oldest time.Time, ok bool, err error)
```

OldestHistory returns when the oldest recorded configuration change was made; ok is false when there are none.

#### func (*Manager) Overview

```go
func (m *Manager) Overview(ctx context.Context) (Overview, error)
```

Overview returns the job overview.

#### func (*Manager) PauseQueue

```go
func (m *Manager) PauseQueue(ctx context.Context, name string) error
```

PauseQueue stops every instance fetching jobs from name. Running jobs finish. It returns [ErrUnknownQueue](#ErrUnknownQueue) or [ErrActorRequired](#ErrActorRequired).

#### func (*Manager) Queues

```go
func (m *Manager) Queues(ctx context.Context) ([]Queue, error)
```

Queues lists active queues by name.

#### func (*Manager) Reload

```go
func (m *Manager) Reload(ctx context.Context) error
```

Reload reads every override from the database and reapplies schedules.

#### func (*Manager) Reset

```go
func (m *Manager) Reset(ctx context.Context, name string, change Change) (DefinitionView, error)
```

Reset returns name to its code defaults. It returns the same errors as [Manager.Update](#Manager.Update).

#### func (*Manager) ResumeQueue

```go
func (m *Manager) ResumeQueue(ctx context.Context, name string) error
```

ResumeQueue resumes a paused queue on every instance.

#### func (*Manager) Retry

```go
func (m *Manager) Retry(ctx context.Context, id int64) (JobRun, error)
```

Retry makes a job available to run again immediately. Running jobs are left alone. It returns [ErrJobNotFound](#ErrJobNotFound) or [ErrActorRequired](#ErrActorRequired).

#### func (*Manager) Run

```go
func (m *Manager) Run(ctx context.Context) error
```

Run keeps job definitions and schedules current until ctx is done, then returns nil. It listens on a dedicated connection for changes committed by any instance, reloads everything after connecting, and reloads everything every resync interval as a fallback. Connection failures are logged and retried with backoff.

#### func (*Manager) RunNow

```go
func (m *Manager) RunNow(ctx context.Context, name string) (JobRun, error)
```

RunNow enqueues a job for an enabled definition immediately, with its current configuration. It returns [ErrUnknownDefinition](#ErrUnknownDefinition), [ErrDefinitionDisabled](#ErrDefinitionDisabled) or [ErrActorRequired](#ErrActorRequired).

#### func (*Manager) Scheduled

```go
func (m *Manager) Scheduled(ctx context.Context) ([]DefinitionView, error)
```

Scheduled returns enabled definitions with a schedule, soonest first.

#### func (*Manager) Update

```go
func (m *Manager) Update(ctx context.Context, name string, patch ConfigPatch, change Change) (DefinitionView, error)
```

Update applies patch to name's configuration. The change reaches every instance within moments; a new schedule takes effect on the leader. It returns [ErrUnknownDefinition](#ErrUnknownDefinition), an [\*InvalidConfigError](#InvalidConfigError), [ErrUnknownQueue](#ErrUnknownQueue), [ErrReasonRequired](#ErrReasonRequired), [ErrActorRequired](#ErrActorRequired) or [ErrVersionConflict](#ErrVersionConflict).

### type ManagerOption

```go
type ManagerOption interface {
	// contains filtered or unexported methods
}
```

A ManagerOption configures [NewManager](#NewManager).

#### func WithManagerLogger

```go
func WithManagerLogger(logger *slog.Logger) ManagerOption
```

WithManagerLogger sets the logger for listener and invalid-override warnings. Default: discard.

#### func WithResyncInterval

```go
func WithResyncInterval(d time.Duration) ManagerOption
```

WithResyncInterval sets how often [Manager.Run](#Manager.Run) reloads every override. Default: [DefaultResyncInterval](#DefaultResyncInterval).

### type Option

```go
type Option interface {
	// contains filtered or unexported methods
}
```

An Option configures [New](#New).

#### func WithDefinitions

```go
func WithDefinitions(defs *Definitions) Option
```

WithDefinitions registers job definitions: their workers, their live configuration and, on working clients, their schedules. Use the same definitions for API and worker processes.

#### func WithJobTimeout

```go
func WithJobTimeout(d time.Duration) Option
```

WithJobTimeout sets the time a job without a definition may run. Default: River's, one minute.

#### func WithLogger

```go
func WithLogger(logger *slog.Logger) Option
```

WithLogger sets the logger for job failures and River's own messages. Default: discard.

#### func WithMaxAttempts

```go
func WithMaxAttempts(n int) Option
```

WithMaxAttempts sets the attempts for jobs without a definition before they are discarded. Default: River's, 25.

#### func WithPropagator

```go
func WithPropagator(p propagation.TextMapPropagator) Option
```

WithPropagator sets how trace context is stored in job metadata. Default: the global OpenTelemetry propagator, which telemetry.Setup configures.

#### func WithQueues

```go
func WithQueues(queues map[string]river.QueueConfig) Option
```

WithQueues makes the client work jobs from queues. Without it the client only enqueues, as an API process does when a separate worker process runs jobs.

#### func WithRetention

```go
func WithRetention(completed, cancelled, discarded time.Duration) Option
```

WithRetention sets how long finished jobs are kept. -1 keeps them forever. Job arguments may contain personal data such as email bodies, so keep completed jobs briefly. Defaults: [DefaultCompletedRetention](#DefaultCompletedRetention), [DefaultCancelledRetention](#DefaultCancelledRetention), [DefaultDiscardedRetention](#DefaultDiscardedRetention).

#### func WithStopTimeout

```go
func WithStopTimeout(d time.Duration) Option
```

WithStopTimeout sets how long running jobs may continue after shutdown starts before their contexts are cancelled. Keep it below the app's shutdown timeout. Default: [DefaultStopTimeout](#DefaultStopTimeout).

#### func WithTracerProvider

```go
func WithTracerProvider(tp trace.TracerProvider) Option
```

WithTracerProvider sets the provider for job spans. Default: the global OpenTelemetry provider. Trace context is propagated with the global propagator.

### type Overview

```go
type Overview struct {
	// Queues are the queues with active workers or unfinished jobs, by name.
	Queues []QueueOverview
	// Failing are the definitions whose most recent run failed: retrying or
	// discarded.
	Failing []DefinitionView
}
```

Overview summarises job work across every instance, for GET /ops/jobs/overview (ADR-0051).

### type Queue

```go
type Queue struct {
	Name      string
	Paused    bool
	PausedAt  *time.Time
	CreatedAt time.Time
	UpdatedAt time.Time
}
```

Queue is a queue workers run or recently ran.

### type QueueOverview

```go
type QueueOverview struct {
	Name   string
	Paused bool
	// Active reports that some instance runs workers for the queue.
	Active                                   bool
	Available, Scheduled, Running, Retryable int64
	// DiscardedLastDay counts jobs that ran out of attempts in the last 24
	// hours.
	DiscardedLastDay int64
}
```

QueueOverview counts one queue's unfinished jobs.