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

mod support;

use std::{sync::Arc, time::Duration};

use rs_zero::resil::{
    AdaptiveShedder, AdaptiveShedderConfig, BreakerConfig, BreakerError, BreakerPolicyConfig,
    BreakerState, ConcurrencyLimit, CpuUsageProvider, RollingWindow, SharedCircuitBreaker,
    WindowConfig, WindowOutcome,
};
use tokio::sync::Barrier;

use support::stress::eventually;

async fn open_breaker_with_one_failure(breaker: &SharedCircuitBreaker) {
    let _ = breaker
        .do_request(|| async { Err::<(), &'static str>("open") })
        .await;
    assert_eq!(breaker.state().await, BreakerState::Open);
}

async fn race_half_open_trials(breaker: SharedCircuitBreaker) -> usize {
    let barrier = Arc::new(Barrier::new(8));
    let mut handles = Vec::new();
    for _ in 0..8 {
        let breaker = breaker.clone();
        let barrier = barrier.clone();
        handles.push(tokio::spawn(async move {
            barrier.wait().await;
            match breaker.allow().await {
                Ok(guard) => {
                    tokio::time::sleep(Duration::from_millis(20)).await;
                    guard.record_success().await;
                    true
                }
                Err(_) => false,
            }
        }));
    }

    let mut allowed = 0usize;
    for handle in handles {
        if handle.await.expect("join") {
            allowed += 1;
        }
    }
    allowed
}

async fn seed_shedder_latency(shedder: &AdaptiveShedder) {
    for _ in 0..4 {
        let guard = shedder.allow().await.expect("seed");
        tokio::time::sleep(Duration::from_millis(2)).await;
        guard.record_success().await;
    }
}

async fn race_shedder_overload(shedder: AdaptiveShedder) -> (usize, usize) {
    let barrier = Arc::new(Barrier::new(12));
    let mut handles = Vec::new();
    for index in 0..12 {
        let shedder = shedder.clone();
        let barrier = barrier.clone();
        handles.push(tokio::spawn(async move {
            barrier.wait().await;
            match shedder.allow().await {
                Ok(guard) => {
                    if index % 2 == 0 {
                        drop(guard);
                    } else {
                        tokio::time::sleep(Duration::from_millis(5)).await;
                        guard.record_failure().await;
                    }
                    true
                }
                Err(_) => false,
            }
        }));
    }

    let mut allowed = 0usize;
    let mut rejected = 0usize;
    for handle in handles {
        if handle.await.expect("join") {
            allowed += 1;
        } else {
            rejected += 1;
        }
    }
    (allowed, rejected)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn concurrency_limit_rejects_contention_and_releases_after_completion() {
    let limit = ConcurrencyLimit::new(4);
    let barrier = Arc::new(Barrier::new(32));
    let mut handles = Vec::new();

    for _ in 0..32 {
        let limit = limit.clone();
        let barrier = barrier.clone();
        handles.push(tokio::spawn(async move {
            barrier.wait().await;
            limit
                .run(async {
                    tokio::time::sleep(Duration::from_millis(25)).await;
                    1usize
                })
                .await
                .map(|value| value == 1)
        }));
    }

    let mut allowed = 0usize;
    let mut rejected = 0usize;
    for handle in handles {
        match handle.await.expect("join") {
            Ok(true) => allowed += 1,
            Ok(false) => panic!("unexpected result"),
            Err(_) => rejected += 1,
        }
    }

    assert_eq!(allowed, 4);
    assert_eq!(rejected, 28);
    assert_eq!(limit.run(async { 99usize }).await.expect("released"), 99);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn shared_breaker_counts_failures_and_recovers_from_half_open_drop() {
    let breaker = SharedCircuitBreaker::new(BreakerConfig {
        failure_threshold: 4,
        reset_timeout: Duration::from_millis(30),
    });

    for _ in 0..4 {
        let error = breaker
            .do_request(|| async { Err::<(), &'static str>("backend down") })
            .await
            .expect_err("backend failure");
        assert!(matches!(error, rs_zero::resil::BreakerCallError::Inner(_)));
    }

    assert_eq!(breaker.state().await, BreakerState::Open);
    assert!(matches!(breaker.allow().await, Err(BreakerError::Open)));

    tokio::time::sleep(Duration::from_millis(40)).await;
    let dropped_trial = breaker.allow().await.expect("half-open trial");
    assert_eq!(breaker.snapshot().await.half_open_in_flight, 1);
    drop(dropped_trial);
    eventually(20, Duration::from_millis(5), || {
        let breaker = breaker.clone();
        async move { breaker.snapshot().await.half_open_in_flight == 0 }
    })
    .await;
    assert_eq!(breaker.state().await, BreakerState::Open);

    tokio::time::sleep(Duration::from_millis(40)).await;
    breaker
        .allow()
        .await
        .expect("second trial")
        .record_success()
        .await;
    assert_eq!(breaker.state().await, BreakerState::Closed);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn shared_breaker_limits_half_open_trial_concurrency() {
    let breaker = SharedCircuitBreaker::with_policy(
        BreakerConfig {
            failure_threshold: 1,
            reset_timeout: Duration::from_millis(20),
        },
        BreakerPolicyConfig {
            half_open_max_calls: 2,
            ..BreakerPolicyConfig::default()
        },
    );

    open_breaker_with_one_failure(&breaker).await;
    tokio::time::sleep(Duration::from_millis(30)).await;
    let allowed = race_half_open_trials(breaker.clone()).await;

    assert_eq!(allowed, 2);
    assert_eq!(breaker.state().await, BreakerState::Closed);
}

#[derive(Debug)]
struct FixedCpu(u32);

impl CpuUsageProvider for FixedCpu {
    fn cpu_usage_millis(&self) -> u32 {
        self.0
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn adaptive_shedder_rejects_overload_and_recovers_in_flight_after_guard_drop() {
    let shedder = AdaptiveShedder::with_cpu_usage_provider(
        AdaptiveShedderConfig {
            max_in_flight: 3,
            min_request_count: 1,
            max_latency: Duration::from_millis(1),
            overload_in_flight_percent: 10,
            cpu_threshold_millis: 800,
            min_overload_factor_percent: 10,
            window: WindowConfig {
                buckets: 4,
                bucket_duration: Duration::from_millis(20),
            },
            cool_off: Duration::from_millis(20),
        },
        Arc::new(FixedCpu(950)),
    );

    seed_shedder_latency(&shedder).await;
    let (allowed, rejected) = race_shedder_overload(shedder.clone()).await;

    assert!(allowed <= 4, "allowed={allowed}, rejected={rejected}");
    assert!(rejected >= 8, "allowed={allowed}, rejected={rejected}");
    eventually(20, Duration::from_millis(5), || {
        let shedder = shedder.clone();
        async move { shedder.snapshot().await.in_flight == 0 }
    })
    .await;

    let snapshot = shedder.snapshot().await;
    assert_eq!(snapshot.in_flight, 0);
    assert!(
        snapshot.window.drops >= rejected as u64,
        "snapshot: {snapshot:?}"
    );
    assert!(snapshot.window.failures >= 1, "snapshot: {snapshot:?}");
}

#[test]
fn rolling_window_rotates_and_preserves_live_bucket_invariants() {
    let mut window = RollingWindow::new(WindowConfig {
        buckets: 4,
        bucket_duration: Duration::from_millis(10),
    });

    for _ in 0..4 {
        window.record_with_latency(WindowOutcome::Success, Duration::from_millis(3));
        window.record(WindowOutcome::Failure);
        window.record(WindowOutcome::Drop);
        std::thread::sleep(Duration::from_millis(11));
    }

    let snapshot = window.snapshot();
    assert!(snapshot.total() <= 12, "snapshot: {snapshot:?}");
    assert!(snapshot.total() >= 6, "snapshot: {snapshot:?}");
    assert!(snapshot.failure_ratio() > 0.0);
    assert!(snapshot.drop_ratio() > 0.0);
    assert!(snapshot.average_latency().is_some());
    assert!(window.max_successes_per_bucket() >= 1);
    assert!(window.min_average_latency().is_some());
}
