//! The load harness (`src/bench.rs`; `draft-evens-moq-bench-00`) run small against a relay on
//! loopback.
//!
//! - **Scenario 2**: one publisher, three subscribers, a video-like stream track and an
//!   audio-like datagram track. Every object sent on the stream track is received by every
//!   subscriber after START and before COMPLETION; of the datagram track's, which nothing
//!   sends again, at most one in twenty of those owed may be missing on a loaded host, and
//!   every subscriber has its START and its COMPLETION. The run's record is signed and reads
//!   back under its key, and the regression guard holds a later run on this host to it.
//!   Planted: the record changed after signing (refused), the same run claimed for another
//!   host (not comparable, never a pass), and a later run that loses objects (named by the
//!   guard).
//! - **Scenario 1**: the datagram track alone.
//! - **Scenario 3**, the meeting: two meetings of three participants, each publishing both
//!   tracks and subscribing to the other two's. Each track's result is over its six
//!   publishers and their two subscribers each, and no participant receives a track of its
//!   own or of the other meeting.
//!
//! - **The cost and the load**: a run given a meter of its relay states the relay's processor
//!   time over the run, the deliveries and the time a delivery, and the host's load before and
//!   after it; its publishers' and subscribers' sessions are opened by the two connectors
//!   given. Planted: a meter that cannot be read fails the run.
//!
//! The figures are this host's on this day and are asserted only against themselves and the
//! profile.

mod common;

use common::*;
use ed25519_dalek::SigningKey;
use joule_moq_relay::bench::{self, Host, Meeting, Meter, Profile, SignedRecord, Tolerances, TrackProfile, TrackResult};
use joule_moq_relay::client::ClientOptions;
use std::time::Duration;
use tokio::time::timeout;

fn track(name: &str, reliable: bool, priority: u8, interval_ms: u32, objects_per_group: u32, first: u32, rest: u32) -> TrackProfile {
    TrackProfile { namespace: vec!["perf".into(), "run".into()], name: name.into(), reliable, priority, interval_ms, objects_per_group, first_object_size: first, remaining_object_size: rest, duration_ms: 1200, start_delay_ms: 300 }
}

fn video() -> TrackProfile {
    track("video", true, 3, 30, 10, 4000, 600)
}

/// The audio-like track, on datagrams.
fn audio() -> TrackProfile {
    track("audio", false, 2, 20, 1, 120, 120)
}

fn profile() -> Profile {
    Profile { name: "smoke".into(), subscribers: 3, tracks: vec![video(), audio()], meeting: None }
}

/// Hold a track's result to what its profile had `publishers` sessions send to `subscribers`
/// each: a stream track whole; a datagram track with at most one in twenty of the objects
/// owed missing, and no subscriber without its START and COMPLETION.
fn held(t: &TrackProfile, r: &TrackResult, publishers: u64, subscribers: u64) {
    assert_eq!((r.publishers, r.objects_sent, r.subscribers_failed, r.unread_messages), (publishers, t.objects(), 0, 0), "{r:?}");
    assert_eq!(r.expected_bps, t.expected_bps());
    let owed = t.objects() * publishers * subscribers;
    println!("{}: {} of {owed} objects owed did not arrive (measured on this host, this run)", r.track, r.lost_objects);
    assert_eq!((r.owed, r.deliveries + r.lost_objects), (owed, owed), "{r:?}");
    // Every delivery's lateness from its publisher's send, each figure at most.
    assert!(r.late_p50_us > 0 && r.late_p50_us <= r.late_p99_us && r.late_p99_us <= r.late_max_us, "{r:?}");
    if t.reliable {
        assert_eq!((r.objects_received_min, r.lost_objects, r.groups_received_min), (r.objects_sent, 0, r.groups_sent), "{r:?}");
    } else {
        assert!(r.lost_objects * 20 <= owed, "{} of {owed} datagram objects did not arrive: {r:?}", r.lost_objects);
        assert!(r.objects_sent - r.objects_received_min <= r.lost_objects, "no subscriber is short by more than was lost in all: {r:?}");
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_run_is_measured_signed_and_guarded() {
    timeout(Duration::from_secs(180), async {
        let server = relay(|_| {}).await;
        let endpoints = std::sync::Mutex::new(Vec::new());
        let connect = || async {
            let (c, ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
            endpoints.lock().unwrap().push(ep);
            Ok(c)
        };
        let p = profile();
        let record = bench::run(&p, Host::here("test-host"), "quic", connect).await.expect("a run");
        assert_eq!(record.kind, bench::KIND);
        assert_eq!(record.date.len(), 10);
        assert_eq!(record.tracks.len(), 2);
        for (t, r) in p.tracks.iter().zip(&record.tracks) {
            held(t, r, 1, 3);
        }

        // Signed, and read only under its key.
        let key = SigningKey::from_bytes(&[0x33; 32]);
        let trusted = key.verifying_key().to_bytes();
        let signed = SignedRecord::sign(&record, &key);
        let text = serde_json::to_string(&signed).unwrap();
        let back: SignedRecord = serde_json::from_str(&text).unwrap();
        assert_eq!(back.verify(&trusted).as_ref(), Ok(&record));
        let forged = SignedRecord { statement: back.statement.replacen("\"subscribers\":3", "\"subscribers\":300", 1), ..back.clone() };
        assert_ne!(forged.statement, back.statement);
        assert_eq!(forged.verify(&trusted).unwrap_err(), "record_signature_invalid");

        // The guard: a second run on this host against the first, with tolerances wide enough
        // for a loaded CI host (the guard's arithmetic is held exactly in the unit tests).
        let mut again = bench::run(&p, Host::here("test-host"), "quic", connect).await.expect("a second run");
        // The guard compares no run of a host above the idle rule (held in the unit tests);
        // the comparison of two runs is held here, on whatever load CI gives, without it.
        let mut record = record.clone();
        (record.load, again.load) = (None, None);
        // A datagram is not sent again, so the datagram track's allowance of this test is the
        // guard's too.
        let loose = Tolerances { lost_objects: 9, receive_variance_us: 2_000_000, max_delta_ms: 5_000, bps_percent: 90, cpu_percent: 1000, delivered_ppm: 50_000 };
        assert_eq!(bench::guard(&record, &again, &loose), Ok(vec![]));
        // Planted: the same run claimed for another host.
        let mut elsewhere = again.clone();
        elsewhere.host.label = "another-host".into();
        assert!(bench::guard(&record, &elsewhere, &loose).unwrap_err().starts_with("not_comparable"));
        // Planted: a run that lost objects.
        let mut lossy = again.clone();
        lossy.tracks[0].lost_objects = record.tracks[0].lost_objects + 10;
        lossy.tracks[0].objects_received_min -= 10;
        let named = bench::guard(&record, &lossy, &loose).unwrap();
        assert_eq!(named.len(), 1);
        assert!(named[0].starts_with("perf/run/video: lost_objects 10"), "{named:?}");
        server.shutdown().await;
    })
    .await
    .expect("the test finishes in time");
}

/// One more session on `server`, its endpoint kept in `endpoints` for as long as the test.
async fn session(server: &joule_moq_relay::RelayServer, endpoints: &std::sync::Mutex<Vec<Option<quinn::Endpoint>>>) -> Result<joule_moq_relay::client::Client, String> {
    let (c, ep) = connect(server, Via::Quic, ClientOptions::default()).await;
    endpoints.lock().unwrap().push(ep);
    Ok(c)
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_datagram_track_alone_is_driven() {
    timeout(Duration::from_secs(120), async {
        let server = relay(|_| {}).await;
        let endpoints = std::sync::Mutex::new(Vec::new());
        let p = Profile { name: "scenario-1".into(), subscribers: 4, tracks: vec![audio()], meeting: None };
        let record = bench::run(&p, Host::here("test-host"), "quic", || session(&server, &endpoints)).await.expect("a run");
        held(&p.tracks[0], &record.tracks[0], 1, 4);
        // A datagram track whose object does not fit a datagram is refused before anything is sent.
        let big = Profile { tracks: vec![TrackProfile { first_object_size: 4000, ..audio() }], ..p.clone() };
        assert!(bench::run(&big, Host::here("test-host"), "quic", || session(&server, &endpoints)).await.unwrap_err().contains("one datagram"));
        server.shutdown().await;
    })
    .await
    .expect("the test finishes in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_meeting_is_driven_every_participant_to_every_other() {
    timeout(Duration::from_secs(180), async {
        let server = relay(|_| {}).await;
        let endpoints = std::sync::Mutex::new(Vec::new());
        let p = Profile { name: "scenario-3".into(), subscribers: 2, tracks: vec![video(), audio()], meeting: Some(Meeting { meetings: 2, participants: 3 }) };
        let record = bench::run(&p, Host::here("test-host"), "quic", || session(&server, &endpoints)).await.expect("a run");
        // Six sessions and no more: each participant publishes and subscribes on one.
        assert_eq!(endpoints.lock().unwrap().len(), 6);
        assert_eq!(record.profile, p);
        for (t, r) in p.tracks.iter().zip(&record.tracks) {
            // Six publishers of each track, each taken by the two others of its meeting: a
            // participant that also received its own, or the other meeting's, would have
            // more than was owed, and one that missed a participant fewer.
            held(t, r, 6, 2);
        }
        // The record reads back with its meeting.
        let key = SigningKey::from_bytes(&[0x34; 32]);
        let back = SignedRecord::sign(&record, &key).verify(&key.verifying_key().to_bytes()).expect("the record");
        assert_eq!(back.profile.meeting, Some(Meeting { meetings: 2, participants: 3 }));
        // A meeting's record is not comparable with a run of another shape.
        let mut other = record.clone();
        other.profile.meeting = Some(Meeting { meetings: 1, participants: 3 });
        assert!(bench::guard(&record, &other, &Tolerances::default()).unwrap_err().starts_with("not_comparable"));
        server.shutdown().await;
    })
    .await
    .expect("the test finishes in time");
}

/// A meter of the relay in this process: its time is read with the clients' here, which a
/// record never does (the tool runs each relay as a process of its own); what is held is
/// that the run reads it and states it.
struct InProcess(bool);

impl Meter for InProcess {
    fn relays(&self) -> Vec<(String, u64)> {
        vec![("relay".into(), 4)]
    }

    fn cpu_us(&self) -> Result<Vec<u64>, String> {
        if self.0 { Ok(vec![bench::own_cpu_us()]) } else { Err("the meter is not read".into()) }
    }
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_run_states_its_cost_and_its_load() {
    timeout(Duration::from_secs(120), async {
        let server = relay(|_| {}).await;
        let endpoints = std::sync::Mutex::new(Vec::new());
        let (pubs, subs) = (std::sync::atomic::AtomicUsize::new(0), std::sync::atomic::AtomicUsize::new(0));
        let p = Profile { name: "cost".into(), subscribers: 3, tracks: vec![video()], meeting: None };
        let publisher = || {
            pubs.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            session(&server, &endpoints)
        };
        let subscriber = || {
            subs.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            session(&server, &endpoints)
        };
        let record = bench::run_with(&p, Host::here("test-host"), "quic", publisher, subscriber, Some(&InProcess(true))).await.expect("a run");
        held(&p.tracks[0], &record.tracks[0], 1, 3);
        assert_eq!((pubs.load(std::sync::atomic::Ordering::Relaxed), subs.load(std::sync::atomic::Ordering::Relaxed)), (1, 3), "each session by its own connector");
        let cost = record.cost.as_ref().expect("a cost, with a meter");
        assert_eq!(cost.deliveries, record.tracks[0].deliveries);
        assert_eq!((cost.relays.len(), cost.relays[0].role.as_str(), cost.relays[0].threads), (1, "relay", 4));
        assert!(cost.relays[0].cpu_us > 0 && cost.cpu_ns_per_delivery == cost.relays[0].cpu_us * 1000 / cost.deliveries, "{cost:?}");
        assert!(cost.wall_ms >= u64::from(p.tracks[0].duration_ms), "from the first object to the last subscriber's end: {cost:?}");
        let load = record.load.as_ref().expect("this host states its load");
        assert_eq!(load.cpus, record.host.cpus);
        println!("the host's one-minute load {}.{:02} before, {}.{:02} after, on {} processors; idle by the bench's rule: {}", load.before_x100 / 100, load.before_x100 % 100, load.after_x100 / 100, load.after_x100 % 100, load.cpus, record.idle());
        // The record reads back with them, and a run without a meter states none.
        let key = SigningKey::from_bytes(&[0x35; 32]);
        assert_eq!(SignedRecord::sign(&record, &key).verify(&key.verifying_key().to_bytes()).as_ref(), Ok(&record));
        // Planted: a meter that cannot be read fails the run.
        let err = bench::run_with(&p, Host::here("test-host"), "quic", || session(&server, &endpoints), || session(&server, &endpoints), Some(&InProcess(false))).await.unwrap_err();
        assert_eq!(err, "the meter is not read");
        server.shutdown().await;
    })
    .await
    .expect("the test finishes in time");
}