chore(river): modularity + readability cleanup across the River systems

internal/config/config.go

@@ -153,7 +153,7 @@ type InboundConfig struct {
 // river-migration.md). Mode="legacy" (the default) drains webhook_events →
 // webhook_subscriber_deliveries via the in-process webhookpub.OutboxWorker
 // (LISTEN/NOTIFY + poll + SKIP-LOCKED lease). Mode="river" opts into the River
-// pipeline: PublishTx/PublishBestEffortTx enqueue a webhook_fan_out job in the event's
+// pipeline: PublishTx/PublishBestEffortTx enqueue a webhook_fanout job in the event's
 // tx and the webhookpub.FanOutWorker does the match/insert/enqueue off the drain loop.
 // Override with E2A_WEBHOOK_FANOUT_MODE. Any value other than "river" is treated as
 // "legacy" (fail-safe to the unchanged path). Wired in Slice 2; unused under legacy.

internal/hitlnotify/jobs.go

@@ -3,7 +3,6 @@ package hitlnotify
 import (
 	"context"
 	"errors"
-	"log"
 	"sync"
 
 	"github.com/jackc/pgx/v5"
@@ -14,12 +13,6 @@ import (
 	"github.com/Mnexa-AI/e2a/internal/jobs"
 )
 
-// reconcileBatch bounds one cutover scan of pending_review rows without a
-// notification job — mirrors outboundsend/inboundprocess. In steady state the set
-// is ~empty (the accept-tx stamps notify_job_id atomically); this caps how many
-// rows one pass re-drives if the feature was just enabled or a crash left a backlog.
-const reconcileBatch = 1000
-
 // Jobs is the HITL-notification integration on the shared River client: a
 // jobs.Registrar (contributes NotifyWorker) plus the transactional enqueue entry
 // point the hold accept-tx calls. Both the shared client and the concrete Deliverer
@@ -103,53 +96,13 @@ func (j *Jobs) EnqueueNotifyTx(ctx context.Context, tx pgx.Tx, messageID string)
 // notify_job_id IS NULL re-check means a re-run (or a concurrent replica) never
 // double-enqueues.
 func (j *Jobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (int, error) {
-	rows, err := pool.Query(ctx,
-		`SELECT id FROM messages
-		  WHERE status='pending_review' AND notify_job_id IS NULL AND notified_at IS NULL
-		  LIMIT $1`, reconcileBatch)
-	if err != nil {
-		return 0, err
-	}
-	var ids []string
-	for rows.Next() {
-		var id string
-		if err := rows.Scan(&id); err != nil {
-			rows.Close()
-			return 0, err
-		}
-		ids = append(ids, id)
-	}
-	rows.Close()
-	if err := rows.Err(); err != nil {
-		return 0, err
-	}
-
-	n := 0
-	for _, id := range ids {
-		if err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
-			// Re-check under a row lock: another process (or a prior run) may have
-			// enqueued it already. Skip if notify_job_id is now set.
-			var jobID *int64
-			if err := tx.QueryRow(ctx,
-				`SELECT notify_job_id FROM messages WHERE id=$1 FOR UPDATE`, id,
-			).Scan(&jobID); err != nil {
-				return err
-			}
-			if jobID != nil {
-				return nil // already enqueued
-			}
-			newJobID, err := j.EnqueueNotifyTx(ctx, tx, id)
-			if err != nil {
-				return err
-			}
-			if err := j.store.StampNotifyJobIDTx(ctx, tx, id, newJobID); err != nil {
-				return err
-			}
-			n++
-			return nil
-		}); err != nil {
-			log.Printf("[hitl-notify] reconcile enqueue %s: %v", id, err)
-		}
-	}
-	return n, nil
+	// The stamp is an inline UPDATE inside jobs.ReconcilePending — identical SQL to
+	// store.StampNotifyJobIDTx (which the accept-tx still uses). The notified_at IS NULL
+	// guard keeps already-emailed holds out of the reconcile set.
+	return jobs.ReconcilePending(ctx, pool, jobs.ReconcileSpec{
+		Table:     "messages",
+		JobColumn: "notify_job_id",
+		Where:     "status='pending_review' AND notified_at IS NULL",
+		LogPrefix: "[hitl-notify] reconcile",
+	}, j.EnqueueNotifyTx)
 }

internal/hitlnotify/worker.go

@@ -99,6 +99,8 @@ func (w *NotifyWorker) NextRetry(job *river.Job[HITLNotifyArgs]) time.Time {
 	return time.Now().Add(notifyRetryBackoffs[i])
 }
 
+// Work intentionally has no Timeout() override — a single SMTP submit of the approval
+// notification fits River's 60s default JobTimeout.
 func (w *NotifyWorker) Work(ctx context.Context, job *river.Job[HITLNotifyArgs]) error {
 	pn, err := w.store.LoadPendingNotify(ctx, job.Args.MessageID)
 	if err != nil {

internal/hitlworker/events.go

@@ -25,7 +25,8 @@ import (
 // reviewed_by_user_id is omitted and `auto_resolved: true` is set; and the
 // routing key (Event.UserID) is the agent OWNER (the sweep is system-scoped).
 
-// publish fires e through the legacy publisher, detached from the sweep's
+// publish fires e through the webhook publisher (the OutboxPublisher, which writes the
+// webhook_events outbox that a fan-out worker drains), detached from the sweep's
 // context so a cancelled sweep doesn't drop an already-committed resolution's
 // notification. No-op when no publisher is wired. The deterministic id (set by
 // the caller) makes a re-swept row's event idempotent.

internal/hitlworker/maintenance.go

@@ -105,7 +105,7 @@ func NewMaintenanceJobs(sweeper Sweeper, asyncSends bool) *MaintenanceJobs {
 // idempotent + cross-replica-safe so scheduling is the only concern). Implements
 // jobs.Registrar.
 func (m *MaintenanceJobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob {
-	river.AddWorker(w, &MaintenanceWorker{sweeper: m.sweeper, asyncSends: m.asyncSends})
+	river.AddWorker(w, NewMaintenanceWorker(m.sweeper, m.asyncSends))
 	return []*river.PeriodicJob{
 		river.NewPeriodicJob(
 			river.PeriodicInterval(maintenanceInterval),

internal/inboundprocess/jobs.go

@@ -3,7 +3,6 @@ package inboundprocess
 import (
 	"context"
 	"errors"
-	"log"
 	"sync"
 
 	"github.com/jackc/pgx/v5"
@@ -14,12 +13,6 @@ import (
 	"github.com/Mnexa-AI/e2a/internal/jobs"
 )
 
-// reconcileBatch bounds one cutover scan of stranded accepted intake rows — mirrors
-// outboundsend.reconcileBatch. In steady state the stranded set is ~empty (the
-// accept-tx stamps process_job_id atomically); this caps how many rows one pass
-// re-drives if a systemic enqueue failure ever left a backlog.
-const reconcileBatch = 1000
-
 // Jobs is the inbound-processing integration on the shared River client: a
 // jobs.Registrar (contributes InboundProcessWorker) plus the transactional enqueue
 // entry point the SMTP accept-tx calls. The shared client + the Processor are both
@@ -96,55 +89,14 @@ func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob {
 // worker's MarkInboundIntakeFailed somehow never lands — rare, and the startup pass
 // re-drives NULL-job rows on the next deploy.
 func (j *Jobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (int, error) {
-	rows, err := pool.Query(ctx,
-		`SELECT id FROM inbound_intake
-		  WHERE status='accepted' AND process_job_id IS NULL
-		  LIMIT $1`, reconcileBatch)
-	if err != nil {
-		return 0, err
-	}
-	var ids []string
-	for rows.Next() {
-		var id string
-		if err := rows.Scan(&id); err != nil {
-			rows.Close()
-			return 0, err
-		}
-		ids = append(ids, id)
-	}
-	rows.Close()
-	if err := rows.Err(); err != nil {
-		return 0, err
-	}
-
-	n := 0
-	for _, id := range ids {
-		if err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
-			// Re-check under a row lock: another process (or a prior run) may have
-			// enqueued it already. Skip if process_job_id is now set.
-			var jobID *int64
-			if err := tx.QueryRow(ctx,
-				`SELECT process_job_id FROM inbound_intake WHERE id=$1 FOR UPDATE`, id,
-			).Scan(&jobID); err != nil {
-				return err
-			}
-			if jobID != nil {
-				return nil // already enqueued
-			}
-			newJobID, err := j.EnqueueInboundProcessTx(ctx, tx, id)
-			if err != nil {
-				return err
-			}
-			if err := j.store.StampInboundIntakeJobIDTx(ctx, tx, id, newJobID); err != nil {
-				return err
-			}
-			n++
-			return nil
-		}); err != nil {
-			log.Printf("[inbound-reconcile] enqueue %s: %v", id, err)
-		}
-	}
-	return n, nil
+	// The stamp is an inline UPDATE inside jobs.ReconcilePending — identical SQL to
+	// store.StampInboundIntakeJobIDTx (which the accept-tx still uses).
+	return jobs.ReconcilePending(ctx, pool, jobs.ReconcileSpec{
+		Table:     "inbound_intake",
+		JobColumn: "process_job_id",
+		Where:     "status='accepted'",
+		LogPrefix: "[inbound-reconcile]",
+	}, j.EnqueueInboundProcessTx)
 }
 
 // EnqueueInboundProcessTx inserts the inbound_process job in the caller's accept-tx

internal/inboundprocess/retention.go

@@ -35,6 +35,12 @@ type RetentionWorker struct {
 	pruner Pruner
 }
 
+// Work intentionally has no Timeout() override — it relies on River's 60s default
+// JobTimeout. PruneProcessedIntake is a single unbounded DELETE; inbound_intake is
+// short-lived (processed rows pruned every tick), so the set is normally tiny. On a
+// large backlog a 60s cut is safe (the DELETE is ctx-aware — it rolls back and the next
+// tick resumes), just slow. TODO: batch it like the janitor's ctid-LIMIT prunes (#397)
+// if a backlog ever makes the single DELETE a problem.
 func (w *RetentionWorker) Work(ctx context.Context, _ *river.Job[InboundRetentionArgs]) error {
 	n, err := w.pruner.PruneProcessedIntake(ctx, retentionWindow)
 	if err != nil {

internal/inboundprocess/worker.go

@@ -93,6 +93,10 @@ func (w *InboundProcessWorker) NextRetry(job *river.Job[InboundProcessArgs]) tim
 	return time.Now().Add(inboundRetryBackoffs[i])
 }
 
+// Work intentionally has no Timeout() override — parse/SPF-DKIM/screen/persist fits
+// River's 60s default JobTimeout. The heaviest step, the content screen, is an external
+// LLM call bounded by its own per-detector timeout (piguard, 5s default) that fails
+// open, so it can't hang the job past 60s.
 func (w *InboundProcessWorker) Work(ctx context.Context, job *river.Job[InboundProcessArgs]) error {
 	it, err := w.store.LoadInboundIntake(ctx, job.Args.IntakeID)
 	if err != nil {

internal/janitor/janitor.go

@@ -244,7 +244,7 @@ func NewMaintenanceJobs(j *Janitor) *MaintenanceJobs { return &MaintenanceJobs{j
 // unbounded-DELETE hazard is resolved — each is now a ctx-aware ctid-LIMIT drain
 // loop (see the DeleteExpired* store methods) bounded by MaintenanceWorker.Timeout.
 func (m *MaintenanceJobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob {
-	river.AddWorker(w, &MaintenanceWorker{janitor: m.janitor})
+	river.AddWorker(w, NewMaintenanceWorker(m.janitor))
 	return []*river.PeriodicJob{
 		river.NewPeriodicJob(
 			river.PeriodicInterval(janitorInterval),

internal/jobs/jobs.go

@@ -107,7 +107,7 @@ func New(pool *pgxpool.Pool, cfg Config, registrars ...Registrar) (*Client, erro
 	}
 
 	rc, err := river.NewClient(riverpgxv5.New(pool), &river.Config{
-		Queues:       defaultQueueConfig(cfg.OutboundWorkers, cfg.InboundWorkers, cfg.WebhookWorkers, cfg.MaintenanceWorkers, cfg.NotifyWorkers, cfg.DefaultWorkers),
+		Queues:       defaultQueueConfig(cfg),
 		Workers:      workers,
 		PeriodicJobs: periodic,
 	})

internal/jobs/queues.go

@@ -34,13 +34,13 @@ const (
 // MaxWorkers is deliberately generous for I/O-bound work (SMTP / HTTP); tune
 // against real throughput. Every queue a domain enqueues into MUST appear here or
 // its jobs sit unworked.
-func defaultQueueConfig(outbound, inbound, webhook, maintenance, notify, deflt int) map[string]river.QueueConfig {
+func defaultQueueConfig(cfg Config) map[string]river.QueueConfig {
 	return map[string]river.QueueConfig{
-		QueueOutbound:    {MaxWorkers: outbound},
-		QueueInbound:     {MaxWorkers: inbound},
-		QueueWebhook:     {MaxWorkers: webhook},
-		QueueMaintenance: {MaxWorkers: maintenance},
-		QueueNotify:      {MaxWorkers: notify},
-		QueueDefault:     {MaxWorkers: deflt},
+		QueueOutbound:    {MaxWorkers: cfg.OutboundWorkers},
+		QueueInbound:     {MaxWorkers: cfg.InboundWorkers},
+		QueueWebhook:     {MaxWorkers: cfg.WebhookWorkers},
+		QueueMaintenance: {MaxWorkers: cfg.MaintenanceWorkers},
+		QueueNotify:      {MaxWorkers: cfg.NotifyWorkers},
+		QueueDefault:     {MaxWorkers: cfg.DefaultWorkers},
 	}
 }

internal/jobs/reconcile.go

@@ -0,0 +1,110 @@
+package jobs
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// 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.
+const DefaultReconcileBatch = 1000
+
+// 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 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
+}
+
+// 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.
+func ReconcilePending(ctx context.Context, pool *pgxpool.Pool, spec ReconcileSpec,
+	enqueueTx func(ctx context.Context, tx pgx.Tx, id string) (int64, error)) (int, error) {
+
+	batch := spec.Batch
+	if batch <= 0 {
+		batch = DefaultReconcileBatch
+	}
+
+	rows, err := pool.Query(ctx,
+		fmt.Sprintf(`SELECT id FROM %s WHERE %s AND %s IS NULL LIMIT $1`,
+			spec.Table, spec.Where, spec.JobColumn),
+		batch)
+	if err != nil {
+		return 0, err
+	}
+	var ids []string
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			rows.Close()
+			return 0, err
+		}
+		ids = append(ids, id)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return 0, err
+	}
+
+	recheckSQL := fmt.Sprintf(`SELECT %s FROM %s WHERE id=$1 FOR UPDATE`, spec.JobColumn, spec.Table)
+	stampSQL := fmt.Sprintf(`UPDATE %s SET %s=$2 WHERE id=$1`, spec.Table, spec.JobColumn)
+
+	n := 0
+	for _, id := range ids {
+		if err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
+			// Re-check under a row lock: another process (or a prior run) may have
+			// enqueued it already. Skip if the job column is now set.
+			var jobID *int64
+			if err := tx.QueryRow(ctx, recheckSQL, id).Scan(&jobID); err != nil {
+				return err
+			}
+			if jobID != nil {
+				return nil // already enqueued
+			}
+			newJobID, err := enqueueTx(ctx, tx, id)
+			if err != nil {
+				return err
+			}
+			if _, err := tx.Exec(ctx, stampSQL, id, newJobID); err != nil {
+				return err
+			}
+			n++
+			return nil
+		}); err != nil {
+			log.Printf("%s enqueue %s: %v", spec.LogPrefix, id, err)
+		}
+	}
+	return n, nil
+}