outbox

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Aug 23, 2026 License: MIT Imports: 30 Imported by: 0

README

Transactional outbox

This package provides a storage-independent, at-least-once transactional outbox. Service is the single application boundary: it exposes storage capabilities, owns destination routing, polls the delivery store, runs one bounded worker pool, renews leases, settles outcomes, and participates in React shutdown.

The implementation deliberately has no instrumentation hooks on append, claim, delivery, retry, or query paths. Services receive react.ILogger from react.LoggerToken and log operational failures only.

Consumer API

Select one adapter and optionally provide the worker-pool configuration:

workerConfig := outbox.DefaultConfig()
workerConfig.WorkerCount = 8

outboxModule := outbox.ForFeature(outbox.Postgres, workerConfig).
    Import(postgresConfigModule)

The selected adapter configuration remains a normal GIOC value:

storeConfig := outbox.DefaultPostgresConfig()
storeConfig.Namespace = "orders"

postgresConfigModule := gioc.NewModule("OrdersOutboxConfig").Provide(
    outbox.ProvidePostgresConfig(storeConfig),
)

Add React's application, logger, and adapter infrastructure modules before running the container. Then resolve the service and register each sink:

service, err := gioc.Get[*outbox.Service](
    container,
    outbox.OutboxServiceToken,
    outboxModule,
)
if err != nil {
    return err
}

err = service.Register(rabbitSink, outbox.DestinationsConfig{
    Destinations: []string{"orders.confirmed", "orders.cancelled"},
    Concurrency:  4,
})
if err != nil {
    return err
}

Register copies the configuration and atomically installs all routes. A destination can be registered only once. Concurrency bounds all deliveries through that registration and defaults to the global worker count. The global Config.WorkerCount remains the hard process-wide bound.

Records for an unregistered destination remain pending. Registering a route wakes the poller, so consumers may register after container startup without a polling delay.

For a small consumer, adapt a function directly with outbox.SinkFunc instead of declaring a separate sink type.

The service itself implements IStore, so producers can append directly:

records, err := service.Append(ctx, outbox.NewRecord{
    ID:             "order-42-confirmed",
    Destination:    "orders.confirmed",
    MessageType:    "order.confirmed",
    AggregateType:  "order",
    AggregateID:    "42",
    OrderingKey:    "42",
    IdempotencyKey: "order.confirmed:42",
    Payload:        payload,
    MaxAttempts:    12,
})

Prefer the narrow tokens when a consumer needs less authority:

  • OutboxAppenderToken provides IAppender.
  • OutboxReaderToken provides IReader.
  • OutboxDeliveryStoreToken provides IDeliveryStore.
  • OutboxMaintenanceStoreToken provides IMaintenanceStore.

All capability providers resolve Service, not the adapter directly. The adapter aggregate remains available at OutboxStoreToken for adapter-specific operations such as PostgreSQL transaction binding and migrations.

Atomicity boundary

A transactional outbox is atomic only when the domain mutation and outbox append commit in the same transactional resource. The package cannot create atomicity across unrelated databases.

For PostgreSQL, bind the outbox store to the caller-owned transaction:

tx, err := database.Begin(ctx)
if err != nil {
    return err
}
defer tx.Rollback(context.Background())

if _, err = tx.Exec(ctx,
    `UPDATE orders SET state='confirmed' WHERE id=$1`, orderID,
); err != nil {
    return err
}

if _, err = postgresStore.Bind(tx).Append(ctx, record); err != nil {
    return err
}
return tx.Commit(ctx)

The caller owns the outer transaction. Bind uses a savepoint for atomic batch validation and never commits or rolls back the outer transaction.

Redis composition is possible only when domain keys and outbox keys share the same Redis Cluster hash slot. See outbox.RedisCompositionRequest. PostgreSQL domain state plus a Redis outbox is not a transactional outbox.

Delivery model

The portable state machine is:

pending   --claim--------------------------> leased
leased    --acknowledge--------------------> delivered
leased    --retry or release---------------> pending
leased    --terminal/exhausted failure-----> dead
leased    --expired lease recovery---------> pending or dead
pending   --cancel--------------------------> cancelled
dead      --requeue-------------------------> pending
pending   --reschedule----------------------> pending

Claims are fenced by record ID, owner, unguessable lease token, version, and storage-authoritative expiry. Stale settlement receives ErrLeaseLost.

Delivery is at least once, never exactly once. A sink can publish successfully and the process can stop before Acknowledge is persisted. The record becomes eligible again after lease recovery. Consumers must deduplicate by stable record ID or implement an idempotent domain operation.

ISink.Deliver must return nil only after the downstream system has reliably accepted the message. For RabbitMQ that normally means a persistent mandatory publish followed by a positive publisher confirmation and no returned message.

Errors are classified as:

  • OutcomeRetryable: schedule another bounded attempt.
  • OutcomeTerminal: move directly to dead.
  • OutcomeAmbiguous: retry because loss is worse than a possible duplicate.

TerminalError selects terminal behavior with the default classifier. A custom IErrorClassifier or IRetryPolicy can be provided in Config.

Worker pool and shutdown

One poller claims only currently registered destinations. Large routing tables are rotated through bounded claim windows so every first-party adapter receives portable requests. Claims never hold a database transaction open while a sink performs network I/O.

Workers enforce delivery timeouts, renew long-running leases before their threshold, and settle with a bounded background context. Shutdown stops new claims first and allows current deliveries to finish. If the deadline is reached, delivery contexts are cancelled and still-current leases are released using their fences.

React registers this work as a pre-shutdown hook. The adapter facade closes in the normal shutdown tier after the worker pool. PostgreSQL pools, Redis clients, and broker connections remain owned by their infrastructure modules.

Storage adapters

PostgreSQL

outbox.PostgresStore provides durable storage with FOR UPDATE SKIP LOCKED, fenced mutations, indexed operational queries, transaction binding, and explicit migrations. Apply PostgresStore.Migrations() or PostgresStore.Migrate(ctx) during deployment before the service begins processing records. The application owns the pgx pool.

Redis

outbox.RedisStore uses bounded atomic Lua scripts and same-slot keys. Redis is authoritative storage, not a cache. Production use requires a deliberate AOF, replication, backup, failover, and noeviction posture. Startup durability checks can warn or fail according to DurabilityMode. The application owns the Redis client.

Logging

react.ILogger is an alias of author.ILogger. Service and each adapter store service resolve it from react.LoggerToken; the standard react.LoggerModule() supplies the implementation. Logging is limited to operational failures and durability warnings; payloads and headers are never logged by first-party code.

Validation

Run unit, lifecycle, routing, lease, retry, and adapter contract tests:

go test ./outbox
go test -race ./outbox

PostgreSQL and Redis integration tests use OUTBOX_POSTGRES_TEST_URL and OUTBOX_REDIS_TEST_URL when those variables are present.

Documentation

Overview

PostgresStore implements durable outbox storage using pgx. The caller owns the pool and every transaction passed to PostgresStore.Bind; the adapter never closes the pool or commits or rolls back the caller's outer transaction.

RedisStore implements storage with bounded, atomic Redis Lua scripts. It treats Redis as authoritative storage, not as a cache. Production deployments must deliberately configure persistence, noeviction, backups, replication, and failover.

Example (PostgresDomainTransaction) 
package main

import (
	"context"
	"os"

	"github.com/0x626f/react/outbox"

	"github.com/jackc/pgx/v5/pgxpool"
)

func main() {
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, os.Getenv("POSTGRES_URL"))
	if err != nil {
		return
	}
	defer pool.Close()
	store, err := outbox.NewPostgresStore(pool, outbox.DefaultPostgresConfig())
	if err != nil {
		return
	}
	tx, err := pool.Begin(ctx)
	if err != nil {
		return
	}
	defer tx.Rollback(context.Background())
	if _, err = tx.Exec(ctx, `UPDATE orders SET state='confirmed' WHERE id=$1`, "42"); err != nil {
		return
	}
	if _, err = store.Bind(tx).Append(ctx, exampleRecord("atomic-postgres")); err != nil {
		return
	}
	_ = tx.Commit(ctx)
}

func exampleRecord(id outbox.ID) outbox.NewRecord {
	return outbox.NewRecord{ID: id, Destination: "orders", MessageType: "order.confirmed", Payload: []byte(`{"order_id":"42"}`), MaxAttempts: 8}
}
Example (PostgresStandalone) 
package main

import (
	"context"
	"os"

	"github.com/0x626f/react/outbox"

	"github.com/jackc/pgx/v5/pgxpool"
)

func main() {
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, os.Getenv("POSTGRES_URL"))
	if err != nil {
		return
	}
	defer pool.Close()
	config := outbox.DefaultPostgresConfig()
	config.Namespace = "orders"
	store, err := outbox.NewPostgresStore(pool, config)
	if err != nil {
		return
	}
	if err = store.Migrate(ctx); err != nil {
		return
	}
	_, _ = store.Append(ctx, exampleRecord("standalone-postgres"))
}

func exampleRecord(id outbox.ID) outbox.NewRecord {
	return outbox.NewRecord{ID: id, Destination: "orders", MessageType: "order.confirmed", Payload: []byte(`{"order_id":"42"}`), MaxAttempts: 8}
}
Example (RabbitMQPublisherConfirms) 
package main

import (
	"context"
	"errors"

	"github.com/0x626f/react/outbox"
)

func main() {
	// IConfirmedPublisher must enable publisher confirms, use a stable message
	// ID, and resolve only after ack/nack plus mandatory-return handling.
	var publisher IConfirmedPublisher
	var sink outbox.ISink = rabbitConfirmedSink{publisher: publisher}
	_ = sink
}

// IConfirmedPublisher publishes a message and waits for its broker outcome.
type IConfirmedPublisher interface {
	PublishAndConfirm(ctx context.Context, destination string, messageID string, headers map[string]string, payload []byte) (routed bool, err error)
}

type rabbitConfirmedSink struct{ publisher IConfirmedPublisher }

func (sink rabbitConfirmedSink) Deliver(ctx context.Context, record outbox.Record) error {
	if sink.publisher == nil {
		return errors.New("publisher unavailable")
	}
	routed, err := sink.publisher.PublishAndConfirm(ctx, record.Destination, string(record.ID), record.Headers, record.Payload)
	if err != nil {
		return err
	}
	if !routed {
		return &outbox.TerminalError{Err: errors.New("message was returned as unroutable")}
	}
	return nil
}
Example (RedisDomainComposition) 
package main

import (
	"context"
	"os"

	"github.com/0x626f/react/outbox"

	goredis "github.com/redis/go-redis/v9"
)

func main() {
	options, err := goredis.ParseURL(os.Getenv("REDIS_URL"))
	if err != nil {
		return
	}
	client := goredis.NewClient(options)
	defer client.Close()
	config := outbox.DefaultRedisConfig()
	config.Namespace = "orders"
	store, err := outbox.NewRedisStore(context.Background(), client, config)
	if err != nil {
		return
	}
	domainKey := "react:outbox:{orders}:domain:order-42"
	_, _ = store.Compose(context.Background(), outbox.RedisCompositionRequest{
		Append:     outbox.AppendRequest{Records: []outbox.NewRecord{exampleRecord("composed-redis")}, DuplicateMode: outbox.RejectDuplicate},
		DomainKeys: []string{domainKey}, DomainArguments: []any{"pending", "confirmed"},
		ValidateLua: `
local value = redis.call('GET', KEYS[DOMAIN_KEY_OFFSET+1])
if value ~= ARGV[DOMAIN_ARG_OFFSET+1] then return {-3} end`,
		ApplyLua: `redis.call('SET', KEYS[DOMAIN_KEY_OFFSET+1], ARGV[DOMAIN_ARG_OFFSET+2])`,
	})
}

func exampleRecord(id outbox.ID) outbox.NewRecord {
	return outbox.NewRecord{ID: id, Destination: "orders", MessageType: "order.confirmed", Payload: []byte(`{"order_id":"42"}`), MaxAttempts: 8}
}
Example (RedisStandalone) 
package main

import (
	"context"
	"os"

	"github.com/0x626f/react/outbox"

	goredis "github.com/redis/go-redis/v9"
)

func main() {
	options, err := goredis.ParseURL(os.Getenv("REDIS_URL"))
	if err != nil {
		return
	}
	client := goredis.NewClient(options)
	defer client.Close()
	config := outbox.DefaultRedisConfig()
	config.Namespace = "orders"
	config.DurabilityMode = outbox.RedisDurabilityRequireAOF
	store, err := outbox.NewRedisStore(context.Background(), client, config)
	if err != nil {
		return
	}
	_, _ = store.Append(context.Background(), exampleRecord("standalone-redis"))
}

func exampleRecord(id outbox.ID) outbox.NewRecord {
	return outbox.NewRecord{ID: id, Destination: "orders", MessageType: "order.confirmed", Payload: []byte(`{"order_id":"42"}`), MaxAttempts: 8}
}

Index

Examples

Constants

View Source 
const (
	// OutboxModuleToken is the base token used by modules returned from
	// ForFeature. The selected storage feature is appended to this value.
	OutboxModuleToken gioc.Token = "OutboxModule"

	// OutboxServiceToken resolves the lifecycle-aware routing and worker hub.
	OutboxServiceToken gioc.Token = "OutboxService"
	// OutboxConfigToken resolves the service-level worker pool configuration.
	OutboxConfigToken gioc.Token = "OutboxConfig"
	// OutboxStoreToken resolves the selected adapter's internal aggregate store.
	OutboxStoreToken gioc.Token = "OutboxStore"
	// OutboxAppenderToken exposes only producer append capability.
	OutboxAppenderToken gioc.Token = "OutboxAppender"
	// OutboxDeliveryStoreToken exposes only fenced delivery state transitions.
	OutboxDeliveryStoreToken gioc.Token = "OutboxDeliveryStore"
	// OutboxReaderToken exposes only operational reads.
	OutboxReaderToken gioc.Token = "OutboxReader"
	// OutboxMaintenanceStoreToken exposes state-aware administrative operations.
	OutboxMaintenanceStoreToken gioc.Token = "OutboxMaintenanceStore"

	// OutboxPostgresConfigToken is consumed by PostgresStoreService.
	OutboxPostgresConfigToken gioc.Token = "OutboxPostgresConfig"
	// OutboxRedisConfigToken is consumed by RedisStoreService.
	OutboxRedisConfigToken gioc.Token = "OutboxRedisConfig"
)
View Source 
const MaxClaimDestinations = 16

MaxClaimDestinations is the portable bound for one storage claim. Service rotates larger routing tables through bounded claim windows.

View Source 
const MaximumTimestampUnixMicro int64 = 1<<53 - 1

MaximumTimestampUnixMicro is the largest portable timestamp. Microseconds through this value are exactly representable by the numeric type used by Redis Lua and are also supported by the other first-party adapters.

Variables

View Source 
var (
	// ErrNotFound means no record exists for the requested ID.
	ErrNotFound = errors.New("outbox: record not found")
	// ErrConflict means existing durable state contradicts an idempotent request.
	ErrConflict = errors.New("outbox: conflict")
	// ErrDuplicateID means an ID or idempotency key already exists in reject mode.
	ErrDuplicateID = errors.New("outbox: duplicate ID or idempotency key")
	// ErrLeaseLost means the supplied lease fence is absent, stale, or expired.
	ErrLeaseLost = errors.New("outbox: lease lost")
	// ErrInvalidTransition means the record state does not permit the operation.
	ErrInvalidTransition = errors.New("outbox: invalid state transition")
	// ErrUnsupportedCriteria means an adapter lacks indexes for a query shape.
	ErrUnsupportedCriteria = errors.New("outbox: unsupported query criteria")
	// ErrClosed means the adapter facade has been closed.
	ErrClosed = errors.New("outbox: store closed")
	// ErrInvalidArgument means validation rejected external input.
	ErrInvalidArgument = errors.New("outbox: invalid argument")
)
View Source 
var (
	// Postgres selects PostgresStoreService. Provide PostgresConfig at
	// OutboxPostgresConfigToken and import React's PostgreSQL module.
	Postgres = DefineStoreFeature("postgres", PostgresStoreServiceProvider)
	// Redis selects RedisStoreService. Provide RedisConfig at
	// OutboxRedisConfigToken and import React's Redis module.
	Redis = DefineStoreFeature("redis", RedisStoreServiceProvider)
)
View Source 
var ErrRedisUnsafeDurability = errors.New("outbox redis: unsafe durability configuration")

ErrRedisUnsafeDurability marks a required persistence or eviction policy failure.

View Source 
var PostgresStoreServiceInjections = []gioc.Token{
	reactpostgres.DataSourceToken,
	OutboxPostgresConfigToken,
	react.LoggerToken,
}

PostgresStoreServiceInjections lists the dependencies used by NewPostgresStoreService.

View Source 
var RedisStoreServiceInjections = []gioc.Token{
	reactredis.ServiceToken,
	react.ApplicationContextToken,
	OutboxRedisConfigToken,
	react.LoggerToken,
}

RedisStoreServiceInjections lists the dependencies used by NewRedisStoreService.

View Source 
var ServiceInjections = []gioc.Token{
	OutboxStoreToken,
	OutboxConfigToken,
	react.ApplicationContextServiceToken,
	react.LoggerToken,
}

ServiceInjections are the complete dependencies resolved by NewService.

Functions

func CanonicalTime 

func CanonicalTime(value time.Time) time.Time

CanonicalTime is the timestamp representation shared by all adapters. PostgreSQL and Redis both preserve microseconds, and Unix microseconds remain exactly representable by Redis Lua numbers for the supported date range.

func CursorForRecord 

func CursorForRecord(record Record, field SortField, direction SortDirection) (string, error)

CursorForRecord encodes a record's complete stable sort tuple.

func EncodeCursor 

func EncodeCursor(cursor Cursor) (string, error)

EncodeCursor creates the opaque base64url continuation value.

func ForFeature 

func ForFeature(storage StoreFeature, configs ...Config) *gioc.Module

ForFeature creates one outbox module backed by storage. The optional Config controls its single service-level worker pool; omitting it uses DefaultConfig. Sinks and routes are registered through Service.Register.

func ImmutableDigest 

func ImmutableDigest(record NewRecord) string

ImmutableDigest returns the canonical digest used for duplicate comparison.

func NormalizeQuery 

func NormalizeQuery(query Query, limits Limits) (Query, Cursor, error)

NormalizeQuery applies defaults and validates portable query bounds.

func PostgresStoreProvider 

func PostgresStoreProvider(token gioc.Token, config PostgresConfig) gioc.IProvider

PostgresStoreProvider creates a singleton store at token from React's PostgreSQL DataSourceToken. Use distinct tokens for independently named outboxes. For standard module-managed wiring prefer ProvidePostgresConfig with ForFeature; that path provides PostgresStoreService and the service-owned worker pool.

func PostgresStoreServiceProvider 

func PostgresStoreServiceProvider(token gioc.Token) gioc.IProvider

PostgresStoreServiceProvider provides a singleton PostgresStoreService at token.

func ProvidePostgresConfig 

func ProvidePostgresConfig(config PostgresConfig) gioc.IProvider

ProvidePostgresConfig returns a provider for OutboxPostgresConfigToken.

func ProvideRedisConfig 

func ProvideRedisConfig(config RedisConfig) gioc.IProvider

ProvideRedisConfig returns a provider for OutboxRedisConfigToken.

func RecordAfterCursor 

func RecordAfterCursor(record Record, cursor Cursor) bool

RecordAfterCursor reports whether a record belongs after a continuation tuple.

func RecordSortTime 

func RecordSortTime(record Record, field SortField) time.Time

RecordSortTime returns the selected stable sort component.

func RedisClusterSlot 

func RedisClusterSlot(key string) uint16

RedisClusterSlot returns the Redis Cluster slot for a key and is useful for deployment validation and third-party composition tests.

func RedisStoreProvider 

func RedisStoreProvider(token gioc.Token, config RedisConfig) gioc.IProvider

RedisStoreProvider creates a singleton store at token from React's Redis service and application context. Use distinct tokens and namespaces for named outboxes. For standard module-managed wiring prefer ProvideRedisConfig with ForFeature; that path provides RedisStoreService and the service-owned worker pool.

func RedisStoreServiceProvider 

func RedisStoreServiceProvider(token gioc.Token) gioc.IProvider

RedisStoreServiceProvider provides a singleton RedisStoreService at token.

func ServiceCapabilityProviders 

func ServiceCapabilityProviders(tokens Tokens) []gioc.IProvider

ServiceCapabilityProviders exposes the service through narrow named capabilities. Ordinary producers should inject IAppender; administrative consumers can request only the stronger interface they need.

func SortRecords 

func SortRecords(records []Record, field SortField, direction SortDirection)

SortRecords orders records by the complete time-plus-ID tuple.

func ValidateDestination 

func ValidateDestination(destination string, limits Limits) error

ValidateDestination checks a portable destination value.

func ValidateID 

func ValidateID(id ID, limits Limits) error

ValidateID checks the portable record ID syntax.

func ValidateLeaseDuration 

func ValidateLeaseDuration(field string, duration, maximum time.Duration) error

ValidateLeaseDuration checks a lease duration against the portable microsecond precision and an adapter's configured upper bound.

func ValidateLeaseOwner 

func ValidateLeaseOwner(owner string, limits Limits) error

ValidateLeaseOwner checks the bounded worker identity stored with a claim.

func ValidateLeaseToken 

func ValidateLeaseToken(token string, limits Limits) error

ValidateLeaseToken checks the bounded shape of a generated lease token.

func ValidateTimestamp 

func ValidateTimestamp(field string, value time.Time) error

ValidateTimestamp checks the portable UTC microsecond range. A zero value is invalid; APIs where zero means "use the default" must apply that default before calling this function.

Types

type AppendRequest 

type AppendRequest struct {
	Records       []NewRecord
	DuplicateMode DuplicateMode
}

AppendRequest selects duplicate semantics for one atomic bounded batch.

type Backlog 

type Backlog struct {
	Pending     int64
	Leased      int64
	Dead        int64
	OldestDueAt *time.Time
}

Backlog is an operational estimate, not a readiness decision by itself.

type ClaimRequest 

type ClaimRequest struct {
	Owner         string
	Limit         int
	LeaseDuration time.Duration
	Destinations  []string
	// RecoveryLimit bounds expired-lease work performed by a claim. Zero uses
	// the adapter's claim limit.
	RecoveryLimit int
}

ClaimRequest describes one bounded due-record lease operation.

type Config 

type Config struct {
	WorkerCount           int
	ClaimBatchSize        int
	LeaseDuration         time.Duration
	RenewalEnabled        bool
	LeaseRenewalThreshold time.Duration
	PollMinimumInterval   time.Duration
	PollMaximumInterval   time.Duration
	DeliveryTimeout       time.Duration
	ShutdownTimeout       time.Duration
	MaximumAttempts       int
	Retry                 ExponentialBackoffConfig
	Limits                Limits

	// These dependencies are optional application delivery policies. The service
	// supplies production defaults when they are nil.
	RetryPolicy     IRetryPolicy
	ErrorClassifier IErrorClassifier
	Owner           string
}

Config controls the service-level worker pool and delivery policy. Storage adapters keep their own namespace and durability configuration.

func DefaultConfig 

func DefaultConfig() Config

DefaultConfig returns conservative, internally valid service defaults.

func (Config) Validate 

func (config Config) Validate() error

Validate rejects unsafe or unbounded configuration relationships.

type Cursor 

type Cursor struct {
	Version   int           `json:"v"`
	Sort      SortField     `json:"s"`
	Direction SortDirection `json:"d"`
	Micros    int64         `json:"t"`
	ID        ID            `json:"i"`
}

Cursor is exported for third-party adapter implementations; applications should treat encoded cursor strings as opaque.

func DecodeCursor 

func DecodeCursor(value string) (Cursor, error)

DecodeCursor validates a continuation value for adapter implementations.

type DeliveryOutcome 

type DeliveryOutcome string

DeliveryOutcome classifies a transport attempt for settlement. 

const (
	OutcomeSuccess   DeliveryOutcome = "success"
	OutcomeRetryable DeliveryOutcome = "retryable"
	OutcomeTerminal  DeliveryOutcome = "terminal"
	OutcomeAmbiguous DeliveryOutcome = "ambiguous"
)

func (DeliveryOutcome) Valid 

func (outcome DeliveryOutcome) Valid() bool

Valid reports whether outcome is one of the bounded built-in labels.

type DeliveryResult 

type DeliveryResult struct{}

DeliveryResult is intentionally transport-neutral. Delivery is considered successful only after the sink's own reliable-delivery condition is met.

type DestinationsConfig 

type DestinationsConfig struct {
	Destinations []string
	Concurrency  int
}

DestinationsConfig binds one sink to a unique set of destinations. Concurrency limits all deliveries through that sink registration; zero uses the service worker count.

type DuplicateMode 

type DuplicateMode int

DuplicateMode controls append behavior when an ID or idempotency key exists. 

const (
	RejectDuplicate DuplicateMode = iota
	AcceptIdentical
)

func (DuplicateMode) Valid 

func (m DuplicateMode) Valid() bool

Valid reports whether m is a defined duplicate mode.

type ErrorClassifierFunc 

type ErrorClassifierFunc func(error) DeliveryOutcome

ErrorClassifierFunc adapts a function to IErrorClassifier.

func (ErrorClassifierFunc) Classify 

func (f ErrorClassifierFunc) Classify(err error) DeliveryOutcome

type ExponentialBackoff 

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

ExponentialBackoff implements bounded exponential delay with optional jitter.

func NewExponentialBackoff 

func NewExponentialBackoff(config ExponentialBackoffConfig) (*ExponentialBackoff, error)

NewExponentialBackoff validates and constructs a retry policy.

func (*ExponentialBackoff) Next 

func (policy *ExponentialBackoff) Next(attempt int, _ error) (time.Duration, bool)

Next returns the bounded jittered delay for the completed attempt number.

type ExponentialBackoffConfig 

type ExponentialBackoffConfig struct {
	Minimum     time.Duration
	Maximum     time.Duration
	Multiplier  float64
	Jitter      float64
	MaxAttempts int
}

ExponentialBackoffConfig configures bounded exponential retry delays.

type Failure 

type Failure struct {
	Code    string
	Message string
}

Failure is bounded, non-sensitive error information persisted operationally.

func BoundFailure 

func BoundFailure(failure Failure, limits Limits) Failure

BoundFailure truncates persisted failure fields without producing invalid UTF-8.

type FieldError 

type FieldError struct {
	Field   string
	Message string
}

FieldError identifies an invalid public input while remaining inspectable as ErrInvalidArgument through errors.Is.

func (*FieldError) Error 

func (e *FieldError) Error() string

func (*FieldError) Unwrap 

func (e *FieldError) Unwrap() error

type Health 

type Health struct {
	Ready            bool
	StorageAvailable bool
	DurabilitySafe   bool
	Message          string
	Backlog          Backlog
}

Health separates storage and durability readiness from backlog signals.

type IAppender 

type IAppender interface {
	Append(ctx context.Context, records ...NewRecord) ([]Record, error)
}

IAppender atomically appends a bounded batch with its configured duplicate mode.

type IBacklogReader 

type IBacklogReader interface {
	Backlog(ctx context.Context) (Backlog, error)
}

IBacklogReader supplies bounded-cardinality backlog observations.

type IBatchAppender 

type IBatchAppender interface {
	AppendBatch(ctx context.Context, request AppendRequest) ([]Record, error)
}

IBatchAppender is implemented by first-party adapters in addition to the convenient IAppender API.

type ID 

type ID string

ID is a stable application-visible record identifier.

type IDeliveryStore 

type IDeliveryStore interface {
	Claim(ctx context.Context, request ClaimRequest) ([]Record, error)
	Renew(ctx context.Context, lease LeaseRef, until time.Time) error
	Acknowledge(ctx context.Context, lease LeaseRef, result DeliveryResult) error
	Retry(ctx context.Context, lease LeaseRef, retry RetryRequest) error
	DeadLetter(ctx context.Context, lease LeaseRef, failure Failure) error
	Release(ctx context.Context, lease LeaseRef, availableAt time.Time) error
}

IDeliveryStore exposes only fenced delivery state-machine operations.

type IErrorClassifier 

type IErrorClassifier interface {
	Classify(err error) DeliveryOutcome
}

IErrorClassifier maps sink errors to transport-neutral outcomes.

func DefaultErrorClassifier 

func DefaultErrorClassifier() IErrorClassifier

DefaultErrorClassifier treats TerminalError as terminal and every other non-nil error as retryable.

type IHealthChecker 

type IHealthChecker interface {
	Health(ctx context.Context) Health
}

IHealthChecker reports readiness separately from backlog pressure. A large backlog is observable but does not by itself make Ready false.

type IMaintenanceStore 

type IMaintenanceStore interface {
	Cancel(ctx context.Context, id ID, reason string) error
	Reschedule(ctx context.Context, id ID, availableAt time.Time) error
	Requeue(ctx context.Context, id ID, options RequeueOptions) error
	Purge(ctx context.Context, request PurgeRequest) (int, error)
}

IMaintenanceStore contains separately injectable administrative operations.

type IPostgresDB 

type IPostgresDB interface {
	Begin(ctx context.Context) (pgx.Tx, error)
	Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}

IPostgresDB is the narrow pgx pool surface owned by the application.

type IPostgresRowScanner 

type IPostgresRowScanner interface{ Scan(dest ...any) error }

IPostgresRowScanner is the shared row and rows scanning surface.

type IReader 

type IReader interface {
	Get(ctx context.Context, id ID) (Record, error)
	Find(ctx context.Context, query Query) (Page, error)
}

IReader retrieves records without granting mutation capability.

type IRetryPolicy 

type IRetryPolicy interface {
	Next(attempt int, failure error) (delay time.Duration, retry bool)
}

IRetryPolicy decides whether and when a failed attempt becomes due again.

func DefaultRetryPolicy 

func DefaultRetryPolicy(maxAttempts int) IRetryPolicy

DefaultRetryPolicy returns the library's general-purpose bounded policy.

type ISink 

type ISink interface {
	Deliver(ctx context.Context, record Record) error
}

ISink delivers one copied record and returns nil only after its reliable transport acceptance condition has been met.

type IStore 

type IStore interface {
	IAppender
	IBatchAppender
	IDeliveryStore
	IReader
	IMaintenanceStore
}

IStore is the aggregate convenience contract implemented by first-party adapters.

type LeaseRef 

type LeaseRef struct {
	ID      ID
	Owner   string
	Token   string
	Version uint64
}

LeaseRef is the complete fence needed to settle one delivery attempt.

type Limits 

type Limits struct {
	MaxIDBytes             int
	MaxDestinationBytes    int
	MaxMessageTypeBytes    int
	MaxAggregateTypeBytes  int
	MaxAggregateIDBytes    int
	MaxOrderingKeyBytes    int
	MaxIdempotencyKeyBytes int
	MaxPayloadBytes        int
	MaxHeaders             int
	MaxHeaderKeyBytes      int
	MaxHeaderValueBytes    int
	MaxHeaderBytes         int
	MaxErrorCodeBytes      int
	MaxErrorMessageBytes   int
	MaxLeaseOwnerBytes     int
	MaxLeaseTokenBytes     int
	MaxBatchSize           int
	MaxClaimBatchSize      int
	MaxPageSize            int
	MaxQueryIDs            int
	MaxQueryValues         int
	MaxPurgeSize           int
	MaxAttempts            int
	MaxWorkerCount         int
	MaxDestinationWorkers  int
}

Limits bounds data and work accepted at public API boundaries.

func DefaultLimits 

func DefaultLimits() Limits

DefaultLimits returns the portable first-party resource bounds.

func (Limits) Normalized 

func (l Limits) Normalized() Limits

Normalized returns a copy with safe defaults applied to zero-valued fields.

type NewRecord 

type NewRecord struct {
	ID             ID
	Destination    string
	MessageType    string
	AggregateType  string
	AggregateID    string
	OrderingKey    string
	IdempotencyKey string
	Headers        map[string]string
	Payload        []byte
	AvailableAt    time.Time
	MaxAttempts    int
}

NewRecord contains immutable message content and its initial delivery policy.

func (NewRecord) Clone 

func (r NewRecord) Clone() NewRecord

Clone returns a deep copy of an append request.

type OperationError 

type OperationError struct {
	Operation string
	Err       error
}

OperationError adds stable operation context without hiding a sentinel returned by an adapter.

func (*OperationError) Error 

func (e *OperationError) Error() string

func (*OperationError) Unwrap 

func (e *OperationError) Unwrap() error

type Page 

type Page struct {
	Records    []Record
	NextCursor string
}

Page contains copied records and an opaque continuation cursor.

type PostgresConfig 

type PostgresConfig struct {
	Namespace          string
	Schema             string
	Table              string
	DuplicateMode      DuplicateMode
	DefaultMaxAttempts int
	MaxLeaseDuration   time.Duration
	Limits             Limits
}

PostgresConfig selects namespace, validated identifiers, and limits.

func DefaultPostgresConfig 

func DefaultPostgresConfig() PostgresConfig

DefaultPostgresConfig returns adapter defaults for one shared table.

type PostgresMigration 

type PostgresMigration struct {
	Version int
	Name    string
	SQL     string
}

PostgresMigration is one ordered adapter-owned schema change.

type PostgresStore 

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

PostgresStore implements the portable contracts with an application-owned pgx pool.

func NewPostgresStore 

func NewPostgresStore(db IPostgresDB, config PostgresConfig) (*PostgresStore, error)

NewPostgresStore validates and constructs a PostgreSQL adapter without running migrations.

func NewPostgresStoreFromDataSource 

func NewPostgresStoreFromDataSource(dataSource *pgxext.DataSource, config PostgresConfig) (*PostgresStore, error)

NewPostgresStoreFromDataSource constructs a store over React's application-owned PostgreSQL pool. Closing the returned facade never closes the data source.

func (*PostgresStore) Acknowledge 

func (store *PostgresStore) Acknowledge(ctx context.Context, lease LeaseRef, _ DeliveryResult) error

func (*PostgresStore) Append 

func (store *PostgresStore) Append(ctx context.Context, records ...NewRecord) ([]Record, error)

func (*PostgresStore) AppendBatch 

func (store *PostgresStore) AppendBatch(ctx context.Context, request AppendRequest) ([]Record, error)

func (*PostgresStore) Backlog 

func (store *PostgresStore) Backlog(ctx context.Context) (Backlog, error)

func (*PostgresStore) Bind 

func (store *PostgresStore) Bind(tx pgx.Tx) *PostgresTxAppender

Bind creates an appender on the caller's pgx transaction. Append uses a nested pgx transaction (a savepoint) so a rejected batch leaves no partial rows. It never commits or rolls back the caller's outer transaction.

func (*PostgresStore) Cancel 

func (store *PostgresStore) Cancel(ctx context.Context, id ID, reason string) error

func (*PostgresStore) Claim 

func (store *PostgresStore) Claim(ctx context.Context, request ClaimRequest) ([]Record, error)

func (*PostgresStore) Close 

func (store *PostgresStore) Close() error

Close only closes this adapter facade. The application-owned PostgreSQL pool remains open.

func (*PostgresStore) DeadLetter 

func (store *PostgresStore) DeadLetter(ctx context.Context, lease LeaseRef, failure Failure) error

func (*PostgresStore) Find 

func (store *PostgresStore) Find(ctx context.Context, query Query) (Page, error)

func (*PostgresStore) Get 

func (store *PostgresStore) Get(ctx context.Context, id ID) (Record, error)

func (*PostgresStore) Health 

func (store *PostgresStore) Health(ctx context.Context) Health

func (*PostgresStore) Migrate 

func (store *PostgresStore) Migrate(ctx context.Context) error

Migrate applies the adapter-owned, versioned migrations. Applications may instead consume Migrations from their normal migration system.

func (*PostgresStore) Migrations 

func (store *PostgresStore) Migrations() []PostgresMigration

Migrations returns rendered, safely quoted adapter migrations in version order.

func (*PostgresStore) Purge 

func (store *PostgresStore) Purge(ctx context.Context, request PurgeRequest) (int, error)

func (*PostgresStore) Release 

func (store *PostgresStore) Release(ctx context.Context, lease LeaseRef, availableAt time.Time) error

func (*PostgresStore) Renew 

func (store *PostgresStore) Renew(ctx context.Context, lease LeaseRef, until time.Time) error

func (*PostgresStore) Requeue 

func (store *PostgresStore) Requeue(ctx context.Context, id ID, options RequeueOptions) error

func (*PostgresStore) Reschedule 

func (store *PostgresStore) Reschedule(ctx context.Context, id ID, availableAt time.Time) error

func (*PostgresStore) Retry 

func (store *PostgresStore) Retry(ctx context.Context, lease LeaseRef, retry RetryRequest) error

type PostgresStoreService 

type PostgresStoreService struct {
	*PostgresStore
	Logger react.ILogger
}

PostgresStoreService is the module-managed PostgreSQL store facade. The application retains ownership of the injected data source and its pool.

func NewPostgresStoreService 

func NewPostgresStoreService(injections gioc.Injections) (*PostgresStoreService, error)

NewPostgresStoreService resolves React's PostgreSQL data source and PostgresConfig from OutboxPostgresConfigToken.

func (*PostgresStoreService) Close 

func (service *PostgresStoreService) Close() error

Close closes the adapter facade and logs an unexpected failure.

type PostgresTxAppender 

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

PostgresTxAppender appends through a caller-owned pgx transaction.

func (*PostgresTxAppender) Append 

func (appender *PostgresTxAppender) Append(ctx context.Context, records ...NewRecord) ([]Record, error)

func (*PostgresTxAppender) AppendBatch 

func (appender *PostgresTxAppender) AppendBatch(ctx context.Context, request AppendRequest) ([]Record, error)

type PurgeRequest 

type PurgeRequest struct {
	States []State
	Before time.Time
	Limit  int
}

PurgeRequest deletes a bounded set of terminal records before a cutoff.

func NormalizePurgeRequest 

func NormalizePurgeRequest(request PurgeRequest, limits Limits) (PurgeRequest, error)

NormalizePurgeRequest validates and canonicalizes bounded terminal cleanup work for adapter implementations.

type Query 

type Query struct {
	IDs            []ID
	States         []State
	Destinations   []string
	MessageTypes   []string
	AggregateType  string
	AggregateID    string
	OrderingKey    string
	IdempotencyKey string
	CreatedAt      TimeRange
	AvailableAt    TimeRange
	Sort           SortField
	Direction      SortDirection
	Limit          int
	Cursor         string
}

Query is the deliberately limited, storage-independent operational query model.

type Record 

type Record struct {
	ID             ID
	Destination    string
	MessageType    string
	AggregateType  string
	AggregateID    string
	OrderingKey    string
	IdempotencyKey string
	Headers        map[string]string
	Payload        []byte

	// ContentDigest is a lowercase SHA-256 digest of immutable message content.
	// It deliberately excludes ID and mutable delivery scheduling fields so an
	// idempotency key can identify the same message submitted with a newly
	// generated ID or after the existing record has been rescheduled.
	ContentDigest string

	State       State
	Attempts    int
	MaxAttempts int
	AvailableAt time.Time

	LeaseOwner string
	LeaseToken string
	LeaseUntil *time.Time

	LastErrorCode    string
	LastErrorMessage string

	CreatedAt   time.Time
	UpdatedAt   time.Time
	DeliveredAt *time.Time
	DeadAt      *time.Time
	CancelledAt *time.Time
	Version     uint64
}

Record is the copied public snapshot of one persisted outbox record.

func PrepareRecord 

func PrepareRecord(input NewRecord, now time.Time, defaultMaxAttempts int, limits Limits) (Record, error)

PrepareRecord validates, defaults, copies, and digests an append request. Adapters call it before making any storage mutation.

func (Record) Clone 

func (r Record) Clone() Record

Clone returns a deep copy suitable for crossing an API boundary.

func (Record) LeaseRef 

func (r Record) LeaseRef() LeaseRef

LeaseRef returns the complete fence from a claimed record snapshot.

type RedisCompositionRequest 

type RedisCompositionRequest struct {
	Append          AppendRequest
	DomainKeys      []string
	DomainArguments []any
	ValidateLua     string
	ApplyLua        string
}

RedisCompositionRequest atomically combines a Redis-backed domain mutation with an outbox append in one server-side script. DomainKeys are appended after the 15 outbox keys and DomainArguments after the two append arguments. Trusted Lua snippets address them with DOMAIN_KEY_OFFSET and DOMAIN_ARG_OFFSET.

ValidateLua must perform every domain precondition and type check without writing. ApplyLua runs only after both domain and outbox duplicate validation have succeeded. It must use only commands whose runtime types were validated; Redis does not roll back writes when a later command raises a runtime error. Both snippets are deployment code, never untrusted request input.

type RedisConfig 

type RedisConfig struct {
	Namespace          string
	DuplicateMode      DuplicateMode
	DefaultMaxAttempts int
	MaxLeaseDuration   time.Duration
	// MaxAppendEncodedBytes bounds the complete JSON batch evaluated by Lua.
	MaxAppendEncodedBytes int
	// MaxClaimResponseBytes bounds record data returned by one claim script.
	MaxClaimResponseBytes int
	Limits                Limits
	DurabilityMode        RedisDurabilityMode
	RequireNoEviction     bool
	// AllowUnsafeEviction must be set deliberately to permit construction when
	// maxmemory-policy is not required to be noeviction.
	AllowUnsafeEviction bool
}

RedisConfig selects the same-slot namespace, resource limits, and required durability and eviction posture.

func DefaultRedisConfig 

func DefaultRedisConfig() RedisConfig

DefaultRedisConfig requires noeviction and warns when AOF is unavailable.

type RedisDurabilityMode 

type RedisDurabilityMode string

RedisDurabilityMode controls startup behavior when persistence cannot meet policy. 

const (
	RedisDurabilityUnchecked  RedisDurabilityMode = "unchecked"
	RedisDurabilityWarn       RedisDurabilityMode = "warn"
	RedisDurabilityRequireAOF RedisDurabilityMode = "require_aof"
)

type RedisDurabilityReport 

type RedisDurabilityReport struct {
	Checked        bool
	AOFEnabled     bool
	AOFFsync       string
	AOFLastWriteOK bool
	EvictionPolicy string
	Role           string
	Warnings       []string
}

RedisDurabilityReport is a copied snapshot of inspected Redis settings and status.

type RedisKeys 

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

RedisKeys centralizes every Redis key. A namespace is also the Redis Cluster hash tag, so every atomic transition remains in exactly one hash slot.

func NewRedisKeys 

func NewRedisKeys(namespace string) (RedisKeys, error)

func (RedisKeys) Cancelled 

func (keys RedisKeys) Cancelled() string

func (RedisKeys) Dead 

func (keys RedisKeys) Dead() string

func (RedisKeys) Delivered 

func (keys RedisKeys) Delivered() string

func (RedisKeys) Idempotency 

func (keys RedisKeys) Idempotency() string

func (RedisKeys) Leased 

func (keys RedisKeys) Leased() string

func (RedisKeys) Namespace 

func (keys RedisKeys) Namespace() string

func (RedisKeys) Pending 

func (keys RedisKeys) Pending() string

func (RedisKeys) PendingDestinations 

func (keys RedisKeys) PendingDestinations() string

func (RedisKeys) QueryAll 

func (keys RedisKeys) QueryAll() string

func (RedisKeys) QueryCancelled 

func (keys RedisKeys) QueryCancelled() string

func (RedisKeys) QueryDead 

func (keys RedisKeys) QueryDead() string

func (RedisKeys) QueryDelivered 

func (keys RedisKeys) QueryDelivered() string

func (RedisKeys) QueryDestinations 

func (keys RedisKeys) QueryDestinations() string

func (RedisKeys) QueryLeased 

func (keys RedisKeys) QueryLeased() string

func (RedisKeys) QueryPending 

func (keys RedisKeys) QueryPending() string

func (RedisKeys) QueryState 

func (keys RedisKeys) QueryState(state State) (string, error)

func (RedisKeys) RecordKey 

func (keys RedisKeys) RecordKey(id ID) string

RecordKey exposes the collision-safe conventional per-record key for domain composition tooling. The first-party representation keeps record blobs in a shared hash so Lua scripts access only explicitly declared keys.

func (RedisKeys) Records 

func (keys RedisKeys) Records() string

func (RedisKeys) ScriptKeys 

func (keys RedisKeys) ScriptKeys() []string

func (RedisKeys) State 

func (keys RedisKeys) State(state State) (string, error)

type RedisStore 

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

RedisStore implements the portable contracts with an application-owned client.

func NewRedisStore 

func NewRedisStore(ctx context.Context, client goredis.UniversalClient, config RedisConfig) (*RedisStore, error)

NewRedisStore validates dependencies, pings Redis, and performs startup durability checks.

func NewRedisStoreFromService 

func NewRedisStoreFromService(ctx context.Context, service *reactredis.Service, config RedisConfig) (*RedisStore, error)

NewRedisStoreFromService constructs a store over React's application-owned Redis client. Closing the returned facade never closes the service client.

func (*RedisStore) Acknowledge 

func (store *RedisStore) Acknowledge(ctx context.Context, lease LeaseRef, _ DeliveryResult) error

func (*RedisStore) Append 

func (store *RedisStore) Append(ctx context.Context, records ...NewRecord) ([]Record, error)

func (*RedisStore) AppendBatch 

func (store *RedisStore) AppendBatch(ctx context.Context, request AppendRequest) ([]Record, error)

func (*RedisStore) Backlog 

func (store *RedisStore) Backlog(ctx context.Context) (Backlog, error)

func (*RedisStore) Cancel 

func (store *RedisStore) Cancel(ctx context.Context, id ID, reason string) error

func (*RedisStore) CheckDurability 

func (store *RedisStore) CheckDurability(ctx context.Context) (report RedisDurabilityReport, resultErr error)

CheckDurability refreshes persistence, eviction, and replication observations.

func (*RedisStore) Claim 

func (store *RedisStore) Claim(ctx context.Context, request ClaimRequest) ([]Record, error)

func (*RedisStore) Close 

func (store *RedisStore) Close() error

Close marks the facade closed without closing the application-owned client.

func (*RedisStore) Compose 

func (store *RedisStore) Compose(ctx context.Context, request RedisCompositionRequest) ([]Record, error)

Compose performs one atomic Redis-backed domain mutation and append. It does not make mutations in PostgreSQL or another Redis hash slot atomic.

func (*RedisStore) DeadLetter 

func (store *RedisStore) DeadLetter(ctx context.Context, lease LeaseRef, failure Failure) error

func (*RedisStore) Find 

func (store *RedisStore) Find(ctx context.Context, query Query) (Page, error)

func (*RedisStore) Get 

func (store *RedisStore) Get(ctx context.Context, id ID) (Record, error)

func (*RedisStore) Health 

func (store *RedisStore) Health(ctx context.Context) Health

func (*RedisStore) Keys 

func (store *RedisStore) Keys() RedisKeys

func (*RedisStore) LastDurabilityReport 

func (store *RedisStore) LastDurabilityReport() RedisDurabilityReport

LastDurabilityReport returns a copy of the most recent startup or health durability inspection.

func (*RedisStore) Purge 

func (store *RedisStore) Purge(ctx context.Context, request PurgeRequest) (int, error)

func (*RedisStore) Release 

func (store *RedisStore) Release(ctx context.Context, lease LeaseRef, availableAt time.Time) error

func (*RedisStore) Renew 

func (store *RedisStore) Renew(ctx context.Context, lease LeaseRef, until time.Time) error

func (*RedisStore) Requeue 

func (store *RedisStore) Requeue(ctx context.Context, id ID, options RequeueOptions) error

func (*RedisStore) Reschedule 

func (store *RedisStore) Reschedule(ctx context.Context, id ID, availableAt time.Time) error

func (*RedisStore) Retry 

func (store *RedisStore) Retry(ctx context.Context, lease LeaseRef, retry RetryRequest) error

type RedisStoreService 

type RedisStoreService struct {
	*RedisStore
	Logger react.ILogger
}

RedisStoreService is the module-managed Redis store facade. The application retains ownership of the injected Redis service and client.

func NewRedisStoreService 

func NewRedisStoreService(injections gioc.Injections) (*RedisStoreService, error)

NewRedisStoreService resolves React's Redis service, the application context, and RedisConfig from OutboxRedisConfigToken.

func (*RedisStoreService) Close 

func (service *RedisStoreService) Close() error

Close closes the adapter facade and logs an unexpected failure.

type RequeueOptions 

type RequeueOptions struct {
	AvailableAt   time.Time
	ResetAttempts bool
	// MaxAttempts optionally replaces the existing limit. Zero preserves it.
	// When preserved attempts have reached that limit, requeue requires either
	// ResetAttempts or a larger valid MaxAttempts value.
	MaxAttempts int
}

RequeueOptions controls replay of a dead record.

type RetryRequest 

type RetryRequest struct {
	AvailableAt time.Time
	Failure     Failure
}

RetryRequest persists one computed retry schedule and failure.

type Service 

type Service struct {
	IStore

	ApplicationService *react.ApplicationService
	Logger             react.ILogger
	// contains filtered or unexported fields
}

Service is the outbox boundary exposed to applications. It provides the storage capabilities, owns all destination routing, and runs one bounded worker pool for every registered sink.

func NewService 

func NewService(injections gioc.Injections) (*Service, error)

NewService resolves the store, delivery capability, worker configuration, application lifecycle, and logger. It starts the service-owned worker pool; consumers then declare routes with Register.

func (*Service) Append 

func (service *Service) Append(ctx context.Context, records ...NewRecord) ([]Record, error)

Append delegates to the selected store and wakes the worker pool after a successful append. Transaction-bound adapter appends remain visible through the bounded polling interval after their outer transaction commits.

func (*Service) AppendBatch 

func (service *Service) AppendBatch(ctx context.Context, request AppendRequest) ([]Record, error)

AppendBatch delegates the atomic batch and wakes the worker pool after a successful append.

func (*Service) Destinations 

func (service *Service) Destinations() []string

Destinations returns a stable copy of the registered routing keys.

func (*Service) Done 

func (service *Service) Done() <-chan struct{}

Done closes after the worker pool and bounded lease cleanup stop.

func (*Service) Register 

func (service *Service) Register(sink ISink, config DestinationsConfig) error

Register atomically routes every configured destination to sink. A destination can belong to only one sink, which makes routing deterministic. Registration is safe while the worker pool is running.

func (*Service) Shutdown 

func (service *Service) Shutdown(ctx context.Context) error

Shutdown stops claims, drains in-flight work within ctx, and releases any still-current leases after forced cancellation. Calls are idempotent.

func (*Service) String 

func (service *Service) String() string

String returns a non-sensitive operational identity.

type SinkFunc 

type SinkFunc func(ctx context.Context, record Record) error

SinkFunc adapts a delivery function to ISink.

func (SinkFunc) Deliver 

func (sink SinkFunc) Deliver(ctx context.Context, record Record) error

type SortDirection 

type SortDirection string

SortDirection selects ascending or descending keyset traversal. 

const (
	SortAscending  SortDirection = "asc"
	SortDescending SortDirection = "desc"
)

type SortField 

type SortField string

SortField selects a portable stable time ordering. 

const (
	SortCreatedAt   SortField = "created_at"
	SortAvailableAt SortField = "available_at"
)

type State 

type State string

State is a durable delivery state. 

const (
	StatePending   State = "pending"
	StateLeased    State = "leased"
	StateDelivered State = "delivered"
	StateDead      State = "dead"
	StateCancelled State = "cancelled"
)

func (State) Terminal 

func (s State) Terminal() bool

Terminal reports whether records in s may be purged by retention policy.

func (State) Valid 

func (s State) Valid() bool

Valid reports whether s is a defined durable state.

type StoreFeature 

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

StoreFeature is an immutable storage adapter description selected by ForFeature.

func DefineStoreFeature 

func DefineStoreFeature(name string, provider StoreProviderFactory) StoreFeature

DefineStoreFeature creates a feature descriptor without global mutable registration or init side effects.

func (StoreFeature) Name 

func (feature StoreFeature) Name() string

Name returns the adapter's stable feature name.

type StoreProviderFactory 

type StoreProviderFactory func(token gioc.Token) gioc.IProvider

StoreProviderFactory creates a fresh provider for an adapter store service at the requested aggregate-store token. Applications normally select one of the first-party feature values.

type TerminalError 

type TerminalError struct{ Err error }

TerminalError lets a sink explicitly identify an error that retrying cannot fix. All unclassified and ambiguous errors are retried for at-least-once delivery.

func (*TerminalError) Error 

func (e *TerminalError) Error() string

func (*TerminalError) Unwrap 

func (e *TerminalError) Unwrap() error

type TimeRange 

type TimeRange struct {
	From *time.Time
	To   *time.Time
}

TimeRange is inclusive at both non-nil bounds.

type Tokens 

type Tokens struct {
	Service          gioc.Token
	Store            gioc.Token
	Appender         gioc.Token
	DeliveryStore    gioc.Token
	Reader           gioc.Token
	MaintenanceStore gioc.Token
}

Tokens gives every named outbox distinct dependency-injection identities. No package-level registry is used.

func ModuleTokens 

func ModuleTokens() Tokens

ModuleTokens returns the fixed service and capability tokens exposed by ForFeature.

func NewTokens 

func NewTokens(name string) (Tokens, error)

NewTokens creates collision-free dependency-injection identities for one named outbox.

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.