#![cfg(all(feature = "cache-redis", feature = "resil"))]

use std::{process::Command, time::Duration};

use rs_zero::{
    cache_redis::RedisCacheConfig,
    resil::{RedisTokenLimiter, RedisTokenLimiterConfig, TokenLimiterRescueConfig},
};

#[tokio::test]
#[ignore = "requires RS_ZERO_TEST_REDIS_URL and a restartable Redis container"]
async fn redis_token_limiter_rescues_and_recovers_after_container_restart() {
    let url = std::env::var("RS_ZERO_TEST_REDIS_URL")
        .unwrap_or_else(|_| "redis://127.0.0.1:16379".to_string());
    let container = std::env::var("RS_ZERO_TEST_REDIS_CONTAINER")
        .expect("RS_ZERO_TEST_REDIS_CONTAINER is required for restart validation");
    let key = format!("recovery-{}", unique_suffix());
    let limiter = RedisTokenLimiter::new(RedisTokenLimiterConfig {
        redis: RedisCacheConfig {
            url,
            connect_timeout: Duration::from_millis(100),
            command_timeout: Duration::from_millis(100),
            ..RedisCacheConfig::default()
        },
        key_prefix: format!("rs-zero:limit:recovery:{}", unique_suffix()),
        rate_per_second: 1,
        burst: 2,
        rescue: TokenLimiterRescueConfig {
            enabled: true,
            ping_interval: Duration::from_millis(100),
        },
        ttl: Duration::from_secs(30),
        ..RedisTokenLimiterConfig::default()
    })
    .expect("limiter");

    assert!(limiter.allow(&key).await.expect("redis allow 1"));
    assert!(limiter.allow(&key).await.expect("redis allow 2"));
    assert!(!limiter.allow(&key).await.expect("redis rejects over quota"));

    docker(&["stop", &container]);
    wait_for_redis_down(&limiter, &key).await;
    assert!(!limiter.redis_alive());

    tokio::time::sleep(Duration::from_millis(1_200)).await;
    let allowed = limiter.allow(&key).await.expect("rescue allow 1");
    let rejected = limiter
        .allow_n(&key, 3)
        .await
        .expect("rescue rejects burst overflow");
    assert!(allowed, "local rescue should allow available burst token");
    assert!(!rejected, "local rescue must not fail-open over burst");

    docker(&["start", &container]);
    wait_for_redis_alive(&limiter).await;
    assert!(limiter.redis_alive());

    tokio::time::sleep(Duration::from_millis(1_200)).await;
    assert!(
        limiter
            .allow(&format!("{key}:after-recovery"))
            .await
            .expect("redis recovered allow")
    );
}

async fn wait_for_redis_down(limiter: &RedisTokenLimiter, key: &str) {
    for _ in 0..30 {
        let _ = limiter.allow(&format!("{key}:probe")).await;
        if !limiter.redis_alive() {
            return;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    panic!("Redis limiter did not enter rescue mode after container stop");
}

async fn wait_for_redis_alive(limiter: &RedisTokenLimiter) {
    for _ in 0..80 {
        if limiter.redis_alive() {
            return;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    panic!("Redis limiter did not recover after container start");
}

fn docker(args: &[&str]) {
    let status = Command::new("docker")
        .args(args)
        .status()
        .expect("docker command should start");
    assert!(status.success(), "docker {args:?} failed with {status}");
}

fn unique_suffix() -> String {
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();
    format!("{nanos}-{}", std::process::id())
}
