### Run Example Services

Source: https://github.com/oagudo/outbox/blob/main/README.md

Commands to navigate to example directories, start infrastructure, and trigger entity creation.

```bash
cd examples/postgres-kafka # or examples/oracle-nats or examples/mysql-rabbitmq
../../scripts/up-and-wait.sh
go run service.go

# In another terminal trigger a POST to trigger entity creation
curl -X POST http://localhost:8080/entity
```

--------------------------------

### Start Dependencies and Run Service

Source: https://github.com/oagudo/outbox/blob/main/examples/postgres-kafka/README.md

Execute this command to start the necessary dependencies and run the Go service. Ensure you are in the correct directory.

```bash
../../scripts/up-and-wait.sh && go run service.go
```

--------------------------------

### Complete Service Example with Outbox

Source: https://context7.com/oagudo/outbox/llms.txt

This example shows a full integration of the Outbox library within an HTTP service, demonstrating Writer and Reader components with PostgreSQL and Kafka. It includes database setup, outbox configuration, error monitoring, an HTTP handler for event publishing, and graceful server shutdown.

```go
package main

import (
    "context"
    "database/sql"
    "encoding/json"
    "log"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/google/uuid"
    _ "github.com/jackc/pgx/v5/stdlib"
    "github.com/oagudo/outbox"
)

type Event struct {
    ID        uuid.UUID `json:"id"`
    Type      string    `json:"type"`
    Data      string    `json:"data"`
    CreatedAt time.Time `json:"created_at"`
}

type LogPublisher struct{}

func (p *LogPublisher) Publish(ctx context.Context, msg *outbox.Message) error {
    log.Printf("Published: %s | Payload: %s", msg.ID, string(msg.Payload))
    return nil
}

func main() {
    // Database setup
    db, err := sql.Open("pgx", "postgres://postgres:postgres@localhost:5432/mydb?sslmode=disable")
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()

    // Outbox setup
    dbCtx := outbox.NewDBContext(db, outbox.SQLDialectPostgres)
    publisher := &LogPublisher{}

    writer := outbox.NewWriter(dbCtx,
        outbox.WithOptimisticPublisher(publisher),
    )

    reader := outbox.NewReader(dbCtx, publisher,
        outbox.WithInterval(5*time.Second),
        outbox.WithMaxAttempts(100),
        outbox.WithExponentialDelay(500*time.Millisecond, 30*time.Minute),
    )

    // Error monitoring
    go func() {
        for err := range reader.Errors() {
            log.Printf("Outbox error: %v", err)
        }
    }()
    go func() {
        for msg := range reader.DiscardedMessages() {
            log.Printf("Discarded message: %s after %d attempts", msg.ID, msg.TimesAttempted)
        }
    }()

    reader.Start()

    // HTTP handler
    mux := http.NewServeMux()
    mux.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) {
        if r.Method != http.MethodPost {
            http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
            return
        }

        event := Event{
            ID:        uuid.New(),
            Type:      "user.created",
            Data:      "example data",
            CreatedAt: time.Now().UTC(),
        }

        payload, _ := json.Marshal(event)
        metadata, _ := json.Marshal(map[string]string{
            "trace_id": uuid.New().String(),
        })

        msg := outbox.NewMessage(payload,
            outbox.WithCreatedAt(event.CreatedAt),
            outbox.WithMetadata(metadata),
        )

        err := writer.WriteOne(r.Context(), msg, func(ctx context.Context, tx outbox.TxQueryer) error {
            _, err := tx.ExecContext(ctx,
                "INSERT INTO events (id, type, data, created_at) VALUES ($1, $2, $3, $4)",
                event.ID, event.Type, event.Data, event.CreatedAt,
            )
            return err
        })

        if err != nil {
            http.Error(w, "Internal error", http.StatusInternalServerError)
            return
        }

        w.Header().Set("Content-Type", "application/json")
        w.WriteHeader(http.StatusCreated)
        json.NewEncoder(w).Encode(event)
    })

    // Server with graceful shutdown
    srv := &http.Server{Addr: ":8080", Handler: mux}

    go func() {
        log.Println("Server starting on :8080")
        if err := srv.ListenAndServe(); err != http.ErrServerClosed {
            log.Fatal(err)
        }
    }()

    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    <-quit

    log.Println("Shutting down...")
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    srv.Shutdown(ctx)
    reader.Stop(ctx)
    log.Println("Shutdown complete")
}
```

--------------------------------

### Configure and Start Background Message Publisher

Source: https://context7.com/oagudo/outbox/llms.txt

Demonstrates how to create and configure an outbox reader for background message publishing. It shows various options for polling intervals, batch sizes, timeouts, retry strategies, and channel sizes. Ensure your database connection and publisher implementation are correctly set up.

```go
package main

import (
    "context"
    "database/sql"
    "log"
    "time"
    "github.com/oagudo/outbox"
    _ "github.com/jackc/pgx/v5/stdlib"
)

type KafkaPublisher struct{}

func (p *KafkaPublisher) Publish(ctx context.Context, msg *outbox.Message) error {
    log.Printf("Publishing to Kafka: %s", msg.ID)
    // Implement your Kafka publishing logic
    return nil
}

func main() {
    db, _ := sql.Open("pgx", "postgres://user:pass@localhost:5432/mydb?sslmode=disable")
    defer db.Close()

    dbCtx := outbox.NewDBContext(db, outbox.SQLDialectPostgres)
    publisher := &KafkaPublisher{}

    // Create reader with all configuration options
    reader := outbox.NewReader(dbCtx, publisher,
        // Polling interval (default: 10s)
        outbox.WithInterval(5*time.Second),

        // Batch sizes
        outbox.WithReadBatchSize(200),    // Messages per poll (default: 100)
        outbox.WithDeleteBatchSize(50),   // Delete batch size (default: 20)

        // Timeouts
        outbox.WithReadTimeout(10*time.Second),    // Default: 5s
        outbox.WithPublishTimeout(30*time.Second), // Default: 5s
        outbox.WithDeleteTimeout(10*time.Second),  // Default: 5s
        outbox.WithUpdateTimeout(5*time.Second),   // Default: 5s

        // Retry configuration
        outbox.WithMaxAttempts(100), // Discard after 100 attempts (default: MaxInt32)

        // Backoff strategy (choose one)
        outbox.WithExponentialDelay(500*time.Millisecond, 30*time.Minute),
        // OR: outbox.WithFixedDelay(5*time.Second),

        // Channel sizes for monitoring
        outbox.WithErrorChannelSize(256),             // Default: 128
        outbox.WithDiscardedMessagesChannelSize(64), // Default: 128
    )

    // Start background processing
    reader.Start()

    // Application runs...
    time.Sleep(1 * time.Minute)

    // Graceful shutdown
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    if err := reader.Stop(ctx); err != nil {
        log.Printf("Reader shutdown timeout: %v", err)
    }
}
```

--------------------------------

### Implement MessagePublisher Interface in Go

Source: https://context7.com/oagudo/outbox/llms.txt

Provides examples for implementing the MessagePublisher interface for Kafka, RabbitMQ, and NATS. Ensure implementations are idempotent as the Publish method may be called multiple times.

```go
package main

import (
    "context"
    "encoding/json"
    "log"
    "github.com/oagudo/outbox"
    "github.com/segmentio/kafka-go"
)

// Kafka implementation
type KafkaPublisher struct {
    writer *kafka.Writer
}

func (p *KafkaPublisher) Publish(ctx context.Context, msg *outbox.Message) error {
    // Parse metadata for headers
    var metadata map[string]string
    if len(msg.Metadata) > 0 {
        json.Unmarshal(msg.Metadata, &metadata)
    }

    headers := make([]kafka.Header, 0)
    for k, v := range metadata {
        headers = append(headers, kafka.Header{Key: k, Value: []byte(v)})
    }

    err := p.writer.WriteMessages(ctx, kafka.Message{
        Key:     []byte(msg.ID.String()),
        Value:   msg.Payload,
        Headers: headers,
    })
    if err != nil {
        log.Printf("Kafka publish failed for %s: %v", msg.ID, err)
        return err
    }

    log.Printf("Published %s to Kafka", msg.ID)
    return nil
}

// RabbitMQ implementation (example structure)
type RabbitMQPublisher struct {
    // channel *amqp.Channel
    exchange string
}

func (p *RabbitMQPublisher) Publish(ctx context.Context, msg *outbox.Message) error {
    // Implement RabbitMQ publishing
    // return p.channel.PublishWithContext(ctx, p.exchange, "", false, false, amqp.Publishing{
    //     MessageId:   msg.ID.String(),
    //     ContentType: "application/json",
    //     Body:        msg.Payload,
    // })
    return nil
}

// NATS implementation (example structure)
type NATSPublisher struct {
    // js nats.JetStreamContext
    subject string
}

func (p *NATSPublisher) Publish(ctx context.Context, msg *outbox.Message) error {
    // Use message ID for deduplication
    // _, err := p.js.Publish(p.subject, msg.Payload, nats.MsgId(msg.ID.String()))
    // return err
    return nil
}
```

--------------------------------

### Implement and Start Outbox Reader

Source: https://github.com/oagudo/outbox/blob/main/README.md

Defines the message publisher interface and configures the reader with polling intervals, batch sizes, and retry logic.

```go
// Create a message publisher implementation
type messagePublisher struct {
    // 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 implementations
    return nil
}

// Create and start the reader
reader := outbox.NewReader(
    dbCtx,                              // Database context
    &messagePublisher{},                // Publisher implementation
    outbox.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()
defer reader.Stop(context.Background()) // Stop during application shutdown
```

--------------------------------

### Initialize Database Context

Source: https://github.com/oagudo/outbox/blob/main/README.md

Configures the database context with a specific SQL dialect for the outbox library.

```go
dbCtx := outbox.NewDBContext(db, outbox.SQLDialectMySQL)
```

--------------------------------

### Instantiate DBContext with pgxpool

Source: https://github.com/oagudo/outbox/blob/main/README.md

How to convert a pgxpool connection into a standard sql.DB instance for use with the outbox library.

```go
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)
```

--------------------------------

### Configure Outbox Writer

Source: https://github.com/oagudo/outbox/blob/main/README.md

Initializes the outbox writer with an optimistic publisher. Publishing occurs asynchronously after transaction commit.

```go
// Create publisher (see Reader section below)
publisher := &messagePublisher{}

// Enable optimistic publishing in writer
writer := outbox.NewWriter(dbCtx, outbox.WithOptimisticPublisher(publisher))
```

--------------------------------

### Configure Retry Backoff Strategies in Go

Source: https://context7.com/oagudo/outbox/llms.txt

Demonstrates how to initialize a reader with fixed, exponential, or custom delay functions to control retry behavior.

```go
package main

import (
    "context"
    "database/sql"
    "time"
    "github.com/oagudo/outbox"
    _ "github.com/jackc/pgx/v5/stdlib"
)

type MyPublisher struct{}

func (p *MyPublisher) Publish(ctx context.Context, msg *outbox.Message) error {
    return nil
}

func main() {
    db, _ := sql.Open("pgx", "postgres://user:pass@localhost:5432/mydb?sslmode=disable")
    defer db.Close()

    dbCtx := outbox.NewDBContext(db, outbox.SQLDialectPostgres)

    // Fixed delay - same delay for all retries
    // Delay: 5s, 5s, 5s, 5s, ...
    readerFixed := outbox.NewReader(dbCtx, &MyPublisher{},
        outbox.WithFixedDelay(5*time.Second),
    )

    // Exponential delay (default) - doubles each attempt up to max
    // With 200ms initial, 1h max:
    // Attempt 0: 200ms, Attempt 1: 400ms, Attempt 2: 800ms, ...
    // Attempt 15+: 1h (capped at max)
    readerExponential := outbox.NewReader(dbCtx, &MyPublisher{},
        outbox.WithExponentialDelay(200*time.Millisecond, 1*time.Hour),
    )

    // Custom delay function
    customDelay := func(attempt int) time.Duration {
        // Example: Linear backoff with jitter
        baseDelay := time.Duration(attempt+1) * 500 * time.Millisecond
        maxDelay := 5 * time.Minute
        if baseDelay > maxDelay {
            return maxDelay
        }
        return baseDelay
    }
    readerCustom := outbox.NewReader(dbCtx, &MyPublisher{},
        outbox.WithDelay(customDelay),
    )

    _ = readerFixed
    _ = readerExponential
    _ = readerCustom
}
```

--------------------------------

### Initialize DBContext for Outbox

Source: https://context7.com/oagudo/outbox/llms.txt

Configures the database connection and SQL dialect required for Writer and Reader components. Supports various SQL dialects including PostgreSQL, MySQL, and SQLite.

```go
package main

import (
    "database/sql"
    "github.com/oagudo/outbox"
    _ "github.com/jackc/pgx/v5/stdlib"
)

func main() {
    // PostgreSQL connection
    db, err := sql.Open("pgx", "postgres://user:pass@localhost:5432/mydb?sslmode=disable")
    if err != nil {
        panic(err)
    }
    defer db.Close()

    // Create DBContext with PostgreSQL dialect (default table name: "outbox")
    dbCtx := outbox.NewDBContext(db, outbox.SQLDialectPostgres)

    // Create DBContext with custom table name
    dbCtxCustom := outbox.NewDBContext(db, outbox.SQLDialectPostgres,
        outbox.WithTableName("my_outbox_table"))

    // Supported dialects:
    // outbox.SQLDialectPostgres
    // outbox.SQLDialectMySQL
    // outbox.SQLDialectMariaDB
    // outbox.SQLDialectSQLite
    // outbox.SQLDialectOracle
    // outbox.SQLDialectSQLServer

    _ = dbCtx
    _ = dbCtxCustom
}
```

--------------------------------

### Configure Optimistic Publishing

Source: https://context7.com/oagudo/outbox/llms.txt

Enable immediate message publishing after transaction commit to reduce latency. The background Reader serves as a fallback mechanism if the optimistic attempt fails.

```go
package main

import (
    "context"
    "database/sql"
    "encoding/json"
    "log"
    "time"
    "github.com/google/uuid"
    "github.com/oagudo/outbox"
    _ "github.com/jackc/pgx/v5/stdlib"
)

type MyPublisher struct{}

func (p *MyPublisher) Publish(ctx context.Context, msg *outbox.Message) error {
    log.Printf("Publishing message %s: %s", msg.ID, string(msg.Payload))
    // Publish to your broker (Kafka, RabbitMQ, etc.)
    return nil
}

func main() {
    db, _ := sql.Open("pgx", "postgres://user:pass@localhost:5432/mydb?sslmode=disable")
    defer db.Close()

    dbCtx := outbox.NewDBContext(db, outbox.SQLDialectPostgres)
    publisher := &MyPublisher{}

    // Enable optimistic publishing with custom timeout
    writer := outbox.NewWriter(dbCtx,
        outbox.WithOptimisticPublisher(publisher),
        outbox.WithOptimisticTimeout(5*time.Second), // Default: 10s
    )

    ctx := context.Background()
    payload, _ := json.Marshal(map[string]string{"event": "order_created", "id": uuid.New().String()})
    msg := outbox.NewMessage(payload)

    // After commit, message is published immediately (async)
    // If optimistic publish fails, Reader handles it later
    err := writer.WriteOne(ctx, msg, func(ctx context.Context, tx outbox.TxQueryer) error {
        _, err := tx.ExecContext(ctx, "INSERT INTO events (id) VALUES ($1)", uuid.New())
        return err
    })

    if err != nil {
        panic(err)
    }
    // Message published immediately after commit (non-blocking)
}
```

--------------------------------

### Initialize Writer and Store Message (User Managed Transactions)

Source: https://github.com/oagudo/outbox/blob/main/README.md

Initializes the database context and writer, then uses an unmanaged writer to store a message within an existing user-managed transaction. Ensure the transaction is properly committed or rolled back.

```go
// Initialise Writer
db, _ := sql.Open("pgx", "postgres://..." )
dbCtx := outbox.NewDBContext(db, outbox.SQLDialectPostgres)
writer := outbox.NewWriter(dbCtx)

tx, _ := db.BeginTx(ctx, nil)
def tx.Rollback()

_, _ = tx.ExecContext(ctx, "INSERT INTO entity (...) VALUES (...)", ...)

// Create message
payload, _ := json.Marshal(entity)
msg := outbox.NewMessage(payload)

// Use unmanaged writer with existing transaction to store the message
unmanagedWriter := writer.Unmanaged()
err = unmanagedWriter.Store(ctx, tx, msg)

tx.Commit()
```

--------------------------------

### Conditional and Multi-Message Publishing with Write

Source: https://context7.com/oagudo/outbox/llms.txt

Use Write for conditional message publishing or storing multiple messages within a single transaction. The callback function receives a MessageWriter to store messages.

```go
package main

import (
    "context"
    "database/sql"
    "encoding/json"
    "errors"
    "time"
    "github.com/google/uuid"
    "github.com/oagudo/outbox"
    _ "github.com/jackc/pgx/v5/stdlib"
)

var ErrOrderNotPending = errors.New("order is not in pending status")

func main() {
    db, _ := sql.Open("pgx", "postgres://user:pass@localhost:5432/mydb?sslmode=disable")
    defer db.Close()

    dbCtx := outbox.NewDBContext(db, outbox.SQLDialectPostgres)
    writer := outbox.NewWriter(dbCtx)

    ctx := context.Background()
    orderID := uuid.New()

    // Conditional message publishing - only publish if update succeeds
    err := writer.Write(ctx, func(ctx context.Context, tx outbox.TxQueryer, msgWriter outbox.MessageWriter) error {
        result, err := tx.ExecContext(ctx,
            "UPDATE orders SET status = 'confirmed' WHERE id = $1 AND status = 'pending'",
            orderID,
        )
        if err != nil {
            return err
        }

        rows, _ := result.RowsAffected()
        if rows == 0 {
            return ErrOrderNotPending // No message published, transaction rolled back
        }

        // Publish confirmation message
        payload, _ := json.Marshal(map[string]interface{}{
            "order_id": orderID,
            "status":   "confirmed",
            "time":     time.Now().UTC(),
        })
        return msgWriter.Store(ctx, outbox.NewMessage(payload))
    })

    if err != nil && !errors.Is(err, ErrOrderNotPending) {
        panic(err)
    }

    // Multiple messages in single transaction
    err = writer.Write(ctx, func(ctx context.Context, tx outbox.TxQueryer, msgWriter outbox.MessageWriter) error {
        // Insert batch of items
        items := []string{"item-1", "item-2", "item-3"}
        for _, item := range items {
            _, err := tx.ExecContext(ctx,
                "INSERT INTO items (id, name) VALUES ($1, $2)",
                uuid.New(), item,
            )
            if err != nil {
                return err
            }

            // Store message for each item
            payload, _ := json.Marshal(map[string]string{"item": item})
            if err := msgWriter.Store(ctx, outbox.NewMessage(payload)); err != nil {
                return err
            }
        }
        return nil
    })

    if err != nil {
        panic(err)
    }
}
```

--------------------------------

### Create Outbox Messages

Source: https://context7.com/oagudo/outbox/llms.txt

Constructs messages with payloads and optional configurations such as custom IDs, timestamps, metadata, and scheduled delivery times.

```go
package main

import (
    "encoding/json"
    "time"
    "github.com/google/uuid"
    "github.com/oagudo/outbox"
)

func main() {
    // Create entity to publish
    entity := map[string]interface{}{
        "id":         uuid.New().String(),
        "name":       "Example Entity",
        "created_at": time.Now().UTC(),
    }
    payload, _ := json.Marshal(entity)

    // Basic message with auto-generated ID and current timestamp
    msg := outbox.NewMessage(payload)

    // Message with custom ID
    customID := uuid.New()
    msgWithID := outbox.NewMessage(payload, outbox.WithID(customID))

    // Message with custom creation time (useful for ordering)
    msgWithTime := outbox.NewMessage(payload,
        outbox.WithCreatedAt(time.Now().Add(-1*time.Hour)))

    // Message with metadata (trace IDs, correlation IDs, etc.)
    metadata, _ := json.Marshal(map[string]string{
        "trace_id":       uuid.New().String(),
        "correlation_id": uuid.New().String(),
    })
    msgWithMetadata := outbox.NewMessage(payload, outbox.WithMetadata(metadata))

    // Scheduled message (publish in the future)
    scheduledTime := time.Now().Add(30 * time.Minute)
    scheduledMsg := outbox.NewMessage(payload, outbox.WithScheduledAt(scheduledTime))

    // Combined options
    fullMsg := outbox.NewMessage(payload,
        outbox.WithID(uuid.New()),
        outbox.WithCreatedAt(time.Now().UTC()),
        outbox.WithScheduledAt(time.Now().Add(5*time.Minute)),
        outbox.WithMetadata(metadata),
    )

    _ = msg
    _ = msgWithID
    _ = msgWithTime
    _ = msgWithMetadata
    _ = scheduledMsg
    _ = fullMsg
}
```

--------------------------------

### Trigger Entity Creation Request

Source: https://github.com/oagudo/outbox/blob/main/examples/postgres-kafka/README.md

Use this curl command in a separate terminal to send a POST request to create an entity. This action initiates the outbox pattern.

```bash
curl -X POST http://localhost:8080/entity
```

--------------------------------

### Publish Single Message with WriteOne

Source: https://context7.com/oagudo/outbox/llms.txt

Use WriteOne to store a single message in the outbox table atomically with a database operation. The transaction commits if the callback succeeds, otherwise it rolls back.

```go
package main

import (
    "context"
    "database/sql"
    "encoding/json"
    "time"
    "github.com/google/uuid"
    "github.com/oagudo/outbox"
    _ "github.com/jackc/pgx/v5/stdlib"
)

type Order struct {
    ID        uuid.UUID `json:"id"`
    CustomerID string   `json:"customer_id"`
    Total     float64   `json:"total"`
    CreatedAt time.Time `json:"created_at"`
}

func main() {
    db, _ := sql.Open("pgx", "postgres://user:pass@localhost:5432/mydb?sslmode=disable")
    defer db.Close()

    dbCtx := outbox.NewDBContext(db, outbox.SQLDialectPostgres)
    writer := outbox.NewWriter(dbCtx)

    ctx := context.Background()
    order := Order{
        ID:        uuid.New(),
        CustomerID: "cust-123",
        Total:     99.99,
        CreatedAt: time.Now().UTC(),
    }

    payload, _ := json.Marshal(order)
    msg := outbox.NewMessage(payload, outbox.WithCreatedAt(order.CreatedAt))

    // WriteOne stores entity and message atomically
    err := writer.WriteOne(ctx, msg, func(ctx context.Context, tx outbox.TxQueryer) error {
        _, err := tx.ExecContext(ctx,
            "INSERT INTO orders (id, customer_id, total, created_at) VALUES ($1, $2, $3, $4)",
            order.ID, order.CustomerID, order.Total, order.CreatedAt,
        )
        return err // Transaction rolls back if error returned
    })

    if err != nil {
        panic(err)
    }
    // Both order and message are committed atomically
}
```

--------------------------------

### Create Outbox Table Schema

Source: https://github.com/oagudo/outbox/blob/main/README.md

SQL schema definitions for the outbox table, tailored for different database engines.

```sql
CREATE TABLE IF NOT EXISTS outbox (
    id UUID PRIMARY KEY,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
    scheduled_at TIMESTAMP WITH TIME ZONE NOT NULL,
    metadata BYTEA,
    payload BYTEA NOT NULL,
    times_attempted INTEGER NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_outbox_created_at ON outbox (created_at);
CREATE INDEX IF NOT EXISTS idx_outbox_scheduled_at ON outbox (scheduled_at);
```

```sql
CREATE TABLE IF NOT EXISTS outbox (
    id BINARY(16) PRIMARY KEY,
    created_at TIMESTAMP(3) NOT NULL,
    scheduled_at TIMESTAMP(3) NOT NULL,
    metadata BLOB,
    payload BLOB NOT NULL,
    times_attempted INT NOT NULL
);

CREATE INDEX idx_outbox_created_at ON outbox (created_at);
CREATE INDEX idx_outbox_scheduled_at ON outbox (scheduled_at);
```

```sql
CREATE TABLE IF NOT EXISTS outbox (
    id UUID PRIMARY KEY,
    created_at TIMESTAMP(3) NOT NULL,
    scheduled_at TIMESTAMP(3) NOT NULL,
    metadata BLOB,
    payload BLOB NOT NULL,
    times_attempted INT NOT NULL
);

CREATE INDEX idx_outbox_created_at ON outbox (created_at);
CREATE INDEX idx_outbox_scheduled_at ON outbox (scheduled_at);
```

```sql
CREATE TABLE IF NOT EXISTS outbox (
    id TEXT PRIMARY KEY,
    created_at DATETIME NOT NULL,
    scheduled_at DATETIME NOT NULL,
    metadata BLOB,
    payload BLOB NOT NULL,
    times_attempted INTEGER NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_outbox_created_at ON outbox (created_at);
CREATE INDEX IF NOT EXISTS idx_outbox_scheduled_at ON outbox (scheduled_at);
```

```sql
CREATE TABLE outbox (
    id RAW(16) PRIMARY KEY,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
    scheduled_at TIMESTAMP WITH TIME ZONE NOT NULL,
    metadata BLOB,
    payload BLOB NOT NULL,
    times_attempted NUMBER(10) NOT NULL
);

CREATE INDEX idx_outbox_created_at ON outbox (created_at);
CREATE INDEX idx_outbox_scheduled_at ON outbox (scheduled_at);
```

```sql
CREATE TABLE outbox (
    id UNIQUEIDENTIFIER PRIMARY KEY,
    created_at DATETIMEOFFSET(3) NOT NULL,
    scheduled_at DATETIMEOFFSET(3) NOT NULL,
    metadata VARBINARY(MAX),
    payload VARBINARY(MAX) NOT NULL,
    times_attempted INT NOT NULL
);

CREATE INDEX idx_outbox_created_at ON outbox (created_at);
CREATE INDEX idx_outbox_scheduled_at ON outbox (scheduled_at);
```

--------------------------------

### Write Multiple Messages (Library Managed Transactions)

Source: https://github.com/oagudo/outbox/blob/main/README.md

Uses the library-managed transaction model to conditionally update an entity and store one or more outbox messages. The transaction is automatically rolled back if the callback function returns an error.

```go
// Use Write function for conditional or multiple message publishing.
// If the callback returns an error the transaction is rolled back
err = writer.Write(ctx,
        func(ctx context.Context, tx outbox.TxQueryer, msgWriter outbox.MessageWriter) error {

    result, err := tx.ExecContext(ctx,
        "UPDATE orders SET status = 'confirmed' WHERE id = $1 AND status = 'pending'", id)
    if err != nil {
        return err
    }
    if rows, _ := result.RowsAffected(); rows == 0 {
        return ErrOrderNotPending // no message, rollback
    }

    // Create and store outbox messages
    payload, _ := json.Marshal(order)
    msg := outbox.NewMessage(order.Payload,
        outbox.WithCreatedAt(order.CreatedAt),
        outbox.WithMetadata(json.RawMessage(`{"trace_id":"abc123"}`)))

    return msgWriter.Store(ctx, msg)
})

```

--------------------------------

### Monitor Reader Errors and Discarded Messages

Source: https://context7.com/oagudo/outbox/llms.txt

Shows how to monitor errors during message publishing, updating, reading, and deletion, as well as how to track messages that have exceeded the maximum retry attempts (poison messages). Connect these channels to your logging, metrics, and alerting systems.

```go
package main

import (
    "context"
    "database/sql"
    "log"
    "time"
    "github.com/oagudo/outbox"
    _ "github.com/jackc/pgx/v5/stdlib"
)

type MyPublisher struct{}

func (p *MyPublisher) Publish(ctx context.Context, msg *outbox.Message) error {
    return nil
}

func main() {
    db, _ := sql.Open("pgx", "postgres://user:pass@localhost:5432/mydb?sslmode=disable")
    defer db.Close()

    dbCtx := outbox.NewDBContext(db, outbox.SQLDialectPostgres)
    reader := outbox.NewReader(dbCtx, &MyPublisher{},
        outbox.WithMaxAttempts(50),
        outbox.WithExponentialDelay(200*time.Millisecond, 1*time.Hour),
    )

    // Monitor errors - connect to your logging/metrics system
    go func() {
        for err := range reader.Errors() {
            switch e := err.(type) {
            case *outbox.PublishError:
                log.Printf("PUBLISH_ERROR | ID: %s | Attempts: %d | Error: %v",
                    e.Message.ID, e.Message.TimesAttempted, e.Err)
                // metrics.Increment("outbox.publish.error")

            case *outbox.UpdateError:
                log.Printf("UPDATE_ERROR | ID: %s | Error: %v",
                    e.Message.ID, e.Err)
                // metrics.Increment("outbox.update.error")

            case *outbox.DeleteError:
                log.Printf("DELETE_ERROR | Count: %d | Error: %v",
                    len(e.Messages), e.Err)
                for _, msg := range e.Messages {
                    log.Printf("  - Failed to delete: %s", msg.ID)
                }
                // metrics.Increment("outbox.delete.error")

            case *outbox.ReadError:
                log.Printf("READ_ERROR | Error: %v", e.Err)
                // metrics.Increment("outbox.read.error")

            default:
                log.Printf("UNKNOWN_ERROR | Error: %v", e)
            }
        }
    }()

    // Monitor poison messages (exceeded max attempts)
    go func() {
        for msg := range reader.DiscardedMessages() {
            log.Printf("POISON_MESSAGE | ID: %s | Attempts: %d | Payload: %s",
                msg.ID, msg.TimesAttempted, string(msg.Payload))

            // Forward to dead-letter queue
            // deadLetterQueue.Send(msg)

            // Alert operations team
            // alerting.SendAlert("Outbox message discarded", msg.ID)

            // metrics.Increment("outbox.message.discarded")
        }
    }()

    reader.Start()
    defer reader.Stop(context.Background())

    // Application runs...
    select {}
}
```

--------------------------------

### Write Single Message (Library Managed Transactions)

Source: https://github.com/oagudo/outbox/blob/main/README.md

A simplified library-managed transaction for storing a single outbox message unconditionally. The transaction is handled by the writer, and any error during the database operation will cause a rollback.

```go
// Use WriteOne function for simple cases that store a single message unconditionally
msg := outbox.NewMessage(payload)
err = writer.WriteOne(ctx, msg, func(ctx context.Context, tx outbox.TxQueryer) error {
    _, err := tx.ExecContext(ctx,
        "INSERT INTO entity (id, created_at) VALUES ($1, $2)",
        entity.ID, entity.CreatedAt)
    return err
})

```

--------------------------------

### Manage Unmanaged Transactions

Source: https://context7.com/oagudo/outbox/llms.txt

Use UnmanagedWriter to integrate with existing transaction management patterns where you control the commit lifecycle. Optimistic publishing is not triggered for these transactions.

```go
package main

import (
    "context"
    "database/sql"
    "encoding/json"
    "time"
    "github.com/google/uuid"
    "github.com/oagudo/outbox"
    _ "github.com/jackc/pgx/v5/stdlib"
)

func main() {
    db, _ := sql.Open("pgx", "postgres://user:pass@localhost:5432/mydb?sslmode=disable")
    defer db.Close()

    dbCtx := outbox.NewDBContext(db, outbox.SQLDialectPostgres)
    writer := outbox.NewWriter(dbCtx)

    ctx := context.Background()

    // Begin your own transaction
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        panic(err)
    }
    defer tx.Rollback() // Rollback if not committed

    // Perform business operations
    entity := map[string]interface{}{
        "id":         uuid.New(),
        "created_at": time.Now().UTC(),
    }
    _, err = tx.ExecContext(ctx,
        "INSERT INTO entities (id, created_at) VALUES ($1, $2)",
        entity["id"], entity["created_at"],
    )
    if err != nil {
        panic(err)
    }

    // Store outbox message using unmanaged writer
    payload, _ := json.Marshal(entity)
    msg := outbox.NewMessage(payload)

    unmanagedWriter := writer.Unmanaged()
    err = unmanagedWriter.Store(ctx, tx, msg)
    if err != nil {
        panic(err)
    }

    // You control when to commit
    err = tx.Commit()
    if err != nil {
        panic(err)
    }
    // Note: Optimistic publishing is not triggered for unmanaged transactions
}
```

--------------------------------

### Monitor Reader Errors and Discarded Messages

Source: https://github.com/oagudo/outbox/blob/main/README.md

Sets up goroutines to handle processing errors and track messages that have exceeded maximum retry attempts.

```go
// Monitor processing errors
go func() {
    for err := range reader.Errors() {
        switch e := 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 := range e.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 poison messages (exceeded max attempts)
go func() {
    for msg := range reader.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
    }
}()
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.