//! E2E performance benchmarks: Protobuf vs JSON codecs,
//! Rust server vs Python server — all using Rust clients.
//!
//! Follows SOP categories G (Load Simulation) and workload patterns.

mod common;

use std::io::{BufRead, BufReader};
use std::net::SocketAddr;
use std::process::{Child, Command, Stdio};
use std::time::Instant;

use tokio::sync::oneshot;
use tokio::task::JoinHandle;

use knafeh::codec::JsonCodec;
use knafeh::error::RpcStatusCode;
use knafeh::transport::tls::TlsConfig;
use knafeh::{Client, Server};

use common::{EchoService, TEST_CERTS};

// ---------------------------------------------------------------------------
// Server helpers
// ---------------------------------------------------------------------------

async fn start_rust_server_with_codec(codec: impl knafeh::Codec) -> (SocketAddr, 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 (ready_tx, ready_rx) = oneshot::channel();
    let handle = tokio::spawn(async move {
        server.serve_with_ready_signal(ready_tx).await.unwrap();
    });
    let addr = ready_rx.await.unwrap();
    (addr, handle)
}

fn start_python_server() -> Option<(String, Child)> {
    let python = [
        "/home/ubuntu/Workspace/aesthetic-score/.venv/bin/python",
        "python3",
        "python",
    ]
    .iter()
    .find(|p| Command::new(p).arg("--version").output().is_ok())
    .map(|s| s.to_string());

    let python = python?;

    let port = 24433u16;
    let mut child = Command::new(&python)
        .arg("tests/bench_python_server.py")
        .arg(&TEST_CERTS.cert_path)
        .arg(&TEST_CERTS.key_path)
        .arg(port.to_string())
        .stdout(Stdio::piped())
        .stderr(Stdio::null())
        .spawn()
        .ok()?;

    let stdout = child.stdout.take().unwrap();
    let mut reader = BufReader::new(stdout);
    let mut line = String::new();
    reader.read_line(&mut line).ok()?;
    if !line.trim().starts_with("READY") {
        child.kill().ok();
        return None;
    }
    std::thread::sleep(std::time::Duration::from_millis(500));
    Some((format!("127.0.0.1:{port}"), child))
}

// ---------------------------------------------------------------------------
// Benchmark runner
// ---------------------------------------------------------------------------

struct BenchResult {
    label: String,
    seq_time: f64,
    seq_n: u64,
    conc_time: f64,
    conc_n: u64,
}

impl BenchResult {
    fn seq_rps(&self) -> f64 {
        self.seq_n as f64 / self.seq_time
    }
    fn conc_rps(&self) -> f64 {
        self.conc_n as f64 / self.conc_time
    }
}

async fn run_benchmark(
    label: &str,
    addr: &str,
    codec: impl knafeh::Codec,
    body: Vec<u8>,
    seq_n: u64,
    conc_n: u64,
) -> BenchResult {
    let client = std::sync::Arc::new(
        Client::builder()
            .endpoint(addr)
            .codec(codec)
            .build()
            .await
            .unwrap(),
    );

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

    // Sequential.
    let start = Instant::now();
    for _ in 0..seq_n {
        let resp = client.call("echo/echo", body.clone()).await.unwrap();
        assert_eq!(resp.status.code, RpcStatusCode::Ok);
    }
    let seq_time = start.elapsed().as_secs_f64();

    // Concurrent.
    let start = Instant::now();
    let mut handles = Vec::new();
    for _ in 0..conc_n {
        let c = std::sync::Arc::clone(&client);
        let b = body.clone();
        handles.push(tokio::spawn(async move {
            c.call("echo/echo", b).await.unwrap();
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    let conc_time = start.elapsed().as_secs_f64();

    BenchResult {
        label: label.to_string(),
        seq_time,
        seq_n,
        conc_time,
        conc_n,
    }
}

fn print_result(r: &BenchResult) {
    println!(
        "  {:30} seq {:>4} calls: {:.3}s {:>6.0} rpc/s | conc {:>3} calls: {:.3}s {:>6.0} rpc/s",
        r.label,
        r.seq_n,
        r.seq_time,
        r.seq_rps(),
        r.conc_n,
        r.conc_time,
        r.conc_rps(),
    );
}

// ---------------------------------------------------------------------------
// Main benchmark test
// ---------------------------------------------------------------------------

#[tokio::test]
async fn test_bench_protobuf_vs_json() {
    let seq_n = 100u64;
    let conc_n = 50u64;

    let json_body = br#"{"bench":true}"#.to_vec();
    let proto_body = b"bench payload bytes".to_vec();

    println!();
    println!("  ╔══════════════════════════════════════════════════════════════════════╗");
    println!("  ║  E2E Benchmark: Protobuf vs JSON, Rust vs Python servers            ║");
    println!(
        "  ║  Rust client for all. {seq_n} seq + {conc_n} conc calls each.                    ║"
    );
    println!("  ╚══════════════════════════════════════════════════════════════════════╝");
    println!();

    // --- Rust server + JSON codec ---
    let (json_addr, _h1) = start_rust_server_with_codec(JsonCodec::new()).await;
    let r_json = run_benchmark(
        "Rust server + JSON",
        &json_addr.to_string(),
        JsonCodec::new(),
        json_body.clone(),
        seq_n,
        conc_n,
    )
    .await;
    print_result(&r_json);

    // --- Rust server + Protobuf codec (default) ---
    let (proto_addr, _h2) = start_rust_server_with_codec(knafeh::ProtobufCodec::new()).await;
    let r_proto = run_benchmark(
        "Rust server + Protobuf",
        &proto_addr.to_string(),
        knafeh::ProtobufCodec::new(),
        proto_body.clone(),
        seq_n,
        conc_n,
    )
    .await;
    print_result(&r_proto);

    // --- Rust server + RawProtobuf codec ---
    let (raw_addr, _h3) = start_rust_server_with_codec(knafeh::RawProtobufCodec::new()).await;
    let r_raw = run_benchmark(
        "Rust server + RawProtobuf",
        &raw_addr.to_string(),
        knafeh::RawProtobufCodec::new(),
        proto_body.clone(),
        seq_n,
        conc_n,
    )
    .await;
    print_result(&r_raw);

    // --- Python server + Protobuf codec (default) ---
    if let Some((py_addr, mut child)) = start_python_server() {
        // Python server uses default protobuf codec.
        let r_py = run_benchmark(
            "Python server + Protobuf",
            &py_addr,
            knafeh::ProtobufCodec::new(),
            proto_body.clone(),
            seq_n,
            conc_n,
        )
        .await;
        print_result(&r_py);
        child.kill().ok();

        // --- Summary table ---
        println!();
        println!("  ── Summary ──────────────────────────────────────────────────────────");
        println!("  {:30} {:>10} {:>10}", "", "seq rpc/s", "conc rpc/s");
        println!(
            "  {:30} {:>10.0} {:>10.0}",
            "Rust + JSON",
            r_json.seq_rps(),
            r_json.conc_rps()
        );
        println!(
            "  {:30} {:>10.0} {:>10.0}",
            "Rust + Protobuf",
            r_proto.seq_rps(),
            r_proto.conc_rps()
        );
        println!(
            "  {:30} {:>10.0} {:>10.0}",
            "Rust + RawProtobuf",
            r_raw.seq_rps(),
            r_raw.conc_rps()
        );
        println!(
            "  {:30} {:>10.0} {:>10.0}",
            "Python + Protobuf",
            r_py.seq_rps(),
            r_py.conc_rps()
        );

        let proto_vs_json_seq = r_proto.seq_rps() / r_json.seq_rps();
        let raw_vs_json_seq = r_raw.seq_rps() / r_json.seq_rps();
        println!();
        println!(
            "  Protobuf/JSON:     seq {proto_vs_json_seq:.2}x | conc {:.2}x",
            r_proto.conc_rps() / r_json.conc_rps()
        );
        println!(
            "  RawProtobuf/JSON:  seq {raw_vs_json_seq:.2}x | conc {:.2}x",
            r_raw.conc_rps() / r_json.conc_rps()
        );
    } else {
        println!();
        println!("  Python server: SKIPPED");
        println!();
        println!("  ── Summary (Rust only) ─────────────────────────────────────────────");
        println!("  {:30} {:>10} {:>10}", "", "seq rpc/s", "conc rpc/s");
        println!(
            "  {:30} {:>10.0} {:>10.0}",
            "JSON",
            r_json.seq_rps(),
            r_json.conc_rps()
        );
        println!(
            "  {:30} {:>10.0} {:>10.0}",
            "Protobuf",
            r_proto.seq_rps(),
            r_proto.conc_rps()
        );
        println!(
            "  {:30} {:>10.0} {:>10.0}",
            "RawProtobuf",
            r_raw.seq_rps(),
            r_raw.conc_rps()
        );
    }
}
