Documentation
¶
Overview ¶
Package webhooknotify sends the two webhook health emails (docs/design/2026-08-08-webhook-health-notifications.md): an early WARNING when a webhook's deliveries start failing at the attempt level, and a DISABLED notice when the auto-disable breaker trips it.
It mirrors internal/hitlnotify's three-layer shape deliberately — that design (docs/design/hitl-notify-river.md) records why the naive inline send fails: a crash or SMTP outage between the state change and the send loses the notification forever. Here the maintenance sweep enqueues the webhook_notify job in the SAME transaction as the state transition (warn stamp / disable flip), and this worker recomposes and submits the email once off the sweep path, with River owning retries.
Index ¶
- Constants
- type DeliverOutcome
- type Deliverer
- type Jobs
- func (j *Jobs) Deliver(ctx context.Context, wh *identity.Webhook, kind string) DeliverOutcome
- func (j *Jobs) EnqueueDisabledTx(ctx context.Context, tx pgx.Tx, webhookID string) error
- func (j *Jobs) EnqueueWarningTx(ctx context.Context, tx pgx.Tx, webhookID string) error
- func (j *Jobs) EnqueueWebhookNotifyTx(ctx context.Context, tx pgx.Tx, webhookID, kind string) (int64, error)
- func (j *Jobs) RegisterJobs(w *river.Workers) []*river.PeriodicJob
- func (j *Jobs) SetDeliverer(d Deliverer)
- func (j *Jobs) SetEnqueuer(e jobs.Enqueuer)
- func (j *Jobs) WithMetrics(m Metrics) *Jobs
- type Metrics
- type Notifier
- type NotifierStore
- type NotifyWorker
- type Store
- type WebhookNotifyArgs
Constants ¶
const ( KindWarning = "warning" KindDisabled = "disabled" )
Notification kinds. One worker, two templates: the guards and the delivery/error triage are identical, and the bodies differ only in copy and severity. The Kind field in the job args is the seam where a third kind lands without new plumbing.
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 owner email — 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 hitlnotify's does, using the shared internal/outbound SMTP classifiers.
type Deliverer ¶
type Deliverer interface {
Deliver(ctx context.Context, wh *identity.Webhook, kind string) DeliverOutcome
}
Deliverer composes and sends one health email. Implemented by *Notifier (compose + SMTPRelay.SendOnce + classify).
type Jobs ¶
type Jobs struct {
// contains filtered or unexported fields
}
Jobs is the webhook health-notification integration on the shared River client: a jobs.Registrar (contributes NotifyWorker) plus the transactional enqueue entry points the maintenance sweep calls. Both the shared client and the concrete Deliverer are injected AFTER construction (two-phase wiring, mirroring hitlnotify.Jobs) — the client via SetEnqueuer, the Deliverer (the Notifier, which needs the relay resolved) via SetDeliverer. Jobs itself is the worker's Deliverer, late-binding to the concrete one, so a job enqueued during the startup window retries instead of hitting a nil deliverer.
Jobs also implements webhook.HealthNotifyEnqueuer (EnqueueDisabledTx / EnqueueWarningTx), which is how the AutoDisableWorker sweep reaches it without importing this package.
func (*Jobs) Deliver ¶
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.
func (*Jobs) EnqueueDisabledTx ¶
func (*Jobs) EnqueueWarningTx ¶
func (*Jobs) EnqueueWebhookNotifyTx ¶
func (j *Jobs) EnqueueWebhookNotifyTx(ctx context.Context, tx pgx.Tx, webhookID, kind string) (int64, error)
EnqueueWebhookNotifyTx inserts one webhook_notify job in the caller's transaction — the maintenance sweep's, so the state transition and its notification job commit atomically (the design's SC2 argument).
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 maintenance sweep is the only producer. Implements jobs.Registrar.
func (*Jobs) SetDeliverer ¶
SetDeliverer injects the concrete Deliverer (the Notifier), built after the relay gating resolves. Guarded so the River worker goroutines read it race-free.
func (*Jobs) SetEnqueuer ¶
SetEnqueuer injects the shared client so the EnqueueTx methods can insert jobs.
func (*Jobs) WithMetrics ¶
WithMetrics wires the observability backend the NotifyWorker emits the notification-outcome counter on. Nil-safe; call before RegisterJobs.
type Metrics ¶
type Metrics interface {
// WebhookNotify records one notification job outcome. kind ∈ {warning,
// disabled}; outcome ∈ {sent, permanent, outage, retryable, skipped}.
// skipped means a staleness guard decided not to send, which is
// counted apart from the failure outcomes on purpose: otherwise a fall
// in sends cannot be told apart from a send path that has died.
WebhookNotify(kind, outcome string)
}
Metrics is the narrow slice of telemetry.Metrics this worker emits (same pattern as internal/webhookdelivery.Metrics — declare only what is used, so tests wire a one-method fake). Satisfied by any telemetry backend.
type Notifier ¶
type Notifier struct {
// contains filtered or unexported fields
}
Notifier composes and sends the two webhook health emails. Construct with New; the NotifyWorker drives Deliver.
func New ¶
func New(store NotifierStore, r relay, fromDomain, fromAddress, replyTo, publicURL string) *Notifier
New returns a Notifier. fromDomain is cfg.OutboundSMTP.FromDomain (must be non-empty — the caller gates on it); fromAddress is the optional notifications.from_address config value, empty = fall back to the fixed local part on fromDomain; replyTo is the optional notifications.reply_to config value, empty = no Reply-To header. publicURL builds the dashboard link; empty degrades to generic copy.
func (*Notifier) Deliver ¶
Deliver composes and sends one health email, classifying the result for the NotifyWorker. Implements Deliverer.
func (*Notifier) FromAddress ¶
FromAddress exposes the resolved sender address (tests + startup log).
func (*Notifier) WithDKIM ¶
func (n *Notifier) WithDKIM(lookup outbound.DKIMKeyLookup) *Notifier
WithDKIM wires per-domain DKIM signing. The SMTP relay itself never DKIM-signs, and an upstream provider only signs identities IT manages — a custom notifications.from_address domain whose DKIM is published as a raw-key TXT record (BYODKIM, e2a holds the private key) is signed here or not at all. Without this leg the notification would ride on SPF alignment alone, and any SPF-breaking forward would land it in spam — the worst failure mode for the one email telling a customer their integration is broken.
nil-safe and fail-open via outbound.SignWithDKIM: no lookup, no stored key for the From domain, or a signing failure all send unsigned (the zero-config self-host path keeps working).
type NotifierStore ¶
type NotifierStore interface {
GetUserByID(ctx context.Context, id string) (*identity.User, error)
RecentWebhookFailureStats(ctx context.Context, webhookID string, window time.Duration) (identity.WebhookFailureStats, error)
}
NotifierStore is the read surface the notifier needs beyond the webhook row itself (which the worker passes in). *identity.Store satisfies it.
type NotifyWorker ¶
type NotifyWorker struct {
river.WorkerDefaults[WebhookNotifyArgs]
// contains filtered or unexported fields
}
NotifyWorker sends one health notification. Mirrors hitlnotify.NotifyWorker.
func NewNotifyWorker ¶
func NewNotifyWorker(store Store, deliverer Deliverer) *NotifyWorker
func (*NotifyWorker) NextRetry ¶
func (w *NotifyWorker) NextRetry(job *river.Job[WebhookNotifyArgs]) time.Time
NextRetry overrides River's default backoff with the notify envelope.
func (*NotifyWorker) WithMetrics ¶
func (w *NotifyWorker) WithMetrics(m Metrics) *NotifyWorker
WithMetrics swaps in a metrics backend. Nil-safe: unset (or nil) means no emission, so tests and self-host builds don't have to wire anything.
func (*NotifyWorker) Work ¶
func (w *NotifyWorker) Work(ctx context.Context, job *river.Job[WebhookNotifyArgs]) error
Work re-reads the webhook and applies the staleness guards before delivering. Every guard is a silent no-op (return nil): the email would be misleading if sent, so dropping it is the correct, fail-closed outcome. An unknown kind also refuses to send (fail-closed).
Every exit emits exactly one WebhookNotify sample, so the counter's total tracks job completions and no branch can go dark.
type Store ¶
type Store interface {
// GetWebhookByIDInternal loads the webhook with no ownership check —
// ownership was established when the sweep selected the row.
GetWebhookByIDInternal(ctx context.Context, webhookID string) (*identity.Webhook, error)
}
Store is the read surface the worker needs. *identity.Store satisfies it.
type WebhookNotifyArgs ¶
type WebhookNotifyArgs struct {
WebhookID string `json:"webhook_id"`
// NotifyKind ∈ {warning, disabled}. (Named NotifyKind because river's
// JobArgs interface reserves the Kind() method name.)
NotifyKind string `json:"kind"`
}
WebhookNotifyArgs drives one health-notification job. Args carry only the webhook id + kind; the worker re-reads the webhook row (the source of truth) each attempt, so the guards always see current state.
func (WebhookNotifyArgs) Kind ¶
func (WebhookNotifyArgs) Kind() string
Source Files