outbox

type BacklogCollector ¶ added in v0.5.0

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

BacklogCollector is a prometheus.Collector that, on each scrape, reads the outbox backlog via a StatsReader and reports it as gauges:

servicekit_outbox_pending             — events awaiting the relay
servicekit_outbox_in_flight           — events leased, publish in progress
servicekit_outbox_failed              — terminal failures awaiting attention
servicekit_outbox_oldest_pending_seconds — age of the oldest pending event

It queries the store on every scrape, so keep the scrape interval sane across replicas. A query error is logged and that scrape emits no outbox gauges (rather than stale values). Register it on the shared registry like any collector.

func NewBacklogCollector(sr StatsReader, log *logger.Logger, opts ...BacklogOption) *BacklogCollector

func (c *BacklogCollector) Collect(ch chan<- prometheus.Metric)

func (c *BacklogCollector) Describe(ch chan<- *prometheus.Desc)

type BacklogOption ¶ added in v0.5.0

type BacklogOption func(*BacklogCollector)

BacklogOption customizes a BacklogCollector.

func WithScrapeTimeout(d time.Duration) BacklogOption

type CleanerConfig ¶

type CleanerConfig struct {
	Name string // default "outbox-cleaner"
	// Interval between cleanups (default 1h).
	Interval time.Duration
	// Retention: sent/failed rows older than this are deleted (default 24h).
	Retention time.Duration
	// Metrics, when set, counts deleted rows.
	Metrics *Metrics
}

CleanerConfig configures the Cleaner worker.

type Event ¶

type Event struct {
	ID            uuid.UUID
	Type          string
	ContentType   string
	Topic         string
	Key           string
	Payload       []byte
	Headers       []byte // JSON object of transport headers ("{}" when none)
	Status        string
	Attempts      int
	MaxAttempts   int
	LastError     string
	NextAttemptAt time.Time
	CreatedAt     time.Time
	SentAt        time.Time
	LeasedAt      time.Time
	LeaseID       uuid.UUID
}

Event is an outbox record as seen by callers — a pure core type with no db tags or Null wrappers (the PG store owns that mapping). Payload is the business data only; the broker.Publisher wraps it in a CloudEvents envelope.

Routing is transport-neutral: Topic is the logical destination and Key an optional routing/ordering key. Each broker maps them to its own concepts (RabbitMQ exchange+routing key, Kafka topic+partition key, NATS subject), so the same event row can be delivered over any transport.

func NewEvent(eventType, topic, key, contentType string, payload []byte, headers map[string]any) (Event, error)

type Marshaler ¶ added in v0.4.0

type Marshaler func(v any) ([]byte, error)

Marshaler serializes a domain event value into the bytes stored as an event's payload. The default is encoding/json.Marshal; override per type with WithMarshaler (e.g. for protobuf).

type Metrics ¶ added in v0.5.0

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

Metrics are the outbox's own throughput counters, owned by this package and registered on the shared registry passed to NewMetrics. Wire it into RelayConfig/SweeperConfig/CleanerConfig to record relay activity; a nil *Metrics is a safe no-op, so metrics stay optional.

Backlog gauges (depth, age) are separate — they need a query, so they live in the BacklogCollector rather than being incremented inline here.

func NewMetrics(reg prometheus.Registerer) *Metrics

type Options ¶

type Options struct {
	// Backoff schedules retries (next_attempt_at) for failed-but-retryable
	// events. Defaults to base 2s, factor 2, max 5m.
	Backoff worker.Backoff
}

Options configures a PG store.

type PG ¶

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

PG is the Postgres-backed implementation of Store (and StatsReader). Every query is an inline const right next to the method that runs it — no string-templated query cache, no fmt.Sprintf — so the SQL reads as SQL. The backing table is fixed (outbox_events); status values are bound as named params from the Status* constants so the SQL and Go agree on one source of truth.

All statements go through the sqldb named-query helpers for consistent tracing and error wrapping. The Store holds its handle as sqlx.ExtContext, so the same type works against either a *sqlx.DB (pool-bound) or a *sqlx.Tx (transaction-bound, via WithTx) — that is how Insert joins the caller's domain transaction in WithinTran.

PG is safe for concurrent use by multiple relay/sweeper replicas: LeasePending and SweepExpiredLeases claim rows with FOR UPDATE SKIP LOCKED.

func NewPG(log *logger.Logger, db *sqlx.DB, opts Options) *PG

func (s *PG) Cleanup(ctx context.Context, retention time.Duration, now time.Time) (int64, error)

func (s *PG) EnsureSchema(ctx context.Context) error

func (s *PG) Insert(ctx context.Context, events ...Event) error

func (s *PG) LeasePending(ctx context.Context, now time.Time, limit int) ([]Event, error)

func (s *PG) MarkFailed(ctx context.Context, ev Event, leaseID uuid.UUID, errMsg string, now time.Time) error

func (s *PG) MarkSent(ctx context.Context, ev Event, leaseID uuid.UUID, now time.Time) error

func (s *PG) Stats(ctx context.Context, now time.Time) (Stats, error)

func (s *PG) SweepExpiredLeases(ctx context.Context, leaseTimeout time.Duration, now time.Time, limit int) (int64, error)

func (s *PG) WithTx(tx sqlx.ExtContext) Store

type PublishFunc ¶ added in v0.4.0

type PublishFunc func(tx *sqlx.Tx, pub Publisher) error

PublishFunc is the caller's transactional closure. It receives the transaction handle for domain writes and a tx-bound Publisher for emitting events. Returning an error rolls back everything — domain writes and the events recorded through pub together.

The domain code inside fn never sees pub's mechanics: it calls pub.Publish(ctx, SomeEvent{...}) with a plain typed value. The tx handle is here so an application can bind its own store to the transaction (e.g. store.WithTx(tx)); domain methods themselves take the bound store, not tx.

type PublishOption ¶ added in v0.4.0

type PublishOption func(*route)

PublishOption overrides the registered route for a single Publish. The Go type still resolves marshaling and the route defaults from the Registry; these options replace individual fields for this call only.

func As(eventType string) PublishOption

func OnTopic(topic string) PublishOption

func WithRouteKey(key string) PublishOption

type Publisher ¶ added in v0.4.0

type Publisher interface {
	// Publish records event in the outbox within the bound transaction. event's
	// Go type must be registered; its route (topic, key, content type) and
	// marshaling are resolved from the Registry. Returns an error if the type is
	// unregistered or marshaling/insert fails.
	//
	// opts override the registered route per call. This is how one Go payload
	// type fans out to several CloudEvents types chosen at runtime: register the
	// type once with a default route, then pass As(...) (and optionally
	// OnTopic/WithRouteKey) to pick the concrete event for this publish.
	Publish(ctx context.Context, event any, opts ...PublishOption) error
}

Publisher records domain events into the outbox. The domain depends only on this interface: it publishes a plain typed value and knows nothing about the transaction, the database, or the transport. A tx-bound Publisher is provided to the WithinTran closure (or built directly with Bind), so the recorded event commits atomically with the domain writes in the same transaction.

This is distinct from broker.Publisher: that one delivers an already-stored event to the wire (used by the Relay); this one writes the event into the outbox table inside the producer's transaction.

func Bind(tx sqlx.ExtContext, store Store, reg *Registry) Publisher

type Registry ¶ added in v0.4.0

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

Registry maps a domain event's Go type to its transport route (event type name, topic, key, content type, and how to marshal it). It lets the domain publish a plain typed value while the routing — which broker topic, which key — stays a wiring concern: the domain never names a topic or exchange.

Populate it once at startup with Register, then hand it to WithinTran / Bind. Registry is safe for concurrent reads after setup.

func NewRegistry() *Registry

type RelayConfig ¶

type RelayConfig struct {
	// Name labels the worker (default "outbox-relay").
	Name string
	// PollInterval is the idle tick rate (default 2s) — the wait after a tick
	// that found no work. The relay paces adaptively: after a full batch it
	// drains again immediately, so a backlog clears without waiting a full
	// interval. It also runs an immediate first tick on startup.
	PollInterval time.Duration
	// MaxPollInterval, when greater than PollInterval, makes the idle wait back
	// off geometrically up to this cap over consecutive empty ticks (a quiet
	// relay polls less often). 0 keeps the idle wait at PollInterval.
	MaxPollInterval time.Duration
	// BatchSize is the max events leased per tick (default 100).
	BatchSize int
	// PublishTimeout bounds each publish + its follow-up mark (default 5s).
	PublishTimeout time.Duration
	// Metrics, when set, records relay throughput (published/failed/latency).
	Metrics *Metrics
}

RelayConfig configures the Relay worker.

type RouteOption ¶ added in v0.4.0

type RouteOption func(*route)

RouteOption customizes a registration.

func WithContentType(ct string) RouteOption

func WithKey(key string) RouteOption

func WithMarshaler(m Marshaler) RouteOption

type Stats ¶ added in v0.5.0

type Stats struct {
	Pending          int64
	InFlight         int64
	Failed           int64
	OldestPendingAge time.Duration // age of the oldest pending row; 0 when none
}

Stats is a point-in-time snapshot of the outbox table used for backlog metrics: how much work is waiting and how old the oldest waiting event is.

type StatsReader ¶ added in v0.5.0

type StatsReader interface {
	Stats(ctx context.Context, now time.Time) (Stats, error)
}

StatsReader reports outbox backlog stats. *PG implements it; it is a separate capability from Store so the core port stays minimal and backlog metrics stay opt-in.

type Store ¶

type Store interface {
	// WithTx returns a sibling Store whose queries run on tx. Used by WithinTran
	// to insert events in the caller's domain transaction.
	WithTx(tx sqlx.ExtContext) Store

	// Insert writes pending events. Bound to a transaction when the Store was
	// derived via WithTx.
	Insert(ctx context.Context, events ...Event) error

	// LeasePending atomically claims up to limit pending events whose
	// next_attempt_at <= now, marks them in_flight with a fresh lease id, and
	// returns them (FOR UPDATE SKIP LOCKED, so relay replicas split work).
	LeasePending(ctx context.Context, now time.Time, limit int) ([]Event, error)

	// MarkSent transitions in_flight -> sent. Guarded by leaseID; returns
	// ErrLeaseLost on a lease mismatch.
	MarkSent(ctx context.Context, ev Event, leaseID uuid.UUID, now time.Time) error

	// MarkFailed transitions in_flight -> pending (attempts+1 < max, with
	// next_attempt_at advanced by backoff) or -> failed (attempts+1 >= max).
	// Guarded by leaseID; returns ErrLeaseLost on a lease mismatch.
	MarkFailed(ctx context.Context, ev Event, leaseID uuid.UUID, errMsg string, now time.Time) error

	// SweepExpiredLeases returns in_flight rows whose lease is older than
	// leaseTimeout back to pending (a crashed-relay signal — attempts is NOT
	// incremented). Returns the number reclaimed.
	SweepExpiredLeases(ctx context.Context, leaseTimeout time.Duration, now time.Time, limit int) (int64, error)

	// Cleanup deletes sent/failed rows created before now-retention. Returns the
	// number of rows deleted.
	Cleanup(ctx context.Context, retention time.Duration, now time.Time) (int64, error)
}

Store is the persistence port for the outbox table. The Postgres implementation is *PG.

type SweeperConfig ¶

type SweeperConfig struct {
	Name string // default "outbox-sweeper"
	// Interval between sweeps (default 30s).
	Interval time.Duration
	// LeaseTimeout: in_flight rows leased longer than this are reclaimed
	// (default 1m). Set comfortably above RelayConfig.PublishTimeout.
	LeaseTimeout time.Duration
	// BatchSize bounds rows reclaimed per sweep (default 100).
	BatchSize int
	// Metrics, when set, counts reclaimed leases.
	Metrics *Metrics
}

SweeperConfig configures the Sweeper worker.