webhooks

type Attempt struct {
	// AttemptedAt is when the request was issued.
	AttemptedAt time.Time `json:"attemptedAt"`
	// ID identifies the attempt.
	ID string `json:"id"`
	// DeliveryID is the delivery this attempted.
	DeliveryID string `json:"deliveryID"`
	// EndpointID is the endpoint it was sent to.
	EndpointID string `json:"endpointID"`
	// Error is the transport or status error, rendered. Empty on success. It is
	// a string because it is stored and read by a human, not re-wrapped.
	Error string `json:"error,omitempty"`
	// Duration is how long the request took.
	Duration time.Duration `json:"duration"`
	// StatusCode is the response status, or 0 if no response was received.
	StatusCode int `json:"statusCode"`
	// AttemptCount is which attempt this was, 1-indexed.
	AttemptCount int `json:"attemptCount"`
}

type Catalog map[EventType]EventDefinition

type CircuitBreakerFactory func(endpointID string) (circuitbreaking.CircuitBreaker, error)

type ClaimedDispatch struct {
	// Endpoint is the subscriber, resolved at claim time rather than at
	// dispatch time — so a secret rotated between the event and its delivery
	// signs with the current key, and an endpoint disabled in between is not
	// delivered to at all.
	Endpoint *Endpoint `json:"endpoint"`
	// Payload is the delivery body, verbatim as dispatched.
	Payload []byte `json:"payload"`
	// Scope is the delivery's scope, read back from the delivery row.
	//
	// The worker does not filter on it — it delivers whatever it claimed, to the
	// endpoint the dispatch names, and the fan-out that produced the row already
	// resolved subscribers within this scope. It is here so that a delivery
	// failure, a dead dispatch, and a slow subscriber are attributable to a
	// tenant in the worker's logs and spans, which is otherwise the one place in
	// the pipeline where whose event it was has been forgotten.
	Scope tenancy.Scope `json:"scope"`
	// EventType is the delivery's event type.
	EventType EventType `json:"eventType"`

	Dispatch
}

type Delivery struct {
	// Scope is whose event this is, and it bounds the fan-out: Dispatch resolves
	// subscribers within it, so an endpoint registered by one account never
	// receives another account's copy of the same event type.
	//
	// Required. An unset scope is refused rather than read as "every
	// subscriber", because the reading that makes a missing filter convenient is
	// the one that leaks a tenant's payload to every other tenant. An
	// application whose events are global says tenancy.Global().
	Scope tenancy.Scope `json:"scope"`
	// ID identifies the delivery, and is what Replay names. Generated when empty.
	ID string `json:"id"`
	// EventType is the catalog event type. Must be in the Catalog.
	EventType EventType `json:"eventType"`
	// OrderingKey groups deliveries that must arrive in order — typically the
	// subject resource's ID. Deliveries sharing a key reach a given endpoint in
	// the order they were dispatched; deliveries with different keys, or with
	// none, are unordered relative to each other.
	//
	// Ordering is per endpoint as well as per key. A subscriber that is timing
	// out delays only its own queue for that key, never another subscriber's.
	OrderingKey string `json:"orderingKey,omitempty"`
	// Payload is the event body, delivered to every subscriber byte for byte as
	// supplied. It is json.RawMessage rather than any so that what is signed and
	// what is sent are the same bytes — re-marshaling between signing and
	// sending is exactly how a signature comes to cover something other than the
	// request body.
	Payload json.RawMessage `json:"payload"`
}

type Dispatch struct {
	// NextAttempt is when this dispatch next becomes claimable.
	NextAttempt time.Time `json:"nextAttempt"`
	// ID identifies the dispatch.
	ID string `json:"id"`
	// DeliveryID is the delivery being sent.
	DeliveryID string `json:"deliveryID"`
	// EndpointID is the subscriber it is being sent to.
	EndpointID string `json:"endpointID"`
	// OrderingKey is denormalized from the delivery so the claim predicate can
	// enforce ordering without joining. See buildSelectClaimable.
	OrderingKey string `json:"orderingKey,omitempty"`
	// LastError is the most recent failure, rendered.
	LastError string `json:"lastError,omitempty"`
	// Attempts is how many attempts have been made.
	Attempts int `json:"attempts"`
	// Dead marks a dispatch that exhausted its attempts. It is skipped by every
	// future claim and is what an operator replays.
	Dead bool `json:"dead"`
}

type Dispatcher interface {
	// Dispatch fans an event out to every endpoint in the delivery's scope that is
	// subscribed to it, writing through the caller's executor so the deliveries
	// commit with the state change that caused them.
	Dispatch(ctx context.Context, q database.SQLQueryExecutor, delivery *Delivery) error
	// Replay re-drives a specific past delivery to a specific one of the scope's
	// endpoints, for operator recovery.
	Replay(ctx context.Context, scope tenancy.Scope, deliveryID, endpointID string) error
	// Register validates and stores an endpoint, under the scope the endpoint
	// carries. Validation is not optional and not separable: an unvalidated
	// endpoint is an SSRF target.
	Register(ctx context.Context, endpoint *Endpoint) error
}

type DispatcherOption func(*StoreDispatcher)

type Endpoint struct {
	// Headers are static headers added to every request to this endpoint, for
	// subscribers that need a routing token or a tenant hint. The signature,
	// timestamp, content type, and event headers this package sets are not
	// overridable from here — see reservedHeaders.
	Headers map[string]string `json:"headers,omitempty"`
	// Scope is whose endpoint this is. Required, and rejected at registration
	// when unset: an endpoint that belongs to nobody in particular is one an
	// application with tenants registered by accident, and it would receive
	// deliveries the account it was meant for never sees.
	//
	// An application whose events are global says tenancy.Global() here, which
	// is a scope like any other and matches only deliveries dispatched in it.
	Scope tenancy.Scope `json:"scope"`
	// ID identifies the endpoint. Generated at registration when empty.
	ID string `json:"id"`
	// URL is the absolute https:// URL deliveries are POSTed to.
	URL string `json:"url"`
	// ContentType is the request's Content-Type. Defaults to application/json.
	ContentType string `json:"contentType"`
	// Secret carries the signing keys. Never serialized: an endpoint travels
	// through API responses and logs, and its secret must not.
	Secret Secret `json:"-"`
	// Events are the catalog event types this endpoint subscribes to.
	Events []EventType `json:"events"`
	// Disabled stops delivery without deleting the endpoint or its history,
	// which is what an operator wants when a subscriber is misbehaving.
	Disabled bool `json:"disabled"`
}

type EventDefinition struct {
	// Description is human-facing prose explaining when the event fires.
	Description string `json:"description"`
}

type EventType string

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

type SQLStoreOption func(*SQLStore)

type Secret = requestsigning.Keyring

type Store interface {
	// SaveEndpoint creates or replaces an endpoint and its subscriptions, under
	// the scope the endpoint carries.
	SaveEndpoint(ctx context.Context, endpoint *Endpoint) error
	// GetEndpoint reads one of scope's endpoints, secrets included. It returns an
	// error wrapping database/sql.ErrNoRows when the endpoint does not exist —
	// including when it exists in another scope, which is the same answer as far
	// as this scope is concerned.
	GetEndpoint(ctx context.Context, scope tenancy.Scope, endpointID string) (*Endpoint, error)
	// ListEndpoints pages through the endpoints registered in scope.
	ListEndpoints(ctx context.Context, scope tenancy.Scope, filter *filtering.QueryFilter) (*filtering.QueryFilteredResult[Endpoint], error)
	// ArchiveEndpoint retires one of scope's endpoints. Its delivery history is
	// retained: the attempts log outlives the endpoint, because "what did we send
	// them" is asked most often after someone has been removed.
	ArchiveEndpoint(ctx context.Context, scope tenancy.Scope, endpointID string) error

	// EndpointsForEvent returns the enabled, unarchived endpoints in scope that
	// are subscribed to eventType, using the caller's executor.
	//
	// The scope is a parameter and not an option: this is the query whose missing
	// filter delivers one account's event to every other account's subscribers.
	EndpointsForEvent(ctx context.Context, q database.SQLQueryExecutor, scope tenancy.Scope, eventType EventType) ([]*Endpoint, error)
	// Enqueue writes a delivery and one dispatch per endpoint, using the
	// caller's executor, so both commit with whatever else that transaction did.
	// The delivery's scope is stored with it.
	Enqueue(ctx context.Context, q database.SQLQueryExecutor, delivery *Delivery, endpointIDs []string, now time.Time) error

	// Claim leases the next batch of due dispatches, incrementing their attempt
	// counts, and returns them ready to send.
	//
	// It spans every scope, deliberately: a worker delivers the whole
	// deployment's backlog, and a per-scope claim would need a list of scopes
	// nothing maintains. What it returns is scoped — each ClaimedDispatch says
	// which scope it came from.
	Claim(ctx context.Context, now time.Time, limit int, leaseUntil time.Time) ([]ClaimedDispatch, error)
	// MarkDelivered retires a dispatch that was accepted.
	MarkDelivered(ctx context.Context, dispatchID string, at time.Time) error
	// RecordFailure releases the lease, schedules the retry, and sets dead once
	// the dispatch has exhausted its attempts.
	//
	// attempts is persisted as given rather than left as Claim incremented it,
	// so the caller can decline to charge an attempt for a failure the
	// subscriber never saw — an open circuit being the case that matters.
	RecordFailure(ctx context.Context, dispatchID string, attempts int, nextAttempt time.Time, lastErr string, dead bool) error
	// RecordAttempt appends to the delivery log.
	RecordAttempt(ctx context.Context, attempt *Attempt) error
	// ListAttempts pages through the attempts recorded for one of scope's
	// deliveries. A delivery in another scope reads as one with no attempts,
	// which is what it is from here.
	ListAttempts(ctx context.Context, scope tenancy.Scope, deliveryID string, filter *filtering.QueryFilter) (*filtering.QueryFilteredResult[Attempt], error)

	// Requeue makes a delivery/endpoint pair claimable again, clearing its dead
	// flag and attempt count. It returns an error wrapping ErrDeliveryNotFound
	// if the pair was never dispatched or has been reaped.
	//
	// It takes no scope because the pair is already one: a dispatch exists only
	// where a delivery fanned out to an endpoint, and Dispatch resolves those
	// within one scope. StoreDispatcher.Replay is the scoped entry point, and it
	// establishes the scope by reading the endpoint in it first.
	Requeue(ctx context.Context, deliveryID, endpointID string, at time.Time) error

	// Backlog reports how many dispatches are waiting and when the oldest was
	// created, for the worker's health gauges. Like Claim, it spans every scope.
	Backlog(ctx context.Context) (depth int64, oldest time.Time, err error)
	// Reap deletes delivered dispatches, their deliveries, and their attempts
	// once they age past the retention window, up to limit rows.
	Reap(ctx context.Context, before time.Time, limit int) (int64, error)
}

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

type URLChecker func(ctx context.Context, rawURL string) error

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

type WorkerConfig struct {
	// UserAgent identifies deliveries to subscribers.
	UserAgent string `env:"USER_AGENT" json:"userAgent,omitempty" yaml:"userAgent,omitempty"`
	// Backoff drives the retry schedule for failed deliveries. MaxAttempts is
	// the threshold past which a dispatch is marked dead.
	Backoff retrycfg.Config `envPrefix:"BACKOFF_" json:"backoff,omitzero" yaml:"backoff,omitempty"`
	// PollInterval is how often the worker looks for work.
	PollInterval time.Duration `env:"POLL_INTERVAL" json:"pollInterval,omitempty" yaml:"pollInterval,omitempty"`
	// LeaseDuration is how long a claim is held before it can be reclaimed.
	LeaseDuration time.Duration `env:"LEASE_DURATION" json:"leaseDuration,omitempty" yaml:"leaseDuration,omitempty"`
	// RequestTimeout bounds one delivery request.
	RequestTimeout time.Duration `env:"REQUEST_TIMEOUT" json:"requestTimeout,omitempty" yaml:"requestTimeout,omitempty"`
	// CircuitOpenRetryDelay is how long a short-circuited dispatch waits before
	// becoming claimable again. It is a flat delay rather than a backoff step,
	// because backing off exponentially against an open circuit means the first
	// delivery after recovery can wait far longer than the outage did.
	CircuitOpenRetryDelay time.Duration `env:"CIRCUIT_OPEN_RETRY_DELAY" json:"circuitOpenRetryDelay,omitempty" yaml:"circuitOpenRetryDelay,omitempty"`
	// Retention is how long delivered dispatches are kept before reaping.
	Retention time.Duration `env:"RETENTION" json:"retention,omitempty" yaml:"retention,omitempty"`
	// ReapInterval is how often the reaper runs.
	ReapInterval time.Duration `env:"REAP_INTERVAL" json:"reapInterval,omitempty" yaml:"reapInterval,omitempty"`
	// BatchSize is how many dispatches one cycle claims.
	BatchSize int `env:"BATCH_SIZE" json:"batchSize,omitempty" yaml:"batchSize,omitempty"`
	// Concurrency is how many deliveries run at once within a batch.
	Concurrency int `env:"CONCURRENCY" json:"concurrency,omitempty" yaml:"concurrency,omitempty"`
	// ReapBatchSize caps how many rows one reap deletes.
	ReapBatchSize int `env:"REAP_BATCH_SIZE" json:"reapBatchSize,omitempty" yaml:"reapBatchSize,omitempty"`
}

type WorkerOption func(*Worker)