Resilient Choreographic Sagas with Kafka and the Outbox Pattern

-- OrderService Database Schema (PostgreSQL)

CREATE TABLE orders (
    id UUID PRIMARY KEY,
    customer_id UUID NOT NULL,
    total_amount DECIMAL(10, 2) NOT NULL,
    status VARCHAR(50) NOT NULL, -- e.g., 'PENDING', 'PAID', 'SHIPPED', 'CANCELLED'
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE outbox_events (
    id UUID PRIMARY KEY,
    aggregate_id UUID NOT NULL, -- The ID of the entity that was changed (e.g., order_id)
    aggregate_type VARCHAR(255) NOT NULL, -- The type of entity (e.g., 'Order')
    topic VARCHAR(255) NOT NULL, -- The Kafka topic to publish to
    payload JSONB NOT NULL, -- The event payload
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

// models.go
package main

import (
	"encoding/json"
	"time"

	"github.com/google/uuid"
)

type Order struct {
	ID          uuid.UUID `json:"id"`
	CustomerID  uuid.UUID `json:"customer_id"`
	TotalAmount float64   `json:"total_amount"`
	Status      string    `json:"status"`
}

type OrderCreatedEventPayload struct {
	OrderID     uuid.UUID `json:"order_id"`
	CustomerID  uuid.UUID `json:"customer_id"`
	TotalAmount float64   `json:"total_amount"`
}

// order_service.go
package main

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

	"github.com/google/uuid"
	_ "github.com/lib/pq"
)

func CreateOrder(ctx context.Context, db *sql.DB, order Order) error {
	// Start a new transaction
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	// Defer a rollback in case of error. The rollback will be ignored if the transaction is committed.
	defer tx.Rollback()

	// 1. Insert the order into the orders table
	order.Status = "PENDING"
	order.ID = uuid.New()

	_, err = tx.ExecContext(ctx, 
		`INSERT INTO orders (id, customer_id, total_amount, status) VALUES ($1, $2, $3, $4)`,
		order.ID, order.CustomerID, order.TotalAmount, order.Status,
	)
	if err != nil {
		return err
	}

	// 2. Create the event payload
	eventPayload := OrderCreatedEventPayload{
		OrderID:     order.ID,
		CustomerID:  order.CustomerID,
		TotalAmount: order.TotalAmount,
	}
	payloadBytes, err := json.Marshal(eventPayload)
	if err != nil {
		return err
	}

	// 3. Insert the event into the outbox_events table
	_, err = tx.ExecContext(ctx,
		`INSERT INTO outbox_events (id, aggregate_id, aggregate_type, topic, payload) VALUES ($1, $2, $3, $4, $5)`,
		uuid.New(), order.ID, "Order", "order.events.created", payloadBytes,
	)
	if err != nil {
		return err
	}

	// 4. Commit the transaction
	// If this fails, the defer Rollback() will handle cleanup.
	if err := tx.Commit(); err != nil {
		return err
	}

	log.Printf("Successfully created order %s and outbox event in a single transaction", order.ID)
	return nil
}

-- Add a 'processed_at' column to the outbox table to track status
ALTER TABLE outbox_events ADD COLUMN processed_at TIMESTAMPTZ NULL;
CREATE INDEX idx_outbox_events_unprocessed ON outbox_events (created_at) WHERE processed_at IS NULL;

// outbox_poller.go
package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"log"
	"time"

	"github.com/confluentinc/confluent-kafka-go/kafka"
	"github.com/google/uuid"
)

// Represents a row from our outbox_events table
type OutboxEvent struct {
	ID          uuid.UUID
	AggregateID uuid.UUID
	Topic       string
	Payload     []byte
}

func StartOutboxPoller(ctx context.Context, db *sql.DB, producer *kafka.Producer, pollInterval time.Duration) {
	ticker := time.NewTicker(pollInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			log.Println("Outbox poller shutting down.")
			return
		case <-ticker.C:
			if err := processOutboxEvents(db, producer); err != nil {
				log.Printf("Error processing outbox events: %v", err)
			}
		}
	}
}

func processOutboxEvents(db *sql.DB, producer *kafka.Producer) error {
	// Use a transaction to ensure we SELECT and UPDATE/DELETE atomically
	tx, err := db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()

	// Select a batch of unprocessed events, locking the rows to prevent other pollers from picking them up
	rows, err := tx.Query(`
		SELECT id, aggregate_id, topic, payload FROM outbox_events 
		WHERE processed_at IS NULL 
		ORDER BY created_at 
		LIMIT 100 
		FOR UPDATE SKIP LOCKED
	`)
	if err != nil {
		return err
	}
	defer rows.Close()

	var events []OutboxEvent
	for rows.Next() {
		var event OutboxEvent
		if err := rows.Scan(&event.ID, &event.AggregateID, &event.Topic, &event.Payload); err != nil {
			return err
		}
		events = append(events, event)
	}

	if len(events) == 0 {
		return tx.Commit() // Nothing to do
	}

	// Publish events to Kafka
	deliveryChan := make(chan kafka.Event)
	for _, event := range events {
		err := producer.Produce(&kafka.Message{
			TopicPartition: kafka.TopicPartition{Topic: &event.Topic, Partition: kafka.PartitionAny},
			Key:            []byte(event.AggregateID.String()), // Crucial for ordering per aggregate
			Value:          event.Payload,
		}, deliveryChan)

		if err != nil {
			log.Printf("Failed to produce message for event %s: %v", event.ID, err)
			// If one fails, we'll roll back and retry the whole batch later.
			return err
		}
	}

	// Wait for all messages to be delivered
	for i := 0; i < len(events); i++ {
		e := <-deliveryChan
		m := e.(*kafka.Message)
		if m.TopicPartition.Error != nil {
			log.Printf("Delivery failed for event: %v", m.TopicPartition.Error)
			return m.TopicPartition.Error
		}
	}

	// Mark events as processed
	for _, event := range events {
		_, err := tx.Exec(`UPDATE outbox_events SET processed_at = NOW() WHERE id = $1`, event.ID)
		if err != nil {
			return err
		}
	}

	return tx.Commit()
}

{
  "name": "order-outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres-orders-db",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "password",
    "database.dbname": "orders_db",
    "database.server.name": "orders-server",
    "table.include.list": "public.outbox_events",
    "tombstones.on.delete": "false",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.route.by.field": "aggregate_type",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.route.topic.replacement": "${routedByValue}.events.created"
  }
}

-- PaymentService Database Schema

CREATE TABLE payments (
    id UUID PRIMARY KEY,
    order_id UUID UNIQUE NOT NULL, -- Ensures we can't process the same order twice
    amount DECIMAL(10, 2) NOT NULL,
    status VARCHAR(50) NOT NULL, -- 'PROCESSED', 'FAILED'
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- A table to track processed event IDs for more general idempotency
CREATE TABLE processed_event_ids (
    event_id UUID PRIMARY KEY,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

// payment_consumer.go
package main

// Assumes incoming message has a unique EventID field in its header or payload
type OrderCreatedEvent struct {
    EventID     uuid.UUID `json:"event_id"`
    OrderID     uuid.UUID `json:"order_id"`
    CustomerID  uuid.UUID `json:"customer_id"`
    TotalAmount float64   `json:"total_amount"`
}

func (s *PaymentService) handleOrderCreated(ctx context.Context, event OrderCreatedEvent) error {
    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    // 1. Idempotency Check
    var exists bool
    err = tx.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM processed_event_ids WHERE event_id = $1)`, event.EventID).Scan(&exists)
    if err != nil {
        return err
    }
    if exists {
        log.Printf("Event %s already processed, skipping.", event.EventID)
        return tx.Commit() // Success, we've handled the duplicate
    }

    // 2. Business Logic: Process the payment
    paymentStatus := "PROCESSED" // Assume success for simplicity
    // In reality, call a payment gateway here. If it fails, set status to 'FAILED'.

    _, err = tx.ExecContext(ctx, 
        `INSERT INTO payments (id, order_id, amount, status) VALUES ($1, $2, $3, $4)`,
        uuid.New(), event.OrderID, event.TotalAmount, paymentStatus,
    )
    if err != nil {
        // This could be a unique constraint violation on order_id, which is another form of idempotency check
        log.Printf("Failed to insert payment for order %s: %v", event.OrderID, err)
        return err
    }

    // 3. Publish the next event in the saga (e.g., PaymentProcessed or PaymentFailed)
    // This would use the same outbox pattern within the PaymentService!
    if err := s.publishPaymentEvent(tx, event.OrderID, paymentStatus); err != nil {
        return err
    }

    // 4. Record the event ID as processed
    _, err = tx.ExecContext(ctx, `INSERT INTO processed_event_ids (event_id) VALUES ($1)`, event.EventID)
    if err != nil {
        return err
    }

    return tx.Commit()
}

// order_service_consumer.go

func (s *OrderService) handlePaymentFailed(ctx context.Context, event PaymentFailedEvent) error {
    // This handler must also be idempotent!
    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil {
        return err
    }
    defer tx.Rollback()

    // Idempotency check omitted for brevity

    // Compensating Transaction: Update the order status to CANCELLED
    result, err := tx.ExecContext(ctx, 
        `UPDATE orders SET status = 'CANCELLED' WHERE id = $1 AND status = 'PENDING'`,
        event.OrderID,
    )
    if err != nil {
        return err
    }

    rowsAffected, _ := result.RowsAffected()
    if rowsAffected == 0 {
        log.Printf("Order %s was not in PENDING state, compensation might not be needed.", event.OrderID)
        // This can happen in complex race conditions. It's important to handle gracefully.
    }

    // We might publish an OrderCancelled event here via the outbox
    // for other services (e.g., notifications) to react to.

    return tx.Commit()
}

// kafka_consumer_wrapper.go

func consumeLoop(consumer *kafka.Consumer, handler func(msg *kafka.Message) error, dlqProducer *kafka.Producer, dlqTopic string, maxRetries int) {
    for {
        msg, err := consumer.ReadMessage(-1)
        if err != nil { /* handle error */ continue }

        var attempt int
        for attempt < maxRetries {
            err = handler(msg)
            if err == nil {
                break // Success
            }
            attempt++
            log.Printf("Handler failed for message %s (attempt %d/%d): %v", msg.TopicPartition, attempt, maxRetries, err)
            time.Sleep(calculateBackoff(attempt)) // Exponential backoff
        }

        if err != nil {
            log.Printf("Message failed after %d retries. Moving to DLQ topic %s.", maxRetries, dlqTopic)
            // Forward to DLQ. Add retry count and error message to headers.
            msg.TopicPartition.Topic = &dlqTopic
            dlqProducer.Produce(msg, nil)
        }

        // Always commit the original offset so we don't get stuck
        consumer.CommitMessage(msg)
    }
}