//! Outbox transactional consistency tests with fault injection.
//!
//! These tests verify that cross-partition writes (sub-orchestration starts,
//! completions) are delivered reliably even when the immediate best-effort
//! delivery fails. The background reconciler picks up pending intents and
//! delivers them.
//!
//! Architecture:
//! 1. ack_orchestration_item writes outbox intents atomically in the source
//!    partition's transactional batch
//! 2. Best-effort delivery attempts to create target documents immediately
//! 3. If best-effort fails (simulated by fault injection), intents remain
//!    as "pending" documents in the source partition
//! 4. The background reconciler queries pending intents and delivers them

use duroxide::runtime::registry::ActivityRegistry;
use duroxide::runtime::{self};
use duroxide::{ActivityContext, Client, OrchestrationContext, OrchestrationRegistry};
use duroxide_cdb::OutboxFaultInjector;
use std::sync::Arc;
use std::time::Duration;

mod common;

/// Sub-orchestration completes even when best-effort outbox delivery fails.
///
/// This test verifies the reconciler recovery path:
/// 1. Parent schedules a sub-orchestration (cross-partition write)
/// 2. Fault injection blocks the immediate delivery of the StartOrchestration
///    message to the child's partition
/// 3. The background reconciler (running every 500ms) picks up the pending
///    intent and delivers it
/// 4. The child runs, completes, and sends SubOrchCompleted back to parent
/// 5. Parent completes successfully
#[tokio::test]
async fn test_outbox_reconciler_delivers_after_best_effort_failure() {
    let (mut provider, container) = common::create_cosmos_provider_fast_reconciler().await;

    // Inject fault: first best-effort delivery attempt will be skipped
    provider.set_outbox_fault_injector(OutboxFaultInjector::fail_next(1));

    let store: Arc<dyn duroxide::providers::Provider> = Arc::new(provider);

    let activity_registry = ActivityRegistry::builder()
        .register("Echo", |_ctx: ActivityContext, input: String| async move {
            Ok(input)
        })
        .build();

    let child = |ctx: OrchestrationContext, input: String| async move {
        let r = ctx.schedule_activity("Echo", input).await?;
        Ok(r)
    };
    let parent = |ctx: OrchestrationContext, input: String| async move {
        let r = ctx
            .schedule_sub_orchestration("Child", input)
            .await
            .unwrap();
        Ok(format!("parent:{r}"))
    };

    let orchestration_registry = OrchestrationRegistry::builder()
        .register("Child", child)
        .register("Parent", parent)
        .build();

    let rt = runtime::Runtime::start_with_store(
        store.clone(),
        activity_registry,
        orchestration_registry,
    )
    .await;
    let client = Client::new(store.clone());

    client
        .start_orchestration("outbox-test-1", "Parent", "hello")
        .await
        .unwrap();

    // Give extra time for reconciler to deliver the intent (500ms interval + processing)
    match client
        .wait_for_orchestration("outbox-test-1", Duration::from_secs(15))
        .await
        .unwrap()
    {
        runtime::OrchestrationStatus::Completed { output, .. } => {
            assert_eq!(output, "parent:hello");
        }
        runtime::OrchestrationStatus::Failed { details, .. } => {
            panic!("Orchestration failed: {}", details.display_message());
        }
        other => panic!("Unexpected status: {:?}", other),
    }

    rt.shutdown(None).await;
    common::cleanup_container(&container).await;
}

/// Sub-orchestration fan-out works when multiple best-effort deliveries fail.
///
/// Parent fans out to 3 children. All 3 StartOrchestration messages have
/// their best-effort delivery blocked. The reconciler must deliver all 3.
#[tokio::test]
async fn test_outbox_reconciler_delivers_multiple_intents() {
    let (mut provider, container) = common::create_cosmos_provider_fast_reconciler().await;

    // Fail the first 3 best-effort delivery batches (one per child)
    provider.set_outbox_fault_injector(OutboxFaultInjector::fail_next(3));

    let store: Arc<dyn duroxide::providers::Provider> = Arc::new(provider);

    let activity_registry = ActivityRegistry::builder()
        .register(
            "Double",
            |_ctx: ActivityContext, input: String| async move {
                let n: i32 = input.parse().unwrap();
                Ok((n * 2).to_string())
            },
        )
        .build();

    let child = |ctx: OrchestrationContext, input: String| async move {
        let r = ctx.schedule_activity("Double", input).await?;
        Ok(r)
    };
    let parent = |ctx: OrchestrationContext, _input: String| async move {
        let a = ctx.schedule_sub_orchestration("ChildDouble", "10");
        let b = ctx.schedule_sub_orchestration("ChildDouble", "20");
        let c = ctx.schedule_sub_orchestration("ChildDouble", "30");
        let results = ctx.join(vec![a, b, c]).await;
        let sum: i32 = results
            .iter()
            .map(|r| r.as_ref().unwrap().parse::<i32>().unwrap())
            .sum();
        Ok(sum.to_string())
    };

    let orchestration_registry = OrchestrationRegistry::builder()
        .register("ChildDouble", child)
        .register("FanoutParent", parent)
        .build();

    let rt = runtime::Runtime::start_with_store(
        store.clone(),
        activity_registry,
        orchestration_registry,
    )
    .await;
    let client = Client::new(store.clone());

    client
        .start_orchestration("outbox-fanout-1", "FanoutParent", "go")
        .await
        .unwrap();

    match client
        .wait_for_orchestration("outbox-fanout-1", Duration::from_secs(20))
        .await
        .unwrap()
    {
        runtime::OrchestrationStatus::Completed { output, .. } => {
            // 10*2 + 20*2 + 30*2 = 120
            assert_eq!(output, "120");
        }
        runtime::OrchestrationStatus::Failed { details, .. } => {
            panic!("Orchestration failed: {}", details.display_message());
        }
        other => panic!("Unexpected status: {:?}", other),
    }

    rt.shutdown(None).await;
    common::cleanup_container(&container).await;
}

/// Normal path (no fault injection): sub-orchestration completes via
/// best-effort delivery without needing the reconciler.
/// This is the baseline to ensure we haven't broken the happy path.
#[tokio::test]
async fn test_outbox_best_effort_delivery_happy_path() {
    let (store, container) = common::create_cosmos_store().await;

    let activity_registry = ActivityRegistry::builder()
        .register(
            "Reverse",
            |_ctx: ActivityContext, input: String| async move {
                Ok(input.chars().rev().collect::<String>())
            },
        )
        .build();

    let child = |ctx: OrchestrationContext, input: String| async move {
        let r = ctx.schedule_activity("Reverse", input).await?;
        Ok(r)
    };
    let parent = |ctx: OrchestrationContext, input: String| async move {
        let r = ctx
            .schedule_sub_orchestration("ChildReverse", input)
            .await
            .unwrap();
        Ok(format!("result:{r}"))
    };

    let orchestration_registry = OrchestrationRegistry::builder()
        .register("ChildReverse", child)
        .register("ParentReverse", parent)
        .build();

    let rt = runtime::Runtime::start_with_store(
        store.clone(),
        activity_registry,
        orchestration_registry,
    )
    .await;
    let client = Client::new(store.clone());

    client
        .start_orchestration("outbox-happy-1", "ParentReverse", "abcdef")
        .await
        .unwrap();

    match client
        .wait_for_orchestration("outbox-happy-1", Duration::from_secs(10))
        .await
        .unwrap()
    {
        runtime::OrchestrationStatus::Completed { output, .. } => {
            assert_eq!(output, "result:fedcba");
        }
        runtime::OrchestrationStatus::Failed { details, .. } => {
            panic!("Orchestration failed: {}", details.display_message());
        }
        other => panic!("Unexpected status: {:?}", other),
    }

    rt.shutdown(None).await;
    common::cleanup_container(&container).await;
}
