//! `--wait-for-ready` run-path wiring.
//!
//! The polling loop and backoff live transport-free in
//! [`mcptest_core::readiness`]. This module supplies the concrete probe the
//! run path uses (a TCP connect against each URL server) and orchestrates the
//! wait across every declared HTTP server before the run connects. Stdio
//! servers are spawned by mcptest itself, so there is nothing to wait for.

use std::net::{TcpStream, ToSocketAddrs};
use std::time::{Duration, Instant};

use mcptest_core::readiness::{wait_for_ready, ReadinessFailFast, ReadinessOutcome, ReadyProbe};
use mcptest_core::suite::{ResolvedSuite, ServerSpec};

use crate::cli::exit_codes::{EXIT_INVALID_INVOCATION, EXIT_NOT_READY};

/// Per-attempt TCP connect timeout. Short so a refused connection retries
/// quickly while the server is still coming up.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(2);

/// Probe that an HTTP(S) MCP server is accepting TCP connections.
///
/// A successful connect means the listener is up, which is the signal the
/// `--wait-for-ready` budget is waiting for. We deliberately stop at the TCP
/// layer: a 401 or protocol error on a server that is already listening is the
/// run's problem to surface (and it now fails fast), not the readiness loop's.
struct TcpReadyProbe {
    /// `host:port` used for both resolution and display.
    authority: String,
}

impl ReadyProbe for TcpReadyProbe {
    fn probe(&self) -> Result<bool, ReadinessFailFast> {
        // Resolve fresh each attempt: a server still coming up may not have a
        // DNS record or a bound port yet, so an unresolved name is a transient
        // "not ready" rather than a hard failure.
        let mut addrs = match self.authority.to_socket_addrs() {
            Ok(addrs) => addrs,
            Err(_) => return Ok(false),
        };
        let Some(addr) = addrs.next() else {
            return Ok(false);
        };
        match TcpStream::connect_timeout(&addr, CONNECT_TIMEOUT) {
            Ok(_) => Ok(true),
            Err(_) => Ok(false),
        }
    }
}

/// The URL a server listens on, or `None` for a stdio server.
fn url_of(spec: &ServerSpec) -> Option<&str> {
    match spec {
        ServerSpec::StreamableHttp { url, .. } | ServerSpec::LegacySse { url, .. } => Some(url),
        _ => None,
    }
}

/// Derive `host:port` from a URL, defaulting the port from the scheme.
fn authority_of(url: &str) -> Option<String> {
    let parsed = url::Url::parse(url).ok()?;
    let host = parsed.host_str()?;
    let port = parsed.port_or_known_default()?;
    Some(format!("{host}:{port}"))
}

/// Poll one `host:port` until it accepts connections or the budget expires.
///
/// Returns `None` once it is ready; `Some(EXIT_NOT_READY)` on timeout,
/// `Some(EXIT_INVALID_INVOCATION)` on a non-retryable probe error,
/// matching the readiness exit-code contract.
fn wait_one(name: &str, authority: &str, budget: Duration, quiet: bool) -> Option<i32> {
    if !quiet {
        println!(
            "wait-for-ready: waiting up to {}s for `{name}` ({authority})",
            budget.as_secs()
        );
    }
    let probe = TcpReadyProbe {
        authority: authority.to_string(),
    };
    let outcome = wait_for_ready(&probe, budget, |elapsed| {
        if !quiet {
            println!("  still waiting for `{name}`... {}s", elapsed.as_secs());
        }
    });
    match outcome {
        ReadinessOutcome::Ready => {
            if !quiet {
                println!("wait-for-ready: `{name}` is accepting connections");
            }
            None
        }
        ReadinessOutcome::Timeout => {
            eprintln!("wait-for-ready: `{name}` ({authority}) not ready within the budget");
            Some(EXIT_NOT_READY)
        }
        ReadinessOutcome::FailFast => {
            eprintln!("wait-for-ready: `{name}` ({authority}) reported a non-retryable error");
            Some(EXIT_INVALID_INVOCATION)
        }
    }
}

/// Wait for every URL server in the suite to accept connections, sharing one
/// budget across them.
///
/// Returns `None` when every server is ready (or there is nothing to probe) so
/// the run proceeds. Returns `Some(exit)` to abort with the readiness exit code.
pub(crate) fn await_servers_ready(
    suite: &ResolvedSuite,
    budget: Duration,
    quiet: bool,
) -> Option<i32> {
    let targets: Vec<(String, String)> = suite
        .servers
        .iter()
        .filter_map(|(name, spec)| {
            let url = url_of(spec)?;
            authority_of(url).map(|authority| (name.clone(), authority))
        })
        .collect();

    if targets.is_empty() {
        if !quiet {
            println!("wait-for-ready: no URL servers to probe; continuing");
        }
        return None;
    }

    let deadline = Instant::now() + budget;
    for (name, authority) in targets {
        let remaining = deadline.saturating_duration_since(Instant::now());
        if remaining.is_zero() {
            eprintln!("wait-for-ready: budget exhausted before `{name}` ({authority}) was ready");
            return Some(EXIT_NOT_READY);
        }
        if let Some(exit) = wait_one(&name, &authority, remaining, quiet) {
            return Some(exit);
        }
    }
    None
}

/// Wait for a single URL to accept connections, used by `doctor --url`.
///
/// Returns `None` when ready or when the URL cannot be parsed (the caller's own
/// probe then surfaces the malformed target); `Some(exit)` on timeout or error.
pub(crate) fn await_url_ready(url: &str, budget: Duration, quiet: bool) -> Option<i32> {
    let authority = authority_of(url)?;
    wait_one("server", &authority, budget, quiet)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn authority_defaults_port_from_scheme() {
        assert_eq!(
            authority_of("https://mcp.example.com/mcp").as_deref(),
            Some("mcp.example.com:443")
        );
        assert_eq!(
            authority_of("http://localhost/mcp").as_deref(),
            Some("localhost:80")
        );
        assert_eq!(
            authority_of("http://127.0.0.1:8931/mcp").as_deref(),
            Some("127.0.0.1:8931")
        );
    }

    #[test]
    fn authority_of_rejects_non_url() {
        assert_eq!(authority_of("not a url"), None);
    }

    #[test]
    fn url_of_skips_stdio_servers() {
        let stdio = ServerSpec::Stdio {
            command: "mcptest".into(),
            args: vec![],
            env: Default::default(),
        };
        assert_eq!(url_of(&stdio), None);
        let http = ServerSpec::StreamableHttp {
            url: "https://x.example/mcp".into(),
            headers: Default::default(),
            oauth: None,
            stateless: false,
        };
        assert_eq!(url_of(&http), Some("https://x.example/mcp"));
    }

    #[test]
    fn probe_reports_false_for_an_unreachable_address() {
        // 192.0.2.0/24 is TEST-NET-1 (RFC 5737): reserved for documentation and
        // guaranteed not to route, so a connect always fails. Deterministic,
        // unlike a freed loopback port that a parallel test could grab.
        let probe = TcpReadyProbe {
            authority: "192.0.2.1:80".to_string(),
        };
        assert!(
            !probe.probe().unwrap(),
            "an unreachable address must not be ready"
        );
    }

    #[test]
    fn probe_reports_true_for_a_live_listener() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        let probe = TcpReadyProbe {
            authority: format!("127.0.0.1:{port}"),
        };
        assert!(probe.probe().unwrap(), "live listener should be ready");
    }
}
