use std::pin::Pin;
use std::sync::{Arc, OnceLock};
use std::task::{Context, Poll};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use anyhow::Context as _;
use serde::Serialize;
use tokio::io::{self, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
use tokio::net::{TcpListener, TcpStream};
use tokio_rustls::TlsConnector;

use crate::relay;
use relay::tune_socket;

const ECHO_ADDR: &str = "127.0.0.1:19876";
const RELAY_ADDR: &str = "127.0.0.1:19877";
const RELAY_TLS_ADDR: &str = "127.0.0.1:19878";

// ─── Stream abstraction ───

/// Wraps either a plain TCP stream or a TLS stream so workload functions
/// can be generic without trait objects.
enum BenchStream {
    Tcp(TcpStream),
    Tls(tokio_rustls::client::TlsStream<TcpStream>),
}

impl AsyncRead for BenchStream {
    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
        match self.get_mut() {
            BenchStream::Tcp(s) => Pin::new(s).poll_read(cx, buf),
            BenchStream::Tls(s) => Pin::new(s).poll_read(cx, buf),
        }
    }
}

impl AsyncWrite for BenchStream {
    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
        match self.get_mut() {
            BenchStream::Tcp(s) => Pin::new(s).poll_write(cx, buf),
            BenchStream::Tls(s) => Pin::new(s).poll_write(cx, buf),
        }
    }
    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        match self.get_mut() {
            BenchStream::Tcp(s) => Pin::new(s).poll_flush(cx),
            BenchStream::Tls(s) => Pin::new(s).poll_flush(cx),
        }
    }
    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
        match self.get_mut() {
            BenchStream::Tcp(s) => Pin::new(s).poll_shutdown(cx),
            BenchStream::Tls(s) => Pin::new(s).poll_shutdown(cx),
        }
    }
}

// ─── Results ───

#[derive(Serialize)]
struct BenchResults {
    timestamp: String,
    git_rev: String,
    iterations: u32,
    workloads: WorkloadResults,
    concurrency_scaling: Vec<ConcurrencyPoint>,
}

#[derive(Serialize)]
struct WorkloadResults {
    interactive: WorkloadComparison,
    terminal_ui: WorkloadComparison,
    stream_up: WorkloadComparison,
    stream_down: WorkloadComparison,
    mixed: WorkloadComparison,
    connection_setup: WorkloadComparison,
}

#[derive(Serialize, Clone)]
struct WorkloadComparison {
    direct: WorkloadMetrics,
    relay: WorkloadMetrics,
    #[serde(skip_serializing_if = "Option::is_none")]
    relay_tls: Option<WorkloadMetrics>,
}

#[derive(Serialize, Clone)]
struct WorkloadMetrics {
    throughput_mbps: f64,
    latency: LatencyPoint,
}

#[derive(Serialize, Clone)]
struct LatencyPoint {
    min_us: u64,
    p50_us: u64,
    p95_us: u64,
    p99_us: u64,
    max_us: u64,
}

#[derive(Serialize)]
struct ConcurrencyPoint {
    connections: usize,
    direct_mbps: f64,
    relay_mbps: f64,
    #[serde(skip_serializing_if = "Option::is_none")]
    relay_tls_mbps: Option<f64>,
}

// ─── Echo server ───

/// Run a standalone echo server (called via `layover bench-echo --listen ADDR`).
pub async fn run_echo(listen: &str) -> anyhow::Result<()> {
    let listener = TcpListener::bind(listen).await?;
    loop {
        let (mut stream, _) = listener.accept().await?;
        stream.set_nodelay(true).ok();
        tune_socket(&stream);
        tokio::spawn(async move {
            let mut buf = [0u8; 65536];
            loop {
                match stream.read(&mut buf).await {
                    Ok(0) | Err(_) => break,
                    Ok(n) => {
                        if stream.write_all(&buf[..n]).await.is_err() {
                            break;
                        }
                    }
                }
            }
        });
    }
}

// ─── Stream creation ───

#[derive(Clone, Copy)]
enum Mode { Direct, Relay, RelayTls }

/// TLS connector initialized once during bench setup, used by all TLS connections.
static TLS_CONNECTOR: OnceLock<TlsConnector> = OnceLock::new();

async fn get_stream(mode: Mode) -> anyhow::Result<BenchStream> {
    match mode {
        Mode::Direct => {
            let stream = TcpStream::connect(ECHO_ADDR).await?;
            stream.set_nodelay(true)?;
            tune_socket(&stream);
            Ok(BenchStream::Tcp(stream))
        }
        Mode::Relay => {
            let mut stream = TcpStream::connect(RELAY_ADDR).await.context("relay connect")?;
            stream.set_nodelay(true)?;
            tune_socket(&stream);
            relay::write_line(&mut stream, &format!("V2 {ECHO_ADDR}")).await?;
            let response = relay::read_line(&mut stream).await?;
            if response != "OK" { anyhow::bail!("relay error: {response}"); }
            Ok(BenchStream::Tcp(stream))
        }
        Mode::RelayTls => {
            let connector = TLS_CONNECTOR.get().context("TLS not initialized")?;
            let tcp = TcpStream::connect(RELAY_TLS_ADDR).await.context("tls relay connect")?;
            tcp.set_nodelay(true)?;
            tune_socket(&tcp);
            let server_name = tokio_rustls::rustls::pki_types::ServerName::try_from("localhost".to_string())
                .map_err(|e| anyhow::anyhow!("bad server name: {e}"))?;
            let mut tls = connector.connect(server_name, tcp).await.context("TLS handshake")?;
            relay::write_line(&mut tls, &format!("V2 {ECHO_ADDR}")).await?;
            let response = relay::read_line(&mut tls).await?;
            if response != "OK" { anyhow::bail!("relay error: {response}"); }
            Ok(BenchStream::Tls(tls))
        }
    }
}

// ─── Workload: Interactive SSH ───

async fn workload_interactive(mode: Mode, rounds: u64) -> anyhow::Result<WorkloadMetrics> {
    let mut stream = get_stream(mode).await?;
    let mut times = Vec::with_capacity(rounds as usize);
    let mut total_bytes = 0u64;
    let start = Instant::now();

    for _ in 0..rounds {
        let t = Instant::now();
        stream.write_all(&[0x61]).await?;
        let mut buf = [0u8; 1];
        stream.read_exact(&mut buf).await?;
        times.push(t.elapsed());
        total_bytes += 2;
    }

    let elapsed = start.elapsed();
    times.sort();
    Ok(WorkloadMetrics {
        throughput_mbps: (total_bytes as f64) / elapsed.as_secs_f64() / 1_048_576.0,
        latency: percentiles(&times),
    })
}

// ─── Workload: Terminal UI / tmux ───

async fn workload_terminal_ui(mode: Mode, rounds: u64) -> anyhow::Result<WorkloadMetrics> {
    let mut stream = get_stream(mode).await?;
    let mut times = Vec::with_capacity(rounds as usize);
    let mut total_bytes = 0u64;
    let start = Instant::now();
    let burst_sizes: Vec<usize> = (0..rounds as usize).map(|i| 2048 + (i % 7) * 1024).collect();

    for &size in &burst_sizes {
        let data = vec![0x42u8; size];
        let t = Instant::now();
        stream.write_all(&data).await?;
        let mut read_buf = vec![0u8; size];
        stream.read_exact(&mut read_buf).await?;
        times.push(t.elapsed());
        total_bytes += (size * 2) as u64;
    }

    let elapsed = start.elapsed();
    times.sort();
    Ok(WorkloadMetrics {
        throughput_mbps: (total_bytes as f64) / elapsed.as_secs_f64() / 1_048_576.0,
        latency: percentiles(&times),
    })
}

// ─── Workload: Stream upload ───
// Uses io::split (not into_split) so it works for both TCP and TLS streams.

async fn workload_stream_up(mode: Mode, bytes: u64) -> anyhow::Result<WorkloadMetrics> {
    let stream = get_stream(mode).await?;
    let (mut reader, mut writer) = io::split(stream);
    let data = vec![0xABu8; 65536];
    let mut remaining = bytes;

    let read_target = bytes;
    let read_handle = tokio::spawn(async move {
        let mut buf = [0u8; 65536];
        let mut total = 0u64;
        while total < read_target {
            match reader.read(&mut buf).await {
                Ok(0) | Err(_) => break,
                Ok(n) => total += n as u64,
            }
        }
        total
    });

    let start = Instant::now();
    while remaining > 0 {
        let chunk = remaining.min(data.len() as u64) as usize;
        writer.write_all(&data[..chunk]).await?;
        remaining -= chunk as u64;
    }
    writer.shutdown().await?;
    let total_read = read_handle.await?;
    let elapsed = start.elapsed();

    Ok(WorkloadMetrics {
        throughput_mbps: ((bytes + total_read) as f64) / elapsed.as_secs_f64() / 1_048_576.0,
        latency: LatencyPoint {
            min_us: elapsed.as_micros() as u64,
            p50_us: elapsed.as_micros() as u64,
            p95_us: elapsed.as_micros() as u64,
            p99_us: elapsed.as_micros() as u64,
            max_us: elapsed.as_micros() as u64,
        },
    })
}

// ─── Workload: Stream download ───

async fn workload_stream_down(mode: Mode, bytes: u64) -> anyhow::Result<WorkloadMetrics> {
    workload_stream_up(mode, bytes).await
}

// ─── Workload: Mixed ───

async fn workload_mixed(mode: Mode, iterations: u64) -> anyhow::Result<WorkloadMetrics> {
    let mut stream = get_stream(mode).await?;
    let mut times = Vec::new();
    let mut total_bytes = 0u64;
    let start = Instant::now();

    for _ in 0..iterations {
        for _ in 0..20 {
            let t = Instant::now();
            stream.write_all(&[0x61]).await?;
            let mut buf = [0u8; 1];
            stream.read_exact(&mut buf).await?;
            times.push(t.elapsed());
            total_bytes += 2;
        }

        let bulk = vec![0xABu8; 4096];
        let bulk_total = 262144u64;
        let mut remaining = bulk_total;
        let t = Instant::now();
        while remaining > 0 {
            let chunk = remaining.min(bulk.len() as u64) as usize;
            stream.write_all(&bulk[..chunk]).await?;
            remaining -= chunk as u64;
        }
        let mut read_remaining = bulk_total;
        let mut read_buf = [0u8; 65536];
        while read_remaining > 0 {
            match stream.read(&mut read_buf).await {
                Ok(0) | Err(_) => break,
                Ok(n) => read_remaining -= n.min(read_remaining as usize) as u64,
            }
        }
        times.push(t.elapsed());
        total_bytes += bulk_total * 2;
    }

    let elapsed = start.elapsed();
    times.sort();
    Ok(WorkloadMetrics {
        throughput_mbps: (total_bytes as f64) / elapsed.as_secs_f64() / 1_048_576.0,
        latency: percentiles(&times),
    })
}

// ─── Workload: Connection setup ───

async fn workload_conn_setup(mode: Mode, rounds: u64) -> anyhow::Result<WorkloadMetrics> {
    let mut times = Vec::with_capacity(rounds as usize);

    for _ in 0..rounds {
        let t = Instant::now();
        let stream = get_stream(mode).await?;
        times.push(t.elapsed());
        drop(stream);
    }

    times.sort();
    let total = times.iter().sum::<Duration>();
    Ok(WorkloadMetrics {
        throughput_mbps: rounds as f64 / total.as_secs_f64(),
        latency: percentiles(&times),
    })
}

// ─── Concurrency scaling ───

async fn bench_concurrency(n: usize, bytes_per_conn: u64, mode: Mode) -> anyhow::Result<f64> {
    let start = Instant::now();

    let mut handles = Vec::with_capacity(n);
    for _ in 0..n {
        handles.push(tokio::spawn(async move {
            let stream = get_stream(mode).await.ok()?;
            let (mut reader, mut writer) = io::split(stream);
            let data = vec![0xABu8; 65536];
            let mut remaining = bytes_per_conn;
            let rt = bytes_per_conn;

            let rh = tokio::spawn(async move {
                let mut buf = [0u8; 65536];
                let mut total = 0u64;
                while total < rt {
                    match reader.read(&mut buf).await {
                        Ok(0) | Err(_) => break,
                        Ok(n) => total += n as u64,
                    }
                }
                total
            });

            while remaining > 0 {
                let chunk = remaining.min(data.len() as u64) as usize;
                if writer.write_all(&data[..chunk]).await.is_err() {
                    break;
                }
                remaining -= chunk as u64;
            }
            let _ = writer.shutdown().await;
            rh.await.ok()?;
            Some(bytes_per_conn)
        }));
    }

    let mut total_bytes = 0u64;
    for h in handles {
        if let Ok(Some(b)) = h.await {
            total_bytes += b;
        }
    }

    let secs = start.elapsed().as_secs_f64();
    Ok((total_bytes as f64) / secs / 1_048_576.0)
}

// ─── Helpers ───

fn percentiles(sorted: &[Duration]) -> LatencyPoint {
    let len = sorted.len();
    if len == 0 {
        return LatencyPoint { min_us: 0, p50_us: 0, p95_us: 0, p99_us: 0, max_us: 0 };
    }
    LatencyPoint {
        min_us: sorted[0].as_micros() as u64,
        p50_us: sorted[len / 2].as_micros() as u64,
        p95_us: sorted[len * 95 / 100].as_micros() as u64,
        p99_us: sorted[len * 99 / 100].as_micros() as u64,
        max_us: sorted[len - 1].as_micros() as u64,
    }
}

fn git_rev() -> String {
    std::process::Command::new("git")
        .args(["rev-parse", "--short", "HEAD"])
        .output()
        .ok()
        .and_then(|o| String::from_utf8(o.stdout).ok())
        .map(|s| s.trim().to_string())
        .unwrap_or_else(|| "unknown".to_string())
}

fn fmt_mbps(v: f64) -> String {
    if v >= 1000.0 { format!("{:.1} GB/s", v / 1024.0) }
    else { format!("{:.1} MB/s", v) }
}

fn fmt_us(v: u64) -> String {
    if v >= 1_000_000 { format!("{:.1}s", v as f64 / 1_000_000.0) }
    else if v >= 1_000 { format!("{:.1}ms", v as f64 / 1_000.0) }
    else { format!("{}us", v) }
}

fn average_comparisons(runs: &[WorkloadComparison]) -> WorkloadComparison {
    let n = runs.len() as f64;
    let avg_metrics = |get: &dyn Fn(&WorkloadComparison) -> &WorkloadMetrics| -> WorkloadMetrics {
        let tp = runs.iter().map(|r| get(r).throughput_mbps).sum::<f64>() / n;
        let lat = |f: &dyn Fn(&LatencyPoint) -> u64| -> u64 {
            (runs.iter().map(|r| f(&get(r).latency) as f64).sum::<f64>() / n) as u64
        };
        WorkloadMetrics {
            throughput_mbps: tp,
            latency: LatencyPoint {
                min_us: lat(&|l| l.min_us),
                p50_us: lat(&|l| l.p50_us),
                p95_us: lat(&|l| l.p95_us),
                p99_us: lat(&|l| l.p99_us),
                max_us: lat(&|l| l.max_us),
            },
        }
    };
    let has_tls = runs.iter().all(|r| r.relay_tls.is_some());
    WorkloadComparison {
        direct: avg_metrics(&|r| &r.direct),
        relay: avg_metrics(&|r| &r.relay),
        relay_tls: if has_tls {
            Some(avg_metrics(&|r| r.relay_tls.as_ref().unwrap()))
        } else {
            None
        },
    }
}

fn print_workload(name: &str, w: &WorkloadComparison) {
    eprintln!("  {name}:");
    eprintln!(
        "    direct  {} | p50={} p99={}",
        fmt_mbps(w.direct.throughput_mbps),
        fmt_us(w.direct.latency.p50_us),
        fmt_us(w.direct.latency.p99_us),
    );
    eprintln!(
        "    relay   {} | p50={} p99={}",
        fmt_mbps(w.relay.throughput_mbps),
        fmt_us(w.relay.latency.p50_us),
        fmt_us(w.relay.latency.p99_us),
    );
    if let Some(tls) = &w.relay_tls {
        eprintln!(
            "    tls     {} | p50={} p99={}",
            fmt_mbps(tls.throughput_mbps),
            fmt_us(tls.latency.p50_us),
            fmt_us(tls.latency.p99_us),
        );
    }
}

// ─── TLS setup ───

fn setup_tls_connector() -> anyhow::Result<()> {
    use tokio_rustls::rustls::ClientConfig;
    use tokio_rustls::rustls::client::danger::{ServerCertVerifier, ServerCertVerified, HandshakeSignatureValid};
    use tokio_rustls::rustls::pki_types::{CertificateDer, ServerName, UnixTime};
    use tokio_rustls::rustls::{DigitallySignedStruct, Error, SignatureScheme};

    /// Accepts any certificate. Only used for benchmarking against self-signed certs.
    #[derive(Debug)]
    struct BenchVerifier;

    impl ServerCertVerifier for BenchVerifier {
        fn verify_server_cert(
            &self, _: &CertificateDer<'_>, _: &[CertificateDer<'_>], _: &ServerName<'_>,
            _: &[u8], _: UnixTime,
        ) -> Result<ServerCertVerified, Error> {
            Ok(ServerCertVerified::assertion())
        }
        fn verify_tls12_signature(
            &self, _: &[u8], _: &CertificateDer<'_>, _: &DigitallySignedStruct,
        ) -> Result<HandshakeSignatureValid, Error> {
            Ok(HandshakeSignatureValid::assertion())
        }
        fn verify_tls13_signature(
            &self, _: &[u8], _: &CertificateDer<'_>, _: &DigitallySignedStruct,
        ) -> Result<HandshakeSignatureValid, Error> {
            Ok(HandshakeSignatureValid::assertion())
        }
        fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
            vec![
                SignatureScheme::ECDSA_NISTP256_SHA256,
                SignatureScheme::ECDSA_NISTP384_SHA384,
                SignatureScheme::ED25519,
                SignatureScheme::RSA_PSS_SHA256,
                SignatureScheme::RSA_PSS_SHA384,
                SignatureScheme::RSA_PSS_SHA512,
                SignatureScheme::RSA_PKCS1_SHA256,
                SignatureScheme::RSA_PKCS1_SHA384,
                SignatureScheme::RSA_PKCS1_SHA512,
            ]
        }
    }

    let config = ClientConfig::builder()
        .dangerous()
        .with_custom_certificate_verifier(Arc::new(BenchVerifier))
        .with_no_client_auth();
    let connector = TlsConnector::from(Arc::new(config));
    TLS_CONNECTOR.set(connector).ok();
    Ok(())
}

// ─── Main entry ───

pub async fn run(
    concurrency_levels: &[usize],
    bytes: u64,
    latency_rounds: u64,
    output: Option<&str>,
) -> anyhow::Result<()> {
    let iterations = 3u32;

    eprintln!("layover bench ({iterations} iterations)");
    eprintln!("─────────────────────────────────────────");

    // Start echo server as separate process (own CPU budget, doesn't compete with bench client)
    let mut echo_proc = std::process::Command::new(std::env::current_exe()?)
        .args(["bench-echo", "--listen", ECHO_ADDR])
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .spawn()
        .context("failed to start echo server subprocess")?;

    // Start no-TLS relay subprocess
    let mut relay_proc = std::process::Command::new(std::env::current_exe()?)
        .args(["serve", "--no-tls", "--listen", RELAY_ADDR, "--allow-private"])
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .spawn()
        .context("failed to start relay subprocess")?;

    // Generate self-signed cert and start TLS relay subprocess
    let cert_dir = std::env::temp_dir().join("layover-bench");
    std::fs::create_dir_all(&cert_dir)?;
    let cert_path = cert_dir.join("cert.pem");
    let key_path = cert_dir.join("key.pem");

    let cert_ok = std::process::Command::new("openssl")
        .args([
            "req", "-x509", "-newkey", "ec",
            "-pkeyopt", "ec_paramgen_curve:prime256v1",
            "-days", "1", "-nodes",
            "-keyout", key_path.to_str().unwrap(),
            "-out", cert_path.to_str().unwrap(),
            "-subj", "/CN=localhost",
            "-addext", "basicConstraints=critical,CA:FALSE",
        ])
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .status()
        .map(|s| s.success())
        .unwrap_or(false);

    let mut relay_tls_proc = if cert_ok {
        setup_tls_connector()?;
        let proc = std::process::Command::new(std::env::current_exe()?)
            .args([
                "serve", "--listen", RELAY_TLS_ADDR, "--allow-private",
                "--cert", cert_path.to_str().unwrap(),
                "--key", key_path.to_str().unwrap(),
            ])
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .spawn()
            .context("failed to start TLS relay subprocess")?;
        Some(proc)
    } else {
        eprintln!("  (openssl not available — skipping TLS benchmarks)");
        None
    };
    let has_tls = relay_tls_proc.is_some();

    tokio::time::sleep(Duration::from_millis(500)).await;
    TcpStream::connect(ECHO_ADDR).await.context("echo server not reachable")?;
    TcpStream::connect(RELAY_ADDR).await.context("relay server not reachable")?;
    if has_tls {
        TcpStream::connect(RELAY_TLS_ADDR).await.context("TLS relay not reachable")?;
    }

    // ── Run workloads N times ──
    let mut interactive_runs = Vec::new();
    let mut terminal_ui_runs = Vec::new();
    let mut stream_up_runs = Vec::new();
    let mut stream_down_runs = Vec::new();
    let mut mixed_runs = Vec::new();
    let mut conn_setup_runs = Vec::new();

    for i in 1..=iterations {
        eprintln!("\n  iteration {i}/{iterations}");

        eprint!("    interactive...");
        let d = workload_interactive(Mode::Direct, latency_rounds).await?;
        let r = workload_interactive(Mode::Relay, latency_rounds).await?;
        let t = if has_tls { Some(workload_interactive(Mode::RelayTls, latency_rounds).await?) } else { None };
        eprintln!(" d:p50={} r:p50={}{}", fmt_us(d.latency.p50_us), fmt_us(r.latency.p50_us),
            t.as_ref().map(|t| format!(" t:p50={}", fmt_us(t.latency.p50_us))).unwrap_or_default());
        interactive_runs.push(WorkloadComparison { direct: d, relay: r, relay_tls: t });

        eprint!("    terminal-ui...");
        let d = workload_terminal_ui(Mode::Direct, latency_rounds.min(200)).await?;
        let r = workload_terminal_ui(Mode::Relay, latency_rounds.min(200)).await?;
        let t = if has_tls { Some(workload_terminal_ui(Mode::RelayTls, latency_rounds.min(200)).await?) } else { None };
        eprintln!(" d:p50={} r:p50={}{}", fmt_us(d.latency.p50_us), fmt_us(r.latency.p50_us),
            t.as_ref().map(|t| format!(" t:p50={}", fmt_us(t.latency.p50_us))).unwrap_or_default());
        terminal_ui_runs.push(WorkloadComparison { direct: d, relay: r, relay_tls: t });

        eprint!("    stream-up...");
        let d = workload_stream_up(Mode::Direct, bytes).await?;
        let r = workload_stream_up(Mode::Relay, bytes).await?;
        let t = if has_tls { Some(workload_stream_up(Mode::RelayTls, bytes).await?) } else { None };
        eprintln!(" d:{} r:{}{}", fmt_mbps(d.throughput_mbps), fmt_mbps(r.throughput_mbps),
            t.as_ref().map(|t| format!(" t:{}", fmt_mbps(t.throughput_mbps))).unwrap_or_default());
        stream_up_runs.push(WorkloadComparison { direct: d, relay: r, relay_tls: t });

        eprint!("    stream-down...");
        let d = workload_stream_down(Mode::Direct, bytes).await?;
        let r = workload_stream_down(Mode::Relay, bytes).await?;
        let t = if has_tls { Some(workload_stream_down(Mode::RelayTls, bytes).await?) } else { None };
        eprintln!(" d:{} r:{}{}", fmt_mbps(d.throughput_mbps), fmt_mbps(r.throughput_mbps),
            t.as_ref().map(|t| format!(" t:{}", fmt_mbps(t.throughput_mbps))).unwrap_or_default());
        stream_down_runs.push(WorkloadComparison { direct: d, relay: r, relay_tls: t });

        eprint!("    mixed...");
        let d = workload_mixed(Mode::Direct, 20).await?;
        let r = workload_mixed(Mode::Relay, 20).await?;
        let t = if has_tls { Some(workload_mixed(Mode::RelayTls, 20).await?) } else { None };
        eprintln!(" d:{} r:{}{}", fmt_mbps(d.throughput_mbps), fmt_mbps(r.throughput_mbps),
            t.as_ref().map(|t| format!(" t:{}", fmt_mbps(t.throughput_mbps))).unwrap_or_default());
        mixed_runs.push(WorkloadComparison { direct: d, relay: r, relay_tls: t });

        eprint!("    conn-setup...");
        let d = workload_conn_setup(Mode::Direct, latency_rounds.min(200)).await?;
        let r = workload_conn_setup(Mode::Relay, latency_rounds.min(200)).await?;
        let t = if has_tls { Some(workload_conn_setup(Mode::RelayTls, latency_rounds.min(200)).await?) } else { None };
        eprintln!(" d:p50={} r:p50={}{}", fmt_us(d.latency.p50_us), fmt_us(r.latency.p50_us),
            t.as_ref().map(|t| format!(" t:p50={}", fmt_us(t.latency.p50_us))).unwrap_or_default());
        conn_setup_runs.push(WorkloadComparison { direct: d, relay: r, relay_tls: t });
    }

    // ── Average results ──
    let interactive = average_comparisons(&interactive_runs);
    let terminal_ui = average_comparisons(&terminal_ui_runs);
    let stream_up = average_comparisons(&stream_up_runs);
    let stream_down = average_comparisons(&stream_down_runs);
    let mixed = average_comparisons(&mixed_runs);
    let connection_setup = average_comparisons(&conn_setup_runs);

    eprintln!("\n─── averaged over {iterations} iterations ───\n");
    print_workload("interactive (keystrokes)", &interactive);
    print_workload("terminal-ui (2-8KB bursts)", &terminal_ui);
    print_workload("stream-up (bulk upload)", &stream_up);
    print_workload("stream-down (bulk download)", &stream_down);
    print_workload("mixed (typing + bulk)", &mixed);
    print_workload("conn-setup (handshake)", &connection_setup);

    // ── Concurrency scaling ──
    eprintln!();
    let bytes_per_conn = bytes / 10;
    let mut scaling = Vec::new();
    for &n in concurrency_levels {
        eprint!("  concurrency n={n}...");
        let direct_mbps = bench_concurrency(n, bytes_per_conn, Mode::Direct).await?;
        let relay_mbps = bench_concurrency(n, bytes_per_conn, Mode::Relay).await?;
        let tls_mbps = if has_tls {
            Some(bench_concurrency(n, bytes_per_conn, Mode::RelayTls).await?)
        } else {
            None
        };
        eprintln!(" d:{} r:{}{}", fmt_mbps(direct_mbps), fmt_mbps(relay_mbps),
            tls_mbps.map(|t| format!(" t:{}", fmt_mbps(t))).unwrap_or_default());
        scaling.push(ConcurrencyPoint { connections: n, direct_mbps, relay_mbps, relay_tls_mbps: tls_mbps });
    }

    // ── Build results ──
    let results = BenchResults {
        timestamp: SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs().to_string(),
        git_rev: git_rev(),
        iterations,
        workloads: WorkloadResults {
            interactive,
            terminal_ui,
            stream_up,
            stream_down,
            mixed,
            connection_setup,
        },
        concurrency_scaling: scaling,
    };

    eprintln!("\n─────────────────────────────────────────");

    let json = serde_json::to_string_pretty(&results)?;

    if let Some(path) = output {
        std::fs::write(path, &json)?;
        eprintln!("results saved to {path}");
    } else {
        println!("{json}");
    }

    echo_proc.kill().ok();
    echo_proc.wait().ok();
    relay_proc.kill().ok();
    relay_proc.wait().ok();
    if let Some(ref mut p) = relay_tls_proc {
        p.kill().ok();
        p.wait().ok();
    }
    let _ = std::fs::remove_dir_all(&cert_dir);

    Ok(())
}
