//! The roster host held against the Python oracle's measured behaviour.
//!
//! Response bytes are only half of it. The redirector cost two live gate
//! attempts to a divergence no response comparison could see — a request body
//! left unread, which turns the close into an RST — so the oracle's lifecycle
//! was measured with a probe and every property is asserted here:
//!
//! ```text
//!   dwell after responding   0 ms
//!   request body             drained before the response is written
//!   close                    immediately after, cleanly
//!   keep-alive               none: one request per connection
//! ```
//!
//! Where a property is observable both from the client and from inside the
//! host, it is asserted **inside the host** via `ConnOutcome`. That distinction
//! is not pedantry: the first attempt to test body-draining asserted that the
//! client still received its response, and that test passed against a host that
//! did not drain at all — the client reads the buffered response and sees
//! `close_notify` before any reset.

use std::io::{Read, Write};
use std::net::TcpStream;
use std::time::{Duration, Instant};

use openfut_adapter_fifa17::roster;
use openfut_roster_host::{bind, BodyRead, RosterConfig};

const PATH: &str = "/fifa17/fut/rosterupdate.xml";
const BODY: &[u8] = b"probe=1&pad=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";

fn cert_pair() -> (String, String) {
    let base = concat!(env!("CARGO_MANIFEST_DIR"), "/../fifa17-recon/tools");
    (
        format!("{base}/redir_cert.pem"),
        format!("{base}/redir_key.pem"),
    )
}

struct Harness {
    addr: std::net::SocketAddr,
    outcomes: openfut_roster_host::Outcomes,
}

fn start() -> Harness {
    start_with(|_| {})
}

fn start_with(tweak: impl FnOnce(&mut RosterConfig)) -> Harness {
    let (c, k) = cert_pair();
    let mut cfg = RosterConfig::for_test(&c, &k);
    tweak(&mut cfg);
    let server = bind(cfg).expect("bind");
    let addr = server.local_addr;
    let outcomes = server.outcomes();
    std::thread::spawn(move || {
        let _ = server.run();
    });
    Harness { addr, outcomes }
}

/// A client that behaves like FIFA: legacy suites, no certificate checking.
fn connect(addr: std::net::SocketAddr) -> openfut_tls::SslStream<TcpStream> {
    use openfut_tls::{SslConnector, SslMethod, SslVerifyMode};
    let mut b = SslConnector::builder(SslMethod::tls()).expect("connector");
    b.set_cipher_list(openfut_adapter_fifa17::tls::OBSERVED_CLIENT_SUITES)
        .expect("ciphers");
    b.set_verify(SslVerifyMode::NONE);
    let sock = TcpStream::connect(addr).expect("connect");
    sock.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    let ssl = b
        .build()
        .configure()
        .and_then(|c| c.verify_hostname(false).into_ssl("roster-test"))
        .expect("ssl");
    let mut s = openfut_tls::SslStream::new(ssl, sock).expect("stream");
    s.connect().expect("handshake");
    s
}

fn request(method: &str, body: Option<&[u8]>) -> Vec<u8> {
    let mut r = format!("{method} {PATH} HTTP/1.1\r\nHost: roster-test\r\nAccept: */*\r\n");
    if let Some(b) = body {
        r.push_str(&format!("Content-Length: {}\r\n", b.len()));
    }
    r.push_str("\r\n");
    let mut out = r.into_bytes();
    if let Some(b) = body {
        out.extend_from_slice(b);
    }
    out
}

fn read_to_eof(s: &mut openfut_tls::SslStream<TcpStream>) -> Vec<u8> {
    let mut out = Vec::new();
    let mut chunk = [0u8; 4096];
    loop {
        match s.read(&mut chunk) {
            Ok(0) | Err(_) => break,
            Ok(n) => out.extend_from_slice(&chunk[..n]),
        }
    }
    out
}

fn exchange(addr: std::net::SocketAddr, method: &str, body: Option<&[u8]>) -> Vec<u8> {
    let mut s = connect(addr);
    s.write_all(&request(method, body)).expect("write");
    s.flush().ok();
    read_to_eof(&mut s)
}

fn mask(raw: &[u8]) -> String {
    String::from_utf8_lossy(raw)
        .split("\r\n")
        .map(|l| {
            if l.starts_with("Date: ") {
                "Date: <MASKED>".to_string()
            } else if l.starts_with("Server: ") {
                "Server: <MASKED>".to_string()
            } else {
                l.to_string()
            }
        })
        .collect::<Vec<_>>()
        .join("\r\n")
}

/// The bytes the adapter says the oracle sends, masked the same way.
fn expected(method: roster::Method) -> String {
    mask(&roster::roster_response(
        method,
        roster::ORACLE_SERVER,
        "IRRELEVANT",
    ))
}

#[test]
fn get_returns_the_oracles_bytes() {
    let h = start();
    let got = exchange(h.addr, "GET", None);
    assert_eq!(mask(&got), expected(roster::Method::Get));
    assert!(String::from_utf8_lossy(&got).starts_with("HTTP/1.0 200 OK\r\n"));
}

#[test]
fn head_and_post_are_headers_only_but_still_advertise_the_body() {
    let h = start();
    for (m, adapter_m) in [
        ("HEAD", roster::Method::Head),
        ("POST", roster::Method::Post),
    ] {
        let body = if m == "POST" { Some(BODY) } else { None };
        let got = exchange(h.addr, m, body);
        assert_eq!(mask(&got), expected(adapter_m), "{m} differs");
        let text = String::from_utf8_lossy(&got);
        assert!(text.contains("Content-Length: 67"), "{m}: {text}");
        assert!(text.ends_with("\r\n\r\n"), "{m} must send no body");
    }
}

/// Asserted on the HOST's record, not on what the client received.
///
/// A client-side assertion cannot distinguish "drained" from "not drained":
/// it reads the buffered response either way. This is the exact mistake that
/// let a mutation survive on the redirector.
#[test]
fn the_request_body_is_drained_before_the_response_is_written() {
    let h = start();
    let _ = exchange(h.addr, "POST", Some(BODY));
    let rec = h.outcomes.lock().unwrap();
    let last = rec.last().expect("a connection was recorded");
    assert_eq!(
        last.body,
        BodyRead::Complete(BODY.len()),
        "the whole body must be read before answering, or the close becomes an RST"
    );
    assert!(last.response_bytes > 0);
}

/// The oracle answers a POST only after the body arrives. Sending the head,
/// pausing, then sending the body proves the host waits rather than replying
/// early and leaving the socket dirty.
#[test]
fn a_late_body_is_waited_for_rather_than_answered_early() {
    let h = start();
    let mut s = connect(h.addr);
    let head = format!(
        "POST {PATH} HTTP/1.1\r\nHost: roster-test\r\nContent-Length: {}\r\n\r\n",
        BODY.len()
    );
    s.write_all(head.as_bytes()).unwrap();
    s.flush().ok();

    s.get_ref()
        .set_read_timeout(Some(Duration::from_millis(600)))
        .unwrap();
    let mut early = [0u8; 64];
    let early_n = s.read(&mut early).unwrap_or(0);
    assert_eq!(
        early_n,
        0,
        "answered before the body arrived: {:?}",
        String::from_utf8_lossy(&early[..early_n])
    );

    s.get_ref()
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    s.write_all(BODY).unwrap();
    s.flush().ok();
    let late = read_to_eof(&mut s);
    assert!(!late.is_empty(), "no response after the body arrived");
    assert_eq!(mask(&late), expected(roster::Method::Post));
}

/// One request per connection: the response says `Connection: close` and the
/// oracle closes. Measured, not assumed — a keep-alive loop would be a
/// behaviour change invisible in any single response.
#[test]
fn the_connection_closes_after_one_request() {
    let h = start();
    let mut s = connect(h.addr);
    s.write_all(&request("GET", None)).unwrap();
    s.flush().ok();
    let first = read_to_eof(&mut s);
    assert!(!first.is_empty());
    // read_to_eof already ran to EOF; a second request must not be answered.
    let _ = s.write_all(&request("GET", None));
    let mut buf = [0u8; 32];
    let n = s.read(&mut buf).unwrap_or(0);
    assert_eq!(n, 0, "a second request was answered on the same connection");
}

/// The oracle has no post-response sleep — unlike the redirector's 300ms.
/// Inheriting the wrong dwell would add delay to a poll that runs every few
/// seconds.
#[test]
fn the_connection_is_not_held_open_after_responding() {
    let h = start();
    let mut s = connect(h.addr);
    s.write_all(&request("GET", None)).unwrap();
    s.flush().ok();
    let start = Instant::now();
    let got = read_to_eof(&mut s);
    let elapsed = start.elapsed();
    assert!(!got.is_empty());
    assert!(
        elapsed < Duration::from_millis(150),
        "closed after {elapsed:?}; the roster oracle closes immediately"
    );
}

/// The dwell stays configurable so the causal experiment — set it and confirm
/// the delay appears — can run without a rebuild. Guards against the knob
/// being quietly ignored.
#[test]
fn the_dwell_is_overridable() {
    let h = start_with(|c| c.close_dwell = Duration::from_millis(400));
    let mut s = connect(h.addr);
    s.write_all(&request("GET", None)).unwrap();
    s.flush().ok();
    let start = Instant::now();
    let _ = read_to_eof(&mut s);
    assert!(
        start.elapsed() >= Duration::from_millis(350),
        "dwell override had no effect"
    );
}

/// The oracle never routes on the path — it logs it and answers the same way.
/// A host that started 404ing unknown paths would be a behaviour change.
#[test]
fn any_path_gets_the_same_answer() {
    let h = start();
    let mut s = connect(h.addr);
    s.write_all(b"GET /something/else HTTP/1.1\r\nHost: x\r\n\r\n")
        .unwrap();
    s.flush().ok();
    let got = read_to_eof(&mut s);
    assert_eq!(mask(&got), expected(roster::Method::Get));
}

/// Masking must not be able to hide a real difference.
#[test]
fn masking_does_not_hide_a_missing_header() {
    let good = roster::roster_response(roster::Method::Get, roster::ORACLE_SERVER, "X");
    let without: Vec<u8> = String::from_utf8_lossy(&good)
        .split("\r\n")
        .filter(|l| !l.starts_with("Date: "))
        .collect::<Vec<_>>()
        .join("\r\n")
        .into_bytes();
    assert_ne!(mask(&good), mask(&without));
}