webhook

package
v1.1.0 Latest Latest
Warning

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

 Go to latest Published: Jul 21, 2026 License: Apache-2.0 Imports: 15 Imported by: 0

Documentation

Index

Constants

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.

Variables

This section is empty.

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 AutoDisableWorker 

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

AutoDisableWorker scans for chronically-failing webhooks and disables them, and clears expired signing_secret_prev rows past their 24h grace window. Decision #12 in the design.

The two passes share a worker because they're both 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) Tick 

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

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

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 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 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
}

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

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

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.

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) 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, and as an ErrorHandler backstop on discard. Records the final attempt count + 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.

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.