hitlnotify

package
v1.9.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 5, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

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 a primary link to the consolidated dashboard review, plus signed quick-action links for reviewers who cannot sign in.

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

View Source 
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 ArgStamper added in v1.9.0 

type ArgStamper func(ctx context.Context, jobID int64, ref sendingpolicy.OperationRef) error

ArgStamper persists a resolved reference into the job's args so a legacy job is resolved once, not once per execution.

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 {
	Compose(ctx context.Context, pn *identity.PendingNotify) (outbound.Envelope, DeliverOutcome)
	Submit(ctx context.Context, env outbound.Envelope, auth sendingpolicy.ProviderAuthorization) DeliverOutcome
}

Deliverer is the two-phase send of one approval email. Compose does every fallible, provider-free step (owner lookup, token signing, MIME, DKIM) and returns the envelope; Submit hands that envelope and a freshly consumed authorization to the provider seam. The split exists so the worker can ConsumeAttempt immediately before the socket opens: a compose failure then costs nothing, instead of a charged ordinal that never reached the provider. Implemented by *Notifier.

type HITLNotifyArgs 

type HITLNotifyArgs struct {
	MessageID string `json:"message_id"`
	// OperationRef is the durable sending operation the hold's accept
	// transaction prepared. A job enqueued by a pre-floor slot carries none
	// and is resolved at fire time through the same Prepare path, then
	// stamped so the derivation happens once.
	OperationRef *sendingpolicy.OperationRef `json:"operation_ref,omitempty"`
}

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 NewJobs 

func NewJobs(store Store) *Jobs

NewJobs builds the integration with just its store (no client, no deliverer yet).

func (*Jobs) Compose added in v1.9.0 

func (j *Jobs) Compose(ctx context.Context, pn *identity.PendingNotify) (outbound.Envelope, DeliverOutcome)

Compose 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 — and because Compose runs before any attempt is charged, that window costs nothing.

func (*Jobs) EnqueueNotifyTx 

func (j *Jobs) EnqueueNotifyTx(ctx context.Context, tx pgx.Tx, messageID string) (int64, error)

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.

With a gate wired the notification's operation is prepared here, in the same transaction, against the locked source row: the triggering account is charged, never the platform, and the worker never derives attribution.

func (*Jobs) Gate added in v1.9.0 

func (j *Jobs) Gate() sendingpolicy.Gate

Gate exposes the wired sending-protection gate (nil when gateless), so the composition root's wiring test can prove the production bundle is armed.

func (*Jobs) NotifyWorker added in v1.9.0 

func (j *Jobs) NotifyWorker() *NotifyWorker

NotifyWorker builds the fully armed worker RegisterJobs registers.

func (*Jobs) ReconcilePending 

func (j *Jobs) ReconcilePending(ctx context.Context, pool *pgxpool.Pool) (int, error)

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) ResolveLegacyOperation added in v1.9.0 

func (j *Jobs) ResolveLegacyOperation(ctx context.Context, messageID string) (sendingpolicy.OperationRef, error)

ResolveLegacyOperation prepares the notification operation for a job that carries no reference, in its own committed transaction, through the same PrepareNotificationTx an enqueue runs.

func (*Jobs) SetDeliverer 

func (j *Jobs) SetDeliverer(d Deliverer)

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 

func (j *Jobs) SetEnqueuer(e jobs.Enqueuer)

SetEnqueuer injects the shared client so EnqueueNotifyTx can insert jobs.

func (*Jobs) Submit added in v1.9.0 

func (j *Jobs) Submit(ctx context.Context, env outbound.Envelope, auth sendingpolicy.ProviderAuthorization) DeliverOutcome

Submit delegates the authorized submission to the concrete Deliverer.

func (*Jobs) WithGate added in v1.9.0 

func (j *Jobs) WithGate(g sendingpolicy.Gate, pool *pgxpool.Pool) *Jobs

WithGate injects the sending-protection gate and the pool its legacy resolver and arg stamp use. Every enqueue then prepares a notification operation in the hold's transaction and every worker execution authorizes through the gate. Chainable; nil keeps the gateless default (tests only).

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, submitter *outbound.ProviderSubmitter, signer *approvaltoken.Signer, fromDomain, fromAddress, replyTo, 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" — combined with notifyLocalPart to produce the From address when fromAddress is empty.

fromAddress and replyTo are notifications.from_address / .reply_to. Setting fromAddress to the same value here and in webhooknotify is what consolidates both senders into one identity; leaving it unset keeps them distinct and separately filterable. Resolution deliberately mirrors webhooknotify.New line for line; it is a copy rather than a shared helper, so changing one means changing the other.

func (*Notifier) Compose added in v1.9.0 

func (n *Notifier) Compose(ctx context.Context, pn *identity.PendingNotify) (outbound.Envelope, DeliverOutcome)

Compose implements Deliverer: the provider-free half, classified like a send so the worker treats a permanent compose failure the same way.

func (*Notifier) NotifyPendingApproval 

func (n *Notifier) NotifyPendingApproval(ctx context.Context, msg *identity.Message, agent *identity.AgentIdentity, auth sendingpolicy.ProviderAuthorization) error

NotifyPendingApproval composes and sends the notification email for a held message with an already-authorized attempt: Compose then Submit in one call, for callers that hold the token up front (tests, the reconciler drill). The worker calls the two phases itself so the token is consumed last.

func (*Notifier) Submit added in v1.9.0 

func (n *Notifier) Submit(ctx context.Context, env outbound.Envelope, auth sendingpolicy.ProviderAuthorization) DeliverOutcome

Submit implements Deliverer: one authorized submission, classified for the River NotifyWorker — a 5xx / validation reject is Permanent (no retry), an unreachable relay is an Outage (snooze), everything else retries.

func (*Notifier) WithDKIM added in v1.7.0 

func (n *Notifier) WithDKIM(lookup outbound.DKIMKeyLookup) *Notifier

WithDKIM wires per-domain DKIM signing, mirroring webhooknotify. The relay itself never signs and an upstream provider only signs identities it manages, so a custom notifications.from_address domain is signed here or not at all. nil-safe and fail-open via outbound.SignWithDKIM.

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

NewNotifyWorker builds a worker with no sending-protection gate. Without one the deliverer receives an empty authorization, which the production submitter refuses before it dials; the composition root installs the gate via WithGate.

func (*NotifyWorker) Gate added in v1.9.0 

func (w *NotifyWorker) Gate() sendingpolicy.Gate

Gate exposes the wired gate (nil when gateless), for the composition root's wiring test.

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) WithArgRestamper added in v1.9.0 

func (w *NotifyWorker) WithArgRestamper(s ArgStamper) *NotifyWorker

WithArgRestamper injects the unconditional re-key used when a job carries a pre-derivation reference.

func (*NotifyWorker) WithArgStamper added in v1.9.0 

func (w *NotifyWorker) WithArgStamper(s ArgStamper) *NotifyWorker

WithArgStamper injects the job-args stamp used after a legacy resolution (adds the reference only when absent).

func (*NotifyWorker) WithGate added in v1.9.0 

func (w *NotifyWorker) WithGate(g sendingpolicy.Gate) *NotifyWorker

WithGate injects the sending-protection gate every notification must pass.

func (*NotifyWorker) WithOperationResolver added in v1.9.0 

func (w *NotifyWorker) WithOperationResolver(r OperationResolver) *NotifyWorker

WithOperationResolver injects the legacy-argument resolver.

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 OperationResolver added in v1.9.0 

type OperationResolver func(ctx context.Context, messageID string) (sendingpolicy.OperationRef, error)

OperationResolver recovers the durable operation for a job that carries no reference, through the same Prepare path an enqueue runs.

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

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.