gobox

README

GoBoX

GoBoX is a Go library that implements the transactional outbox pattern on top of PostgreSQL. It is designed for use with Debezium Outbox SMT, so application events written into the outbox table can be reliably published to Kafka through CDC.

What It Does

The library provides:

• an Outbox service for writing domain events into a Postgres outbox table;
• embedded SQL migrations for creating the outbox schema;
• a small configuration object for controlling the Goose migration table name.

Intended Usage

The typical flow is:

1. your application writes domain data and an outbox event in the same database transaction;
2. Debezium reads the outbox table from PostgreSQL;
3. Debezium Outbox SMT transforms those rows into Kafka messages.

This keeps database state changes and message publication aligned without requiring the application to publish directly to Kafka inside the business transaction.

Example

package main

import (
	"context"

	"github.com/fr33dman/gobox/pkg"
)

func produceEvent(ctx context.Context, dbtx someDBTX) error {
	outbox := pkg.NewOutbox(dbtx)

	_, err := outbox.Produce(ctx, pkg.Event{
		AggregateType: "documents",
		AggregateId:   "doc-42",
		EventType:     "document.created",
		Payload:       `{"id":"doc-42"}`,
		Headers:       `{"trace_id":"abc-123"}`,
		DedupKey:      "document.created:doc-42",
		Seq:           1,
	})
	return err
}

Migrations

Use the embedded migrations to create the outbox table:

cfg := pkg.OutboxConfiguration{}
if err := cfg.LoadFromEnv(); err != nil {
	return err
}

pkg.MigrateOutboxTable(ctx, db, cfg)

Environment variables:

• GOBOX_GOOSE_TABLE_NAME overrides the Goose migration table name.

Debezium Outbox SMT

This library is written for use with Debezium Outbox SMT.

Recommended connector and SMT settings:

# Use PostgreSQL built-in logical decoding plugin.
# pgoutput does not require installing additional PostgreSQL extensions.
plugin.name: "pgoutput"

# Required logical identifier of this Debezium source.
# Normally used as a prefix for Debezium-generated topics and internal metadata.
# Outbox Event Router overrides the final business event topic name below.
topic.prefix: "<setup-your-topic-prefix>"

# PostgreSQL logical replication slot used by Debezium to track
# its position in the WAL. Must be unique per connector.
slot.name: "<setup-slot-name>"

# PostgreSQL logical replication publication used by pgoutput.
publication.name: "<setup-publication-name>"

# Automatically create/update the publication so it contains only
# tables matched by table.include.list.
publication.autocreate.mode: "filtered"

# Capture changes only from the outbox table.
table.include.list: "<setup-your-schema>.outbox"


# Serialize Kafka message values as JSON.
value.converter: "org.apache.kafka.connect.json.JsonConverter"

# Do not include Kafka Connect schema metadata in every JSON message.
# The Kafka value will contain only the actual event payload.
value.converter.schemas.enable: "false"

# Serialize Kafka message keys as JSON.
key.converter: "org.apache.kafka.connect.json.JsonConverter"

# Do not include Kafka Connect schema metadata in message keys.
key.converter.schemas.enable: "false"

# Register an SMT named "outbox".
transforms: "outbox"

# Transform raw Debezium CDC records into domain events using
# Debezium's Transactional Outbox Event Router.
transforms.outbox.type: "io.debezium.transforms.outbox.EventRouter"

# Use outbox.id as the unique event ID.
# Debezium puts it into the Kafka "id" header.
transforms.outbox.table.field.event.id: "id"

# Use aggregate_id as the Kafka message key.
# Events with the same aggregate_id therefore go to the same partition,
# preserving their order within that partition.
transforms.outbox.table.field.event.key: "aggregate_id"

# Use payload as the Kafka message value.
transforms.outbox.table.field.event.payload: "payload"

# Use created_at as the Kafka record timestamp instead of
# the timestamp at which Debezium processed the WAL record.
transforms.outbox.table.field.event.timestamp: "created_at"

# Parse the PostgreSQL JSONB payload and emit it as a JSON object
# instead of an escaped JSON string.
transforms.outbox.table.expand.json.payload: "true"

# Use aggregate_type to determine the destination topic.
transforms.outbox.route.by.field: "aggregate_type"

# Example:
# aggregate_type = "documents" -> <topic-prefix>.documents
# aggregate_type = "repositories" -> <topic-prefix>.repositories
transforms.outbox.route.topic.replacement: "<setup-your-topic-prefix>.${routedByValue}"

# Put `event_type` into the Kafka header named `type`, `headers` into the Kafka header named `headers`
transforms.outbox.table.fields.additional.placement: "event_type:header:type,headers:header:headers"

Documentation

Index

• func MigrateOutboxTable(ctx context.Context, db *sql.DB, cfg OutboxConfiguration) (err error)
• type Event
• type Outbox
  • func NewOutbox(dbtx repo2.DBTX) *Outbox
  • func (o *Outbox) Produce(ctx context.Context, ev Event) (createdEv Event, err error)
• type OutboxConfiguration
  • func (c *OutboxConfiguration) LoadFromEnv() (err error)

Functions

func MigrateOutboxTable

func MigrateOutboxTable(ctx context.Context, db *sql.DB, cfg OutboxConfiguration) (err error)

MigrateOutboxTable applies the embedded outbox schema migrations to the target database.

Types

type Event

type Event struct {
	Id            uuid.UUID
	AggregateType string
	AggregateId   string
	EventType     string
	Payload       string
	Headers       string
	CreatedAt     time.Time
	DedupKey      string
	Seq           int
}

Event is the library-level representation of a single outbox record.

type Outbox

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

Outbox persists domain events into the Postgres outbox table.

func NewOutbox

func NewOutbox(dbtx repo2.DBTX) *Outbox

NewOutbox binds the service to any pgx-compatible query executor or transaction.

func (*Outbox) Produce

func (o *Outbox) Produce(ctx context.Context, ev Event) (createdEv Event, err error)

Produce stores an event row that Debezium can later publish to Kafka.

type OutboxConfiguration

type OutboxConfiguration struct {
	GooseTableName string `env:"GOBOX_GOOSE_TABLE_NAME" envDefault:"goose_db_version_outbox"`
}

OutboxConfiguration holds migration-related settings for the library.

func (*OutboxConfiguration) LoadFromEnv

func (c *OutboxConfiguration) LoadFromEnv() (err error)

LoadFromEnv fills the config from environment variables declared in the struct tags.