Saga Pattern Implementation with Temporal for Resilient Microservices

// file: shared/types.go
package shared

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

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

// OrderProcessingWorkflow is the entry point for our Saga.
func OrderProcessingWorkflow(ctx workflow.Context, order OrderDetails) (string, error) {
	// Workflow implementation here...
}

// Activities defines the contract for our microservice interactions.
type Activities struct{}

func (a *Activities) ReserveInventory(ctx context.Context, itemID string, quantity int) (string, error) { /* ... */ }
func (a *Activities) ProcessPayment(ctx context.Context, userID string, amount float64, idempotencyKey string) (string, error) { /* ... */ }
func (a *Activities) ShipOrder(ctx context.Context, orderID string) (string, error) { /* ... */ }
func (a *Activities) MarkOrderAsCompleted(ctx context.Context, orderID string) error { /* ... */ }

// Compensation Activities
func (a *Activities) UnreserveInventory(ctx context.Context, reservationID string) error { /* ... */ }
func (a *Activities) RefundPayment(ctx context.Context, transactionID string, idempotencyKey string) error { /* ... */ }

// file: workflow/workflow.go
package workflow

import (
	"time"

	"go.temporal.io/sdk/workflow"
	"example.com/saga/shared"
)

func OrderProcessingWorkflow(ctx workflow.Context, order shared.OrderDetails) (string, error) {
	// Set aggressive timeouts for activities. In a real app, these would be configurable.
	options := workflow.ActivityOptions{
		StartToCloseTimeout: time.Second * 10,
	}
	ctx = workflow.WithActivityOptions(ctx, options)

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

	var a *shared.Activities
	var compensationFuncs []func()

	// Use defer to run all registered compensations on any workflow failure.
	// Note: The deferred function captures the 'compensationFuncs' slice by reference.
	defer func() {
		if workflow.IsCancelRequested(ctx) || (workflow.GetInfo(ctx).GetContinueAsNewSuggested() && workflow.IsReplaying()) {
			// Don't run compensations if the workflow is being cancelled or continued as new.
			// Cancellation handling would have its own logic.
			return
		}
		// If there's an error, run compensations in reverse order.
		if workflow.GetInfo(ctx).GetCurrentAttempt() > 0 || (len(compensationFuncs) > 0 && workflow.GetInfo(ctx).GetLastCompletionResult() == nil) {
			for i := len(compensationFuncs) - 1; i >= 0; i-- {
				compensationFuncs[i]()
			}
		}
	}()

	// 1. Reserve Inventory
	var reservationID string
	err := workflow.ExecuteActivity(ctx, a.ReserveInventory, order.ItemID, order.Quantity).Get(ctx, &reservationID)
	if err != nil {
		logger.Error("Failed to reserve inventory", "Error", err)
		return "", err
	}
	compensationFuncs = append(compensationFuncs, func() {
		logger.Info("Executing compensation: UnreserveInventory", "ReservationID", reservationID)
		_ = workflow.ExecuteActivity(ctx, a.UnreserveInventory, reservationID).Get(ctx, nil) // Best-effort compensation
	})

	// 2. Process Payment
	var transactionID string
	paymentIdempotencyKey := workflow.GetInfo(ctx).WorkflowExecution.ID + "-payment"
	err = workflow.ExecuteActivity(ctx, a.ProcessPayment, order.UserID, order.Price, paymentIdempotencyKey).Get(ctx, &transactionID)
	if err != nil {
		logger.Error("Failed to process payment", "Error", err)
		return "", err
	}
	compensationFuncs = append(compensationFuncs, func() {
		refundIdempotencyKey := workflow.GetInfo(ctx).WorkflowExecution.ID + "-refund"
		logger.Info("Executing compensation: RefundPayment", "TransactionID", transactionID)
		_ = workflow.ExecuteActivity(ctx, a.RefundPayment, transactionID, refundIdempotencyKey).Get(ctx, nil)
	})

	// 3. Ship Order
	var shipmentID string
	err = workflow.ExecuteActivity(ctx, a.ShipOrder, order.OrderID).Get(ctx, &shipmentID)
	if err != nil {
		logger.Error("Failed to ship order", "Error", err)
		return "", err
	}
	// Note: Shipping is often considered a point of no return. We won't add a compensation for it.
	// In a real-world scenario, this might involve a complex return process, which could be another saga.

	// 4. Mark Order as Completed
	err = workflow.ExecuteActivity(ctx, a.MarkOrderAsCompleted, order.OrderID).Get(ctx, nil)
	if err != nil {
		logger.Error("Failed to mark order as completed", "Error", err)
		// This is a critical failure. The order is shipped but not marked complete.
		// This might require manual intervention. We'll still fail the workflow to alert operators.
		return "", err
	}

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

// file: activities/activities.go
package activities

import (
	"context"
	"fmt"
	// ... other imports for your HTTP client, etc.
)

// Mock Payment Service Client
type PaymentServiceClient struct{}

func (c *PaymentServiceClient) Charge(ctx context.Context, userID string, amount float64, idempotencyKey string) (string, error) {
	// In a real implementation, this would be an HTTP/gRPC call.
	// The downstream service MUST use the idempotencyKey to de-duplicate requests.
	fmt.Printf("Calling Payment Service: Charge user %s for %.2f with key %s\n", userID, amount, idempotencyKey)
	// Simulate a transient failure to demonstrate retries.
	if shouldFailSometimes() {
		return "", fmt.Errorf("payment gateway timeout")
	}
	return "txn-" + idempotencyKey, nil
}

// ... other service clients

type Activities struct {
	PaymentClient *PaymentServiceClient
	// ... other clients
}

func (a *Activities) ProcessPayment(ctx context.Context, userID string, amount float64, idempotencyKey string) (string, error) {
	// The activity's only job is to call the downstream service with the key.
	// Temporal's retry policy, configured in the workflow, will handle retries.
	return a.PaymentClient.Charge(ctx, userID, amount, idempotencyKey)
}

// PSEUDOCODE for the Payment Service endpoint
func handleChargeRequest(request ChargeRequest) {
    // 1. Check a cache (e.g., Redis) or a dedicated DB table for the idempotency key.
    cachedResponse, err := cache.Get(request.IdempotencyKey)
    if err == nil && cachedResponse != nil {
        // Request has been processed before, return the stored result.
        return cachedResponse
    }

    // 2. If not in cache, begin a transaction.
    tx := db.Begin()
    
    // 3. Within the transaction, check a persistent table for the idempotency key.
    var existingTxn Transaction
    tx.Where("idempotency_key = ?", request.IdempotencyKey).First(&existingTxn)
    if existingTxn.ID != 0 {
        tx.Rollback()
        return existingTxn.Response // Return previous result
    }

    // 4. Key not found. This is a new request. Process the payment.
    paymentResult, err := paymentGateway.Charge(request.CardDetails, request.Amount)
    if err != nil {
        tx.Rollback()
        // Return an error that Temporal can use for retry decisions
        return InternalServerError(err)
    }

    // 5. Store the result and the idempotency key in the persistent table.
    newTxn := Transaction{
        IdempotencyKey: request.IdempotencyKey,
        Response:       paymentResult,
        Status:         "COMPLETED",
    }
    tx.Create(&newTxn)
    tx.Commit()

    // 6. Optionally, write the result to the cache for faster lookups.
    cache.Set(request.IdempotencyKey, paymentResult, 24 * time.Hour)

    return paymentResult
}

// In ProcessPayment Activity...
chargeResult, err := a.PaymentClient.Charge(...)
if err != nil {
    // Check for a specific application error that indicates fraud review
    if IsFraudReviewError(err) {
        // Return a special error type that the workflow can interpret.
        return "", temporal.NewApplicationError("Fraud review required", "FraudReviewRequired")
    }
    return "", err
}

// In OrderProcessingWorkflow...

// ... after trying to process payment
err = workflow.ExecuteActivity(ctx, a.ProcessPayment, ...).Get(ctx, &transactionID)
if err != nil {
    var appErr *temporal.ApplicationError
    if errors.As(err, &appErr) && appErr.Type() == "FraudReviewRequired" {
        logger.Info("Payment requires fraud review. Awaiting signal.")
        
        // Create a channel to receive the signal on.
        signalChan := workflow.GetSignalChannel(ctx, "fraud-review-signal")

        var signal struct {
            Approved bool
            Reason   string
        }

        // Wait for the signal. The workflow will be durable and idle here.
        // We can add a timer to timeout the wait.
        signalReceived := false
        timer := workflow.NewTimer(ctx, time.Hour*24*7) // 7-day timeout for review

        workflow.NewSelector(ctx). 
            AddReceive(signalChan, func(c workflow.ReceiveChannel, more bool) {
                c.Receive(ctx, &signal)
                signalReceived = true
            }).
            AddFuture(timer, func(f workflow.Future) {
                // Timer fired, do nothing, let the loop exit
            }).
            Select(ctx)

        if !signalReceived || !signal.Approved {
            logger.Warn("Fraud review timed out or was rejected", "Reason", signal.Reason)
            // This will trigger the compensation logic (refund, unreserve inventory).
            return "", fmt.Errorf("payment rejected in fraud review: %s", signal.Reason)
        }

        logger.Info("Fraud review approved. Continuing workflow.")
        // The payment is now considered complete, so we proceed.
        // We might need to get a new transaction ID from the signal payload.

    } else {
        logger.Error("Failed to process payment", "Error", err)
        return "", err
    }
}
// ... continue with ShipOrder, etc.

# Using the CLI
tctl workflow signal --workflow_id 'your-workflow-id' --name 'fraud-review-signal' --input '{"Approved": true, "Reason": "Manual review by operator Jane Doe"}'

// file: workflow/subscription.go
package workflow

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

type SubscriptionState struct {
	UserID      string
	BillingCycle int
	Amount      float64
}

func SubscriptionWorkflow(ctx workflow.Context, state SubscriptionState) error {
	logger := workflow.GetLogger(ctx)
	logger.Info("Subscription workflow started", "UserID", state.UserID, "Cycle", state.BillingCycle)

	// Sleep until it's time to bill. In a real app, this would be a calculated duration.
	awakeTime := workflow.Now(ctx).Add(30 * 24 * time.Hour)
	err := workflow.Sleep(ctx, awakeTime.Sub(workflow.Now(ctx)))
	if err != nil {
		return err // Handle cancellation
	}

	// Bill the user
	var a *shared.Activities
	err = workflow.ExecuteActivity(ctx, a.ProcessPayment, state.UserID, state.Amount, ...).Get(ctx, nil)
	if err != nil {
		// Handle payment failure (e.g., send dunning emails in another saga)
		logger.Error("Subscription payment failed", "Error", err)
		// Don't continue-as-new if payment fails
		return err
	}

	logger.Info("Billing successful for cycle", "Cycle", state.BillingCycle)

	// Prepare for the next cycle
	nextState := state
	nextState.BillingCycle++

	// Continue as a new workflow to keep history short
	return workflow.NewContinueAsNewError(ctx, SubscriptionWorkflow, nextState)
}