fraiseql-server 2.3.0

//! Federation Saga Validation Tests
//!
//! Tests saga orchestration across multiple services/databases:
//! 1. Multi-step saga execution (happy path and failures)
//! 2. Compensation and rollback (LIFO order)
//! 3. Deadletter queue for failed sagas
//! 4. Observer notifications on saga completion
//! 5. Trace context propagation
//! 6. Concurrent saga execution
//! 7. Idempotency guarantees
//!
//! # Scenario: `CreateOrderWithInventoryReservation`
//!
//! Across 2 databases:
//! - Order Service (Database 1): Create order
//! - Inventory Service (Database 2): Reserve inventory
//! - Compensation: Reverse both operations in LIFO order
//!
//! # Running Tests
//!
//! ```bash
//! cargo test --test federation_saga_validation_test -- --nocapture
//! ```
//!
//! **Execution engine:** none
//! **Infrastructure:** none
//! **Parallelism:** safe

#![cfg(test)]
#![allow(dead_code)]
#![allow(clippy::unwrap_used, clippy::panic)] // Reason: test code, panics acceptable
#![allow(clippy::cast_precision_loss)] // Reason: test metrics use usize/u64→f64 for reporting
#![allow(clippy::cast_sign_loss)] // Reason: test data uses small positive integers
#![allow(clippy::cast_possible_truncation)] // Reason: test data values are small and bounded
#![allow(clippy::cast_possible_wrap)] // Reason: test data values are small and bounded
#![allow(clippy::cast_lossless)] // Reason: test code readability
#![allow(clippy::missing_panics_doc)] // Reason: test helper functions, panics are expected
#![allow(clippy::missing_errors_doc)] // Reason: test helper functions
#![allow(missing_docs)] // Reason: test code does not require documentation
#![allow(clippy::items_after_statements)] // Reason: test helpers defined near use site
#![allow(clippy::used_underscore_binding)] // Reason: test variables prefixed with _ by convention
#![allow(clippy::needless_pass_by_value)] // Reason: test helper signatures follow test patterns

use std::collections::HashMap;

// Common test utilities
mod common;
#[allow(unused_imports)]
// Reason: shared harness re-exports many symbols; saga tests use a subset
use common::{SagaStepDef, SagaStepResult, TestSagaExecutor};

// ============================================================================
// Saga Domain Model
// ============================================================================

/// Unique identifier for a saga instance
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
struct SagaId(u64);

impl SagaId {
    /// Generate a new saga ID
    fn new(id: u64) -> Self {
        Self(id)
    }
}

/// A single step in a saga
#[derive(Debug, Clone)]
struct SagaStep {
    /// Step name (e.g., "`CreateOrder`", "`ReserveInventory`")
    name: String,

    /// Service that handles this step
    service: String,

    /// Database this step operates on
    database: String,

    /// Input data for this step
    input: HashMap<String, String>,

    /// Compensation step (to reverse this operation)
    compensation_name: Option<String>,
}

impl SagaStep {
    /// Create a new saga step
    fn new(name: &str, service: &str, database: &str) -> Self {
        Self {
            name:              name.to_string(),
            service:           service.to_string(),
            database:          database.to_string(),
            input:             HashMap::new(),
            compensation_name: None,
        }
    }

    /// Add input data
    fn with_input(mut self, key: &str, value: &str) -> Self {
        self.input.insert(key.to_string(), value.to_string());
        self
    }

    /// Set compensation step
    fn with_compensation(mut self, comp_name: &str) -> Self {
        self.compensation_name = Some(comp_name.to_string());
        self
    }
}

/// Status of a saga step execution
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StepStatus {
    /// Step not yet executed
    Pending,
    /// Step completed successfully
    Completed,
    /// Step failed
    Failed,
    /// Step is being compensated (rolled back)
    Compensating,
    /// Compensation completed
    Compensated,
}

/// Result of a saga step execution
#[derive(Debug, Clone)]
struct StepResult {
    /// Step name
    step_name: String,
    /// Execution status
    status:    StepStatus,
    /// Result data (if successful)
    result:    Option<HashMap<String, String>>,
    /// Error message (if failed)
    error:     Option<String>,
}

impl StepResult {
    /// Create a successful step result
    fn success(step_name: &str, result: HashMap<String, String>) -> Self {
        Self {
            step_name: step_name.to_string(),
            status:    StepStatus::Completed,
            result:    Some(result),
            error:     None,
        }
    }

    /// Create a failed step result
    fn failure(step_name: &str, error: &str) -> Self {
        Self {
            step_name: step_name.to_string(),
            status:    StepStatus::Failed,
            result:    None,
            error:     Some(error.to_string()),
        }
    }

    /// Create a compensated step result
    fn compensated(step_name: &str) -> Self {
        Self {
            step_name: step_name.to_string(),
            status:    StepStatus::Compensated,
            result:    None,
            error:     None,
        }
    }
}

/// Overall status of a saga
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SagaStatus {
    /// Saga not yet started
    Pending,
    /// Saga running (steps being executed)
    Running,
    /// Saga completed successfully
    Completed,
    /// Saga failed and is rolling back
    Compensating,
    /// Saga compensated (rolled back)
    Compensated,
    /// Saga moved to deadletter queue (unrecoverable failure)
    DeadLettered,
}

/// A saga execution tracking all steps
#[derive(Debug, Clone)]
struct SagaExecution {
    /// Saga ID
    id: SagaId,

    /// Saga name
    name: String,

    /// Steps in the saga
    steps: Vec<SagaStep>,

    /// Results of executed steps (in order)
    results: Vec<StepResult>,

    /// Current status
    status: SagaStatus,

    /// Trace ID for this saga (for logging/tracing)
    trace_id: String,

    /// When the saga was created
    created_at: u64,

    /// When the saga completed (or failed)
    completed_at: Option<u64>,
}

impl SagaExecution {
    /// Create a new saga execution
    fn new(id: SagaId, name: &str) -> Self {
        Self {
            id,
            name: name.to_string(),
            steps: vec![],
            results: vec![],
            status: SagaStatus::Pending,
            trace_id: format!("trace_{}", id.0),
            created_at: 0,
            completed_at: None,
        }
    }

    /// Add a step to the saga
    fn add_step(mut self, step: SagaStep) -> Self {
        self.steps.push(step);
        self
    }

    /// Mark saga as running
    fn mark_running(mut self) -> Self {
        self.status = SagaStatus::Running;
        self
    }

    /// Mark saga as completed
    fn mark_completed(mut self) -> Self {
        self.status = SagaStatus::Completed;
        self.completed_at = Some(1);
        self
    }

    /// Mark saga as failed
    fn mark_failed(mut self) -> Self {
        self.status = SagaStatus::Compensating;
        self
    }

    /// Mark saga as compensated
    fn mark_compensated(mut self) -> Self {
        self.status = SagaStatus::Compensated;
        self.completed_at = Some(1);
        self
    }

    /// Mark saga as dead lettered
    fn mark_deadlettered(mut self) -> Self {
        self.status = SagaStatus::DeadLettered;
        self.completed_at = Some(1);
        self
    }

    /// Add a step result
    fn add_result(mut self, result: StepResult) -> Self {
        self.results.push(result);
        self
    }

    /// Check if saga succeeded
    fn is_successful(&self) -> bool {
        self.status == SagaStatus::Completed
            && self.results.iter().all(|r| r.status == StepStatus::Completed)
    }

    /// Check if saga was compensated
    fn is_compensated(&self) -> bool {
        self.status == SagaStatus::Compensated
    }

    /// Get the order of steps that were executed
    fn executed_steps(&self) -> Vec<&str> {
        self.results.iter().map(|r| r.step_name.as_str()).collect()
    }

    /// Get compensation order (LIFO)
    fn compensation_order(&self) -> Vec<&str> {
        // Compensation should be in reverse order (LIFO)
        self.executed_steps().into_iter().rev().collect()
    }
}

// ============================================================================
// Saga Execution Tests
// ============================================================================

/// Test 1: Two-step saga success (happy path)
#[test]
fn test_saga_two_step_success() {
    // Create order saga: CreateOrder -> ReserveInventory
    let saga = SagaExecution::new(SagaId::new(1), "CreateOrderWithInventoryReservation")
        .add_step(
            SagaStep::new("CreateOrder", "OrderService", "orders_db")
                .with_input("orderId", "order_123")
                .with_input("customerId", "customer_456")
                .with_compensation("CancelOrder"),
        )
        .add_step(
            SagaStep::new("ReserveInventory", "InventoryService", "inventory_db")
                .with_input("productId", "product_789")
                .with_input("quantity", "5")
                .with_compensation("ReleaseInventory"),
        );

    // Saga should have 2 steps
    assert_eq!(saga.steps.len(), 2);

    // First step should be CreateOrder
    assert_eq!(saga.steps[0].name, "CreateOrder");
    assert_eq!(saga.steps[0].service, "OrderService");
    assert_eq!(saga.steps[0].database, "orders_db");

    // Second step should be ReserveInventory
    assert_eq!(saga.steps[1].name, "ReserveInventory");
    assert_eq!(saga.steps[1].service, "InventoryService");
    assert_eq!(saga.steps[1].database, "inventory_db");

    // Both steps should have compensation
    assert!(saga.steps[0].compensation_name.is_some());
    assert!(saga.steps[1].compensation_name.is_some());
}

/// Test 2: Saga partial success (one step fails)
#[test]
fn test_saga_partial_success() {
    // Simulate: CreateOrder succeeds, ReserveInventory fails
    let saga = SagaExecution::new(SagaId::new(2), "CreateOrderWithInventoryReservation")
        .add_step(
            SagaStep::new("CreateOrder", "OrderService", "orders_db")
                .with_compensation("CancelOrder"),
        )
        .add_step(
            SagaStep::new("ReserveInventory", "InventoryService", "inventory_db")
                .with_compensation("ReleaseInventory"),
        )
        .mark_running()
        .add_result(StepResult::success("CreateOrder", {
            let mut r = HashMap::new();
            r.insert("orderId".to_string(), "order_123".to_string());
            r
        }))
        .add_result(StepResult::failure("ReserveInventory", "Insufficient inventory"));

    // Saga should be in failed state
    assert!(saga.status == SagaStatus::Running || saga.status == SagaStatus::Compensating);

    // First step should have completed
    assert_eq!(saga.results[0].status, StepStatus::Completed);

    // Second step should have failed
    assert_eq!(saga.results[1].status, StepStatus::Failed);
    assert_eq!(saga.results[1].error, Some("Insufficient inventory".to_string()));

    // Saga should not be successful
    assert!(!saga.is_successful());
}

// ============================================================================
// Compensation & Rollback Tests
// ============================================================================

/// Test 3: Saga compensation in LIFO order
#[test]
fn test_saga_compensation_rollback() {
    // Build a saga that executed multiple steps successfully,
    // but then needs to be rolled back
    let saga = SagaExecution::new(SagaId::new(3), "CreateOrderWithInventoryReservation")
        .add_step(
            SagaStep::new("CreateOrder", "OrderService", "orders_db")
                .with_compensation("CancelOrder"),
        )
        .add_step(
            SagaStep::new("ReserveInventory", "InventoryService", "inventory_db")
                .with_compensation("ReleaseInventory"),
        )
        .mark_running()
        .add_result(StepResult::success("CreateOrder", HashMap::new()))
        .add_result(StepResult::success("ReserveInventory", HashMap::new()))
        .mark_failed(); // Now it's in compensating state

    // Verify compensation order is LIFO (Last In, First Out)
    // Compensation should be: ReleaseInventory first, then CancelOrder
    let comp_order = saga.compensation_order();
    assert_eq!(comp_order.len(), 2);
    assert_eq!(comp_order[0], "ReserveInventory"); // Most recent first
    assert_eq!(comp_order[1], "CreateOrder"); // Earlier step second
}

/// Test 4: Deadletter queue for unrecoverable failures
#[test]
fn test_saga_deadletter_queue() {
    // Saga that failed and even compensation failed
    let saga = SagaExecution::new(SagaId::new(4), "CreateOrderWithInventoryReservation")
        .add_step(
            SagaStep::new("CreateOrder", "OrderService", "orders_db")
                .with_compensation("CancelOrder"),
        )
        .add_step(
            SagaStep::new("ReserveInventory", "InventoryService", "inventory_db")
                .with_compensation("ReleaseInventory"),
        )
        .mark_running()
        .add_result(StepResult::success("CreateOrder", HashMap::new()))
        .add_result(StepResult::failure("ReserveInventory", "Network timeout"))
        .add_result(StepResult::failure("ReleaseInventory", "Service unreachable"))
        .mark_deadlettered(); // Moved to DLQ

    // Saga should be in deadletter state
    assert_eq!(saga.status, SagaStatus::DeadLettered);

    // Should have completed_at timestamp
    assert!(saga.completed_at.is_some());

    // Should have multiple failed results
    assert!(saga.results.iter().any(|r| r.error.is_some()));
}

// ============================================================================
// Saga Orchestration Tests
// ============================================================================

/// Test 5: Observer notification on saga completion
#[test]
fn test_saga_observer_notification() {
    // After saga completes, observers should be notified
    let saga = SagaExecution::new(SagaId::new(5), "CreateOrderWithInventoryReservation")
        .add_step(SagaStep::new("CreateOrder", "OrderService", "orders_db"))
        .add_step(SagaStep::new("ReserveInventory", "InventoryService", "inventory_db"))
        .mark_running()
        .add_result(StepResult::success("CreateOrder", HashMap::new()))
        .add_result(StepResult::success("ReserveInventory", HashMap::new()))
        .mark_completed();

    // Saga should be completed
    assert_eq!(saga.status, SagaStatus::Completed);

    // Should have completed_at timestamp for observer notification
    assert!(saga.completed_at.is_some());

    // All steps should be successful
    assert!(saga.is_successful());
}

/// Test 6: Trace context propagation across steps
#[test]
fn test_saga_tracing_context() {
    // Trace IDs should be propagated across all steps
    let saga = SagaExecution::new(SagaId::new(6), "CreateOrderWithInventoryReservation")
        .add_step(SagaStep::new("CreateOrder", "OrderService", "orders_db"))
        .add_step(SagaStep::new("ReserveInventory", "InventoryService", "inventory_db"));

    // Saga should have a trace ID
    assert!(!saga.trace_id.is_empty());
    assert!(saga.trace_id.contains("trace_"));

    // Each step should be able to use this trace ID for logging
    for (i, step) in saga.steps.iter().enumerate() {
        // Trace ID should be consistent across steps
        assert_eq!(step.service, ["OrderService", "InventoryService"][i]);
    }
}

// ============================================================================
// Advanced Scenarios
// ============================================================================

/// Test 7: Concurrent saga execution
#[test]
fn test_saga_concurrent_execution() {
    // Multiple sagas running concurrently should not interfere
    let saga1 = SagaExecution::new(SagaId::new(7), "CreateOrderWithInventoryReservation")
        .add_step(SagaStep::new("CreateOrder", "OrderService", "orders_db"))
        .mark_running();

    let saga2 = SagaExecution::new(SagaId::new(8), "CreateOrderWithInventoryReservation")
        .add_step(SagaStep::new("CreateOrder", "OrderService", "orders_db"))
        .mark_running();

    // Each saga should have unique ID and trace ID
    assert_ne!(saga1.id, saga2.id);
    assert_ne!(saga1.trace_id, saga2.trace_id);

    // Both should be in running state
    assert_eq!(saga1.status, SagaStatus::Running);
    assert_eq!(saga2.status, SagaStatus::Running);
}

/// Test 8: Saga idempotency
#[test]
fn test_saga_idempotency() {
    // Saga executed twice with same inputs should produce same results

    // First execution
    let saga1 = SagaExecution::new(SagaId::new(9), "CreateOrderWithInventoryReservation")
        .add_step(
            SagaStep::new("CreateOrder", "OrderService", "orders_db")
                .with_input("orderId", "order_same_123")
                .with_input("customerId", "customer_456"),
        )
        .mark_running()
        .add_result(StepResult::success("CreateOrder", {
            let mut r = HashMap::new();
            r.insert("orderId".to_string(), "order_same_123".to_string());
            r
        }))
        .mark_completed();

    // Second execution with same inputs
    let saga2 = SagaExecution::new(SagaId::new(10), "CreateOrderWithInventoryReservation")
        .add_step(
            SagaStep::new("CreateOrder", "OrderService", "orders_db")
                .with_input("orderId", "order_same_123") // Same order ID
                .with_input("customerId", "customer_456"),
        )
        .mark_running()
        .add_result(StepResult::success("CreateOrder", {
            let mut r = HashMap::new();
            r.insert("orderId".to_string(), "order_same_123".to_string());
            r
        }))
        .mark_completed();

    // Both sagas should have same input (same order ID)
    assert_eq!(saga1.steps[0].input.get("orderId"), saga2.steps[0].input.get("orderId"));

    // Both should produce successful results
    assert!(saga1.is_successful());
    assert!(saga2.is_successful());

    // Results should match
    assert_eq!(saga1.results[0].status, saga2.results[0].status);
}

// ============================================================================
// Saga Execution (Async)
// ============================================================================

/// Execute two-step saga forward phase
#[tokio::test]
async fn test_saga_forward_phase_execution() {
    let mut executor = TestSagaExecutor::new();

    let steps = vec![
        SagaStepDef::new(1, "order-service", "orders", serde_json::json!({"orderId": "order_123"}))
            .with_compensation("cancel_order"),
        SagaStepDef::new(
            2,
            "inventory-service",
            "inventory",
            serde_json::json!({"productId": "prod_456"}),
        )
        .with_compensation("release_inventory"),
    ];

    let results = executor
        .execute_saga("saga-123", steps.clone())
        .await
        .unwrap_or_else(|e| panic!("Saga execution should succeed: {e}"));
    assert_eq!(results.len(), 2, "Should have executed 2 steps");
    assert!(results[0].success, "First step should succeed");
    assert!(results[1].success, "Second step should succeed");
    assert_eq!(results[0].step_number, 1);
    assert_eq!(results[1].step_number, 2);
}

/// Verify LIFO compensation order
#[tokio::test]
async fn test_saga_lifo_compensation_order() {
    let executor = TestSagaExecutor::new();

    let forward_steps = vec![
        SagaStepDef::new(1, "order-service", "orders", serde_json::json!({}))
            .with_compensation("cancel_order"),
        SagaStepDef::new(2, "inventory-service", "inventory", serde_json::json!({}))
            .with_compensation("release_inventory"),
        SagaStepDef::new(3, "payment-service", "payments", serde_json::json!({}))
            .with_compensation("refund_payment"),
    ];

    // Simulate compensation results in LIFO order (step 3, 2, 1)
    let compensation_results = vec![
        SagaStepResult::success(3, serde_json::json!({"refunded": true})),
        SagaStepResult::success(2, serde_json::json!({"released": true})),
        SagaStepResult::success(1, serde_json::json!({"cancelled": true})),
    ];

    // Verify LIFO order
    let order_check = executor.verify_lifo_order(&forward_steps, &compensation_results);
    if let Err(e) = order_check {
        panic!("Compensation order verification failed: {}", e);
    }
}

/// Verify multi-step saga with correct step execution
#[tokio::test]
async fn test_multi_step_saga_execution() {
    let mut executor = TestSagaExecutor::new();

    let steps = vec![
        SagaStepDef::new(
            1,
            "order-service",
            "orders",
            serde_json::json!({"customerId": "cust_123"}),
        ),
        SagaStepDef::new(
            2,
            "inventory-service",
            "inventory",
            serde_json::json!({"productId": "prod_456", "quantity": 5}),
        ),
        SagaStepDef::new(3, "payment-service", "payments", serde_json::json!({"amount": 100.00})),
    ];

    let results = executor
        .execute_saga("multi-saga", steps)
        .await
        .unwrap_or_else(|e| panic!("expected Ok executing multi-step saga: {e}"));
    assert_eq!(results.len(), 3);

    // Verify each step has correct metadata
    for (idx, result) in results.iter().enumerate() {
        assert!(result.success);
        assert_eq!(result.step_number, idx + 1);
        assert!(result.data.is_some());
    }
}

// ============================================================================
// Summary: 11 tests covering saga execution, compensation, orchestration,
// and advanced scenarios (concurrent execution, idempotency).
// ============================================================================