//! `GET /metrics`, a resource of the gateway's table: it reads strictly as the text format; a
//! refused request, a WHIP session and a WHEP session move the requests, the refusals, the
//! sessions running and opened and the records signed, as each session opens and as it ends; an
//! RTMP connection counts as a session once its publish is admitted, and not before; an SRT
//! session counts while it runs; the energy marker says no meter backs the gateway; `HEAD` is
//! answered as `GET` is, without the text; every resource of the gateway's table, asked once, is
//! counted by its pattern; through a front the path is one the gateway does not have, to a request
//! with a grant too; and no room, principal, grant, session or address of those requests is in it.

mod common;

use common::rtmp::{stream_key, Client};
use common::*;
use ed25519_dalek::SigningKey;
use joule_contrib::srt::{self, CallConfig};
use joule_moq_ingest::frame::Codec;
use std::collections::BTreeSet;
use std::time::{Duration, Instant};
use str0m::media::Direction;
use tokio::net::TcpStream;
use tokio::time::timeout;
use wai_metrics::read::{identifier_like, label_values, parse, sum, value, Family};

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn metrics_move_with_a_session_and_name_no_party() {
    timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time");
}

async fn scrape(w: &World) -> (String, Vec<Family>) {
    let r = http("GET", &format!("{}/metrics", w.base()), &[], b"").await;
    assert_eq!(r.status, 200, "{}", r.text());
    assert_eq!(r.header("content-type"), Some(wai_metrics::CONTENT_TYPE));
    assert!(r.header("access-control-allow-origin").is_none(), "no page of another origin reads it");
    let text = r.text();
    let f = parse(&text).unwrap_or_else(|e| panic!("{e}\n{text}"));
    (text, f)
}

fn requests(f: &[Family], route: &str, method: &str, code: &str) -> Option<f64> {
    value(f, "wai_ingest_requests_total", &[("route", route), ("method", method), ("code", code)])
}

/// The sessions of `protocol` running, as the gauge states them.
fn running(f: &[Family], protocol: &str) -> Option<f64> {
    value(f, "wai_ingest_sessions", &[("protocol", protocol)])
}

/// The requests answered with a client or a server error.
fn errors(f: &[Family]) -> f64 {
    f.iter().flat_map(|x| &x.samples).filter(|s| s.name == "wai_ingest_requests_total" && s.labels.iter().any(|(k, v)| k == "code" && (v.starts_with('4') || v.starts_with('5')))).map(|s| s.value).sum()
}

async fn run() {
    let w = World::start(|_| {}).await;
    let (_, f) = scrape(&w).await;
    assert_eq!(value(&f, "wai_ingest_sessions", &[("protocol", "whip")]), Some(0.0));
    assert_eq!(value(&f, "wai_ingest_sessions_opened_total", &[("protocol", "whip")]), Some(0.0));
    assert_eq!(value(&f, "wai_ingest_energy_unmetered", &[]), Some(1.0));
    assert_eq!(value(&f, "wai_ingest_records_total", &[]), Some(0.0));
    // A scrape that carries an Authorization field is not read as a grant.
    let r = http("GET", &format!("{}/metrics", w.base()), &[("authorization", "Bearer bm90LWEtZ3JhbnQ")], b"").await;
    assert_eq!(r.status, 200, "{}", r.text());
    assert!(r.header("wai-room-challenge").is_none());
    let r = http("POST", &format!("{}/metrics", w.base()), &[], b"").await;
    assert_eq!((r.status, r.header("allow")), (405, Some("GET, HEAD")));
    let r = http("HEAD", &format!("{}/metrics", w.base()), &[], b"").await;
    assert_eq!((r.status, r.body.is_empty()), (200, true));
    assert_eq!(r.header("content-type"), Some(wai_metrics::CONTENT_TYPE));
    // Through a front, the scrape is no resource here.
    let r = http("GET", &format!("{}/metrics", w.base()), &[("x-forwarded-for", "198.51.100.9")], b"").await;
    assert_eq!(r.status, 404, "{}", r.text());
    assert!(!r.text().contains("wai_ingest"), "{}", r.text());
    assert!(r.header("access-control-allow-origin").is_some(), "as any other resource the gateway does not have");
    // And with a grant's field, as at a path the gateway does not have: read, and answered so.
    let grant = ("authorization", "Bearer bm90LWEtZ3JhbnQ");
    let absent = http("GET", &format!("{}/no-such-resource", w.base()), &[grant], b"").await;
    let r = http("GET", &format!("{}/metrics", w.base()), &[grant, ("x-forwarded-for", "198.51.100.9")], b"").await;
    assert_eq!(r.status, absent.status, "{}", r.text());
    assert_eq!(r.text(), absent.text());
    assert_eq!(r.header("wai-room-challenge").is_some(), absent.header("wai-room-challenge").is_some(), "and each takes the next challenge alike");

    // A viewer's grant may not publish: refused.
    let whip = format!("{}/whip", w.base());
    let viewer = SigningKey::from_bytes(&[0x61; 32]);
    let va = w.auth(&viewer, 1);
    let o = offer(&[Codec::Avc, Codec::Opus], Direction::SendOnly, true, true).await;
    assert_eq!(va.send("POST", &whip, &[("content-type", "application/sdp")], o.offer.as_bytes()).await.status, 403);

    // A publisher's session, while it runs.
    let a = w.auth(&w.publisher, 0);
    let p = publish(&w, &a).await;
    let (text, f) = scrape(&w).await;
    assert_eq!(value(&f, "wai_ingest_sessions", &[("protocol", "whip")]), Some(1.0));
    assert_eq!(value(&f, "wai_ingest_sessions", &[("protocol", "whep")]), Some(0.0));
    assert_eq!(value(&f, "wai_ingest_sessions_opened_total", &[("protocol", "whip")]), Some(1.0));
    assert_eq!((requests(&f, "/whip", "POST", "201"), requests(&f, "/whip", "POST", "403")), (Some(1.0), Some(1.0)));
    assert_eq!((requests(&f, "/metrics", "GET", "200"), requests(&f, "/metrics", "POST", "405")), (Some(2.0), Some(1.0)));
    assert_eq!(requests(&f, "/metrics", "HEAD", "200"), Some(1.0));
    assert_eq!(value(&f, "wai_ingest_refused_total", &[]), Some(errors(&f)), "every refusal is a request answered with an error");
    let mut texts = vec![text];

    // A player's session, once the publisher's catalog is there: the gateway's fan-out.
    let mut sent = Sent::default();
    send(&p, &mut sent, 1, 15).await;
    let deadline = Instant::now() + STEP;
    let (o, resp) = loop {
        let o = offer(&[Codec::Avc, Codec::Opus], Direction::RecvOnly, true, true).await;
        let r = va.send("POST", &w.whep_url(), &[("content-type", "application/sdp")], o.offer.as_bytes()).await;
        if r.status == 201 {
            break (o, r);
        }
        assert_eq!(r.status, 409, "{}", r.text());
        assert!(Instant::now() < deadline, "the catalog arrives");
        tokio::time::sleep(Duration::from_millis(200)).await;
    };
    let player = o.start(&resp);
    player.connected().await;
    let (text, f) = scrape(&w).await;
    assert_eq!((running(&f, "whep"), running(&f, "whip")), (Some(1.0), Some(1.0)));
    assert_eq!(value(&f, "wai_ingest_sessions_opened_total", &[("protocol", "whep")]), Some(1.0));
    assert_eq!(requests(&f, "/whep/{room}/{owner}", "POST", "201"), Some(1.0));
    assert_eq!(value(&f, "wai_ingest_refused_total", &[]), Some(errors(&f)), "a 409 while the catalog is awaited is a refusal too");
    texts.push(text);

    // The player leaves: the fan-out falls, what was opened stays counted, and its record is signed.
    assert_eq!(delete(&w, &player.location, &va).await.status, 200);
    assert!(until(STEP, || w.gateway.gateway.records().len() == 1).await, "the player's record");
    assert!(until(STEP, || w.gateway.gateway.whep.lock().unwrap().is_empty()).await, "the player's session is gone");
    let (text, f) = scrape(&w).await;
    assert_eq!((running(&f, "whep"), running(&f, "whip")), (Some(0.0), Some(1.0)));
    assert_eq!(value(&f, "wai_ingest_sessions_opened_total", &[("protocol", "whep")]), Some(1.0));
    assert_eq!(requests(&f, "/whep/s/{id}", "DELETE", "200"), Some(1.0));
    texts.push(text);

    // The publisher's session ends: its record is signed.
    assert_eq!(delete(&w, &p.location, &a).await.status, 200);
    assert!(until(STEP, || w.gateway.gateway.records().len() == 2).await, "the session's record");
    assert!(until(STEP, || w.gateway.gateway.whip.lock().unwrap().is_empty()).await, "the session is gone");
    let r = http("GET", &format!("{}/whep/s/{}", w.base(), "ab".repeat(16)), &[], b"").await;
    assert_eq!(r.status, 404);
    assert!(r.header("access-control-allow-origin").is_some(), "a page reads the gateway's other answers");
    let (text, f) = scrape(&w).await;
    assert_eq!(value(&f, "wai_ingest_sessions", &[("protocol", "whip")]), Some(0.0));
    assert_eq!(value(&f, "wai_ingest_records_total", &[]), Some(2.0));
    assert_eq!(requests(&f, "/whip/s/{id}", "DELETE", "200"), Some(1.0));
    assert_eq!(requests(&f, "/whep/s/{id}", "GET", "404"), Some(1.0), "a session's resource is counted by its pattern");
    assert_eq!(value(&f, "wai_ingest_refused_total", &[]), Some(errors(&f)));
    assert!(sum(&f, "wai_ingest_requests_total", &[]) >= 8.0);
    texts.push(text);

    // No party and no thing of one: not a label, and nowhere in the text.
    let session = p.location.rsplit('/').next().unwrap().to_string();
    let played = player.location.rsplit('/').next().unwrap().to_string();
    let ids = [
        hex::encode(w.room),
        w.owner(),
        hex::encode(viewer.verifying_key().to_bytes()),
        hex::encode(w.gateway_key.verifying_key().to_bytes()),
        hex::encode(w.authority.verifying_key().to_bytes()),
        a.token(),
        va.token(),
        session,
        played,
        "ab".repeat(16),
        "127.0.0.1".to_string(),
    ];
    for (text, f) in texts.iter().map(|t| (t, parse(t).unwrap())) {
        for l in label_values(&f) {
            assert!(!identifier_like(l), "a label value {l:?}");
        }
        for id in &ids {
            assert!(!text.contains(id.as_str()), "{id} is in the metrics");
        }
    }
    p.close();
    player.close();
    w.stop().await;
}

/// The gateway holds an RTMP connection from its first byte. One that says nothing, and one that
/// connects and presents no stream key, are no session; a publish is one once it is admitted,
/// beside them, and not after it ends.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_rtmp_connection_is_a_session_once_its_publish_is_admitted() {
    timeout(Duration::from_secs(120), async {
        let w = World::start(|c| {
            c.rtmp.listen = Some("127.0.0.1:0".parse().unwrap());
            // The idle connections below outlast the test on a loaded host.
            c.rtmp.connect_timeout_ms = 100_000;
        })
        .await;
        let addr = w.gateway.rtmp_addr.expect("an RTMP listener");
        let connections = || w.gateway.gateway.rtmp.lock().unwrap().len();

        let silent = TcpStream::connect(addr).await.unwrap();
        let mut idle = Client::start(TcpStream::connect(addr).await.unwrap()).await;
        idle.connect(0).await;
        assert!(until(STEP, || connections() == 2).await, "both connections are held");
        let (_, f) = scrape(&w).await;
        assert_eq!(running(&f, "rtmp"), Some(0.0), "neither presented a stream key");

        let auth = w.auth(&w.publisher, 0);
        let key = stream_key(&w, &auth).await;
        let mut c = Client::start(TcpStream::connect(addr).await.unwrap()).await;
        c.connect(0).await;
        let s = c.publish(&key).await;
        assert_eq!(s.code, "NetStream.Publish.Start", "{}", s.description);
        assert_eq!(connections(), 3);
        let (_, f) = scrape(&w).await;
        assert_eq!(running(&f, "rtmp"), Some(1.0), "the admitted publish, and neither idle connection");

        c.delete_stream().await;
        c.shutdown().await;
        assert!(until(STEP, || connections() == 2).await, "the publish's connection is gone");
        let (_, f) = scrape(&w).await;
        assert_eq!(running(&f, "rtmp"), Some(0.0));

        drop((silent, idle));
        w.stop().await;
    })
    .await
    .expect("the test finishes in time");
}

/// An SRT session counts from its call being admitted until it ends, and not while it waits.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_contribution_session_counts_while_it_runs() {
    timeout(Duration::from_secs(120), async {
        let w = World::start(|c| {
            c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap());
            c.contrib.srt.latency_ms = 250;
        })
        .await;
        let auth = w.auth(&w.publisher, 0);
        let r = auth.send("POST", &format!("{}/srt", w.base()), &[("content-type", "application/json")], b"").await;
        assert_eq!(r.status, 201, "POST /srt: {}", r.text());
        let s: serde_json::Value = serde_json::from_slice(&r.body).unwrap();
        let (_, f) = scrape(&w).await;
        assert_eq!(running(&f, "contribution"), Some(0.0), "opened, and not called yet");

        let call = CallConfig {
            stream_id: s["stream_id"].as_str().map(str::to_string),
            passphrase: s["passphrase"].as_str().map(str::to_string),
            keylen: s["pbkeylen"].as_u64().unwrap() as usize,
            latency_ms: 300,
            ..CallConfig::default()
        };
        let link = timeout(STEP, srt::call(w.gateway.srt_addr.unwrap(), call)).await.unwrap().expect("the call is admitted");
        assert!(contrib_live(&w, s["session"].as_str().unwrap()).await, "the session publishes its tracks");
        let (_, f) = scrape(&w).await;
        assert_eq!(running(&f, "contribution"), Some(1.0));

        link.close();
        assert!(until(STEP, || w.gateway.gateway.contrib.sessions.lock().unwrap().is_empty()).await, "the session is gone");
        let (_, f) = scrape(&w).await;
        assert_eq!(running(&f, "contribution"), Some(0.0));
        w.stop().await;
    })
    .await
    .expect("the test finishes in time");
}

/// A path of each resource `Gateway::answer` matches, read from its table: each slice pattern
/// there, every alternative of a segment taken, any other segment a sample one. A resource added
/// to the table is asked with no test changed.
fn resources() -> Vec<String> {
    let source = include_str!("../src/gateway.rs");
    let table = source.split_once("let r = match (seg.as_slice(), &m) {").expect("the table").1;
    let table = table.split_once("_ => problem(StatusCode::NOT_FOUND").expect("its end").0;
    let mut paths = BTreeSet::new();
    for arm in table.split("([").skip(1) {
        let pattern = arm.split_once(']').expect("a slice pattern").0;
        let mut found = vec![String::new()];
        for segment in pattern.split(", ") {
            let literals: Vec<&str> = segment.split('"').skip(1).step_by(2).collect();
            let choices = if literals.is_empty() { vec!["0"] } else { literals };
            found = found.iter().flat_map(|p| choices.iter().map(move |c| format!("{p}/{c}"))).collect();
        }
        paths.extend(found);
    }
    paths.into_iter().collect()
}

/// Every resource of the gateway's table, asked once, is counted by its pattern: one left out of
/// `route_of` would be counted as `other`.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn every_resource_of_the_table_is_counted_by_its_pattern() {
    timeout(Duration::from_secs(120), async {
        let w = World::start(|_| {}).await;
        let paths = resources();
        assert!(paths.len() >= 10, "{paths:?}");
        for p in &paths {
            http("GET", &format!("{}{p}", w.base()), &[], b"").await;
        }
        let (text, f) = scrape(&w).await;
        assert_eq!(sum(&f, "wai_ingest_requests_total", &[("route", "other")]), 0.0, "{text}");
        assert!(sum(&f, "wai_ingest_requests_total", &[]) >= paths.len() as f64, "{text}");
        w.stop().await;
    })
    .await
    .expect("the test finishes in time");
}