The Unbreakable Transaction: The Outbox Pattern in Microservices

// DO NOT DO THIS. This is the dual-write anti-pattern.
func (s *OrderService) CreateOrder_Naive(ctx context.Context, orderDetails models.Order) (string, error) {
    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil {
        return "", fmt.Errorf("could not begin transaction: %w", err)
    }
    defer tx.Rollback() // Rollback on any error

    // 1. Write to the database
    var orderID string
    err = tx.QueryRowContext(ctx, 
        "INSERT INTO orders (customer_id, total_price, status) VALUES ($1, $2, 'PENDING') RETURNING id",
        orderDetails.CustomerID, orderDetails.TotalPrice).Scan(&orderID)
    if err != nil {
        return "", fmt.Errorf("could not insert order: %w", err)
    }

    // 2. Publish the event
    event := events.OrderCreated{
        OrderID:    orderID,
        CustomerID: orderDetails.CustomerID,
        TotalPrice: orderDetails.TotalPrice,
    }
    err = s.eventPublisher.Publish(ctx, "orders.created", event)
    if err != nil {
        // The transaction will be rolled back, but the event might have been published!
        // Or, more likely, this fails and the DB write is rolled back.
        return "", fmt.Errorf("could not publish event: %w", err)
    }

    // If both succeed, commit the transaction
    if err := tx.Commit(); err != nil {
        return "", fmt.Errorf("could not commit transaction: %w", err)
    }

    return orderID, nil
}

CREATE TABLE outbox (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    aggregate_type VARCHAR(255) NOT NULL,
    aggregate_id VARCHAR(255) NOT NULL,
    event_type VARCHAR(255) NOT NULL,
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    -- This column is crucial for the message relay to track progress.
    -- It's nullable to easily query for unprocessed events.
    processed_at TIMESTAMPTZ
);

-- An index is critical for the relay's polling query performance.
CREATE INDEX idx_outbox_unprocessed ON outbox (created_at) WHERE processed_at IS NULL;

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"

	"github.com/jmoiron/sqlx"
)

// Assume models and events packages are defined elsewhere

// OrderService with a database connection
type OrderService struct {
	db *sqlx.DB
}

// CreateOrder reliably creates an order and an outbox event in a single transaction.
func (s *OrderService) CreateOrder(ctx context.Context, orderDetails models.Order) (string, error) {
	tx, err := s.db.BeginTxx(ctx, nil)
	if err != nil {
		return "", fmt.Errorf("could not begin transaction: %w", err)
	}
	// Ensure rollback on any path that returns an error
	defer func() {
		if p := recover(); p != nil {
			tx.Rollback()
			panic(p) // re-panic after rollback
		} else if err != nil {
			tx.Rollback() // err is non-nil, rollback
		}
	}()

	// 1. Insert the business data
	var orderID string
	err = tx.QueryRowxContext(ctx,
		"INSERT INTO orders (customer_id, total_price, status) VALUES ($1, $2, 'PENDING') RETURNING id",
		orderDetails.CustomerID, orderDetails.TotalPrice).Scan(&orderID)
	if err != nil {
		return "", fmt.Errorf("could not insert order: %w", err)
	}

	// 2. Create the event payload
	event := events.OrderCreated{
		OrderID:    orderID,
		CustomerID: orderDetails.CustomerID,
		TotalPrice: orderDetails.TotalPrice,
	}
	payload, err := json.Marshal(event)
	if err != nil {
		return "", fmt.Errorf("could not marshal event payload: %w", err)
	}

	// 3. Insert the event into the outbox table
	_, err = tx.ExecContext(ctx,
		`INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
		 VALUES ('order', $1, 'OrderCreated', $2)`,
		orderID, payload)
	if err != nil {
		return "", fmt.Errorf("could not insert into outbox: %w", err)
	}

	// 4. Commit the single, atomic transaction
	err = tx.Commit()
	if err != nil {
		return "", fmt.Errorf("could not commit transaction: %w", err)
	}

	return orderID, nil
}

// MessageRelay polls the outbox table and publishes events.
type MessageRelay struct {
	db             *sqlx.DB
	eventPublisher EventPublisher // Interface to Kafka, RabbitMQ, etc.
	pollingInterval time.Duration
	batchSize      int
}

// Run starts the polling loop.
func (r *MessageRelay) Run(ctx context.Context) {
	ticker := time.NewTicker(r.pollingInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			log.Println("MessageRelay shutting down.")
			return
		case <-ticker.C:
			err := r.processOutboxBatch(ctx)
			if err != nil {
				log.Printf("Error processing outbox batch: %v", err)
			}
		}
	}
}

// processOutboxBatch fetches and processes a batch of events.
func (r *MessageRelay) processOutboxBatch(ctx context.Context) error {
	tx, err := r.db.BeginTxx(ctx, nil)
	if err != nil {
		return fmt.Errorf("could not begin transaction: %w", err)
	}
	defer tx.Rollback()

	// Advanced: Use FOR UPDATE SKIP LOCKED for concurrent relay instances
	rows, err := tx.QueryxContext(ctx,
		`SELECT id, aggregate_type, aggregate_id, event_type, payload
		 FROM outbox
		 WHERE processed_at IS NULL
		 ORDER BY created_at ASC
		 LIMIT $1
		 FOR UPDATE SKIP LOCKED`,
		r.batchSize)
	if err != nil {
		return fmt.Errorf("could not query outbox: %w", err)
	}
	defer rows.Close()

	var eventIDsToUpdate []uuid.UUID
	for rows.Next() {
		var event models.OutboxEvent
		if err := rows.StructScan(&event); err != nil {
			return fmt.Errorf("could not scan outbox event: %w", err)
		}

		// In a real implementation, you'd route based on event_type, etc.
		topic := fmt.Sprintf("%s.%s", event.AggregateType, event.EventType)
		
		// This is the critical step. Publish to the broker.
		if err := r.eventPublisher.Publish(ctx, topic, event.Payload); err != nil {
			log.Printf("Failed to publish event %s: %v. It will be retried.", event.ID, err)
			// If one message fails, we rollback the entire batch.
			// More advanced logic could selectively skip/DLQ messages.
			return fmt.Errorf("publishing failed: %w", err)
		}

		eventIDsToUpdate = append(eventIDsToUpdate, event.ID)
	}

	if len(eventIDsToUpdate) == 0 {
		return nil // No events to process
	}

	// Mark the processed events in the same transaction
	query, args, err := sqlx.In("UPDATE outbox SET processed_at = NOW() WHERE id IN (?)", eventIDsToUpdate)
	if err != nil {
		return fmt.Errorf("could not build update query: %w", err)
	}
	query = r.db.Rebind(query)
	_, err = tx.ExecContext(ctx, query, args...)
	if err != nil {
		return fmt.Errorf("could not update outbox events: %w", err)
	}

	return tx.Commit()
}

{
    "name": "outbox-connector",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "database.hostname": "postgres",
        "database.port": "5432",
        "database.user": "postgres",
        "database.password": "password",
        "database.dbname" : "orders_db",
        "database.server.name": "orders_server",
        "table.include.list": "public.outbox",
        "tombstones.on.delete": "false",
        // Unpack the JSON and promote headers
        "transforms": "outbox",
        "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
        "transforms.outbox.route.by.field": "aggregate_type",
        "transforms.outbox.route.topic.replacement": "${routedByValue}_events",
        "transforms.outbox.table.fields.additional.placement": "event_type:header:eventType"
    }
}

-- In the NotificationService's database
CREATE TABLE processed_events (
    event_id UUID PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

// NotificationService consumes events and sends notifications.
type NotificationService struct {
	db *sqlx.DB
	emailer EmailSender // Interface for sending emails
}

// HandleOrderCreated is the idempotent message handler.
func (s *NotificationService) HandleOrderCreated(ctx context.Context, event events.OrderCreated, eventID uuid.UUID) error {
	tx, err := s.db.BeginTxx(ctx, nil)
	if err != nil {
		return fmt.Errorf("could not begin transaction: %w", err)
	}
	defer tx.Rollback()

	// 1. Idempotency Check: Try to insert the event ID.
	// Using ON CONFLICT DO NOTHING is an efficient way to check for existence.
	// If the row already exists, the insert is a no-op.
	var alreadyProcessed bool
	err = tx.QueryRowxContext(ctx, 
		`INSERT INTO processed_events (event_id) VALUES ($1)
		 ON CONFLICT (event_id) DO NOTHING
		 RETURNING (xmax = 0)`,
		eventID).Scan(&alreadyProcessed)

	// A real insert sets xmax to a non-zero value. ON CONFLICT makes it 0.
	// So if xmax is 0, it means the row was already there.
	if err != nil && err != sql.ErrNoRows {
		return fmt.Errorf("idempotency check failed: %w", err)
	}
    
    // If the insert was a no-op (row existed), then we've seen this event.
	if !alreadyProcessed && err != sql.ErrNoRows {
		log.Printf("Event %s already processed, skipping.", eventID)
		return tx.Commit() // Commit the empty transaction and acknowledge the message
	}

	// 2. Business Logic: Only execute if the event is new.
	log.Printf("Processing new event %s for order %s", eventID, event.OrderID)
	err = s.emailer.SendOrderConfirmation(ctx, event.CustomerID, event.OrderID)
	if err != nil {
		// Don't commit the processed_events insert, so the message will be retried.
		return fmt.Errorf("failed to send email: %w", err)
	}

	// 3. Commit the transaction, atomically saving the business result (if any)
	// and the record of the processed event.
	return tx.Commit()
}