//! Scoped outbound HTTP client.
//!
//! Provides a builder-style API for outbound HTTP requests. All operations go
//! through `scope.http_out()` — engine code must not use `reqwest` directly.
//!
//! # Design
//!
//! - **Transparent passthrough**: bytes in / bytes out, no automatic JSON
//!   deserialization. Suitable for proxy-style use cases.
//! - **4xx/5xx are not errors**: caller inspects `status()` and decides.
//!   Only network-layer failures (connect/timeout/TLS/URL) become `Critical` errors.
//! - **Native errors are sealed**: raw `reqwest::Error` / `hyper::Error` / `io::Error`
//!   messages are never exposed to the engine — they are wrapped in `Sealed<String>`.
//! - **Streaming**: `send_stream()` returns `HttpStream` where the body is an
//!   async `Stream<Item = Result<Bytes, HttpOutError>>` for chunk-by-chunk consumption.
//! - **Connection pooling**: internal shared `reqwest::Client` (see `shared_http_client`).

use crate::critical::{Critical, Transaction, TxScope};
use crate::sealed::Sealed;
use bytes::Bytes;
use futures_util::Stream;
use std::pin::Pin;
use std::time::Duration;

/// Default request timeout if not overridden by the caller.
const DEFAULT_TIMEOUT_SECS: u64 = 60;

/// Operation name used in `Critical` errors from this module.
const OP_NAME: &str = "http_out";

// ─── Errors ─────────────────────────────────────────────────────────────

/// Outbound HTTP error classification.
///
/// All variants carry a `Sealed<String>` with implementation-detail diagnostics.
/// Engine code can match on the variant for failover decisions but cannot read
/// the detail string.
#[derive(Debug)]
pub enum HttpOutError {
    /// Failed to establish TCP connection (DNS resolution, connection refused, etc.)
    Connection { detail: Sealed<String> },
    /// Request exceeded the configured timeout.
    Timeout { detail: Sealed<String> },
    /// URL format is invalid (scheme, host, port, etc.)
    InvalidUrl { detail: Sealed<String> },
    /// TLS handshake / certificate error.
    Tls { detail: Sealed<String> },
    /// Any other error (kept to evolve the enum without breaking callers).
    Other { detail: Sealed<String> },
}

impl std::fmt::Display for HttpOutError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Connection { detail } => write!(f, "connection: {}", detail),
            Self::Timeout { detail } => write!(f, "timeout: {}", detail),
            Self::InvalidUrl { detail } => write!(f, "invalid url: {}", detail),
            Self::Tls { detail } => write!(f, "tls: {}", detail),
            Self::Other { detail } => write!(f, "other: {}", detail),
        }
    }
}

impl std::error::Error for HttpOutError {}

impl HttpOutError {
    /// Returns true if this error indicates a transient condition that may
    /// succeed on retry / failover to a different endpoint.
    pub fn is_retryable(&self) -> bool {
        matches!(
            self,
            Self::Connection { .. } | Self::Timeout { .. } | Self::Tls { .. }
        )
    }

    fn summary(&self) -> &'static str {
        match self {
            Self::Connection { .. } => "connection",
            Self::Timeout { .. } => "timeout",
            Self::InvalidUrl { .. } => "invalid_url",
            Self::Tls { .. } => "tls",
            Self::Other { .. } => "other",
        }
    }
}

/// Classify a `reqwest::Error` into `HttpOutError`.
///
/// We intentionally never propagate the raw error — only a sealed diagnostic
/// string plus a category tag for failover decisions.
fn classify_reqwest_error(e: reqwest::Error) -> HttpOutError {
    let msg = format!("{}", e);
    let detail = Sealed::new(msg);
    if e.is_timeout() {
        HttpOutError::Timeout { detail }
    } else if e.is_connect() {
        HttpOutError::Connection { detail }
    } else if e.is_builder() {
        HttpOutError::InvalidUrl { detail }
    } else {
        // reqwest doesn't expose a direct TLS predicate; fall back to string match
        // for common TLS-related errors.
        let lc = e.to_string().to_ascii_lowercase();
        if lc.contains("tls") || lc.contains("certificate") || lc.contains("ssl") {
            HttpOutError::Tls { detail }
        } else {
            HttpOutError::Other { detail }
        }
    }
}

// ─── Response ───────────────────────────────────────────────────────────

/// Non-streaming HTTP response.
///
/// The full body is buffered in memory. For large or streaming responses,
/// use `send_stream()` instead.
pub struct HttpResponse {
    status: u16,
    headers: Vec<(String, Vec<u8>)>,
    body: Bytes,
}

impl HttpResponse {
    /// HTTP status code (e.g. 200, 429, 503). `4xx`/`5xx` are NOT errors —
    /// the caller inspects this to decide what to do (e.g. failover on 429/5xx).
    pub fn status(&self) -> u16 {
        self.status
    }

    /// Iterate over response headers. Values are raw bytes (headers may be
    /// non-UTF8 for some protocols).
    pub fn headers(&self) -> impl Iterator<Item = (&str, &[u8])> {
        self.headers.iter().map(|(k, v)| (k.as_str(), v.as_slice()))
    }

    /// Get the first header value by name (case-insensitive).
    pub fn header(&self, name: &str) -> Option<&[u8]> {
        self.headers
            .iter()
            .find(|(k, _)| k.eq_ignore_ascii_case(name))
            .map(|(_, v)| v.as_slice())
    }

    /// Raw body bytes.
    pub fn body_bytes(&self) -> &[u8] {
        &self.body
    }

    /// Consume the response and return the body.
    pub fn into_body(self) -> Bytes {
        self.body
    }
}

// ─── Stream ─────────────────────────────────────────────────────────────

/// Streaming HTTP response.
///
/// Headers are available immediately; the body is consumed chunk-by-chunk.
/// Typical use: feed `body_stream()` into `axum::body::Body::from_stream` to
/// proxy upstream LLM SSE responses.
pub struct HttpStream {
    status: u16,
    headers: Vec<(String, Vec<u8>)>,
    body: Pin<Box<dyn Stream<Item = Result<Bytes, HttpOutError>> + Send>>,
}

impl HttpStream {
    /// HTTP status code from response headers. `4xx`/`5xx` are NOT errors.
    pub fn status(&self) -> u16 {
        self.status
    }

    /// Iterate over response headers.
    pub fn headers(&self) -> impl Iterator<Item = (&str, &[u8])> {
        self.headers.iter().map(|(k, v)| (k.as_str(), v.as_slice()))
    }

    /// Get the first header value by name (case-insensitive).
    pub fn header(&self, name: &str) -> Option<&[u8]> {
        self.headers
            .iter()
            .find(|(k, _)| k.eq_ignore_ascii_case(name))
            .map(|(_, v)| v.as_slice())
    }

    /// Consume the stream and return the body chunk stream.
    ///
    /// Each item is `Result<Bytes, HttpOutError>`: the chunk bytes or a
    /// chunk-level error (e.g. mid-stream network drop).
    pub fn body_stream(self) -> Pin<Box<dyn Stream<Item = Result<Bytes, HttpOutError>> + Send>> {
        self.body
    }
}

// ─── HTTP method ────────────────────────────────────────────────────────

#[derive(Debug, Clone, Copy)]
enum Method {
    Get,
    Post,
    Put,
    Delete,
    Patch,
}

impl Method {
    fn as_reqwest(&self) -> reqwest::Method {
        match self {
            Self::Get => reqwest::Method::GET,
            Self::Post => reqwest::Method::POST,
            Self::Put => reqwest::Method::PUT,
            Self::Delete => reqwest::Method::DELETE,
            Self::Patch => reqwest::Method::PATCH,
        }
    }

    fn as_str(&self) -> &'static str {
        match self {
            Self::Get => "GET",
            Self::Post => "POST",
            Self::Put => "PUT",
            Self::Delete => "DELETE",
            Self::Patch => "PATCH",
        }
    }
}

// ─── Scoped entry point ─────────────────────────────────────────────────

/// Outbound HTTP handle scoped to a transaction.
///
/// Construct via `scope.http_out()`.
pub struct ScopedHttpOut<'a, T: Transaction> {
    pub(crate) scope: &'a TxScope<T>,
}

impl<'a, T: Transaction> ScopedHttpOut<'a, T> {
    /// Start a GET request.
    pub fn get(self, url: &str) -> HttpRequestBuilder<'a, T> {
        HttpRequestBuilder::new(self.scope, Method::Get, url)
    }

    /// Start a POST request.
    pub fn post(self, url: &str) -> HttpRequestBuilder<'a, T> {
        HttpRequestBuilder::new(self.scope, Method::Post, url)
    }

    /// Start a PUT request.
    pub fn put(self, url: &str) -> HttpRequestBuilder<'a, T> {
        HttpRequestBuilder::new(self.scope, Method::Put, url)
    }

    /// Start a DELETE request.
    pub fn delete(self, url: &str) -> HttpRequestBuilder<'a, T> {
        HttpRequestBuilder::new(self.scope, Method::Delete, url)
    }

    /// Start a PATCH request.
    pub fn patch(self, url: &str) -> HttpRequestBuilder<'a, T> {
        HttpRequestBuilder::new(self.scope, Method::Patch, url)
    }
}

// ─── Request builder ────────────────────────────────────────────────────

/// Builder for an outbound HTTP request.
///
/// Chain `.header()` / `.body_bytes()` / `.timeout_secs()` and finish with
/// `.send()` (buffered) or `.send_stream()` (streaming).
pub struct HttpRequestBuilder<'a, T: Transaction> {
    scope: &'a TxScope<T>,
    method: Method,
    url: String,
    headers: Vec<(String, String)>,
    body: Option<Bytes>,
    timeout: Duration,
}

impl<'a, T: Transaction> HttpRequestBuilder<'a, T> {
    fn new(scope: &'a TxScope<T>, method: Method, url: &str) -> Self {
        Self {
            scope,
            method,
            url: url.to_string(),
            headers: Vec::new(),
            body: None,
            timeout: Duration::from_secs(DEFAULT_TIMEOUT_SECS),
        }
    }

    /// Add a request header. Repeat for multiple values of the same name.
    pub fn header(mut self, name: &str, value: &str) -> Self {
        self.headers.push((name.to_string(), value.to_string()));
        self
    }

    /// Set the request body (raw bytes, transparent passthrough).
    pub fn body_bytes(mut self, bytes: impl Into<Bytes>) -> Self {
        self.body = Some(bytes.into());
        self
    }

    /// Override the request timeout. Default is 60 seconds.
    pub fn timeout_secs(mut self, secs: u64) -> Self {
        self.timeout = Duration::from_secs(secs);
        self
    }

    /// Send the request and buffer the full response body in memory.
    ///
    /// Returns `Critical<HttpResponse>`. `4xx`/`5xx` statuses are **not** errors
    /// — only network-layer failures become `Critical::error`.
    pub async fn send(self) -> Critical<HttpResponse> {
        let method_str = self.method.as_str();
        let (req, timeout) = match self.build_request() {
            Ok(v) => v,
            Err(e) => {
                return Self::error_to_critical(e, method_str);
            }
        };
        let fut = req.send();
        let response = match tokio::time::timeout(timeout, fut).await {
            Ok(Ok(r)) => r,
            Ok(Err(e)) => {
                return Self::error_to_critical(classify_reqwest_error(e), method_str);
            }
            Err(_) => {
                return Self::error_to_critical(
                    HttpOutError::Timeout {
                        detail: Sealed::new(format!("deadline exceeded after {:?}", timeout)),
                    },
                    method_str,
                );
            }
        };

        let status = response.status().as_u16();
        let headers = collect_headers(response.headers());
        let body = match response.bytes().await {
            Ok(b) => b,
            Err(e) => {
                return Self::error_to_critical(classify_reqwest_error(e), method_str);
            }
        };

        let result = Critical::ok(HttpResponse { status, headers, body });
        result.with_tx_context(self.scope.raw_trace_id(), T::name())
    }

    /// Send the request and return a streaming response.
    ///
    /// Headers are available immediately; body chunks arrive via
    /// `HttpStream::body_stream()`.
    pub async fn send_stream(self) -> Critical<HttpStream> {
        let method_str = self.method.as_str();
        let (req, timeout) = match self.build_request() {
            Ok(v) => v,
            Err(e) => {
                return Self::error_to_critical(e, method_str);
            }
        };
        // Only the initial response (status+headers) is subject to `timeout`.
        // Body-chunk reads run without a wall-clock deadline — callers that
        // need per-chunk timeouts can layer their own on top of the stream.
        let fut = req.send();
        let response = match tokio::time::timeout(timeout, fut).await {
            Ok(Ok(r)) => r,
            Ok(Err(e)) => {
                return Self::error_to_critical(classify_reqwest_error(e), method_str);
            }
            Err(_) => {
                return Self::error_to_critical(
                    HttpOutError::Timeout {
                        detail: Sealed::new(format!("deadline exceeded after {:?}", timeout)),
                    },
                    method_str,
                );
            }
        };

        let status = response.status().as_u16();
        let headers = collect_headers(response.headers());
        let body_stream = response.bytes_stream();
        let mapped: Pin<Box<dyn Stream<Item = Result<Bytes, HttpOutError>> + Send>> =
            Box::pin(futures_util::StreamExt::map(body_stream, |chunk| {
                chunk.map_err(classify_reqwest_error)
            }));

        let result = Critical::ok(HttpStream {
            status,
            headers,
            body: mapped,
        });
        result.with_tx_context(self.scope.raw_trace_id(), T::name())
    }

    fn build_request(&self) -> Result<(reqwest::RequestBuilder, Duration), HttpOutError> {
        let client = shared_http_client();
        let mut req = client.request(self.method.as_reqwest(), &self.url);
        for (k, v) in &self.headers {
            req = req.header(k, v);
        }
        if let Some(body) = &self.body {
            req = req.body(body.clone().to_vec());
        }
        Ok((req, self.timeout))
    }

    fn error_to_critical<V>(err: HttpOutError, method: &'static str) -> Critical<V> {
        let tag = err.summary();
        Critical::error(
            OP_NAME,
            format!("{} {}: {} — {}", method, tag, tag, err),
        )
    }
}

// ─── Internal helpers ───────────────────────────────────────────────────

fn collect_headers(h: &reqwest::header::HeaderMap) -> Vec<(String, Vec<u8>)> {
    h.iter()
        .map(|(k, v)| (k.as_str().to_string(), v.as_bytes().to_vec()))
        .collect()
}

/// Shared `reqwest::Client` for connection pooling across all outbound HTTP
/// calls made through `scope.http_out()`.
///
/// `reqwest::Client` is internally `Arc`-wrapped, so `.clone()` is cheap.
pub(crate) fn shared_http_client() -> reqwest::Client {
    static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
    CLIENT
        .get_or_init(|| {
            reqwest::Client::builder()
                .connect_timeout(Duration::from_secs(30))
                .build()
                .expect("shared_http_client: failed to build reqwest::Client")
        })
        .clone()
}

// ─── TxScope extension ──────────────────────────────────────────────────

impl<T: Transaction> TxScope<T> {
    /// Get a scoped outbound HTTP client.
    ///
    /// All `Critical` results carry `tx_context` (trace_id + tx_name) so
    /// failures report which transaction they belong to.
    ///
    /// # Example
    /// ```ignore
    /// let resp = scope.http_out()
    ///     .post("https://api.example.com/v1/chat")
    ///     .header("Authorization", "Bearer xxx")
    ///     .body_bytes(body)
    ///     .timeout_secs(60)
    ///     .send().await
    ///     .suppress(MyToken::HttpFailed, "upstream unavailable");
    /// ```
    pub fn http_out(&self) -> ScopedHttpOut<'_, T> {
        ScopedHttpOut { scope: self }
    }
}