//! `GET /metrics` on the relay's control listener (`src/control.rs`), on loopback:
//!
//! - it reads strictly as the text format, every family present from the start;
//! - its series move with real sessions: a publisher and a subscriber over the WebSocket binding
//!   are two sessions open and two opened, the publisher's two tracks are held, the subscriber's
//!   two subscriptions are the fan-out, and the objects published are counted in; once both
//!   leave, no session is open and the opened count stands;
//! - `HEAD` is answered as `GET`, without the text; another method is `405`; another path, and a
//!   request through a front, `404`;
//! - a JSON-lines request on the same listener is answered as before, and counted by its `op`;
//! - no label value in it is a key, a digest, an address or an account.

mod common;

use common::*;
use ed25519_dalek::SigningKey;
use joule_moq::draft22::Parameters;
use joule_moq_relay::client::ClientOptions;
use serde_json::Value;
use std::net::SocketAddr;
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::time::timeout;
use wai_metrics::read::{identifier_like, label_values, parse, value, Family};

/// One request on the control listener at `addr`, and every byte it is answered with until the
/// relay closes the connection.
async fn http(addr: SocketAddr, req: &str) -> String {
    let mut s = tokio::net::TcpStream::connect(addr).await.expect("the control listener");
    s.write_all(req.as_bytes()).await.unwrap();
    let mut out = Vec::new();
    timeout(STEP, s.read_to_end(&mut out)).await.expect("the connection ends").expect("the answer, then the end of the connection");
    String::from_utf8(out).expect("UTF-8")
}

/// `GET /metrics`: its status line, and its families as the format's reader reads them.
async fn scrape(addr: SocketAddr) -> Vec<Family> {
    let said = http(addr, "GET /metrics HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n").await;
    let (head, body) = said.split_once("\r\n\r\n").unwrap_or_else(|| panic!("an HTTP answer: {said:?}"));
    assert!(head.starts_with("HTTP/1.1 200 OK\r\n"), "{head}");
    assert!(head.contains(&format!("Content-Type: {}\r\n", wai_metrics::CONTENT_TYPE)), "{head}");
    assert!(head.contains(&format!("Content-Length: {}\r\n", body.len())), "{head}");
    let f = parse(body).unwrap_or_else(|e| panic!("the exposition does not read: {e:?}\n{body}"));
    for v in label_values(&f) {
        assert!(!identifier_like(v), "a label value that names a party: {v}\n{body}");
    }
    f
}

fn at(f: &[Family], name: &str) -> f64 {
    value(f, name, &[]).unwrap_or_else(|| panic!("{name} is not in the exposition"))
}

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

async fn run() {
    let server = relay(|c| {
        c.control.listen = Some(loopback());
        c.metering.energy = joule_moq_relay::config::EnergyMode::Off;
    })
    .await;
    let addr = server.control_addr.expect("the loopback listener");

    // Before any session: every family, at 0.
    let f = scrape(addr).await;
    for name in ["wai_relay_sessions", "wai_relay_sessions_opened_total", "wai_relay_tracks", "wai_relay_subscriptions", "wai_relay_objects_in_total", "wai_relay_queued"] {
        assert_eq!(at(&f, name), 0.0, "{name}");
    }
    for name in [
        "wai_relay_fetches_served_total",
        "wai_relay_door_refused_total",
        "wai_relay_requests_refused_total",
        "wai_relay_upstream_total",
        "wai_relay_ratings_held",
        "wai_relay_going_away",
        "wai_relay_control_requests_total",
        "wai_relay_control_refused_total",
    ] {
        assert!(f.iter().any(|x| x.name == name), "{name} is not in the exposition");
    }
    assert_eq!(at(&f, "wai_relay_energy_unmetered"), 1.0, "no meter samples a relay with energy off");
    assert_eq!(value(&f, "wai_relay_control_requests_total", &[("op", "show")]), Some(0.0), "every op's series is written from the start");

    // A publisher of two tracks and a subscriber to both, over the WebSocket binding.
    let key = SigningKey::from_bytes(&[0x5a; 32]);
    let t = Tracks::new(ns(&["metrics", "room"]), "cam", 10, 8, 64);
    let (p, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let _published = publish_tracks(&p, &t, None).await;
    let (s, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let _sub = subscribe_both(&s, &t, Parameters::new()).await;
    for g in 0..3 {
        publish_group(&p, &key, &t, g, &payload("m", g, 64)).await;
    }
    let moved = until(STEP, || server.relay().stats().objects_in >= 6).await;
    assert!(moved, "the relay counted the objects published");
    let f = scrape(addr).await;
    assert_eq!(at(&f, "wai_relay_sessions"), 2.0);
    assert_eq!(at(&f, "wai_relay_sessions_opened_total"), 2.0);
    assert_eq!(at(&f, "wai_relay_tracks"), 2.0, "the media track and its receipts track");
    assert_eq!(at(&f, "wai_relay_subscriptions"), 2.0, "one subscriber to each track");
    assert!(at(&f, "wai_relay_objects_in_total") >= 6.0, "three groups of two tracks");

    // A JSON-lines request on the same listener is answered as before, and counted by its op.
    let c = tokio::net::TcpStream::connect(addr).await.unwrap();
    let (r, mut w) = c.into_split();
    w.write_all(b"{\"op\":\"show\"}\n").await.unwrap();
    let mut line = String::new();
    timeout(STEP, BufReader::new(r).read_line(&mut line)).await.expect("an answer").unwrap();
    let shown: Value = serde_json::from_str(&line).unwrap_or_else(|e| panic!("{e}: {line:?}"));
    assert_eq!(shown["ok"], true, "{shown}");
    assert!(shown["auth"].is_object(), "{shown}");
    drop(w);
    let f = scrape(addr).await;
    assert_eq!(value(&f, "wai_relay_control_requests_total", &[("op", "show")]), Some(1.0));
    assert_eq!(value(&f, "wai_relay_control_requests_total", &[("op", "metrics")]), Some(2.0), "the two scrapes before this one");

    // HEAD as GET, without the text; another method 405; another path, and through a front, 404.
    let said = http(addr, "HEAD /metrics HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n").await;
    let (head, body) = said.split_once("\r\n\r\n").unwrap_or_else(|| panic!("{said:?}"));
    assert!(head.starts_with("HTTP/1.1 200 OK\r\n") && head.contains(wai_metrics::CONTENT_TYPE), "{head}");
    assert!(body.is_empty(), "HEAD was answered with a body: {body:?}");
    let said = http(addr, "POST /metrics HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 0\r\n\r\n").await;
    assert!(said.starts_with("HTTP/1.1 405 ") && said.contains("\r\nAllow: GET, HEAD\r\n"), "{said}");
    let said = http(addr, "GET /grants HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n").await;
    assert!(said.starts_with("HTTP/1.1 404 "), "{said}");
    for front in wai_metrics::FRONT_FIELDS {
        let said = http(addr, &format!("GET /metrics HTTP/1.1\r\nHost: relay.example\r\n{front}: 192.0.2.7\r\n\r\n")).await;
        assert!(said.starts_with("HTTP/1.1 404 ") && !said.contains("# HELP"), "{front}: a request through a front was answered the counters: {said}");
    }
    let f = scrape(addr).await;
    assert_eq!(value(&f, "wai_relay_control_refused_total", &[("why", "http")]), Some(2.0 + wai_metrics::FRONT_FIELDS.len() as f64), "the 405, the 404 and each front");

    // Both leave: no session is open, and the opened count stands.
    p.close(0, "done");
    s.close(0, "done");
    let gone = until(STEP, || server.relay().stats().sessions_open == 0).await;
    assert!(gone, "the sessions ended");
    let f = scrape(addr).await;
    assert_eq!((at(&f, "wai_relay_sessions"), at(&f, "wai_relay_sessions_opened_total")), (0.0, 2.0));
    server.shutdown().await;
}