Documentation
¶
Overview ¶
Package jobs is the single River composition root for e2a's background work. All durable background jobs — outbound sends, webhook delivery, HITL hold resolution, sender-identity provisioning, janitors — run on ONE shared river.Client against ONE river_job table, instead of a bespoke Postgres queue per subsystem. Each domain contributes its workers as a Registrar; queues are central (queues.go); business code enqueues through the Enqueuer interface so it stays testable and River lives at the edges.
This replaces the hand-rolled claim/lease/retry/sweep machinery: River's SKIP LOCKED claim, per-job MaxAttempts + retry policy, and built-in reaper do the plumbing, so domains only write job args + a Work() handler.
Index ¶
Constants ¶
const ( // QueueOutbound carries outbound send jobs (API → SES). QueueOutbound = "outbound" // QueueInbound carries inbound message-processing jobs (accepted raw MIME → // parse/screen/persist/deliver). Isolated so an inbound spike (incl. the // per-message Gemini screen) can't starve outbound sends or webhook delivery. QueueInbound = "inbound" // QueueWebhook carries customer webhook-delivery jobs. Isolated from outbound // so a slow/failing endpoint's backlog never delays sends. QueueWebhook = "webhook" // QueueMaintenance carries low-urgency periodic/janitor work (reapers, // hold-TTL resolution, auto-disable sweeps). QueueMaintenance = "maintenance" // QueueNotify carries HITL approval-notification emails (the owner's review // alert when an outbound message enters pending_review). Small, isolated pool // so a burst of held sends notifying — or a stuck notification to a bad owner // address retrying — never competes with customer outbound delivery. QueueNotify = "notify" // QueueDefault is River's built-in default — anything not explicitly routed. QueueDefault = river.QueueDefault )
Named queues. Jobs are assigned a queue at enqueue time (InsertOpts.Queue); the shared client works all of these. Separate queues give INDEPENDENT concurrency per lane, so a backlog in one (e.g. a slow customer webhook endpoint) can never starve another (e.g. outbound sends) — the isolation the hand-rolled queues couldn't express cleanly. Keep these names stable: they are persisted on river_job rows.
const DefaultReconcileBatch = 1000
DefaultReconcileBatch bounds one reconcile scan. In steady state the stranded set is ~empty; under a systemic enqueue failure it caps how many rows one pass re-drives (one tx each) so an unhealthy River isn't amplified by fanning the whole backlog every tick — the remainder is picked up on the next pass.
Variables ¶
This section is empty.
Functions ¶
func Migrate ¶
Migrate applies River's own schema (river_job et al.) to the pool. River tracks its migrations in its own river_migration table, separate from e2a's schema_migrations, so this is idempotent and safe alongside identity migrations. Call once at startup, after the e2a migrations. (Generalized from the per-manager Migrate senderidentity used to own.)
func ReconcilePending ¶
func ReconcilePending(ctx context.Context, pool *pgxpool.Pool, spec ReconcileSpec, enqueueTx func(ctx context.Context, tx pgx.Tx, id string) (int64, error)) (int, error)
ReconcilePending re-drives every row matching spec.Where whose spec.JobColumn IS NULL: it scans up to spec.Batch ids, then per id opens a tx, re-checks the job column under FOR UPDATE (skipping rows another process or a prior pass already enqueued), calls enqueueTx to insert the River job in that tx, and stamps the returned job id back onto the row — all atomically. Idempotent: the FOR UPDATE + IS NULL guard means a re-run (or a concurrent replica) never double-enqueues. A per-row failure is logged and skipped (the next pass retries it); the returned count is the rows enqueued.
This is the shared body behind every domain's startup cutover + live reconcile worker (outboundsend, inboundprocess, hitlnotify, webhookdelivery, webhookpub fan-out). Each domain supplies a ReconcileSpec + its own EnqueueXTx.
Types ¶
type Client ¶
Client is the shared River client plus e2a's lifecycle helpers. It embeds *river.Client[pgx.Tx], so it IS an Enqueuer and exposes Start/Stop directly.
type Config ¶
type Config struct {
OutboundWorkers int // QueueOutbound concurrency (default 8)
InboundWorkers int // QueueInbound concurrency (default 8)
WebhookWorkers int // QueueWebhook concurrency (default 16)
MaintenanceWorkers int // QueueMaintenance concurrency (default 2)
NotifyWorkers int // QueueNotify concurrency (default 4)
DefaultWorkers int // QueueDefault concurrency (default 5)
}
Config sizes the per-queue worker pools. Zero values get sane defaults.
type Enqueuer ¶
type Enqueuer interface {
Insert(ctx context.Context, args river.JobArgs, opts *river.InsertOpts) (*rivertype.JobInsertResult, error)
InsertTx(ctx context.Context, tx pgx.Tx, args river.JobArgs, opts *river.InsertOpts) (*rivertype.JobInsertResult, error)
}
Enqueuer is the narrow insert surface business code depends on, rather than the whole river.Client — keeps the store/agent layer testable with a fake and River swappable. Insert enqueues immediately; InsertTx enqueues WITHIN the caller's transaction (the outbox pattern: the job commits atomically with the business write and can never be lost). *river.Client[pgx.Tx] satisfies this directly.
type ReconcileSpec ¶
type ReconcileSpec struct {
// Table is the row table (e.g. "messages", "webhook_events").
Table string
// JobColumn is the nullable bigint column holding the enqueued River job id
// (e.g. "send_job_id", "fanout_job_id"). A row is "stranded" when it matches
// Where AND JobColumn IS NULL.
JobColumn string
// Where is the domain predicate identifying rows that SHOULD carry a job
// (e.g. "direction='outbound' AND delivery_status='accepted'"). ANDed with
// "<JobColumn> IS NULL"; pass no leading/trailing "AND".
Where string
// LogPrefix tags per-row enqueue-failure logs (e.g. "[outbound-reconcile]").
LogPrefix string
// Batch caps rows scanned per pass; 0 uses DefaultReconcileBatch.
Batch int
}
ReconcileSpec describes one table's stranded-row reconcile.
SECURITY: Table, JobColumn, and Where are COMPILE-TIME CONSTANTS supplied by the calling package — never runtime or user input. They are interpolated directly into SQL (there is no parameter form for identifiers), so callers must pass only literal strings. The only runtime value (the row id) is always a bound parameter.
type Registrar ¶
type Registrar interface {
RegisterJobs(w *river.Workers) []*river.PeriodicJob
}
Registrar is what each domain implements to contribute its jobs to the shared client. RegisterJobs adds the domain's workers to the shared bundle (via river.AddWorker) and returns any periodic jobs it wants scheduled. Called once, at client construction — River requires all workers registered before Start.
Source Files