Saga pattern trong Microservices

// Order Service — publishes event after creating order
@Service
@RequiredArgsConstructor
public class OrderService {
 
   private final OrderRepository orderRepository;
   private final RabbitTemplate rabbitTemplate;
 
   @Transactional
   public OrderDTO createOrder(CreateOrderRequest request) {
       Order order = Order.builder()
           .customerId(request.getCustomerId())
           .items(request.getItems())
           .status(OrderStatus.PENDING)
           .build();
       order = orderRepository.save(order);
 
       // Publish event → Inventory Service listens
       // Lưu ý: gửi message ngay trong @Transactional là "dual write": message đi trước khi DB commit,
       // DB rollback thì message vẫn đã gửi. Production nên dùng Transactional Outbox (xem đoạn ngay dưới)
       rabbitTemplate.convertAndSend("order.exchange", "order.created",
           new OrderCreatedEvent(order.getId(), order.getItems(), order.getTotalAmount()));
 
       return toDTO(order);
   }
}
 
// Inventory Service — listens and reacts
@Component
@RequiredArgsConstructor
@Slf4j
public class InventoryEventListener {
 
   private final InventoryService inventoryService;
   private final RabbitTemplate rabbitTemplate;
 
   @RabbitListener(queues = "inventory.order-created")
   public void onOrderCreated(OrderCreatedEvent event) {
       try {
           inventoryService.reserve(event.getOrderId(), event.getItems());
 
           // Success → notify Payment Service
           rabbitTemplate.convertAndSend("order.exchange", "inventory.reserved",
               new InventoryReservedEvent(event.getOrderId(), event.getTotalAmount()));
 
       } catch (InsufficientStockException e) {
           // Fail → compensate: cancel order
           rabbitTemplate.convertAndSend("order.exchange", "inventory.failed",
               new InventoryFailedEvent(event.getOrderId(), e.getMessage()));
       }
   }
 
   // Compensation: release inventory khi payment fails
   @RabbitListener(queues = "inventory.payment-failed")
   public void onPaymentFailed(PaymentFailedEvent event) {
       inventoryService.release(event.getOrderId());
       log.info("Inventory released for order: {}", event.getOrderId());
   }
}

/**
* Saga Orchestrator — biết toàn bộ flow, điều phối step-by-step.
*
* Implement dưới dạng state machine hoặc sequential steps.
* Mỗi step: execute → success → next step / fail → compensate previous steps.
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class CreateOrderSagaOrchestrator {
 
   private final OrderService orderService;
   private final InventoryClient inventoryClient;
   private final PaymentClient paymentClient;
   private final NotificationClient notificationClient;
 
   /**
    * Execute saga: từng step, rollback nếu fail.
    *
    * Flow: Create Order → Reserve Inventory → Charge Payment → Confirm
    * Compensate: Cancel Order ← Release Inventory ← Refund Payment
    */
   public OrderDTO execute(CreateOrderRequest request) {
       // Step 1: Create order (PENDING status)
       OrderDTO order = orderService.create(request);
       log.info("Saga step 1/4: Order created | orderId={}", order.getId());
 
       try {
           // Step 2: Reserve inventory
           inventoryClient.reserve(order.getId(), request.getItems());
           log.info("Saga step 2/4: Inventory reserved | orderId={}", order.getId());
 
           try {
               // Step 3: Charge payment
               paymentClient.charge(order.getId(), order.getTotalAmount(), request.getPaymentMethod());
               log.info("Saga step 3/4: Payment charged | orderId={}", order.getId());
 
               // Step 4: Confirm order
               orderService.confirm(order.getId());
               notificationClient.sendOrderConfirmation(order.getCustomerId(), order.getId());
               log.info("Saga step 4/4: Order confirmed | orderId={}", order.getId());
 
               return orderService.getById(order.getId());
 
           } catch (PaymentException e) {
               // Compensate step 2: release inventory
               log.warn("Saga compensation: payment failed, releasing inventory | orderId={}",
                   order.getId());
               inventoryClient.release(order.getId());
               orderService.cancel(order.getId(), "Payment failed: " + e.getMessage());
               throw new OrderCreationFailedException("Payment failed", e);
           }
 
       } catch (InsufficientStockException e) {
           // Compensate step 1: cancel order
           log.warn("Saga compensation: insufficient stock, cancelling order | orderId={}",
               order.getId());
           orderService.cancel(order.getId(), "Insufficient stock: " + e.getMessage());
           throw new OrderCreationFailedException("Insufficient stock", e);
       }
   }
}

@Service
@RequiredArgsConstructor
@Slf4j
public class PaymentService {
 
   private final PaymentRepository paymentRepository;
 
   @Transactional
   public PaymentResult charge(UUID orderId, BigDecimal amount) {
       // Check idempotency: đã charge cho order này chưa?
       Optional<Payment> existing = paymentRepository.findByOrderId(orderId);
       if (existing.isPresent()) {
           log.info("Payment already processed for order: {}", orderId);
           return existing.get().toResult();  // Return existing result, don't charge again
       }
 
       // First time → execute charge
       // Check-then-act ở trên chưa đủ: 2 message trùng tới cùng lúc vẫn lọt qua cả hai.
       // Cần UNIQUE(order_id) ở bảng payment để insert thứ hai fail, và gửi orderId làm
       // idempotency key cho payment gateway để chính gateway không trừ tiền 2 lần
       Payment payment = processCharge(orderId, amount);
       return payment.toResult();
   }
}

// Orchestrator với timeout per step
CompletableFuture<InventoryResult> inventoryFuture = CompletableFuture
   .supplyAsync(() -> inventoryClient.reserve(orderId, items))
   .orTimeout(10, TimeUnit.SECONDS);  // 10s timeout
 
try {
   InventoryResult result = inventoryFuture.join();
} catch (CompletionException e) {
   // orTimeout làm future fail với TimeoutException, join() bọc nó trong CompletionException
   if (!(e.getCause() instanceof TimeoutException)) {
       throw e;
   }
   // Timeout = treat as failure → compensate.
   // Nhưng request có thể đã tới Inventory và reserve thành công, nên release luôn
   // (release phải idempotent: chưa reserve thì không làm gì)
   inventoryClient.release(orderId);
   orderService.cancel(orderId, "Inventory service timeout");
   throw new SagaTimeoutException("Inventory reservation timed out");
}

public enum OrderStatus {
   PENDING,              // Saga đang chạy — chưa final
   INVENTORY_RESERVED,   // Step 2 done
   PAYMENT_CHARGED,      // Step 3 done
   CONFIRMED,            // Saga complete — happy path
   CANCELLED,            // Saga compensated — sad path
   COMPENSATION_FAILED   // Compensation cũng fail — cần manual intervention
}

<!-- BPMN as Saga Orchestrator (rút gọn: bỏ khai báo namespace và phần diagram) -->
<error id="PAYMENT_FAILED" errorCode="PAYMENT_FAILED"/>
 
<process id="createOrderSaga">
   <startEvent id="start"/>
   <sequenceFlow id="f1" sourceRef="start" targetRef="createOrder"/>
 
   <!-- Step 1: Create Order -->
   <serviceTask id="createOrder" activiti:delegateExpression="${createOrderDelegate}"/>
   <boundaryEvent id="createOrderCompensation" attachedToRef="createOrder">
       <compensateEventDefinition/>
   </boundaryEvent>
   <serviceTask id="cancelOrder" isForCompensation="true"
                activiti:delegateExpression="${cancelOrderDelegate}"/>
   <!-- compensation handler nối với boundary event bằng association, không phải sequenceFlow -->
   <association id="a1" sourceRef="createOrderCompensation" targetRef="cancelOrder"
                associationDirection="One"/>
   <sequenceFlow id="f2" sourceRef="createOrder" targetRef="reserveInventory"/>
 
   <!-- Step 2: Reserve Inventory -->
   <serviceTask id="reserveInventory" activiti:delegateExpression="${reserveInventoryDelegate}"/>
   <boundaryEvent id="reserveCompensation" attachedToRef="reserveInventory">
       <compensateEventDefinition/>
   </boundaryEvent>
   <serviceTask id="releaseInventory" isForCompensation="true"
                activiti:delegateExpression="${releaseInventoryDelegate}"/>
   <association id="a2" sourceRef="reserveCompensation" targetRef="releaseInventory"
                associationDirection="One"/>
   <sequenceFlow id="f3" sourceRef="reserveInventory" targetRef="chargePayment"/>
 
   <!-- Step 3: Charge Payment (may fail, delegate ném BpmnError("PAYMENT_FAILED")) -->
   <serviceTask id="chargePayment" activiti:delegateExpression="${chargePaymentDelegate}"/>
   <boundaryEvent id="paymentError" attachedToRef="chargePayment">
       <errorEventDefinition errorRef="PAYMENT_FAILED"/>
   </boundaryEvent>
   <sequenceFlow id="f4" sourceRef="chargePayment" targetRef="success"/>
 
   <!-- Happy path end -->
   <endEvent id="success"/>
 
   <!-- Error path: trigger compensation for all completed steps (ngược thứ tự hoàn thành) -->
   <sequenceFlow id="f5" sourceRef="paymentError" targetRef="compensateAll"/>
   <intermediateThrowEvent id="compensateAll">
       <compensateEventDefinition/>
   </intermediateThrowEvent>
   <sequenceFlow id="f6" sourceRef="compensateAll" targetRef="sagaFailed"/>
   <endEvent id="sagaFailed"/>
</process>