//! E2E load simulation tests: throughput, concurrency, mixed workloads.
//!
//! SOP Category G (Load Simulation).

mod common;

use std::net::SocketAddr;

use async_trait::async_trait;

use knafeh::codec::JsonCodec;
use knafeh::error::{KnafehError, RpcStatusCode};
use knafeh::rpc::message::{Metadata, RpcRequest, RpcResponse};
use knafeh::rpc::service::{MethodDescriptor, MethodKind, Service};
use knafeh::rpc::stream::{rpc_stream_channel, RpcStreamRequest, RpcStreamResponse};
use knafeh::{Client, Server};

// ---------------------------------------------------------------------------
// TLS certs
// ---------------------------------------------------------------------------

use common::{start_test_server, tls_server, EchoService};

// ---------------------------------------------------------------------------
// Test services
// ---------------------------------------------------------------------------

struct CountService;

#[async_trait]
impl Service for CountService {
    fn name(&self) -> &str {
        "count"
    }
    fn methods(&self) -> Vec<MethodDescriptor> {
        vec![MethodDescriptor {
            name: "count".to_string(),
            kind: MethodKind::ServerStreaming,
        }]
    }
    async fn call_unary(&self, _: &str, _: RpcRequest) -> Result<RpcResponse, KnafehError> {
        unimplemented!()
    }
    async fn call_server_stream(
        &self,
        _: &str,
        req: RpcRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        let body: serde_json::Value =
            serde_json::from_slice(&req.body).map_err(|e| KnafehError::Codec(e.to_string()))?;
        let count = body["count"].as_u64().unwrap_or(0);
        let (sender, receiver) = rpc_stream_channel(32);
        tokio::spawn(async move {
            for i in 0..count {
                let chunk = serde_json::json!({"i": i}).to_string().into_bytes();
                if sender.send(chunk).await.is_err() {
                    break;
                }
            }
        });
        Ok(RpcStreamResponse {
            metadata: Metadata::new(),
            receiver,
        })
    }
    async fn call_client_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcResponse, KnafehError> {
        unimplemented!()
    }
    async fn call_bidi_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }
}

/// Sleeps for a configurable duration then echoes.
struct SlowService;

#[async_trait]
impl Service for SlowService {
    fn name(&self) -> &str {
        "slow"
    }
    fn methods(&self) -> Vec<MethodDescriptor> {
        vec![MethodDescriptor {
            name: "call".to_string(),
            kind: MethodKind::Unary,
        }]
    }
    async fn call_unary(&self, _: &str, req: RpcRequest) -> Result<RpcResponse, KnafehError> {
        let body: serde_json::Value =
            serde_json::from_slice(&req.body).map_err(|e| KnafehError::Codec(e.to_string()))?;
        let delay_ms = body["delay_ms"].as_u64().unwrap_or(100);
        tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
        Ok(RpcResponse::ok(req.body))
    }
    async fn call_server_stream(
        &self,
        _: &str,
        _: RpcRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }
    async fn call_client_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcResponse, KnafehError> {
        unimplemented!()
    }
    async fn call_bidi_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }
}

// ===========================================================================
// Category G: Load Simulation
// ===========================================================================

/// G1: Sustained throughput — 100 sequential requests.
#[tokio::test]
async fn test_e2e_sustained_throughput() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(EchoService)
        .build()
        .unwrap();
    let (addr, _handle) = start_test_server(server).await;
    let client = Client::builder()
        .endpoint(addr.to_string())
        .codec(JsonCodec::new())
        .build()
        .await
        .unwrap();

    for i in 0..100u64 {
        let body = format!(r#"{{"seq":{i}}}"#);
        let resp = client
            .call("echo/echo", body.as_bytes().to_vec())
            .await
            .unwrap();
        assert_eq!(resp.status.code, RpcStatusCode::Ok);
        assert_eq!(resp.body, body.as_bytes());
    }

    // Should reuse a single connection for all 100 calls.
    assert_eq!(client.pool().size(), 1);
}

/// G2: Burst concurrency — 50 concurrent requests.
#[tokio::test]
async fn test_e2e_burst_concurrency() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(EchoService)
        .build()
        .unwrap();
    let (addr, _handle) = start_test_server(server).await;
    let client = std::sync::Arc::new(
        Client::builder()
            .endpoint(addr.to_string())
            .codec(JsonCodec::new())
            .build()
            .await
            .unwrap(),
    );

    let mut handles = Vec::new();
    for i in 0..50u64 {
        let client = std::sync::Arc::clone(&client);
        handles.push(tokio::spawn(async move {
            let body = format!(r#"{{"i":{i}}}"#);
            let resp = client
                .call("echo/echo", body.as_bytes().to_vec())
                .await
                .unwrap();
            assert_eq!(resp.status.code, RpcStatusCode::Ok);
            assert_eq!(resp.body, body.as_bytes());
        }));
    }

    for h in handles {
        h.await.unwrap();
    }

    // Pool should stay within configured limits (default 4).
    assert!(client.pool().size() <= 4);
}

/// G3: Mixed workload — 70% unary + 20% streaming + 10% error calls.
#[tokio::test]
async fn test_e2e_mixed_workload() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(EchoService)
        .add_service(CountService)
        .build()
        .unwrap();
    let (addr, _handle) = start_test_server(server).await;
    let client = std::sync::Arc::new(
        Client::builder()
            .endpoint(addr.to_string())
            .codec(JsonCodec::new())
            .build()
            .await
            .unwrap(),
    );

    let mut handles = Vec::new();

    // 70% unary
    for i in 0..70u64 {
        let client = std::sync::Arc::clone(&client);
        handles.push(tokio::spawn(async move {
            let body = format!(r#"{{"i":{i}}}"#);
            let resp = client
                .call("echo/echo", body.as_bytes().to_vec())
                .await
                .unwrap();
            assert_eq!(resp.status.code, RpcStatusCode::Ok);
        }));
    }

    // 20% streaming
    for _ in 0..20u64 {
        let client = std::sync::Arc::clone(&client);
        handles.push(tokio::spawn(async move {
            let mut stream = client
                .server_stream("count/count", br#"{"count":5}"#.to_vec())
                .await
                .unwrap();
            let mut n = 0;
            while let Some(Ok(_)) = stream.next().await {
                n += 1;
            }
            assert_eq!(n, 5);
        }));
    }

    // 10% error (method not found)
    for _ in 0..10u64 {
        let client = std::sync::Arc::clone(&client);
        handles.push(tokio::spawn(async move {
            let resp = client
                .call("nonexistent/method", br#"{}"#.to_vec())
                .await
                .unwrap();
            assert_eq!(resp.status.code, RpcStatusCode::NotFound);
        }));
    }

    for h in handles {
        h.await.unwrap();
    }
}

/// G4: Slow service under load — 20 concurrent calls to a 50ms-latency service.
#[tokio::test]
async fn test_e2e_slow_service_under_load() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(SlowService)
        .build()
        .unwrap();
    let (addr, _handle) = start_test_server(server).await;
    let client = std::sync::Arc::new(
        Client::builder()
            .endpoint(addr.to_string())
            .codec(JsonCodec::new())
            .build()
            .await
            .unwrap(),
    );

    let start = std::time::Instant::now();
    let mut handles = Vec::new();
    for i in 0..20u64 {
        let client = std::sync::Arc::clone(&client);
        handles.push(tokio::spawn(async move {
            let body = format!(r#"{{"delay_ms":50,"i":{i}}}"#);
            let resp = client
                .call("slow/call", body.as_bytes().to_vec())
                .await
                .unwrap();
            assert_eq!(resp.status.code, RpcStatusCode::Ok);
        }));
    }

    for h in handles {
        h.await.unwrap();
    }

    // 20 concurrent 50ms calls should complete much faster than 20*50ms=1s
    // thanks to QUIC multiplexing. Allow generous 5s ceiling.
    let elapsed = start.elapsed();
    assert!(
        elapsed < std::time::Duration::from_secs(5),
        "expected < 5s, got {:?}",
        elapsed
    );

    // Pool should not be exhausted.
    assert!(client.pool().size() <= 4);
}
