Documentation
¶
Overview ¶
Package txoutbox provides reusable transaction hook and outbox dispatch helpers.
Index ¶
- Constants
- Variables
- func ContextWithHooks(ctx context.Context, hooks *Hooks) context.Context
- func WithTxHooks[Tx any](ctx context.Context, txManager TxManager[Tx], ...) error
- func WithTxHooksContext[Tx any](ctx context.Context, txManager TxManager[Tx], ...) error
- type ClaimInput
- type DispatchInput
- type DispatchResult
- type Hooks
- type Message
- type PostCommitHook
- type Publisher
- type Query
- type Store
- type TxManager
Constants ¶
const ( OutboxStatusPending = "pending" OutboxStatusProcessing = "processing" OutboxStatusRetrying = "retrying" OutboxStatusSucceeded = "succeeded" OutboxStatusFailed = "failed" )
Variables ¶
var ( ErrStoreNotConfigured = errors.New("txoutbox: store not configured") ErrPublisherNotConfigured = errors.New("txoutbox: publisher not configured") )
Functions ¶
func ContextWithHooks ¶ added in v0.27.0
ContextWithHooks injects hook collectors into a context for nested tx-hook aware operations.
func WithTxHooks ¶
func WithTxHooks[Tx any](ctx context.Context, txManager TxManager[Tx], fn func(tx Tx, hooks *Hooks) error) error
WithTxHooks executes fn in a write transaction and runs queued post-commit hooks only when commit succeeds.
func WithTxHooksContext ¶ added in v0.27.0
func WithTxHooksContext[Tx any](ctx context.Context, txManager TxManager[Tx], fn func(ctx context.Context, tx Tx, hooks *Hooks) error) error
WithTxHooksContext is the context-aware variant of WithTxHooks. It passes a context carrying the active hook collector to fn so nested calls can defer side effects until the outermost commit boundary.
Types ¶
type ClaimInput ¶
type ClaimInput struct {
Consumer string `json:"consumer"`
Now time.Time `json:"now"`
Limit int `json:"limit"`
Topic string `json:"topic"`
LockUntil *time.Time `json:"lock_until"`
}
ClaimInput controls batched outbox claiming.
type DispatchInput ¶
type DispatchInput struct {
Consumer string `json:"consumer"`
Topic string `json:"topic"`
Limit int `json:"limit"`
Now time.Time `json:"now"`
RetryDelay time.Duration `json:"retry_delay"`
ClaimLockTo *time.Time `json:"claim_lock_to"`
}
DispatchInput controls outbox dispatch behavior.
type DispatchResult ¶
type DispatchResult struct {
Claimed int `json:"claimed"`
Published int `json:"published"`
Retrying int `json:"retrying"`
Failed int `json:"failed"`
}
DispatchResult captures batch dispatch outcomes.
func DispatchBatch ¶
func DispatchBatch[Scope any](ctx context.Context, store Store[Scope], scope Scope, publisher Publisher, input DispatchInput) (DispatchResult, error)
DispatchBatch claims pending outbox messages and publishes them.
type Hooks ¶
type Hooks struct {
// contains filtered or unexported fields
}
Hooks tracks post-commit hooks collected during a transaction callback.
func HooksFromContext ¶ added in v0.27.0
HooksFromContext returns the active hook collector from context when present.
func (*Hooks) AfterCommit ¶
func (h *Hooks) AfterCommit(hook PostCommitHook)
AfterCommit registers a hook to run after transaction commit.
type Message ¶
type Message struct {
ID string `json:"id"`
TenantID string `json:"tenant_id"`
OrgID string `json:"org_id"`
Topic string `json:"topic"`
MessageKey string `json:"message_key"`
PayloadJSON string `json:"payload_json"`
HeadersJSON string `json:"headers_json"`
CorrelationID string `json:"correlation_id"`
Status string `json:"status"`
AttemptCount int `json:"attempt_count"`
MaxAttempts int `json:"max_attempts"`
LastError string `json:"last_error"`
AvailableAt time.Time `json:"available_at"`
LockedAt *time.Time `json:"locked_at"`
LockedBy string `json:"locked_by"`
PublishedAt *time.Time `json:"published_at"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
Message stores durable side-effect events for post-commit dispatch.
type PostCommitHook ¶
type PostCommitHook func() error
PostCommitHook defines work executed only after a successful transaction commit.
type Query ¶
type Query struct {
Topic string `json:"topic"`
Status string `json:"status"`
Limit int `json:"limit"`
Offset int `json:"offset"`
SortDesc bool `json:"sort_desc"`
}
Query controls outbox listing filters.
type Store ¶
type Store[Scope any] interface { EnqueueOutboxMessage(ctx context.Context, scope Scope, record Message) (Message, error) ClaimOutboxMessages(ctx context.Context, scope Scope, input ClaimInput) ([]Message, error) MarkOutboxMessageSucceeded(ctx context.Context, scope Scope, id string, publishedAt time.Time) (Message, error) MarkOutboxMessageFailed(ctx context.Context, scope Scope, id, failureReason string, nextAttemptAt *time.Time, failedAt time.Time) (Message, error) ListOutboxMessages(ctx context.Context, scope Scope, query Query) ([]Message, error) }
Store defines durable post-commit side-effect message persistence.
Source Files