#![cfg(all(feature = "rest", feature = "resil"))]

use std::time::Duration;

use axum::{Router, http::StatusCode, routing::get};
use rs_zero::rest::{ApiResponse, RestConfig, RestResilienceConfig, RestServer};
use tower::ServiceExt;

#[cfg(feature = "cache-redis")]
use rs_zero::{
    cache_redis::RedisCacheConfig, resil::RedisTokenLimiterConfig, rest::RestRateLimiterConfig,
};

async fn body_text(response: axum::response::Response) -> String {
    let bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
        .await
        .expect("body");
    String::from_utf8(bytes.to_vec()).expect("utf8")
}

#[test]
fn go_zero_rest_defaults_enable_sre_breaker() {
    let config = RestConfig::go_zero_defaults("hello");

    assert!(config.middlewares.resilience.breaker_enabled);
    assert!(config.middlewares.resilience.breaker_sre_enabled);
}

#[tokio::test]
async fn concurrency_limit_rejects_excess_requests() {
    let mut config = RestConfig::default();
    config.middlewares.resilience.max_concurrency = Some(1);
    let app = RestServer::new(
        config,
        Router::new().route(
            "/slow",
            get(|| async {
                tokio::time::sleep(Duration::from_millis(50)).await;
                ApiResponse::success("ok")
            }),
        ),
    )
    .into_router();

    let first_app = app.clone();
    let first = tokio::spawn(async move {
        first_app
            .oneshot(
                axum::http::Request::builder()
                    .uri("/slow")
                    .body(axum::body::Body::empty())
                    .expect("request"),
            )
            .await
            .expect("response")
    });
    tokio::time::sleep(Duration::from_millis(5)).await;

    let second = app
        .oneshot(
            axum::http::Request::builder()
                .uri("/slow")
                .body(axum::body::Body::empty())
                .expect("request"),
        )
        .await
        .expect("response");

    assert!(body_text(second).await.contains("CONCURRENCY_LIMIT"));
    assert!(
        body_text(first.await.expect("join"))
            .await
            .contains("success")
    );
}

#[cfg(feature = "cache-redis")]
#[tokio::test]
async fn redis_token_limiter_fail_open_allows_request() {
    let mut config = RestConfig::default();
    config.middlewares.resilience.rate_limiter = invalid_token_limiter(true);
    let app = RestServer::new(
        config,
        Router::new().route("/ready", get(|| async { ApiResponse::success("ok") })),
    )
    .into_router();

    let response = app
        .oneshot(
            axum::http::Request::builder()
                .uri("/ready")
                .body(axum::body::Body::empty())
                .expect("request"),
        )
        .await
        .expect("response");

    assert!(body_text(response).await.contains("success"));
}

#[cfg(feature = "cache-redis")]
#[tokio::test]
async fn redis_token_limiter_fail_closed_rejects_request() {
    let mut config = RestConfig::default();
    config.middlewares.resilience.rate_limiter = invalid_token_limiter(false);
    let app = RestServer::new(
        config,
        Router::new().route("/ready", get(|| async { ApiResponse::success("ok") })),
    )
    .into_router();

    let response = app
        .oneshot(
            axum::http::Request::builder()
                .uri("/ready")
                .body(axum::body::Body::empty())
                .expect("request"),
        )
        .await
        .expect("response");

    assert!(body_text(response).await.contains("SERVICE_UNAVAILABLE"));
}

#[cfg(feature = "cache-redis")]
fn invalid_token_limiter(fail_open: bool) -> RestRateLimiterConfig {
    RestRateLimiterConfig::RedisToken(RedisTokenLimiterConfig {
        redis: RedisCacheConfig {
            url: "http://invalid".to_string(),
            ..RedisCacheConfig::default()
        },
        fail_open,
        ..RedisTokenLimiterConfig::default()
    })
}

#[tokio::test]
async fn breaker_rejects_after_route_failure() {
    let mut config = RestConfig::default();
    config.middlewares.resilience = RestResilienceConfig {
        breaker_enabled: true,
        breaker_failure_threshold: 1,
        breaker_reset_timeout: Duration::from_secs(60),
        max_concurrency: None,
        ..RestResilienceConfig::default()
    };
    let app = RestServer::new(
        config,
        Router::new().route("/fail", get(|| async { StatusCode::INTERNAL_SERVER_ERROR })),
    )
    .into_router();

    let first = app
        .clone()
        .oneshot(
            axum::http::Request::builder()
                .uri("/fail")
                .body(axum::body::Body::empty())
                .expect("request"),
        )
        .await
        .expect("response");
    assert!(body_text(first).await.contains("500"));

    let second = app
        .oneshot(
            axum::http::Request::builder()
                .uri("/fail")
                .body(axum::body::Body::empty())
                .expect("request"),
        )
        .await
        .expect("response");
    assert!(body_text(second).await.contains("SERVICE_UNAVAILABLE"));
}

#[tokio::test]
async fn adaptive_shedder_rejects_excess_requests() {
    let mut config = RestConfig::default();
    config.middlewares.resilience = RestResilienceConfig {
        shedding_enabled: true,
        shedding_max_in_flight: Some(1),
        shedding_min_request_count: 1,
        ..RestResilienceConfig::default()
    };
    let app = RestServer::new(
        config,
        Router::new().route(
            "/slow",
            get(|| async {
                tokio::time::sleep(Duration::from_millis(50)).await;
                ApiResponse::success("ok")
            }),
        ),
    )
    .into_router();

    let first_app = app.clone();
    let first = tokio::spawn(async move {
        first_app
            .oneshot(
                axum::http::Request::builder()
                    .uri("/slow")
                    .body(axum::body::Body::empty())
                    .expect("request"),
            )
            .await
            .expect("response")
    });
    tokio::time::sleep(Duration::from_millis(5)).await;

    let second = app
        .oneshot(
            axum::http::Request::builder()
                .uri("/slow")
                .body(axum::body::Body::empty())
                .expect("request"),
        )
        .await
        .expect("response");

    assert!(body_text(second).await.contains("OVERLOADED"));
    assert!(
        body_text(first.await.expect("join"))
            .await
            .contains("success")
    );
}

#[cfg(feature = "observability")]
fn rest_pressure_config(registry: rs_zero::observability::MetricsRegistry) -> RestConfig {
    let mut middlewares = rs_zero::rest::RestMiddlewareConfig::default();
    middlewares.metrics.enabled = true;
    middlewares.resilience = RestResilienceConfig {
        breaker_enabled: true,
        breaker_failure_threshold: 2,
        breaker_reset_timeout: Duration::from_secs(60),
        max_concurrency: Some(1),
        ..RestResilienceConfig::default()
    };
    RestConfig {
        metrics_registry: Some(registry),
        middlewares,
        ..RestConfig::default()
    }
}

#[cfg(feature = "observability")]
async fn rest_text(app: Router, uri: &str) -> String {
    body_text(
        app.oneshot(
            axum::http::Request::builder()
                .uri(uri)
                .body(axum::body::Body::empty())
                .expect("request"),
        )
        .await
        .expect("response"),
    )
    .await
}

#[cfg(feature = "observability")]
#[tokio::test]
async fn rest_resilience_pressure_records_low_cardinality_metrics() {
    let registry = rs_zero::observability::MetricsRegistry::new();
    let config = rest_pressure_config(registry.clone());
    let app = RestServer::new(
        config,
        Router::new()
            .route(
                "/slow",
                get(|| async {
                    tokio::time::sleep(Duration::from_millis(40)).await;
                    ApiResponse::success("ok")
                }),
            )
            .route("/fail", get(|| async { StatusCode::INTERNAL_SERVER_ERROR })),
    )
    .into_router();

    let first_app = app.clone();
    let first = tokio::spawn(async move { rest_text(first_app, "/slow").await });
    tokio::time::sleep(Duration::from_millis(5)).await;
    let rejected_body = rest_text(app.clone(), "/slow").await;
    let first_body = first.await.expect("join");
    assert!(first_body.contains("success"));
    assert!(rejected_body.contains("CONCURRENCY_LIMIT"));

    for _ in 0..3 {
        let _ = rest_text(app.clone(), "/fail").await;
    }

    let text = registry.render_prometheus();
    assert!(text.contains("component=\"concurrency\",outcome=\"allowed\",transport=\"http\""));
    assert!(text.contains("component=\"concurrency\",outcome=\"rejected\",transport=\"http\""));
    assert!(text.contains("component=\"breaker\",outcome=\"failure\",transport=\"http\""));
}
