webhooks

type Attempt struct {
	// CreatedAt is when the request was issued, which is the instant the log
	// line was written.
	CreatedAt time.Time `json:"createdAt"`
	// 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 the claim in
	// webhooks/internal/queries, which is where the guarantee is written down.
	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 transaction so the deliveries
	// commit with the state change that caused them.
	Dispatch(ctx context.Context, q database.Tx, 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
	// Subscribe adds one event type to one of the scope's endpoints, gating on the
	// catalog, and returns the subscription.
	Subscribe(ctx context.Context, scope tenancy.Scope, endpointID string, eventType EventType) (*Subscription, error)
	// Unsubscribe retires one of the scope's subscriptions by ID, leaving the
	// endpoint and its other subscriptions alone.
	Unsubscribe(ctx context.Context, scope tenancy.Scope, subscriptionID string) error
}

type DispatcherOption func(*StoreDispatcher)

type Endpoint struct {
	// CreatedAt is when the endpoint was registered, stamped by the Store.
	CreatedAt time.Time `json:"createdAt"`
	// LastUpdatedAt is when the endpoint was last re-registered.
	LastUpdatedAt *time.Time `json:"lastUpdatedAt"`
	// ArchivedAt is when the endpoint was retired by ArchiveEndpoint. The row is
	// kept rather than deleted: the attempts log outlives the endpoint, because
	// "what did we send them" is asked most often after someone has been removed.
	//
	// Re-registering an archived endpoint clears it — SaveEndpoint is an upsert,
	// and a caller who registers an ID that exists means to have it live.
	ArchivedAt *time.Time `json:"archivedAt"`
	// 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"`
	// CreatedBy names the principal that registered the endpoint — the user who
	// clicked the button, where an application tracks that.
	//
	// It is a tenancy.Scope rather than a bare user ID on purpose. "Whose is
	// this" already has a type in this module, and the answer here is finer
	// grained than Scope rather than a different kind of answer: an application
	// that lists "the endpoints I created" filters on the same value type it
	// filters tenants by, and a second string field would be a parallel owner
	// dimension that nothing else in the module understands.
	//
	// Unlike Scope it is optional. An endpoint belongs to its Scope, which is
	// what bounds fan-out and every read; who registered it is provenance, and an
	// application with no per-user attribution leaves this unset. It is written
	// once, at registration, and is not part of what a re-registration updates —
	// an endpoint does not change hands, and neither does its provenance.
	CreatedBy tenancy.Scope `json:"createdBy"`
	// ID identifies the endpoint. Generated at registration when empty.
	ID string `json:"id"`
	// Name is the label an endpoint-management UI shows. Optional, and free-form:
	// this package never reads it, and two endpoints in one scope may share one.
	Name string `json:"name"`
	// 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:"-"`
	// Subscriptions are the endpoint's live subscriptions — one identified,
	// individually archivable row per catalog event type it wants.
	//
	// On the way in it is the set to register: SubscribeTo builds one from event
	// types, and SaveEndpoint reconciles the stored rows against it, so an entry
	// that is not there any more is archived rather than deleted. On the way out
	// the Store fills it with the live rows, IDs and timestamps included.
	//
	// It is rows rather than a []EventType because a subscription is something an
	// application's own API lets a user retire one of, and a flat list can only be
	// rewritten: "unsubscribe from order.created" becomes "here is the list
	// again, minus one", which loses when it happened, and there is no identifier
	// to name in the request that asked for it. EventTypes derives the flat form
	// where one is wanted, so nothing has to keep two copies agreeing.
	Subscriptions []Subscription `json:"subscriptions"`
	// 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, under the scope the endpoint
	// carries, and reconciles its subscriptions against the set it names.
	//
	// Reconciles rather than replaces: a subscription the endpoint already has is
	// kept, with its identity and its creation time; one it names for the first
	// time is created; one it no longer names is archived rather than deleted, so
	// that a subscription an endpoint has ended is still something the delivery
	// log can be read against. It fills the endpoint's Subscriptions with the rows
	// that are live afterwards, IDs included.
	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.
	//
	// Its subscriptions are left as they are. An archived endpoint is excluded
	// from fan-out by its own archived_at, so archiving them too would buy
	// nothing and would lose which event types it was subscribed to if it is ever
	// re-registered.
	ArchiveEndpoint(ctx context.Context, scope tenancy.Scope, endpointID string) error

	// AddSubscription subscribes one of scope's endpoints to eventType and
	// returns the resulting row.
	//
	// It is idempotent on the (endpoint, event type) pair: subscribing to
	// something the endpoint already subscribes to returns the existing row, and
	// re-subscribing to something it archived revives that row rather than
	// minting a second one for the same pair. It returns an error wrapping
	// database/sql.ErrNoRows when the endpoint does not exist in scope.
	//
	// The catalog is not checked here — a Store has none. StoreDispatcher.Subscribe
	// is the entry point that gates on it, and it is the one an application should
	// call, for the reason Register exists rather than SaveEndpoint being public
	// API: an accepted subscription to an event type nothing publishes is an
	// endpoint that never fires and no signal explaining why.
	AddSubscription(ctx context.Context, scope tenancy.Scope, endpointID string, eventType EventType) (*Subscription, error)
	// GetSubscription reads one of scope's subscriptions, archived ones included —
	// "when did they stop receiving this" is a question about an archived row. It
	// returns an error wrapping database/sql.ErrNoRows when the subscription does
	// not exist under one of scope's endpoints.
	GetSubscription(ctx context.Context, scope tenancy.Scope, subscriptionID string) (*Subscription, error)
	// ListSubscriptions pages the live subscriptions of one of scope's endpoints.
	ListSubscriptions(ctx context.Context, scope tenancy.Scope, endpointID string, filter *filtering.QueryFilter) (*filtering.QueryFilteredResult[Subscription], error)
	// ArchiveSubscription retires one of scope's subscriptions, so the endpoint
	// stops receiving that event type without its other subscriptions, its
	// delivery history, or its identity being touched.
	//
	// This is the method a flat event list cannot offer. Against one, "stop
	// sending me order.created" can only be expressed as a rewrite of the whole
	// set, which loses when it happened and races any concurrent edit of the same
	// endpoint.
	ArchiveSubscription(ctx context.Context, scope tenancy.Scope, subscriptionID 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, in the caller's
	// transaction, so both commit with whatever else that transaction did.
	// The delivery's scope is stored with it.
	Enqueue(ctx context.Context, q database.Tx, 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 Subscription struct {
	// CreatedAt is when the subscription was made, stamped by the Store.
	CreatedAt time.Time `json:"createdAt"`
	// LastUpdatedAt is when a write last named this subscription: the save or the
	// AddSubscription that would have created it had it not already existed.
	// Nil until one has, so a subscription made once and left alone has none.
	LastUpdatedAt *time.Time `json:"lastUpdatedAt"`
	// ArchivedAt is when the subscription was retired. Archived subscriptions
	// are excluded from fan-out and from the endpoint's Subscriptions, and are
	// kept so that "when did they stop receiving this" has an answer.
	ArchivedAt *time.Time `json:"archivedAt"`
	// ID identifies the subscription, and is what ArchiveSubscription names.
	// Generated by the Store when empty.
	ID string `json:"id"`
	// EndpointID is the endpoint that subscribed.
	EndpointID string `json:"endpointID"`
	// EventType is the catalog event type subscribed to.
	EventType EventType `json:"eventType"`
}

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)