Documentation
¶
Overview ¶
Package inboundprocess is Layer 3 of the queue-first inbound pipeline (docs/design/inbound-message-pipeline-river.md): the River execution stage that takes an accepted inbound_intake row and runs the full processing chain — parse, SPF/DKIM/DMARC, ingestion gate, content screening, persist, and event publish — off the SMTP critical path. It mirrors internal/outboundsend: a River Worker on the shared `inbound` queue, with River owning claim / retry / rescue.
At-least-once from the SMTP edge: the intake row is committed (and the job enqueued) in the same tx before 250, so an accepted message is never lost. The worker's terminal write flips intake.status='processed' ATOMICALLY with the messages insert + event publish (inside processInbound's tx), so a crash re-drive finds 'processed' and no-ops — the idempotency gate.
One processing pass per job attempt — River owns the multi-attempt envelope via NextRetry, so Work() stays a single ProcessIntake call, not an internal loop.
Index ¶
- Constants
- type InboundProcessArgs
- type InboundProcessWorker
- type InboundRetentionArgs
- type Jobs
- func (j *Jobs) EnqueueInboundProcessTx(ctx context.Context, tx pgx.Tx, intakeID string) (int64, error)
- func (j *Jobs) ProcessIntake(ctx context.Context, it *identity.InboundIntake) error
- func (j *Jobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (int, error)
- func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
- func (j *Jobs) SetEnqueuer(e jobs.Enqueuer)
- func (j *Jobs) SetProcessor(p Processor)
- type Processor
- type Pruner
- type RetentionWorker
- type Store
Constants ¶
const MaxInboundAttempts = 6
MaxInboundAttempts caps the retry tail before the intake is marked failed.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type InboundProcessArgs ¶
type InboundProcessArgs struct {
IntakeID string `json:"intake_id"`
}
InboundProcessArgs drives one inbound processing job. Args carry only the intake id; the worker re-reads the inbound_intake row (the source of truth) each attempt.
func (InboundProcessArgs) Kind ¶
func (InboundProcessArgs) Kind() string
type InboundProcessWorker ¶
type InboundProcessWorker struct {
river.WorkerDefaults[InboundProcessArgs]
// contains filtered or unexported fields
}
InboundProcessWorker processes an accepted intake row. Mirrors outboundsend.SendWorker.
func NewInboundProcessWorker ¶
func NewInboundProcessWorker(store Store, processor Processor) *InboundProcessWorker
func (*InboundProcessWorker) NextRetry ¶
func (w *InboundProcessWorker) NextRetry(job *river.Job[InboundProcessArgs]) time.Time
NextRetry overrides River's default backoff with the decided inbound envelope.
func (*InboundProcessWorker) Work ¶
func (w *InboundProcessWorker) Work(ctx context.Context, job *river.Job[InboundProcessArgs]) error
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.
type InboundRetentionArgs ¶
type InboundRetentionArgs struct{}
InboundRetentionArgs drives one retention sweep (no payload — the worker prunes the whole processed-and-old set).
func (InboundRetentionArgs) Kind ¶
func (InboundRetentionArgs) Kind() string
type Jobs ¶
type Jobs struct {
// contains filtered or unexported fields
}
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 injected AFTER construction (two-phase wiring) — the client via SetEnqueuer, the Processor (the relay Server) via SetProcessor, because the relay is built after jobs.New. Jobs itself is the worker's Processor, late-binding to the concrete one.
func (*Jobs) EnqueueInboundProcessTx ¶
func (j *Jobs) EnqueueInboundProcessTx(ctx context.Context, tx pgx.Tx, intakeID string) (int64, error)
EnqueueInboundProcessTx inserts the inbound_process job in the caller's accept-tx (the same tx as the inbound_intake insert), returning the River job id to stamp on the intake row so a committed accepted row always has its job.
func (*Jobs) ProcessIntake ¶
ProcessIntake makes Jobs itself the worker's Processor, delegating to the concrete one set via SetProcessor. Until that is wired (the brief startup window before the relay is built) it returns a retryable error, so a pending job simply retries rather than panicking on a nil processor.
func (*Jobs) ReconcilePending ¶
ReconcilePending enqueues an inbound_process job for every accepted intake row that has no job yet (process_job_id IS NULL). Run ONCE at startup as the cutover.
Because the accept-tx is a single transaction (intake insert + job enqueue + job-id stamp commit together), a committed accepted row in steady state ALWAYS has its job — so this set is normally empty. It exists to enqueue (a) any accepted rows at the moment the mode is first flipped on, and (b) rows stranded by a crash between insert and enqueue. Idempotent: the per-row FOR UPDATE + process_job_id IS NULL guard means a re-run (or a concurrent replica) never double-enqueues.
NOTE (follow-up, matching outboundsend): no live periodic reconciler yet. The one residual it would close is an intake left accepted-with-a-terminal/dead-job if the worker's MarkInboundIntakeFailed somehow never lands — rare, and the startup pass re-drives NULL-job rows on the next deploy.
func (*Jobs) RegisterJobs ¶
func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
RegisterJobs adds the InboundProcessWorker (with Jobs as the late-binding Processor) + the RetentionWorker, and schedules the retention sweep as a periodic on QueueMaintenance. Implements jobs.Registrar.
func (*Jobs) SetEnqueuer ¶
SetEnqueuer injects the shared client so EnqueueInboundProcessTx can insert jobs.
func (*Jobs) SetProcessor ¶
SetProcessor injects the concrete Processor (the relay Server), built after jobs.New. Guarded so the River worker goroutines read it race-free.
type Processor ¶
type Processor interface {
ProcessIntake(ctx context.Context, intake *identity.InboundIntake) error
}
Processor runs the full inbound chain for one intake row and marks the intake processed atomically with the messages insert. Implemented over the relay Server (which owns processInbound) in the binary.
type Pruner ¶
type Pruner interface {
PruneProcessedIntake(ctx context.Context, olderThan time.Duration) (int64, error)
}
Pruner is the retention surface (a subset of Store).
type RetentionWorker ¶
type RetentionWorker struct {
river.WorkerDefaults[InboundRetentionArgs]
// contains filtered or unexported fields
}
RetentionWorker prunes processed intake older than retentionWindow. Scheduled as a River periodic on QueueMaintenance (mirrors the webhook auto-disable janitor).
func (*RetentionWorker) Work ¶
func (w *RetentionWorker) Work(ctx context.Context, _ *river.Job[InboundRetentionArgs]) error
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.
type Store ¶
type Store interface {
// LoadInboundIntake returns the row, or (nil, nil) if pruned — a no-op.
LoadInboundIntake(ctx context.Context, intakeID string) (*identity.InboundIntake, error)
// MarkInboundIntakeFailed records a terminal failure after the retry tail.
MarkInboundIntakeFailed(ctx context.Context, intakeID, detail string) error
// StampInboundIntakeJobIDTx records the job id on a reconciled row.
StampInboundIntakeJobIDTx(ctx context.Context, tx pgx.Tx, intakeID string, jobID int64) error
// PruneProcessedIntake deletes processed rows older than olderThan (retention).
PruneProcessedIntake(ctx context.Context, olderThan time.Duration) (int64, error)
}
Store is the intake surface the worker + reconciler need. Implemented over internal/identity.
Source Files