Saga Pattern with Temporal: A Production-Grade Implementation Guide

/order-saga
|-- /activities
|   |-- inventory_activity.go
|   |-- order_activity.go
|   |-- payment_activity.go
|   `-- shipping_activity.go
|-- /shared
|   `-- types.go
|-- /worker
|   `-- main.go
|-- /starter
|   `-- main.go
`-- workflow.go

// activities/payment_activity.go
package activities

import (
	"context"
	"fmt"
	"time"

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

// Simulating a call to a payment microservice
func ProcessPayment(ctx context.Context, orderID string, amount float64, idempotencyKey string) (string, error) {
	logger := activity.GetLogger(ctx)
	logger.Info("Processing payment", "OrderID", orderID, "Amount", amount)

	// In a real-world scenario, this would be an RPC call to the payment service.
	// The idempotencyKey would be passed in the request header/body.

	// Simulate a long-running API call that might time out
	for i := 0; i < 5; i++ {
		activity.RecordHeartbeat(ctx, i)
		time.Sleep(1 * time.Second)
	}

	// Simulate a potential failure
	if amount > 1000 {
		return "", fmt.Errorf("payment gateway declined for amount > 1000")
	}

	transactionID := fmt.Sprintf("txn-%s", idempotencyKey)
	logger.Info("Payment processed successfully", "TransactionID", transactionID)
	return transactionID, nil
}

func RefundPayment(ctx context.Context, transactionID string, idempotencyKey string) error {
	logger := activity.GetLogger(ctx)
	logger.Info("Refunding payment", "TransactionID", transactionID)

	// RPC call to the payment service to issue a refund.
	// The idempotencyKey ensures we don't double-refund.

	logger.Info("Payment refunded successfully")
	return nil
}

// workflow.go
package main

import (
	"time"

	"your_project/activities"
	"your_project/shared"

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

func OrderProcessingWorkflow(ctx workflow.Context, orderDetails shared.OrderDetails) (string, error) {
	// Configure activity options
	ao := workflow.ActivityOptions{
		StartToCloseTimeout: 10 * time.Second,
		// HeartbeatTimeout is crucial for long-running activities
		HeartbeatTimeout: 2 * time.Second, 
		RetryPolicy: &temporal.RetryPolicy{
			InitialInterval:    time.Second,
			BackoffCoefficient: 2.0,
			MaximumInterval:    100 * time.Second,
			MaximumAttempts:    5, // Or 0 for unlimited
		},
	}
	ctx = workflow.WithActivityOptions(ctx, ao)

	logger := workflow.GetLogger(ctx)
	logger.Info("Order processing workflow started", "OrderID", orderDetails.OrderID)

	// 1. Create Order in DB
	var orderID string
	err := workflow.ExecuteActivity(ctx, activities.CreateOrder, orderDetails).Get(ctx, &orderID)
	if err != nil {
		logger.Error("Failed to create order", "Error", err)
		return "", err
	}

	// Compensation stack
	var compensations []func()
	defer func() {
		// If the workflow failed (non-nil err), execute compensations
		if err != nil {
			logger.Warn("Workflow failed, starting compensation logic", "OrderID", orderID)
			for _, compensation := range compensations {
				// Execute compensation activities in a new, detached context
				// This ensures they run even if the main workflow context is cancelled
				disconnectedCtx, _ := workflow.NewDisconnectedContext(ctx)
				compensationCtx, _ := workflow.NewDisconnectedContext(ctx)
				err := workflow.ExecuteActivity(compensationCtx, compensation).Get(compensationCtx, nil)
				if err != nil {
					logger.Error("Compensation activity failed", "Error", err)
					// Handle non-compensatable errors here (e.g., signal for manual intervention)
				}
			}
		}
	}()

	// 2. Reserve Inventory
	err = workflow.ExecuteActivity(ctx, activities.ReserveInventory, orderID, orderDetails.Items).Get(ctx, nil)
	if err != nil {
		logger.Error("Failed to reserve inventory", "Error", err)
		// No compensation needed yet, as only the initial order was created.
		// We might update the order status to FAILED here.
		workflow.ExecuteActivity(ctx, activities.UpdateOrderStatus, orderID, "FAILED_INVENTORY_RESERVATION").Get(ctx, nil)
		return "", err
	}
	// Add compensation to the stack
	compensations = append(compensations, func() {
		workflow.ExecuteActivity(ctx, activities.ReleaseInventory, orderID, orderDetails.Items).Get(ctx, nil)
	})

	// 3. Process Payment
	var transactionID string
	idempotencyKey := workflow.GetInfo(ctx).WorkflowExecution.ID
	err = workflow.ExecuteActivity(ctx, activities.ProcessPayment, orderID, orderDetails.TotalAmount, idempotencyKey).Get(ctx, &transactionID)
	if err != nil {
		logger.Error("Failed to process payment", "Error", err)
		return "", err // Defer will trigger compensations
	}
	// Add compensation to the stack
	compensations = append(compensations, func() {
		refundIdempotencyKey := fmt.Sprintf("refund-%s", idempotencyKey)
		workflow.ExecuteActivity(ctx, activities.RefundPayment, transactionID, refundIdempotencyKey).Get(ctx, nil)
	})

	// 4. Update Order Status
	err = workflow.ExecuteActivity(ctx, activities.UpdateOrderStatus, orderID, "CONFIRMED").Get(ctx, nil)
	if err != nil {
		logger.Error("Failed to update order status", "Error", err)
		return "", err // Defer will trigger compensations
	}
	// No direct compensation for status update, as refunding is the business compensation.

	// 5. Dispatch Shipping
	err = workflow.ExecuteActivity(ctx, activities.DispatchShipping, orderID).Get(ctx, nil)
	if err != nil {
		logger.Error("Failed to dispatch shipping", "Error", err)
		// This is a partial failure. The order is paid. We don't want to refund yet.
		// We can update status and let another process handle it.
		workflow.ExecuteActivity(ctx, activities.UpdateOrderStatus, orderID, "PAYMENT_CONFIRMED_SHIPPING_FAILED").Get(ctx, nil)
		// We don't return the error here, so compensations are NOT triggered.
		// The workflow succeeds from a technical perspective.
		return "Order completed with shipping failure. Manual intervention required.", nil
	}

	logger.Info("Workflow completed successfully", "OrderID", orderID)
	return "Order processed successfully", nil
}

// Get a unique ID for this specific workflow execution
idempotencyKey := workflow.GetInfo(ctx).WorkflowExecution.ID
// Pass it to the activity
err = workflow.ExecuteActivity(ctx, activities.ProcessPayment, orderID, orderDetails.TotalAmount, idempotencyKey).Get(ctx, &transactionID)

func (s *PaymentService) ChargeCard(request ChargeRequest) (*ChargeResponse, error) {
    // 1. Check if we've seen this idempotency key before
    dbTx, err := s.db.Begin()
    existingResult, err := dbTx.QueryRow("SELECT result FROM processed_transactions WHERE key = ? FOR UPDATE", request.IdempotencyKey)
    if err == nil && existingResult != nil {
        // We've already processed this. Return the stored result.
        return existingResult, nil
    }
    if err != sql.ErrNoRows {
        return nil, err // Some other DB error
    }

    // 2. Not seen before. Process the payment.
    gatewayResponse, err := s.paymentGateway.Charge(request.CardDetails, request.Amount)
    if err != nil {
        // Store the failure result so we don't retry a failing card indefinitely
        s.storeResult(dbTx, request.IdempotencyKey, "FAILED")
        dbTx.Commit()
        return nil, err
    }

    // 3. Store the successful result before returning
    s.storeResult(dbTx, request.IdempotencyKey, gatewayResponse.TransactionID)
    dbTx.Commit()

    return &ChargeResponse{TransactionID: gatewayResponse.TransactionID}, nil
}

// ... inside the defer block ...
for _, compensation := range compensations {
    compensationCtx, _ := workflow.NewDisconnectedContext(ctx)
    // Use a specific retry policy for compensations. Maybe fewer retries.
    compensationAo := workflow.ActivityOptions{StartToCloseTimeout: 30 * time.Second}
    compensationCtx = workflow.WithActivityOptions(compensationCtx, compensationAo)

    err := workflow.ExecuteActivity(compensationCtx, compensation).Get(compensationCtx, nil)
    if err != nil {
        logger.Error("CRITICAL: Compensation activity failed. Manual intervention required.", "Error", err)
        // The workflow will now fail and remain in a failed state in the Temporal UI.
        // An external monitoring system should alert on workflows failing in this specific state.
        // Alternatively, we can use Signals to attempt a fix.
    }
}

// Add this to your workflow logic

// A channel to receive the signal
retrySignalChan := workflow.GetSignalChannel(ctx, "RetryCompensationSignal")

// ... inside the compensation defer block's error handling ...
if err != nil {
    logger.Error("Compensation failed, awaiting manual signal to retry...", "Error", err)
    // Block until a signal is received
    retrySignalChan.Receive(ctx, nil)
    logger.Info("Retry signal received, re-attempting compensation.")
    // Re-run the failed activity
    workflow.ExecuteActivity(compensationCtx, compensation).Get(compensationCtx, nil)
}

tctl workflow signal --workflow_id <your_workflow_id> --name RetryCompensationSignal

// ...
err = workflow.ExecuteActivity(ctx, activities.DispatchShipping, orderID).Get(ctx, nil)
if err != nil {
    logger.Error("Failed to dispatch shipping", "Error", err)
    // Update status to a specific error state
    workflow.ExecuteActivity(ctx, activities.UpdateOrderStatus, orderID, "PAYMENT_CONFIRMED_SHIPPING_FAILED").Get(ctx, nil)
    
    // DO NOT return the error. This prevents the compensation stack from running.
    return "Order completed with shipping failure. Manual intervention required.", nil
}

func SubscriptionWorkflow(ctx workflow.Context, subDetails Subscription) error {
    // Run this month's billing logic (a mini-Saga)
    err := runBillingCycle(ctx, subDetails)
    if err != nil {
        // Handle billing failure, maybe signal for dunning process
        return err
    }

    // Wait for the next billing cycle
    nextBillingDate := workflow.Now(ctx).Add(30 * 24 * time.Hour)
    await_ctx, _ := workflow.NewDisconnectedContext(ctx)
    err = workflow.Sleep(await_ctx, nextBillingDate.Sub(workflow.Now(ctx)))
	if err != nil {
		return err
	}

    // Instead of looping, continue as a new workflow
    return workflow.NewContinueAsNewError(ctx, SubscriptionWorkflow, subDetails)
}