outbox

package
v1.27.0 Latest Latest
Warning

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

 Go to latest Published: Sep 11, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Overview

Package outbox provides a transactional outbox pattern implementation with support for external message delivery through configurable bridges.

The bridge system allows outbox messages to be delivered to external systems like Kafka, webhooks, RabbitMQ, etc. based on topic-based routing rules.

Package outbox provides a small SQL-backed transactional outbox runtime.

Index

Constants

View Source 
const (
	// PayloadEncodingBase64 declares the classic wire shape: the "payload"
	// field is a JSON string holding the base64 encoding of the raw payload
	// bytes (Go's default encoding/json representation of []byte).
	PayloadEncodingBase64 = "base64"

	// PayloadEncodingJSON declares the embedded shape: the "payload" field
	// is the payload's JSON document itself, embedded verbatim.
	PayloadEncodingJSON = "json"
)

Values of WebhookPayloadEncodingHeader and of WebhookConfig.PayloadEncoding.

Under either declared encoding, a message with no payload puts JSON null in the "payload" field.

View Source 
const DefaultTableName = "nucleus_outbox"
View Source 
const GracefulStopTimeout = 5 * time.Second

GracefulStopTimeout bounds how long Stop waits for the dispatch pass in flight to finish before it cancels the run context. It applies when the caller's Stop context carries no deadline of its own — a graceful stop must not become an unbounded one, because a pass can be waiting on a bridge that is not answering.

View Source 
const WebhookPayloadEncodingHeader = "X-Outbox-Payload-Encoding"

WebhookPayloadEncodingHeader declares, on every webhook delivery, how the "payload" field of that message's body is encoded. Its value is one of PayloadEncodingBase64 or PayloadEncodingJSON. The header is always present, so a consumer never has to guess the payload shape.

The header is INFORMATIONAL and deliberately NOT part of the signed material (SEC-3). The bridge signs the body alone — byte-for-byte the module-webhook scheme (nucleus.SignWebhookBody), which is exactly why one verifier serves both surfaces; binding this header into the signature would fork that scheme and break every consumer that verifies outbox deliveries with the module-webhook verifier. Because the header is unsigned, a consumer must never let it drive a decoding or security decision: decode by the encoding you were configured to expect, and use CheckPayloadEncoding to reject — as defense in depth — any delivery whose declared encoding disagrees with that expectation. Flipping the header in transit does not forge anything (it leaves the body signature valid and, at worst, makes a consumer that trusts it misparse a legitimate body into a 400), which is why closing this hole is a hardening, not a fix for an exploitable bug.

View Source 
const WebhookSignatureHeader = "X-Nucleus-Signature"

WebhookSignatureHeader carries the HMAC-SHA256 signature of the webhook body when the bridge is configured with a Secret. It is the SAME header, with the SAME "sha256=<hex>" value format, that module webhooks (pkg/nucleus, WebhookSpec.Secret) verify on inbound requests — one signing scheme across the framework, so a consumer can verify outbox deliveries with the same code (and the same helper, nucleus.SignWebhookBody) it already uses for module webhooks.

Variables

View Source 
var (
	ErrLeaseOwnerRequired = fmt.Errorf("outbox: lease owner is required")
	ErrNoRouteMatched     = fmt.Errorf("outbox: no bridge route matched message topic")
)
View Source 
var (
	ErrNilDB          = fmt.Errorf("outbox: nil db")
	ErrNilTx          = fmt.Errorf("outbox: nil tx")
	ErrEmptyTopic     = fmt.Errorf("outbox: topic is required")
	ErrNilStore       = fmt.Errorf("outbox: store is nil")
	ErrHandlerMissing = fmt.Errorf("outbox: handler is required")
)
View Source 
var ErrPayloadEncodingMismatch = errors.New("outbox: payload encoding mismatch")

ErrPayloadEncodingMismatch reports that a webhook delivery declared, in its WebhookPayloadEncodingHeader, a payload encoding different from the one the consumer was configured to expect. Returned (wrapped) by CheckPayloadEncoding.

Functions

func CheckPayloadEncoding 

func CheckPayloadEncoding(expected, delivered string) error

CheckPayloadEncoding is the consumer-side defense-in-depth check for the WebhookPayloadEncodingHeader (SEC-3). Since the header is not part of the signed material, a consumer decodes by the encoding it was configured to expect and MAY call CheckPayloadEncoding to reject a delivery whose declared encoding disagrees with that expectation: 

enc := r.Header.Get(outbox.WebhookPayloadEncodingHeader)
if err := outbox.CheckPayloadEncoding(cfgEncoding, enc); err != nil {
    http.Error(w, "bad request", http.StatusBadRequest)
    return
}

Both arguments are normalized the way NewWebhookBridge normalizes PayloadEncoding: surrounding whitespace trimmed, case folded, and an empty value taken as PayloadEncodingBase64 (the default wire shape). A match returns nil; a mismatch returns an error wrapping ErrPayloadEncodingMismatch and naming both encodings.

One legitimate divergence exists: a bridge in PayloadEncodingJSON mode downgrades a payload that is not valid JSON to the base64 shape for that delivery (see Send), declaring "base64" accordingly. That fallback only affects hand-built messages — a producer that enqueues through the store always emits valid JSON — so a store-backed consumer never sees the divergence and can reject the mismatch; a consumer that also accepts hand-built payloads should tolerate a base64 delivery under a json expectation.

Types

type Bridge 

type Bridge interface {
	// Name returns the unique identifier for this bridge.
	// This name is used for registration and routing configuration.
	Name() string

	// Send delivers an outbox message to the external system.
	// The context can be used for cancellation and timeout control.
	// Returns an error if delivery fails, which will trigger retry logic.
	Send(ctx context.Context, msg Message) error

	// Healthy checks if the bridge is operational.
	// This is called during health checks and can be used to verify
	// connectivity to the external system.
	Healthy(ctx context.Context) error

	// Close gracefully shuts down the bridge.
	// Called during application shutdown to release resources.
	Close() error
}

Bridge defines an external message delivery destination (Kafka, Webhook, RabbitMQ, etc.).

Implementations of this interface can be registered with a BridgeRegistry and used by the Dispatcher to deliver outbox messages to external systems. The router determines which bridges receive a message based on topic patterns.

type BridgeRegistry 

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

BridgeRegistry manages a collection of registered bridges.

The registry provides thread-safe operations for registering, retrieving, and closing bridges. It is used by the ManagedOutbox to coordinate multiple external delivery destinations.

func NewBridgeRegistry 

func NewBridgeRegistry() *BridgeRegistry

NewBridgeRegistry creates an empty bridge registry.

func (*BridgeRegistry) Close 

func (r *BridgeRegistry) Close() error

Close shuts down all registered bridges.

This method calls Close() on each registered bridge and collects any errors. If any bridge fails to close, a combined error is returned. This method is thread-safe.

func (*BridgeRegistry) Get 

func (r *BridgeRegistry) Get(name string) (Bridge, bool)

Get retrieves a bridge by name.

Returns the bridge and true if found, nil and false otherwise. This method is thread-safe.

func (*BridgeRegistry) List 

func (r *BridgeRegistry) List() []Bridge

List returns all registered bridges.

The returned slice is a copy and safe to modify. This method is thread-safe.

func (*BridgeRegistry) Register 

func (r *BridgeRegistry) Register(bridge Bridge) error

Register adds a bridge to the registry.

The bridge name must be unique and non-empty. This method is thread-safe and will return an error if a bridge with the same name is already registered.

type Config 

type Config struct {
	DatabaseURL string
	TableName   string
	Flavor      Flavor
}

type DispatchResult 

type DispatchResult struct {
	Attempted int `json:"attempted"`
	Delivered int `json:"delivered"`
	Retried   int `json:"retried"`
	Failed    int `json:"failed"`
}

DispatchResult summarizes one dispatcher pass.

Attempted is the total number of messages processed in this pass. Delivered is the number of messages successfully delivered. Retried is the number of messages that failed and will be retried. Failed is the number of messages that exceeded MaxAttempts and were marked as failed.

type Dispatcher 

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

Dispatcher polls the outbox table, leases pending messages, and delivers them through a handler.

The dispatcher uses a leasing mechanism to ensure that multiple instances can run concurrently without duplicate processing. Messages are claimed with a lease duration, and if delivery fails, they are retried with exponential backoff.

When Registry and Router are configured, the dispatcher uses bridge-based routing instead of the traditional HandlerFunc.

func NewDispatcher 

func NewDispatcher(store *Store, handler HandlerFunc, cfg DispatcherConfig) (*Dispatcher, error)

NewDispatcher creates a new dispatcher with the given store, handler, and configuration.

The store must be non-nil and the handler must be non-nil unless bridge-based routing is configured. The configuration is normalized to fill in any missing values with defaults.

Returns an error if the store or handler is nil, or if the lease owner is empty.

func (*Dispatcher) Run 

func (d *Dispatcher) Run(ctx context.Context) error

Run starts the dispatcher and blocks until the context is canceled.

This method performs an initial dispatch pass, then polls at the configured interval until the context is canceled. It is designed to be run in a goroutine.

Example: 

go dispatcher.Run(ctx)

Returns an error if the initial dispatch pass fails.

func (*Dispatcher) RunGraceful 

func (d *Dispatcher) RunGraceful(ctx context.Context, stopAfterPass <-chan struct{}) error

RunGraceful is Run with a soft-stop signal: when stopAfterPass is closed, the dispatcher finishes the pass it is in and returns, instead of having its SQL cancelled mid-statement.

The distinction matters at shutdown. Cancelling the run context is abrupt by design: it aborts the in-flight statement, so a pass can end having claimed messages it never attempted to deliver (they wait for the lease to expire), and the driver tears the result set down under cancellation. The soft signal makes the common shutdown path finish its work first; the caller keeps the context for the case where waiting is no longer an option (see ManagedOutbox.Stop, which escalates on its deadline).

A nil stopAfterPass makes this exactly Run: only the context stops it.

func (*Dispatcher) RunOnce 

func (d *Dispatcher) RunOnce(ctx context.Context) (DispatchResult, error)

RunOnce performs a single dispatch pass.

This method claims available messages, delivers them through the handler or bridges, and updates their status based on the result. It returns a summary of the pass.

If bridge-based routing is configured (Registry and Router are non-nil), messages are delivered to matching bridges. Otherwise, the traditional HandlerFunc is used.

Messages that fail delivery are retried with exponential backoff until MaxAttempts is reached, at which point they are marked as failed.

type DispatcherConfig 

type DispatcherConfig struct {
	LeaseOwner         string
	LeaseDuration      time.Duration
	PollInterval       time.Duration
	BatchSize          int
	MaxAttempts        int
	BaseDelay          time.Duration
	MaxDelay           time.Duration
	Registry           *BridgeRegistry
	Router             *Router
	MissingRoutePolicy MissingRoutePolicy
}

DispatcherConfig configures delivery attempts and polling behaviour.

LeaseOwner is a unique identifier for this dispatcher instance, used for distributed locking when multiple instances are running. LeaseDuration is how long a message lease is held before it can be claimed by another instance. PollInterval is how often the dispatcher polls for new messages. BatchSize is the maximum number of messages to process in one poll cycle. MaxAttempts is the maximum number of delivery attempts before marking as failed. BaseDelay is the initial retry delay for exponential backoff. MaxDelay is the maximum retry delay. Registry is the bridge registry for external message delivery (optional). Router is the topic router for determining which bridges receive messages (optional). MissingRoutePolicy controls whether an unrouted bridge message is an error or intentionally ignored.

If Registry and Router are configured, the dispatcher will use bridge-based routing. Otherwise, it will use the traditional HandlerFunc for message delivery.

func DefaultDispatcherConfig 

func DefaultDispatcherConfig() DispatcherConfig

DefaultDispatcherConfig returns sensible defaults for dispatcher configuration.

These defaults are suitable for development and can be overridden for production.

type Entry 

type Entry struct {
	ID          string
	Topic       string
	Payload     any
	AvailableAt time.Time
}

type Flavor 

type Flavor string
const (
	FlavorSQLite   Flavor = "sqlite"
	FlavorPostgres Flavor = "postgres"
	FlavorMySQL    Flavor = "mysql"
)

type HandlerFunc 

type HandlerFunc func(context.Context, Message) error

HandlerFunc delivers one claimed outbox message.

This function type is used for traditional message delivery when bridge-based routing is not configured. The function should handle the message (e.g., send to an external system) and return an error if delivery fails. Errors trigger retry logic in the dispatcher.

type KafkaBridge 

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

KafkaBridge is reserved for a future Kafka implementation. It is intentionally disabled until the package wires a maintained Kafka client and real delivery/health semantics. Applications must not configure Kafka as a production bridge in this release.

func NewKafkaBridge 

func NewKafkaBridge(cfg KafkaConfig) (*KafkaBridge, error)

NewKafkaBridge validates Kafka bridge configuration and then returns a clear error because Kafka delivery is not implemented in this release.

func (*KafkaBridge) Close 

func (b *KafkaBridge) Close() error

Close is a no-op because no Kafka resources are acquired.

func (*KafkaBridge) Healthy 

func (b *KafkaBridge) Healthy(ctx context.Context) error

Healthy returns an explicit unsupported error.

func (*KafkaBridge) Name 

func (b *KafkaBridge) Name() string

Name returns the bridge name.

func (*KafkaBridge) Send 

func (b *KafkaBridge) Send(ctx context.Context, msg Message) error

Send returns an explicit unsupported error.

type KafkaConfig 

type KafkaConfig struct {
	Name    string
	Brokers []string
	Topic   string
}

KafkaConfig configures a Kafka bridge.

Brokers is a list of Kafka broker addresses in the format "host:port". Topic is the Kafka topic to which messages will be published.

type ManagedConfig 

type ManagedConfig struct {
	DB                 *sql.DB
	TableName          string
	Flavor             Flavor
	LeaseOwner         string
	LeaseDuration      time.Duration
	PollInterval       time.Duration
	BatchSize          int
	MaxAttempts        int
	BaseDelay          time.Duration
	MaxDelay           time.Duration
	Logger             *slog.Logger
	MissingRoutePolicy MissingRoutePolicy
}

ManagedConfig configures a managed outbox instance.

DB is the SQL database connection used for the outbox table. TableName is the name of the outbox table (defaults to "nucleus_outbox"). Flavor is the database flavor (SQLite, Postgres, MySQL) for SQL dialect differences. LeaseOwner is a unique identifier for this instance (used for distributed locking). LeaseDuration is how long a message lease is held before it can be claimed by another instance. PollInterval is how often the dispatcher polls for new messages. BatchSize is the maximum number of messages to process in one poll cycle. MaxAttempts is the maximum number of delivery attempts before marking as failed. BaseDelay is the initial retry delay for exponential backoff. MaxDelay is the maximum retry delay. Logger is the structured logger for operational events.

MissingRoutePolicy controls what happens when a leased message's topic has no registered bridge (QCD-FW-5). The default, MissingRouteError, fails the message — correct for a homogeneous fleet where every instance registers every bridge. In a deliberately heterogeneous fleet (only some processes register bridges for some topics), set MissingRouteIgnore so an instance leaves unrouted messages for the instance that can deliver them, instead of leasing and failing them (stealing attempts from the real deliverer).

type ManagedOutbox 

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

ManagedOutbox wraps the outbox components with lifecycle management.

This type provides a high-level interface for managing the outbox pattern within an application. It combines the store, dispatcher, bridge registry, and topic router into a single managed component that can be started and stopped.

The managed outbox is typically created by the app.App when outbox is enabled in the configuration, but can also be created manually for custom use cases.

Example usage: 

managed, err := outbox.NewManagedOutbox(outbox.ManagedConfig{
    DB:        sqlDB,
    TableName: "nucleus_outbox",
    Flavor:    outbox.FlavorSQLite,
    Logger:    logger,
})
if err != nil {
    log.Fatal(err)
}

// Register bridges
managed.RegisterBridge(webhookBridge)
managed.AddRoute("notifications.*", "webhook")

// Start the dispatcher
if err := managed.Start(context.Background()); err != nil {
    log.Fatal(err)
}

// Enqueue messages
managed.Enqueue(ctx, outbox.Entry{
    Topic:   "notifications.email",
    Payload: map[string]any{"to": "user@example.com"},
})

func NewManagedOutbox 

func NewManagedOutbox(cfg ManagedConfig) (*ManagedOutbox, error)

NewManagedOutbox creates a new managed outbox instance.

This method initializes the store, bridge registry, topic router, and dispatcher. The dispatcher is configured with a fallback handler that returns an error if no bridges are configured for a message topic.

Returns an error if the database connection is nil or if store/dispatcher creation fails.

func (*ManagedOutbox) AddRoute 

func (m *ManagedOutbox) AddRoute(pattern string, bridgeNames ...string)

AddRoute adds a topic routing rule.

Messages matching the pattern will be sent to all specified bridges. This method is thread-safe and can be called before or after Start(). Example: AddRoute("billing.*", "kafka-billing", "webhook-alerts")

func (*ManagedOutbox) Enqueue 

func (m *ManagedOutbox) Enqueue(ctx context.Context, entry Entry) (Message, error)

Enqueue adds a message to the outbox.

The message will be stored in the outbox table and will be dispatched by the background dispatcher when it becomes available. Returns the created message with its ID and status.

func (*ManagedOutbox) EnqueueTx 

func (m *ManagedOutbox) EnqueueTx(ctx context.Context, tx *sql.Tx, entry Entry) (Message, error)

EnqueueTx adds a message to the outbox within a transaction.

This is the recommended method for transactional outbox pattern usage. The message is enqueued within the provided transaction, ensuring atomicity with other database operations. If the transaction is rolled back, the message will not be persisted.

Example: 

tx, _ := db.BeginTx(ctx, nil)
// ... perform domain writes ...
managed.EnqueueTx(ctx, tx, outbox.Entry{Topic: "order.created", Payload: ...})
tx.Commit()

func (*ManagedOutbox) RegisterBridge 

func (m *ManagedOutbox) RegisterBridge(bridge Bridge) error

RegisterBridge adds a bridge to the registry.

This method is thread-safe and can be called before or after Start(). Bridges registered after Start() will be used for subsequent message dispatches.

func (*ManagedOutbox) Registry 

func (m *ManagedOutbox) Registry() *BridgeRegistry

Registry returns the bridge registry.

This provides access to the bridge registry for advanced use cases such as dynamically adding/removing bridges at runtime.

func (*ManagedOutbox) RequeueFailed 

func (m *ManagedOutbox) RequeueFailed(ctx context.Context, ids ...string) (int64, error)

RequeueFailed forwards to the managed store — see Store.RequeueFailed.

func (*ManagedOutbox) Router 

func (m *ManagedOutbox) Router() *Router

Router returns the topic router.

This provides access to the topic router for advanced use cases such as dynamically adding/removing routing rules at runtime.

func (*ManagedOutbox) Snapshot 

func (m *ManagedOutbox) Snapshot(ctx context.Context) RuntimeSnapshot

Snapshot returns the current outbox state.

This method queries the outbox table and returns counts of messages by status (pending, processing, delivered, failed) along with timestamps for the oldest pending and last delivered messages. Useful for monitoring and health checks.

func (*ManagedOutbox) Start 

func (m *ManagedOutbox) Start(ctx context.Context) error

Start begins the dispatcher in a background goroutine.

The dispatcher will poll the outbox table for pending messages and deliver them to configured bridges. This method is non-blocking and returns immediately after starting the goroutine.

The context is used to stop the dispatcher when the application shuts down. Canceling the context will cause the dispatcher to stop gracefully.

Returns an error if the dispatcher is already running.

func (*ManagedOutbox) Stop 

func (m *ManagedOutbox) Stop(ctx context.Context) error

Stop gracefully shuts down the outbox.

This method stops the dispatcher and closes all registered bridges. It is safe to call multiple times. If the dispatcher is not running, this method returns nil immediately.

func (*ManagedOutbox) Store 

func (m *ManagedOutbox) Store() *Store

Store returns the underlying store for direct access if needed.

This provides access to the low-level store interface for advanced use cases that require direct database operations. Most applications should use the Enqueue/EnqueueTx methods instead.

type Message 

type Message struct {
	ID          string
	Topic       string
	Payload     []byte
	Status      Status
	Attempts    int
	AvailableAt time.Time
	CreatedAt   time.Time
	DeliveredAt time.Time
	LastError   string
}

type MissingRoutePolicy 

type MissingRoutePolicy string

MissingRoutePolicy controls how bridge dispatch handles a message whose topic has no configured bridge route. 

const (
	// MissingRouteError keeps the message durable by retrying/failing it.
	MissingRouteError MissingRoutePolicy = "error"
	// MissingRouteIgnore preserves the old drop-on-the-floor behaviour only
	// when an application has explicitly opted into it.
	MissingRouteIgnore MissingRoutePolicy = "ignore"
)

type Router 

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

Router determines which bridges should receive a message based on topic patterns.

The router maintains a list of routing rules and matches incoming message topics against these rules to determine which bridges should receive the message. Multiple bridges can receive the same message, enabling fan-out patterns.

func NewRouter 

func NewRouter() *Router

NewRouter creates an empty topic router.

func (*Router) AddRoute 

func (r *Router) AddRoute(pattern string, bridgeNames ...string)

AddRoute adds a topic pattern to bridge mapping.

When a message with a matching topic is dispatched, all bridges listed in bridgeNames will receive the message. This method is thread-safe.

func (*Router) Match 

func (r *Router) Match(topic string) []string

Match returns the list of bridge names that match the given topic.

The topic is matched against all registered routing rules in order. Bridge names are deduplicated so each bridge appears at most once in the result. This method is thread-safe.

type RuntimeSnapshot 

type RuntimeSnapshot struct {
	Enabled         bool   `json:"enabled"`
	Table           string `json:"table"`
	Flavor          string `json:"flavor,omitempty"`
	Reason          string `json:"reason,omitempty"`
	Pending         int    `json:"pending"`
	Processing      int    `json:"processing"`
	Delivered       int    `json:"delivered"`
	Failed          int    `json:"failed"`
	Total           int    `json:"total"`
	OldestPendingAt string `json:"oldest_pending_at,omitempty"`
	LastDeliveredAt string `json:"last_delivered_at,omitempty"`
}

func InspectRuntime 

func InspectRuntime(db *sql.DB, cfg Config) RuntimeSnapshot

type Status 

type Status string
const (
	StatusPending    Status = "pending"
	StatusProcessing Status = "processing"
	StatusDelivered  Status = "delivered"
	StatusFailed     Status = "failed"
)

type Store 

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

func NewStore 

func NewStore(db *sql.DB, cfg Config) (*Store, error)

func (*Store) Enqueue 

func (s *Store) Enqueue(ctx context.Context, entry Entry) (Message, error)

func (*Store) EnqueueTx 

func (s *Store) EnqueueTx(ctx context.Context, tx *sql.Tx, entry Entry) (Message, error)

func (*Store) RequeueFailed 

func (s *Store) RequeueFailed(ctx context.Context, ids ...string) (int64, error)

RequeueFailed returns messages in the "failed" state to "pending" so the dispatcher picks them up again (NF-8: a message that exhausted MaxAttempts stayed failed forever, and the only remedy was hand-written SQL).

With no ids, every failed message is requeued; with ids, only those — an id that is not currently failed is left untouched (delivered or in-flight messages must not be re-run by a typo'd id).

A requeued message gets its full retry budget back (attempts reset to 0) and becomes available immediately. last_error is preserved until the next attempt overwrites it, so the reason it failed remains inspectable right up to the retry.

Returns the number of messages actually requeued.

func (*Store) Snapshot 

func (s *Store) Snapshot(ctx context.Context) RuntimeSnapshot

type TopicRouting 

type TopicRouting struct {
	Pattern string   `json:"pattern"` // e.g., "billing.*" or "orders.created"
	Bridges []string `json:"bridges"` // bridge names to route matching messages to
}

TopicRouting defines how messages are routed to bridges based on topic patterns.

The pattern field supports wildcard matching:

  • "*" matches any single segment
  • "prefix.*" matches any topic with the given prefix
  • Exact string matches the topic exactly

Example patterns:

  • "billing.*" matches "billing.invoice.created", "billing.payment.received"
  • "orders.created" matches only "orders.created"
  • "*" matches all topics

type WebhookBridge 

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

WebhookBridge delivers outbox messages via HTTP webhooks.

This bridge sends outbox messages as HTTP POST requests to a configured URL. The message payload is serialized as JSON and includes the message ID, topic, payload, status, and metadata. Custom headers can be configured for authentication and other purposes.

Every delivery carries the WebhookPayloadEncodingHeader declaring the payload shape, and — when a Secret is configured — the WebhookSignatureHeader with an HMAC-SHA256 signature of the body.

Example usage: 

bridge, err := outbox.NewWebhookBridge(outbox.WebhookConfig{
    Name:   "notifications",
    URL:    "https://api.example.com/webhooks",
    Secret: os.Getenv("NOTIFICATIONS_WEBHOOK_SECRET"),
    Headers: map[string]string{
        "Authorization": "Bearer token",
    },
})

func NewWebhookBridge 

func NewWebhookBridge(cfg WebhookConfig) (*WebhookBridge, error)

NewWebhookBridge creates a new webhook bridge.

Returns an error if the name or URL is empty, or if PayloadEncoding is neither empty (meaning PayloadEncodingBase64), PayloadEncodingBase64 nor PayloadEncodingJSON. The HTTP client is configured with the specified timeout (default 30 seconds).

func (*WebhookBridge) Close 

func (b *WebhookBridge) Close() error

Close closes the HTTP client.

This method closes idle connections to release resources. It is safe to call multiple times.

func (*WebhookBridge) Healthy 

func (b *WebhookBridge) Healthy(ctx context.Context) error

Healthy checks if the webhook endpoint is reachable.

This method performs a GET request to the configured URL and checks the response. Status codes 2xx-4xx are considered healthy (the endpoint is responding). Status code 5xx indicates a server error and is considered unhealthy.

Note: Some webhook endpoints may not support GET requests. In such cases, consider disabling health checks or implementing a custom health check endpoint.

func (*WebhookBridge) Name 

func (b *WebhookBridge) Name() string

Name returns the bridge name.

func (*WebhookBridge) Send 

func (b *WebhookBridge) Send(ctx context.Context, msg Message) error

Send delivers a message via HTTP POST.

The message is serialized as JSON with the following structure: 

{
  "id": "message-id",
  "topic": "event.topic",
  "payload": "eyJvcmRlcl9pZCI6NDJ9",
  "status": "pending",
  "attempts": 1,
  "available_at": "2024-01-01T00:00:00Z",
  "created_at": "2024-01-01T00:00:00Z"
}

The shape of the "payload" field is governed by WebhookConfig.PayloadEncoding and declared per delivery in the WebhookPayloadEncodingHeader, which is always present:

  • PayloadEncodingBase64 (the default): the field is a JSON string with the base64 encoding of the raw payload bytes — byte-for-byte the wire shape of every tagged release up to v1.4.0, so existing consumers keep working without changes. The header is "base64".
  • PayloadEncodingJSON (opt-in): the field embeds the payload verbatim as JSON — Store.Enqueue encodes Entry.Payload with encoding/json, so Message.Payload is a JSON document by construction and consumers read it directly. The header is "json". A payload that is not valid JSON — possible only for a Message built by hand rather than read from the store — falls back to the base64-string form for that delivery, and the header declares "base64" accordingly.

Under either mode a message with no payload puts JSON null in the field.

When WebhookConfig.Secret is set, the request also carries WebhookSignatureHeader with the HMAC-SHA256 signature of the exact body bytes ("sha256=<hex>", identical to module webhooks).

Returns an error if the HTTP request fails or returns a non-2xx status code. The response body is included in error messages for debugging.

type WebhookConfig 

type WebhookConfig struct {
	Name            string
	URL             string
	Headers         map[string]string
	Timeout         time.Duration
	Secret          string
	PayloadEncoding string
}

WebhookConfig configures a webhook bridge.

The URL field is required and must be a valid HTTP/HTTPS endpoint. Headers can be used for authentication (e.g., Bearer tokens) or custom metadata. Timeout defaults to 30 seconds if not specified.

Secret, when non-empty, makes the bridge sign every delivery body with HMAC-SHA256 and send the result as WebhookSignatureHeader ("sha256=<hex>") — the exact scheme module webhooks verify, so consumers can share one verifier. An empty Secret sends unsigned deliveries: the consumer must authenticate the caller by other means.

PayloadEncoding selects the wire shape of the "payload" field: PayloadEncodingBase64 (the default when empty — the classic shape every tagged release up to v1.4.0 emits) or PayloadEncodingJSON (opt-in: the payload JSON document is embedded verbatim, saving the consumer a base64 round-trip). Any other value is rejected by NewWebhookBridge. Whatever the mode, each delivery declares its actual payload shape in WebhookPayloadEncodingHeader.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.