Queues

Background work is River on the same PostgreSQL database. The web process only inserts jobs; cmd/queue runs the workers. Both processes use pkg/storage, and the wiring is explicit in internal/runtime.

Jobs and workers

Job arguments live in queue/jobs/. Keep them small and give each one a stable Kind; add JSON tags when the wire key should differ from the Go field name:

1package jobs
2
3import "uuid"
4
5type SendVerificationEmailArgs struct {
6    UserID uuid.UUID
7}
8
9func (SendVerificationEmailArgs) Kind() string { return "send_verification_email" }

The worker lives in queue/ and is registered explicitly. Workers call a service or a domain receiver; they do not issue SQL:

 1type SendVerificationEmailWorker struct {
 2    river.WorkerDefaults[jobs.SendVerificationEmailArgs]
 3    identity services.Identity
 4    sender   email.TransactionalSender
 5}
 6
 7func NewSendVerificationEmailWorker(
 8    identity services.Identity,
 9    sender email.TransactionalSender,
10) *SendVerificationEmailWorker {
11    return &SendVerificationEmailWorker{identity: identity, sender: sender}
12}
13
14func (w *SendVerificationEmailWorker) Work(
15    ctx context.Context,
16    job *river.Job[jobs.SendVerificationEmailArgs],
17) error {
18    ctx, span := telemetry.Start(ctx, "queue.send_verification_email", "user.id", job.Args.UserID)
19    defer span.End()
20
21    data, err := w.identity.VerificationEmail(ctx, job.Args.UserID)
22    if err != nil {
23        if errors.Is(err, services.ErrNoEmailToSend) {
24            return nil
25        }
26
27        return telemetry.Fail(ctx, err)
28    }
29
30    return storage.SendTransactional(ctx, data, w.sender)
31}

 1func NewWorkers(
 2    identity services.Identity,
 3    tokens domains.Tokens,
 4    transactional email.TransactionalSender,
 5    marketing email.MarketingSender,
 6) *river.Workers {
 7    workers := river.NewWorkers()
 8    storage.AddEmailWorkers(workers, transactional, marketing)
 9    river.AddWorker(workers, NewSendVerificationEmailWorker(identity, transactional))
10    river.AddWorker(workers, NewSendPasswordResetEmailWorker(identity, transactional))
11    river.AddWorker(workers, NewDeleteExpiredTokensWorker(tokens))
12
13    return workers
14}

andurel inspect jobs --json lists job and worker files, and andurel patterns show job is the reference.

Enqueue from a service

Handlers do not insert jobs; services do, through a storage.InsertQueue. When the job depends on data written in the same use case, enqueue on the bound queue inside storage.AtomicOps (here data is the use case's input):

 1err := storage.AtomicOps(ctx, i.transactor, func(ctx context.Context, tx storage.Tx) error {
 2    users, queue := i.users.InTx(tx), i.queue.InTx(tx)
 3
 4    created, err := users.Create(ctx, i.pepper, domains.CreateUserData{
 5        Email: data.Email,
 6        PasswordPair: domains.PasswordPair{
 7            Password:        data.Password,
 8            ConfirmPassword: data.ConfirmPassword,
 9        },
10    })
11    if err != nil {
12        return fmt.Errorf("create user: %w", err)
13    }
14
15    if _, err := queue.Insert(ctx, jobs.SendVerificationEmailArgs{UserID: created.ID}, nil); err != nil {
16        return fmt.Errorf("queue verification email: %w", err)
17    }
18
19    return nil
20})

Returning nil commits both the row and the job; an error rolls back both. Pass *river.InsertOpts as the third argument to control MaxAttempts, Queue, or Priority.

Job arguments persist in river_job for the retention period. Never enqueue a secret: send a user id and let the worker mint codes, tokens, or signed links at send time.

Recurring work

A periodic job reuses the same args and worker types. Return each schedule from queue/PeriodicJobs(); runtime.NewQueue passes the list to storage.WithRiverPeriodicJobs:

 1func PeriodicJobs() []*river.PeriodicJob {
 2    return []*river.PeriodicJob{
 3        river.NewPeriodicJob(
 4            river.PeriodicInterval(time.Hour),
 5            func() (river.JobArgs, *river.InsertOpts) {
 6                return jobs.DeleteExpiredTokensArgs{}, nil
 7            },
 8            &river.PeriodicJobOpts{RunOnStart: true},
 9        ),
10    }
11}

The scaffold's hourly job calls domains.Tokens.DestroyExpired; returning nil args from the constructor skips an insertion for that tick.

Two processes

runtime.New builds the web graph with an insert-only queue:

1queueInsert, err := storage.NewQueueInsert(db, cfg.QueueInsert.Config)

runtime.NewQueue builds the worker graph with the processor, workers, periodic jobs, telemetry, and email senders:

1processor, err := storage.NewQueueProcessor(
2    db,
3    cfg.QueueWorker.Config,
4    storage.WithRiverWorkers(queue.NewWorkers(identity, tokens, transactional, marketing)),
5    storage.WithRiverPeriodicJobs(queue.PeriodicJobs()...),
6    storage.WithRiverLogger(tel.Logger()),
7)

Deploy cmd/app and cmd/queue as separate processes. Each owns its lifecycle; Queue.Shutdown stops the processor before closing the database and telemetry.

Configuration

config.NewQueueInsert reads the settings needed to enqueue, and config.NewQueueWorker clones that base and adds worker policy:

QUEUE_MAX_ATTEMPTS=25
QUEUE_SCHEMA=river
QUEUE_MAX_WORKERS=10
QUEUE_POLL_INTERVAL=1s
QUEUE_JOB_TIMEOUT=1m
QUEUE_SOFT_STOP_TIMEOUT=30s

Unset keys keep the storage.DefaultQueueConfig values. Invalid settings fail during startup because QueueConfig.Validate runs in the config constructor.

Workers may be retried, so keep them idempotent: load current records inside the worker, keep arguments small, and return classified errors so River can apply the right retry policy.