Documentation
¶
Overview ¶
Package hitlworker runs the periodic sweep that finalizes pending_review holds whose TTL has elapsed. Outbound holds become sent (auto-approved) or review_expired_rejected; inbound holds become review_expired_approved (released to the agent) or review_expired_rejected — per the owning agent's hitl_expiration_action column. Message content remains retained in both cases.
Index ¶
- Constants
- type HITLMaintenanceArgs
- type MaintenanceJobs
- type MaintenanceWorker
- type ManagedUnsubscribeIssuer
- type OutboundEnqueuer
- type Sweeper
- type WebSocketHub
- type Worker
- func (w *Worker) RunOnce(ctx context.Context) error
- func (w *Worker) SetManagedUnsubscribeIssuer(i ManagedUnsubscribeIssuer)
- func (w *Worker) SetOutboundEnqueuer(e OutboundEnqueuer)
- func (w *Worker) SetOutboundFooterResolver(r func(ctx context.Context, userID string) bool)
- func (w *Worker) SetOutbox(o webhookpub.Outbox)
- func (w *Worker) SetPublisher(p webhookpub.Publisher)
- func (w *Worker) SetQuotaCheck(f func(ctx context.Context, userID string, units int) error)
- func (w *Worker) SetWebSocketHub(h WebSocketHub)
Constants ¶
const DefaultBatchSize = 100
DefaultBatchSize caps how many rows one sweep will try to finalize. The partial index on (approval_expires_at) WHERE status='pending_review' keeps the list query cheap regardless of total table size.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type HITLMaintenanceArgs ¶
type HITLMaintenanceArgs struct{}
HITLMaintenanceArgs drives the periodic HITL TTL sweep. No fields — the worker sweeps every expired hold each run.
func (HITLMaintenanceArgs) Kind ¶
func (HITLMaintenanceArgs) Kind() string
type MaintenanceJobs ¶
type MaintenanceJobs struct {
// contains filtered or unexported fields
}
MaintenanceJobs is the jobs.Registrar for the HITL TTL sweep: it contributes the MaintenanceWorker and a periodic that fires it on QueueMaintenance. No enqueuer needed — the schedule is the only trigger.
func NewMaintenanceJobs ¶
func NewMaintenanceJobs(sweeper Sweeper) *MaintenanceJobs
NewMaintenanceJobs builds the registrar around a Sweeper (the *Worker).
func (*MaintenanceJobs) RegisterJobs ¶
func (m *MaintenanceJobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
RegisterJobs adds the maintenance worker + its periodic schedule. Mirrors the webhook janitor + inbound retention periodics: routed to QueueMaintenance, no UniqueOpts (River's periodic scheduler already inserts at most one per interval and a completed run must not dedup-block the next), RunOnStart:false (first sweep after one interval — a conscious minor change from the old ticker's immediate first sweep, consistent with the other periodics; the per-row store operations are idempotent + cross-replica-safe so scheduling is the only concern). Implements jobs.Registrar.
type MaintenanceWorker ¶
type MaintenanceWorker struct {
river.WorkerDefaults[HITLMaintenanceArgs]
// contains filtered or unexported fields
}
MaintenanceWorker runs the TTL sweep once per scheduled job. Any sweep error is logged and swallowed (Work returns nil) so a transient DB blip never spins River's retry machinery for a best-effort idempotent sweep — the next interval picks it up. Mirrors webhookdelivery.MaintenanceWorker.
func NewMaintenanceWorker ¶
func NewMaintenanceWorker(sweeper Sweeper) *MaintenanceWorker
NewMaintenanceWorker builds the worker around a Sweeper. Exported so tests can drive Work directly (RegisterJobs builds an identical one).
func (*MaintenanceWorker) Timeout ¶
func (w *MaintenanceWorker) Timeout(*river.Job[HITLMaintenanceArgs]) time.Duration
Timeout caps the sweep's runtime. River's client default is 1 minute. Auto-approve enqueues onto QueueOutbound instead of blocking, so a bounded timeout safely protects slots in the shared maintenance pool.
func (*MaintenanceWorker) Work ¶
func (w *MaintenanceWorker) Work(ctx context.Context, _ *river.Job[HITLMaintenanceArgs]) error
type OutboundEnqueuer ¶
type OutboundEnqueuer interface {
EnqueueSendTx(ctx context.Context, tx pgx.Tx, messageID string) (int64, error)
// EnqueueScheduledSendTx enqueues the send to run no earlier than `at`, used
// when an auto-approved hold carried a still-future send_at (#815). Same
// outbox tx as EnqueueSendTx; only the job's first-run time differs.
EnqueueScheduledSendTx(ctx context.Context, tx pgx.Tx, messageID string, at time.Time) (int64, error)
}
OutboundEnqueuer inserts an outbound_send job (QueueOutbound) in the caller's transaction. Satisfied by *outboundsend.Jobs. The sweep hands an approved outbound send to the queue-first pipeline — transitioning the hold to review_expired_approved + delivery_status='accepted' and enqueuing — instead of blocking on Sender.Send. Self-sends never use it (they loopback).
type Sweeper ¶
Sweeper runs one TTL-expiration sweep of both hold queues (outbound holds + inbound review holds). Satisfied by *Worker — the River worker just drives its RunOnce on a schedule instead of a hand-rolled time.Ticker.
type WebSocketHub ¶
type Worker ¶
type Worker struct {
// contains filtered or unexported fields
}
Worker runs the TTL sweep. Construct with New; its RunOnce is driven on a schedule by the River maintenance periodic (see maintenance.go).
func New ¶
func New(store *identity.Store, sender *outbound.Sender, usageTracker usage.UsageTracker, fromDomain string) *Worker
New constructs a Worker. fromDomain is the deployment's outbound from-domain (cfg.OutboundSMTP.FromDomain) — used by the self-send loopback branch to stamp the synthetic Message-ID / Received headers the same way internal/agent does on the user-driven approve path. Pass "" if the deployment has no outbound relay configured; the loopback path falls back to "e2a.local" for the host portion.
func (*Worker) RunOnce ¶
RunOnce performs a single sweep of both queues (outbound holds, then inbound review holds). This is the sweep body the River maintenance periodic drives on a schedule (see maintenance.go); it's also called directly by tests for deterministic behavior. Returns nil — both sweeps log and swallow their own per-row/query errors internally (a transient DB blip should not spin River's retry machinery); the error return satisfies the Sweeper interface.
func (*Worker) SetManagedUnsubscribeIssuer ¶
func (w *Worker) SetManagedUnsubscribeIssuer(i ManagedUnsubscribeIssuer)
func (*Worker) SetOutboundEnqueuer ¶
func (w *Worker) SetOutboundEnqueuer(e OutboundEnqueuer)
SetOutboundEnqueuer wires the mandatory outbound send enqueuer. Two-phase wiring: pass the *outboundsend.Jobs pointer; its shared River client is injected later via the jobs client's SetEnqueuer.
func (*Worker) SetOutboundFooterResolver ¶ added in v1.6.0
SetOutboundFooterResolver wires the per-account outbound-footer decision for TTL auto-approved sends. See the footerResolver field for semantics.
func (*Worker) SetOutbox ¶
func (w *Worker) SetOutbox(o webhookpub.Outbox)
SetOutbox wires the transactional outcome-event writer for providerless local delivery. Production uses the same unconditional outbox as all other message triggers.
func (*Worker) SetPublisher ¶
func (w *Worker) SetPublisher(p webhookpub.Publisher)
SetPublisher wires the webhook publisher used to emit review-resolution events on TTL auto-resolution. Without it the sweep transitions rows silently.
func (*Worker) SetQuotaCheck ¶ added in v1.8.0
SetQuotaCheck installs the flow-cap re-check consulted before a TTL auto-approve releases a hold into delivery. Held drafts are not metered at accept, so expiry time is the enforcement point — exactly like the human approve funnel's checkApproveQuota. The func returns a *limits.LimitExceededError-shaped error when over cap; a transient error is treated as fail-open by the caller. Wired in main from the limits enforcer; nil (tests, self-host without limits) disables the check.
func (*Worker) SetWebSocketHub ¶
func (w *Worker) SetWebSocketHub(h WebSocketHub)
Source Files