xoutbox

xoutbox

A production-grade Transactional Outbox for Go. Write events in the same database transaction as your business data, and xoutbox delivers them to Kafka, NATS JetStream, RabbitMQ or Redis with no lost messages, crash recovery and per-key ordering. You don't need distributed transactions.

┌──────────── your service ────────────┐          ┌──────────────┐
│ BEGIN                                │          │    Kafka     │
│   INSERT INTO orders ...             │  Worker  │    NATS      │
│   INSERT INTO outbox ...  ◄──────────┼──claim──►│  RabbitMQ    │
│ COMMIT                               │  publish │    Redis     │
└──────────────────────────────────────┘          └──────────────┘

Features

Install

go get github.com/Ali127Dev/xoutbox

Quick start (Postgres → Kafka)

db, _ := sql.Open("pgx", os.Getenv("DATABASE_URL"))

store := postgres.New(db)                // postgres.WithTable("events_outbox"), postgres.WithOrdering(), ...
if err := store.Migrate(ctx); err != nil { // or copy postgres/schema.sql into your migrations
    log.Fatal(err)
}

pub, err := kafka.New(kafka.Config{Brokers: []string{"localhost:9092"}})
if err != nil {
    log.Fatal(err)
}
defer pub.Close()

worker := xoutbox.NewWorker(store, pub,
    xoutbox.WithBatchSize(200),
    xoutbox.WithConcurrency(16),
    xoutbox.WithCleanup(time.Minute, 72*time.Hour), // delete published rows after 3 days
)
go worker.Run(ctx) // returns nil once ctx is cancelled and the in-flight batch is committed

Writing events in your transaction

func (s *Service) CreateOrder(ctx context.Context, o Order) error {
    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    if _, err := tx.ExecContext(ctx, `INSERT INTO orders (id, total) VALUES ($1, $2)`, o.ID, o.Total); err != nil {
        return err
    }

    evt, err := xoutbox.NewJSONEvent("orders", "order.created", o) // topic, type, payload
    if err != nil {
        return err
    }
    evt = evt.WithKey(o.CustomerID).WithHeader("trace-id", traceID(ctx))

    if err := s.outbox.Insert(ctx, tx, evt); err != nil {
        return err
    }
    if err := tx.Commit(); err != nil {
        return err
    }
    s.worker.Notify() // optional: publish right away instead of waiting for the next poll
    return nil
}

Other brokers

// NATS JetStream (exactly-once into the stream thanks to Nats-Msg-Id dedup)
nc, _ := nats.Connect(nats.DefaultURL)
js, _ := jetstream.New(nc)
pub := xnats.New(js, xnats.WithSubjectPrefix("events."))     // topic "orders" → subject "events.orders"

// RabbitMQ (confirms + persistent + auto-reconnect)
pub, _ := rabbitmq.New(rabbitmq.Config{URL: "amqp://guest:guest@localhost:5672/", Exchange: "events"})

// Redis Streams
rdb := goredis.NewUniversalClient(&goredis.UniversalOptions{Addrs: []string{"localhost:6379"}})
pub := xredis.New(rdb, xredis.Config{MaxLen: 1_000_000})     // xredis.PubSub for fire-and-forget

// Anything else
pub := xoutbox.PublisherFunc(func(ctx context.Context, e xoutbox.Event) error {
    return myClient.Send(ctx, e.Topic, e.Payload)
})

How Event fields map to each broker:

To route to different brokers, write a PublisherFunc that switches on e.Topic or e.Type.

Delivery guarantees

xoutbox gives you at-least-once delivery. A message can be published twice if the broker accepts it but the worker crashes before marking it published. Make consumers idempotent by using the outbox-id header / message ID as a dedup key. NATS JetStream does this for you.

Ordering. Inside one batch, events that share a Key are published one after another. If one fails, the rest of its group is put back without using up an attempt. For strict ordering across retries and multiple workers, enable postgres.WithOrdering() / mysql.WithOrdering(): an event is only claimed once every older event with the same key has been published or dead-lettered. Events with an empty key are never blocked.

Worker options

worker.RunOnce(ctx) processes a single batch, which is handy for cron jobs, Lambdas and tests.

Metrics (Prometheus example)

published := promauto.NewCounterVec(prometheus.CounterOpts{Name: "outbox_published_total"}, []string{"topic"})
failed    := promauto.NewCounterVec(prometheus.CounterOpts{Name: "outbox_failed_total"}, []string{"topic", "dead"})
lag       := promauto.NewHistogram(prometheus.HistogramOpts{Name: "outbox_lag_seconds"})

worker := xoutbox.NewWorker(store, pub, xoutbox.WithHooks(xoutbox.Hooks{
    OnPublished: func(e xoutbox.Event) {
        published.WithLabelValues(e.Topic).Inc()
        lag.Observe(time.Since(e.CreatedAt).Seconds())
    },
    OnFailed: func(e xoutbox.Event, err error, dead bool) {
        failed.WithLabelValues(e.Topic, strconv.FormatBool(dead)).Inc()
    },
}))

// Backlog gauges: scrape store.Stats(ctx) → Pending, Processing, Dead, OldestPending

Operations

n, err := store.Requeue(ctx)            // replay every dead event
n, err := store.Requeue(ctx, id1, id2)  // replay specific events
st, err := store.Stats(ctx)             // backlog snapshot
deleted, err := store.Cleanup(ctx, 24*time.Hour)
fmt.Println(store.SchemaSQL())          // DDL for your migration tool

-- inspect the dead-letter queue
SELECT id, topic, event_type, attempts, last_error, created_at FROM outbox WHERE status = 'dead';

Schema

store.Migrate(ctx) creates the table and indexes idempotently. You can also copy postgres/schema.sql or mysql/schema.sql into your migrations (replace {t}/{n} with the table name). Key columns: seq (insertion order), id, topic, partition_key, payload, headers, status, attempts, max_attempts, locked_by, locked_until, available_at, last_error.

All timestamps come from the database clock, so app-server clock skew and timezone settings don't matter.

Custom backends

Implement xoutbox.Store for another database, or xoutbox.Publisher / xoutbox.BatchPublisher for another broker. internal/storetest holds the conformance suite used by the SQL stores.

type Store interface {
    Claim(ctx context.Context, owner string, limit int, lease time.Duration) ([]Event, error)
    MarkPublished(ctx context.Context, ids []string) error
    MarkFailed(ctx context.Context, owner string, f Failure) error
}

Development

make test               # unit tests
make test-integration   # starts Postgres, MySQL, Kafka, NATS, RabbitMQ, Redis via docker compose

License

MIT