jobs

type Client ¶

type Client struct {
	*river.Client[pgx.Tx]
}

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.

func New(pool *pgxpool.Pool, cfg Config, registrars ...Registrar) (*Client, error)

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 QueueStatsArgs ¶ added in v1.2.2

type QueueStatsArgs struct{}

QueueStatsArgs drives one gauge sampling pass.

func (QueueStatsArgs) Kind() string

type QueueStatsJobs ¶ added in v1.2.2

type QueueStatsJobs struct {
	// contains filtered or unexported fields
}

QueueStatsJobs contributes the periodic queue-gauge sampler to the shared client. Implements Registrar (same shape as sendramp.MaintenanceJobs).

func NewQueueStatsJobs(pool *pgxpool.Pool, metrics QueueStatsMetrics) *QueueStatsJobs

func (q *QueueStatsJobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob

type QueueStatsMetrics ¶ added in v1.2.2

type QueueStatsMetrics interface {
	SetQueueDepth(queue, state string, n int)
	SetQueueOldestAge(queue string, seconds float64)
}

QueueStatsMetrics is the narrow gauge surface the sampler sets. Satisfied by telemetry.Metrics; injectable so tests assert with a fake.

type QueueStatsWorker ¶ added in v1.2.2

type QueueStatsWorker struct {
	river.WorkerDefaults[QueueStatsArgs]
	// contains filtered or unexported fields
}

QueueStatsWorker samples river_job depth and oldest-runnable-age gauges per queue. Read-only and bounded (two single-round-trip aggregates over the indexed state column), so River's default JobTimeout is ample.

func NewQueueStatsWorker(pool *pgxpool.Pool, metrics QueueStatsMetrics) *QueueStatsWorker

func (w *QueueStatsWorker) Sample(ctx context.Context) error

func (w *QueueStatsWorker) Work(ctx context.Context, _ *river.Job[QueueStatsArgs]) error

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.