Documentation
¶
Overview ¶
Package outbox is wowapi's transactional outbox: modules write domain events into events_outbox in the SAME transaction as their business writes, so an event is emitted if and only if the write commits (no lost or phantom events). A relay later claims pending events and dispatches them to idempotent handlers, deduped via the processed_events inbox. Contract: blueprint 07 §3/§7.
Index ¶
- func CountDeadEvents(ctx context.Context, pool *pgxpool.Pool) (int64, error)
- func DiscardDeadEvent(ctx context.Context, pool *pgxpool.Pool, id uuid.UUID) error
- func PublishDLQDepth(ctx context.Context, pool *pgxpool.Pool, m observability.Metrics) error
- func ReplayDeadEvent(ctx context.Context, pool *pgxpool.Pool, id uuid.UUID) error
- type DeadEventEntry
- type DispatchedEvent
- type Event
- type Handler
- type HandlerRegistry
- type Relay
- type RelayOption
- type Writer
- type WriterOption
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func CountDeadEvents ¶
CountDeadEvents returns the number of dead-lettered (dispatch_status='dead') outbox events — the events contribution to DLQ depth (roadmap CA-1 / backlog B-8). Runs on the platform pool (events_outbox is read cross-tenant as app_platform via the outbox_relay_all policy).
func DiscardDeadEvent ¶
DiscardDeadEvent permanently deletes a dead event. Returns KindNotFound if id is not a dead event.
func PublishDLQDepth ¶
PublishDLQDepth counts dead-lettered events and sets the dlq_depth{queue="events"} gauge on m. Drive it from the leader-safe scheduler so the depth is counted once, not once per replica. m may be nil (no emission); otherwise it is the shared sink — observability.NoOp when no adapter is wired.
Types ¶
type DeadEventEntry ¶
type DeadEventEntry struct {
ID uuid.UUID
TenantID uuid.UUID
EventType string
Attempts int
MaxAttempts int
LastError string
FailedAt *time.Time
Payload []byte
}
DeadEventEntry is a dead-lettered outbox event for inspection.
func ListDeadEvents ¶
ListDeadEvents returns dead-lettered events, most recently failed first.
type DispatchedEvent ¶
type DispatchedEvent struct {
ID uuid.UUID
Type string
SchemaVersion int
Resource resource.Ref
Actor json.RawMessage
Payload json.RawMessage
TenantID uuid.UUID
}
DispatchedEvent is what a handler receives: the envelope plus the raw payload bytes (the handler unmarshals into its own typed struct).
type Event ¶
type Event struct {
ID uuid.UUID
Type string
SchemaVersion int
Resource resource.Ref
Actor json.RawMessage // opaque actor descriptor; never a secret
Payload any
// TenantID is set by the writer from the tx's tenant; callers leave it zero.
TenantID uuid.UUID
}
Event is the outbox envelope. Type is "module.resource.verb_past"; Payload is the module's event struct (additive within a SchemaVersion). ID and OccurredAt are assigned on write when zero.
type Handler ¶
Handler processes a dispatched event within a tenant transaction. It must be idempotent (the inbox dedups redelivery, but a handler should tolerate it).
type HandlerRegistry ¶
type HandlerRegistry struct {
// contains filtered or unexported fields
}
HandlerRegistry collects event subscriptions during module registration.
func NewHandlerRegistry ¶
func NewHandlerRegistry() *HandlerRegistry
NewHandlerRegistry returns an empty registry.
func (*HandlerRegistry) Err ¶
func (r *HandlerRegistry) Err() error
Err returns accumulated subscription errors joined, or nil.
func (*HandlerRegistry) Subscribe ¶
func (r *HandlerRegistry) Subscribe(eventType, handlerName string, fn Handler)
Subscribe registers an idempotent handler for an event type. handlerName must be unique per event type and stable across deploys (it keys the inbox).
type Relay ¶
type Relay struct {
// contains filtered or unexported fields
}
Relay claims pending outbox events and dispatches them to registered handlers. It reads across tenants on a platform-privileged pool (app_platform — the relay RLS policy admits all rows, D-0048), then RE-ENTERS a tenant transaction bound to each event's tenant to run handlers under normal tenant RLS and the processed_events inbox.
Ordering is per-aggregate (blueprint 07 §7): the claim only picks the earliest still-undispatched event for each (tenant, resource), so a later event never overtakes an earlier pending/failed one, and a transaction-scoped advisory lock keyed on the aggregate serializes concurrent relays. Handlers get an exactly-once DB EFFECT via the inbox (dedup + effect share the tenant tx); external side effects in a handler are still at-least-once. A poison event dead-letters ('dead') after max_attempts rather than retrying forever.
func NewRelay ¶
func NewRelay(pool *pgxpool.Pool, txm database.TxManager, registry *HandlerRegistry, batchSize int, opts ...RelayOption) *Relay
NewRelay builds the relay. pool must authenticate as the relay role (app_platform); txm runs handler transactions per tenant.
func (*Relay) DispatchOnce ¶
DispatchOnce claims up to batch pending events (FOR UPDATE SKIP LOCKED, so concurrent relays never double-claim), dispatches each to its handlers, and marks it dispatched (or failed, to retry later). It returns the number of events processed. A relay loop calls this until it returns 0, then sleeps.
func (*Relay) RequeueFailed ¶
RequeueFailed resets 'failed' events (not 'dead' — those are terminal) back to 'pending' once the last failure is older than cooldown. The cooldown is keyed on failed_at (the actual failure time), not occurred_at (write time), so a just-failed event actually waits (review finding ARCH-55). Dead-lettered events are left for the admin requeue path.
type RelayOption ¶
type RelayOption func(*Relay)
RelayOption customizes the relay.
func WithRelayTracer ¶
func WithRelayTracer(tr observability.Tracer) RelayOption
WithRelayTracer wires a tracer so the relay continues the originating request's trace when it dispatches an event (roadmap O1/CA-9): it extracts the event's stored traceparent and runs the handler under a child span. Default: NoOpTracer.
type Writer ¶
Writer writes events into the outbox within the caller's tenant transaction. Stateless: Write takes the tx's TenantDB so the event commits atomically with the business write.
type WriterOption ¶
type WriterOption func(*pgWriter)
WriterOption customizes the outbox writer.
func WithWriterTracer ¶
func WithWriterTracer(tr observability.Tracer) WriterOption
WithWriterTracer wires a tracer so each written event captures the current request's W3C traceparent (roadmap O1/CA-9); the relay continues that trace when it dispatches the event. Default: NoOpTracer (empty trace context).
Source Files