outbound

package
v0.3.3 Latest Latest
Warning

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

Go to latest
Published: Nov 1, 2026 License: MIT Imports: 37 Imported by: 0

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

View Source
const (
	MigrationOrigin migrate.Origin  = "foundry.webhooks.outbound"
	CreateTables    migrate.ID      = "000001_create_outbound_webhooks"
	Introduced      migrate.Version = "v0.1.0"
)
View Source
const (
	MaxPayloadBytes   = 1 << 20
	MaxEndpointEvents = 64
	MaxSigningSecrets = 4
	MaxSecretGrace    = 7 * 24 * time.Hour
)
View Source
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.

func (Config) Validate

func (c Config) Validate() error

Validate checks the SQL schema and endpoint, payload, concurrency, timeout and secret-grace bounds.

type Delivery

type Delivery struct{}

Delivery identifies the model type for persisted outbound deliveries.

type DeliveryID

type DeliveryID = model.ID[Delivery]

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.

func (DeliveryJob) ToOutbox

func (j DeliveryJob) ToOutbox(service *Service, producer *jobs.Outbox) (Queue, error)

ToOutbox validates and binds the service and durable job producer into a queue.

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.

func (DeliveryRequest) Format

func (DeliveryRequest) Format(s fmt.State, _ rune)

Format writes a request label without exposing delivery data.

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 Endpoint

type Endpoint struct{}

Endpoint and Delivery are identity markers for typed IDs.

type EndpointID

type EndpointID = model.ID[Endpoint]

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.

func (EndpointInfo) Format

func (EndpointInfo) Format(s fmt.State, _ rune)

Format writes an endpoint label without exposing endpoint data.

type EndpointOptions

type EndpointOptions struct {
	URL    string
	Events []EventType
}

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

func DefineEvent[P any](kind EventType, payload contract.JSON[P]) Event[P]

DefineEvent declares a typed webhook event using the supplied JSON contract.

func (Event[P]) Format

func (Event[P]) Format(s fmt.State, _ rune)

Format writes an event label without exposing its payload contract.

func (Event[P]) Type

func (e Event[P]) Type() EventType

Type returns the declared webhook event type.

func (Event[P]) Validate

func (e Event[P]) Validate() error

Validate checks the event type and JSON contract.

type EventType

type EventType string

EventType names a delivered event, for example "invoice.paid".

func (EventType) Validate

func (t EventType) Validate() error

Validate requires a semantic event identifier.

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) Format

func (Queue) Format(s fmt.State, _ rune)

Format writes a queue label without exposing its dependencies.

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.

func (Queue) Service

func (q Queue) Service() *Service

Service returns the bound service, for example to inspect deliveries.

func (Queue) Validate

func (q Queue) Validate() error

Validate checks the service, job declaration and outbox destination.

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

func (s *Service) Close(ctx context.Context) error

Close stops admitting operations and waits for actual exit; Done reports it.

func (*Service) Config

func (s *Service) Config() Config

Config returns the service's settings, or zero settings for a nil service.

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) Format

func (*Service) Format(state fmt.State, _ rune)

Format writes a service label without exposing its dependencies.

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

func (s *Service) RotateSecret(ctx context.Context, id EndpointID) (secret.String, error)

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.

func (*Service) Validate

func (s *Service) Validate() error

Validate checks that the service has a database, operation group and HTTP client.

Directories

Path Synopsis
Package command supplies outbound webhook delivery inspection and replay.
Package command supplies outbound webhook delivery inspection and replay.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL