Saga Pattern with Temporal: A Fault-Tolerant Orchestration Deep Dive

// file: shared/definitions.go
package shared

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

// OrderDetails contains all the information for the order.
type OrderDetails struct {
	UserID    string
	ItemID    string
	Quantity  int
	Price     float64
	OrderID   string
}

// PaymentDetails contains payment processing info.
type PaymentDetails struct {
	CardToken string
	Amount    float64
}

// ShipmentDetails contains shipping info.
type ShipmentDetails struct {
	OrderID string
	Address string
}

// OrderProcessingWorkflow defines the saga's orchestration logic.
func OrderProcessingWorkflow(ctx workflow.Context, order OrderDetails) (string, error) {
    // Workflow implementation will go here
}

// file: activities/inventory_activity.go
package activities

import (
	"context"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/my-app/shared"
)

// Simulating an inventory service client

func ReserveInventory(ctx context.Context, order shared.OrderDetails) (string, error) {
	fmt.Printf("Reserving %d of item %s for order %s\n", order.Quantity, order.ItemID, order.OrderID)
	// In a real app, this would call the inventory service via gRPC/REST.
	// Simulate some work.
	time.Sleep(50 * time.Millisecond)

	// Return a unique reservation ID for compensation purposes.
	reservationID := uuid.New().String()
	fmt.Printf("Inventory reserved with ID: %s\n", reservationID)
	return reservationID, nil
}

func ReleaseInventory(ctx context.Context, reservationID string, orderID string) error {
	fmt.Printf("COMPENSATION: Releasing inventory for reservation %s (order %s)\n", reservationID, orderID)
	// Call inventory service to release the hold.
	time.Sleep(50 * time.Millisecond)
	return nil
}

// file: activities/payment_activity.go
package activities

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/my-app/shared"
	"go.temporal.io/sdk/activity"
	"go.temporal.io/sdk/temporal"
)

func ProcessPayment(ctx context.Context, payment shared.PaymentDetails, orderID string) (string, error) {
	logger := activity.GetLogger(ctx)
	logger.Info("Processing payment", "OrderID", orderID, "Amount", payment.Amount)

    // Example of a business logic failure vs a transient one
    if payment.CardToken == "tok_insufficient_funds" {
        // This is a non-retryable application error.
        return "", temporal.NewApplicationError("Insufficient funds", "INSUFFICIENT_FUNDS")
    }
    if payment.CardToken == "tok_payment_gateway_timeout" {
        // This is a transient error that should be retried.
        return "", errors.New("payment gateway timed out")
    }

	time.Sleep(100 * time.Millisecond)
	transactionID := "txn_" + uuid.New().String()
	logger.Info("Payment successful", "TransactionID", transactionID)
	return transactionID, nil
}

func RefundPayment(ctx context.Context, transactionID string, orderID string) error {
	logger := activity.GetLogger(ctx)
	logger.Info("COMPENSATION: Refunding payment", "TransactionID", transactionID, "OrderID", orderID)
	time.Sleep(100 * time.Millisecond)
	return nil
}

// file: workflow/order_workflow.go
package workflow

import (
	"time"

	"github.com/my-app/activities"
	"github.com/my-app/shared"
	"go.temporal.io/sdk/temporal"
	"go.temporal.io/sdk/workflow"
)

func OrderProcessingWorkflow(ctx workflow.Context, order shared.OrderDetails) (string, error) {
	logger := workflow.GetLogger(ctx)
	logger.Info("Order processing workflow started", "OrderID", order.OrderID)

	// Configure activity options with timeouts.
	// These are crucial for production systems.
	ao := workflow.ActivityOptions{
		StartToCloseTimeout: 10 * time.Second,
	}
	ctx = workflow.WithActivityOptions(ctx, ao)

	// 1. Reserve Inventory
	var reservationID string
	var compensations []func()
	
	// Use a separate child context for the main logic path.
	// This allows the defer functions to run even if the main path context is canceled.
	childCtx, cancelHandler := workflow.NewDisconnectedContext(ctx)

	var workflowError error
	defer func() {
		// If the main workflow logic failed, run all compensations.
		if workflowError != nil {
			logger.Warn("Workflow failed, executing compensations.", "Error", workflowError)
			// Run compensations in reverse order.
			for i := len(compensations) - 1; i >= 0; i-- {
				compensations[i]()
			}
		}
	}()

	err := workflow.ExecuteActivity(childCtx, activities.ReserveInventory, order).Get(childCtx, &reservationID)
	if err != nil {
		logger.Error("Failed to reserve inventory", "Error", err)
		workflowError = err
		return "", err
	}
	compensations = append(compensations, func() {
		// Use a new context for compensation to ensure it runs even if the main context was canceled.
		compensationCtx, _ := workflow.NewDisconnectedContext(ctx)
		compensationCtx = workflow.WithActivityOptions(compensationCtx, workflow.ActivityOptions{
			StartToCloseTimeout: 10 * time.Second,
		})
		wfErr := workflow.ExecuteActivity(compensationCtx, activities.ReleaseInventory, reservationID, order.OrderID).Get(compensationCtx, nil)
        if wfErr != nil {
            logger.Error("Failed to execute ReleaseInventory compensation", "Error", wfErr)
        }
	})

	// 2. Process Payment
	paymentDetails := shared.PaymentDetails{CardToken: "tok_visa", Amount: order.Price}
	var transactionID string
	err = workflow.ExecuteActivity(childCtx, activities.ProcessPayment, paymentDetails, order.OrderID).Get(childCtx, &transactionID)
	if err != nil {
		logger.Error("Failed to process payment", "Error", err)
		workflowError = err
		return "", err
	}
	compensations = append(compensations, func() {
		compensationCtx, _ := workflow.NewDisconnectedContext(ctx)
		compensationCtx = workflow.WithActivityOptions(compensationCtx, workflow.ActivityOptions{
			StartToCloseTimeout: 10 * time.Second,
		})
		wfErr := workflow.ExecuteActivity(compensationCtx, activities.RefundPayment, transactionID, order.OrderID).Get(compensationCtx, nil)
        if wfErr != nil {
            logger.Error("Failed to execute RefundPayment compensation", "Error", wfErr)
        }
	})

	// 3. Create Shipment
	shipmentDetails := shared.ShipmentDetails{OrderID: order.OrderID, Address: "123 Temporal Lane"}
	var shipmentID string
	err = workflow.ExecuteActivity(childCtx, activities.CreateShipment, shipmentDetails).Get(childCtx, &shipmentID)
	if err != nil {
		logger.Error("Failed to create shipment", "Error", err)
		workflowError = err
		return "", err
	}
    // NOTE: We don't add a compensation for shipment itself until it's confirmed.
    // The compensation would be `CancelShipment`.

	// 4. Update Order Status
	err = workflow.ExecuteActivity(childCtx, activities.UpdateOrderStatus, order.OrderID, "COMPLETED").Get(childCtx, nil)
	if err != nil {
		// This is a critical failure. The order is paid and shipped but we can't update status.
		// We might need a human-in-the-loop via a different activity.
		logger.Error("CRITICAL: Failed to update order status post-shipment", "Error", err)
		workflowError = err
		return "", err
	}

	// 5. Send Confirmation Email (fire and forget)
	_ = workflow.ExecuteActivity(childCtx, activities.SendConfirmationEmail, order.OrderID)

	// Cancel the disconnected context to clean up resources.
	cancelHandler()

	logger.Info("Workflow completed successfully", "OrderID", order.OrderID)
	return "Order Completed: " + order.OrderID, nil
}

// Inside the workflow
	paymentIdempotencyKey := workflow.GetInfo(ctx).WorkflowExecution.ID + "-payment"
	err = workflow.ExecuteActivity(childCtx, activities.ProcessPayment, paymentDetails, order.OrderID, paymentIdempotencyKey).Get(childCtx, &transactionID)

// file: activities/payment_activity_idempotent.go

// The activity signature now accepts the idempotency key.
func ProcessPaymentIdempotent(ctx context.Context, payment shared.PaymentDetails, orderID string, idempotencyKey string) (string, error) {
	logger := activity.GetLogger(ctx)
	logger.Info("Processing payment with idempotency key", "Key", idempotencyKey)

    // 1. Check if this operation has already been completed.
    // In a real system, you'd query a database table like `processed_transactions`.
    // db.Query("SELECT transaction_id FROM processed_transactions WHERE idempotency_key = ?", idempotencyKey)
    // if transaction_id exists, return it immediately without reprocessing.

	// ... (payment processing logic as before)

    // 2. After successful payment, atomically store the result with the key.
    // db.Exec("INSERT INTO processed_transactions (idempotency_key, transaction_id) VALUES (?, ?)", idempotencyKey, transactionID)

	return transactionID, nil
}

// ... inside ProcessPayment activity
if payment.CardToken == "tok_insufficient_funds" {
    // This is a non-retryable application error.
    // The first argument is the message, the second is the error type.
    return "", temporal.NewApplicationError("Insufficient funds", "INSUFFICIENT_FUNDS")
}

// file: activities/long_running_activity.go

func ComplexShipmentCreation(ctx context.Context, details shared.ShipmentDetails) (string, error) {
    logger := activity.GetLogger(ctx)

    // Step 1: Validate address (1 minute)
    time.Sleep(1 * time.Minute)
    activity.RecordHeartbeat(ctx, "Address validated")

    // Step 2: Find best carrier (2 minutes)
    time.Sleep(2 * time.Minute)
    activity.RecordHeartbeat(ctx, "Carrier found")

    // Step 3: Book with carrier API (can be slow)
    time.Sleep(5 * time.Minute)
    activity.RecordHeartbeat(ctx, "Booked with carrier")

    logger.Info("Shipment created")
    return "shipment_abc123", nil
}

// file: workflow/order_workflow_test.go
package workflow

import (
	"errors"
	"testing"

	"github.com/my-app/activities"
	"github.com/my-app/shared"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"go.temporal.io/sdk/testsuite"
)

func TestOrderWorkflow_ShipmentFailure(t *testing.T) {
	// Set up the test suite and environment
	ts := &testsuite.WorkflowTestSuite{}
	env := ts.NewTestWorkflowEnvironment()

	// Mock activities
	env.OnActivity(activities.ReserveInventory, mock.Anything, mock.Anything).Return("res-123", nil)
	env.OnActivity(activities.ProcessPayment, mock.Anything, mock.Anything, mock.Anything).Return("txn-456", nil)
	
	// Mock the failure
	env.OnActivity(activities.CreateShipment, mock.Anything, mock.Anything).Return("", errors.New("shipping provider API is down"))

	// Mock the expected compensation calls
	env.OnActivity(activities.RefundPayment, mock.Anything, "txn-456", "order-abc").Return(nil).Once()
	env.OnActivity(activities.ReleaseInventory, mock.Anything, "res-123", "order-abc").Return(nil).Once()

	// Execute the workflow
	order := shared.OrderDetails{OrderID: "order-abc"}
	env.ExecuteWorkflow(OrderProcessingWorkflow, order)

	// Assertions
	require.True(t, env.IsWorkflowCompleted())
	require.Error(t, env.GetWorkflowError())

	// Verify that all mocked activities were called as expected
	env.AssertExpectations(t)
}