Key Features

Usage

The Writer

// Setup database connectiondb,_:=sql.Open("pgx","postgres://user:password@localhost:5432/outbox?sslmode=disable")// Create a DBContext and Writer instancedbCtx:=outbox.NewDBContext(db,outbox.SQLDialectPostgres)writer:=outbox.NewWriter(dbCtx)// In your business logic://// Create your entity and outbox messageentity:=Entity{ID:uuid.New(),CreatedAt:time.Now().UTC(),}payload,_:=json.Marshal(entity)metadata:=json.RawMessage(`{"trace_id":"abc123","correlation_id":"xyz789"}`)msg:=outbox.NewMessage(payload,outbox.WithCreatedAt(entity.CreatedAt),outbox.WithMetadata(metadata))// Write message and entity in a single transactionerr=writer.Write(ctx,msg,// This user-defined callback executes queries within the// same transaction that stores the outbox messagefunc(ctx context.Context,txQueryer outbox.TxQueryer)error {_,err:=txQueryer.ExecContext(ctx,"INSERT INTO entity (id, created_at) VALUES ($1, $2)",entity.ID,entity.CreatedAt,        )returnerr    })

The Reader

// Create a message publisher implementationtypemessagePublisherstruct {// Your message broker client (e.g., Kafka, RabbitMQ)}func (p*messagePublisher)Publish(ctx context.Context,msg*outbox.Message)error {// Publish the message to your broker. See examples below for specific implementationsreturnnil}// Create and start the readerreader:=outbox.NewReader(dbCtx,// Database context&messagePublisher{},// Publisher implementationoutbox.WithInterval(5*time.Second),// Polling interval (default: 10s)outbox.WithReadBatchSize(200),// Read batch size (default: 100)outbox.WithDeleteBatchSize(50),// Delete batch size (default: 20)outbox.WithMaxAttempts(300),// Discard after 300 attempts (default: MaxInt32)outbox.WithExponentialDelay(// Delay between attempts (default: Exponential; can also use Fixed or Custom)500*time.Millisecond,// Initial delay (default: 200ms)30*time.Minute),// Maximum delay (default: 1h))reader.Start()deferreader.Stop(context.Background())// Stop during application shutdown// Monitor standard processing errors (publish / update / delete / read).gofunc() {forerr:=rangereader.Errors() {switche:=err.(type) {case*outbox.PublishError:log.Printf("Failed to publish message | ID: %s | Error: %v",e.Message.ID,e.Err)case*outbox.UpdateError:log.Printf("Failed to update message | ID: %s | Error: %v",e.Message.ID,e.Err)case*outbox.DeleteError:log.Printf("Batch message deletion failed | Count: %d | Error: %v",len(e.Messages),e.Err)for_,msg:=rangee.Messages {log.Printf("Failed to delete message | ID: %s",msg.ID)            }case*outbox.ReadError:log.Printf("Failed to read outbox messages | Error: %v",e.Err)default:log.Printf("Unexpected error occurred | Error: %v",e)        }    }}()// Monitor discarded messages (hit the max-attempts threshold).gofunc() {formsg:=rangereader.DiscardedMessages() {log.Printf("outbox message %s discarded after %d attempts",msg.ID,msg.TimesAttempted)// Example next steps://   • forward to a dead-letter topic//   • raise an alert / metric//   • persist for manual inspection    }}()

Database Setup

1. Choose Your Database Dialect

// Example creating a DBContext with MySQL dialectdbCtx:=outbox.NewDBContext(db,outbox.SQLDialectMySQL)

2. Create the Outbox Table

Examples

cd examples/postgres-kafka# or examples/oracle-nats or examples/mysql-rabitmq../../scripts/up-and-wait.shgo run service.go# In another terminal trigger a POST to trigger entity creationcurl -X POST http://localhost:8080/entity

FAQ

What happens when multiple instances of my service use the library?

How to instantiate aDBContext when usingpgxpool ?

import ("github.com/jackc/pgx/v5/pgxpool""github.com/jackc/pgx/v5/stdlib""github.com/oagudo/outbox")// ...pool,_:=pgxpool.New(context.Background(),os.Getenv("DATABASE_URL"))db:=stdlib.OpenDBFromPool(pool)dbCtx:=outbox.NewDBContext(db,outbox.SQLDialectPostgres)

Contributing