Documentation
¶
Overview ¶
Package contactdue owns the outreach wake-up: it finds engagements whose next_action_at has passed and emits contact.due so an agent that is not running gets woken.
This is deliberately NOT part of the janitor. Everything the janitor does is pruning — expired messages, purged agents, dead sessions — and this is a scheduled product event with user-visible latency. Folding it in would have meant inheriting an hourly cadence chosen for TTL cleanup, sharing one error list and retry policy across two very different severities ("cleanup was skipped" vs "an agent was not woken"), and reporting through pruning metrics. It gets its own queue lane, interval, and metrics for the same reason every other domain here does.
Index ¶
Constants ¶
const Batch = 200
Batch bounds one sweep. Anything not claimed is picked up next pass, so a campaign coming due all at once spreads across runs instead of flooding a subscriber in a single burst.
const SweepInterval = 5 * time.Minute
SweepInterval is how often due engagements are checked.
Minutes, not the janitor's hour: this bounds how late a wake-up can be. A five-day follow-up cadence tolerates a few minutes of slack and would tolerate an hour, but the latency should be a decision rather than a side effect of which sweep the code happened to live in.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type BatchPublisher ¶
type BatchPublisher interface {
PublishDueBatch(ctx context.Context, now time.Time, limit int) (attempted int, err error)
}
BatchPublisher claims schedules and persists their durable events atomically. attempted is the number of schedules affected by a failed transaction (zero when the failure happened before the claim).
type DueClaimer ¶
type DueClaimer interface {
ClaimDueEngagementsTx(ctx context.Context, tx pgx.Tx, now time.Time, limit int) ([]identity.DueEngagement, error)
}
DueClaimer selects due schedules and advances their notification marker in a caller-owned transaction.
type Jobs ¶
type Jobs struct {
// contains filtered or unexported fields
}
Jobs registers the worker and its periodic schedule.
func (*Jobs) RegisterJobs ¶
func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
RegisterJobs adds the sweep worker and its periodic.
RunOnStart is true: a deploy or restart should pick up anything that came due while the process was down, rather than leaving agents unwoken until the next interval. The claim is idempotent per schedule, so an extra run is harmless.
type Metrics ¶
Metrics reports sweep outcomes. Separate from the janitor's row-deletion counters because nothing here is a deletion.
type OutboxPublisher ¶
type OutboxPublisher struct {
// contains filtered or unexported fields
}
OutboxPublisher owns the schedule-claim → durable-event invariant. Both writes use one PostgreSQL transaction, eliminating the loss window where a schedule was previously marked notified before a best-effort publisher tried to create the event.
func NewOutboxPublisher ¶
func NewOutboxPublisher(pool *pgxpool.Pool, claimer DueClaimer, outbox TxOutbox) *OutboxPublisher
NewOutboxPublisher builds the atomic batch publisher.
func (*OutboxPublisher) PublishDueBatch ¶
func (p *OutboxPublisher) PublishDueBatch(ctx context.Context, now time.Time, limit int) (attempted int, err error)
PublishDueBatch claims at most limit schedules and persists every wake-up in the same transaction. On error, attempted reports how many schedules had been selected before rollback so observability can count the affected batch.
type SweepArgs ¶
type SweepArgs struct{}
SweepArgs drives the periodic. No fields — each run claims the next batch.
func (SweepArgs) InsertOpts ¶
func (SweepArgs) InsertOpts() river.InsertOpts
InsertOpts routes the sweep to the maintenance lane: it is periodic background work and must never compete with customer sends or webhook delivery for worker slots.
type Sweeper ¶
type Sweeper struct {
// contains filtered or unexported fields
}
Sweeper is the unit of work. Exported and directly callable so an integration test can drive a real sweep without waiting on a periodic schedule.
func NewSweeper ¶
func NewSweeper(p BatchPublisher, m Metrics) *Sweeper
NewSweeper builds the sweeper. metrics may be nil.
type TxOutbox ¶
TxOutbox is the one outbox capability this module needs. The narrow seam makes the atomic rollback behavior directly testable while the production webhookpub.Outbox satisfies it without an adapter.
type Worker ¶
type Worker struct {
river.WorkerDefaults[SweepArgs]
// contains filtered or unexported fields
}
Worker runs one sweep per fire.
Source Files