The queue behind our own ports

Status: released in v0.1.76 (2026-09-27) — design and as-built record in docs/design/event-bus-outbox-webhooks.md in the StandMeet repo.

The upper layers (domains, the admin panel, MCP) depend only on our own interfaces: events.Recorder, and jobs.Jobs, jobs.Inspector and jobs.Runtime in internal/infra/jobs. River is the current implementation behind the job ports; a later adapter would not change any upper-layer code. The transport is swappable. The outbox is not: whatever runs underneath, an event is first written to the outbox in the same transaction as the domain change.

Interface	Methods	Implementations
events.Recorder (fixed)	With(tx) Recorder, Record(ctx, ownerID, type, subject, data) error	the outbox writer (Postgres), the only one
jobs.Jobs	With(tx) Jobs, Enqueue(ctx, kind, args, EnqueueOpts) (JobID, error)	River adapter
jobs.Inspector	Overview(kind), List(filter), Get, Retry, Cancel, Periodic, RunPeriodic	River adapter
jobs.Runtime	Jobs + Inspector + Wait(ctx, id, max), Start, Stop	River adapter
jobs.Handler	func(ctx, args json.RawMessage) error, idempotent	job kinds and subscriptions

• There is no separate Transport interface. Subscriptions are data (events.Subscription) that the bus turns into job kinds; the relay enqueues through jobs.Jobs. Another transport would replace the relay inside internal/infra/events; no domain code would change.
• Job kinds (jobs.Kind) and periodic jobs (jobs.Periodic{Name, Every, Run}) are data, handed to the runtime at construction. There is no Periodic(...) registration method.
• EnqueueOpts has RunAt (schedule later) and UniqueByArgs (one job per args, used by webhook fan-out).
• States: pending, running, retryable, completed, discarded, cancelled.

classDiagram
  direction TB
  class Recorder {
    «interface · never changes»
    +With(tx) Recorder
    +Record(ctx, ownerID, type, subject, data) error
  }
  class Jobs {
    «interface · swappable»
    +With(tx) Jobs
    +Enqueue(ctx, kind, args, opts) JobID
  }
  class Inspector {
    «interface · swappable»
    +Overview(kind) / List(filter) / Get(id)
    +Retry(id) / Cancel(id) / Periodic() / RunPeriodic(name)
  }
  class Runtime {
    «interface · swappable»
    +Wait(ctx, id, max) State
    +Start(ctx) / Stop(ctx)
  }
  class Bus {
    «infra/events · relay loop»
    +Kinds() one per Subscription
    +Periodics() retention and sweep
  }
  class OutboxRecorder {
    «Postgres · only implementation»
    INSERT events in caller tx
  }
  class RiverAdapter {
    «infra/jobs/river · now»
  }
  class Domains {
    «upper layer»
    corpus / access / owner …
  }
  class TasksPanel {
    «upper layer»
    admin · MCP tasks.*
  }
  Recorder <|.. OutboxRecorder
  Runtime --|> Jobs
  Runtime --|> Inspector
  Runtime <|.. RiverAdapter
  Domains ..> Recorder
  Domains ..> Jobs : Enqueue
  Domains ..> Bus : Subscription data
  Bus ..> Jobs : relay enqueues
  TasksPanel ..> Inspector

The rules that keep it swappable

• Strings and JSON only. type and kind are strings; data and args are JSON. On River, kinds stay strings through one raw-JSON args type whose KindAliases() lists every declared kind (internal/infra/jobs/river/worker.go). River's generic arg types never leak upward.
• The contract does not change with the implementation: at-least-once delivery, idempotent handlers, no ordering across events.
• A gate keeps River behind the port. check-queue-behind-port.sh fails when github.com/riverqueue/** is imported outside internal/infra/jobs/river (tests included). No exclusion list.
• A conformance suite states the contract. About 27 UTs cover the Jobs / Inspector / Runtime contract. They run against River now and against any future implementation unchanged. The suite plus a green full e2e acceptance is the proof that upper layers do not notice.

Where it lives

internal/infra/jobs holds the ports, the queue table, the failure classes and DefaultBackoff. internal/infra/jobs/river is the adapter: River's migrator (jobsriver.Migrate, run at boot right after pgstore.Migrate), the worker, periodic jobs, the inspector and Wait. Neither knows a domain. internal/infra/events holds the Recorder, the relay, retention and webhook signing (event-model).

River in brief

River (riverqueue.com) v0.47 is a Go job queue that runs on Postgres. It is pgx/v5 native. Its license is MPL-2.0, which is compatible with AGPL.

River feature	What we use it for
InsertTx (transactional insert)	a job commits with the change that caused it
SKIP LOCKED claims	workers on many processes never take the same job
LISTEN/NOTIFY wake-up	jobs start without waiting for a poll
retries with backoff	the single retry owner (retry-has-one-owner)
unique-by-args inserts	one webhook.deliver per (endpoint, event); not used for coalescing (relay-claims-rows-not-cursor)
leader-elected periodic jobs	the periodic jobs run once, not once per process (the relay is not one of them)
rescuer	a job stuck in running past the threshold is rescued and retried
job cleaner	completed rows go after 24 h, discarded after 7 days (River's defaults)
job rows as a queryable log	the tasks-panel reads them

• No new service: River adds a few tables (river_job, river_leader, …) to the Postgres we already run.
• River owns its DDL: its own migrator creates the tables at boot, and schema.sql does not copy them (a copy would drift on the next River upgrade).
• River UI is not adopted: it is a separate service with its own auth. The tasks-panel follows its information layout on dispatcher ops.

Why a queue on Postgres at all: why-not-a-broker.