//! Integration tests for the stdio transport.
//!
//! These exercise the full reader / writer / shutdown loop end to end.
//! The "server" is a small POSIX shell script we write to a tempfile,
//! so the test does not depend on any binary outside the standard
//! toolchain. Windows is skipped because the harness uses `/bin/sh`.

#![cfg(unix)]

use std::os::unix::fs::PermissionsExt;
use std::path::PathBuf;
use std::time::Duration;

use mcptest_core::protocol::{JsonRpcId, JsonRpcMessage};
use mcptest_core::transport::{StdioConfig, StdioError, StdioTransport, STDIO_CHANNEL_CAPACITY};
use tempfile::TempDir;

/// Generous ceiling for waits that verify behavior, not speed: spawning a
/// child, receiving a line, or a graceful exit. These tests do not measure
/// performance, so the bound only needs to be high enough that a busy box
/// (a full `cargo test --workspace` run spawns many child processes in
/// parallel) never trips it, while still failing if something truly hangs.
/// A tight 2s/3s bound here false-fails under that contention even though
/// the same test passes in about 1.6s in isolation.
const GENEROUS_WAIT: Duration = Duration::from_secs(30);

/// Write the supplied script body to `dir/name.sh` and mark it
/// executable. Returns the full path so tests can hand it to
/// [`StdioConfig`].
fn write_script(dir: &TempDir, name: &str, body: &str) -> PathBuf {
    let path = dir.path().join(name);
    std::fs::write(&path, body).expect("write script");
    let mut perms = std::fs::metadata(&path).expect("meta").permissions();
    perms.set_mode(0o755);
    std::fs::set_permissions(&path, perms).expect("chmod");
    path
}

/// Build a shell-script echo server that frames JSON-RPC over stdio.
///
/// The script reads each line from stdin and writes it back to stdout
/// verbatim. It exits cleanly on EOF, which is how the transport asks
/// it to shut down.
fn write_echo_script(dir: &TempDir) -> PathBuf {
    let body = r#"#!/bin/sh
# Newline-delimited JSON-RPC echo server. Reads one line at a time and
# writes the same line back. Used by the mcptest stdio transport
# integration tests.
while IFS= read -r line; do
  printf '%s\n' "$line"
done
"#;
    write_script(dir, "echo.sh", body)
}

fn make_request(id: i64, method: &str) -> JsonRpcMessage {
    JsonRpcMessage {
        jsonrpc: "2.0".to_string(),
        id: Some(JsonRpcId::Number(id)),
        method: Some(method.into()),
        params: None,
        result: None,
        error: None,
    }
}

#[tokio::test]
async fn echoes_messages_in_order() {
    let tmp = TempDir::new().unwrap();
    let script = write_echo_script(&tmp);
    let cfg = StdioConfig {
        command: script.to_string_lossy().into(),
        args: vec![],
        env: Default::default(),
        working_dir: None,
        shutdown_timeout: GENEROUS_WAIT,
    };
    let (mut transport, handle) = StdioTransport::spawn(cfg).expect("spawn");

    // Send three requests.
    for n in 1..=3 {
        transport
            .outbound_tx
            .send(make_request(n, "ping"))
            .expect("send");
    }

    // Expect three responses, in order.
    for n in 1..=3 {
        let got = tokio::time::timeout(GENEROUS_WAIT, transport.inbound_rx.recv())
            .await
            .expect("timed out waiting for echo")
            .expect("channel closed");
        assert_eq!(got.id, Some(JsonRpcId::Number(n)));
        assert_eq!(got.method.as_deref(), Some("ping"));
    }

    drop(transport);
    handle.shutdown().await.expect("graceful shutdown");
}

#[tokio::test]
async fn graceful_shutdown_terminates_child() {
    let tmp = TempDir::new().unwrap();
    let script = write_echo_script(&tmp);
    let cfg = StdioConfig {
        command: script.to_string_lossy().into(),
        shutdown_timeout: GENEROUS_WAIT,
        ..Default::default()
    };
    let (transport, handle) = StdioTransport::spawn(cfg).expect("spawn");

    // Drop the outbound sender, which closes stdin and lets the script
    // exit on its own. The shutdown call must return promptly.
    drop(transport);
    let start = std::time::Instant::now();
    let outcome = handle.shutdown().await;
    assert!(
        outcome.is_ok(),
        "expected graceful shutdown, got {outcome:?}"
    );
    assert!(
        start.elapsed() < Duration::from_secs(15),
        "shutdown took too long: {:?}",
        start.elapsed()
    );
}

#[tokio::test]
async fn premature_exit_closes_inbound_channel() {
    let tmp = TempDir::new().unwrap();
    let path = write_script(
        &tmp,
        "exit_now.sh",
        r#"#!/bin/sh
# Server that exits immediately.
exit 0
"#,
    );

    let cfg = StdioConfig {
        command: path.to_string_lossy().into(),
        shutdown_timeout: GENEROUS_WAIT,
        ..Default::default()
    };
    let (mut transport, handle) = StdioTransport::spawn(cfg).expect("spawn");

    // The inbound channel should close as soon as the reader sees EOF.
    let msg = tokio::time::timeout(GENEROUS_WAIT, transport.inbound_rx.recv())
        .await
        .expect("timed out waiting for channel close");
    assert!(msg.is_none(), "premature exit should close inbound channel");

    drop(transport);
    handle
        .shutdown()
        .await
        .expect("shutdown after premature exit");
}

#[tokio::test]
async fn malformed_lines_are_skipped() {
    let tmp = TempDir::new().unwrap();
    // Emit a malformed line, then a valid envelope. The reader must
    // discard the bad line and surface the valid one.
    let path = write_script(
        &tmp,
        "mixed.sh",
        r#"#!/bin/sh
printf 'not json at all\n'
printf '{"jsonrpc":"2.0","id":42,"method":"after"}\n'
# Hold the pipe open so the test can shut down deterministically.
sleep 5
"#,
    );

    let cfg = StdioConfig {
        command: path.to_string_lossy().into(),
        shutdown_timeout: Duration::from_secs(1),
        ..Default::default()
    };
    let (mut transport, handle) = StdioTransport::spawn(cfg).expect("spawn");

    let got = tokio::time::timeout(GENEROUS_WAIT, transport.inbound_rx.recv())
        .await
        .expect("timed out")
        .expect("channel closed");
    assert_eq!(got.id, Some(JsonRpcId::Number(42)));
    assert_eq!(got.method.as_deref(), Some("after"));

    drop(transport);
    // The sleep will keep the child alive past `shutdown_timeout`, so
    // shutdown must escalate to a forced kill and report it.
    let outcome = handle.shutdown().await;
    assert!(
        matches!(outcome, Err(StdioError::ShutdownTimeout(_))),
        "expected ShutdownTimeout, got {outcome:?}"
    );
}

#[tokio::test]
async fn shutdown_timeout_escalates_to_forced_kill() {
    let tmp = TempDir::new().unwrap();
    // Script ignores stdin closing by sleeping. The transport must
    // forcibly kill it after the configured timeout.
    let path = write_script(
        &tmp,
        "stubborn.sh",
        r#"#!/bin/sh
sleep 30
"#,
    );

    let cfg = StdioConfig {
        command: path.to_string_lossy().into(),
        shutdown_timeout: Duration::from_millis(250),
        ..Default::default()
    };
    let (transport, handle) = StdioTransport::spawn(cfg).expect("spawn");
    drop(transport);

    let start = std::time::Instant::now();
    let outcome = handle.shutdown().await;
    let elapsed = start.elapsed();
    assert!(
        matches!(outcome, Err(StdioError::ShutdownTimeout(_))),
        "expected ShutdownTimeout, got {outcome:?}"
    );
    assert!(
        elapsed < Duration::from_secs(15),
        "forced kill should land quickly after the timeout, took {elapsed:?}"
    );
}

#[tokio::test]
async fn flooding_server_closes_inbound_channel_instead_of_buffering_unbounded() {
    // WOR-1313: a server that spews notifications faster than anyone
    // consumes them must fill the bounded inbound queue and get cut
    // off, not grow process memory without limit. The test deliberately
    // reads nothing until the reader has had time to declare the flood.
    let tmp = TempDir::new().unwrap();
    let path = write_script(
        &tmp,
        "flood.sh",
        r#"#!/bin/sh
# Notification firehose: no ids, no reads, just output forever.
while :; do
  printf '{"jsonrpc":"2.0","method":"notifications/flood"}\n'
done
"#,
    );

    let cfg = StdioConfig {
        command: path.to_string_lossy().into(),
        shutdown_timeout: Duration::from_millis(250),
        ..Default::default()
    };
    let (mut transport, handle) = StdioTransport::spawn(cfg).expect("spawn");

    // Sit past the reader's flood window (2s) without consuming, so the
    // queue stays full the whole time and the reader gives up.
    tokio::time::sleep(Duration::from_secs(4)).await;

    let mut received = 0usize;
    loop {
        match tokio::time::timeout(GENEROUS_WAIT, transport.inbound_rx.recv()).await {
            Ok(Some(_)) => received += 1,
            Ok(None) => break,
            Err(_) => panic!("inbound channel neither yielded nor closed"),
        }
    }
    assert!(
        received <= STDIO_CHANNEL_CAPACITY + 1,
        "flooded transport buffered {received} envelopes; the queue must stay bounded"
    );
    assert!(
        received >= STDIO_CHANNEL_CAPACITY / 2,
        "expected the queue to have filled before the cutoff, got {received}"
    );

    drop(transport);
    // The firehose ignores EOF, so shutdown escalates to a kill.
    let _ = handle.shutdown().await;
}

#[tokio::test]
async fn term_ignoring_child_is_force_killed_and_reaped() {
    // WOR-1314: a child that ignores polite signals is killed and then
    // actively reaped on a short poll loop. The distinguishable outcome
    // is ShutdownTimeout (killed and cleaned up), never ZombieChild
    // (SIGKILL is not ignorable) and never an indefinite hang. A true
    // unreapable zombie needs a kernel-wedged process, which cannot be
    // fabricated portably; the ZombieChild rendering is pinned by a
    // unit test next to the error enum instead.
    let tmp = TempDir::new().unwrap();
    let path = write_script(
        &tmp,
        "ignore_term.sh",
        r#"#!/bin/sh
trap '' TERM
sleep 30
"#,
    );

    let cfg = StdioConfig {
        command: path.to_string_lossy().into(),
        shutdown_timeout: Duration::from_millis(250),
        ..Default::default()
    };
    let (transport, handle) = StdioTransport::spawn(cfg).expect("spawn");
    drop(transport);

    let start = std::time::Instant::now();
    let outcome = handle.shutdown().await;
    let elapsed = start.elapsed();
    assert!(
        matches!(outcome, Err(StdioError::ShutdownTimeout(_))),
        "expected killed-and-reaped (ShutdownTimeout), got {outcome:?}"
    );
    assert!(
        elapsed < Duration::from_secs(15),
        "kill escalation and reap should land quickly, took {elapsed:?}"
    );
}

#[tokio::test]
async fn env_vars_reach_child() {
    let tmp = TempDir::new().unwrap();
    let path = write_script(
        &tmp,
        "env_echo.sh",
        r#"#!/bin/sh
printf '{"jsonrpc":"2.0","id":1,"method":"%s"}\n' "$MCPTEST_PROBE"
sleep 5
"#,
    );

    let mut env = std::collections::HashMap::new();
    env.insert("MCPTEST_PROBE".to_string(), "from-env".to_string());
    let cfg = StdioConfig {
        command: path.to_string_lossy().into(),
        env,
        shutdown_timeout: Duration::from_millis(250),
        ..Default::default()
    };
    let (mut transport, handle) = StdioTransport::spawn(cfg).expect("spawn");
    let got = tokio::time::timeout(GENEROUS_WAIT, transport.inbound_rx.recv())
        .await
        .expect("timed out")
        .expect("channel closed");
    assert_eq!(got.method.as_deref(), Some("from-env"));
    drop(transport);
    // We expect a forced kill because of the sleep, but that is not
    // what this test asserts; we just need to release the child.
    let _ = handle.shutdown().await;
}

#[tokio::test]
async fn shutdown_closes_stdin_so_eof_respecting_child_exits_without_kill() {
    let tmp = TempDir::new().unwrap();
    // A well-behaved server: read stdin until EOF, then exit. `read`
    // returns non-zero on EOF, so the loop ends and the script exits.
    let path = write_script(
        &tmp,
        "eof_respecting.sh",
        r#"#!/bin/sh
while IFS= read -r _line; do :; done
"#,
    );

    let cfg = StdioConfig {
        command: path.to_string_lossy().into(),
        shutdown_timeout: Duration::from_secs(5),
        ..Default::default()
    };
    let (_transport, handle) = StdioTransport::spawn(cfg).expect("spawn");

    // Deliberately keep `_transport` (and its outbound Sender) alive across
    // the shutdown call. shutdown() must close the child's stdin itself
    // rather than relying on the caller to drop the channel, so an
    // EOF-respecting child exits well before the timeout.
    let start = std::time::Instant::now();
    let outcome = handle.shutdown().await;
    let elapsed = start.elapsed();

    assert!(outcome.is_ok(), "expected clean shutdown, got {outcome:?}");
    assert!(
        elapsed < Duration::from_secs(2),
        "EOF-respecting child should exit before the 5s timeout, took {elapsed:?}"
    );
}
