Паттерн Saga в C#: управление распределенными транзакциями в микросервисной архитектуре

public class OrderSagaOrchestrator
{
    private readonly IOrderService _orderService;
    private readonly IInventoryService _inventoryService;
    private readonly IPaymentService _paymentService;
 
    public OrderSagaOrchestrator(
        IOrderService orderService,
        IInventoryService inventoryService,
        IPaymentService paymentService)
    {
        _orderService = orderService;
        _inventoryService = inventoryService;
        _paymentService = paymentService;
    }
 
    public async Task<bool> ProcessOrderAsync(OrderData order)
    {
        try
        {
            // Шаг 1: Создание заказа
            await _orderService.CreateOrderAsync(order);
            
            // Шаг 2: Резервация товаров
            await _inventoryService.ReserveInventoryAsync(order.OrderId, order.Items);
            
            // Шаг 3: Проведение оплаты
            await _paymentService.ChargeAsync(order.OrderId, order.TotalAmount);
            
            return true;
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Saga failed: {ex.Message}. Starting compensation...");
            
            // Компенсирующие действия в обратном порядке
            await _paymentService.RefundAsync(order.OrderId);
            await _inventoryService.ReleaseInventoryAsync(order.OrderId);
            await _orderService.CancelOrderAsync(order.OrderId);
            
            return false;
        }
    }
}

public async Task<bool> TryReserveInventoryAsync(string productId, int requestedQuantity, string reservationId)
{
    // Транзакция на уровне сервиса инвентаря
    using (var transaction = await _database.BeginTransactionAsync())
    {
        var product = await _database.Products.FindAsync(productId);
        
        // Проверка и изменение выполняются атомарно
        if (product.AvailableQuantity >= requestedQuantity)
        {
            product.AvailableQuantity -= requestedQuantity;
            
            // Создаем запись о резервации
            var reservation = new Reservation
            {
                ReservationId = reservationId,
                ProductId = productId,
                Quantity = requestedQuantity,
                ExpirationTime = DateTime.UtcNow.AddMinutes(10)
            };
            
            _database.Reservations.Add(reservation);
            await _database.SaveChangesAsync();
            await transaction.CommitAsync();
            return true;
        }
        
        await transaction.RollbackAsync();
        return false;
    }
}

public interface ISagaStep<TData>
{
    Task ExecuteAsync(TData data);
    Task CompensateAsync(TData data);
    bool IsCompensable { get; }
}

public class CreateOrderStep : ISagaStep<OrderSagaData>
{
    private readonly IOrderRepository _orderRepository;
    
    public CreateOrderStep(IOrderRepository orderRepository)
    {
        _orderRepository = orderRepository;
    }
    
    public async Task ExecuteAsync(OrderSagaData data)
    {
        var order = new Order
        {
            Id = data.OrderId,
            CustomerId = data.CustomerId,
            Items = data.Items.Select(i => new OrderItem
            {
                ProductId = i.ProductId,
                Quantity = i.Quantity,
                Price = i.Price
            }).ToList(),
            Status = OrderStatus.Pending
        };
        
        await _orderRepository.CreateAsync(order);
        data.OrderCreated = true;
    }
    
    public async Task CompensateAsync(OrderSagaData data)
    {
        if (data.OrderCreated)
        {
            await _orderRepository.UpdateStatusAsync(data.OrderId, OrderStatus.Cancelled);
        }
    }
    
    public bool IsCompensable => true;
}

public class SagaCoordinator<TData> where TData : class
{
    private readonly List<ISagaStep<TData>> _steps = new List<ISagaStep<TData>>();
    
    public SagaCoordinator<TData> AddStep(ISagaStep<TData> step)
    {
        _steps.Add(step);
        return this;
    }
    
    public async Task<bool> ExecuteAsync(TData data)
    {
        var executedSteps = new List<ISagaStep<TData>>();
        
        try
        {
            foreach (var step in _steps)
            {
                await step.ExecuteAsync(data);
                if (step.IsCompensable)
                {
                    executedSteps.Add(step);
                }
            }
            
            return true;
        }
        catch (Exception ex)
        {
            // Здесь можно добавить логирование
            Console.WriteLine($"Saga failed at step {executedSteps.Count}. Error: {ex.Message}");
            
            // Выполняем компенсации в обратном порядке
            for (int i = executedSteps.Count - 1; i >= 0; i--)
            {
                try
                {
                    await executedSteps[i].CompensateAsync(data);
                }
                catch (Exception compensationEx)
                {
                    // Логирование ошибок компенсации
                    Console.WriteLine($"Compensation failed: {compensationEx.Message}");
                }
            }
            
            return false;
        }
    }
}

var sagaData = new OrderSagaData { 
    OrderId = Guid.NewGuid().ToString(),
    CustomerId = "customer123",
    Items = new List<OrderItemData> { ... }
};
 
var saga = new SagaCoordinator<OrderSagaData>()
    .AddStep(new CreateOrderStep(orderRepository))
    .AddStep(new ReserveInventoryStep(inventoryService))
    .AddStep(new ProcessPaymentStep(paymentService))
    .AddStep(new UpdateOrderStatusStep(orderRepository));
 
bool success = await saga.ExecuteAsync(sagaData);

public enum OrderState { Created, InventoryReserved, PaymentProcessed, Completed, Cancelled }
 
public enum OrderEvent { Create, ReserveInventory, ProcessPayment, Complete, Cancel }
 
public class OrderSaga
{
    private readonly StateMachine<OrderState, OrderEvent> _machine;
    private OrderState _currentState;
    
    public OrderSaga(OrderSagaData data)
    {
        _machine = new StateMachine<OrderState, OrderEvent>(() => _currentState, s => _currentState = s);
        
        _machine.Configure(OrderState.Created)
            .Permit(OrderEvent.ReserveInventory, OrderState.InventoryReserved)
            .Permit(OrderEvent.Cancel, OrderState.Cancelled);
        
        _machine.Configure(OrderState.InventoryReserved)
            .Permit(OrderEvent.ProcessPayment, OrderState.PaymentProcessed)
            .Permit(OrderEvent.Cancel, OrderState.Cancelled)
            .OnExit(OrderEvent.Cancel, () => ReleaseInventory(data));
        
        _machine.Configure(OrderState.PaymentProcessed)
            .Permit(OrderEvent.Complete, OrderState.Completed)
            .Permit(OrderEvent.Cancel, OrderState.Cancelled)
            .OnExit(OrderEvent.Cancel, () => RefundPayment(data));
        
        _machine.Configure(OrderState.Completed)
            .Ignore(OrderEvent.Cancel); // Завершенный заказ нельзя отменить
        
        _machine.Configure(OrderState.Cancelled);
        
        _currentState = OrderState.Created;
    }
    
    private void ReleaseInventory(OrderSagaData data)
    {
        // Реализация освобождения ресурсов
    }
    
    private void RefundPayment(OrderSagaData data)
    {
        // Реализация возврата платежа
    }
    
    public OrderState CurrentState => _currentState;
    
    public async Task<bool> MoveNext(OrderEvent @event, OrderSagaData data)
    {
        try
        {
            _machine.Fire(@event);
            return true;
        }
        catch (InvalidOperationException)
        {
            return false;
        }
    }
}

public class PersistentSagaCoordinator<TData> where TData : class, ISagaData
{
    private readonly List<ISagaStep<TData>> _steps = new List<ISagaStep<TData>>();
    private readonly ISagaRepository<TData> _repository;
 
    public PersistentSagaCoordinator(ISagaRepository<TData> repository)
    {
        _repository = repository;
    }
 
    public PersistentSagaCoordinator<TData> AddStep(ISagaStep<TData> step)
    {
        _steps.Add(step);
        return this;
    }
 
    public async Task<bool> ExecuteAsync(TData data)
    {
        // Загружаем существующую сагу или создаем новую
        var existingSaga = await _repository.GetByIdAsync(data.SagaId);
        if (existingSaga != null)
        {
            // Копируем состояние существующей саги
            data = existingSaga;
        }
        else
        {
            // Сохраняем начальное состояние
            await _repository.SaveAsync(data);
        }
 
        // Выполняем шаги, начиная с текущего
        for (int i = data.CurrentStep; i < _steps.Count; i++)
        {
            try
            {
                await _steps[i].ExecuteAsync(data);
                
                // Обновляем текущий шаг и сохраняем
                data.CurrentStep = i + 1;
                await _repository.SaveAsync(data);
            }
            catch (Exception ex)
            {
                // Логируем ошибку
                data.ErrorMessage = ex.Message;
                await _repository.SaveAsync(data);
 
                // Компенсируем выполненные шаги
                for (int j = i - 1; j >= 0; j--)
                {
                    if (_steps[j].IsCompensable)
                    {
                        await _steps[j].CompensateAsync(data);
                        data.CompensatedSteps.Add(j);
                        await _repository.SaveAsync(data);
                    }
                }
                
                return false;
            }
        }
 
        return true;
    }
}
 
public interface ISagaData
{
    Guid SagaId { get; set; }
    int CurrentStep { get; set; }
    List<int> CompensatedSteps { get; set; }
    string ErrorMessage { get; set; }
}
 
public interface ISagaRepository<TData> where TData : ISagaData
{
    Task<TData> GetByIdAsync(Guid sagaId);
    Task SaveAsync(TData data);
}

public class EfSagaRepository<TData> : ISagaRepository<TData> 
    where TData : class, ISagaData
{
    private readonly DbContext _dbContext;
    
    public EfSagaRepository(DbContext dbContext)
    {
        _dbContext = dbContext;
    }
    
    public async Task<TData> GetByIdAsync(Guid sagaId)
    {
        return await _dbContext.Set<TData>()
            .FirstOrDefaultAsync(s => s.SagaId == sagaId);
    }
    
    public async Task SaveAsync(TData data)
    {
        var existing = await GetByIdAsync(data.SagaId);
        
        if (existing == null)
        {
            _dbContext.Set<TData>().Add(data);
        }
        else
        {
            _dbContext.Entry(existing).CurrentValues.SetValues(data);
        }
        
        await _dbContext.SaveChangesAsync();
    }
}

public class StateMachineSaga<TState, TEvent, TData>
    where TState : struct, Enum
    where TEvent : struct, Enum
    where TData : class, ISagaData
{
    private readonly StateMachine<TState, TEvent> _stateMachine;
    private readonly ISagaRepository<TData> _repository;
    private TData _data;
    
    public StateMachineSaga(
        TData data,
        ISagaRepository<TData> repository,
        Func<StateMachine<TState, TEvent>, TData, Task> configureStateMachine)
    {
        _data = data;
        _repository = repository;
        
        _stateMachine = new StateMachine<TState, TEvent>(
            () => (TState)Enum.Parse(typeof(TState), _data.CurrentState),
            state => 
            {
                _data.CurrentState = state.ToString();
                _repository.SaveAsync(_data).GetAwaiter().GetResult();
            });
        
        configureStateMachine(_stateMachine, _data).GetAwaiter().GetResult();
    }
    
    public async Task FireEventAsync(TEvent @event)
    {
        // Загружаем актуальное состояние из БД
        _data = await _repository.GetByIdAsync(_data.SagaId);
        
        // Проверяем, можно ли выполнить переход
        if (_stateMachine.CanFire(@event))
        {
            await Task.Run(() => _stateMachine.Fire(@event));
            return;
        }
        
        throw new InvalidOperationException(
            $"Cannot fire event {@event} in state {_data.CurrentState}");
    }
}

// Определение состояний и событий
public enum OrderState { New, InventoryReserved, PaymentProcessed, Completed, Cancelled }
public enum OrderEvent { ReserveInventory, ProcessPayment, Complete, Cancel }
 
// Создание и настройка саги
var sagaData = new OrderSagaData { SagaId = Guid.NewGuid(), CurrentState = OrderState.New.ToString() };
var saga = new StateMachineSaga<OrderState, OrderEvent, OrderSagaData>(
    sagaData,
    new EfSagaRepository<OrderSagaData>(dbContext),
    ConfigureStateMachine);
 
// Настройка машины состояний
async Task ConfigureStateMachine(
    StateMachine<OrderState, OrderEvent> machine, 
    OrderSagaData data)
{
    machine.Configure(OrderState.New)
        .Permit(OrderEvent.ReserveInventory, OrderState.InventoryReserved)
        .OnEntryAsync(() => LogStateChange(data, "New"))
        .OnExitAsync(() => LogStateChange(data, "Leaving New"));
        
    machine.Configure(OrderState.InventoryReserved)
        .Permit(OrderEvent.ProcessPayment, OrderState.PaymentProcessed)
        .Permit(OrderEvent.Cancel, OrderState.Cancelled)
        .OnEntryAsync(() => ReserveInventory(data))
        .OnExitAsync(OrderEvent.Cancel, () => ReleaseInventory(data));
    
    // И так далее...
}
 
// Использование
await saga.FireEventAsync(OrderEvent.ReserveInventory);
await saga.FireEventAsync(OrderEvent.ProcessPayment);
await saga.FireEventAsync(OrderEvent.Complete);

public class OrderSagaData : ISagaData
{
    public Guid SagaId { get; set; }
    public string CurrentState { get; set; }
    public int CurrentStep { get; set; }
    public List<int> CompensatedSteps { get; set; } = new List<int>();
    public string ErrorMessage { get; set; }
    
    // Бизнес-данные
    public string OrderId { get; set; }
    public string CustomerId { get; set; }
    public List<OrderItem> Items { get; set; } = new List<OrderItem>();
    public decimal TotalAmount { get; set; }
    public string PaymentTransactionId { get; set; }
    public bool InventoryReserved { get; set; }
    public bool PaymentProcessed { get; set; }
}
 
public class OrderItem
{
    public string ProductId { get; set; }
    public int Quantity { get; set; }
    public decimal Price { get; set; }
}
 
public class SagaDbContext : DbContext
{
    public SagaDbContext(DbContextOptions options) : base(options) { }
    
    public DbSet<OrderSagaData> OrderSagas { get; set; }
    
    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.Entity<OrderSagaData>()
            .HasKey(s => s.SagaId);
            
        modelBuilder.Entity<OrderSagaData>()
            .Property(s => s.CurrentState)
            .IsRequired();
            
        modelBuilder.Entity<OrderSagaData>()
            .Property(s => s.CompensatedSteps)
            .HasConversion(
                v => JsonSerializer.Serialize(v, null),
                v => JsonSerializer.Deserialize<List<int>>(v, null) ?? new List<int>());
                
        modelBuilder.Entity<OrderSagaData>()
            .Property(s => s.Items)
            .HasConversion(
                v => JsonSerializer.Serialize(v, null),
                v => JsonSerializer.Deserialize<List<OrderItem>>(v, null) ?? new List<OrderItem>());
    }
}

public class MongoSagaRepository<TData> : ISagaRepository<TData>
    where TData : class, ISagaData
{
    private readonly IMongoCollection<TData> _collection;
    
    public MongoSagaRepository(IMongoDatabase database, string collectionName)
    {
        _collection = database.GetCollection<TData>(collectionName);
    }
    
    public async Task<TData> GetByIdAsync(Guid sagaId)
    {
        return await _collection.Find(s => s.SagaId == sagaId)
            .FirstOrDefaultAsync();
    }
    
    public async Task SaveAsync(TData data)
    {
        await _collection.ReplaceOneAsync(
            s => s.SagaId == data.SagaId,
            data,
            new ReplaceOptions { IsUpsert = true });
    }
}

public class Order
{
    public Guid Id { get; set; }
    public string CustomerId { get; set; }
    public OrderStatus Status { get; set; }
    public List<OrderItem> Items { get; set; }
    public decimal TotalAmount { get; set; }
    public DateTime CreatedAt { get; set; }
}
 
public enum OrderStatus
{
    Created,
    InventoryReserved,
    PaymentProcessed,
    Completed,
    Cancelled
}

public class OrderProcessingSagaData : ISagaData
{
    public Guid SagaId { get; set; }
    public string CurrentState { get; set; }
    
    // Бизнес-данные
    public Guid OrderId { get; set; }
    public string CustomerId { get; set; }
    public List<OrderItemDto> Items { get; set; }
    public decimal TotalAmount { get; set; }
    public string PaymentId { get; set; }
    public string InventoryReservationId { get; set; }
    public DateTime CreatedAt { get; set; }
    
    // Флаги состояния
    public bool OrderCreated { get; set; }
    public bool InventoryReserved { get; set; }
    public bool PaymentProcessed { get; set; }
    public bool NotificationSent { get; set; }
}

public class ReserveInventoryStep : ISagaStep<OrderProcessingSagaData>
{
    private readonly IInventoryService _inventoryService;
    private readonly ILogger<ReserveInventoryStep> _logger;
 
    public ReserveInventoryStep(
        IInventoryService inventoryService,
        ILogger<ReserveInventoryStep> logger)
    {
        _inventoryService = inventoryService;
        _logger = logger;
    }
 
    public async Task ExecuteAsync(OrderProcessingSagaData data)
    {
        _logger.LogInformation("Reserving inventory for order {OrderId}", data.OrderId);
        
        var items = data.Items.Select(i => new InventoryItemDto
        {
            ProductId = i.ProductId,
            Quantity = i.Quantity
        }).ToList();
        
        var reservationId = await _inventoryService.ReserveInventoryAsync(
            data.OrderId, items);
            
        data.InventoryReservationId = reservationId;
        data.InventoryReserved = true;
        
        _logger.LogInformation("Inventory reserved successfully for order {OrderId}", data.OrderId);
    }
 
    public async Task CompensateAsync(OrderProcessingSagaData data)
    {
        if (data.InventoryReserved)
        {
            _logger.LogInformation("Releasing inventory for order {OrderId}", data.OrderId);
            await _inventoryService.ReleaseInventoryAsync(data.InventoryReservationId);
            data.InventoryReserved = false;
            _logger.LogInformation("Inventory released for order {OrderId}", data.OrderId);
        }
    }
 
    public bool IsCompensable => true;
}

var sagaCoordinator = new PersistentSagaCoordinator<OrderProcessingSagaData>(repository)
    .AddStep(new CreateOrderStep(orderService, logger))
    .AddStep(new ReserveInventoryStep(inventoryService, logger))
    .AddStep(new ProcessPaymentStep(paymentService, logger))
    .AddStep(new SendNotificationStep(notificationService, logger));
 
var sagaData = new OrderProcessingSagaData
{
    SagaId = Guid.NewGuid(),
    OrderId = Guid.NewGuid(),
    CustomerId = "customer123",
    Items = items,
    TotalAmount = items.Sum(i => i.Price * i.Quantity),
    CreatedAt = DateTime.UtcNow
};
 
bool success = await sagaCoordinator.ExecuteAsync(sagaData);

public class OrderProcessingSaga : MassTransitStateMachine<OrderProcessingSagaState>
{
    public OrderProcessingSaga()
    {
        InstanceState(x => x.CurrentState);
        
        // Определяем события
        Event(() => OrderSubmitted, x => x.CorrelateById(context => context.Message.OrderId));
        Event(() => InventoryReserved, x => x.CorrelateById(context => context.Message.OrderId));
        Event(() => PaymentProcessed, x => x.CorrelateById(context => context.Message.OrderId));
        Event(() => OrderCompleted, x => x.CorrelateById(context => context.Message.OrderId));
        Event(() => OrderFaulted, x => x.CorrelateById(context => context.Message.OrderId));
        
        // Определяем состояния
        Initially(
            When(OrderSubmitted)
                .Then(context => {
                    context.Instance.OrderId = context.Data.OrderId;
                    context.Instance.CustomerId = context.Data.CustomerId;
                    context.Instance.TotalAmount = context.Data.TotalAmount;
                    context.Instance.Items = context.Data.Items;
                })
                .PublishAsync(context => context.Init<ReserveInventory>(new {
                    OrderId = context.Instance.OrderId,
                    Items = context.Instance.Items
                }))
                .TransitionTo(Reserving));
                
        During(Reserving,
            When(InventoryReserved)
                .PublishAsync(context => context.Init<ProcessPayment>(new {
                    OrderId = context.Instance.OrderId,
                    CustomerId = context.Instance.CustomerId,
                    Amount = context.Instance.TotalAmount
                }))
                .TransitionTo(ProcessingPayment),
            When(OrderFaulted)
                .TransitionTo(Faulted));
                
        // Продолжаем определять остальные переходы
    }
    
    // Определяем события и состояния
    public Event<OrderSubmittedEvent> OrderSubmitted { get; private set; }
    public Event<InventoryReservedEvent> InventoryReserved { get; private set; }
    public Event<PaymentProcessedEvent> PaymentProcessed { get; private set; }
    public Event<OrderCompletedEvent> OrderCompleted { get; private set; }
    public Event<OrderFaultedEvent> OrderFaulted { get; private set; }
    
    public State Reserving { get; private set; }
    public State ProcessingPayment { get; private set; }
    public State Completed { get; private set; }
    public State Faulted { get; private set; }
}

services.AddMassTransit(x =>
{
    x.AddSagaStateMachine<OrderProcessingSaga, OrderProcessingSagaState>()
        .EntityFrameworkRepository(r =>
        {
            r.ConcurrencyMode = ConcurrencyMode.Pessimistic;
            r.AddDbContext<DbContext, SagaDbContext>((provider, builder) =>
            {
                builder.UseSqlServer(connectionString);
            });
        });
        
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("localhost", "/", h =>
        {
            h.Username("guest");
            h.Password("guest");
        });
        
        cfg.ConfigureEndpoints(context);
    });
});

public class OrderProcessingSaga : Saga<OrderProcessingSagaData>,
    IAmInitiatedBy<OrderSubmittedEvent>,
    IHandleMessages<InventoryReservedEvent>,
    IHandleMessages<PaymentProcessedEvent>,
    IHandleMessages<OrderFaultedEvent>
{
    private readonly ILogger<OrderProcessingSaga> _logger;
    
    public OrderProcessingSaga(ILogger<OrderProcessingSaga> logger)
    {
        _logger = logger;
    }
    
    protected override void CorrelateMessages(ICorrelationConfig<OrderProcessingSagaData> config)
    {
        config.Correlate<OrderSubmittedEvent>(m => m.OrderId, d => d.OrderId);
        config.Correlate<InventoryReservedEvent>(m => m.OrderId, d => d.OrderId);
        config.Correlate<PaymentProcessedEvent>(m => m.OrderId, d => d.OrderId);
        config.Correlate<OrderFaultedEvent>(m => m.OrderId, d => d.OrderId);
    }
    
    public async Task Handle(OrderSubmittedEvent message)
    {
        _logger.LogInformation("Handling OrderSubmittedEvent for order {OrderId}", message.OrderId);
        
        Data.OrderId = message.OrderId;
        Data.CustomerId = message.CustomerId;
        Data.Items = message.Items;
        Data.TotalAmount = message.TotalAmount;
        Data.CreatedAt = DateTime.UtcNow;
        
        await bus.Send(new ReserveInventoryCommand
        {
            OrderId = Data.OrderId,
            Items = Data.Items
        });
    }
    
    // Реализация других обработчиков...
}

services.AddRebus(configure => configure
    .Transport(t => t.UseRabbitMq("amqp://guest:guest@localhost:5672", "order-processing"))
    .Routing(r => r.TypeBased()
        .Map<ReserveInventoryCommand>("inventory")
        .Map<ProcessPaymentCommand>("payment")
        .Map<SendNotificationCommand>("notification"))
    .Sagas(s => s.StoreInSqlServer(connectionString, "Sagas", "SagaIndex"))
    .Options(o => o.SetNumberOfWorkers(5)));

services.AddMassTransit(x =>
{
    x.AddSagaStateMachine<OrderProcessingSaga, OrderProcessingSagaState>()
        .MessageSessionRepository();
        
    x.UsingAzureServiceBus((context, cfg) =>
    {
        cfg.Host(new Uri("sb://your-namespace.servicebus.windows.net/"), h =>
        {
            h.TokenProvider = TokenProvider.CreateManagedIdentityTokenProvider();
        });
        
        cfg.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter("order", false));
        
        // Настройка топологии сообщений
        cfg.Message<OrderSubmittedEvent>(m => m.SetEntityName("order-events"));
        cfg.Message<InventoryReservedEvent>(m => m.SetEntityName("inventory-events"));
        cfg.Message<PaymentProcessedEvent>(m => m.SetEntityName("payment-events"));
    });
});

public class AzureServiceBusSagaRepository<TSaga> : ISagaRepository<TSaga>
    where TSaga : class, ISagaData
{
    private readonly IMessageSession _messageSession;
    private readonly string _sessionId;
 
    public AzureServiceBusSagaRepository(IMessageSession messageSession, string sessionId)
    {
        _messageSession = messageSession;
        _sessionId = sessionId;
    }
 
    public async Task<TSaga> GetByIdAsync(Guid sagaId)
    {
        var session = await _messageSession.AcceptMessageSessionAsync(_sessionId);
        var state = await session.GetStateAsync();
        
        if (state == null)
            return null;
            
        return JsonSerializer.Deserialize<TSaga>(state);
    }
 
    public async Task SaveAsync(TSaga saga)
    {
        var session = await _messageSession.AcceptMessageSessionAsync(_sessionId);
        var state = JsonSerializer.SerializeToUtf8Bytes(saga);
        await session.SetStateAsync(state);
    }
}

public static IHostBuilder CreateHostBuilder(string[] args) =>
    Host.CreateDefaultBuilder(args)
        .UseSerilog((context, services, configuration) => configuration
            .ReadFrom.Configuration(context.Configuration)
            .ReadFrom.Services(services)
            .Enrich.FromLogContext()
            .Enrich.WithProperty("Application", "OrderService")
            .WriteTo.Console()
            .WriteTo.Seq("http://seq-server:5341"))
        .ConfigureWebHostDefaults(webBuilder =>
        {
            webBuilder.UseStartup<Startup>();
        });

public async Task ExecuteAsync(OrderProcessingSagaData data)
{
    _logger.LogInformation(
        "Starting payment processing for order {OrderId}, amount: {Amount}", 
        data.OrderId, 
        data.TotalAmount);
    
    using (_logger.BeginScope(new Dictionary<string, object>
    {
        ["SagaId"] = data.SagaId,
        ["OrderId"] = data.OrderId,
        ["CustomerId"] = data.CustomerId
    }))
    {
        try
        {
            var paymentId = await _paymentService.ProcessPaymentAsync(
                data.CustomerId, 
                data.OrderId,
                data.TotalAmount);
                
            data.PaymentId = paymentId;
            data.PaymentProcessed = true;
            
            _logger.LogInformation(
                "Payment processed successfully for order {OrderId}, paymentId: {PaymentId}", 
                data.OrderId, 
                paymentId);
        }
        catch (Exception ex)
        {
            _logger.LogError(
                ex, 
                "Failed to process payment for order {OrderId}", 
                data.OrderId);
                
            throw;
        }
    }
}

public class ReserveInventoryStep : ISagaStep<OrderProcessingSagaData>
{
    private readonly IInventoryService _inventoryService;
    private readonly ILogger<ReserveInventoryStep> _logger;
    private readonly AsyncPolicy _retryPolicy;
 
    public ReserveInventoryStep(
        IInventoryService inventoryService,
        ILogger<ReserveInventoryStep> logger)
    {
        _inventoryService = inventoryService;
        _logger = logger;
        
        // Определяем политику повторных попыток
        _retryPolicy = Policy
            .Handle<HttpRequestException>()
            .Or<TimeoutException>()
            .WaitAndRetryAsync(
                3, // количество повторов
                retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)), // экспоненциальный отступ
                (exception, timeSpan, retryCount, context) =>
                {
                    _logger.LogWarning(
                        exception,
                        "Attempt {RetryCount} to reserve inventory failed. Retrying in {RetryTimeSpan}...",
                        retryCount,
                        timeSpan);
                });
    }
 
    public async Task ExecuteAsync(OrderProcessingSagaData data)
    {
        await _retryPolicy.ExecuteAsync(async () =>
        {
            _logger.LogInformation("Reserving inventory for order {OrderId}", data.OrderId);
            
            var items = data.Items.Select(i => new InventoryItemDto
            {
                ProductId = i.ProductId,
                Quantity = i.Quantity
            }).ToList();
            
            var reservationId = await _inventoryService.ReserveInventoryAsync(
                data.OrderId, items);
                
            data.InventoryReservationId = reservationId;
            data.InventoryReserved = true;
            
            _logger.LogInformation("Inventory reserved successfully for order {OrderId}", data.OrderId);
        });
    }
 
    public async Task CompensateAsync(OrderProcessingSagaData data)
    {
        // Также применяем политику повторов к компенсации
        await _retryPolicy.ExecuteAsync(async () =>
        {
            if (data.InventoryReserved)
            {
                _logger.LogInformation("Releasing inventory for order {OrderId}", data.OrderId);
                await _inventoryService.ReleaseInventoryAsync(data.InventoryReservationId);
                data.InventoryReserved = false;
                _logger.LogInformation("Inventory released for order {OrderId}", data.OrderId);
            }
        });
    }
 
    public bool IsCompensable => true;
}

// Политика таймаута
var timeoutPolicy = Policy
    .TimeoutAsync(30, TimeoutStrategy.Pessimistic);
 
// Политика прерывателя цепи
var circuitBreakerPolicy = Policy
    .Handle<Exception>()
    .CircuitBreakerAsync(
        exceptionsAllowedBeforeBreaking: 5,
        durationOfBreak: TimeSpan.FromMinutes(1),
        onBreak: (ex, breakDelay) => 
            _logger.LogWarning(
                ex, 
                "Circuit broken for {BreakDelay}!", 
                breakDelay),
        onReset: () => 
            _logger.LogInformation("Circuit reset!"),
        onHalfOpen: () => 
            _logger.LogInformation("Circuit half-open"));
 
// Комбинируем политики
var resiliencePolicy = Policy.WrapAsync(
    circuitBreakerPolicy,
    _retryPolicy,
    timeoutPolicy);

public async Task<PaymentResult> ProcessPaymentAsync(string paymentId, decimal amount)
{
    // Проверяем, был ли уже обработан этот платеж
    var existingPayment = await _paymentRepository.GetByIdAsync(paymentId);
    if (existingPayment != null)
    {
        _logger.LogWarning("Payment {PaymentId} already processed. Returning existing result.", paymentId);
        return new PaymentResult
        {
            PaymentId = paymentId,
            Status = existingPayment.Status,
            TransactionId = existingPayment.TransactionId
        };
    }
 
    // Выполняем платеж и сохраняем результат
    var result = await _paymentGateway.ChargeAsync(paymentId, amount);
    await _paymentRepository.SaveAsync(new Payment
    {
        Id = paymentId,
        Amount = amount,
        Status = result.Status,
        TransactionId = result.TransactionId,
        ProcessedAt = DateTime.UtcNow
    });
 
    return result;
}

public async Task SendOrderConfirmationAsync(Guid orderId)
{
    // Сохраняем в Outbox
    var message = new OutboxMessage
    {
        Id = Guid.NewGuid(),
        Type = "OrderConfirmation",
        Content = JsonSerializer.Serialize(new { OrderId = orderId }),
        Status = OutboxStatus.Pending,
        CreatedAt = DateTime.UtcNow
    };
    
    await _outboxRepository.SaveAsync(message);
    
    // Отправку выполнит фоновый процесс, который гарантирует однократную доставку
}

[Fact]
public async Task ShouldCompensateWhenPaymentFails()
{
    // Arrange
    var mockInventoryService = new Mock<IInventoryService>();
    var mockPaymentService = new Mock<IPaymentService>();
    
    mockPaymentService
        .Setup(s => s.ChargeAsync(It.IsAny<string>(), It.IsAny<decimal>()))
        .ThrowsAsync(new Exception("Payment gateway unavailable"));
    
    var saga = new OrderSagaOrchestrator(
        new OrderService(_orderRepository),
        mockInventoryService.Object,
        mockPaymentService.Object);
    
    // Act
    var result = await saga.ProcessOrderAsync(new OrderData { /* ... */ });
    
    // Assert
    Assert.False(result);
    mockInventoryService.Verify(
        s => s.ReleaseInventoryAsync(It.IsAny<string>()),
        Times.Once);
}

// Анти-паттерн: Синхронный вызов без таймаута
public async Task<bool> ReserveInventoryAsync(string productId, int quantity)
{
    var response = await _httpClient.PostAsJsonAsync(
        "/api/inventory/reserve",
        new { ProductId = productId, Quantity = quantity });
    
    response.EnsureSuccessStatusCode();
    return true;
}
 
// Лучше:
public async Task<bool> ReserveInventoryAsync(string productId, int quantity)
{
    using var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(5));
    
    try 
    {
        var response = await _httpClient.PostAsJsonAsync(
            "/api/inventory/reserve",
            new { ProductId = productId, Quantity = quantity },
            cancellationTokenSource.Token);
        
        response.EnsureSuccessStatusCode();
        return true;
    }
    catch (OperationCanceledException)
    {
        _logger.LogWarning("Inventory reservation timed out");
        return false;
    }
}

public class MessageDeduplicationService : IMessageDeduplicationService
{
    private readonly IDistributedCache _cache;
    private readonly ILogger<MessageDeduplicationService> _logger;
    private readonly TimeSpan _deduplicationWindow;
 
    public MessageDeduplicationService(
        IDistributedCache cache,
        ILogger<MessageDeduplicationService> logger,
        TimeSpan? deduplicationWindow = null)
    {
        _cache = cache;
        _logger = logger;
        _deduplicationWindow = deduplicationWindow ?? TimeSpan.FromHours(24);
    }
 
    public async Task<bool> IsNewMessageAsync(string messageId)
    {
        // Пытаемся добавить messageId в кеш
        // Если такой ID уже есть - значит, это дубликат
        if (await _cache.GetAsync(GetCacheKey(messageId)) != null)
        {
            _logger.LogWarning("Duplicate message detected: {MessageId}", messageId);
            return false;
        }
 
        // Добавляем ID сообщения в кеш с указанным временем жизни
        await _cache.SetAsync(
            GetCacheKey(messageId),
            Encoding.UTF8.GetBytes("1"),
            new DistributedCacheEntryOptions
            {
                AbsoluteExpirationRelativeToNow = _deduplicationWindow
            });
 
        return true;
    }
 
    private string GetCacheKey(string messageId) => $"msg:{messageId}";
}

public async Task Handle(PaymentProcessedEvent message, CancellationToken cancellationToken)
{
    // Проверяем, не дубликат ли это
    if (!await _deduplicationService.IsNewMessageAsync(message.MessageId))
    {
        _logger.LogInformation("Skipping duplicate payment event: {MessageId}", message.MessageId);
        return;
    }
 
    // Обрабатываем сообщение
    await _paymentProcessor.ProcessAsync(message);
}

public class LoanApprovalSaga : IStateMachine<LoanApplicationState>
{
    private readonly IScheduler _scheduler;
 
    // Остальной код...
 
    private async Task StartApprovalProcess(LoanApplicationData data)
    {
        // Отправляем запрос на проверку
        await _verificationService.StartVerificationAsync(data.ApplicationId);
 
        // Планируем проверку статуса через 24 часа
        await _scheduler.ScheduleAsync(
            new CheckVerificationStatusCommand 
            { 
                ApplicationId = data.ApplicationId 
            },
            DateTime.UtcNow.AddHours(24));
 
        // Переходим в состояние ожидания
        data.CurrentState = LoanApplicationState.VerificationInProgress;
        await _repository.SaveAsync(data);
    }
}

public class OrderQueryService
{
    private readonly IOrderRepository _repository;
 
    public OrderQueryService(IOrderRepository repository)
    {
        _repository = repository;
    }
 
    public async Task<OrderDto> GetOrderAsync(Guid orderId)
    {
        var order = await _repository.GetByIdAsync(orderId);
        
        if (order == null)
            return null;
 
        var dto = new OrderDto
        {
            Id = order.Id,
            CustomerId = order.CustomerId,
            Items = order.Items.Select(i => new OrderItemDto 
            {
                ProductId = i.ProductId,
                Quantity = i.Quantity,
                Price = i.Price
            }).ToList(),
            TotalAmount = order.TotalAmount,
            Status = order.Status,
            // Добавляем метаданные о состоянии саги
            ProcessingState = GetUserFriendlyState(order)
        };
 
        return dto;
    }
 
    private string GetUserFriendlyState(Order order)
    {
        return order.Status switch
        {
            OrderStatus.Created => "Ваш заказ создан",
            OrderStatus.InventoryReserved => "Товары зарезервированы, ожидается оплата",
            OrderStatus.PaymentProcessed => "Оплата получена, заказ обрабатывается",
            OrderStatus.Completed => "Заказ выполнен",
            OrderStatus.Cancelled => "Заказ отменен",
            _ => "Обрабатывается"
        };
    }
}

public async Task<bool> ReserveInventoryAsync(string productId, int quantity)
{
    var product = await _productRepository.GetByIdAsync(productId);
    
    if (product.AvailableQuantity < quantity)
        return false;
 
    // Запоминаем версию ресурса
    long originalVersion = product.Version;
    
    product.AvailableQuantity -= quantity;
    
    try
    {
        // Сохраняем только если версия не изменилась
        return await _productRepository.UpdateIfVersionMatchesAsync(
            product, originalVersion);
    }
    catch (DbUpdateConcurrencyException)
    {
        _logger.LogInformation(
            "Concurrency conflict when reserving product {ProductId}", 
            productId);
        return false;
    }
}

services.AddMassTransit(x =>
{
    x.AddConsumer<ReserveInventoryConsumer>()
        .Partition(4, p =>
        {
            // Сообщения с одинаковым ProductId попадают в одну и ту же очередь
            p.KeySelector = context => 
                context.Message.ProductId;
        });
});

public class SagaMonitoringService : BackgroundService
{
    private readonly ISagaRepository<OrderSagaData> _repository;
    private readonly ILogger<SagaMonitoringService> _logger;
    private readonly TimeSpan _staleThreshold;
 
    public SagaMonitoringService(
        ISagaRepository<OrderSagaData> repository,
        ILogger<SagaMonitoringService> logger,
        TimeSpan? staleThreshold = null)
    {
        _repository = repository;
        _logger = logger;
        _staleThreshold = staleThreshold ?? TimeSpan.FromHours(1);
    }
 
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                // Находим все саги, которые не обновлялись дольше порогового значения
                var staleSagas = await _repository.FindStaleSagasAsync(_staleThreshold);
                
                foreach (var saga in staleSagas)
                {
                    _logger.LogWarning(
                        "Found stale saga: {SagaId}, state: {State}, last updated: {LastUpdated}",
                        saga.SagaId,
                        saga.CurrentState,
                        saga.UpdatedAt);
                    
                    // Можно добавить автоматический откат или уведомление команды поддержки
                }
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Error in saga monitoring service");
            }
 
            // Проверяем раз в 5 минут
            await Task.Delay(TimeSpan.FromMinutes(5), stoppingToken);
        }
    }
}