webhook

package
v1.3.0 Latest Latest
Warning

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

 Go to latest Published: Jul 28, 2026 License: MIT Imports: 21 Imported by: 0

Documentation

Overview

Package webhook delivers events to external services over managed subscriptions: consumers register an HTTPS endpoint and receive every matching envelope as a signed POST, retried with exponential backoff and dead-lettered after a cap. One delivery row exists per envelope × subscription, so a dead consumer never affects a healthy one. Design: docs/design/event-delivery.md.

Index

Constants

View Source 
const (
	StatusPending   = "pending"
	StatusInflight  = "inflight"
	StatusDelivered = "delivered"
	StatusDead      = "dead"
)

Delivery statuses.

View Source 
const EntityName = "webhook_subscription"

EntityName is the activity-log entity for subscription changes.

Variables

This section is empty.

Functions

func Backoff 

func Backoff(attempts int) time.Duration

Backoff returns the exponential, jittered delay before attempt n+1: 1s, 4s, 16s, … capped at 15 minutes, ±20% jitter so replicas don't retry in lockstep.

Types

type ClaimedDelivery 

type ClaimedDelivery struct {
	Delivery
	Envelope events.Envelope
	// Endpoint is the subscription's URL/secret snapshot at claim time.
	URL    string
	Secret string
	// LeaseExpiresAt is the lease this claim took. It is the worker's proof
	// of ownership: it travels back on the Outcome so Record can refuse to
	// write the result of a delivery whose lease has since been released and
	// re-claimed by another worker.
	LeaseExpiresAt time.Time
}

ClaimedDelivery is a delivery the worker owns, joined with everything needed to POST without further reads.

type CreateInput 

type CreateInput struct {
	Name       string
	URL        string
	Secret     string
	EventTypes []string
	Active     *bool // nil defaults to true
}

CreateInput registers a new subscription.

type Delivery 

type Delivery struct {
	ID             ulid.ID   `json:"id"`
	SubscriptionID ulid.ID   `json:"subscription_id"`
	EnvelopeID     string    `json:"envelope_id"`
	TenantID       string    `json:"tenant_id"`
	EventType      string    `json:"event_type"`
	FeedSeq        int64     `json:"feed_seq"`
	Status         string    `json:"status"`
	Attempts       int       `json:"attempts"`
	NextAttemptAt  time.Time `json:"next_attempt_at"`
	LastError      string    `json:"last_error,omitempty"`
	ResponseCode   int       `json:"response_code,omitempty"`
	CreatedAt      time.Time `json:"created_at"`
	UpdatedAt      time.Time `json:"updated_at"`
}

Delivery is one attempt-tracked (envelope × subscription) pair.

type DeliveryFilter 

type DeliveryFilter struct {
	TenantID       valueobjects.TenantID
	SubscriptionID ulid.ID
	Status         string
}

DeliveryFilter narrows delivery listings.

type DeliveryStore 

type DeliveryStore interface {
	// ClaimDue atomically marks up to limit due deliveries inflight (with a
	// lease) and returns them joined with envelope and endpoint. At most
	// one delivery per subscription is claimed, in feed order, and never
	// while another delivery of the same subscription is inflight — this
	// is what keeps per-subscription ordering in the happy path.
	ClaimDue(ctx context.Context, limit int, leaseFor time.Duration, now time.Time) ([]ClaimedDelivery, error)

	// Record persists attempt outcomes for the deliveries the caller still
	// owns. An outcome whose lease no longer matches the stored one is
	// skipped, and reported in the returned count of lost claims.
	//
	// Without the ownership predicate a worker whose lease lapsed mid-POST
	// overwrote the new owner's row — including rewinding a delivered row to
	// pending with a next_attempt_at, which caused a third send.
	Record(ctx context.Context, now time.Time, outcomes ...Outcome) (lost int, err error)

	// ReleaseExpired returns inflight deliveries whose lease lapsed (a
	// worker crashed mid-delivery) to pending.
	ReleaseExpired(ctx context.Context, now time.Time) (int, error)

	List(ctx context.Context, filter DeliveryFilter, page db.Page) ([]Delivery, int, error)

	// Redeliver returns a dead (or delivered) delivery to pending now.
	Redeliver(ctx context.Context, tenant valueobjects.TenantID, id ulid.ID, now time.Time) error

	// RedeliverMatching returns every dead delivery matching the filter to
	// pending, and reports how many it moved.
	//
	// Recovery used to be one API call per dead delivery. After an endpoint
	// outage the dead letters number in the thousands, so the only recovery
	// path was a script that discovers ids and calls the API once each —
	// which is the tool an operator has to write while the incident is
	// still open.
	RedeliverMatching(ctx context.Context, filter DeliveryFilter, now time.Time) (int, error)
}

DeliveryStore persists delivery state.

type Interactor 

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

Interactor implements the subscription-management usecases.

func NewInteractor 

func NewInteractor(u uow.UnitOfWork, subs SubscriptionStore, deliveries DeliveryStore, policy URLPolicy) *Interactor

NewInteractor wires the webhook usecases. policy governs which subscription URLs are accepted (SSRF guard).

func (*Interactor) Create 

func (i *Interactor) Create(ctx context.Context, in CreateInput) (*Subscription, error)

Create registers a webhook subscription.

func (*Interactor) Delete 

func (i *Interactor) Delete(ctx context.Context, rawID string) error

Delete removes a subscription and its delivery history.

func (*Interactor) Ensure 

func (i *Interactor) Ensure(ctx context.Context, in CreateInput) (*Subscription, error)

Ensure upserts a subscription by name — the bootstrap path for environment-configured endpoints. Existing subscriptions get the given URL/secret/filters; missing ones are created.

func (*Interactor) Get 

func (i *Interactor) Get(ctx context.Context, rawID string) (*Subscription, error)

Get returns one subscription.

func (*Interactor) List 

func (i *Interactor) List(ctx context.Context) ([]Subscription, error)

List returns the tenant's subscriptions.

func (*Interactor) ListDeliveries 

func (i *Interactor) ListDeliveries(ctx context.Context, in ListDeliveriesInput) (*ListDeliveriesOutput, error)

ListDeliveries pages a subscription's delivery log.

func (*Interactor) Redeliver 

func (i *Interactor) Redeliver(ctx context.Context, rawID string) error

Redeliver returns a dead or delivered delivery to the queue.

func (*Interactor) RedeliverDead added in v1.3.0 

func (i *Interactor) RedeliverDead(ctx context.Context, rawSubscriptionID string) (int, error)

RedeliverDead returns every dead delivery of a subscription to pending, and reports how many it moved. An empty subscription id covers the tenant.

This is the bulk counterpart of Redeliver. After an endpoint outage the dead letters number in the thousands, and one API call each is not a recovery path — it is a script an operator writes during the incident.

func (*Interactor) Update 

func (i *Interactor) Update(ctx context.Context, in UpdateInput) (*Subscription, error)

Update mutates a subscription.

type ListDeliveriesInput 

type ListDeliveriesInput struct {
	SubscriptionID string
	Status         string
	Page           db.PageArgs
}

ListDeliveriesInput filters the delivery log.

type ListDeliveriesOutput 

type ListDeliveriesOutput struct {
	Items    []Delivery
	PageInfo db.PageInfo
}

ListDeliveriesOutput is one page of deliveries.

type Outcome 

type Outcome struct {
	DeliveryID ulid.ID
	// LeaseExpiresAt is the lease the worker held when it made the attempt.
	// Record writes only while it still matches the stored lease, so a worker
	// whose lease lapsed cannot overwrite the state of the worker that took
	// over.
	LeaseExpiresAt time.Time
	Delivered      bool
	ResponseCode   int
	Err            string
	// NextAttemptAt schedules the retry when not delivered; ignored when
	// Dead is set.
	NextAttemptAt time.Time
	Dead          bool
}

Outcome records one delivery attempt.

type Subscription 

type Subscription struct {
	ID       ulid.ID               `json:"id"`
	TenantID valueobjects.TenantID `json:"tenant_id"`
	Name     string                `json:"name"`
	URL      string                `json:"url"`
	Secret   string                `json:"-"`
	// PreviousSecret is always empty. It is neither read nor written.
	//
	// Deprecated: a delivery carries one signature, computed with Secret,
	// so a second stored secret was never consulted when signing. The
	// rotation grace window is on the receiving side — see VerifyRequest,
	// which accepts a list of secrets — and the rotation order is in
	// docs/design/event-delivery.md. The field and its column are retained
	// until the next major version, so that a rollback to an older binary
	// still finds the column.
	PreviousSecret string `json:"-"`
	// EventTypes filters deliveries; empty means every event.
	EventTypes []string  `json:"event_types"`
	Active     bool      `json:"active"`
	CreatedAt  time.Time `json:"created_at"`
	UpdatedAt  time.Time `json:"updated_at"`
}

Subscription is one registered webhook endpoint.

func (Subscription) Matches 

func (s Subscription) Matches(eventType string) bool

Matches reports whether the subscription wants this event type.

func (Subscription) Validate 

func (s Subscription) Validate(policy URLPolicy) error

Validate checks the subscription's shape and its URL against the policy.

type SubscriptionStore 

type SubscriptionStore interface {
	WithTx(tx db.Tx) SubscriptionStore
	Get(ctx context.Context, tenant valueobjects.TenantID, id ulid.ID) (Subscription, error)
	GetByName(ctx context.Context, tenant valueobjects.TenantID, name string) (Subscription, error)
	List(ctx context.Context, tenant valueobjects.TenantID) ([]Subscription, error)
	// ListActive returns every active subscription across tenants — the
	// expansion step fans envelopes out against this set.
	ListActive(ctx context.Context) ([]Subscription, error)
	Create(ctx context.Context, s Subscription) error
	Update(ctx context.Context, s Subscription) error
	Delete(ctx context.Context, tenant valueobjects.TenantID, id ulid.ID) error
}

SubscriptionStore persists subscriptions. WithTx binds the store to a transaction for writes inside a unit of work.

type URLPolicy 

type URLPolicy struct {
	// AllowPrivate permits http and private/loopback/link-local hosts —
	// for on-prem deployments whose consumers live on internal networks.
	// The delivery worker's dialer guard must be relaxed in step.
	AllowPrivate bool
}

URLPolicy governs which subscription URLs are accepted. The zero value is the safe default: https only, no private/loopback/link-local hosts.

type UpdateInput 

type UpdateInput struct {
	ID         string
	URL        *string
	EventTypes *[]string
	Active     *bool
	// RotateSecret installs a new secret. It is a HARD CUTOVER: the previous
	// secret stops signing immediately, so update the receiver first, then
	// rotate. An earlier design kept a grace window and this comment
	// outlived it by eleven lines — an embedder who read the field would
	// rotate first and update receivers afterwards, which is exactly the
	// outage the grace window had existed to prevent.
	RotateSecret *string
}

UpdateInput mutates a subscription. Nil pointers leave fields unchanged.

type Worker 

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

Worker drains due deliveries: claim (short tx) → POST (no tx) → record (short tx). Crash recovery comes from the inflight lease — expired leases return to pending via ReleaseExpired. Every flexitype replica runs workers; SKIP LOCKED claims keep them from colliding.

func NewWorker 

func NewWorker(deliveries DeliveryStore, opts ...WorkerOption) *Worker

NewWorker builds a delivery worker over the store. The default HTTP client refuses non-public targets (SSRF guard); override with WithHTTPClient (e.g. safedial.NewClient with AllowPrivate for on-prem).

func (*Worker) Nudge 

func (w *Worker) Nudge()

Nudge wakes the worker immediately — the relay calls it after expansion so happy-path latency stays milliseconds.

func (*Worker) Run 

func (w *Worker) Run(ctx context.Context)

Run processes deliveries until ctx ends.

type WorkerOption 

type WorkerOption func(*Worker)

WorkerOption customises a Worker.

func WithHTTPClient 

func WithHTTPClient(c *http.Client) WorkerOption

WithHTTPClient overrides the delivery client (default 10s timeout).

func WithMaxAttempts 

func WithMaxAttempts(n int) WorkerOption

WithMaxAttempts sets the dead-letter cap.

The default of 25 gives a retry window of about 5 hours, not the 3 days this comment used to claim: the backoff is 1s, 4s, 16s, 64s, 256s and then capped at 15 minutes, so 25 attempts is 5m45s + 19x15m. Sizing a window for an overnight outage means raising the cap or the ceiling, and the arithmetic is stated here so the next person does not have to rediscover it.

func WithWorkerConcurrency 

func WithWorkerConcurrency(n int) WorkerOption

WithWorkerConcurrency sets parallel deliveries per pass (default 4).

func WithWorkerErrorObserver 

func WithWorkerErrorObserver(fn func(error)) WorkerOption

WithWorkerErrorObserver receives worker-level failures (claim/record errors — delivery failures are recorded per row).

func WithWorkerInterval 

func WithWorkerInterval(d time.Duration) WorkerOption

WithWorkerInterval sets the poll interval (default 1s).

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.