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

use std::time::Duration;

use rs_zero::rpc::{RpcResilienceConfig, RpcResilienceLayer};

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

#[test]
fn go_zero_rpc_defaults_enable_sre_breaker() {
    let config = RpcResilienceConfig::go_zero_defaults();

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

#[tokio::test]
async fn rpc_resilience_maps_timeout() {
    let layer = RpcResilienceLayer::new(
        "hello",
        RpcResilienceConfig {
            request_timeout: Duration::from_millis(5),
            ..RpcResilienceConfig::default()
        },
    );

    let result = layer
        .run_unary("Say", || async {
            tokio::time::sleep(Duration::from_millis(20)).await;
            Ok::<_, tonic::Status>("ok")
        })
        .await;

    assert_eq!(
        result.expect_err("timeout").code(),
        tonic::Code::DeadlineExceeded
    );
}

#[cfg(feature = "observability")]
#[tokio::test]
async fn rpc_resilience_records_rpc_and_decision_metrics() {
    let registry = rs_zero::observability::MetricsRegistry::new();
    let layer = RpcResilienceLayer::new(
        "hello",
        RpcResilienceConfig {
            max_concurrency: Some(1),
            ..RpcResilienceConfig::default()
        },
    )
    .with_metrics(registry.clone());

    let result = layer
        .run_unary("Say", || async { Ok::<_, tonic::Status>("ok") })
        .await;

    assert_eq!(result.expect("rpc"), "ok");
    let text = registry.render_prometheus();
    assert!(text.contains("rs_zero_rpc_requests_total"));
    assert!(text.contains("service=\"hello\",method=\"Say\",code=\"Ok\""));
    assert!(text.contains("component=\"concurrency\",outcome=\"allowed\",transport=\"grpc\""));
}

#[tokio::test]
async fn rpc_resilience_maps_shedder_rejection() {
    let layer = RpcResilienceLayer::new(
        "hello",
        RpcResilienceConfig {
            shedding_enabled: true,
            shedding_max_in_flight: Some(1),
            ..RpcResilienceConfig::default()
        },
    );
    let held = layer.clone();
    let first = tokio::spawn(async move {
        held.run_unary("Say", || async {
            tokio::time::sleep(Duration::from_millis(30)).await;
            Ok::<_, tonic::Status>("ok")
        })
        .await
    });
    tokio::time::sleep(Duration::from_millis(5)).await;

    let rejected = layer.run_unary("Say", || async { Ok("ok") }).await;

    assert_eq!(
        rejected.expect_err("rejected").code(),
        tonic::Code::ResourceExhausted
    );
    assert_eq!(first.await.expect("join").expect("first"), "ok");
}

#[tokio::test]
async fn rpc_resilience_maps_breaker_rejection() {
    let layer = RpcResilienceLayer::new(
        "hello",
        RpcResilienceConfig {
            breaker_enabled: true,
            breaker_failure_threshold: 1,
            breaker_reset_timeout: Duration::from_secs(60),
            ..RpcResilienceConfig::default()
        },
    );

    let _ = layer
        .run_unary("Say", || async {
            Err::<(), _>(tonic::Status::internal("down"))
        })
        .await;
    let rejected = layer.run_unary("Say", || async { Ok(()) }).await;

    assert_eq!(rejected.expect_err("open").code(), tonic::Code::Unavailable);
}

#[cfg(feature = "cache-redis")]
#[tokio::test]
async fn rpc_redis_token_limiter_fail_open_allows_call() {
    let layer = RpcResilienceLayer::new(
        "hello",
        RpcResilienceConfig {
            rate_limiter: invalid_token_limiter(true),
            ..RpcResilienceConfig::default()
        },
    );

    let result = layer.run_unary("Say", || async { Ok("ok") }).await;

    assert_eq!(result.expect("allowed"), "ok");
}

#[cfg(feature = "cache-redis")]
#[tokio::test]
async fn rpc_redis_token_limiter_fail_closed_rejects_call() {
    let layer = RpcResilienceLayer::new(
        "hello",
        RpcResilienceConfig {
            rate_limiter: invalid_token_limiter(false),
            ..RpcResilienceConfig::default()
        },
    );

    let result = layer.run_unary("Say", || async { Ok("ok") }).await;

    assert_eq!(
        result.expect_err("rejected").code(),
        tonic::Code::Unavailable
    );
}

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

#[cfg(feature = "observability")]
fn rpc_pressure_layer(registry: rs_zero::observability::MetricsRegistry) -> RpcResilienceLayer {
    RpcResilienceLayer::new(
        "hello",
        RpcResilienceConfig {
            max_concurrency: Some(2),
            shedding_enabled: true,
            shedding_max_in_flight: Some(2),
            shedding_min_request_count: 1,
            breaker_enabled: true,
            breaker_failure_threshold: 2,
            breaker_reset_timeout: Duration::from_secs(60),
            ..RpcResilienceConfig::default()
        },
    )
    .with_metrics(registry)
}

#[cfg(feature = "observability")]
async fn rpc_pressure_call(layer: RpcResilienceLayer, index: usize) -> Result<(), tonic::Status> {
    layer
        .run_unary("Say", || async move {
            tokio::time::sleep(Duration::from_millis(25)).await;
            if index.is_multiple_of(5) {
                Err::<(), _>(tonic::Status::internal("planned"))
            } else {
                Ok(())
            }
        })
        .await
}

#[cfg(feature = "observability")]
#[tokio::test]
async fn rpc_resilience_pressure_records_low_cardinality_metrics() {
    let registry = rs_zero::observability::MetricsRegistry::new();
    let layer = rpc_pressure_layer(registry.clone());

    let mut handles = Vec::new();
    for index in 0..8 {
        let layer = layer.clone();
        handles.push(tokio::spawn(async move {
            rpc_pressure_call(layer, index).await
        }));
    }

    let mut ok = 0usize;
    let mut resource_exhausted = 0usize;
    let mut internal = 0usize;
    for handle in handles {
        match handle.await.expect("join") {
            Ok(()) => ok += 1,
            Err(status) if status.code() == tonic::Code::ResourceExhausted => {
                resource_exhausted += 1;
            }
            Err(status) if status.code() == tonic::Code::Internal => internal += 1,
            Err(status) => panic!("unexpected status: {status:?}"),
        }
    }

    let text = registry.render_prometheus();
    assert!(
        ok > 0,
        "ok={ok}, exhausted={resource_exhausted}, internal={internal}"
    );
    assert!(
        resource_exhausted > 0,
        "ok={ok}, exhausted={resource_exhausted}"
    );
    assert!(text.contains("component=\"concurrency\",outcome=\"allowed\",transport=\"grpc\""));
    assert!(text.contains("component=\"concurrency\",outcome=\"rejected\",transport=\"grpc\""));
    assert!(text.contains("service=\"hello\",method=\"Say\""));
    assert!(!text.contains("planned"));
}

#[tokio::test]
async fn rpc_server_layer_stack_maps_timeout_without_manual_helper() {
    use http::{Request, Response};
    use rs_zero::rpc::{RpcServerConfig, RpcServerLayerStack};
    use std::{convert::Infallible, task::Poll};
    use tower::{Layer, Service};

    #[derive(Clone)]
    struct SlowService;

    impl Service<Request<tonic::body::Body>> for SlowService {
        type Response = Response<tonic::body::Body>;
        type Error = Infallible;
        type Future =
            std::pin::Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

        fn poll_ready(
            &mut self,
            _cx: &mut std::task::Context<'_>,
        ) -> Poll<Result<(), Self::Error>> {
            Poll::Ready(Ok(()))
        }

        fn call(&mut self, _request: Request<tonic::body::Body>) -> Self::Future {
            Box::pin(async move {
                tokio::time::sleep(Duration::from_millis(20)).await;
                Ok(Response::new(tonic::body::Body::empty()))
            })
        }
    }

    let config = RpcServerConfig {
        resilience: RpcResilienceConfig {
            request_timeout: Duration::from_millis(1),
            ..RpcResilienceConfig::default()
        },
        ..RpcServerConfig::new("hello", "127.0.0.1:0".parse().expect("addr"))
    };
    let mut service = RpcServerLayerStack::new(config)
        .into_layer()
        .layer(SlowService);

    let response = service
        .call(
            Request::builder()
                .uri("/hello.Hello/SayHello")
                .body(tonic::body::Body::empty())
                .expect("request"),
        )
        .await
        .expect("infallible");

    assert_eq!(response.status(), http::StatusCode::OK);
    assert_eq!(
        response
            .headers()
            .get("grpc-status")
            .and_then(|value| value.to_str().ok()),
        Some("4")
    );
}

#[cfg(feature = "observability")]
#[tokio::test]
async fn rpc_server_layer_stack_reads_request_id_from_metadata() {
    use http::{Request, Response};
    use rs_zero::{
        observability::MetricsRegistry,
        rpc::{RpcServerConfig, RpcServerLayerStack},
    };
    use std::{convert::Infallible, task::Poll};
    use tower::{Layer, Service};

    #[derive(Clone)]
    struct OkService;

    impl Service<Request<tonic::body::Body>> for OkService {
        type Response = Response<tonic::body::Body>;
        type Error = Infallible;
        type Future =
            std::pin::Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

        fn poll_ready(
            &mut self,
            _cx: &mut std::task::Context<'_>,
        ) -> Poll<Result<(), Self::Error>> {
            Poll::Ready(Ok(()))
        }

        fn call(&mut self, _request: Request<tonic::body::Body>) -> Self::Future {
            Box::pin(async move { Ok(Response::new(tonic::body::Body::empty())) })
        }
    }

    let registry = MetricsRegistry::new();
    let config = RpcServerConfig::new("hello", "127.0.0.1:0".parse().expect("addr"));
    let mut service = RpcServerLayerStack::new(config)
        .with_metrics(registry.clone())
        .into_layer()
        .layer(OkService);

    let response = service
        .call(
            Request::builder()
                .uri("/hello.Hello/SayHello")
                .header("x-request-id", "rpc-layer-req-1")
                .body(tonic::body::Body::empty())
                .expect("request"),
        )
        .await
        .expect("infallible");

    assert_eq!(response.status(), http::StatusCode::OK);
    assert!(
        registry
            .render_prometheus()
            .contains("service=\"hello\",method=\"/hello.Hello/SayHello\",code=\"Ok\"")
    );
}
