Documentation
¶
Overview ¶
Package hitlnotify sends the approval notification email that fires whenever a new outbound message enters pending_review.
The notification is the reviewer's primary touchpoint with HITL — it arrives in the account owner's inbox with a preview of the held message and one-click approve / reject magic links, plus a link back to the dashboard for edit-before-approve.
Delivery is durable, on River: the hold accept-tx enqueues a hitl_notify job (QueueNotify) in the same transaction as the pending_review row, and the NotifyWorker (worker.go) recomposes and submits the email ONCE off the request path — River owns the retry envelope (docs/design/hitl-notify-river.md). This replaced the earlier detached, best-effort goroutine, which lost the notification on a crash or SMTP outage between the 202 response and the send.
worker.go is Layer 3 of the durable HITL approval-notification pipeline (docs/design/hitl-notify-river.md): the River execution stage that takes a pending_review message, recomposes the reviewer's approve/reject email, and submits it ONCE off the request path. It mirrors internal/outboundsend — a River Worker on the shared `notify` queue, with River owning claim/retry/rescue.
At-least-once from the 202 response: the hitl_notify job is enqueued in the same tx as the pending_review row (the hold accept-tx) before the API answers, so an accepted hold's notification is never lost. The worker's notified_at stamp (written only AFTER a successful send) makes a crash-after-send re-drive a no-op; loss is impossible, a rare duplicate "please review" email is benign.
Index ¶
- Constants
- type DeliverOutcome
- type Deliverer
- type HITLNotifyArgs
- type Jobs
- func (j *Jobs) Deliver(ctx context.Context, pn *identity.PendingNotify) DeliverOutcome
- func (j *Jobs) EnqueueNotifyTx(ctx context.Context, tx pgx.Tx, messageID string) (int64, 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) SetDeliverer(d Deliverer)
- func (j *Jobs) SetEnqueuer(e jobs.Enqueuer)
- type Notifier
- type NotifyWorker
- type Store
Constants ¶
const MaxNotifyAttempts = 6
MaxNotifyAttempts caps the retry tail before River discards the job.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type DeliverOutcome ¶
type DeliverOutcome struct {
Err error
Permanent bool // 5xx / validation — no retry
Outage bool // relay unreachable — snooze without spending an attempt
}
DeliverOutcome is the classified result of one notification send. Permanent and Outage split the retry decision exactly as the outbound worker's does, using the shared internal/outbound SMTP classifiers.
type Deliverer ¶
type Deliverer interface {
Deliver(ctx context.Context, pn *identity.PendingNotify) DeliverOutcome
}
Deliverer composes and sends the approval email for one held message. Implemented by *Notifier (compose + SMTPRelay.SendOnce + classify).
type HITLNotifyArgs ¶
type HITLNotifyArgs struct {
MessageID string `json:"message_id"`
}
HITLNotifyArgs drives one approval-notification job. Args carry only the message id; the worker re-reads the message + agent (the source of truth) each attempt, so the job row stays tiny and always reflects the current hold state.
func (HITLNotifyArgs) Kind ¶
func (HITLNotifyArgs) Kind() string
type Jobs ¶
type Jobs struct {
// contains filtered or unexported fields
}
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 are injected AFTER construction (two-phase wiring) — the client via SetEnqueuer, the Deliverer (the Notifier, which needs the relay + signer resolved) via SetDeliverer. Jobs itself is the worker's Deliverer, late-binding to the concrete one, mirroring inboundprocess's late-bound Processor.
func (*Jobs) Deliver ¶
func (j *Jobs) Deliver(ctx context.Context, pn *identity.PendingNotify) DeliverOutcome
Deliver makes Jobs itself the worker's Deliverer, delegating to the concrete one set via SetDeliverer. Until that is wired (the brief startup window before the notifier is built) it returns a retryable outcome, so a pending job simply retries rather than dropping on a nil deliverer.
func (*Jobs) EnqueueNotifyTx ¶
EnqueueNotifyTx inserts the hitl_notify job in the caller's hold accept-tx (the same tx as the pending_review insert), returning the River job id to stamp on the message so a committed pending_review row always has its notification job.
func (*Jobs) ReconcilePending ¶
ReconcilePending enqueues a hitl_notify job for every pending_review message that has no job yet AND was never notified (notify_job_id IS NULL AND notified_at IS NULL). Run ONCE at startup as the cutover.
Because the accept-tx is a single transaction (message insert + job enqueue + job-id stamp commit together), a committed pending_review row in steady state ALWAYS has its job — so this set is normally empty. It exists to enqueue holds created on the no-notifier plain path (notified_at NULL) if a relay is later configured, plus any row stranded by a crash between insert and enqueue.
The `notified_at IS NULL` guard is what makes the feature's very first deploy safe: every hold already pending_review at cutover was notified by the old code path, and migration 057 stamps notified_at on exactly those rows, so this scan skips them — no owner is emailed twice. Idempotent: the per-row FOR UPDATE + notify_job_id IS NULL re-check means a re-run (or a concurrent replica) never double-enqueues.
func (*Jobs) RegisterJobs ¶
func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
RegisterJobs adds the NotifyWorker (with Jobs as the late-binding Deliverer). No periodics — the reconciler is a one-shot startup cutover. Implements jobs.Registrar.
func (*Jobs) SetDeliverer ¶
SetDeliverer injects the concrete Deliverer (the Notifier), built after the relay/signer gating resolves. Guarded so the River worker goroutines read it race-free.
func (*Jobs) SetEnqueuer ¶
SetEnqueuer injects the shared client so EnqueueNotifyTx can insert jobs.
type Notifier ¶
type Notifier struct {
// contains filtered or unexported fields
}
Notifier sends approval notification emails. Construct with New, then call NotifyPendingApproval from the HITL gate right after the pending row is written. Errors are logged, never returned upstream.
func New ¶
func New(store *identity.Store, relay *outbound.SMTPRelay, signer *approvaltoken.Signer, fromDomain, publicURL string) *Notifier
New returns a Notifier that sends mail through relay using the given public URL to build magic-link URLs. fromDomain is the platform relay's from-domain — e.g. "send.example.com" — which is combined with the fixed local-part to produce the From address.
func (*Notifier) Deliver ¶
func (n *Notifier) Deliver(ctx context.Context, pn *identity.PendingNotify) DeliverOutcome
Deliver composes and sends the approval email for one held message, classifying the result for the River NotifyWorker: a 5xx / validation reject is Permanent (no retry), an unreachable relay is an Outage (snooze), everything else retries. Implements hitlnotify.Deliverer. The classifiers key on the SMTP code / net error preserved through NotifyPendingApproval's %w wrapping.
func (*Notifier) NotifyPendingApproval ¶
func (n *Notifier) NotifyPendingApproval(ctx context.Context, msg *identity.Message, agent *identity.AgentIdentity) error
NotifyPendingApproval composes and sends the notification email for a held message, submitting once (SendOnce). It is the compose+send core the River NotifyWorker drives via Deliver; the returned error is classified there into retry/permanent/outage.
type NotifyWorker ¶
type NotifyWorker struct {
river.WorkerDefaults[HITLNotifyArgs]
// contains filtered or unexported fields
}
NotifyWorker sends the approval notification for one pending_review message. Mirrors outboundsend.SendWorker.
func NewNotifyWorker ¶
func NewNotifyWorker(store Store, deliverer Deliverer) *NotifyWorker
func (*NotifyWorker) NextRetry ¶
func (w *NotifyWorker) NextRetry(job *river.Job[HITLNotifyArgs]) time.Time
NextRetry overrides River's default backoff with the decided notify envelope.
func (*NotifyWorker) Work ¶
func (w *NotifyWorker) Work(ctx context.Context, job *river.Job[HITLNotifyArgs]) error
Work intentionally has no Timeout() override — a single SMTP submit of the approval notification fits River's 60s default JobTimeout.
type Store ¶
type Store interface {
// LoadPendingNotify returns the held message + owning agent, or (nil, nil) when
// there is nothing to notify about (message or agent gone) — a no-op.
LoadPendingNotify(ctx context.Context, messageID string) (*identity.PendingNotify, error)
// MarkMessageNotified stamps notified_at after a successful send (the dedup marker).
MarkMessageNotified(ctx context.Context, messageID string) error
// StampNotifyJobIDTx records the job id on a reconciled row (accept-tx + reconciler).
StampNotifyJobIDTx(ctx context.Context, tx pgx.Tx, messageID string, jobID int64) error
}
Store is the message surface the worker + reconciler need. Implemented over internal/identity (*identity.Store).
Source Files