webhook

package
v1.7.1 Latest Latest
Warning

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

 Go to latest Published: Aug 8, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Index

Constants

View Source 
const DeliveryRetention = 30 * 24 * time.Hour

DeliveryRetention mirrors the 30-day expires_at default on webhook_subscriber_deliveries (migration 025) and the janitor that enforces it. Unlike messages — which are retained until trashed — delivery history is genuinely pruned, so a window reaching further back than this is reporting on rows that no longer exist.

View Source 
const DeliveryTTL = 48 * time.Hour
View Source 
const LeaseDuration = 5 * time.Minute

LeaseDuration is how long a leased delivery is hidden from other workers. On a clean delivery the row's next_retry_at is overwritten by the success path (MarkDelivered) or a real backoff (RecordFailure). The lease only matters as a recovery mechanism when a worker dies mid-delivery — after LeaseDuration the row becomes eligible again and another worker picks it up. Long enough that a legitimate slow webhook won't be double-fired, short enough that a crashed worker doesn't strand its rows for hours.

View Source 
const MaxDeliveryMetricsEndpoints = 50

MaxDeliveryMetricsEndpoints bounds the per-endpoint breakdown.

Variables

View Source 
var ErrDisallowedWebhookIP = errors.New("resolved to disallowed IP")

ErrDisallowedWebhookIP marks a dial-guard rejection (the URL resolved to a private/loopback/link-local address). Sentinel so transportErrorLabel can classify it into its customer-facing vocabulary without string matching; the wrapping error carries the concrete IP for process logs only — the customer-facing string never echoes it (a DNS-rebinding probe would otherwise learn internal addressing from the delivery history).

Functions

func IsDisallowedWebhookIP 

func IsDisallowedWebhookIP(ip net.IP) bool

IsDisallowedWebhookIP reports whether ip is in a range a webhook must never reach (loopback, RFC-1918 private, link-local incl. the cloud metadata endpoint 169.254.169.254, CGNAT shared space, multicast, unspecified, IPv6 ULA). Shared by registration-time validation (agent.ValidateWebhookURL) and the delivery-time dial guard below so the two can never drift — closing the DNS-rebinding window where a host validates as public at registration then re-resolves to an internal address before delivery.

Types

type AccountDeliveryMetrics added in v1.7.0 

type AccountDeliveryMetrics struct {
	Totals    DeliveryOutcomeCounts
	Endpoints []EndpointDeliveryMetrics
	// EndpointsTruncated reports that the account has more endpoints with
	// traffic than Endpoints contains. Totals stay complete.
	EndpointsTruncated bool
	// EndpointsAutoDisabled counts endpoints e2a has auto-disabled. Computed
	// across every endpoint the account owns, not just the listed ones, and
	// independent of whether the endpoint had traffic in this window.
	EndpointsAutoDisabled int64

	// WindowExceedsRetention is true when the requested window starts before
	// the delivery-retention horizon, so older rows have been pruned and the
	// counts below understate that stretch. Without this a 90-day view looks
	// like a collapse in webhook volume rather than a retention boundary.
	WindowExceedsRetention bool
}

AccountDeliveryMetrics is the account-wide webhook aggregate.

type AutoDisableWorker 

type AutoDisableWorker struct {
	// contains filtered or unexported fields
}

AutoDisableWorker scans for chronically-failing webhooks and disables them, warns owners whose webhooks have started failing at the attempt level (long before the breaker can trip), and clears expired signing_secret_prev rows past their 24h grace window.

The passes share a worker because they're all cheap, idempotent, and run on the same low cadence. The schedule is owned by River (webhookdelivery.MaintenanceJobs, a periodic on QueueMaintenance) which drives Tick; this type is the sweep body.

func NewAutoDisableWorker 

func NewAutoDisableWorker(store *identity.Store) *AutoDisableWorker

NewAutoDisableWorker constructs the sweep. River drives Tick on a periodic schedule; tests can call Tick directly.

func (*AutoDisableWorker) SetNotifier added in v1.7.0 

func (w *AutoDisableWorker) SetNotifier(n HealthNotifyEnqueuer)

SetNotifier injects the health-notification enqueuer (two-phase wiring: the concrete *webhooknotify.Jobs is built alongside the registrars, before the shared River client exists). nil-safe — unset means the sweep transitions state without enqueuing notifications.

func (*AutoDisableWorker) Tick 

func (w *AutoDisableWorker) Tick(ctx context.Context)

Tick runs the maintenance passes once. Driven by the River periodic (and directly by tests).

Order matters: the disable pass runs BEFORE the warn pass so that a burst crossing both thresholds inside one sweep interval produces only the disable email — the warn pass's enabled = true predicate excludes the rows this tick just disabled. (The notify worker's kind=warning guard drops a stale warning against a since-disabled webhook as a second line of defense.)

type Delivery 

type Delivery struct {
	AgentID       string     `json:"agent_id"`
	MessageID     string     `json:"message_id"`
	Status        string     `json:"status"`
	Attempts      int        `json:"attempts"`
	MaxAttempts   int        `json:"max_attempts"`
	LastError     string     `json:"last_error"`
	LastAttemptAt *time.Time `json:"last_attempt_at,omitempty"`
	NextRetryAt   time.Time  `json:"next_retry_at"`
	CreatedAt     time.Time  `json:"created_at"`
	ExpiresAt     time.Time  `json:"expires_at"`
}

type DeliveryOutcome 

type DeliveryOutcome struct {
	Success    bool
	StatusCode int
	Error      string
}

DeliveryOutcome is what the deliverer returns to the caller for status accounting. statusCode is 0 when there was no HTTP response (connection error, timeout, DNS failure).

type DeliveryOutcomeCounts added in v1.7.0 

type DeliveryOutcomeCounts struct {
	Total            int64
	Delivered        int64
	Pending          int64
	EndpointRejected int64
	NoResponse       int64
}

DeliveryOutcomeCounts is one population's delivery tally.

The two failure buckets are what the TABLE can actually prove, which is not the same split the internal telemetry makes. The worker labels its metrics endpoint_failure vs e2a_failure, but both write status='failed' with no stored discriminator, so that attribution cannot be recovered here. What survives is whether the endpoint ANSWERED:

  • EndpointRejected: the endpoint returned a non-2xx (last_status_code is a real HTTP status). Unambiguously the subscriber's own response.
  • NoResponse: no HTTP response was ever received — connect/DNS/TLS failure, an SSRF-blocked URL, a delivery that expired while pending, or (rarely) an e2a-side error. Predominantly an unreachable endpoint, but NOT a clean "your fault" bucket, and it must not be presented as one.

func (DeliveryOutcomeCounts) Failed added in v1.7.0 

func (c DeliveryOutcomeCounts) Failed() int64

Failed is the terminal failure count across both attributable buckets.

type DeliveryStore 

type DeliveryStore struct {
	// contains filtered or unexported fields
}

func NewDeliveryStore 

func NewDeliveryStore(pool *pgxpool.Pool) *DeliveryStore

func (*DeliveryStore) CreateDelivery 

func (s *DeliveryStore) CreateDelivery(ctx context.Context, messageID string, lastError string) (*Delivery, error)

func (*DeliveryStore) DeleteExpiredDeliveries 

func (s *DeliveryStore) DeleteExpiredDeliveries(ctx context.Context) (int64, error)

func (*DeliveryStore) GetPendingDeliveries 

func (s *DeliveryStore) GetPendingDeliveries(ctx context.Context, limit int) ([]Delivery, error)

GetPendingDeliveries atomically claims up to `limit` due deliveries. Each returned row's next_retry_at is pushed by LeaseDuration so other workers (in this process or a different replica) won't grab the same row. The standard `WHERE status='pending' AND next_retry_at <= now()` filter then naturally excludes leased rows.

This must run inside a transaction: `FOR UPDATE SKIP LOCKED` only holds the row lock for the lifetime of the surrounding transaction. pool.Query (autocommit) would release the lock as soon as the SELECT completed, leaving a window where two callers could each return the same row.

func (*DeliveryStore) MarkAttemptFailed 

func (s *DeliveryStore) MarkAttemptFailed(ctx context.Context, messageID, errMsg string, nextRetry time.Time) error

func (*DeliveryStore) MarkDelivered 

func (s *DeliveryStore) MarkDelivered(ctx context.Context, messageID string) error

func (*DeliveryStore) MarkFailed 

func (s *DeliveryStore) MarkFailed(ctx context.Context, messageID, errMsg string) error

type EndpointDeliveryMetrics added in v1.7.0 

type EndpointDeliveryMetrics struct {
	WebhookID string
	// URLHost is the endpoint's host only. The full URL can carry credentials
	// in its path or query string, and a metrics payload is the last place
	// that should be re-emitted — the ID identifies the webhook well enough.
	URLHost string
	Counts  DeliveryOutcomeCounts
	// LastStatusCode is the most recent HTTP status observed for this
	// endpoint in the window, or nil when nothing ever answered. A constant
	// 405 or 401 tells a customer exactly what to fix.
	LastStatusCode *int32

	// Enabled / AutoDisabledAt / AutoDisableReason mirror the webhook's health
	// state. An endpoint e2a auto-disabled after sustained failure is the most
	// actionable fact on this whole page — it means events are being dropped
	// right now, not merely retried — so it travels with the counts rather
	// than living only on the webhooks screen.
	Enabled           bool
	AutoDisabledAt    *time.Time
	AutoDisableReason string
}

EndpointDeliveryMetrics is one subscriber endpoint's slice.

type HealthNotifyEnqueuer added in v1.7.0 

type HealthNotifyEnqueuer interface {
	// EnqueueDisabledTx enqueues the "we disabled your webhook" email job.
	EnqueueDisabledTx(ctx context.Context, tx pgx.Tx, webhookID string) error
	// EnqueueWarningTx enqueues the "your webhook is failing" early-warning
	// email job.
	EnqueueWarningTx(ctx context.Context, tx pgx.Tx, webhookID string) error
}

HealthNotifyEnqueuer enqueues webhook health-notification jobs inside the sweep's transaction — one per state transition, atomically with it. Implemented by *webhooknotify.Jobs; nil (the default) leaves the sweep's state transitions running with no notifications, exactly the pre-feature behavior (self-hosts without an SMTP relay).

type Payload 

type Payload struct {
	MessageID      string `json:"message_id,omitempty"`
	ConversationID string `json:"conversation_id,omitempty"`
	From           string `json:"from"`
	// To is the parsed To: header from the inbound message — every fan-out
	// delivery for one inbound message carries the same list. delivered_to is
	// this delivery's per-agent target — the envelope Delivered-To address
	// (always one of the addressed agents, not necessarily in To: when the
	// agent was Bcc'd).
	To []string `json:"to"`
	CC []string `json:"cc,omitempty"`
	// ReplyTo is the parsed Reply-To: header (RFC 5322 § 3.6.2 — list, single
	// value is typical but multi is legal). Empty list when the header is
	// absent; the relay never silently falls back to From: so consumers can
	// distinguish "sender didn't request a different reply mailbox" from
	// "sender explicitly named these mailboxes".
	ReplyTo     []string          `json:"reply_to,omitempty"`
	Recipient   string            `json:"delivered_to"`
	RawMessage  []byte            `json:"raw_message"`
	AuthHeaders map[string]string `json:"auth_headers"`
	ReceivedAt  time.Time         `json:"received_at"`
}

type SubscriberDeliverer 

type SubscriberDeliverer struct {
	// contains filtered or unexported fields
}

SubscriberDeliverer performs the HTTP POST for a webhook_subscriber_deliveries row, signs the request with the per-webhook HMAC secret, and reports success / failure to the caller. Distinct from the retired legacy per-agent delivery path.

Slice 1 carries only the current secret. Slice 4 will extend this to dual-sign during the 24h rotation grace window.

func NewSubscriberDeliverer 

func NewSubscriberDeliverer(requireHTTPS bool, internalSinkURL string) *SubscriberDeliverer

NewSubscriberDeliverer constructs the deliverer with the 15s per-attempt timeout chosen in design decision #6.

requireHTTPS gates against plaintext URLs in production. The same flag installs a dial-time IP guard (guardedDialControl): registration-time ValidateWebhookURL validates DNS once, but a hostname can re-resolve to an internal IP before delivery (DNS rebinding). The guard re-checks the actual resolved IP at connect time, closing that window. It is gated to production so local/CI deliveries to 127.0.0.1 still work.

internalSinkURL (usually empty) names ONE trusted internal sink — the e2a-prober's /sink — reached over plain HTTP on an internal host. Deliveries to that EXACT URL bypass the HTTPS + SSRF guards via a separate exemptClient. This is safe because: (1) the value is server-operator config, never attacker input; (2) it is matched by exact string equality, so it grants access to no other internal address; and (3) the probe webhook that targets it is created by the privileged prober `seed`, not the public registration API (which rejects http:// + private hosts). Empty disables the exemption entirely.

func (*SubscriberDeliverer) Deliver 

func (d *SubscriberDeliverer) Deliver(ctx context.Context, url string, body []byte, secret, secretPrev, eventType, schemaVersion string) DeliveryOutcome

Deliver performs one POST attempt. It signs the request body with the supplied HMAC secret in Stripe-style header format: 

X-E2A-Signature: t=<unix>,v1=<hex(hmac-sha256(secret, "<t>.<body>"))>

secretPrev (if non-empty) adds a second v1=... signature for the receiver to verify against during the 24h rotation grace window. Slice 1 always passes secretPrev="" (no grace logic yet); slice 4 wires this up.

2xx responses are success. Anything else (including 3xx, since redirects are blocked) is a failure with the HTTP status code reported back. Connection errors return Success=false and StatusCode=0.

type SubscriberDelivery 

type SubscriberDelivery struct {
	ID             string
	WebhookID      string
	EventType      string
	EventPayload   []byte // pre-marshalled envelope bytes; POSTed verbatim
	MessageID      *string
	Status         string // pending | delivered | failed
	Attempts       int
	MaxAttempts    int
	LastError      string
	LastStatusCode *int
	LastAttemptAt  *time.Time
	NextRetryAt    time.Time
	CreatedAt      time.Time
	ExpiresAt      time.Time
	// ReplayID is non-nil for customer-initiated replay rows (a re-delivery
	// of an old event), nil for first-delivery rows. The first-attempt
	// latency SLI observes first-delivery rows only: a replay's baseline
	// would be the ORIGINAL event's created_at, recording the replay lag
	// as a false outlier.
	ReplayID *string
	// EventCreatedAt is the originating webhook_events row's created_at —
	// the baseline of the event→first-attempt latency SLI. Nil for rows
	// without an event link (the /test endpoint's deliveries).
	EventCreatedAt *time.Time
}

SubscriberDelivery is one row in webhook_subscriber_deliveries. Distinct from the legacy Delivery struct (which is keyed by message_id and tracks legacy single-URL delivery state).

type SubscriberStore 

type SubscriberStore struct {
	// contains filtered or unexported fields
}

SubscriberStore manages webhook_subscriber_deliveries. Parallel to the legacy DeliveryStore (which manages webhook_deliveries).

func NewSubscriberStore 

func NewSubscriberStore(pool *pgxpool.Pool) *SubscriberStore

func (*SubscriberStore) CountDeliveriesForAccount added in v1.7.0 

func (s *SubscriberStore) CountDeliveriesForAccount(ctx context.Context, userID string, start, end time.Time) (AccountDeliveryMetrics, error)

CountDeliveriesForAccount aggregates webhook delivery outcomes for one account over a window.

The window is anchored on the DELIVERY row's own created_at, not on the originating message's. Deliveries outlive their messages by design — message_id is ON DELETE SET NULL precisely so history survives the message janitor — so anchoring on the message would silently drop every delivery whose message had been pruned.

The grain is one row per (event, subscriber) pair: an account with three webhooks matching the same event produces three rows for one message. These counts therefore exceed message counts and must never be presented as duplicate mail.

func (*SubscriberStore) DeleteExpiredSubscriberDeliveries 

func (s *SubscriberStore) DeleteExpiredSubscriberDeliveries(ctx context.Context) (deleted, marked int, err error)

DeleteExpiredSubscriberDeliveries enforces the 30-day TTL (migration 025) in two phases; without it the table grows monotonically and query plans degrade. Mirrors DeliveryStore.DeleteExpiredDeliveries for the legacy table, with one refinement:

  1. DELETE expired rows already in a terminal state (delivered/failed) — as before.
  2. Expired rows still 'pending' are NOT deleted but marked terminally 'failed' ("expired before delivery"), so a delivery record never silently vanishes while pending: the transition is counted (the caller logs it and emits the WebhookExpiredPending metric) and visible in the delivery-history API until the NEXT sweep deletes the now-terminal row.

Phase 2's population should be ~empty now that the delivery reconciler also rescues dead-job strands; what remains is essentially rows snoozing behind a webhook disabled for longer than the TTL. Marking those failed (instead of a bare `status <> 'pending'` guard on the DELETE) keeps them from becoming immortal — unbounded growth behind a permanently-disabled webhook. Phase 2 runs AFTER phase 1 so a freshly marked row survives until the following sweep. The DeliverWorker treats any non-pending row as a no-op, so a still-snoozing job waking up later never POSTs a row marked here.

Returns the rows deleted (phase 1) and the rows marked failed (phase 2).

func (*SubscriberStore) GetSubscriberDeliveryByID 

func (s *SubscriberStore) GetSubscriberDeliveryByID(ctx context.Context, deliveryID string) (*SubscriberDelivery, error)

GetSubscriberDeliveryByID loads a single delivery row by id — the River DeliverWorker's entry point (it holds only the delivery id and reads the payload + webhook_id here). Returns pgx.ErrNoRows if the row is gone. The replay_id + joined webhook_events.created_at feed the event→first-attempt latency SLI (LEFT JOIN: rows without an event link, e.g. /test deliveries, load a nil EventCreatedAt).

func (*SubscriberStore) InsertPendingForTest 

func (s *SubscriberStore) InsertPendingForTest(ctx context.Context, webhookID, eventType string, envelope []byte) (string, error)

InsertPendingForTest creates a single delivery row tied to the given webhook + event type with the supplied envelope bytes. The retry worker picks it up on the next tick. Used by the POST /v1/webhooks/{id}/test endpoint to schedule a one-off delivery without going through the publisher's filter-matching path.

func (*SubscriberStore) ListDeliveriesByWebhook 

func (s *SubscriberStore) ListDeliveriesByWebhook(ctx context.Context, webhookID, status string, limit int, afterCreatedAt time.Time, afterID string) ([]SubscriberDelivery, error)

ListDeliveriesByWebhook returns one page of delivery rows for the webhook, most-recent first, keyset-paginated on (created_at, id). When status is non-empty, restricts to that status (pending|delivered|failed). The caller passes limit (fetch limit+1 to detect a further page) and the after-key from the previous page's last row (zero afterCreatedAt = first page). The delivery log grows unbounded on a busy webhook, so it needs real pagination rather than silently truncating at a fixed cap. Limit is bounded by the caller; this method does not enforce a cap.

func (*SubscriberStore) MarkDelivered 

func (s *SubscriberStore) MarkDelivered(ctx context.Context, deliveryID string, statusCode int) error

MarkDelivered transitions a row to status='delivered' and stamps last_attempt_at + last_status_code. Also bumps webhooks.last_delivered_at in the same transaction so list views show the freshest activity.

func (*SubscriberStore) MarkDeliveredIfPending added in v1.4.1 

func (s *SubscriberStore) MarkDeliveredIfPending(ctx context.Context, deliveryID string, statusCode int) (bool, error)

MarkDeliveredIfPending is the transition-aware form used by the delivery worker's terminal SLI. It returns true only when this call changed a pending row to delivered, so a duplicate River execution cannot count the same delivery twice.

func (*SubscriberStore) MarkSubscriberFailed 

func (s *SubscriberStore) MarkSubscriberFailed(ctx context.Context, deliveryID string, attemptN int, errMsg string, statusCode int) error

MarkSubscriberFailed transitions a delivery to terminal 'failed' — called by the DeliverWorker on the last (River-exhausted) attempt. Records the final attempt count + error.

func (*SubscriberStore) MarkSubscriberFailedIfPending added in v1.3.0 

func (s *SubscriberStore) MarkSubscriberFailedIfPending(ctx context.Context, deliveryID string, attemptN int, errMsg string, statusCode int) error

MarkSubscriberFailedIfPending is MarkSubscriberFailed conditional on the row still being 'pending' — the BLIND terminal write for the final-attempt row-load failure, where the read error means the row's true state is unknown. It must never clobber a row that already reached a terminal state (e.g. delivered by a path the failed read couldn't see). A missing row is a no-op, not an error.

func (*SubscriberStore) RecordSubscriberAttempt 

func (s *SubscriberStore) RecordSubscriberAttempt(ctx context.Context, deliveryID string, attemptN int, errMsg string, statusCode int) error

RecordSubscriberAttempt records ONE failed attempt without deciding retry or terminality — under River the retry schedule and the give-up decision belong to the job, not the store. Status stays 'pending'; attempts/last_error/ last_status_code/last_attempt_at are updated. (Contrast RecordAttemptFailure, the hand-rolled path, which also computed next_retry_at and flipped to 'failed' at the cap — retired with the legacy worker.) attemptN is the River job's attempt number, written verbatim so the history API's attempts count matches River's.

The status='pending' guard makes the write a no-op on a row that terminalized mid-attempt: without it, a worker mid-POST racing the expiry janitor's mark-failed would resurrect the row to 'pending' when its non-final attempt failed. Semantically the method only ever applies to in-flight rows anyway.

func (*SubscriberStore) TransitionSubscriberFailedIfPending added in v1.4.1 

func (s *SubscriberStore) TransitionSubscriberFailedIfPending(ctx context.Context, deliveryID string, attemptN int, errMsg string, statusCode int) (bool, error)

TransitionSubscriberFailedIfPending is the transition-aware form used by terminal SLI emission. A false result means the row was already terminal or gone, so callers must not increment the per-delivery terminal counter.

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.