outbox

README

go-outbox

go-outbox implements the outbox pattern for Go applications. It stores messages in a database table as part of the same transaction as your business logic and dispatches them asynchronously to your message broker.

Features

• Generic API that works with any message type
• Built in PostgreSQL store with notification triggers
• Message broker integrations for Kafka (franz-go) and NATS (including JetStream)
• Pluggable dead letter queue interface
• Configurable concurrency and retry behaviour
• Uses encoding/gob for message serialization by default

Installation

go get github.com/nrfta/go-outbox

Quick start

The snippet below shows how to configure an outbox using PostgreSQL and NATS JetStream:

package main

import (
    "database/sql"
    "log/slog"

    "github.com/nats-io/nats.go"
    "github.com/nats-io/nats.go/jetstream"

    "github.com/nrfta/go-outbox"
    mbNats "github.com/nrfta/go-outbox/mb/nats"
    outboxPg "github.com/nrfta/go-outbox/store/pg"
)

func NewNatsOutbox(db *sql.DB, connStr string) (outbox.Outbox[*nats.Msg], error) {
    nc, err := nats.Connect(nats.DefaultURL)
    if err != nil {
        return nil, err
    }

    js, err := jetstream.New(nc)
    if err != nil {
        return nil, err
    }

    broker, err := mbNats.NewJetstream(js)
    if err != nil {
        return nil, err
    }

    store, err := outboxPg.NewStore(db, connStr)
    if err != nil {
        return nil, err
    }

    ob := outbox.New[*nats.Msg](
        store,
        broker,
        logDeadLetterQueue{},
        outbox.WithLogger[*nats.Msg](slog.Default()),
        outbox.WithMaxRetries[*nats.Msg](10),
        outbox.WithNumberOfRoutines[*nats.Msg](5),
    )
    return ob, nil
}

Sending messages

Messages are queued by storing them in the outbox table inside the same database transaction as your business logic. The outbox library then dispatches them asynchronously in the background. Use SendTx to add a message to the queue:

ctx := context.Background()
tx, _ := db.Begin()

msg := &nats.Msg{Subject: "hello", Data: []byte("world")}
if err := ob.SendTx(ctx, tx, msg); err != nil {
    tx.Rollback()
    return err
}
return tx.Commit()

Options

The outbox.New constructor accepts several optional configuration functions:

• WithNumberOfRoutines(n int) – limits the number of goroutines used to dispatch messages
• WithMaxRetries(n int) – maximum number of attempts before sending a message to the dead letter queue
• WithLogger(*slog.Logger) – provide a custom logger

You can provide your own store or message broker by implementing the Store and MessageBroker interfaces found in outbox.go.

Testing

The testing package provides synchronous implementations of outbox interfaces for deterministic, race-condition-free testing. Instead of processing messages asynchronously with background workers, these implementations deliver messages synchronously when they're created.

Benefits

• Deterministic execution – No async race conditions or timing issues
• Immediate processing – Events handled synchronously, no waiting required
• Full stack traces – See the complete call chain in test failures
• No infrastructure – Tests run without NATS or background workers
• Simpler debugging – Step through event handling in debugger

Usage Example

import (
    "github.com/nats-io/nats.go/jetstream"
    "github.com/nrfta/go-outbox"
    "github.com/nrfta/go-outbox/testing"
)

// Setup synchronous testing infrastructure
syncJS := testing.NewSyncJetStream()
syncBroker := testing.NewSyncBroker(syncJS)
syncStore := testing.NewSyncStore(syncBroker)

// Create outbox with sync components
ob, err := outbox.New(syncStore, syncBroker)
if err != nil {
    panic(err)
}

// Override JetStream in your DI container
// Events will now be processed synchronously

The sync implementations work together:

1. SyncStore decodes messages and sends them to the broker immediately (no DB storage)
2. SyncBroker routes messages to SyncJetStream
3. SyncJetStream delivers messages directly to registered consumers

See testing/doc.go for detailed documentation.

Running tests

Integration tests require a running PostgreSQL instance. You can run all tests with:

go test ./...

The CI workflow spins up Postgres automatically.

Documentation

Index

• Variables
• func WithLogger[T any](logger *slog.Logger) option[T]
• func WithMaxRetries[T any](n int) option[T]
• func WithNumberOfRoutines[T any](n int) option[T]
• type DeadLetterQueue
• type EncodeDecoder
• type MessageBroker
• type Outbox
  • func New[T any](s Store, mb MessageBroker[T], dlq DeadLetterQueue, opts ...option[T]) Outbox[T]
• type Record
• type Store

Variables

var (
	ErrRecordNotFound = errors.New("record not found")
)

Functions

func WithLogger

func WithLogger[T any](logger *slog.Logger) option[T]

func WithMaxRetries

func WithMaxRetries[T any](n int) option[T]

func WithNumberOfRoutines

func WithNumberOfRoutines[T any](n int) option[T]

Types

type DeadLetterQueue

type DeadLetterQueue interface {
	Send(context.Context, any, error) error
}

type EncodeDecoder

type EncodeDecoder[T any] interface {
	Encode(data T) ([]byte, error)
	Decode(raw []byte) (T, error)
}

type MessageBroker

type MessageBroker[T any] interface {
	Send(context.Context, T) error
}

type Outbox

type Outbox[T any] interface {
	// SendTx stores the provided message within the provided Tx
	SendTx(ctx context.Context, tx *sql.Tx, msg T) error
}

Outbox implements the outbox pattern.

func New

func New[T any](s Store, mb MessageBroker[T], dlq DeadLetterQueue, opts ...option[T]) Outbox[T]

type Record

type Record struct {
	ID               xid.ID
	Message          []byte
	CreatedAt        time.Time
	NumberOfAttempts int
	LastAttemptAt    *time.Time
}

type Store

type Store interface {
	// CreateRecordTx stores the Record within the provided Tx.
	CreateRecordTx(context.Context, *sql.Tx, Record) (*Record, error)

	// Listen creates a channel of record IDs to process.
	Listen() <-chan xid.ID

	// GetWithLock finds a Record in the Store by the provided ID. This method
	// must lock the returned Record while being processed to ensure
	// concurrent integrity.
	GetWithLock(context.Context, xid.ID) (*Record, error)

	// Delete removes a record from the Store by the provided ID.
	Delete(context.Context, xid.ID) error

	// ProcessTx performs the function provided inside a transaction.
	ProcessTx(context.Context, func(Store) bool) error

	// Update sends the updated Record to the store
	Update(context.Context, *Record) error
}