Saga pattern

<?php

declare(strict_types=1);

namespace App\Saga\Order;

/**
 * Persistent saga state stored in DB.
 * Each step result is recorded for idempotency and recovery.
 */
enum OrderSagaStatus: string
{
    case Started = 'started';
    case PaymentCompleted = 'payment_completed';
    case StockReserved = 'stock_reserved';
    case ShippingCreated = 'shipping_created';
    case Completed = 'completed';
    case Compensating = 'compensating';
    case Failed = 'failed';
}

final class OrderSagaState
{
    public function __construct(
        public readonly string $sagaId,
        public readonly string $orderId,
        public OrderSagaStatus $status,
        public ?string $paymentId = null,
        public ?string $reservationId = null,
        public ?string $shipmentId = null,
        public ?string $failureReason = null,
        public int $version = 0,
    ) {}
}

<?php

declare(strict_types=1);

namespace App\Saga\Order;

use Symfony\Component\Messenger\MessageBusInterface;

/**
 * Order saga orchestrator. Runs locally transacted steps.
 * On failure triggers compensations in reverse order.
 */
final class OrderSagaOrchestrator
{
    public function __construct(
        private readonly OrderSagaRepository $repo,
        private readonly PaymentClient $payment,
        private readonly InventoryClient $inventory,
        private readonly ShippingClient $shipping,
        private readonly MessageBusInterface $bus,
    ) {}

    public function start(string $orderId, int $amountCents, string $sku, int $qty): string
    {
        $state = new OrderSagaState(
            sagaId: bin2hex(random_bytes(16)),
            orderId: $orderId,
            status: OrderSagaStatus::Started,
        );
        $this->repo->save($state);

        // Push first step as a message (async, retryable)
        $this->bus->dispatch(new ExecuteChargePayment($state->sagaId, $amountCents));

        return $state->sagaId;
    }

    public function handleChargePayment(string $sagaId, int $amountCents): void
    {
        $state = $this->repo->findOrFail($sagaId);
        if ($state->status !== OrderSagaStatus::Started) {
            return; // Idempotent: already processed
        }

        try {
            // Idempotency key = sagaId + step name
            $paymentId = $this->payment->charge(
                idempotencyKey: $sagaId . ':payment',
                amountCents: $amountCents,
            );

            $state->paymentId = $paymentId;
            $state->status = OrderSagaStatus::PaymentCompleted;
            $this->repo->save($state);

            $this->bus->dispatch(new ExecuteReserveStock($sagaId));
        } catch (PaymentException $e) {
            $this->fail($state, 'payment_failed: ' . $e->getMessage());
        }
    }

    public function handleReserveStock(string $sagaId): void
    {
        $state = $this->repo->findOrFail($sagaId);
        if ($state->status !== OrderSagaStatus::PaymentCompleted) {
            return;
        }

        try {
            $reservationId = $this->inventory->reserve(
                idempotencyKey: $sagaId . ':stock',
                orderId: $state->orderId,
            );

            $state->reservationId = $reservationId;
            $state->status = OrderSagaStatus::StockReserved;
            $this->repo->save($state);

            $this->bus->dispatch(new ExecuteCreateShipping($sagaId));
        } catch (InventoryException $e) {
            // Compensate: refund payment
            $this->compensate($state, 'stock_unavailable: ' . $e->getMessage());
        }
    }

    public function handleCreateShipping(string $sagaId): void
    {
        $state = $this->repo->findOrFail($sagaId);
        if ($state->status !== OrderSagaStatus::StockReserved) {
            return;
        }

        try {
            $shipmentId = $this->shipping->create(
                idempotencyKey: $sagaId . ':shipping',
                orderId: $state->orderId,
            );

            $state->shipmentId = $shipmentId;
            $state->status = OrderSagaStatus::Completed;
            $this->repo->save($state);
        } catch (ShippingException $e) {
            $this->compensate($state, 'shipping_failed: ' . $e->getMessage());
        }
    }

    /**
     * Compensation: run reverse actions for completed steps.
     * Each compensation must be idempotent.
     */
    private function compensate(OrderSagaState $state, string $reason): void
    {
        $state->status = OrderSagaStatus::Compensating;
        $state->failureReason = $reason;
        $this->repo->save($state);

        // Release stock if reserved
        if ($state->reservationId !== null) {
            $this->inventory->release(
                idempotencyKey: $state->sagaId . ':release',
                reservationId: $state->reservationId,
            );
        }

        // Refund payment if charged
        if ($state->paymentId !== null) {
            $this->payment->refund(
                idempotencyKey: $state->sagaId . ':refund',
                paymentId: $state->paymentId,
            );
        }

        $state->status = OrderSagaStatus::Failed;
        $this->repo->save($state);
    }

    private function fail(OrderSagaState $state, string $reason): void
    {
        $state->status = OrderSagaStatus::Failed;
        $state->failureReason = $reason;
        $this->repo->save($state);
    }
}

package saga

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

// Status represents the saga state machine.
type Status string

const (
	StatusStarted         Status = "started"
	StatusPaid            Status = "paid"
	StatusStockReserved   Status = "stock_reserved"
	StatusShippingCreated Status = "shipping_created"
	StatusCompleted       Status = "completed"
	StatusCompensating    Status = "compensating"
	StatusFailed          Status = "failed"
)

// State persists saga progress in DB.
type State struct {
	SagaID         string
	OrderID        string
	Status         Status
	PaymentID      string
	ReservationID  string
	ShipmentID     string
	FailureReason  string
	Version        int
	UpdatedAt      time.Time
}

type Orchestrator struct {
	repo      StateRepo
	payment   PaymentClient
	inventory InventoryClient
	shipping  ShippingClient
}

// Run executes the saga forward. On failure, compensations are triggered.
// All steps are idempotent via keys derived from sagaID.
func (o *Orchestrator) Run(ctx context.Context, sagaID, orderID string, amount int64) error {
	state, err := o.repo.Load(ctx, sagaID)
	if err != nil {
		return fmt.Errorf("load state: %w", err)
	}

	// Resume from wherever we left off (recovery after crash)
	switch state.Status {
	case StatusStarted:
		if err := o.stepCharge(ctx, state, amount); err != nil {
			return o.compensate(ctx, state, err)
		}
		fallthrough
	case StatusPaid:
		if err := o.stepReserve(ctx, state, orderID); err != nil {
			return o.compensate(ctx, state, err)
		}
		fallthrough
	case StatusStockReserved:
		if err := o.stepShipping(ctx, state, orderID); err != nil {
			return o.compensate(ctx, state, err)
		}
		state.Status = StatusCompleted
		return o.repo.Save(ctx, state)
	case StatusCompleted:
		return nil
	case StatusFailed:
		return errors.New("saga already failed")
	}
	return nil
}

func (o *Orchestrator) stepCharge(ctx context.Context, s *State, amount int64) error {
	pid, err := o.payment.Charge(ctx, s.SagaID+":payment", amount)
	if err != nil {
		return fmt.Errorf("charge: %w", err)
	}
	s.PaymentID = pid
	s.Status = StatusPaid
	return o.repo.Save(ctx, s)
}

func (o *Orchestrator) stepReserve(ctx context.Context, s *State, orderID string) error {
	rid, err := o.inventory.Reserve(ctx, s.SagaID+":stock", orderID)
	if err != nil {
		return fmt.Errorf("reserve: %w", err)
	}
	s.ReservationID = rid
	s.Status = StatusStockReserved
	return o.repo.Save(ctx, s)
}

func (o *Orchestrator) stepShipping(ctx context.Context, s *State, orderID string) error {
	sid, err := o.shipping.Create(ctx, s.SagaID+":shipping", orderID)
	if err != nil {
		return fmt.Errorf("shipping: %w", err)
	}
	s.ShipmentID = sid
	s.Status = StatusShippingCreated
	return o.repo.Save(ctx, s)
}

// compensate runs reverse actions for completed steps.
// Each compensation must be idempotent via idempotency keys.
func (o *Orchestrator) compensate(ctx context.Context, s *State, cause error) error {
	s.Status = StatusCompensating
	s.FailureReason = cause.Error()
	_ = o.repo.Save(ctx, s)

	if s.ReservationID != "" {
		if err := o.inventory.Release(ctx, s.SagaID+":release", s.ReservationID); err != nil {
			// Log and retry via outbox; do not abort compensation chain
			_ = err
		}
	}
	if s.PaymentID != "" {
		if err := o.payment.Refund(ctx, s.SagaID+":refund", s.PaymentID); err != nil {
			_ = err
		}
	}

	s.Status = StatusFailed
	return o.repo.Save(ctx, s)
}

using Microsoft.Extensions.Logging;

namespace App.Saga;

public enum SagaStatus
{
    Started,
    Paid,
    StockReserved,
    ShippingCreated,
    Completed,
    Compensating,
    Failed,
}

/// State persists saga progress in DB.
public sealed class SagaState
{
    public required string SagaId { get; init; }
    public required string OrderId { get; init; }
    public SagaStatus Status { get; set; }
    public string? PaymentId { get; set; }
    public string? ReservationId { get; set; }
    public string? ShipmentId { get; set; }
    public string? FailureReason { get; set; }
    public int Version { get; set; }
    public DateTimeOffset UpdatedAt { get; set; }
}

public sealed class SagaFailedException(string message) : Exception(message);

public sealed class Orchestrator(
    IStateRepo repo,
    IPaymentClient payment,
    IInventoryClient inventory,
    IShippingClient shipping,
    ILogger<Orchestrator> log)
{
    /// Runs the saga forward. On failure, compensations are triggered.
    /// All steps are idempotent via keys derived from sagaId.
    public async Task RunAsync(string sagaId, string orderId, long amount, CancellationToken ct)
    {
        var state = await repo.LoadAsync(sagaId, ct);

        try
        {
            // Resume from wherever we left off (recovery after crash);
            // cases deliberately fall through to the next step
            switch (state.Status)
            {
                case SagaStatus.Started:
                    await StepChargeAsync(state, amount, ct);
                    goto case SagaStatus.Paid;

                case SagaStatus.Paid:
                    await StepReserveAsync(state, orderId, ct);
                    goto case SagaStatus.StockReserved;

                case SagaStatus.StockReserved:
                    await StepShippingAsync(state, orderId, ct);
                    state.Status = SagaStatus.Completed;
                    await repo.SaveAsync(state, ct);
                    return;

                case SagaStatus.ShippingCreated:
                    state.Status = SagaStatus.Completed;
                    await repo.SaveAsync(state, ct);
                    return;

                case SagaStatus.Completed:
                    return;

                case SagaStatus.Compensating:
                    await CompensateAsync(state, state.FailureReason ?? "resumed", ct);
                    return;

                case SagaStatus.Failed:
                    throw new SagaFailedException("saga already failed");

                default:
                    throw new InvalidOperationException($"unknown status: {state.Status}");
            }
        }
        catch (Exception e) when (e is not SagaFailedException and not OperationCanceledException)
        {
            await CompensateAsync(state, e.Message, ct);
            throw;
        }
    }

    private async Task StepChargeAsync(SagaState s, long amount, CancellationToken ct)
    {
        s.PaymentId = await payment.ChargeAsync($"{s.SagaId}:payment", amount, ct);
        s.Status = SagaStatus.Paid;
        await repo.SaveAsync(s, ct);
    }

    private async Task StepReserveAsync(SagaState s, string orderId, CancellationToken ct)
    {
        s.ReservationId = await inventory.ReserveAsync($"{s.SagaId}:stock", orderId, ct);
        s.Status = SagaStatus.StockReserved;
        await repo.SaveAsync(s, ct);
    }

    private async Task StepShippingAsync(SagaState s, string orderId, CancellationToken ct)
    {
        s.ShipmentId = await shipping.CreateAsync($"{s.SagaId}:shipping", orderId, ct);
        s.Status = SagaStatus.ShippingCreated;
        await repo.SaveAsync(s, ct);
    }

    /// Runs reverse actions for completed steps.
    /// Each compensation must be idempotent via idempotency keys.
    private async Task CompensateAsync(SagaState s, string cause, CancellationToken ct)
    {
        s.Status = SagaStatus.Compensating;
        s.FailureReason = cause;
        await repo.SaveAsync(s, ct);

        if (s.ReservationId is not null)
        {
            try
            {
                await inventory.ReleaseAsync($"{s.SagaId}:release", s.ReservationId, ct);
            }
            catch (Exception e)
            {
                // Log and retry via outbox; do not abort the compensation chain
                log.LogError(e, "release failed; saga={SagaId}", s.SagaId);
            }
        }

        if (s.PaymentId is not null)
        {
            try
            {
                await payment.RefundAsync($"{s.SagaId}:refund", s.PaymentId, ct);
            }
            catch (Exception e)
            {
                log.LogError(e, "refund failed; saga={SagaId}", s.SagaId);
            }
        }

        s.Status = SagaStatus.Failed;
        await repo.SaveAsync(s, ct);
    }
}

import logging
from dataclasses import dataclass
from datetime import datetime
from enum import StrEnum

log = logging.getLogger(__name__)

class SagaStatus(StrEnum):
    STARTED = "started"
    PAID = "paid"
    STOCK_RESERVED = "stock_reserved"
    SHIPPING_CREATED = "shipping_created"
    COMPLETED = "completed"
    COMPENSATING = "compensating"
    FAILED = "failed"

@dataclass(slots=True)
class SagaState:
    """Persists saga progress in DB."""

    saga_id: str
    order_id: str
    status: SagaStatus
    payment_id: str | None = None
    reservation_id: str | None = None
    shipment_id: str | None = None
    failure_reason: str | None = None
    version: int = 0
    updated_at: datetime | None = None

class SagaFailedError(Exception):
    """Terminal state reached; no forward progress is possible."""

class Orchestrator:
    def __init__(
        self,
        repo: StateRepo,
        payment: PaymentClient,
        inventory: InventoryClient,
        shipping: ShippingClient,
    ) -> None:
        self._repo = repo
        self._payment = payment
        self._inventory = inventory
        self._shipping = shipping

    async def run(self, saga_id: str, order_id: str, amount: int) -> None:
        """Execute the saga forward; compensate on failure.

        All steps are idempotent via keys derived from saga_id.
        """
        state = await self._repo.load(saga_id)

        if state.status is SagaStatus.COMPLETED:
            return
        if state.status is SagaStatus.FAILED:
            raise SagaFailedError("saga already failed")
        if state.status is SagaStatus.COMPENSATING:
            await self._compensate(state, state.failure_reason or "resumed")
            return

        # Ordered pipeline; resume from wherever we left off (recovery after crash)
        steps = [
            (SagaStatus.STARTED, lambda: self._step_charge(state, amount)),
            (SagaStatus.PAID, lambda: self._step_reserve(state, order_id)),
            (SagaStatus.STOCK_RESERVED, lambda: self._step_shipping(state, order_id)),
        ]
        start_at = next(i for i, (st, _) in enumerate(steps) if st is state.status)

        try:
            for _, step in steps[start_at:]:
                await step()
        except Exception as e:
            await self._compensate(state, str(e))
            raise

        state.status = SagaStatus.COMPLETED
        await self._repo.save(state)

    async def _step_charge(self, s: SagaState, amount: int) -> None:
        s.payment_id = await self._payment.charge(f"{s.saga_id}:payment", amount)
        s.status = SagaStatus.PAID
        await self._repo.save(s)

    async def _step_reserve(self, s: SagaState, order_id: str) -> None:
        s.reservation_id = await self._inventory.reserve(f"{s.saga_id}:stock", order_id)
        s.status = SagaStatus.STOCK_RESERVED
        await self._repo.save(s)

    async def _step_shipping(self, s: SagaState, order_id: str) -> None:
        s.shipment_id = await self._shipping.create(f"{s.saga_id}:shipping", order_id)
        s.status = SagaStatus.SHIPPING_CREATED
        await self._repo.save(s)

    async def _compensate(self, s: SagaState, cause: str) -> None:
        """Reverse completed steps. Each compensation is idempotent."""
        s.status = SagaStatus.COMPENSATING
        s.failure_reason = cause
        await self._repo.save(s)

        if s.reservation_id is not None:
            try:
                await self._inventory.release(f"{s.saga_id}:release", s.reservation_id)
            except Exception:
                # Log and retry via outbox; do not abort the compensation chain
                log.exception("release failed", extra={"saga_id": s.saga_id})

        if s.payment_id is not None:
            try:
                await self._payment.refund(f"{s.saga_id}:refund", s.payment_id)
            except Exception:
                log.exception("refund failed", extra={"saga_id": s.saga_id})

        s.status = SagaStatus.FAILED
        await self._repo.save(s)