Documentation
¶
Overview ¶
Package outbound delivers signed webhooks to registered endpoints. Events are typed by their generated JSON contracts, signed with Standard Webhooks (webhook-id, webhook-timestamp, webhook-signature) and sent through typed jobs from the transactional outbox with retries and a delivery log.
Index ¶
- Constants
- func Migrations() []migrate.Definition
- func Module(name foundation.ProviderID, key foundation.Key[*Service], ...) foundation.Module
- func WebhookID(id DeliveryID) string
- type Config
- type Delivery
- type DeliveryID
- type DeliveryInfo
- type DeliveryJob
- func (j DeliveryJob) Declare(service *Service) (jobs.Declaration, error)
- func (j DeliveryJob) Definition() jobs.Definition[DeliveryRequest]
- func (j DeliveryJob) FailureSink(service *Service) (jobs.FailureSink, error)
- func (j DeliveryJob) ToOutbox(service *Service, producer *jobs.Outbox) (Queue, error)
- type DeliveryQuery
- type DeliveryRequest
- type DeliveryState
- type Dependencies
- type Endpoint
- type EndpointID
- type EndpointInfo
- type EndpointOptions
- type Event
- type EventType
- type Queue
- type Service
- func (s *Service) Close(ctx context.Context) error
- func (s *Service) Config() Config
- func (s *Service) CreateEndpoint(ctx context.Context, options EndpointOptions) (EndpointID, secret.String, error)
- func (s *Service) Deliver(ctx context.Context, id DeliveryID) error
- func (s *Service) Deliveries(ctx context.Context, selection DeliveryQuery) ([]DeliveryInfo, error)
- func (s *Service) Done() <-chan struct{}
- func (s *Service) Endpoint(ctx context.Context, id EndpointID) (EndpointInfo, error)
- func (s *Service) Endpoints(ctx context.Context, after EndpointID, limit int) ([]EndpointInfo, error)
- func (*Service) Format(state fmt.State, _ rune)
- func (s *Service) PruneDeliveries(ctx context.Context, olderThan time.Duration, limit int) (int, error)
- func (s *Service) RotateSecret(ctx context.Context, id EndpointID) (secret.String, error)
- func (s *Service) RotateSecretWithGrace(ctx context.Context, id EndpointID, grace time.Duration) (secret.String, error)
- func (s *Service) UpdateEndpoint(ctx context.Context, id EndpointID, options EndpointOptions, active bool) (EndpointInfo, error)
- func (s *Service) Validate() error
Constants ¶
const ( MigrationOrigin migrate.Origin = "foundry.webhooks.outbound" CreateTables migrate.ID = "000001_create_outbound_webhooks" Introduced migrate.Version = "v0.1.0" )
const ( MaxPayloadBytes = 1 << 20 MaxEndpointEvents = 64 MaxSigningSecrets = 4 MaxSecretGrace = 7 * 24 * time.Hour )
const MaxPruneBatch = 10000
MaxPruneBatch bounds one PruneDeliveries call.
Variables ¶
This section is empty.
Functions ¶
func Migrations ¶
func Migrations() []migrate.Definition
Migrations returns the owned endpoint, secret and delivery-log tables. Run them in the same schema as the job outbox that delivery jobs are enqueued to.
func Module ¶
func Module(name foundation.ProviderID, key foundation.Key[*Service], requires []foundation.ProviderID, construct func(foundation.Resolver) (*Service, error)) foundation.Module
Module owns the service and requires the providers owning its borrowed database, keyring and HTTP client. Workers running delivery jobs must stop first; shutdown waits for in-flight operations to exit.
func WebhookID ¶
func WebhookID(id DeliveryID) string
WebhookID is the stable Standard Webhooks message ID of a delivery. It is identical across retries and replays, so receivers deduplicate by it.
Types ¶
type Config ¶
type Config struct {
Schema string
MaxEndpoints int
MaxPayloadBytes int64
MaxActive int
Timeout time.Duration
SecretGrace time.Duration
}
Config bounds one outbound webhook service. MaxEndpoints bounds the registry and one Publish fan-out. MaxPayloadBytes bounds the signed body, including the envelope. SecretGrace is the default overlap during secret rotation, in which the previous secret still signs. Operations queue for capacity for at most min(Timeout, 5s) and then fail as retryable overload.
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns default schema, endpoint, payload, concurrency, timeout and secret-grace settings.
type Delivery ¶
type Delivery struct{}
Delivery identifies the model type for persisted outbound deliveries.
type DeliveryID ¶
DeliveryID identifies a persisted outbound delivery.
func Publish ¶
func Publish[P any](ctx context.Context, tx *database.Tx, queue Queue, event Event[P], payload P) ([]DeliveryID, error)
Publish fans event out to every active endpoint subscribed to its type, in the caller's transaction. The payload is encoded once and each endpoint gets its own delivery ID.
func Send ¶
func Send[P any](ctx context.Context, tx *database.Tx, queue Queue, endpoint EndpointID, event Event[P], payload P) (DeliveryID, error)
Send logs one delivery of event to an active endpoint and enqueues its job in the caller's transaction: both commit or roll back with the business change. It performs no network I/O.
type DeliveryInfo ¶
type DeliveryInfo struct {
ID DeliveryID
Endpoint EndpointID
Event EventType
State DeliveryState
Attempts uint32
LastStatus int
LastFailure string
CreatedAt temporal.DateTime
DeliveredAt value.Nullable[temporal.DateTime]
}
DeliveryInfo is one delivery log entry. It never contains the payload, response body, secrets or transport error text; LastFailure is a stable category and LastStatus the last HTTP status (0 when none was received).
type DeliveryJob ¶
type DeliveryJob struct {
// contains filtered or unexported fields
}
DeliveryJob sends one logged delivery. Retries follow the job policy's attempts and backoff; the final attempt records the delivery as failed.
func DefineDeliveryJob ¶
func DefineDeliveryJob(name jobs.Name, policy jobs.Policy) DeliveryJob
DefineDeliveryJob declares the version-one delivery job with the supplied policy.
func (DeliveryJob) Declare ¶
func (j DeliveryJob) Declare(service *Service) (jobs.Declaration, error)
Declare binds the job handler to service; a zero delivery ID is a permanent job failure.
func (DeliveryJob) Definition ¶
func (j DeliveryJob) Definition() jobs.Definition[DeliveryRequest]
Definition returns the typed job definition, for registry inspection and explicit dispatch by operational tooling.
func (DeliveryJob) FailureSink ¶
func (j DeliveryJob) FailureSink(service *Service) (jobs.FailureSink, error)
FailureSink records deliveries whose job the runtime finalized as failed (attempts, retry deadline or exception limit exhausted, timeout, lost lease) as failed. Register it for every worker that runs this job (jobs.WithFailureSink or jobs.RegisterFailureSink); otherwise such a delivery keeps its last failure category but stays pending. Sinks run at most once per terminal failure.
type DeliveryQuery ¶
type DeliveryQuery struct {
Endpoint value.Optional[EndpointID]
State value.Optional[DeliveryState]
After DeliveryID
Limit int
}
DeliveryQuery selects delivery log entries in ID order.
type DeliveryRequest ¶
type DeliveryRequest struct {
Delivery DeliveryID `json:"delivery"`
}
DeliveryRequest is the typed job payload: only the durable delivery ID.
type DeliveryState ¶
type DeliveryState string
DeliveryState is the delivery log state of one message.
const ( Pending DeliveryState = "pending" Succeeded DeliveryState = "succeeded" Failed DeliveryState = "failed" )
type Dependencies ¶
type Dependencies struct {
DB *database.DB
Keys *encryption.Keyring
Client *httpclient.Client
Clock clock.Clock
}
Dependencies are borrowed and must outlive the service. Keys encrypts endpoint signing secrets at rest. Client must enforce a restricted destination policy, because endpoint URLs are usually customer input.
type EndpointID ¶
EndpointID identifies a persisted outbound endpoint.
type EndpointInfo ¶
type EndpointInfo struct {
ID EndpointID
URL string
Events []EventType
Active bool
CreatedAt temporal.DateTime
}
EndpointInfo describes a registered endpoint. It never contains secrets.
type EndpointOptions ¶
EndpointOptions registers one receiver. URL is checked against the service client's destination policy before it is stored and again for every send. Events are the event types Publish fans out to this endpoint; Send can target any active endpoint directly.
type Event ¶
type Event[P any] struct { // contains filtered or unexported fields }
Event declares one outbound event and the generated JSON contract of its payload. Declare it once and reuse it for every Send and Publish.
func DefineEvent ¶
DefineEvent declares a typed webhook event using the supplied JSON contract.
type EventType ¶
type EventType string
EventType names a delivered event, for example "invoice.paid".
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue binds a service to its delivery job and outbox producer. Construct the service, declare its job, assemble the dispatcher and outbox, then bind.
func (Queue) Replay ¶
func (q Queue) Replay(ctx context.Context, id DeliveryID) error
Replay re-enqueues one delivery with the same webhook-id, so the receiver can deduplicate it. The delivery returns to pending; its attempt count and last result are kept for inspection.
func (Queue) ReplayFailed ¶
func (q Queue) ReplayFailed(ctx context.Context, endpoint value.Optional[EndpointID], limit int) (int, error)
ReplayFailed re-enqueues up to limit failed deliveries of active endpoints, optionally of one endpoint, in one transaction and reports how many were replayed. Deliveries of inactive endpoints are skipped; naming an inactive endpoint explicitly is a conflict. The count covers committed work only.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service owns bounded registry and delivery operations. Construction performs no I/O; apply Migrations with the ordinary migration runner first.
func New ¶
func New(dependencies Dependencies, config Config) (*Service, error)
New validates dependencies and bounds and creates a service using a destination-restricted HTTP client.
func (*Service) Close ¶
Close stops admitting operations and waits for actual exit; Done reports it.
func (*Service) CreateEndpoint ¶
func (s *Service) CreateEndpoint(ctx context.Context, options EndpointOptions) (EndpointID, secret.String, error)
CreateEndpoint registers an active endpoint and returns its first signing secret ("whsec_" + base64). The secret is shown once: give it to the receiver for verification; only its encrypted form is stored.
func (*Service) Deliver ¶
func (s *Service) Deliver(ctx context.Context, id DeliveryID) error
Deliver sends one pending delivery now and records the attempt. A 2xx response succeeds (the response body is drained up to a small bound and never stored). Transport failures, timeouts, 408, 429 and 5xx, and failures before sending (storage, signing secrets), are retryable: the attempt is logged with its category and the delivery stays pending until the job runtime finalizes it (see DeliveryJob.FailureSink). Any other status, a denied destination or an inactive endpoint records the delivery as failed. Succeeded or failed deliveries are not resent; use Replay. The job handler calls it; it is exported for tools and tests.
func (*Service) Deliveries ¶
func (s *Service) Deliveries(ctx context.Context, selection DeliveryQuery) ([]DeliveryInfo, error)
Deliveries inspects the delivery log without payloads.
func (*Service) Done ¶
func (s *Service) Done() <-chan struct{}
Done returns the channel that closes when the service's operation group finishes shutting down.
func (*Service) Endpoint ¶
func (s *Service) Endpoint(ctx context.Context, id EndpointID) (EndpointInfo, error)
Endpoint returns one registered endpoint.
func (*Service) Endpoints ¶
func (s *Service) Endpoints(ctx context.Context, after EndpointID, limit int) ([]EndpointInfo, error)
Endpoints lists registered endpoints in ID order after the given ID.
func (*Service) PruneDeliveries ¶
func (s *Service) PruneDeliveries(ctx context.Context, olderThan time.Duration, limit int) (int, error)
PruneDeliveries deletes up to limit succeeded or failed deliveries (with their stored payloads) last updated more than olderThan ago, and reports how many were deleted. Pending deliveries are never pruned. Run it from an ordinary scheduled task until it returns fewer than limit.
func (*Service) RotateSecret ¶
RotateSecret adds a new current signing secret and returns it once. Previous secrets keep signing, alongside the new one, for Config.SecretGrace, so receivers can switch keys without rejecting deliveries.
func (*Service) RotateSecretWithGrace ¶
func (s *Service) RotateSecretWithGrace(ctx context.Context, id EndpointID, grace time.Duration) (secret.String, error)
RotateSecretWithGrace is RotateSecret with an explicit overlap. A zero grace revokes previous secrets immediately. At most MaxSigningSecrets secrets sign at a time.
func (*Service) UpdateEndpoint ¶
func (s *Service) UpdateEndpoint(ctx context.Context, id EndpointID, options EndpointOptions, active bool) (EndpointInfo, error)
UpdateEndpoint replaces an endpoint's URL, subscriptions and active flag. An inactive endpoint receives no new deliveries; pending ones fail.