Saga Pattern in Go — Temporal, Outbox Pattern & Debezium

package saga

import (
    "fmt"
    "time"

    "go.temporal.io/sdk/temporal"
    "go.temporal.io/sdk/workflow"
)

type OrderSagaInput struct {
    OrderID  string
    UserID   string
    Items    []OrderItem
    Amount   float64
    Currency string
}

type OrderItem struct {
    ProductID string
    Quantity  int
}

// OrderSagaWorkflow orchestrates the full order fulfillment saga
func OrderSagaWorkflow(ctx workflow.Context, input OrderSagaInput) error {
    activityOpts := workflow.ActivityOptions{
        StartToCloseTimeout: 30 * time.Second,
        RetryPolicy: &temporal.RetryPolicy{
            MaximumAttempts:    5,
            InitialInterval:    time.Second,
            MaximumInterval:    30 * time.Second,
            BackoffCoefficient: 2.0,
            // Do NOT retry on business errors — only on transient failures
            NonRetryableErrorTypes: []string{"PAYMENT_DECLINED", "INVENTORY_PERMANENTLY_UNAVAILABLE"},
        },
    }
    ctx = workflow.WithActivityOptions(ctx, activityOpts)

    var saga workflow.Saga
    saga.SetParallelCompensation(false) // Sequential compensation (LIFO order)

    // ─── Step 1: Create Order ─────────────────────────────────────────────
    var orderResult CreateOrderResult
    if err := workflow.ExecuteActivity(ctx, CreateOrderActivity, input).Get(ctx, &orderResult); err != nil {
        return fmt.Errorf("create order: %w", err)
    }
    // Register compensation IMMEDIATELY after each successful step
    saga.AddCompensation(CancelOrderActivity, orderResult.OrderID)

    // ─── Step 2: Reserve Inventory ────────────────────────────────────────
    var reserveResult ReserveInventoryResult
    if err := workflow.ExecuteActivity(ctx, ReserveInventoryActivity, orderResult.OrderID, input.Items).Get(ctx, &reserveResult); err != nil {
        saga.Compensate(ctx) // Triggers: CancelOrderActivity
        return fmt.Errorf("reserve inventory: %w", err)
    }
    saga.AddCompensation(ReleaseInventoryActivity, reserveResult.ReservationID)

    // ─── Step 3: Process Payment ──────────────────────────────────────────
    var paymentResult ProcessPaymentResult
    if err := workflow.ExecuteActivity(ctx, ProcessPaymentActivity, orderResult.OrderID, input.Amount).Get(ctx, &paymentResult); err != nil {
        saga.Compensate(ctx) // Triggers LIFO: ReleaseInventoryActivity → CancelOrderActivity
        return fmt.Errorf("payment: %w", err)
    }
    saga.AddCompensation(RefundPaymentActivity, paymentResult.TransactionID)

    // ─── Step 4: Notify Fulfillment ───────────────────────────────────────
    if err := workflow.ExecuteActivity(ctx, NotifyFulfillmentActivity, orderResult.OrderID).Get(ctx, nil); err != nil {
        saga.Compensate(ctx) // Triggers LIFO: RefundPaymentActivity → ReleaseInventory → CancelOrder
        return fmt.Errorf("fulfillment notification: %w", err)
    }

    workflow.GetLogger(ctx).Info("Order saga completed", "orderID", orderResult.OrderID)
    return nil
}

// ─── Activity stubs (must be registered on a Temporal Worker) ─────────────

type CreateOrderResult struct{ OrderID string }
type ReserveInventoryResult struct{ ReservationID string }
type ProcessPaymentResult struct{ TransactionID string }

func CreateOrderActivity(input OrderSagaInput) (CreateOrderResult, error) {
    // INSERT INTO orders ... ON CONFLICT DO NOTHING (idempotent)
    return CreateOrderResult{OrderID: "order-uuid"}, nil
}

func CancelOrderActivity(orderID string) error {
    // UPDATE orders SET status='cancelled' WHERE id=orderID (idempotent)
    return nil
}

func ReserveInventoryActivity(orderID string, items []OrderItem) (ReserveInventoryResult, error) {
    return ReserveInventoryResult{ReservationID: fmt.Sprintf("res-%s", orderID)}, nil
}

func ReleaseInventoryActivity(reservationID string) error {
    // UPDATE inventory_reservations SET status='released' WHERE id=reservationID (idempotent)
    return nil
}

func ProcessPaymentActivity(orderID string, amount float64) (ProcessPaymentResult, error) {
    return ProcessPaymentResult{TransactionID: fmt.Sprintf("tx-%s", orderID)}, nil
}

func RefundPaymentActivity(transactionID string) error {
    // POST /v1/refunds {transaction_id: transactionID} (idempotent via idempotency key)
    return nil
}

func NotifyFulfillmentActivity(orderID string) error {
    return nil
}

-- Outbox table
CREATE TABLE outbox_table (
    id             UUID         NOT NULL DEFAULT gen_random_uuid(),
    aggregate_type VARCHAR(100) NOT NULL,  -- e.g., 'order', 'payment'
    aggregate_id   VARCHAR(255) NOT NULL,  -- e.g., order UUID
    event_type     VARCHAR(100) NOT NULL,  -- e.g., 'ORDER_CREATED'
    payload        JSONB        NOT NULL,
    created_at     TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
    PRIMARY KEY (id)
);

// Application code — atomic write: order + outbox event in one transaction
func (s *OrderService) CreateOrder(ctx context.Context, userID string, amount float64) (string, error) {
    tx, err := s.db.BeginTx(ctx, nil)
    if err != nil {
        return "", err
    }
    defer tx.Rollback()

    // 1. Insert business record
    var orderID string
    err = tx.QueryRowContext(ctx,
        `INSERT INTO orders (user_id, amount, status) VALUES ($1, $2, 'pending') RETURNING id`,
        userID, amount,
    ).Scan(&orderID)
    if err != nil {
        return "", err
    }

    // 2. Insert outbox event — SAME TRANSACTION
    payload, _ := json.Marshal(map[string]interface{}{
        "order_id": orderID, "user_id": userID, "amount": amount,
    })
    _, err = tx.ExecContext(ctx,
        `INSERT INTO outbox_table (aggregate_type, aggregate_id, event_type, payload)
         VALUES ('order', $1, 'ORDER_CREATED', $2)`,
        orderID, payload,
    )
    if err != nil {
        return "", err
    }

    // 3. Both records committed atomically — Debezium will pick up the outbox row
    return orderID, tx.Commit()
}

{
  "name": "postgres-outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres-db.internal",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${file:/secrets/debezium.properties:db.password}",
    "database.dbname": "orders_db",
    "database.server.name": "orders-dbserver",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_outbox_slot",
    "table.include.list": "public.outbox_table",
    "tombstones.on.delete": "false",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.table.field.event.id": "id",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.table.field.event.payload": "payload",
    "transforms.outbox.route.by.field": "aggregate_type",
    "transforms.outbox.route.topic.replacement": "events.${routedByValue}",
    "transforms.outbox.table.expand.json.payload": "true",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}