#![cfg(feature = "worker")]

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

use scalo::worker::{
    AdaptiveWorkerPool, FanOutPolicy, FanOutResult, ScalingDecision, ScalingInput, WorkerPoolConfig,
};

// --- Config tests ---

#[test]
fn test_default_config_has_sensible_values() {
    let cfg = WorkerPoolConfig::default();
    assert_eq!(cfg.min_threads, 2);
    assert_eq!(cfg.max_threads, 0);
    assert!((cfg.grow_below - 0.60).abs() < f64::EPSILON);
    assert!((cfg.shrink_above - 0.85).abs() < f64::EPSILON);
    assert!((cfg.emergency_above - 0.95).abs() < f64::EPSILON);
    assert!((cfg.memory_pressure_cap - 0.80).abs() < f64::EPSILON);
    assert_eq!(cfg.scale_interval_secs, 5);
    assert_eq!(cfg.async_concurrency, 32);
    assert_eq!(cfg.health_saturation_timeout_secs, 30);
}

#[test]
fn test_config_validation_rejects_min_greater_than_max() {
    let cfg = WorkerPoolConfig {
        min_threads: 16,
        max_threads: 4,
        ..Default::default()
    };
    assert!(cfg.validate().is_err());
}

#[test]
fn test_try_new_rejects_invalid_config() {
    // try_new validates after resolving max_threads -- an invalid config is
    // rejected at construction, not surfaced as a later scaler clamp panic.
    let bad = WorkerPoolConfig {
        min_threads: 16,
        max_threads: 4,
        ..Default::default()
    };
    assert!(AdaptiveWorkerPool::try_new(bad).is_err());

    let good = WorkerPoolConfig {
        min_threads: 1,
        max_threads: 0,
        ..Default::default()
    };
    assert!(AdaptiveWorkerPool::try_new(good).is_ok());
}

#[test]
fn test_config_validation_accepts_max_zero_auto_detect() {
    let cfg = WorkerPoolConfig {
        min_threads: 2,
        max_threads: 0,
        ..Default::default()
    };
    assert!(cfg.validate().is_ok());
}

#[test]
fn test_from_cascade_without_config_setup_returns_defaults() {
    // Regression test for hyperi-io/dfe-loader#19:
    // from_cascade() must not panic when config::setup() hasn't been called.
    // It should fall back to defaults gracefully.
    let cfg = WorkerPoolConfig::from_cascade("worker_pool")
        .expect("from_cascade should not panic without config::setup()");
    assert_eq!(cfg.min_threads, 2);
    assert_eq!(cfg.max_threads, 0); // auto-detect default
}

#[test]
fn test_config_validation_rejects_thresholds_out_of_order() {
    let cfg = WorkerPoolConfig {
        grow_below: 0.90,
        shrink_above: 0.50,
        ..Default::default()
    };
    assert!(cfg.validate().is_err());
}

#[test]
fn test_resolve_max_threads_auto_detect() {
    let mut cfg = WorkerPoolConfig::default();
    cfg.resolve_max_threads();
    assert!(cfg.max_threads >= 1);
}

// --- process_batch tests ---

#[test]
fn test_process_batch_executes_on_multiple_threads() {
    let config = WorkerPoolConfig {
        min_threads: 4,
        max_threads: 4,
        ..Default::default()
    };
    let pool = AdaptiveWorkerPool::new(config);

    let thread_ids = Arc::new(parking_lot::Mutex::new(std::collections::HashSet::new()));
    let items: Vec<i32> = (0..100).collect();

    let tids = thread_ids.clone();
    let results: Vec<Result<i32, String>> = pool.process_batch(&items, |&item| {
        tids.lock().insert(std::thread::current().id());
        std::thread::sleep(std::time::Duration::from_millis(1));
        Ok(item * 2)
    });

    assert_eq!(results.len(), 100);
    for (i, result) in results.iter().enumerate() {
        assert_eq!(
            *result.as_ref().unwrap(),
            (i32::try_from(i).unwrap()) * 2,
            "Wrong result at index {i}"
        );
    }
    let unique_threads = thread_ids.lock().len();
    assert!(
        unique_threads > 1,
        "Expected multiple threads, got {unique_threads}"
    );
}

#[test]
fn test_process_batch_respects_semaphore_throttle() {
    let config = WorkerPoolConfig {
        min_threads: 2,
        max_threads: 4,
        ..Default::default()
    };
    let pool = AdaptiveWorkerPool::new(config);

    let concurrent = Arc::new(AtomicUsize::new(0));
    let max_concurrent = Arc::new(AtomicUsize::new(0));
    let items: Vec<i32> = (0..20).collect();

    let c = concurrent.clone();
    let mc = max_concurrent.clone();
    let _results: Vec<Result<i32, String>> = pool.process_batch(&items, |&item| {
        let current = c.fetch_add(1, Ordering::SeqCst) + 1;
        mc.fetch_max(current, Ordering::SeqCst);
        std::thread::sleep(std::time::Duration::from_millis(10));
        c.fetch_sub(1, Ordering::SeqCst);
        Ok(item)
    });

    let observed_max = max_concurrent.load(Ordering::SeqCst);
    assert!(
        observed_max <= 2,
        "Expected max 2 concurrent, got {observed_max}"
    );
}

#[test]
fn test_process_batch_handles_errors() {
    let config = WorkerPoolConfig {
        min_threads: 2,
        max_threads: 2,
        ..Default::default()
    };
    let pool = AdaptiveWorkerPool::new(config);

    let items: Vec<i32> = (0..10).collect();
    let results: Vec<Result<i32, String>> = pool.process_batch(&items, |&item| {
        if item % 3 == 0 {
            Err(format!("bad item: {item}"))
        } else {
            Ok(item * 2)
        }
    });

    assert_eq!(results.len(), 10);
    assert!(results[0].is_err());
    assert!(results[1].is_ok());
    assert!(results[3].is_err());
}

#[test]
fn test_process_batch_empty_input() {
    let pool = AdaptiveWorkerPool::new(WorkerPoolConfig::default());
    let items: Vec<i32> = vec![];
    let results: Vec<Result<i32, String>> = pool.process_batch(&items, |&x| Ok(x));
    assert!(results.is_empty());
}

// --- fan_out_async tests ---

#[tokio::test]
async fn test_fan_out_async_preserves_order() {
    let config = WorkerPoolConfig {
        min_threads: 2,
        max_threads: 4,
        async_concurrency: 4,
        ..Default::default()
    };
    let pool = AdaptiveWorkerPool::new(config);

    let items: Vec<i32> = (0..20).collect();
    let results: Vec<Option<Result<i32, String>>> = pool
        .fan_out_async(&items, |&item| async move {
            tokio::time::sleep(std::time::Duration::from_millis(
                u64::try_from(20 - item).unwrap_or(0) % 10,
            ))
            .await;
            Ok(item * 3)
        })
        .await;

    assert_eq!(results.len(), 20);
    for (i, result) in results.iter().enumerate() {
        let val = result
            .as_ref()
            .expect("task should not panic")
            .as_ref()
            .expect("task returned Err");
        assert_eq!(
            *val,
            (i32::try_from(i).unwrap()) * 3,
            "Result at index {i} has wrong value"
        );
    }
}

#[tokio::test]
async fn test_fan_out_async_respects_concurrency_limit() {
    let config = WorkerPoolConfig {
        async_concurrency: 3,
        ..Default::default()
    };
    let pool = AdaptiveWorkerPool::new(config);

    let concurrent = Arc::new(AtomicUsize::new(0));
    let max_concurrent = Arc::new(AtomicUsize::new(0));
    let items: Vec<i32> = (0..12).collect();

    let c = concurrent.clone();
    let mc = max_concurrent.clone();
    let _results: Vec<Option<Result<i32, String>>> = pool
        .fan_out_async(&items, |&item| {
            let c = c.clone();
            let mc = mc.clone();
            async move {
                let current = c.fetch_add(1, Ordering::SeqCst) + 1;
                mc.fetch_max(current, Ordering::SeqCst);
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
                c.fetch_sub(1, Ordering::SeqCst);
                Ok(item)
            }
        })
        .await;

    let observed = max_concurrent.load(Ordering::SeqCst);
    assert!(observed <= 3, "Expected max 3 concurrent, got {observed}");
}

#[tokio::test]
async fn test_fan_out_async_empty_input() {
    let pool = AdaptiveWorkerPool::new(WorkerPoolConfig::default());
    let items: Vec<i32> = vec![];
    let results: Vec<Option<Result<i32, String>>> =
        pool.fan_out_async(&items, |&x| async move { Ok(x) }).await;
    assert!(results.is_empty());
}

// --- Scaling decision tests ---

fn scaling_input(cpu: f64, mem: f64, current: usize) -> ScalingInput {
    ScalingInput {
        cpu_util: cpu,
        memory_pressure: mem,
        current,
        min_threads: 2,
        max_threads: 8,
        grow_below: 0.60,
        shrink_above: 0.85,
        emergency_above: 0.95,
        memory_pressure_cap: 0.80,
    }
}

#[test]
fn test_scaling_decision_grow_when_cpu_low() {
    let decision = ScalingDecision::evaluate(&scaling_input(0.40, 0.20, 4));
    assert_eq!(decision.target, 6);
    assert_eq!(decision.direction, "up");
}

#[test]
fn test_scaling_decision_steady_in_dead_band() {
    let decision = ScalingDecision::evaluate(&scaling_input(0.72, 0.20, 4));
    assert_eq!(decision.target, 4);
    assert_eq!(decision.direction, "steady");
}

#[test]
fn test_scaling_decision_shrink_when_cpu_high() {
    let decision = ScalingDecision::evaluate(&scaling_input(0.90, 0.20, 6));
    assert_eq!(decision.target, 5);
    assert_eq!(decision.direction, "down");
}

#[test]
fn test_scaling_decision_emergency_shrink() {
    let decision = ScalingDecision::evaluate(&scaling_input(0.97, 0.20, 6));
    assert_eq!(decision.target, 4);
    assert_eq!(decision.direction, "emergency_down");
}

#[test]
fn test_scaling_decision_memory_cap_overrides_cpu() {
    let decision = ScalingDecision::evaluate(&scaling_input(0.40, 0.90, 6));
    assert_eq!(decision.target, 2);
    assert_eq!(decision.direction, "memory_cap");
}

#[test]
fn test_scaling_decision_respects_min_max_bounds() {
    // Try to grow past max
    let decision = ScalingDecision::evaluate(&scaling_input(0.30, 0.20, 7));
    assert_eq!(decision.target, 8);

    // Try to shrink below min
    let decision = ScalingDecision::evaluate(&scaling_input(0.97, 0.20, 3));
    assert_eq!(decision.target, 2);
}

// --- Graceful shutdown test ---

#[tokio::test]
async fn test_graceful_shutdown_drains_work() {
    let config = WorkerPoolConfig {
        min_threads: 2,
        max_threads: 4,
        ..Default::default()
    };
    let pool = Arc::new(AdaptiveWorkerPool::new(config));
    let cancel = tokio_util::sync::CancellationToken::new();

    pool.start_scaling_loop(cancel.clone());

    // Submit some work
    let items: Vec<i32> = (0..10).collect();
    let results: Vec<Result<i32, String>> = pool.process_batch(&items, |&x| Ok(x));
    assert_eq!(results.len(), 10);

    // Shutdown scaler
    cancel.cancel();
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    // Pool still usable after scaler stops — just no more scaling adjustments
    let more: Vec<Result<i32, String>> = pool.process_batch(&[42], |&x| Ok(x));
    assert_eq!(more.len(), 1);
    assert_eq!(*more[0].as_ref().unwrap(), 42);
}

// --- Pool active_threads test ---

#[test]
fn test_map_owned_throttles_to_target_not_rayon_size() {
    // Pool with 4 rayon threads but target=1 (min_threads=1). map_owned must
    // cap concurrent execution at the target (1), proving the parsed path is
    // throttled by the scaler -- not by the rayon pool size. Without the
    // semaphore (the old install() path) peak concurrency would reach 4.
    let config = WorkerPoolConfig {
        min_threads: 1,
        max_threads: 4,
        ..Default::default()
    };
    let pool = AdaptiveWorkerPool::new(config);
    assert_eq!(pool.target_threads(), 1);

    let in_flight = Arc::new(AtomicUsize::new(0));
    let peak = Arc::new(AtomicUsize::new(0));
    let items: Vec<usize> = (0..40).collect();

    let inflight2 = Arc::clone(&in_flight);
    let peak2 = Arc::clone(&peak);
    let out: Vec<usize> = pool.map_owned(items, move |i| {
        let now = inflight2.fetch_add(1, Ordering::SeqCst) + 1;
        peak2.fetch_max(now, Ordering::SeqCst);
        // Hold the permit briefly so overlap would occur if unthrottled.
        std::thread::sleep(std::time::Duration::from_millis(2));
        inflight2.fetch_sub(1, Ordering::SeqCst);
        i * 2
    });

    assert_eq!(out.len(), 40);
    assert_eq!(out[10], 20, "map_owned preserves per-item mapping/order");
    assert_eq!(
        peak.load(Ordering::SeqCst),
        1,
        "concurrent execution never exceeded target=1"
    );
}

#[test]
fn test_thread_accounting_target_leased_available() {
    let config = WorkerPoolConfig {
        min_threads: 3,
        max_threads: 8,
        ..Default::default()
    };
    let pool = AdaptiveWorkerPool::new(config);
    // Idle pool: zero in-flight, regardless of the scaler target.
    assert_eq!(pool.active_threads(), 0, "idle pool reports zero active");
    // Target starts at min_threads; headroom equals the target when idle.
    assert_eq!(pool.target_threads(), 3, "initial target is min_threads");
    assert_eq!(pool.available_threads(), 3, "idle headroom equals target");
    assert_eq!(pool.max_threads(), 8);
}

// =============================================================================
// Adversarial / edge case tests
// =============================================================================

#[test]
fn test_process_batch_panic_in_closure_does_not_crash_pool() {
    // If a closure panics, rayon propagates it. The pool should still be usable after.
    let config = WorkerPoolConfig {
        min_threads: 2,
        max_threads: 2,
        ..Default::default()
    };
    let pool = AdaptiveWorkerPool::new(config);

    // First batch: one item panics
    let items: Vec<i32> = vec![1, 2, 3];
    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        pool.process_batch(&items, |&item| -> Result<i32, String> {
            assert!(item != 2, "deliberate panic in closure");
            Ok(item)
        })
    }));
    assert!(result.is_err(), "panic should propagate from process_batch");

    // Pool should still be usable after the panic
    let items2: Vec<i32> = vec![10, 20, 30];
    let results: Vec<Result<i32, String>> = pool.process_batch(&items2, |&item| Ok(item * 2));
    assert_eq!(results.len(), 3);
    assert_eq!(*results[0].as_ref().unwrap(), 20);
    assert_eq!(*results[1].as_ref().unwrap(), 40);
    assert_eq!(*results[2].as_ref().unwrap(), 60);
}

#[test]
fn test_process_batch_all_errors() {
    let pool = AdaptiveWorkerPool::new(WorkerPoolConfig::default());
    let items: Vec<i32> = (0..50).collect();
    let results: Vec<Result<i32, String>> =
        pool.process_batch(&items, |&item| Err(format!("every item fails: {item}")));
    assert_eq!(results.len(), 50);
    assert!(results.iter().all(Result::is_err));
}

#[test]
fn test_process_batch_single_item() {
    let pool = AdaptiveWorkerPool::new(WorkerPoolConfig::default());
    let items = vec![42];
    let results: Vec<Result<i32, String>> = pool.process_batch(&items, |&x| Ok(x * 2));
    assert_eq!(results.len(), 1);
    assert_eq!(*results[0].as_ref().unwrap(), 84);
}

#[test]
fn test_process_batch_large_batch_stress() {
    // 10,000 items should process without issues
    let config = WorkerPoolConfig {
        min_threads: 4,
        max_threads: 4,
        ..Default::default()
    };
    let pool = AdaptiveWorkerPool::new(config);
    let items: Vec<i32> = (0..10_000).collect();
    let results: Vec<Result<i64, String>> = pool.process_batch(&items, |&item| {
        // Some CPU work to exercise thread scheduling
        Ok(i64::from(item) * i64::from(item))
    });
    assert_eq!(results.len(), 10_000);
    // Verify ordering preserved
    for (i, result) in results.iter().enumerate() {
        let expected = i64::try_from(i).unwrap() * i64::try_from(i).unwrap();
        assert_eq!(
            *result.as_ref().unwrap(),
            expected,
            "ordering broken at index {i}"
        );
    }
}

#[test]
fn test_config_validation_grow_equals_shrink_rejected() {
    let cfg = WorkerPoolConfig {
        grow_below: 0.80,
        shrink_above: 0.80, // equal — no dead band
        ..Default::default()
    };
    assert!(cfg.validate().is_err());
}

#[test]
fn test_config_validation_emergency_below_shrink_rejected() {
    let cfg = WorkerPoolConfig {
        shrink_above: 0.95,
        emergency_above: 0.90, // below shrink
        ..Default::default()
    };
    assert!(cfg.validate().is_err());
}

#[test]
fn test_config_min_threads_zero_rejected_by_validate() {
    // Codex F11: min_threads=0 leaves the rayon semaphore spinning
    // in yield_now() forever. validate() now rejects upfront.
    let config = WorkerPoolConfig {
        min_threads: 0,
        max_threads: 4,
        ..Default::default()
    };
    assert!(config.validate().is_err());
}

#[test]
fn test_scaling_decision_boundary_exactly_at_grow_below() {
    // CPU exactly at grow_below threshold — should be in steady band (not grow)
    let input = ScalingInput {
        cpu_util: 0.60, // exactly at grow_below
        memory_pressure: 0.20,
        current: 4,
        min_threads: 2,
        max_threads: 8,
        grow_below: 0.60,
        shrink_above: 0.85,
        emergency_above: 0.95,
        memory_pressure_cap: 0.80,
    };
    let decision = ScalingDecision::evaluate(&input);
    // At exactly grow_below: cpu_util < grow_below is FALSE (0.60 < 0.60 = false)
    // So it falls to cpu_util <= shrink_above (0.60 <= 0.85 = true) → steady
    assert_eq!(decision.direction, "steady");
}

#[test]
fn test_scaling_decision_boundary_exactly_at_shrink_above() {
    let input = ScalingInput {
        cpu_util: 0.85, // exactly at shrink_above
        memory_pressure: 0.20,
        current: 4,
        min_threads: 2,
        max_threads: 8,
        grow_below: 0.60,
        shrink_above: 0.85,
        emergency_above: 0.95,
        memory_pressure_cap: 0.80,
    };
    let decision = ScalingDecision::evaluate(&input);
    // cpu_util <= shrink_above (0.85 <= 0.85 = true) → steady
    assert_eq!(decision.direction, "steady");
}

#[test]
fn test_scaling_decision_boundary_exactly_at_emergency() {
    let input = ScalingInput {
        cpu_util: 0.95, // exactly at emergency_above
        memory_pressure: 0.20,
        current: 4,
        min_threads: 2,
        max_threads: 8,
        grow_below: 0.60,
        shrink_above: 0.85,
        emergency_above: 0.95,
        memory_pressure_cap: 0.80,
    };
    let decision = ScalingDecision::evaluate(&input);
    // cpu_util <= emergency_above (0.95 <= 0.95 = true) → down (not emergency)
    assert_eq!(decision.direction, "down");
}

#[test]
fn test_scaling_decision_memory_exactly_at_cap() {
    let input = ScalingInput {
        cpu_util: 0.40,
        memory_pressure: 0.80, // exactly at memory_pressure_cap
        current: 6,
        min_threads: 2,
        max_threads: 8,
        grow_below: 0.60,
        shrink_above: 0.85,
        emergency_above: 0.95,
        memory_pressure_cap: 0.80,
    };
    let decision = ScalingDecision::evaluate(&input);
    // memory_pressure > memory_pressure_cap (0.80 > 0.80 = false) → NOT memory_cap
    // Falls through to cpu check: 0.40 < 0.60 → grow
    assert_eq!(decision.direction, "up");
}

#[tokio::test]
async fn test_fan_out_async_with_all_failures() {
    let pool = AdaptiveWorkerPool::new(WorkerPoolConfig::default());
    let items: Vec<i32> = (0..10).collect();
    let results: Vec<Option<Result<i32, String>>> = pool
        .fan_out_async(&items, |&item| async move { Err(format!("fail: {item}")) })
        .await;
    assert_eq!(results.len(), 10);
    // Every slot is `Some(Err(...))` — no task panicked, every one returned Err.
    assert!(results.iter().all(|r| matches!(r, Some(Err(_)))));
}

#[tokio::test]
async fn test_fan_out_async_mixed_success_failure_preserves_order() {
    let pool = AdaptiveWorkerPool::new(WorkerPoolConfig {
        async_concurrency: 4,
        ..Default::default()
    });
    let items: Vec<i32> = (0..20).collect();
    let results: Vec<Option<Result<i32, String>>> = pool
        .fan_out_async(&items, |&item| async move {
            // Variable delay to stress ordering
            tokio::time::sleep(std::time::Duration::from_millis(
                u64::try_from(item % 5).unwrap_or(0),
            ))
            .await;
            if item % 3 == 0 {
                Err(format!("fail: {item}"))
            } else {
                Ok(item * 10)
            }
        })
        .await;

    assert_eq!(results.len(), 20);
    // Verify ordering: indices that are multiples of 3 should be Err
    assert!(matches!(results[0], Some(Err(_)))); // 0 % 3 == 0
    assert!(matches!(results[3], Some(Err(_)))); // 3 % 3 == 0
    assert!(matches!(results[6], Some(Err(_)))); // 6 % 3 == 0
    assert_eq!(*results[1].as_ref().unwrap().as_ref().unwrap(), 10);
    assert_eq!(*results[2].as_ref().unwrap().as_ref().unwrap(), 20);
    assert_eq!(*results[4].as_ref().unwrap().as_ref().unwrap(), 40);
}

// --- fan_out_async_with_policy ---

#[tokio::test]
async fn fan_out_with_policy_times_out_slow_items_preserving_order() {
    let pool = AdaptiveWorkerPool::new(WorkerPoolConfig::default());
    let policy = FanOutPolicy {
        per_item_timeout: Some(std::time::Duration::from_millis(100)),
        cancel: None,
    };
    // Item 1 sleeps past the timeout; 0 and 2 are fast.
    let items = vec![0u64, 1, 2];
    let results: Vec<FanOutResult<u64, ()>> = pool
        .fan_out_async_with_policy(&items, &policy, |&n| async move {
            if n == 1 {
                tokio::time::sleep(std::time::Duration::from_millis(500)).await;
            }
            Ok(n * 10)
        })
        .await;

    assert_eq!(results.len(), 3);
    assert!(matches!(results[0], FanOutResult::Ok(0)));
    assert!(
        matches!(results[1], FanOutResult::TimedOut),
        "slow item times out"
    );
    assert!(matches!(results[2], FanOutResult::Ok(20)));
}

#[tokio::test]
async fn fan_out_with_policy_cancellation_stops_spawning() {
    let pool = AdaptiveWorkerPool::new(WorkerPoolConfig::default());
    let token = tokio_util::sync::CancellationToken::new();
    token.cancel(); // pre-cancelled
    let policy = FanOutPolicy {
        per_item_timeout: None,
        cancel: Some(token),
    };
    let items = vec![0u64, 1, 2];
    let results: Vec<FanOutResult<u64, ()>> = pool
        .fan_out_async_with_policy(&items, &policy, |&n| async move { Ok(n) })
        .await;
    // Nothing spawned -> all Cancelled.
    assert!(results.iter().all(|r| matches!(r, FanOutResult::Cancelled)));
}
