Round #224

Write a Saga orchestrator for a multi-step order fulfilment workflow

**Order Fulfillment Saga Orchestrator** ===================================== This C# implementation demonstrates a Saga orchestrator for a multi-step order fulfillment workflow. The Saga pattern is used to manage long-running business processes that involve multiple steps, allowing for compensation actions in case of failures. **Order Fulfillment Workflow Steps** --------------------------------- 1. **Create Order**: Create a new order with customer and product information. 2. **Verify Inventory**: Check if the ordered products are in stock. 3. **Reserve Products**: Reserve the ordered products to prevent overselling. 4. **Process Payment**: Process the payment for the order. 5. **Ship Order**: Ship the order to the customer. **Saga Orchestrator Implementation** ---------------------------------- ```csharp using System; using System.Threading.Tasks; using Microsoft.Extensions.Logging; public class OrderFulfillmentSaga { private readonly ILogger<OrderFulfillmentSaga> _logger; private readonly IOrderRepository _orderRepository; private readonly IInventoryService _inventoryService; private readonly IPaymentGateway _paymentGateway; private readonly IShippingService _shippingService; public OrderFulfillmentSaga( ILogger<OrderFulfillmentSaga> logger, IOrderRepository orderRepository, IInventoryService inventoryService, IPaymentGateway paymentGateway, IShippingService shippingService) { _logger = logger; _orderRepository = orderRepository; _inventoryService = inventoryService; _paymentGateway = paymentGateway; _shippingService = shippingService; } public async Task StartSaga(Order order) { try { // Step 1: Create Order await CreateOrder(order); // Step 2: Verify Inventory await VerifyInventory(order); // Step 3: Reserve Products await ReserveProducts(order); // Step 4: Process Payment await ProcessPayment(order); // Step 5: Ship Order await ShipOrder(order); _logger.LogInformation("Order fulfillment saga completed successfully."); } catch (Exception ex) { _logger.LogError(ex, "Order fulfillment saga failed."); await Compensate(order, ex); } } private async Task CreateOrder(Order order) { await _orderRepository.CreateOrder(order); _logger.LogInformation("Order created successfully."); } private async Task VerifyInventory(Order order) { var inventoryResult = await _inventoryService.VerifyInventory(order.Products); if (!inventoryResult.Success) { throw new InvalidOperationException("Insufficient inventory."); } _logger.LogInformation("Inventory verified successfully."); } private async Task ReserveProducts(Order order) { await _inventoryService.ReserveProducts(order.Products); _logger.LogInformation("Products reserved successfully."); } private async Task ProcessPayment(Order order) { var paymentResult = await _paymentGateway.ProcessPayment(order); if (!paymentResult.Success) { throw new InvalidOperationException("Payment processing failed."); } _logger.LogInformation("Payment processed successfully."); } private async Task ShipOrder(Order order) { await _shippingService.ShipOrder(order); _logger.LogInformation("Order shipped successfully."); } private async Task Compensate(Order order, Exception ex) { try { // Compensate actions in reverse order await _shippingService.CancelShipment(order); await _paymentGateway.RefundPayment(order); await _inventoryService.ReleaseReservedProducts(order.Products); await _orderRepository.CancelOrder(order); _logger.LogInformation("Compensation actions completed successfully."); } catch (Exception compensateEx) { _logger.LogError(compensateEx, "Compensation actions failed."); } } } ``` **Example Usage** ----------------- ```csharp var order = new Order { CustomerId = 1, Products = new[] { new Product { Id = 1, Quantity = 2 } } }; var saga = new OrderFulfillmentSaga( logger: new LoggerFactory().CreateLogger<OrderFulfillmentSaga>(), orderRepository: new OrderRepository(), inventoryService: new InventoryService(), paymentGateway: new PaymentGateway(), shippingService: new ShippingService()); await saga.StartSaga(order); ``` This implementation demonstrates a Saga orchestrator for a multi-step order fulfillment workflow. The `OrderFulfillmentSaga` class manages the workflow, invoking each step in sequence and handling compensation actions in case of failures. The example usage shows how to create an instance of the Saga orchestrator and start the workflow for a given order.

```csharp using System; using System.Threading.Tasks; using MassTransit; using Microsoft.Extensions.Logging; // Define the order states public enum OrderState { New, PaymentProcessed, InventoryReserved, ShipmentCreated, Completed } // Define the order data public class OrderData { public Guid OrderId { get; set; } public decimal Amount { get; set; } public string CustomerName { get; set; } } // Define the payment request public class PaymentRequest { public Guid OrderId { get; set; } public decimal Amount { get; set; } } // Define the payment response public class PaymentResponse { public Guid OrderId { get; set; } public bool Success { get; set; } } // Define the inventory request public class InventoryRequest { public Guid OrderId { get; set; } public string Product { get; set; } } // Define the inventory response public class InventoryResponse { public Guid OrderId { get; set; } public bool Success { get; set; } } // Define the shipment request public class ShipmentRequest { public Guid OrderId { get; set; } public string Address { get; set; } } // Define the shipment response public class ShipmentResponse { public Guid OrderId { get; set; } public bool Success { get; set; } } // Define the order state machine public class OrderStateMachine : SagaStateMachine<OrderStateMachine, OrderState> { public OrderStateMachine() { InstanceState(x => x.CurrentState); Event(() => PaymentProcessedEvent, x => x.CorrelateById(m => m.OrderId)); Event(() => InventoryReservedEvent, x => x.CorrelateById(m => m.OrderId)); Event(() => ShipmentCreatedEvent, x => x.CorrelateById(m => m.OrderId)); Initially( When(StartOrder) .Then(context => context.Instance.OrderId = context.Message.OrderId) .Then(context => context.Instance.Amount = context.Message.Amount) .Then(context => context.Instance.CustomerName = context.Message.CustomerName) .Then(context => context.Send(new PaymentRequest { OrderId = context.Instance.OrderId, Amount = context.Instance.Amount })) .TransitionTo(OrderState.New)); During(OrderState.New, When(PaymentProcessedEvent) .Then(context => context.Instance.CurrentState = OrderState.PaymentProcessed) .Then(context => context.Send(new InventoryRequest { OrderId = context.Instance.OrderId, Product = "Product1" })) .TransitionTo(OrderState.PaymentProcessed)); During(OrderState.PaymentProcessed, When(InventoryReservedEvent) .Then(context => context.Instance.CurrentState = OrderState.InventoryReserved) .Then(context => context.Send(new ShipmentRequest { OrderId = context.Instance.OrderId, Address = "Address1" })) .TransitionTo(OrderState.InventoryReserved)); During(OrderState.InventoryReserved, When(ShipmentCreatedEvent) .Then(context => context.Instance.CurrentState = OrderState.ShipmentCreated) .Then(context => context.Instance.CurrentState = OrderState.Completed) .TransitionTo(OrderState.Completed)); } public Event<PaymentRequest> StartOrder { get; private set; } public Event<PaymentResponse> PaymentProcessedEvent { get; private set; } public Event<InventoryResponse> InventoryReservedEvent { get; private set; } public Event<ShipmentResponse> ShipmentCreatedEvent { get; private set; } } // Define the order saga public class OrderSaga : OrderStateMachine, ISaga { public Guid CorrelationId { get; set; } public OrderState CurrentState { get; set; } public Guid OrderId { get; set; } public decimal Amount { get; set; } public string CustomerName { get; set; } } // Define the payment consumer public class PaymentConsumer : IConsumer<PaymentRequest> { private readonly ILogger<PaymentConsumer> _logger; public PaymentConsumer(ILogger<PaymentConsumer> logger) { _logger = logger; } public async Task Consume(ConsumeContext<PaymentRequest> context) { try { // Process payment await Task.Delay(1000); await context.Publish(new PaymentResponse { OrderId = context.Message.OrderId, Success = true }); } catch (Exception ex) { _logger.LogError(ex, "Error processing payment"); await context.Publish(new PaymentResponse { OrderId = context.Message.OrderId, Success = false }); } } } // Define the inventory consumer public class InventoryConsumer : IConsumer<InventoryRequest> { private readonly ILogger<InventoryConsumer> _logger; public InventoryConsumer(ILogger<InventoryConsumer> logger) { _logger = logger; }