//! End-to-end tests: spawn the real `echoircd` binary on ephemeral ports and drive it
//! as a client would. Covers the reactor pool (cross-worker delivery), TLS in the
//! reactor (handshake, cross-transport, the secure marker, the stalled-handshake reap),
//! and core routing (PRIVMSG, nick collision).
//!
//! Each test owns its own server on its own ports and kills the child in `Drop` — by
//! PID, never by process name — so the suite can't touch anything it didn't start.

use std::io::{self, Read, Write};
use std::net::TcpStream;
use std::path::PathBuf;
use std::process::{Child, Command, Stdio};
use std::thread;
use std::time::{Duration, Instant};

use openssl::ssl::{SslConnector, SslMethod, SslStream, SslVerifyMode};

/// Grab a free localhost port by binding :0 and releasing it.
fn free_port() -> u16 {
    std::net::TcpListener::bind("127.0.0.1:0")
        .unwrap()
        .local_addr()
        .unwrap()
        .port()
}

/// A throwaway self-signed cert/key (PEM) for the TLS listener.
fn gen_cert() -> (String, String) {
    use openssl::asn1::Asn1Time;
    use openssl::bn::{BigNum, MsbOption};
    use openssl::hash::MessageDigest;
    use openssl::pkey::PKey;
    use openssl::rsa::Rsa;
    use openssl::x509::{X509NameBuilder, X509};

    let key = PKey::from_rsa(Rsa::generate(2048).unwrap()).unwrap();
    let mut name = X509NameBuilder::new().unwrap();
    name.append_entry_by_text("CN", "echo.test").unwrap();
    let name = name.build();
    let mut b = X509::builder().unwrap();
    b.set_version(2).unwrap();
    let serial = {
        let mut bn = BigNum::new().unwrap();
        bn.rand(128, MsbOption::MAYBE_ZERO, false).unwrap();
        bn.to_asn1_integer().unwrap()
    };
    b.set_serial_number(&serial).unwrap();
    b.set_subject_name(&name).unwrap();
    b.set_issuer_name(&name).unwrap();
    b.set_pubkey(&key).unwrap();
    b.set_not_before(&Asn1Time::days_from_now(0).unwrap()).unwrap();
    b.set_not_after(&Asn1Time::days_from_now(1).unwrap()).unwrap();
    b.sign(&key, MessageDigest::sha256()).unwrap();
    let cert = b.build();
    (
        String::from_utf8(cert.to_pem().unwrap()).unwrap(),
        String::from_utf8(key.private_key_to_pem_pkcs8().unwrap()).unwrap(),
    )
}

/// A running echoircd child plus the ports it listens on. Killed + cleaned on drop.
struct Server {
    child: Child,
    plain: u16,
    tls: u16,
    dir: PathBuf,
}

impl Drop for Server {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
        let _ = std::fs::remove_dir_all(&self.dir);
    }
}

impl Server {
    fn start(io_threads: usize, tls: bool, hs_timeout: u32) -> Server {
        Server::start_full(io_threads, tls, hs_timeout, 0)
    }

    fn start_full(io_threads: usize, tls: bool, hs_timeout: u32, accept_rate: usize) -> Server {
        let (plain, tlsp, s2s) = (free_port(), free_port(), free_port());
        let dir = std::env::temp_dir().join(format!("echoircd-it-{}-{plain}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let mut conf = format!(
            "servername = it.test\nnetwork = itNet\nbind = 127.0.0.1:{plain}\n\
             bind_server = 127.0.0.1:{s2s}\nsid = 1AA\nmotd = hi\nio_threads = {io_threads}\n"
        );
        if accept_rate > 0 {
            conf.push_str(&format!("accept_rate = {accept_rate}\naccept_burst = {accept_rate}\n"));
        }
        if tls {
            let (cert, key) = gen_cert();
            let (cp, kp) = (dir.join("cert.pem"), dir.join("key.pem"));
            std::fs::write(&cp, cert).unwrap();
            std::fs::write(&kp, key).unwrap();
            conf.push_str(&format!(
                "bind_tls = 127.0.0.1:{tlsp}\ntls_cert = {}\ntls_key = {}\n\
                 tls_handshake_timeout = {hs_timeout}\n",
                cp.display(),
                kp.display()
            ));
        }
        let cpath = dir.join("echoircd.conf");
        std::fs::write(&cpath, conf).unwrap();
        let child = Command::new(env!("CARGO_BIN_EXE_echoircd"))
            .arg(&cpath)
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .expect("spawn echoircd");
        let srv = Server { child, plain, tls: tlsp, dir };
        srv.wait_ready();
        srv
    }

    /// Block until a full NICK/USER registration succeeds, so the whole pipeline
    /// (acceptor + reactor + core) is proven up before the test proceeds.
    fn wait_ready(&self) {
        let deadline = Instant::now() + Duration::from_secs(20);
        while Instant::now() < deadline {
            if let Ok(mut s) = TcpStream::connect(("127.0.0.1", self.plain)) {
                s.set_read_timeout(Some(Duration::from_millis(400))).ok();
                let _ = s.write_all(b"NICK probe\r\nUSER probe 0 * :probe\r\n");
                if read_until(&mut s, " 001 ", Duration::from_secs(2)) {
                    let _ = s.write_all(b"QUIT :ready\r\n");
                    drop(s);
                    thread::sleep(Duration::from_millis(200));
                    return;
                }
            }
            thread::sleep(Duration::from_millis(100));
        }
        panic!("server never became ready");
    }

    fn plain_client(&self, nick: &str) -> TcpStream {
        let mut s = TcpStream::connect(("127.0.0.1", self.plain)).unwrap();
        s.set_read_timeout(Some(Duration::from_millis(400))).unwrap();
        register(&mut s, nick);
        s
    }

    fn tls_client(&self, nick: &str) -> SslStream<TcpStream> {
        let tcp = TcpStream::connect(("127.0.0.1", self.tls)).unwrap();
        let mut b = SslConnector::builder(SslMethod::tls()).unwrap();
        b.set_verify(SslVerifyMode::NONE);
        let mut s = b
            .build()
            .configure()
            .unwrap()
            .verify_hostname(false)
            .connect("echo.test", tcp)
            .unwrap();
        s.get_ref()
            .set_read_timeout(Some(Duration::from_millis(400)))
            .unwrap();
        register(&mut s, nick);
        s
    }
}

/// Read until `needle` appears or `timeout` elapses; returns whether it was seen.
/// Relies on the stream having a read timeout so reads don't block forever.
fn read_until<S: Read>(s: &mut S, needle: &str, timeout: Duration) -> bool {
    let deadline = Instant::now() + timeout;
    let mut buf = Vec::new();
    let mut chunk = [0u8; 8192];
    while Instant::now() < deadline {
        match s.read(&mut chunk) {
            Ok(0) => break,
            Ok(n) => {
                buf.extend_from_slice(&chunk[..n]);
                if String::from_utf8_lossy(&buf).contains(needle) {
                    return true;
                }
            }
            Err(ref e)
                if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut => {}
            Err(_) => break,
        }
    }
    String::from_utf8_lossy(&buf).contains(needle)
}

/// Accumulate everything readable within `timeout` into one string, so a test can
/// assert on several lines that arrived in a single batch (read_until discards its
/// buffer per call, which loses lines sent back-to-back).
fn read_collect<S: Read>(s: &mut S, timeout: Duration) -> String {
    let deadline = Instant::now() + timeout;
    let mut buf = Vec::new();
    let mut chunk = [0u8; 8192];
    while Instant::now() < deadline {
        match s.read(&mut chunk) {
            Ok(0) => break,
            Ok(n) => buf.extend_from_slice(&chunk[..n]),
            Err(ref e)
                if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut => {}
            Err(_) => break,
        }
    }
    String::from_utf8_lossy(&buf).into_owned()
}

fn register<S: Read + Write>(s: &mut S, nick: &str) {
    s.write_all(format!("NICK {nick}\r\nUSER {nick} 0 * :{nick}\r\n").as_bytes())
        .unwrap();
    // Generous: read_until returns the instant 001 arrives, so a big deadline only
    // buys patience on a box saturated by the parallel test servers, never latency.
    assert!(
        read_until(s, " 001 ", Duration::from_secs(15)),
        "no 001 welcome for {nick}"
    );
}

fn line<S: Write>(s: &mut S, l: &str) {
    s.write_all(format!("{l}\r\n").as_bytes()).unwrap();
}

#[test]
fn plaintext_registration_privmsg_and_collision() {
    let srv = Server::start(2, false, 0);
    let mut a = srv.plain_client("alice");
    let mut b = srv.plain_client("bob");
    line(&mut a, "JOIN #x");
    line(&mut b, "JOIN #x");
    read_until(&mut a, "JOIN", Duration::from_secs(2));
    read_until(&mut b, "JOIN", Duration::from_secs(2));
    line(&mut a, "PRIVMSG #x :hello-bob");
    assert!(
        read_until(&mut b, "hello-bob", Duration::from_secs(3)),
        "bob never received alice's channel message"
    );

    // a third client can't steal alice's nick
    let mut c = TcpStream::connect(("127.0.0.1", srv.plain)).unwrap();
    c.set_read_timeout(Some(Duration::from_millis(400))).unwrap();
    line(&mut c, "NICK alice");
    assert!(
        read_until(&mut c, " 433 ", Duration::from_secs(3)),
        "expected 433 nick-in-use"
    );
}

#[test]
fn reactor_pool_cross_worker_delivery() {
    // 3 workers: the clients land on different reactors, yet a channel message from one
    // must reach every other — proving the single core routes across workers.
    let srv = Server::start(3, false, 0);
    let mut clients: Vec<TcpStream> = (0..6).map(|i| srv.plain_client(&format!("u{i}"))).collect();
    for c in clients.iter_mut() {
        line(c, "JOIN #pool");
    }
    for c in clients.iter_mut() {
        read_until(c, "JOIN", Duration::from_secs(2));
    }
    line(&mut clients[0], "PRIVMSG #pool :ping-all");
    for (i, c) in clients.iter_mut().enumerate().skip(1) {
        assert!(
            read_until(c, "ping-all", Duration::from_secs(3)),
            "client u{i} (a different reactor worker) missed the message"
        );
    }
}

#[test]
fn tls_in_reactor_handshake_and_cross_transport() {
    let srv = Server::start(2, true, 15);
    let mut t = srv.tls_client("secure1"); // full handshake happens inside a worker
    let mut p = srv.plain_client("plain1");
    line(&mut t, "JOIN #z");
    line(&mut p, "JOIN #z");
    read_until(&mut t, "JOIN", Duration::from_secs(2));
    read_until(&mut p, "JOIN", Duration::from_secs(2));

    line(&mut t, "PRIVMSG #z :from-tls");
    assert!(
        read_until(&mut p, "from-tls", Duration::from_secs(3)),
        "TLS -> plaintext delivery failed"
    );
    line(&mut p, "PRIVMSG #z :from-plain");
    assert!(
        read_until(&mut t, "from-plain", Duration::from_secs(3)),
        "plaintext -> TLS delivery failed"
    );

    // the TLS user is flagged secure (671 in WHOIS)
    line(&mut p, "WHOIS secure1");
    assert!(
        read_until(&mut p, " 671 ", Duration::from_secs(3)),
        "TLS user not reported as using a secure connection"
    );
}

/// Register `nick` having negotiated capability `cap` (IRCv3 CAP LS/REQ/END).
fn register_with_cap<S: Read + Write>(s: &mut S, nick: &str, cap: &str) {
    s.write_all(
        format!("CAP LS 302\r\nNICK {nick}\r\nUSER {nick} 0 * :{nick}\r\nCAP REQ :{cap}\r\nCAP END\r\n")
            .as_bytes(),
    )
    .unwrap();
    assert!(
        read_until(s, "ACK", Duration::from_secs(10)),
        "no CAP ACK for {cap}"
    );
    assert!(
        read_until(s, " 001 ", Duration::from_secs(15)),
        "no 001 welcome for {nick}"
    );
}

#[test]
fn channel_rename_notifies_by_cap_and_needs_ops() {
    let srv = Server::start(2, false, 0);
    // alice negotiates draft/channel-rename; bob does not.
    let mut alice = TcpStream::connect(("127.0.0.1", srv.plain)).unwrap();
    alice.set_read_timeout(Some(Duration::from_millis(400))).unwrap();
    register_with_cap(&mut alice, "alice", "draft/channel-rename");
    let mut bob = srv.plain_client("bob");

    // Generous timeouts: the full integration suite runs many echoircd processes in
    // parallel, so a saturated CI box can slip past tight socket deadlines.
    let t = Duration::from_secs(10);
    // Serialize the joins: alice must create #old (and become op) before bob joins,
    // or a reactor-scheduling race could make bob the creator instead.
    line(&mut alice, "JOIN #old"); // alice creates -> op
    assert!(read_until(&mut alice, "JOIN #old", t), "alice join");
    line(&mut bob, "JOIN #old"); // joins the existing channel -> non-op
    assert!(read_until(&mut bob, "JOIN #old", t), "bob join");

    // A non-op can't rename.
    line(&mut bob, "RENAME #old #nope");
    assert!(
        read_until(&mut bob, " 482 ", t),
        "non-op RENAME should get 482 CHANOPRIVSNEEDED"
    );

    // The op renames; alice (cap) gets a RENAME line, bob (no cap) is walked PART -> JOIN.
    line(&mut alice, "RENAME #old #new :moving");
    assert!(
        read_until(&mut alice, "RENAME #old #new", t),
        "cap client did not receive RENAME"
    );
    // bob's PART and JOIN arrive in one batch — collect and check both.
    let bobseen = read_collect(&mut bob, Duration::from_secs(4));
    assert!(bobseen.contains("PART #old"), "plain client not PARTed: {bobseen:?}");
    assert!(bobseen.contains("JOIN #new"), "plain client not re-JOINed: {bobseen:?}");
    assert!(!bobseen.contains("RENAME"), "plain client should not see RENAME: {bobseen:?}");

    // The channel now answers under the new name (and not the old).
    line(&mut alice, "PRIVMSG #new :landed");
    assert!(
        read_until(&mut bob, "landed", t),
        "message to the renamed channel didn't reach members"
    );
    line(&mut alice, "NAMES #old");
    assert!(
        read_until(&mut alice, " 366 ", t),
        "NAMES on the old name should just end (channel is gone)"
    );
}

#[test]
fn accept_rate_limit_drops_connection_churn() {
    // rate/burst = 5: a rapid burst of 20 connections from one IP must be partly dropped
    // at the accept edge — some register, but not all 20.
    let srv = Server::start_full(2, false, 5, 5);
    let mut socks = Vec::new();
    for i in 0..20 {
        if let Ok(mut s) = TcpStream::connect(("127.0.0.1", srv.plain)) {
            s.set_read_timeout(Some(Duration::from_millis(600))).unwrap();
            let _ = s.write_all(format!("NICK n{i}\r\nUSER n{i} 0 * :n\r\n").as_bytes());
            socks.push(s);
        }
    }
    let mut registered = 0;
    for s in socks.iter_mut() {
        if read_until(s, " 001 ", Duration::from_millis(800)) {
            registered += 1;
        }
    }
    assert!(
        registered < 20,
        "rate limit didn't drop any of a 20-connection burst ({registered} registered)"
    );
    assert!(
        registered >= 3,
        "rate limit dropped too much — burst of 5 should let at least a few through ({registered})"
    );
}

#[test]
fn tls_stalled_handshake_is_reaped() {
    // 2s handshake timeout: a raw TCP connection to the TLS port that never negotiates
    // must be closed by the server, not leaked.
    let srv = Server::start(2, true, 2);
    let mut raw = TcpStream::connect(("127.0.0.1", srv.tls)).unwrap();
    raw.set_read_timeout(Some(Duration::from_secs(6))).unwrap();
    let start = Instant::now();
    let mut buf = [0u8; 16];
    let closed = matches!(raw.read(&mut buf), Ok(0)); // server closed → EOF
    let elapsed = start.elapsed();
    assert!(closed, "stalled TLS handshake was not closed by the server");
    assert!(
        elapsed >= Duration::from_millis(1500) && elapsed < Duration::from_secs(5),
        "reap timing off: {elapsed:?} (expected ~2s)"
    );
}