//! The gateway's throughput and energy on a contribution link: a sender in another
//! process (`joule-contrib-send`, named by `WAI_CONTRIB_SEND`) sends a synthetic program at
//! `WAI_BENCH_KBPS` (20 000 by default) for `WAI_BENCH_SECONDS` (20) over SRT, encrypted, into
//! the gateway, which publishes every object with its receipts to a relay whose door checks
//! every receipt; both are this process. It prints one JSON line: the payload rate carried,
//! the objects published, the process's CPU time and its energy by the counter's profile,
//! class and scope (the gateway and the relay; the sender's process is not counted).
//!
//! ```text
//! WAI_CONTRIB_SEND=…/release/joule-contrib-send cargo test --release -p joule-moq-ingest \
//!   --test contrib_bench -- --ignored --nocapture
//! ```

mod common;

use common::*;
use joule_contrib::energy;
use joule_moq_ingest::linkrec;
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};

fn env_u64(k: &str, d: u64) -> u64 {
    std::env::var(k).ok().and_then(|v| v.parse().ok()).unwrap_or(d)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[ignore = "a measurement: set WAI_CONTRIB_SEND, run with --release"]
async fn contribution_throughput_and_energy() {
    // What the relay refuses at its door, and why, goes to the test's output.
    let _ = tracing_subscriber::fmt().with_max_level(tracing::Level::WARN).with_test_writer().try_init();
    let send = std::env::var("WAI_CONTRIB_SEND").expect("WAI_CONTRIB_SEND names joule-contrib-send");
    let kbps = env_u64("WAI_BENCH_KBPS", 20_000);
    let seconds = env_u64("WAI_BENCH_SECONDS", 20);
    let w = World::start(|c| {
        c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap());
        c.contrib.srt.latency_ms = 200;
        c.contrib.max_bitrate_kbps = 2_000_000;
        c.limits.max_frame_bytes = 16 << 20;
        c.limits.max_queue_bytes = 512 << 20;
    })
    .await;
    let auth = w.auth(&w.publisher, 0);
    let r = auth.send("POST", &format!("{}/srt", w.base()), &[], b"").await;
    assert_eq!(r.status, 201, "{}", r.text());
    let s: serde_json::Value = serde_json::from_slice(&r.body).unwrap();
    let url = format!(
        "srt://127.0.0.1:{}?latency=200&streamid={}&passphrase={}&pbkeylen=16",
        w.gateway.srt_addr.unwrap().port(),
        s["stream_id"].as_str().unwrap().replace('#', "%23").replace(',', "%2C").replace('=', "%3D").replace('/', "%2F"),
        s["passphrase"].as_str().unwrap()
    );
    // On a counter every user of the package shares (linux-rapl), the figure is a bracket:
    // the run's reading less an idle reading taken just before it, scaled to the run's length.
    let idle = match energy::read() {
        Some(r) if r.scope == "bracket" => {
            let t = Instant::now();
            tokio::time::sleep(Duration::from_secs(seconds)).await;
            Some((r.to(&energy::read().unwrap()), t.elapsed().as_secs_f64()))
        }
        _ => None,
    };
    let e0 = energy::read();
    let c0 = energy::cpu_us();
    let t0 = Instant::now();
    let mut child = Command::new(&send).args([&url, "--synthetic", &seconds.to_string(), "--kbps", &kbps.to_string(), "--linger", "1000"]).stdout(Stdio::piped()).stderr(Stdio::null()).spawn().expect("the sender runs");
    let out = tokio::task::spawn_blocking(move || {
        let mut o = String::new();
        std::io::Read::read_to_string(child.stdout.as_mut().unwrap(), &mut o).unwrap();
        let st = child.wait().unwrap();
        (st, o)
    })
    .await
    .unwrap();
    assert!(out.0.success(), "the sender finished: {}", out.1);
    assert!(until(Duration::from_secs(30), || w.gateway.gateway.link_records().iter().any(|b| linkrec::LinkRecord::read(b).is_ok_and(|r| r.last))).await, "the session ended");
    let wall = t0.elapsed().as_secs_f64();
    let c1 = energy::cpu_us();
    let e1 = energy::read();
    let parts = linkrec::verify_series(&w.gateway.gateway.link_records()).expect("the link record verifies");
    let bytes: u64 = parts.iter().map(|p| p.account.bytes).sum();
    let packets: u64 = parts.iter().map(|p| p.account.delivered).sum();
    // The session's record is signed once its tracks have ended at the relay, after its
    // last link part.
    assert!(until(Duration::from_secs(30), || w.gateway.gateway.records().iter().any(|r| r.record.protocol == "srt")).await, "the session record");
    let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.protocol == "srt").unwrap();
    let objects: u64 = rec.record.tracks.iter().filter(|t| t.dir == 2).map(|t| t.objects).sum();
    let raw = match (&e0, &e1) {
        (Some(a), Some(b)) => Some(a.to(b) as f64 / 1e9),
        _ => None,
    };
    let idle_j = idle.map(|(nj, s)| nj as f64 / 1e9 * wall / s);
    let joules = match (raw, idle_j) {
        (Some(r), Some(i)) => Some(r - i),
        (r, _) => r,
    };
    let gb = bytes as f64 / 1e9;
    println!(
        "{}",
        serde_json::json!({
            "bench": "gateway-srt-aes128",
            "kbps_target": kbps,
            "seconds": seconds,
            "wall_s": wall,
            "payload_bytes": bytes,
            "packets": packets,
            "objects_published": objects,
            "payload_mbps": bytes as f64 * 8.0 / 1e6 / seconds as f64,
            "cpu_s": (c1 - c0) as f64 / 1e6,
            "refused_at_door": w.relay.relay().stats().refused_at_door,
            "sender": serde_json::from_str::<serde_json::Value>(out.1.trim()).unwrap_or(serde_json::Value::Null),
            "energy": e1.map(|e| serde_json::json!({"profile": e.profile, "class": e.class, "scope": e.scope, "joules": joules, "raw_joules": raw, "idle_joules": idle_j, "j_per_gb": joules.map(|j| j / gb.max(1e-9)), "covers": if e.scope == "task" { "the gateway and the relay, one process; not the sender" } else { "the package less an idle bracket: the gateway, the relay and the sender together" }})),
        })
    );
    assert_eq!(w.relay.relay().stats().refused_at_door, 0);
    w.stop().await;
}