trypema 1.0.1

use std::{env, time::Duration};

use super::runtime;

use crate::common::SuppressionFactorCacheMs;
use crate::hybrid::SyncIntervalMs;
use crate::{
    HardLimitFactor, LocalRateLimiterOptions, RateGroupSizeMs, RateLimit, RateLimitDecision,
    RateLimiter, RateLimiterOptions, RedisKey, RedisRateLimiterOptions, WindowSizeSeconds,
};

fn redis_url() -> String {
    env::var("REDIS_URL").unwrap_or_else(|_| {
        panic!(
            "REDIS_URL env var must be set for Redis-backed tests (e.g. REDIS_URL=redis://127.0.0.1:16379/)"
        )
    })
}

fn unique_prefix() -> RedisKey {
    let n: u64 = rand::random();
    RedisKey::try_from(format!("trypema_test_{n}")).unwrap()
}

fn key(s: &str) -> RedisKey {
    RedisKey::try_from(s.to_string()).unwrap()
}

async fn build_limiter(
    url: &str,
    window_size_seconds: u64,
    rate_group_size_ms: u64,
    hard_limit_factor: f64,
) -> std::sync::Arc<RateLimiter> {
    build_limiter_with_cache_ms(
        url,
        window_size_seconds,
        rate_group_size_ms,
        hard_limit_factor,
        *SuppressionFactorCacheMs::default(),
    )
    .await
}

async fn build_limiter_with_cache_ms(
    url: &str,
    window_size_seconds: u64,
    rate_group_size_ms: u64,
    hard_limit_factor: f64,
    suppression_factor_cache_ms: u64,
) -> std::sync::Arc<RateLimiter> {
    let client = redis::Client::open(url).unwrap();
    let cm = client.get_connection_manager().await.unwrap();
    let prefix = unique_prefix();

    let options = RateLimiterOptions {
        local: LocalRateLimiterOptions {
            window_size_seconds: WindowSizeSeconds::try_from(window_size_seconds).unwrap(),
            rate_group_size_ms: RateGroupSizeMs::try_from(rate_group_size_ms).unwrap(),
            hard_limit_factor: HardLimitFactor::try_from(hard_limit_factor).unwrap(),
            suppression_factor_cache_ms: SuppressionFactorCacheMs::try_from(
                suppression_factor_cache_ms,
            )
            .unwrap(),
        },
        redis: RedisRateLimiterOptions {
            connection_manager: cm,
            prefix: Some(prefix.clone()),
            window_size_seconds: WindowSizeSeconds::try_from(window_size_seconds).unwrap(),
            rate_group_size_ms: RateGroupSizeMs::try_from(rate_group_size_ms).unwrap(),
            hard_limit_factor: HardLimitFactor::try_from(hard_limit_factor).unwrap(),
            suppression_factor_cache_ms: SuppressionFactorCacheMs::try_from(
                suppression_factor_cache_ms,
            )
            .unwrap(),
            sync_interval_ms: SyncIntervalMs::default(),
        },
    };

    std::sync::Arc::new(RateLimiter::new(options))
}

#[test]
fn get_suppression_factor_fresh_key_returns_zero_and_sets_cache_ttl() {
    let url = redis_url();

    runtime::block_on(async {
        let rl = build_limiter_with_cache_ms(&url, 10, 100, 2f64, 500).await;
        let k = key("k");

        // With no usage, suppression factor resolves to 0.
        let sf = rl
            .redis()
            .suppressed()
            .get_suppression_factor(&k)
            .await
            .unwrap();
        assert!((sf - 0.0).abs() < 1e-12, "sf: {sf}");
    })
}

#[test]
fn get_suppression_factor_computed_uses_last_second_peak_rate_at_threshold_boundary() {
    let url = redis_url();

    runtime::block_on(async {
        // window_size=10s, hard_limit_factor=2 => window_limit=20, soft_limit=10.
        // Under the new semantics, accepted == soft with soft < hard returns sf=0 (no suppression
        // yet). We must drive accepted *past* soft (to 11) so the ramp zone is entered and
        // perceived_rate uses last-second peak instead of average.
        let cache_ms = 50_u64;
        let rl = build_limiter_with_cache_ms(&url, 10, 100, 2f64, cache_ms).await;
        let k = key("k");
        let rate_limit = RateLimit::try_from(1f64).unwrap();

        // Drive accepted to 11 in a burst < 1s. All 11 calls are pre-increment checks:
        // call 11 sees pre-increment total=10. accepted=10 == soft=10, soft(10) != hard(20) => Allowed.
        // After all 11 calls, total_count=11, total_declined=0, accepted=11.
        for _ in 0..11 {
            let _ = rl
                .redis()
                .suppressed()
                .inc(&k, &rate_limit, 1)
                .await
                .unwrap();
        }

        // Ensure any cached suppression_factor from the burst expires so we recompute.
        std::thread::sleep(Duration::from_millis(cache_ms + 25));

        let sf = rl
            .redis()
            .suppressed()
            .get_suppression_factor(&k)
            .await
            .unwrap();

        // accepted(11) > soft(10) and total(11) < hard(20): enter ramp zone.
        // rate_in_last_1s = 11, average_rate = 11/10 = 1.1.
        // perceived_rate = max(1.1, 11) = 11.
        // rate_limit = window_limit / window_size / hard_limit_factor = 20 / 10 / 2 = 1.
        // sf = 1 - (1 / 11)
        let expected = 1.0_f64 - (1.0_f64 / 11.0_f64);
        assert!(
            (sf - expected).abs() < 1e-12,
            "sf: {sf}, expected: {expected}"
        );
    });
}

#[test]
fn get_suppression_factor_evicts_out_of_window_usage_and_resets_admission() {
    let url = redis_url();

    runtime::block_on(async {
        let window_size_seconds = 1_u64;
        let cache_ms = 50_u64;
        let rl = build_limiter_with_cache_ms(&url, window_size_seconds, 1000, 2f64, cache_ms).await;

        let k = key("k");
        let rate_limit = RateLimit::try_from(1f64).unwrap();

        // window_limit = window_size * rate_limit * hard_limit_factor = 1 * 1 * 2 = 2.
        // We record 2 calls in the window, then wait for the full window to pass. If eviction
        // does not occur, the next increment would see total_count >= window_limit and go to full
        // suppression (sf=1.0). If eviction occurs, the next increment is Allowed.
        let d1 = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, 2)
            .await
            .unwrap();

        assert!(matches!(d1, RateLimitDecision::Allowed), "d1: {:?}", d1);

        std::thread::sleep(Duration::from_millis(window_size_seconds * 1000 + 50));
        std::thread::sleep(Duration::from_millis(cache_ms + 25));

        let sf = rl
            .redis()
            .suppressed()
            .get_suppression_factor(&k)
            .await
            .unwrap();
        assert!((sf - 0.0).abs() < 1e-12, "sf: {sf}");

        let d2 = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, 1)
            .await
            .unwrap();
        assert!(matches!(d2, RateLimitDecision::Allowed), "d2: {:?}", d2);
    });
}

#[test]
fn suppression_factor_gt_one_is_invalid_and_is_recomputed() {
    let url = redis_url();

    runtime::block_on(async {
        let rl = build_limiter(&url, 1, 1000, 10f64).await;
        let k = key("k");
        let rate_limit = RateLimit::try_from(5f64).unwrap();

        // This should never error; any invalid cached value handling is internal.
        let decision = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, 1)
            .await
            .unwrap();
        assert!(matches!(
            decision,
            RateLimitDecision::Allowed | RateLimitDecision::Suppressed { .. }
        ));
    });
}

#[test]
fn suppression_factor_negative_is_invalid_and_is_recomputed() {
    let url = redis_url();

    runtime::block_on(async {
        let rl = build_limiter(&url, 1, 1000, 10f64).await;
        let k = key("k");
        let rate_limit = RateLimit::try_from(5f64).unwrap();

        let decision = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, 1)
            .await
            .unwrap();
        assert!(matches!(
            decision,
            RateLimitDecision::Allowed | RateLimitDecision::Suppressed { .. }
        ));
    });
}

#[test]
fn verify_suppression_factor_calculation_spread_redis() {
    let url = redis_url();

    runtime::block_on(async {
        let rl = build_limiter(&url, 10, 100, 10f64).await;
        let k = key("k");
        let rate_limit = RateLimit::try_from(1f64).unwrap();

        // fill up in the first 3 seconds
        for _ in 0..20 {
            let _ = rl
                .redis()
                .suppressed()
                .inc(&k, &rate_limit, 1)
                .await
                .unwrap();
            runtime::async_sleep(Duration::from_millis(3000 / 20)).await;
        }

        // wait for 1.5 seconds
        runtime::async_sleep(Duration::from_millis(1200)).await;

        let expected_suppression_factor = 1f64 - (1f64 / 2f64);

        let decision = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, 1)
            .await
            .unwrap();

        eprintln!("decision: {:?}", decision);

        assert!(
            matches!(
                decision,
                RateLimitDecision::Suppressed {
                    suppression_factor,
                    ..
                } if (suppression_factor - expected_suppression_factor).abs() < 1e-12
            ),
            "decision: {:?}",
            decision
        );
    });
}

#[test]
fn verify_suppression_factor_calculation_last_second_redis() {
    let url = redis_url();

    runtime::block_on(async {
        let rl = build_limiter(&url, 10, 100, 10f64).await;
        let k = key("k");
        let rate_limit = RateLimit::try_from(1f64).unwrap();

        let _ = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, 10)
            .await
            .unwrap();
        // wait for 1s to pass
        runtime::async_sleep(Duration::from_millis(1001)).await;

        let _ = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, 20)
            .await
            .unwrap();
        // Allow time for the suppression_factor to expire
        runtime::async_sleep(Duration::from_millis(101)).await;

        let expected_suppression_factor = 1f64 - (1f64 / 20f64);

        let decision = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, 1)
            .await
            .unwrap();

        assert!(
            matches!(
                decision,
                RateLimitDecision::Suppressed {
                    suppression_factor,
                    ..
                } if (suppression_factor - expected_suppression_factor).abs() < 1e-12
            ),
            "decision: {:?}, expected sf: {expected_suppression_factor}",
            decision
        );
    });
}

#[test]
fn verify_hard_limit_rejects() {
    let url = redis_url();

    runtime::block_on(async {
        let rl = build_limiter(&url, 10, 100, 10f64).await;
        let k = key("k");
        let rate_limit = RateLimit::try_from(1f64).unwrap();

        let _ = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, 100)
            .await
            .unwrap();
        // wait for 1s to pass
        runtime::async_sleep(Duration::from_millis(1001)).await;

        let _ = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, 20)
            .await
            .unwrap();

        let decision = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, 1)
            .await
            .unwrap();

        assert!(
            matches!(
                decision,
                RateLimitDecision::Suppressed {
                    suppression_factor,
                    is_allowed: false,
                } if suppression_factor == 1.0f64
            ),
            "decision: {:?}",
            decision
        );
    });
}

#[test]
fn suppressed_is_deterministically_allowed_until_base_capacity_boundary_redis() {
    let url = redis_url();

    runtime::block_on(async {
        let window_size_seconds = 10_u64;
        let hard_limit_factor = 2f64;
        let rl = build_limiter(&url, window_size_seconds, 1000, hard_limit_factor).await;

        let k = key("k_base");
        let rate_limit = RateLimit::try_from(1f64).unwrap();

        // Base capacity = 10s * 1 req/s = 10.
        let base_capacity = window_size_seconds;

        // Pre-increment total is below base capacity.
        let d1 = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, base_capacity - 1)
            .await
            .unwrap();
        assert!(matches!(d1, RateLimitDecision::Allowed), "d1: {d1:?}");

        // Still below base capacity pre-increment, so suppression must not start.
        let d2 = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, 1)
            .await
            .unwrap();
        assert!(matches!(d2, RateLimitDecision::Allowed), "d2: {d2:?}");
    });
}

#[test]
fn suppressed_is_fully_denied_after_hard_limit_observed_redis() {
    let url = redis_url();

    runtime::block_on(async {
        let window_size_seconds = 10_u64;
        let hard_limit_factor = 2f64;

        // Use a tiny cache to deterministically observe suppression-factor recompute.
        let rl = build_limiter_with_cache_ms(&url, window_size_seconds, 1000, hard_limit_factor, 1)
            .await;

        let k = key("k_hard");
        let rate_limit = RateLimit::try_from(1f64).unwrap();

        let hard_capacity = window_size_seconds * 2;

        // Drive observed count to the hard limit in one call. This call itself is Allowed
        // because suppression_factor is computed from pre-increment state.
        let d1 = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, hard_capacity)
            .await
            .unwrap();
        assert!(matches!(d1, RateLimitDecision::Allowed), "d1: {d1:?}");

        // Ensure cached suppression factor expires before the next call.
        runtime::async_sleep(Duration::from_millis(5)).await;

        for i in 0..5u64 {
            let d = rl
                .redis()
                .suppressed()
                .inc(&k, &rate_limit, 1)
                .await
                .unwrap();

            assert!(
                matches!(
                    d,
                    RateLimitDecision::Suppressed {
                        suppression_factor,
                        is_allowed: false,
                    } if (suppression_factor - 1.0).abs() < 1e-12
                ),
                "i={i} d={d:?}"
            );
        }
    });
}

/// After the window elapses, previously committed usage must be evicted from Redis so that
/// a fresh burst is admitted at the full rate again.
///
/// This test catches the bug where `read_state` passes `window_size_ms` to the Lua script
/// instead of `window_size_seconds`. With the wrong value the eviction threshold is pushed
/// ~16 minutes into the past, old buckets are never removed, `total_count` accumulates
/// indefinitely, and suppression stays at 1.0 even after the window has expired.
#[test]
fn suppressed_redis_window_eviction_allows_fresh_burst_after_expiry() {
    let url = redis_url();

    runtime::block_on(async {
        let window_size_seconds = 1_u64;
        // hard_limit_factor=1.0 so hard_window_limit == soft_window_limit == window capacity.
        let hard_limit_factor = 1.0_f64;
        let cache_ms = 5_u64;

        let rate_limit = RateLimit::try_from(5f64).unwrap();
        // window_limit = 1s * 5 req/s * 1.0 = 5
        let window_limit = (window_size_seconds as f64 * *rate_limit * hard_limit_factor) as u64;

        let rl = build_limiter_with_cache_ms(
            &url,
            window_size_seconds,
            1_000,
            hard_limit_factor,
            cache_ms,
        )
        .await;

        let k = key("k_evict");

        // Drive observed count to the hard limit in one call.
        let d1 = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, window_limit)
            .await
            .unwrap();
        assert!(matches!(d1, RateLimitDecision::Allowed), "d1: {d1:?}");

        // Let the suppression_factor cache expire so the next read recomputes from Redis.
        runtime::async_sleep(Duration::from_millis(cache_ms + 50)).await;

        // Confirm full suppression before the window expires.
        let sf_before = rl
            .redis()
            .suppressed()
            .get_suppression_factor(&k)
            .await
            .unwrap();
        assert!(
            (sf_before - 1.0).abs() < 1e-12,
            "expected sf=1.0 before window expiry, got {sf_before}"
        );

        // Wait for the full window to expire, then let the suppression cache expire again.
        std::thread::sleep(Duration::from_millis(window_size_seconds * 1_000 + 50));
        runtime::async_sleep(Duration::from_millis(cache_ms + 50)).await;

        // After the window has expired, eviction must have cleared the old buckets.
        let sf_after = rl
            .redis()
            .suppressed()
            .get_suppression_factor(&k)
            .await
            .unwrap();
        assert!(
            (sf_after - 0.0).abs() < 1e-12,
            "expected sf=0.0 after window expiry but got {sf_after} — \
             old buckets were not evicted (window_size_ms passed instead of window_size_seconds?)"
        );

        // A new request must be admitted.
        let d2 = rl
            .redis()
            .suppressed()
            .inc(&k, &rate_limit, 1)
            .await
            .unwrap();
        assert!(
            matches!(d2, RateLimitDecision::Allowed),
            "expected Allowed after window expiry, got {d2:?}"
        );
    });
}

/// Run for three consecutive windows at well above the rate limit and assert that the total
/// admitted volume across all windows is close to `rate * num_windows`.
///
/// If window eviction is broken, `total_count` accumulates and the hard limit is hit after
/// the first window — every subsequent request is suppressed, giving total_allowed ≈
/// window_limit instead of ≈ rate * num_windows.
#[test]
fn suppressed_redis_throughput_over_multiple_windows_stays_at_rate_limit() {
    let url = redis_url();

    runtime::block_on(async {
        let window_size_seconds = 1_u64;
        let hard_limit_factor = 1.5_f64;
        let cache_ms = 5_u64;
        let num_windows = 3_u64;

        let rate_limit = RateLimit::try_from(10f64).unwrap();
        // soft_limit = 10, hard_limit = 15
        let soft_limit = (window_size_seconds as f64 * *rate_limit) as u64;
        let hard_limit = (soft_limit as f64 * hard_limit_factor) as u64;

        let rl = build_limiter_with_cache_ms(
            &url,
            window_size_seconds,
            100,
            hard_limit_factor,
            cache_ms,
        )
        .await;

        let k = key("k_multi");
        let mut total_allowed: u64 = 0;

        for _window in 0..num_windows {
            // Hammer at 10× the rate limit to ensure we hit the ceiling each window.
            let burst = soft_limit * 10;
            for _ in 0..burst {
                let d = rl
                    .redis()
                    .suppressed()
                    .inc(&k, &rate_limit, 1)
                    .await
                    .unwrap();
                match d {
                    RateLimitDecision::Allowed => total_allowed += 1,
                    RateLimitDecision::Suppressed { is_allowed, .. } => {
                        if is_allowed {
                            total_allowed += 1;
                        }
                    }
                    RateLimitDecision::Rejected { .. } => {
                        panic!("suppressed strategy must never return Rejected")
                    }
                }
            }

            // Wait for the window to expire and the suppression cache to clear.
            std::thread::sleep(Duration::from_millis(window_size_seconds * 1_000 + 50));
            runtime::async_sleep(Duration::from_millis(cache_ms + 50)).await;
        }

        // Over num_windows windows the total must be at least soft_limit * num_windows.
        // If eviction is broken, total_allowed ≈ hard_limit (15) instead of ≈ 30+.
        let expected_min = soft_limit * num_windows;
        assert!(
            total_allowed >= expected_min,
            "total_allowed={total_allowed} but expected >= {expected_min} over {num_windows} windows \
             (hard_limit={hard_limit}) — window eviction is likely broken"
        );
    });
}