webhooknotify

package
v1.9.0 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: 16 Imported by: 0

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

View Source 
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.

View Source 
const MaxNotifyAttempts = 6

MaxNotifyAttempts caps the retry tail before River discards the job.

Variables

This section is empty.

Functions

func ExpectedOperationID added in v1.9.0 

func ExpectedOperationID(wh *identity.Webhook, kind string) string

ExpectedOperationID is the operation a notice of the given kind for this webhook's current episode must carry: the same derivation the gate's PrepareNotificationTx uses. Empty when the episode was never stamped.

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.

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

Deliverer is the two-phase send of one health email. Compose does every fallible, provider-free step (owner lookup, failure stats, MIME, DKIM) and returns the envelope; Submit hands that envelope and a freshly consumed authorization to the provider seam. The split lets the worker ConsumeAttempt immediately before the socket opens, so a compose failure costs no charged ordinal. Implemented by *Notifier.

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 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, wh *identity.Webhook, kind string) (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) EnqueueDisabledTx 

func (j *Jobs) EnqueueDisabledTx(ctx context.Context, tx pgx.Tx, webhookID string) error

func (*Jobs) EnqueueWarningTx 

func (j *Jobs) EnqueueWarningTx(ctx context.Context, tx pgx.Tx, webhookID string) error

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).

With a gate wired the notification's operation is prepared here, against the locked webhook row, so the owning account is charged 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) 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) ResolveLegacyOperation added in v1.9.0 

func (j *Jobs) ResolveLegacyOperation(ctx context.Context, webhookID, kind 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 the sweep's enqueue runs.

func (*Jobs) SetDeliverer 

func (j *Jobs) SetDeliverer(d Deliverer)

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 

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

SetEnqueuer injects the shared client so the EnqueueTx methods 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. Chainable; nil keeps the gateless default.

func (*Jobs) WithMetrics 

func (j *Jobs) WithMetrics(m Metrics) *Jobs

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, s submitter, 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) Compose added in v1.9.0 

func (n *Notifier) Compose(ctx context.Context, wh *identity.Webhook, kind string) (outbound.Envelope, DeliverOutcome)

Compose implements Deliverer: the provider-free half (owner lookup, failure stats, MIME, Message-ID, DKIM), classified like a send so the worker treats a permanent compose failure the same way.

func (*Notifier) Deliver 

func (n *Notifier) Deliver(ctx context.Context, wh *identity.Webhook, kind string, auth sendingpolicy.ProviderAuthorization) DeliverOutcome

Deliver composes and sends one health email with an already-authorized attempt: Compose then Submit in one call, for callers that hold the token up front (tests). The worker runs the two phases itself so the token is consumed last.

func (*Notifier) FromAddress 

func (n *Notifier) FromAddress() string

FromAddress exposes the resolved sender address (tests + startup log).

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 NotifyWorker.

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) 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[WebhookNotifyArgs]) time.Time

NextRetry overrides River's default backoff with the 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

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. WithGate injects the sending-protection gate every notification must pass.

func (*NotifyWorker) WithMetrics 

func (w *NotifyWorker) WithMetrics(m Metrics) *NotifyWorker

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

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

OperationResolver recovers the durable operation for a job that carries no reference, through the same Prepare path the sweep's enqueue runs. The kind selects the episode (warning or disable) the operation is keyed by.

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"`
	// OperationRef is the durable sending operation the sweep's transaction
	// prepared; a job from a pre-floor slot carries none and is resolved at
	// fire time, then stamped.
	OperationRef *sendingpolicy.OperationRef `json:"operation_ref,omitempty"`
	// 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

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.