//! Integration tests for `scope.http_out()`.
//!
//! Uses a minimal TcpListener-based mock server (no external HTTP mocking deps)
//! to exercise non-streaming, streaming, header passthrough, and error paths.

use bytes::Bytes;
use futures_util::StreamExt;
use mad_hatter::critical::{
    ConceptMapToken, Critical, FatalToken, SuppressibleToken, Transaction, TxScope,
};
use mad_hatter::{HttpOutError, HttpResponse, HttpStream};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use tokio::task::JoinHandle;

// ─── Test Transaction type ──────────────────────────────────────────────

#[derive(Debug, Clone, Copy, PartialEq)]
enum HttpTestFatal {
    Upstream,
}

impl ConceptMapToken for HttpTestFatal {
    fn operation_name(&self) -> &'static str {
        match self {
            HttpTestFatal::Upstream => "http_test::upstream",
        }
    }
}

impl FatalToken for HttpTestFatal {}

#[derive(Debug, Clone, Copy, PartialEq)]
enum HttpTestSuppressible {
    BestEffort,
}

impl ConceptMapToken for HttpTestSuppressible {
    fn operation_name(&self) -> &'static str {
        match self {
            HttpTestSuppressible::BestEffort => "http_test::best_effort",
        }
    }
}

impl SuppressibleToken for HttpTestSuppressible {}

struct HttpTestTx;
impl Transaction for HttpTestTx {
    type Fatal = HttpTestFatal;
    type Suppressible = HttpTestSuppressible;
    fn name() -> &'static str {
        "http_test"
    }
}

// ─── Mock server helpers ────────────────────────────────────────────────

/// Start an HTTP server that responds with the given status, headers, and body.
///
/// Also captures the first request's method/path/headers/body into `captured`.
#[allow(clippy::type_complexity)]
async fn start_mock_server(
    status: u16,
    reason: &'static str,
    resp_headers: Vec<(&'static str, &'static str)>,
    resp_body: Vec<u8>,
    captured: std::sync::Arc<std::sync::Mutex<Option<CapturedRequest>>>,
) -> (u16, JoinHandle<()>) {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    let handle = tokio::spawn(async move {
        loop {
            let (mut stream, _) = match listener.accept().await {
                Ok(v) => v,
                Err(_) => break,
            };
            let captured = captured.clone();
            let resp_headers = resp_headers.clone();
            let resp_body = resp_body.clone();
            tokio::spawn(async move {
                let mut buf = vec![0u8; 16384];
                let mut total = 0usize;
                loop {
                    let n = match stream.read(&mut buf[total..]).await {
                        Ok(0) => break,
                        Ok(n) => n,
                        Err(_) => return,
                    };
                    total += n;
                    if let Some(hdr_end) = find_subslice(&buf[..total], b"\r\n\r\n") {
                        let head_bytes = &buf[..hdr_end];
                        let head_str = String::from_utf8_lossy(head_bytes);
                        let mut lines = head_str.lines();
                        let first = lines.next().unwrap_or("").to_string();
                        let parts: Vec<&str> = first.split_whitespace().collect();
                        let method = parts.first().copied().unwrap_or("").to_string();
                        let path = parts.get(1).copied().unwrap_or("").to_string();
                        let mut headers: Vec<(String, String)> = Vec::new();
                        let mut content_length: usize = 0;
                        for line in lines {
                            if let Some((k, v)) = line.split_once(':') {
                                let k = k.trim().to_string();
                                let v = v.trim().to_string();
                                if k.eq_ignore_ascii_case("content-length") {
                                    content_length = v.parse().unwrap_or(0);
                                }
                                headers.push((k, v));
                            }
                        }
                        let body_start = hdr_end + 4;
                        while total - body_start < content_length && total < buf.len() {
                            let n = match stream.read(&mut buf[total..]).await {
                                Ok(0) => break,
                                Ok(n) => n,
                                Err(_) => return,
                            };
                            total += n;
                        }
                        let body_end = (body_start + content_length).min(total);
                        let body = buf[body_start..body_end].to_vec();
                        let mut guard = captured.lock().unwrap();
                        if guard.is_none() {
                            *guard = Some(CapturedRequest {
                                method,
                                path,
                                headers,
                                body,
                            });
                        }
                        break;
                    }
                    if total >= buf.len() {
                        return;
                    }
                }

                let mut response = format!("HTTP/1.1 {} {}\r\n", status, reason);
                response.push_str(&format!("Content-Length: {}\r\n", resp_body.len()));
                for (k, v) in &resp_headers {
                    response.push_str(&format!("{}: {}\r\n", k, v));
                }
                response.push_str("\r\n");
                let mut bytes = response.into_bytes();
                bytes.extend_from_slice(&resp_body);
                let _ = stream.write_all(&bytes).await;
                let _ = stream.shutdown().await;
            });
        }
    });
    (port, handle)
}

#[derive(Debug)]
struct CapturedRequest {
    method: String,
    path: String,
    headers: Vec<(String, String)>,
    body: Vec<u8>,
}

fn find_subslice(haystack: &[u8], needle: &[u8]) -> Option<usize> {
    haystack.windows(needle.len()).position(|w| w == needle)
}

/// Start a chunked-encoding server that sends `chunks` with small delays.
async fn start_chunked_server(chunks: Vec<&'static str>) -> (u16, JoinHandle<()>) {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    let handle = tokio::spawn(async move {
        loop {
            let (mut stream, _) = match listener.accept().await {
                Ok(v) => v,
                Err(_) => break,
            };
            let chunks = chunks.clone();
            tokio::spawn(async move {
                let mut buf = vec![0u8; 8192];
                let _ = tokio::time::timeout(
                    std::time::Duration::from_millis(200),
                    stream.read(&mut buf),
                )
                .await;

                let head = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\n\r\n";
                let _ = stream.write_all(head.as_bytes()).await;

                for chunk in chunks {
                    let chunk_header = format!("{:x}\r\n", chunk.len());
                    let _ = stream.write_all(chunk_header.as_bytes()).await;
                    let _ = stream.write_all(chunk.as_bytes()).await;
                    let _ = stream.write_all(b"\r\n").await;
                    tokio::time::sleep(std::time::Duration::from_millis(20)).await;
                }
                let _ = stream.write_all(b"0\r\n\r\n").await;
                let _ = stream.shutdown().await;
            });
        }
    });
    (port, handle)
}

/// Start a server that accepts but never responds (for timeout tests).
async fn start_slow_server() -> (u16, JoinHandle<()>) {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    let handle = tokio::spawn(async move {
        loop {
            let (stream, _) = match listener.accept().await {
                Ok(v) => v,
                Err(_) => break,
            };
            tokio::spawn(async move {
                tokio::time::sleep(std::time::Duration::from_secs(30)).await;
                drop(stream);
            });
        }
    });
    (port, handle)
}

// ─── Tests ──────────────────────────────────────────────────────────────

#[tokio::test]
async fn test_non_streaming_200_success() {
    let captured = std::sync::Arc::new(std::sync::Mutex::new(None));
    let (port, _h) = start_mock_server(
        200,
        "OK",
        vec![("X-Custom", "hello")],
        b"response body".to_vec(),
        captured.clone(),
    )
    .await;

    let scope: TxScope<HttpTestTx> = TxScope::new();
    let url = format!("http://127.0.0.1:{}/test", port);
    let c: Critical<HttpResponse> = scope
        .http_out()
        .post(&url)
        .header("X-Request", "abc")
        .body_bytes(Bytes::from_static(b"request body"))
        .timeout_secs(5)
        .send()
        .await;

    let resp = c
        .suppress(HttpTestSuppressible::BestEffort, "unexpected failure")
        .expect("expected success response, got error");

    assert_eq!(resp.status(), 200);
    assert_eq!(resp.body_bytes(), b"response body");
    assert_eq!(resp.header("X-Custom"), Some(&b"hello"[..]));

    let req = captured
        .lock()
        .unwrap()
        .take()
        .expect("no request captured");
    assert_eq!(req.method, "POST");
    assert_eq!(req.path, "/test");
    assert_eq!(req.body, b"request body");
    let has_custom = req
        .headers
        .iter()
        .any(|(k, v)| k.eq_ignore_ascii_case("x-request") && v == "abc");
    assert!(has_custom, "X-Request header not passed through");
}

#[tokio::test]
async fn test_4xx_not_error_passthrough() {
    let captured = std::sync::Arc::new(std::sync::Mutex::new(None));
    let (port, _h) = start_mock_server(
        404,
        "Not Found",
        vec![],
        b"{\"error\":\"not_found\"}".to_vec(),
        captured.clone(),
    )
    .await;

    let scope: TxScope<HttpTestTx> = TxScope::new();
    let url = format!("http://127.0.0.1:{}/missing", port);
    let c: Critical<HttpResponse> = scope.http_out().get(&url).timeout_secs(5).send().await;

    // 4xx must NOT be Critical error — caller inspects status.
    let resp = c
        .suppress(HttpTestSuppressible::BestEffort, "should not be error")
        .expect("4xx was classified as Critical error (wrong)");
    assert_eq!(resp.status(), 404);
    assert_eq!(resp.body_bytes(), b"{\"error\":\"not_found\"}");
}

#[tokio::test]
async fn test_5xx_not_error_passthrough() {
    let captured = std::sync::Arc::new(std::sync::Mutex::new(None));
    let (port, _h) = start_mock_server(
        503,
        "Service Unavailable",
        vec![],
        b"down".to_vec(),
        captured.clone(),
    )
    .await;

    let scope: TxScope<HttpTestTx> = TxScope::new();
    let url = format!("http://127.0.0.1:{}/", port);
    let c: Critical<HttpResponse> = scope.http_out().get(&url).timeout_secs(5).send().await;

    let resp = c
        .suppress(HttpTestSuppressible::BestEffort, "should not be error")
        .expect("5xx was classified as Critical error (wrong)");
    assert_eq!(resp.status(), 503);
    assert_eq!(resp.body_bytes(), b"down");
}

#[tokio::test]
async fn test_connection_refused_is_critical_error() {
    // Bind a port, then drop the listener so the port is free.
    let port = {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        listener.local_addr().unwrap().port()
    };

    let scope: TxScope<HttpTestTx> = TxScope::new();
    let url = format!("http://127.0.0.1:{}/", port);
    let c: Critical<HttpResponse> = scope.http_out().get(&url).timeout_secs(3).send().await;

    // suppress returns None on error — confirms connection failure was classified
    // as a Critical error (not passed through as 2xx/4xx/5xx).
    let opt = c.suppress(HttpTestSuppressible::BestEffort, "connection refused expected");
    assert!(
        opt.is_none(),
        "expected Critical error on connection refused"
    );
}

#[tokio::test]
async fn test_timeout_is_critical_error() {
    let (port, _h) = start_slow_server().await;
    let scope: TxScope<HttpTestTx> = TxScope::new();
    let url = format!("http://127.0.0.1:{}/", port);
    let c: Critical<HttpResponse> = scope.http_out().get(&url).timeout_secs(1).send().await;

    let opt = c.suppress(HttpTestSuppressible::BestEffort, "timeout expected");
    assert!(
        opt.is_none(),
        "expected timeout to surface as Critical error"
    );
}

#[tokio::test]
async fn test_invalid_url_is_critical_error() {
    let scope: TxScope<HttpTestTx> = TxScope::new();
    let c: Critical<HttpResponse> = scope
        .http_out()
        .get("not-a-valid-url-at-all")
        .send()
        .await;
    let opt = c.suppress(HttpTestSuppressible::BestEffort, "invalid url expected");
    assert!(
        opt.is_none(),
        "expected invalid URL to surface as Critical error"
    );
}

#[tokio::test]
async fn test_streaming_chunks() {
    let (port, _h) = start_chunked_server(vec![
        "data: one\n\n",
        "data: two\n\n",
        "data: three\n\n",
    ])
    .await;

    let scope: TxScope<HttpTestTx> = TxScope::new();
    let url = format!("http://127.0.0.1:{}/sse", port);
    let c: Critical<HttpStream> = scope
        .http_out()
        .get(&url)
        .timeout_secs(5)
        .send_stream()
        .await;

    let stream = c
        .suppress(HttpTestSuppressible::BestEffort, "stream open failed")
        .expect("expected stream to open");

    assert_eq!(stream.status(), 200);
    let ct = stream.header("Content-Type").unwrap_or_default();
    assert_eq!(ct, b"text/event-stream");

    let mut body_stream = stream.body_stream();
    let mut collected = Vec::new();
    while let Some(item) = body_stream.next().await {
        match item {
            Ok(chunk) => collected.extend_from_slice(&chunk),
            Err(e) => panic!("chunk error: {}", e),
        }
    }

    let text = String::from_utf8_lossy(&collected);
    assert!(text.contains("one"), "missing 'one' in stream: {}", text);
    assert!(text.contains("two"), "missing 'two' in stream: {}", text);
    assert!(text.contains("three"), "missing 'three' in stream: {}", text);
}

#[tokio::test]
async fn test_http_out_error_is_retryable() {
    use mad_hatter::sealed::Sealed;
    let e = HttpOutError::Connection {
        detail: Sealed::new("refused".to_string()),
    };
    assert!(e.is_retryable());

    let e = HttpOutError::Timeout {
        detail: Sealed::new("dl".to_string()),
    };
    assert!(e.is_retryable());

    let e = HttpOutError::Tls {
        detail: Sealed::new("cert".to_string()),
    };
    assert!(e.is_retryable());

    let e = HttpOutError::InvalidUrl {
        detail: Sealed::new("bad".to_string()),
    };
    assert!(!e.is_retryable());

    let e = HttpOutError::Other {
        detail: Sealed::new("meh".to_string()),
    };
    assert!(!e.is_retryable());
}

#[tokio::test]
async fn test_request_headers_passthrough_multiple() {
    let captured = std::sync::Arc::new(std::sync::Mutex::new(None));
    let (port, _h) = start_mock_server(200, "OK", vec![], b"ok".to_vec(), captured.clone()).await;

    let scope: TxScope<HttpTestTx> = TxScope::new();
    let url = format!("http://127.0.0.1:{}/", port);
    let c: Critical<HttpResponse> = scope
        .http_out()
        .post(&url)
        .header("Authorization", "Bearer secret-token-123")
        .header("HTTP-Referer", "https://example.com")
        .header("X-Title", "my app")
        .body_bytes(Bytes::from_static(b"{}"))
        .timeout_secs(5)
        .send()
        .await;
    let _ = c.suppress(HttpTestSuppressible::BestEffort, "not under test");

    let req = captured
        .lock()
        .unwrap()
        .take()
        .expect("no request captured");
    let find = |name: &str| {
        req.headers
            .iter()
            .find(|(k, _)| k.eq_ignore_ascii_case(name))
            .map(|(_, v)| v.clone())
    };
    assert_eq!(
        find("Authorization"),
        Some("Bearer secret-token-123".to_string())
    );
    assert_eq!(
        find("HTTP-Referer"),
        Some("https://example.com".to_string())
    );
    assert_eq!(find("X-Title"), Some("my app".to_string()));
}