//! High-volume performance test matching quic-rpc and bxp-core methodology.
//!
//! quic-rpc benchmarks: 50,000 requests, 32 parallel
//! bxp-core benchmarks: 10,000 requests, latency percentiles (p50/p90/p99)
//!
//! We run similar workloads on Knafeh to enable direct comparison.

mod common;

use std::net::SocketAddr;
use std::time::{Duration, Instant};

use tokio::sync::oneshot;

use knafeh::codec::JsonCodec;
use knafeh::rpc::middleware::MiddlewareStack;
use knafeh::rpc::router::MethodRouter;
use knafeh::transport::quic_native::{QuicClient, QuicServer};
use knafeh::transport::tls::TlsConfig;
use knafeh::{Client, ProtobufCodec, RawProtobufCodec, Server};

use common::{EchoService, TEST_CERTS};

async fn start_server(codec: impl knafeh::Codec) -> (SocketAddr, tokio::task::JoinHandle<()>) {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .codec(codec)
        .tls(TlsConfig::server(
            &TEST_CERTS.cert_path,
            &TEST_CERTS.key_path,
        ))
        .add_service(EchoService)
        .build()
        .unwrap();
    let (tx, rx) = oneshot::channel();
    let h = tokio::spawn(async move {
        server.serve_with_ready_signal(tx).await.unwrap();
    });
    (rx.await.unwrap(), h)
}

// ---------------------------------------------------------------------------
// Latency measurement
// ---------------------------------------------------------------------------

fn percentile(sorted: &[Duration], p: f64) -> Duration {
    let idx = ((sorted.len() as f64) * p / 100.0).ceil() as usize;
    sorted[idx.min(sorted.len() - 1)]
}

fn print_latency_stats(label: &str, latencies: &mut [Duration], total: Duration, n: usize) {
    latencies.sort();
    let rps = n as f64 / total.as_secs_f64();
    let p50 = percentile(latencies, 50.0);
    let p90 = percentile(latencies, 90.0);
    let p99 = percentile(latencies, 99.0);
    println!(
        "  {label:35} {n:>6} calls | {rps:>8.0} rpc/s | p50 {p50:>8.1?} | p90 {p90:>8.1?} | p99 {p99:>8.1?}",
    );
}

// ---------------------------------------------------------------------------
// Benchmark: matches quic-rpc methodology (high volume, parallel)
// ---------------------------------------------------------------------------

#[tokio::test]
#[ignore = "high-volume benchmark; run with `cargo test -- --ignored`"]
async fn test_perf_high_volume() {
    println!();
    println!("  ╔══════════════════════════════════════════════════════════════════════════════╗");
    println!("  ║  High-Volume Performance (quic-rpc / bxp-core methodology)                  ║");
    println!("  ╠══════════════════════════════════════════════════════════════════════════════╣");
    println!("  ║  quic-rpc: 50K seq, 50K×32 parallel (quinn + postcard)                     ║");
    println!("  ║  bxp-core: 10K seq (quinn + cap'n proto) → 13,249 rpc/s, p50=64µs          ║");
    println!("  ╚══════════════════════════════════════════════════════════════════════════════╝");
    println!();

    // --- RawProtobuf (minimal codec overhead, closest to raw QUIC) ---
    {
        let (addr, _h) = start_server(RawProtobufCodec::new()).await;
        let client = std::sync::Arc::new(
            Client::builder()
                .endpoint(addr.to_string())
                .codec(RawProtobufCodec::new())
                .build()
                .await
                .unwrap(),
        );
        let body = b"ping".to_vec();

        // Warmup
        client.call("echo/echo", body.clone()).await.unwrap();

        // Sequential (10,000 — matches bxp-core)
        let n = 10_000usize;
        let mut latencies = Vec::with_capacity(n);
        let total_start = Instant::now();
        for _ in 0..n {
            let t = Instant::now();
            client.call("echo/echo", body.clone()).await.unwrap();
            latencies.push(t.elapsed());
        }
        let total = total_start.elapsed();
        print_latency_stats("RawProtobuf sequential", &mut latencies, total, n);

        // Parallel 32 (matches quic-rpc pattern)
        let par = 32usize;
        let n_par = 5_000usize;
        let total_start = Instant::now();
        let mut handles = Vec::new();
        for _ in 0..n_par {
            let c = std::sync::Arc::clone(&client);
            let b = body.clone();
            handles.push(tokio::spawn(async move {
                let t = Instant::now();
                c.call("echo/echo", b).await.unwrap();
                t.elapsed()
            }));
            // Limit in-flight to `par` at a time.
            if handles.len() >= par {
                let batch: Vec<_> = handles.drain(..par).collect();
                for h in batch {
                    h.await.unwrap();
                }
            }
        }
        for h in handles {
            h.await.unwrap();
        }
        let total = total_start.elapsed();
        let rps = n_par as f64 / total.as_secs_f64();
        println!(
            "  {:35} {:>6} calls | {:>8.0} rpc/s | parallelism={}",
            "RawProtobuf parallel", n_par, rps, par
        );
    }

    println!();

    // --- Protobuf (default codec with envelope) ---
    {
        let (addr, _h) = start_server(ProtobufCodec::new()).await;
        let client = std::sync::Arc::new(
            Client::builder()
                .endpoint(addr.to_string())
                .codec(ProtobufCodec::new())
                .build()
                .await
                .unwrap(),
        );
        let body = b"ping".to_vec();

        client.call("echo/echo", body.clone()).await.unwrap();

        let n = 10_000usize;
        let mut latencies = Vec::with_capacity(n);
        let total_start = Instant::now();
        for _ in 0..n {
            let t = Instant::now();
            client.call("echo/echo", body.clone()).await.unwrap();
            latencies.push(t.elapsed());
        }
        let total = total_start.elapsed();
        print_latency_stats("Protobuf sequential", &mut latencies, total, n);
    }

    println!();

    // --- JSON (for reference) ---
    {
        let (addr, _h) = start_server(JsonCodec::new()).await;
        let client = std::sync::Arc::new(
            Client::builder()
                .endpoint(addr.to_string())
                .codec(JsonCodec::new())
                .build()
                .await
                .unwrap(),
        );
        let body = br#"{"p":"ping"}"#.to_vec();

        client.call("echo/echo", body.clone()).await.unwrap();

        let n = 10_000usize;
        let mut latencies = Vec::with_capacity(n);
        let total_start = Instant::now();
        for _ in 0..n {
            let t = Instant::now();
            client.call("echo/echo", body.clone()).await.unwrap();
            latencies.push(t.elapsed());
        }
        let total = total_start.elapsed();
        print_latency_stats("JSON sequential", &mut latencies, total, n);
    }

    println!();

    // --- Raw QUIC (quinn, no HTTP/3 — same approach as quic-rpc / bxp-core) ---
    {
        let ck = rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).unwrap();
        let cert_pem = ck.cert.pem();
        let key_pem = ck.key_pair.serialize_pem();

        let mut router = MethodRouter::new();
        router.add_service(std::sync::Arc::new(EchoService));

        let quic_server = QuicServer::bind(
            "127.0.0.1:0".parse().unwrap(),
            &cert_pem,
            &key_pem,
            router,
            std::sync::Arc::new(RawProtobufCodec::new()),
            MiddlewareStack::new(),
        )
        .unwrap();
        let quic_addr = quic_server.local_addr().unwrap();

        tokio::spawn(async move {
            quic_server.serve().await.unwrap();
        });

        // Give the server time to start accepting.
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        let quic_client = QuicClient::connect_insecure(
            quic_addr,
            std::sync::Arc::new(RawProtobufCodec::new()),
            std::sync::Arc::new(MiddlewareStack::new()),
        )
        .await
        .unwrap();

        // Warmup.
        quic_client
            .call("echo/echo", b"warmup".to_vec())
            .await
            .unwrap();

        // Sequential.
        let n = 10_000usize;
        let mut latencies = Vec::with_capacity(n);
        let total_start = Instant::now();
        for _ in 0..n {
            let t = Instant::now();
            quic_client
                .call("echo/echo", b"ping".to_vec())
                .await
                .unwrap();
            latencies.push(t.elapsed());
        }
        let total = total_start.elapsed();
        print_latency_stats("Raw QUIC (quinn) sequential", &mut latencies, total, n);

        // Parallel 32.
        let par = 32usize;
        let n_par = 5_000usize;
        let quic_client = std::sync::Arc::new(quic_client);
        let total_start = Instant::now();
        let mut handles = Vec::new();
        for _ in 0..n_par {
            let c = std::sync::Arc::clone(&quic_client);
            handles.push(tokio::spawn(async move {
                c.call("echo/echo", b"ping".to_vec()).await.unwrap();
            }));
            if handles.len() >= par {
                let batch: Vec<_> = handles.drain(..par).collect();
                for h in batch {
                    h.await.unwrap();
                }
            }
        }
        for h in handles {
            h.await.unwrap();
        }
        let total = total_start.elapsed();
        let rps = n_par as f64 / total.as_secs_f64();
        println!(
            "  {:35} {:>6} calls | {:>8.0} rpc/s | parallelism={}",
            "Raw QUIC (quinn) parallel", n_par, rps, par
        );
    }

    println!();
    println!("  ── Reference: bxp-core reports 13,249 rpc/s, p50=64.1µs (quinn + cap'n proto)");
    println!("  ── Reference: quic-rpc uses 50K requests, 32 parallel (quinn + postcard)");
}
