Round #508

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

Certainly! A Saga pattern is useful for managing complex workflows that involve multiple steps, where each step can fail independently. Below is an example of a Saga orchestrator in C# for a multi-step order fulfillment workflow. ### Workflow Steps: 1. **Place Order**: Create an order record. 2. **Process Payment**: Charge the customer's credit card. 3. **Ship Order**: Ship the order to the customer. 4. **Confirm Delivery**: Confirm that the order has been delivered. ### Saga States: - `OrderPlaced` - `PaymentProcessed` - `OrderShipped` - `DeliveryConfirmed` ### Saga Orchestrator: ```csharp using System; using System.Threading.Tasks; public class OrderFulfillmentSaga { private readonly IOrderRepository _orderRepository; private readonly IPaymentService _paymentService; private readonly IShippingService _shippingService; private readonly IDeliveryConfirmationService _deliveryConfirmationService; public OrderFulfillmentSaga( IOrderRepository orderRepository, IPaymentService paymentService, IShippingService shippingService, IDeliveryConfirmationService deliveryConfirmationService) { _orderRepository = orderRepository; _paymentService = paymentService; _shippingService = shippingService; _deliveryConfirmationService = deliveryConfirmationService; } public async Task ExecuteAsync(Order order) { try { // Step 1: Place Order await PlaceOrderAsync(order); order.State = "OrderPlaced"; await _orderRepository.UpdateAsync(order); // Step 2: Process Payment await ProcessPaymentAsync(order); order.State = "PaymentProcessed"; await _orderRepository.UpdateAsync(order); // Step 3: Ship Order await ShipOrderAsync(order); order.State = "OrderShipped"; await _orderRepository.UpdateAsync(order); // Step 4: Confirm Delivery await ConfirmDeliveryAsync(order); order.State = "DeliveryConfirmed"; await _orderRepository.UpdateAsync(order); Console.WriteLine("Order fulfillment completed successfully."); } catch (Exception ex) { Console.WriteLine($"Error during order fulfillment: {ex.Message}"); // Revert the saga if any step fails await RevertAsync(order); } } private async Task PlaceOrderAsync(Order order) { // Logic to place an order in the database Console.WriteLine("Placing order..."); // Simulate placing order await Task.Delay(1000); } private async Task ProcessPaymentAsync(Order order) { // Logic to process payment Console.WriteLine("Processing payment..."); if (!_paymentService.Charge(order.CustomerId, order.TotalAmount)) { throw new Exception("Failed to process payment."); } await Task.Delay(1000); } private async Task ShipOrderAsync(Order order) { // Logic to ship the order Console.WriteLine("Shipping order..."); if (!_shippingService.Ship(order.OrderId, order.CustomerAddress)) { throw new Exception("Failed to ship order."); } await Task.Delay(1000); } private async Task ConfirmDeliveryAsync(Order order) { // Logic to confirm delivery Console.WriteLine("Confirming delivery..."); if (!_deliveryConfirmationService.Confirm(order.OrderId)) { throw new Exception("Failed to confirm delivery."); } await Task.Delay(1000); } private async Task RevertAsync(Order order) { // Revert the saga in case of failure Console.WriteLine("Reverting order fulfillment..."); if (order.State == "DeliveryConfirmed") { await ConfirmDeliveryAsync(order); // Revert delivery confirmation } if (order.State == "OrderShipped") { await ShipOrderAsync(order); // Revert shipping } if (order.State == "PaymentProcessed") { await ProcessPaymentAsync(order); // Revert payment processing } if (order.State == "OrderPlaced") { await PlaceOrderAsync(order); // Revert order placement } Console.WriteLine("Order fulfillment reverted successfully."); } } public class Order { public int OrderId { get; set; } public string CustomerId { get; set; } public decimal TotalAmount { get; set; } public string CustomerAddress { get; set; } public string State { get; set; } } public interface IOrderRepository { Task<Order> GetAsync(int orderId); Task UpdateAsync(Order order); } public class OrderRepository : IOrderRepository { // Implementation of the repository methods public async Task<Order> GetAsync(int orderId) { // Simulate getting an order from the database return new Order { OrderId = orderId, CustomerId = "C123", TotalAmount = 100.0m, CustomerAddress = "123 Main St" }; } public async Task UpdateAsync(Order order) { // Simulate updating an order in the database Console.WriteLine($"Updating order {order.OrderId} to state {order.State}"); } } public interface IPaymentService { bool Charge(string customerId, decimal amount); } public class PaymentService : IPaymentService { public bool Charge(string customerId, decimal amount) { // Simulate processing payment Console.WriteLine($"Charging customer {customerId} for ${amount}"); return true; // Simulate successful charge } } public interface IShippingService { bool Ship(int orderId, string address); } public class ShippingService : IShippingService { public bool Ship(int orderId, string address) { // Simulate shipping the order Console.WriteLine($"Shipping order {orderId} to {address}"); return true; // Simulate successful shipment } } public interface IDeliveryConfirmationService { bool Confirm(int orderId); } public class DeliveryConfirmationService : IDeliveryConfirmationService { public bool Confirm(int orderId) { // Simulate confirming delivery Console.WriteLine($"Confirming delivery for order {orderId}"); return true; // Simulate successful confirmation } } ``` ### Explanation: 1. **OrderFulfillmentSaga**: This class orchestrates the entire order fulfillment workflow. 2. **ExecuteAsync**: The main method that executes the saga steps and handles exceptions by reverting the saga if any step fails. 3. **PlaceOrderAsync, ProcessPaymentAsync, ShipOrderAsync, ConfirmDeliveryAsync**: These methods simulate each step of the workflow. 4. **RevertAsync**: This method reverts the saga in case of failure, ensuring that all previous steps are undone. ### Usage: To use this Saga orchestrator, you would create an instance of `OrderFulfillmentSaga` and call the `ExecuteAsync` method with an `Order` object. ```csharp var order = new Order { OrderId = 1, CustomerId = "C123", TotalAmount = 100.0m, CustomerAddress = "123 Main St" }; var saga = new OrderFulfillmentSaga( new OrderRepository(), new PaymentService(), new ShippingService(), new DeliveryConfirmationService()); await saga.ExecuteAsync(order); ``` This example provides a basic implementation of a Saga orchestrator for order fulfillment. In a real-world scenario, you would need to integrate with actual services and handle more complex error scenarios.

To implement a Saga orchestrator for a multi-step order fulfillment workflow, we'll use the MassTransit library, which is a popular choice for building message-driven applications and implementing sagas in .NET. First, let's define the necessary interfaces and messages: 1. **OrderCreatedEvent**: This event will be published when an order is created. 2. **InventoryCheckedEvent**: This event will be published after checking inventory. 3. **PaymentProcessedEvent**: This event will be published after processing payment. 4. **ShipmentScheduledEvent**: This event will be published after scheduling shipment. Next, we'll define the saga state and events: ```csharp using MassTransit; using System; public class OrderFulfillmentState : SagaStateMachineInstance { public Guid CorrelationId { get; set; } public string CurrentState { get; set; } } public enum OrderFulfillmentStates { Created, InventoryChecked, PaymentProcessed, ShipmentScheduled, Completed, Failed } public class OrderCreatedEvent : CorrelatedBy<Guid> { public Guid OrderId { get; set; } public DateTime OrderDate { get; set; } public decimal TotalAmount { get; set; } } public class InventoryCheckedEvent : CorrelatedBy<Guid> { public Guid OrderId { get; set; } public bool IsInventoryAvailable { get; set; } } public class PaymentProcessedEvent : CorrelatedBy<Guid> { public Guid OrderId { get; set; } public bool IsPaymentSuccessful { get; set; } } public class ShipmentScheduledEvent : CorrelatedBy<Guid> { public Guid OrderId { get; set; } public DateTime ScheduledShipmentDate { get; set; } } ``` Now, let's define the saga: ```csharp using MassTransit; using System; public class OrderFulfillmentSaga : SagaStateMachine<OrderFulfillmentState>, IVersionedSagaStateMachine { public const string StateMachineDefinitionName = "OrderFulfillmentSaga"; public const int SchemaVersion = 1; public OrderFulfillmentSaga() { InstanceState(x => x.CurrentState); Event(() => OrderCreated, x => { x.CorrelateBy(context => context.Message.OrderId); x.OnException<Exception>(context => context.Complete()); }); Event(() => InventoryChecked, x => { x.CorrelateBy(context => context.Message.OrderId); x.OnException<Exception>(context => context.Complete()); }); Event(() => PaymentProcessed, x => { x.CorrelateBy(context => context.Message.OrderId); x.OnException<Exception>(context => context.Complete()); }); Event(() => ShipmentScheduled, x => { x.CorrelateBy(context => context.Message.OrderId); x.OnException<Exception>(context => context.Complete()); }); Initially( When(OrderCreated) .Then(context => context.Instance.CurrentState = OrderFulfillmentStates.Created) .Publish(context => new InventoryCheckedEvent { OrderId = context.Message.OrderId }) .TransitionTo(OrderFulfillmentStates.InventoryChecked)); During(OrderFulfillmentStates.InventoryChecked, When(InventoryChecked) .Then(context => { if (context.Message.IsInventoryAvailable) context.Instance.CurrentState = OrderFulfillmentStates.PaymentProcessed; else context.Instance.CurrentState = OrderFulfillmentStates.Failed; }) .Publish(context => new PaymentProcessedEvent { OrderId = context.Message.OrderId }) .TransitionTo(OrderFulfillmentStates.PaymentProcessed)); During(OrderFulfillmentStates.PaymentProcessed, When(PaymentProcessed) .Then(context => { if (context.Message.IsPaymentSuccessful) context.Instance.CurrentState = OrderFulfillmentStates.ShipmentScheduled; else context.Instance.CurrentState = OrderFulfillmentStates.Failed; }) .Publish(context => new ShipmentScheduledEvent { OrderId = context.Message.OrderId, ScheduledShipmentDate = DateTime.UtcNow.AddDays(3) }) .TransitionTo(OrderFulfillmentStates.ShipmentScheduled)); During(OrderFulfillmentStates.ShipmentScheduled, When(ShipmentScheduled) .Then(context => { context.Instance.CurrentState = OrderFulfillmentStates.Completed; }) .Finalize()); DuringAny( When(OrderCreated, InventoryChecked, PaymentProcessed, ShipmentScheduled) .Catch<Exception>(context => context.Complete())); } public Guid CorrelationId { get; set; } } ``` Finally, let's configure MassTransit to use the saga: ```csharp using MassTransit; using MassTransit.EntityFrameworkCoreIntegration; using Microsoft.EntityFrameworkCore; public class OrderFulfillmentDbContext : SagaDbContext<OrderFulfillmentState> { public DbSet<OrderFulfillmentState> OrderFulfillmentStates { get; set; } public OrderFulfillmentDbContext(DbContextOptions<OrderFulfillmentDbContext> options) : base(options) { } } public class OrderFulfillmentBusConfigurator : IReceiveEndpointConfigurator { public void Configure(IReceiveEndpointBuilder builder) { builder.UseEntityFrameworkCoreRepository<OrderFulfillmentState, OrderFulfillmentDbContext>(options => { options.UseSqlServer(builder.Connection); }); builder.ConfigureConsumers(builder.Container); } } public class Program { public static async Task Main(string[] args) { var bus = Bus.Factory.CreateUsingRabbitMq(cfg => { cfg.Host("localhost", "/", h => { }); cfg.ReceiveEndpoint("order_fulfillment_queue", ConfigureBus); }); await bus.StartAsync(); Console.WriteLine("Press any key to exit..."); Console.ReadKey(); await bus.StopAsync(); } private static void ConfigureBus(IReceiveEndpointConfigurator endpointConfigurator) { endpointConfigurator.ConfigureConsumeTopology = false; endpointConfigurator.ReceiveEndpoint<OrderFulfillmentBusConfigurator>(); } } ``` ### Explanation 1. **OrderFulfillmentState**: This class represents the state of the saga, including the correlation ID and the current state. 2. **OrderFulfillmentStates**: An enumeration representing the different states of the order fulfillment process. 3. **OrderCreatedEvent**, **InventoryCheckedEvent**, **PaymentProcessedEvent**, **ShipmentScheduledEvent**: These classes represent the events that trigger transitions in the saga. 4. **OrderFulfillmentSaga**: This class defines the state machine and its behavior, including how to handle each event and transition between states. 5. **OrderFulfillmentDbContext**: This class represents the database context for storing the saga state. 6. **OrderFulfillmentBusConfigurator**: This class configures MassTransit to use Entity Framework Core for persistence and RabbitMQ as the message broker. 7. **Program**: The entry point of the application, which sets up and starts the MassTransit bus. This implementation provides a basic structure for an order fulfillment saga using MassTransit. You can extend it by adding more events, states, and logic as needed for your specific use case.