distributed 2.3.4

//! Saga Orchestration Tests
//!
//! ## The Order Fulfillment Saga
//!
//! This example implements an order fulfillment process:
//!
//! 1. **Start**: Saga receives order details
//! 2. **Reserve Inventory**: Reserve items from inventory
//! 3. **Process Payment**: Charge the customer
//! 4. **Complete**: Mark order as fulfilled
//!
//! If any step fails, the saga compensates by:
//! - Releasing inventory reservations
//! - Refunding payments
//! - Cancelling the order

use super::order::{
    Inventory, Order, OrderFulfillmentSaga, OrderItem, OrderStatus, Payment, PaymentStatus,
    SagaStatus,
};
use distributed::{AggregateBuilder, HashMapRepository};

#[tokio::test]
async fn saga_happy_path_completes_order() {
    // Set up repositories for each aggregate type
    let order_repo = HashMapRepository::new().aggregate::<Order>();
    let inventory_repo = HashMapRepository::new().aggregate::<Inventory>();
    let payment_repo = HashMapRepository::new().aggregate::<Payment>();
    let saga_repo = HashMapRepository::new().aggregate::<OrderFulfillmentSaga>();

    // === Setup: Initialize inventory ===
    let mut widget_inventory = Inventory::new();
    widget_inventory
        .initialize("WIDGET-001".to_string(), 100)
        .unwrap();
    inventory_repo.commit(&mut widget_inventory).await.unwrap();

    // === Step 1: Create Order ===
    let order_id = "order-123".to_string();
    let items = vec![OrderItem {
        sku: "WIDGET-001".to_string(),
        quantity: 5,
        price_cents: 1000,
    }];

    let mut order = Order::new();
    order
        .create(order_id.clone(), "customer-456".to_string(), items.clone())
        .unwrap();
    order_repo.commit(&mut order).await.unwrap();

    // === Step 2: Start Saga ===
    let mut order_fulfillment_saga = OrderFulfillmentSaga::new();
    order_fulfillment_saga
        .start(
            "saga-123".to_string(),
            order_id.clone(),
            "customer-456".to_string(),
            items,
            5000, // 5 widgets * 1000 cents
        )
        .unwrap();
    assert_eq!(order_fulfillment_saga.status(), SagaStatus::Started);
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();

    // === Step 3: Reserve Inventory ===
    let mut inventory = inventory_repo.get("WIDGET-001").await.unwrap().unwrap();
    assert!(inventory.can_reserve(5));
    inventory.reserve(order_id.clone(), 5).unwrap();
    inventory_repo.commit(&mut inventory).await.unwrap();

    // Update saga state
    let mut order_fulfillment_saga = saga_repo.get("saga-123").await.unwrap().unwrap();
    order_fulfillment_saga.inventory_reserved().unwrap();
    assert_eq!(
        order_fulfillment_saga.status(),
        SagaStatus::InventoryReserved
    );
    assert!(order_fulfillment_saga.compensation().inventory_reserved);
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();

    // Update order state
    let mut order = order_repo.get(&order_id).await.unwrap().unwrap();
    order.mark_inventory_reserved().unwrap();
    order_repo.commit(&mut order).await.unwrap();

    // === Step 4: Process Payment ===
    let mut payment = Payment::new();
    payment
        .initiate("payment-789".to_string(), order_id.clone(), 5000)
        .unwrap();
    payment.authorize("txn-abc123".to_string()).unwrap();
    payment.capture().unwrap();
    assert!(payment.is_successful());
    payment_repo.commit(&mut payment).await.unwrap();

    // Update saga state
    let mut order_fulfillment_saga = saga_repo.get("saga-123").await.unwrap().unwrap();
    order_fulfillment_saga.payment_succeeded().unwrap();
    assert_eq!(
        order_fulfillment_saga.status(),
        SagaStatus::PaymentProcessed
    );
    assert!(order_fulfillment_saga.compensation().payment_processed);
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();

    // Update order state
    let mut order = order_repo.get(&order_id).await.unwrap().unwrap();
    order.mark_payment_processed().unwrap();
    order_repo.commit(&mut order).await.unwrap();

    // === Step 5: Complete Saga ===
    let mut order_fulfillment_saga = saga_repo.get("saga-123").await.unwrap().unwrap();
    order_fulfillment_saga.complete().unwrap();
    assert_eq!(order_fulfillment_saga.status(), SagaStatus::Completed);
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();

    // Commit the inventory reservation (no longer reversible)
    let mut inventory = inventory_repo.get("WIDGET-001").await.unwrap().unwrap();
    inventory.commit_reservation(order_id.clone()).unwrap();
    inventory_repo.commit(&mut inventory).await.unwrap();

    // Complete the order
    let mut order = order_repo.get(&order_id).await.unwrap().unwrap();
    order.complete().unwrap();
    order_repo.commit(&mut order).await.unwrap();

    // === Verify Final State ===
    let final_order_fulfillment_saga = saga_repo.get("saga-123").await.unwrap().unwrap();
    assert_eq!(final_order_fulfillment_saga.status(), SagaStatus::Completed);
    assert!(final_order_fulfillment_saga.is_complete());

    let final_order = order_repo.get(&order_id).await.unwrap().unwrap();
    assert_eq!(final_order.status(), OrderStatus::Completed);

    let final_inventory = inventory_repo.get("WIDGET-001").await.unwrap().unwrap();
    assert_eq!(final_inventory.available(), 95); // 100 - 5
    assert_eq!(final_inventory.reserved(), 0);
    assert!(final_inventory.reservation_for_order(&order_id).is_none());
}

#[tokio::test]
async fn saga_compensates_on_payment_failure() {
    // Set up repositories
    let order_repo = HashMapRepository::new().aggregate::<Order>();
    let inventory_repo = HashMapRepository::new().aggregate::<Inventory>();
    let payment_repo = HashMapRepository::new().aggregate::<Payment>();
    let saga_repo = HashMapRepository::new().aggregate::<OrderFulfillmentSaga>();

    // === Setup ===
    let mut widget_inventory = Inventory::new();
    widget_inventory
        .initialize("WIDGET-002".to_string(), 50)
        .unwrap();
    inventory_repo.commit(&mut widget_inventory).await.unwrap();

    let order_id = "order-fail-456".to_string();
    let items = vec![OrderItem {
        sku: "WIDGET-002".to_string(),
        quantity: 10,
        price_cents: 500,
    }];

    let mut order = Order::new();
    order
        .create(order_id.clone(), "customer-789".to_string(), items.clone())
        .unwrap();
    order_repo.commit(&mut order).await.unwrap();

    // === Start Saga ===
    let mut order_fulfillment_saga = OrderFulfillmentSaga::new();
    order_fulfillment_saga
        .start(
            "saga-fail-456".to_string(),
            order_id.clone(),
            "customer-789".to_string(),
            items,
            5000,
        )
        .unwrap();
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();

    // === Reserve Inventory (succeeds) ===
    let mut inventory = inventory_repo.get("WIDGET-002").await.unwrap().unwrap();
    inventory.reserve(order_id.clone(), 10).unwrap();
    inventory_repo.commit(&mut inventory).await.unwrap();

    let mut order_fulfillment_saga = saga_repo.get("saga-fail-456").await.unwrap().unwrap();
    order_fulfillment_saga.inventory_reserved().unwrap();
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();

    let mut order = order_repo.get(&order_id).await.unwrap().unwrap();
    order.mark_inventory_reserved().unwrap();
    order_repo.commit(&mut order).await.unwrap();

    // Verify inventory is reserved
    let inventory = inventory_repo.get("WIDGET-002").await.unwrap().unwrap();
    assert_eq!(inventory.available(), 40); // 50 - 10
    assert_eq!(inventory.reserved(), 10);

    // === Payment Fails ===
    let mut payment = Payment::new();
    payment
        .initiate("payment-fail-xyz".to_string(), order_id.clone(), 5000)
        .unwrap();
    payment.fail("Insufficient funds".to_string()).unwrap();
    assert!(!payment.is_successful());
    assert_eq!(payment.status(), PaymentStatus::Failed);
    payment_repo.commit(&mut payment).await.unwrap();

    // === Saga enters compensation mode ===
    let mut order_fulfillment_saga = saga_repo.get("saga-fail-456").await.unwrap().unwrap();
    order_fulfillment_saga
        .step_failed("Payment".to_string(), "Insufficient funds".to_string())
        .unwrap();
    assert_eq!(order_fulfillment_saga.status(), SagaStatus::Compensating);
    assert!(order_fulfillment_saga.needs_inventory_compensation());
    assert!(!order_fulfillment_saga.needs_payment_compensation()); // Payment wasn't successful
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();

    // === Compensate: Release Inventory ===
    let mut inventory = inventory_repo.get("WIDGET-002").await.unwrap().unwrap();
    inventory.release_reservation(order_id.clone()).unwrap();
    inventory_repo.commit(&mut inventory).await.unwrap();

    let mut order_fulfillment_saga = saga_repo.get("saga-fail-456").await.unwrap().unwrap();
    order_fulfillment_saga.inventory_compensated().unwrap();
    assert!(!order_fulfillment_saga.needs_inventory_compensation());
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();

    // === Cancel Order ===
    let mut order = order_repo.get(&order_id).await.unwrap().unwrap();
    order
        .cancel("Payment failed: Insufficient funds".to_string())
        .unwrap();
    order_repo.commit(&mut order).await.unwrap();

    // === Mark Saga as Failed ===
    let mut order_fulfillment_saga = saga_repo.get("saga-fail-456").await.unwrap().unwrap();
    order_fulfillment_saga.mark_failed().unwrap();
    assert_eq!(order_fulfillment_saga.status(), SagaStatus::Failed);
    assert!(order_fulfillment_saga.is_complete());
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();

    // === Verify Final State ===
    let final_order_fulfillment_saga = saga_repo.get("saga-fail-456").await.unwrap().unwrap();
    assert_eq!(final_order_fulfillment_saga.status(), SagaStatus::Failed);
    assert_eq!(
        final_order_fulfillment_saga.failure_reason(),
        Some("Payment: Insufficient funds")
    );

    let final_order = order_repo.get(&order_id).await.unwrap().unwrap();
    assert_eq!(final_order.status(), OrderStatus::Cancelled);

    // Inventory should be restored
    let final_inventory = inventory_repo.get("WIDGET-002").await.unwrap().unwrap();
    assert_eq!(final_inventory.available(), 50); // Back to original
    assert_eq!(final_inventory.reserved(), 0);
}

#[tokio::test]
async fn saga_compensates_on_inventory_failure() {
    // Set up repositories
    let order_repo = HashMapRepository::new().aggregate::<Order>();
    let inventory_repo = HashMapRepository::new().aggregate::<Inventory>();
    let saga_repo = HashMapRepository::new().aggregate::<OrderFulfillmentSaga>();

    // === Setup: Low inventory ===
    let mut widget_inventory = Inventory::new();
    widget_inventory
        .initialize("WIDGET-003".to_string(), 5)
        .unwrap(); // Only 5 available
    inventory_repo.commit(&mut widget_inventory).await.unwrap();

    let order_id = "order-inv-fail-789".to_string();
    let items = vec![OrderItem {
        sku: "WIDGET-003".to_string(),
        quantity: 10, // Trying to order 10
        price_cents: 500,
    }];

    let mut order = Order::new();
    order
        .create(order_id.clone(), "customer-xyz".to_string(), items.clone())
        .unwrap();
    order_repo.commit(&mut order).await.unwrap();

    // === Start Saga ===
    let mut order_fulfillment_saga = OrderFulfillmentSaga::new();
    order_fulfillment_saga
        .start(
            "saga-inv-fail".to_string(),
            order_id.clone(),
            "customer-xyz".to_string(),
            items,
            5000,
        )
        .unwrap();
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();

    // === Try to Reserve Inventory (fails - not enough stock) ===
    let inventory = inventory_repo.get("WIDGET-003").await.unwrap().unwrap();
    assert!(!inventory.can_reserve(10)); // Can't reserve 10 when only 5 available

    // Saga fails at first step
    let mut order_fulfillment_saga = saga_repo.get("saga-inv-fail").await.unwrap().unwrap();
    order_fulfillment_saga
        .step_failed(
            "Inventory".to_string(),
            "Insufficient stock for WIDGET-003".to_string(),
        )
        .unwrap();
    assert_eq!(order_fulfillment_saga.status(), SagaStatus::Compensating);
    // No compensation needed - nothing was reserved yet
    assert!(!order_fulfillment_saga.needs_inventory_compensation());
    assert!(!order_fulfillment_saga.needs_payment_compensation());
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();

    // Cancel order
    let mut order = order_repo.get(&order_id).await.unwrap().unwrap();
    order.cancel("Insufficient stock".to_string()).unwrap();
    order_repo.commit(&mut order).await.unwrap();

    // Mark saga as failed (no compensation needed)
    let mut order_fulfillment_saga = saga_repo.get("saga-inv-fail").await.unwrap().unwrap();
    order_fulfillment_saga.mark_failed().unwrap();
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();

    // === Verify Final State ===
    let final_order_fulfillment_saga = saga_repo.get("saga-inv-fail").await.unwrap().unwrap();
    assert_eq!(final_order_fulfillment_saga.status(), SagaStatus::Failed);
    assert!(final_order_fulfillment_saga.is_complete());

    let final_order = order_repo.get(&order_id).await.unwrap().unwrap();
    assert_eq!(final_order.status(), OrderStatus::Cancelled);

    // Inventory unchanged
    let final_inventory = inventory_repo.get("WIDGET-003").await.unwrap().unwrap();
    assert_eq!(final_inventory.available(), 5);
    assert_eq!(final_inventory.reserved(), 0);
}

#[tokio::test]
async fn saga_is_replayable_from_events() {
    let saga_repo = HashMapRepository::new().aggregate::<OrderFulfillmentSaga>();

    let items = vec![OrderItem {
        sku: "WIDGET-REPLAY".to_string(),
        quantity: 3,
        price_cents: 1500,
    }];

    // Create and progress a saga
    let mut order_fulfillment_saga = OrderFulfillmentSaga::new();
    order_fulfillment_saga
        .start(
            "saga-replay".to_string(),
            "order-replay".to_string(),
            "customer-replay".to_string(),
            items,
            4500,
        )
        .unwrap();
    order_fulfillment_saga.inventory_reserved().unwrap();
    order_fulfillment_saga.payment_succeeded().unwrap();

    // Commit to repository
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();

    // Retrieve and verify state is reconstructed from events
    let restored = saga_repo.get("saga-replay").await.unwrap().unwrap();

    assert_eq!(restored.order_id(), "order-replay");
    assert_eq!(restored.customer_id(), "customer-replay");
    assert_eq!(restored.total_cents(), 4500);
    assert_eq!(restored.status(), SagaStatus::PaymentProcessed);
    assert!(restored.compensation().inventory_reserved);
    assert!(restored.compensation().payment_processed);

    // Can continue from restored state
    let mut restored = restored;
    restored.complete().unwrap();
    saga_repo.commit(&mut restored).await.unwrap();

    let final_order_fulfillment_saga = saga_repo.get("saga-replay").await.unwrap().unwrap();
    assert_eq!(final_order_fulfillment_saga.status(), SagaStatus::Completed);
}

#[tokio::test]
async fn saga_tracks_compensation_state_correctly() {
    let saga_repo = HashMapRepository::new().aggregate::<OrderFulfillmentSaga>();

    let items = vec![OrderItem {
        sku: "WIDGET-COMP".to_string(),
        quantity: 1,
        price_cents: 100,
    }];

    // Progress saga through multiple steps
    let mut order_fulfillment_saga = OrderFulfillmentSaga::new();
    order_fulfillment_saga
        .start(
            "saga-comp".to_string(),
            "order-comp".to_string(),
            "customer-comp".to_string(),
            items,
            100,
        )
        .unwrap();

    // Initially no compensation needed
    assert!(!order_fulfillment_saga.compensation().inventory_reserved);
    assert!(!order_fulfillment_saga.compensation().payment_processed);

    // After inventory reserved
    order_fulfillment_saga.inventory_reserved().unwrap();
    assert!(order_fulfillment_saga.compensation().inventory_reserved);
    assert!(!order_fulfillment_saga.compensation().payment_processed);

    // After payment processed
    order_fulfillment_saga.payment_succeeded().unwrap();
    assert!(order_fulfillment_saga.compensation().inventory_reserved);
    assert!(order_fulfillment_saga.compensation().payment_processed);

    // Fail after both steps completed
    order_fulfillment_saga
        .step_failed("FinalStep".to_string(), "Something went wrong".to_string())
        .unwrap();
    assert_eq!(order_fulfillment_saga.status(), SagaStatus::Compensating);
    assert!(order_fulfillment_saga.needs_inventory_compensation());
    assert!(order_fulfillment_saga.needs_payment_compensation());

    // Compensate payment first
    order_fulfillment_saga.payment_compensated().unwrap();
    assert!(order_fulfillment_saga.needs_inventory_compensation());
    assert!(!order_fulfillment_saga.needs_payment_compensation());

    // Compensate inventory
    order_fulfillment_saga.inventory_compensated().unwrap();
    assert!(!order_fulfillment_saga.needs_inventory_compensation());
    assert!(!order_fulfillment_saga.needs_payment_compensation());

    // Now can mark as failed
    order_fulfillment_saga.mark_failed().unwrap();
    assert_eq!(order_fulfillment_saga.status(), SagaStatus::Failed);
    assert!(order_fulfillment_saga.is_complete());

    // Verify it persists correctly
    saga_repo.commit(&mut order_fulfillment_saga).await.unwrap();
    let restored = saga_repo.get("saga-comp").await.unwrap().unwrap();
    assert_eq!(restored.status(), SagaStatus::Failed);
    assert!(!restored.compensation().inventory_reserved);
    assert!(!restored.compensation().payment_processed);
}