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[string]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"`
	// EventType is the delivery's event type.
	EventType string `json:"eventType"`

	Dispatch
}

type Delivery struct {
	// 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 string `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 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 endpoint, for
	// operator recovery.
	Replay(ctx context.Context, deliveryID, endpointID string) error
	// Register validates and stores an endpoint. Validation is not optional and
	// not separable: an unvalidated endpoint is an SSRF target.
	Register(ctx context.Context, endpoint *Endpoint) error
}

type DispatcherOption func(*dispatcher)

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"`
	// 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 []string `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 SQLStoreOption func(*sqlStore)

type Secret struct {
	// Current is the key new signatures are minted under. Required.
	Current []byte `json:"-"`
	// Previous is an outgoing key still emitted alongside Current during a
	// rotation window. Empty outside one.
	Previous []byte `json:"-"`
}

type Store interface {
	// SaveEndpoint creates or replaces an endpoint and its subscriptions.
	SaveEndpoint(ctx context.Context, endpoint *Endpoint) error
	// GetEndpoint reads one endpoint, secrets included. It returns an error
	// wrapping database/sql.ErrNoRows when the endpoint does not exist.
	GetEndpoint(ctx context.Context, endpointID string) (*Endpoint, error)
	// ListEndpoints pages through registered endpoints.
	ListEndpoints(ctx context.Context, filter *filtering.QueryFilter) (*filtering.QueryFilteredResult[Endpoint], error)
	// ArchiveEndpoint retires an endpoint. 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, endpointID string) error

	// EndpointsForEvent returns the enabled, unarchived endpoints subscribed to
	// eventType, using the caller's executor.
	EndpointsForEvent(ctx context.Context, q database.SQLQueryExecutor, eventType string) ([]*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.
	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.
	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 delivery.
	ListAttempts(ctx context.Context, 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.
	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.
	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 URLChecker func(ctx context.Context, rawURL string) error

type VerifyOption func(*verifyConfig)

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)