Documentation
¶
Overview ¶
Package outbox implements at-least-once EXTERNAL event delivery: the unit of work writes envelopes into an outbox table in the same transaction as the change, and a relay dispatches them to the registered external hooks, retrying on failure. Without the outbox, a crash between commit and dispatch loses events; with it, every external consumer (webhooks, pub/sub) sees each committed change at least once. Internal projections (computed attributes, search index, GraphQL cache) do NOT ride the relay — they are maintained synchronously in the writing request regardless of the outbox (see application/uow and issue #211).
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Relay ¶
type Relay struct {
// contains filtered or unexported fields
}
Relay drains the outbox: on a nudge (post-commit) or on the interval it fetches pending envelopes, dispatches them and records outcomes. Failed envelopes stay pending and retry on later passes.
func NewRelay ¶
func NewRelay(store Store, dispatcher *events.Dispatcher, opts ...RelayOption) *Relay
NewRelay builds a relay over the store and dispatcher.
func (*Relay) DrainOnce ¶
DrainOnce performs one drain pass — claiming, dispatching and finalizing batches until the outbox is empty. Run loops these on a nudge or the interval; DrainOnce is exposed for one-shot draining and tests.
type RelayOption ¶
type RelayOption func(*Relay)
RelayOption customises a Relay.
func WithAfterExpand ¶
func WithAfterExpand(fn func()) RelayOption
WithAfterExpand installs a hook invoked after a pass that expanded envelopes — the delivery worker's nudge, keeping webhook latency in milliseconds.
func WithBatchSize ¶
func WithBatchSize(n int) RelayOption
WithBatchSize sets how many envelopes one pass claims (default 100).
func WithErrorObserver ¶
func WithErrorObserver(fn func(err error)) RelayOption
WithErrorObserver receives relay-level failures (fetch errors — dispatch failures are recorded per envelope).
func WithInterval ¶
func WithInterval(d time.Duration) RelayOption
WithInterval sets the poll interval (default 2s).
func WithLeaseTTL ¶
func WithLeaseTTL(d time.Duration) RelayOption
WithLeaseTTL sets how long a claimed batch stays leased to this relay before another relay may reclaim it (default 1m). It should comfortably exceed the slowest expected dispatch of one batch.
func WithRelayID ¶
func WithRelayID(id string) RelayOption
WithRelayID sets the identifier stamped on leases (default a random ULID). Only useful for deterministic tests.
type Store ¶
type Store interface {
// Write persists envelopes inside the caller's transaction — the unit
// of work's pre-commit handler.
Write(ctx context.Context, tx db.Tx, envs []events.Envelope) error
// Claim leases up to limit undispatched envelopes for the given relay
// and returns them for dispatch. Claiming takes a short row lease
// (claimed_by/claimed_at) so no other relay grabs the same rows while
// this relay dispatches them outside any transaction; a lease older
// than leaseTTL is treated as abandoned (crashed relay) and reclaimed.
// It does NOT hold the sequencer lock — no network I/O happens here.
Claim(ctx context.Context, relayID string, limit int, leaseTTL time.Duration) ([]events.Envelope, error)
// Finalize records the outcome of dispatching a claimed batch. Under
// the single-sequencer advisory lock (DB-only, no network I/O) it
// assigns feed_seq to each success in claim order, fans out one
// webhook-delivery row per matching subscription and marks the
// envelope dispatched; failures have their attempt counted and their
// lease cleared so a later pass retries them.
Finalize(ctx context.Context, results []Result) error
}
Store is the persistence port for the outbox.
Source Files