//! Listener URL synthesis and reachability waiting, shared by every SDK.

use std::time::{Duration, Instant};

use tokio::io::AsyncReadExt;
use tokio::net::TcpStream;

use crate::config::{Config, IngressScheme};
use crate::error::{SailError, TransportKind};
use crate::worker::Listener;
use crate::Client;

/// How long a TCP probe waits, after connecting, for the guest to prove the
/// tunnel holds.
const TCP_PROBE_READ_WINDOW: Duration = Duration::from_secs(5);

/// Per-poll bound on the control-plane request when the overall wait is
/// unbounded, so a hung request still cannot stall the loop indefinitely.
const MAX_POLL_RPC_BUDGET: Duration = Duration::from_mins(1);

/// The public URL for an HTTP listener the server left blank: synthesized from
/// the client's ingress config (the server omits URLs in local/path mode).
/// Returns an empty string when no base is configured.
pub(crate) fn synthesized_public_url(config: &Config, sailbox_id: &str, guest_port: u32) -> String {
    let base = config.ingress_base.trim().trim_end_matches('/');
    if base.is_empty() {
        return String::new();
    }
    if config.ingress_scheme != IngressScheme::Subdomain {
        return format!("{base}/_sailbox/{sailbox_id}/{guest_port}");
    }
    let parsed = url::Url::parse(base)
        .ok()
        .filter(|url| url.host_str().is_some())
        .or_else(|| url::Url::parse(&format!("https://{base}")).ok());
    let Some(parsed) = parsed else {
        return format!("{base}/_sailbox/{sailbox_id}/{guest_port}");
    };
    let Some(host) = parsed.host_str() else {
        return format!("{base}/_sailbox/{sailbox_id}/{guest_port}");
    };
    let label = dns_safe_listener_label(sailbox_id);
    let port_suffix = parsed
        .port()
        .map(|port| format!(":{port}"))
        .unwrap_or_default();
    format!(
        "{}://{label}-{guest_port}.{host}{port_suffix}",
        parsed.scheme()
    )
}

/// Sailbox ids become DNS labels in subdomain URLs: lowercase, `_` → `-`.
fn dns_safe_listener_label(sailbox_id: &str) -> String {
    sailbox_id.trim().to_lowercase().replace('_', "-")
}

/// One reachability check against a listener's endpoint. Returns
/// `(ready, detail)` where `detail` explains a not-ready result.
async fn probe_once(listener: &Listener, budget: Duration) -> (bool, String) {
    if !listener.is_active() {
        return (false, listener.route_status.as_str().to_string());
    }
    if listener.protocol == crate::sailbox::types::ListenerProtocol::Tcp {
        if listener.public_host.is_empty() || listener.public_port == 0 {
            return (false, "listener has no endpoint yet".to_string());
        }
        return tcp_endpoint_accepts(&listener.public_host, listener.public_port, budget).await;
    }
    if listener.public_url.is_empty() {
        return (false, "listener has no endpoint yet".to_string());
    }
    http_endpoint_ready(&listener.public_url, budget).await
}

/// Whether an HTTP listener answers end to end. Any response counts — even an
/// error status means the request reached a server — except the ingress edge's
/// 502/503 "upstream unavailable", which means the guest is not answering yet.
async fn http_endpoint_ready(public_url: &str, budget: Duration) -> (bool, String) {
    let request = reqwest::Client::new()
        .get(public_url)
        .timeout(budget)
        .send();
    match request.await {
        Ok(response) => {
            let status = response.status().as_u16();
            if status == 502 || status == 503 {
                let body = response.text().await.unwrap_or_default();
                let body = body.trim();
                if body.eq_ignore_ascii_case("upstream unavailable") {
                    return (false, format!("HTTP {status}: {body}"));
                }
            }
            (true, String::new())
        }
        Err(err) => (false, err.to_string()),
    }
}

/// Whether a raw-TCP listener's guest port accepted the connection. The ingress
/// edge passes raw bytes through, so a completed connect only proves the edge
/// accepted us; the platform drops the connection if the guest isn't accepting.
/// Bytes from the guest (e.g. an SSH banner) or a quiet-but-open socket both
/// mean the guest accepted, while a prompt EOF/reset means it could not be
/// dialed.
async fn tcp_endpoint_accepts(host: &str, port: u32, budget: Duration) -> (bool, String) {
    let addr = format!("{host}:{port}");
    let started = Instant::now();
    let mut stream = match tokio::time::timeout(budget, TcpStream::connect(&addr)).await {
        Ok(Ok(stream)) => stream,
        Ok(Err(err)) => return (false, err.to_string()),
        Err(_) => return (false, format!("timed out connecting to {addr}")),
    };
    // The read spends what the connect left of the budget, so the probe never
    // outlives the caller's deadline. A near-zero window can miss an ingress
    // EOF and report a quiet-open socket ready, which is the bias quiet-open
    // already carries; overshooting the advertised deadline would be worse.
    let read_window = budget
        .saturating_sub(started.elapsed())
        .min(TCP_PROBE_READ_WINDOW);
    let mut buf = [0u8; 1];
    match tokio::time::timeout(read_window, stream.read(&mut buf)).await {
        Ok(Ok(0)) => (
            false,
            "ingress closed the connection; guest port not accepting".to_string(),
        ),
        Ok(Ok(_)) => (true, String::new()),
        Ok(Err(err)) => (false, err.to_string()),
        // A quiet-but-open socket: the guest accepted and is waiting for the
        // client to speak first.
        Err(_) => (true, String::new()),
    }
}

impl Client {
    /// Id-form of [`Sailbox::wait_for_listener`](crate::Sailbox::wait_for_listener),
    /// which documents the probe contract.
    #[doc(hidden)]
    pub async fn wait_for_listener(
        &self,
        sailbox_id: &str,
        guest_port: u32,
        timeout: Duration,
        poll_interval: Duration,
    ) -> Result<Listener, SailError> {
        if poll_interval.is_zero() {
            return Err(SailError::InvalidArgument {
                message: "poll_interval must be positive".to_string(),
            });
        }
        let deadline = Instant::now().checked_add(timeout);
        let remaining = |fallback: Duration| {
            deadline.map_or(fallback, |deadline| {
                deadline.saturating_duration_since(Instant::now())
            })
        };
        let mut last_detail = String::new();
        loop {
            // Bound each control-plane poll by the remaining budget so a hung
            // request cannot outlive the advertised deadline.
            let poll_budget = remaining(MAX_POLL_RPC_BUDGET).max(Duration::from_millis(1));
            let listener =
                match tokio::time::timeout(poll_budget, self.get_listener(sailbox_id, guest_port))
                    .await
                {
                    Ok(result) => result?,
                    // A hung poll ends a bounded wait (the budget was the time
                    // remaining), but an unbounded wait keeps polling: the cap
                    // only recycles a stuck control-plane request.
                    Err(_) if deadline.is_some() => {
                        return Err(wait_timeout_error(
                            sailbox_id,
                            guest_port,
                            timeout,
                            &last_detail,
                        ))
                    }
                    Err(_) => {
                        last_detail = "control-plane poll timed out".to_string();
                        tokio::time::sleep(poll_interval).await;
                        continue;
                    }
                };
            // An unbounded wait still bounds each probe; the RPC-budget cap is
            // the whole-probe allowance (the TCP read window is applied inside
            // the TCP probe itself).
            let (ready, detail) = probe_once(&listener, remaining(MAX_POLL_RPC_BUDGET)).await;
            if ready {
                return Ok(listener);
            }
            last_detail = detail;
            let left = remaining(Duration::MAX);
            if left.is_zero() {
                return Err(wait_timeout_error(
                    sailbox_id,
                    guest_port,
                    timeout,
                    &last_detail,
                ));
            }
            tokio::time::sleep(poll_interval.min(left)).await;
        }
    }
}

fn wait_timeout_error(
    sailbox_id: &str,
    guest_port: u32,
    timeout: Duration,
    last_detail: &str,
) -> SailError {
    let detail = if last_detail.is_empty() {
        String::new()
    } else {
        format!("; last error was {last_detail}")
    };
    SailError::Transport {
        kind: TransportKind::Timeout,
        message: format!(
            "listener {sailbox_id}:{guest_port} did not become reachable within {:.0} \
             seconds{detail}",
            timeout.as_secs_f64()
        ),
        source: None,
    }
}

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

    fn config(base: &str, scheme: IngressScheme) -> Config {
        Config {
            api_key: String::new(),
            api_url: String::new(),
            sailbox_api_url: String::new(),
            imagebuilder_url: String::new(),
            ingress_base: base.to_string(),
            ingress_scheme: scheme,
        }
    }

    #[test]
    fn synthesizes_path_and_subdomain_urls() {
        assert_eq!(
            synthesized_public_url(
                &config("http://localhost:8080/", IngressScheme::Path),
                "sb_1",
                3000
            ),
            "http://localhost:8080/_sailbox/sb_1/3000"
        );
        assert_eq!(
            synthesized_public_url(
                &config("ingress.example.com", IngressScheme::Subdomain),
                "sb_ab_1",
                3000
            ),
            "https://sb-ab-1-3000.ingress.example.com"
        );
        assert_eq!(
            synthesized_public_url(
                &config("http://ingress.local:8443", IngressScheme::Subdomain),
                "sb_1",
                80
            ),
            "http://sb-1-80.ingress.local:8443"
        );
        assert_eq!(
            synthesized_public_url(&config("", IngressScheme::Path), "sb_1", 80),
            ""
        );
        // A path-suffixed or ip:port base is preserved verbatim in path mode.
        assert_eq!(
            synthesized_public_url(
                &config(
                    "https://tunnel.example.net/sailbox-ingress",
                    IngressScheme::Path
                ),
                "sb_1",
                3000
            ),
            "https://tunnel.example.net/sailbox-ingress/_sailbox/sb_1/3000"
        );
        assert_eq!(
            synthesized_public_url(
                &config("http://10.0.0.5:18080", IngressScheme::Path),
                "sb_1",
                3000
            ),
            "http://10.0.0.5:18080/_sailbox/sb_1/3000"
        );
    }

    #[tokio::test]
    async fn tcp_probe_reads_the_guest_signal() {
        use tokio::io::AsyncWriteExt;
        use tokio::net::TcpListener;

        // Bytes from the guest → ready.
        let banner = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = banner.local_addr().unwrap();
        tokio::spawn(async move {
            let (mut sock, _) = banner.accept().await.unwrap();
            sock.write_all(b"SSH-2.0-OpenSSH").await.unwrap();
        });
        let (ready, _) =
            tcp_endpoint_accepts("127.0.0.1", addr.port().into(), Duration::from_secs(5)).await;
        assert!(ready);

        // Prompt EOF → the guest port was not dialable.
        let refuser = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = refuser.local_addr().unwrap();
        tokio::spawn(async move {
            let (sock, _) = refuser.accept().await.unwrap();
            drop(sock);
        });
        let (ready, detail) =
            tcp_endpoint_accepts("127.0.0.1", addr.port().into(), Duration::from_secs(5)).await;
        assert!(!ready);
        assert!(detail.contains("not accepting"), "{detail}");

        // Refused connection → not ready.
        let (ready, _) = tcp_endpoint_accepts("127.0.0.1", 1, Duration::from_secs(1)).await;
        assert!(!ready);

        // A quiet-but-open socket: the guest accepted and is waiting for the
        // client to speak first → ready once the read window lapses.
        let quiet = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = quiet.local_addr().unwrap();
        tokio::spawn(async move {
            let (_sock, _) = quiet.accept().await.unwrap();
            tokio::time::sleep(Duration::from_secs(30)).await;
        });
        let (ready, _) =
            tcp_endpoint_accepts("127.0.0.1", addr.port().into(), Duration::from_millis(300)).await;
        assert!(ready);
    }

    #[tokio::test]
    async fn http_probe_distinguishes_upstream_unavailable() {
        use tokio::io::AsyncWriteExt;
        use tokio::net::TcpListener;

        async fn serve_once(response: &'static str) -> String {
            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = listener.local_addr().unwrap();
            tokio::spawn(async move {
                let (mut sock, _) = listener.accept().await.unwrap();
                let mut buf = [0u8; 1024];
                let _ = sock.read(&mut buf).await;
                sock.write_all(response.as_bytes()).await.unwrap();
            });
            format!("http://{addr}/")
        }

        let url = serve_once(
            "HTTP/1.1 503 Service Unavailable\r\ncontent-length: 20\r\n\r\nupstream unavailable",
        )
        .await;
        let (ready, detail) = http_endpoint_ready(&url, Duration::from_secs(5)).await;
        assert!(!ready);
        assert!(detail.contains("503"), "{detail}");

        // Any other response — even an error status — means the guest answered.
        let url =
            serve_once("HTTP/1.1 500 Internal Server Error\r\ncontent-length: 0\r\n\r\n").await;
        let (ready, _) = http_endpoint_ready(&url, Duration::from_secs(5)).await;
        assert!(ready);

        let (ready, _) = http_endpoint_ready("http://127.0.0.1:1/", Duration::from_secs(1)).await;
        assert!(!ready);
    }
}
