//! Poison Message Handling Tests
//!
//! These tests validate the runtime behavior when messages exceed max_attempts.

mod common;

use std::sync::Arc;
use std::time::Duration;

use common::fault_injection::PoisonInjectingProvider;
use duroxide::providers::sqlite::SqliteProvider;
use duroxide::providers::{Provider, WorkItem};
use duroxide::runtime::registry::ActivityRegistry;
use duroxide::runtime::{self, OrchestrationStatus, RuntimeOptions};
use duroxide::{
    ActivityContext, Client, ErrorDetails, INITIAL_EXECUTION_ID, OrchestrationContext, OrchestrationRegistry,
    PoisonMessageType,
};

/// Test that attempt_count increments correctly when abandoning orchestration items
#[tokio::test]
async fn orchestration_attempt_count_increments_on_abandon() {
    let provider = Arc::new(
        SqliteProvider::new_in_memory()
            .await
            .expect("Failed to create provider"),
    );
    let lock_timeout = Duration::from_secs(5);

    // Enqueue a start orchestration
    provider
        .enqueue_for_orchestrator(
            WorkItem::StartOrchestration {
                instance: "attempt-test-1".to_string(),
                orchestration: "TestOrch".to_string(),
                input: "{}".to_string(),
                version: Some("1.0.0".to_string()),
                parent_instance: None,
                parent_id: None,
                execution_id: INITIAL_EXECUTION_ID,
            },
            None,
        )
        .await
        .expect("enqueue should succeed");

    // First fetch - attempt_count should be 1
    let (_item1, lock_token1, attempt_count1) = provider
        .fetch_orchestration_item(lock_timeout, Duration::ZERO)
        .await
        .expect("fetch should succeed")
        .expect("item should be present");
    assert_eq!(attempt_count1, 1, "First fetch should have attempt_count = 1");

    // Abandon and fetch again
    provider
        .abandon_orchestration_item(&lock_token1, None, false)
        .await
        .expect("abandon should succeed");

    let (_item2, lock_token2, attempt_count2) = provider
        .fetch_orchestration_item(lock_timeout, Duration::ZERO)
        .await
        .expect("fetch should succeed")
        .expect("item should be present");
    assert_eq!(attempt_count2, 2, "Second fetch should have attempt_count = 2");

    // Abandon and fetch a third time
    provider
        .abandon_orchestration_item(&lock_token2, None, false)
        .await
        .expect("abandon should succeed");

    let (_item3, _lock_token3, attempt_count3) = provider
        .fetch_orchestration_item(lock_timeout, Duration::ZERO)
        .await
        .expect("fetch should succeed")
        .expect("item should be present");
    assert_eq!(attempt_count3, 3, "Third fetch should have attempt_count = 3");
}

/// Test that attempt_count increments correctly for worker items
#[tokio::test]
async fn worker_attempt_count_increments_on_lock_expiry() {
    let provider = Arc::new(
        SqliteProvider::new_in_memory()
            .await
            .expect("Failed to create provider"),
    );
    let short_timeout = Duration::from_secs(1);

    // Enqueue an activity
    provider
        .enqueue_for_worker(WorkItem::ActivityExecute {
            instance: "attempt-test-2".to_string(),
            execution_id: INITIAL_EXECUTION_ID,
            id: 1,
            name: "TestActivity".to_string(),
            input: "{}".to_string(),
        })
        .await
        .expect("enqueue should succeed");

    // First fetch - attempt_count should be 1
    let (_item1, _token1, count1) = provider
        .fetch_work_item(short_timeout, Duration::ZERO)
        .await
        .expect("fetch should succeed")
        .expect("item should be present");
    assert_eq!(count1, 1, "First fetch should have attempt_count = 1");

    // Wait for lock to expire
    tokio::time::sleep(Duration::from_millis(1100)).await;

    // Second fetch after expiry - attempt_count should be 2
    let (_item2, token2, count2) = provider
        .fetch_work_item(short_timeout, Duration::ZERO)
        .await
        .expect("fetch should succeed")
        .expect("item should be present after lock expiry");
    assert_eq!(count2, 2, "Second fetch should have attempt_count = 2");

    // Ack to clean up
    provider
        .ack_work_item(
            &token2,
            WorkItem::ActivityCompleted {
                instance: "attempt-test-2".to_string(),
                execution_id: INITIAL_EXECUTION_ID,
                id: 1,
                result: "done".to_string(),
            },
        )
        .await
        .expect("ack should succeed");
}

/// Test that PoisonMessageType correctly identifies orchestrations
#[tokio::test]
async fn poison_message_type_orchestration() {
    let poison_type = PoisonMessageType::Orchestration {
        instance: "test-instance".to_string(),
        execution_id: 1,
    };

    let error = ErrorDetails::Poison {
        attempt_count: 11,
        max_attempts: 10,
        message_type: poison_type,
        message: "{}".to_string(),
    };

    assert_eq!(error.category(), "poison");
    assert!(!error.is_retryable());
    assert!(error.display_message().contains("orchestration"));
    assert!(error.display_message().contains("test-instance"));
    assert!(error.display_message().contains("11"));
    assert!(error.display_message().contains("10"));
}

/// Test that PoisonMessageType correctly identifies activities
#[tokio::test]
async fn poison_message_type_activity() {
    let poison_type = PoisonMessageType::Activity {
        instance: "test-instance".to_string(),
        execution_id: 1,
        activity_name: "TestActivity".to_string(),
        activity_id: 42,
    };

    let error = ErrorDetails::Poison {
        attempt_count: 5,
        max_attempts: 3,
        message_type: poison_type,
        message: "{}".to_string(),
    };

    assert_eq!(error.category(), "poison");
    assert!(!error.is_retryable());
    assert!(error.display_message().contains("activity"));
    assert!(error.display_message().contains("TestActivity"));
    assert!(error.display_message().contains("42"));
}

/// Test that RuntimeOptions has correct default max_attempts
#[tokio::test]
async fn runtime_options_default_max_attempts() {
    let options = RuntimeOptions::default();
    assert_eq!(options.max_attempts, 10, "Default max_attempts should be 10");
}

/// Test that max_attempts can be configured
#[tokio::test]
async fn runtime_options_configurable_max_attempts() {
    let options = RuntimeOptions {
        max_attempts: 5,
        ..Default::default()
    };
    assert_eq!(options.max_attempts, 5, "max_attempts should be configurable");
}

// =============================================================================
// E2E Tests with Fault Injection
// =============================================================================

fn fast_runtime_options(max_attempts: u32) -> RuntimeOptions {
    RuntimeOptions {
        max_attempts,
        dispatcher_min_poll_interval: Duration::from_millis(5),
        ..Default::default()
    }
}

/// E2E: Orchestration item detected as poison fails the orchestration
#[tokio::test]
async fn e2e_orchestration_item_poison_fails_orchestration() {
    let sqlite = Arc::new(
        SqliteProvider::new_in_memory()
            .await
            .expect("Failed to create provider"),
    );
    let provider = Arc::new(PoisonInjectingProvider::new(sqlite));

    // Register a simple orchestration that should never execute
    let orchestrations = OrchestrationRegistry::builder()
        .register("SimpleOrch", |_ctx: OrchestrationContext, _input: String| async move {
            panic!("Orchestration code should not run when detected as poison");
        })
        .build();

    let activities = ActivityRegistry::builder().build();

    // Inject poison BEFORE starting runtime - next fetch will see high attempt count
    provider.inject_orchestration_poison(11);

    let rt = runtime::Runtime::start_with_options(
        provider.clone() as Arc<dyn Provider>,
        Arc::new(activities),
        orchestrations,
        fast_runtime_options(10),
    )
    .await;

    let client = Client::new(provider.clone() as Arc<dyn Provider>);

    // Start orchestration
    let instance = "poison-orch-test-1";
    client
        .start_orchestration(instance, "SimpleOrch", "{}")
        .await
        .expect("start should succeed");

    // Wait for orchestration to complete (should fail as poison)
    let status = client
        .wait_for_orchestration(instance, Duration::from_secs(5))
        .await
        .expect("wait should succeed");

    // Should be failed with complete poison error details
    match status {
        OrchestrationStatus::Failed { details } => {
            if let ErrorDetails::Poison {
                attempt_count,
                max_attempts,
                message_type,
                message,
            } = details
            {
                assert_eq!(attempt_count, 11);
                assert_eq!(max_attempts, 10);

                // Verify message type contains correct instance and execution_id
                if let PoisonMessageType::Orchestration {
                    instance: inst,
                    execution_id,
                } = message_type
                {
                    assert_eq!(inst, instance);
                    assert_eq!(execution_id, INITIAL_EXECUTION_ID);
                } else {
                    panic!("Expected Orchestration message type");
                }

                // Message should be serialized work items
                assert!(!message.is_empty(), "Message should contain serialized data");
            } else {
                panic!("Expected Poison error, got: {:?}", details);
            }
        }
        other => panic!("Expected Failed status, got: {:?}", other),
    }

    rt.shutdown(None).await;
}

/// E2E: Activity item detected as poison fails the orchestration
#[tokio::test]
async fn e2e_activity_item_poison_fails_orchestration() {
    let sqlite = Arc::new(
        SqliteProvider::new_in_memory()
            .await
            .expect("Failed to create provider"),
    );
    let provider = Arc::new(PoisonInjectingProvider::new(sqlite));

    // Register orchestration that schedules an activity
    let orchestrations = OrchestrationRegistry::builder()
        .register(
            "OrchWithActivity",
            |ctx: OrchestrationContext, _input: String| async move {
                ctx.schedule_activity("TestActivity", "{}").into_activity().await
            },
        )
        .build();

    // Register activity that would succeed normally
    let activities = ActivityRegistry::builder()
        .register("TestActivity", |_ctx: ActivityContext, _input: String| async move {
            Ok::<_, String>("activity-result".to_string())
        })
        .build();

    let rt = runtime::Runtime::start_with_options(
        provider.clone() as Arc<dyn Provider>,
        Arc::new(activities),
        orchestrations,
        fast_runtime_options(10),
    )
    .await;

    let client = Client::new(provider.clone() as Arc<dyn Provider>);

    // Inject activity poison BEFORE starting - persistent mode ensures all activity fetches
    // will see the high attempt count
    provider.inject_activity_poison_persistent(11);

    let instance = "poison-activity-test-1";
    client
        .start_orchestration(instance, "OrchWithActivity", "{}")
        .await
        .expect("start should succeed");

    // Wait for orchestration to complete
    let status = client
        .wait_for_orchestration(instance, Duration::from_secs(5))
        .await
        .expect("wait should succeed");

    // Orchestration should fail with poison error details
    match status {
        OrchestrationStatus::Failed { details } => {
            if let ErrorDetails::Poison {
                attempt_count,
                max_attempts,
                message_type,
                ..
            } = details
            {
                assert_eq!(attempt_count, 11);
                assert_eq!(max_attempts, 10);

                // Verify activity message type
                if let PoisonMessageType::Activity {
                    instance: inst,
                    execution_id,
                    activity_name,
                    activity_id,
                } = message_type
                {
                    assert_eq!(inst, instance);
                    assert_eq!(execution_id, INITIAL_EXECUTION_ID);
                    assert_eq!(activity_name, "TestActivity");
                    assert!(activity_id > 0, "Activity ID should be positive");
                } else {
                    panic!("Expected Activity message type, got: {:?}", message_type);
                }
            } else {
                panic!("Expected Poison error, got: {:?}", details);
            }
        }
        other => panic!("Expected Failed status, got: {:?}", other),
    }

    rt.shutdown(None).await;
}

/// E2E: Sub-orchestration detected as poison fails the parent
#[tokio::test]
async fn e2e_sub_orchestration_poison_fails_parent() {
    let sqlite = Arc::new(
        SqliteProvider::new_in_memory()
            .await
            .expect("Failed to create provider"),
    );
    let provider = Arc::new(PoisonInjectingProvider::new(sqlite));

    // Register parent orchestration that calls child
    let orchestrations = OrchestrationRegistry::builder()
        .register("ParentOrch", |ctx: OrchestrationContext, _input: String| async move {
            ctx.schedule_sub_orchestration("ChildOrch", "{}")
                .into_sub_orchestration()
                .await
        })
        .register("ChildOrch", |_ctx: OrchestrationContext, _input: String| async move {
            Ok::<_, String>("child-result".to_string())
        })
        .build();

    let activities = ActivityRegistry::builder().build();

    let rt = runtime::Runtime::start_with_options(
        provider.clone() as Arc<dyn Provider>,
        Arc::new(activities),
        orchestrations,
        fast_runtime_options(10),
    )
    .await;

    let client = Client::new(provider.clone() as Arc<dyn Provider>);

    // Skip the first orchestration fetch (the parent), then poison the next (the child).
    provider.inject_orchestration_poison_after_skip(1, 11);

    let instance = "poison-suborg-test-1";
    client
        .start_orchestration(instance, "ParentOrch", "{}")
        .await
        .expect("start should succeed");

    // Wait for parent to complete
    let status = client
        .wait_for_orchestration(instance, Duration::from_secs(5))
        .await
        .expect("wait should succeed");

    // Parent should fail with poison error from child
    match status {
        OrchestrationStatus::Failed { details } => {
            if let ErrorDetails::Poison {
                attempt_count,
                max_attempts,
                message_type,
                message,
            } = details
            {
                assert_eq!(attempt_count, 11);
                assert_eq!(max_attempts, 10);

                // Verify child orchestration message type
                if let PoisonMessageType::Orchestration {
                    instance: child_inst,
                    execution_id,
                } = message_type
                {
                    // Child instance should be derived from parent
                    assert!(
                        child_inst.starts_with(instance),
                        "Child instance should start with parent instance"
                    );
                    assert_eq!(execution_id, INITIAL_EXECUTION_ID);
                } else {
                    panic!("Expected Orchestration message type, got: {:?}", message_type);
                }

                assert!(!message.is_empty(), "Message should contain serialized data");
            } else {
                panic!("Expected Poison error, got: {:?}", details);
            }
        }
        other => panic!("Expected Failed status, got: {:?}", other),
    }

    rt.shutdown(None).await;
}

/// E2E: One poisoned activity among several (sequential) fails the orchestration
#[tokio::test]
async fn e2e_one_poisoned_activity_among_many() {
    let sqlite = Arc::new(
        SqliteProvider::new_in_memory()
            .await
            .expect("Failed to create provider"),
    );
    let provider = Arc::new(PoisonInjectingProvider::new(sqlite));

    // Register orchestration that schedules multiple activities sequentially
    let orchestrations = OrchestrationRegistry::builder()
        .register(
            "MultiActivityOrch",
            |ctx: OrchestrationContext, _input: String| async move {
                // First activity
                let _r1 = ctx.schedule_activity("Activity1", "{}").into_activity().await?;
                // Second activity - this one will be poisoned
                let _r2 = ctx.schedule_activity("Activity2", "{}").into_activity().await?;
                // Third activity - should not run
                let r3 = ctx.schedule_activity("Activity3", "{}").into_activity().await?;
                Ok::<_, String>(r3)
            },
        )
        .build();

    // Register activities
    let activities = ActivityRegistry::builder()
        .register("Activity1", |_ctx: ActivityContext, _input: String| async move {
            Ok::<_, String>("result1".to_string())
        })
        .register("Activity2", |_ctx: ActivityContext, _input: String| async move {
            Ok::<_, String>("result2".to_string())
        })
        .register("Activity3", |_ctx: ActivityContext, _input: String| async move {
            Ok::<_, String>("result3".to_string())
        })
        .build();

    let rt = runtime::Runtime::start_with_options(
        provider.clone() as Arc<dyn Provider>,
        Arc::new(activities),
        orchestrations,
        fast_runtime_options(5),
    )
    .await;

    let client = Client::new(provider.clone() as Arc<dyn Provider>);

    // Inject activity poison in persistent mode BEFORE starting.
    // This will cause the first activity fetch to be detected as poison.
    provider.inject_activity_poison_persistent(6);

    let instance = "poison-multi-activity-test";
    client
        .start_orchestration(instance, "MultiActivityOrch", "{}")
        .await
        .expect("start should succeed");

    let status = client
        .wait_for_orchestration(instance, Duration::from_secs(5))
        .await
        .expect("wait should succeed");

    assert!(
        matches!(status, OrchestrationStatus::Failed { .. }),
        "Orchestration should fail due to one poisoned activity, got: {:?}",
        status
    );

    rt.shutdown(None).await;
}

/// E2E: Custom max_attempts is respected
#[tokio::test]
async fn e2e_custom_max_attempts_respected() {
    let sqlite = Arc::new(
        SqliteProvider::new_in_memory()
            .await
            .expect("Failed to create provider"),
    );
    let provider = Arc::new(PoisonInjectingProvider::new(sqlite));

    let orchestrations = OrchestrationRegistry::builder()
        .register(
            "CustomMaxOrch",
            |_ctx: OrchestrationContext, _input: String| async move { Ok::<_, String>("success".to_string()) },
        )
        .build();

    let activities = ActivityRegistry::builder().build();

    // Inject attempt_count = 4 which exceeds custom max of 3
    provider.inject_orchestration_poison(4);

    let rt = runtime::Runtime::start_with_options(
        provider.clone() as Arc<dyn Provider>,
        Arc::new(activities),
        orchestrations,
        fast_runtime_options(3), // custom max_attempts = 3
    )
    .await;

    let client = Client::new(provider.clone() as Arc<dyn Provider>);

    let instance = "custom-max-attempts-test";
    client
        .start_orchestration(instance, "CustomMaxOrch", "{}")
        .await
        .expect("start should succeed");

    let status = client
        .wait_for_orchestration(instance, Duration::from_secs(5))
        .await
        .expect("wait should succeed");

    match status {
        OrchestrationStatus::Failed { details } => {
            if let ErrorDetails::Poison {
                attempt_count,
                max_attempts,
                ..
            } = details
            {
                assert_eq!(attempt_count, 4);
                assert_eq!(max_attempts, 3);
            } else {
                panic!("Expected Poison error with custom max_attempts");
            }
        }
        other => panic!("Expected Failed status, got: {:?}", other),
    }

    rt.shutdown(None).await;
}

/// E2E: Activity in sub-orchestration is poisoned, fails the sub-orch, which fails the parent
#[tokio::test]
async fn e2e_activity_poisons_suborchestration_poisons_parent() {
    let sqlite = Arc::new(
        SqliteProvider::new_in_memory()
            .await
            .expect("Failed to create provider"),
    );
    let provider = Arc::new(PoisonInjectingProvider::new(sqlite));

    // Register parent orchestration that calls child
    // Child orchestration calls an activity
    let orchestrations = OrchestrationRegistry::builder()
        .register(
            "GrandparentOrch",
            |ctx: OrchestrationContext, _input: String| async move {
                ctx.schedule_sub_orchestration("ChildWithActivityOrch", "{}")
                    .into_sub_orchestration()
                    .await
            },
        )
        .register(
            "ChildWithActivityOrch",
            |ctx: OrchestrationContext, _input: String| async move {
                // This activity will be poisoned
                ctx.schedule_activity("ChildActivity", "{}").into_activity().await
            },
        )
        .build();

    // Register activity that would succeed normally
    let activities = ActivityRegistry::builder()
        .register("ChildActivity", |_ctx: ActivityContext, _input: String| async move {
            Ok::<_, String>("activity-result".to_string())
        })
        .build();

    let rt = runtime::Runtime::start_with_options(
        provider.clone() as Arc<dyn Provider>,
        Arc::new(activities),
        orchestrations,
        fast_runtime_options(10),
    )
    .await;

    let client = Client::new(provider.clone() as Arc<dyn Provider>);

    // Inject activity poison - this will poison the activity in the child orchestration
    provider.inject_activity_poison_persistent(11);

    let instance = "poison-chain-test";
    client
        .start_orchestration(instance, "GrandparentOrch", "{}")
        .await
        .expect("start should succeed");

    // Wait for grandparent to complete
    let status = client
        .wait_for_orchestration(instance, Duration::from_secs(5))
        .await
        .expect("wait should succeed");

    // Grandparent should fail: activity poison → child fails → grandparent fails
    match status {
        OrchestrationStatus::Failed { details } => {
            // The error propagated should be the activity poison error
            if let ErrorDetails::Poison {
                attempt_count,
                max_attempts,
                message_type,
                ..
            } = details
            {
                assert_eq!(attempt_count, 11);
                assert_eq!(max_attempts, 10);

                // Should be an Activity poison type (from the poisoned activity in child)
                if let PoisonMessageType::Activity { activity_name, .. } = message_type {
                    assert_eq!(activity_name, "ChildActivity");
                } else {
                    panic!("Expected Activity message type, got: {:?}", message_type);
                }
            } else {
                panic!("Expected Poison error, got: {:?}", details);
            }
        }
        other => panic!("Expected Failed status, got: {:?}", other),
    }

    rt.shutdown(None).await;
}
