qubit-http 0.7.0

/*******************************************************************************
 *
 *    Copyright (c) 2025 - 2026 Haixing Hu.
 *
 *    SPDX-License-Identifier: Apache-2.0
 *
 *    Licensed under the Apache License, Version 2.0.
 *
 ******************************************************************************/
//! Unified HTTP response type and helpers.

use std::sync::{
    Arc,
    Mutex,
};
use std::time::Duration;

use async_stream::stream;
use bytes::Bytes;
use futures_util::{
    stream as futures_stream,
    StreamExt,
};
use http::header::{
    CONTENT_LENGTH,
    CONTENT_TYPE,
};
use http::{
    HeaderMap,
    Method,
    StatusCode,
};
use serde::de::DeserializeOwned;
use tokio_util::sync::CancellationToken;
use url::Url;

use crate::error::{
    backend_error_mapper::map_reqwest_error,
    ReqwestErrorPhase,
};
use crate::sse::{
    DoneMarkerPolicy,
    SseChunkStream,
    SseEventStream,
    SseJsonMode,
};
use crate::{
    HttpByteStream,
    HttpError,
    HttpErrorKind,
    HttpResult,
};

use super::{
    HttpResponseMeta,
    HttpResponseOptions,
};

/// Snapshot of a body read failure retained after the backend body is consumed.
#[derive(Debug, Clone)]
struct BodyReadFailure {
    /// Error category from the original failure.
    kind: HttpErrorKind,
    /// Original failure message.
    message: String,
    /// Optional request method context.
    method: Option<Method>,
    /// Optional request URL context.
    url: Option<Url>,
    /// Optional HTTP status context.
    status: Option<StatusCode>,
    /// Optional body preview context.
    response_body_preview: Option<String>,
    /// Optional retry-after context.
    retry_after: Option<Duration>,
}

impl BodyReadFailure {
    /// Captures cloneable diagnostic fields from a read error.
    ///
    /// # Parameters
    /// - `error`: Original body read failure.
    ///
    /// # Returns
    /// Cloneable failure snapshot for subsequent reads.
    fn from_error(error: &HttpError) -> Self {
        Self {
            kind: error.kind,
            message: error.message.clone(),
            method: error.method.clone(),
            url: error.url.clone(),
            status: error.status,
            response_body_preview: error.response_body_preview.clone(),
            retry_after: error.retry_after,
        }
    }

    /// Rebuilds an [`HttpError`] for a later read attempt.
    ///
    /// # Returns
    /// Error preserving the original kind and cloneable context.
    fn to_error(&self) -> HttpError {
        let mut error = HttpError::new(
            self.kind,
            format!("previous response body read failed: {}", self.message),
        );
        if let Some(method) = self.method.as_ref() {
            error = error.with_method(method);
        }
        if let Some(url) = self.url.as_ref() {
            error = error.with_url(url);
        }
        if let Some(status) = self.status {
            error = error.with_status(status);
        }
        if let Some(preview) = self.response_body_preview.as_ref() {
            error = error.with_response_body_preview(preview.clone());
        }
        if let Some(retry_after) = self.retry_after {
            error = error.with_retry_after(retry_after);
        }
        error
    }
}

/// Shared response body failure state.
type BodyReadFailureState = Arc<Mutex<Option<BodyReadFailure>>>;

/// Maps one backend response body read error into this crate's error model.
///
/// # Parameters
/// - `error`: Backend error returned while reading response body bytes.
/// - `method`: Request method used for diagnostics.
/// - `url`: Request URL used for diagnostics.
///
/// # Returns
/// [`HttpErrorKind::ReadTimeout`] for timeout errors, otherwise
/// [`HttpErrorKind::Transport`] for backend body read failures.
fn map_response_read_error(error: reqwest::Error, method: Method, url: Url) -> HttpError {
    if error.is_timeout() {
        return map_reqwest_error(
            error,
            HttpErrorKind::Transport,
            ReqwestErrorPhase::Read,
            Some(method),
            Some(url),
        );
    }

    HttpError::transport(format!("Failed to read response body: {}", error))
        .with_method(&method)
        .with_url(&url)
        .with_source(error)
}

/// Runtime state bound to one response instance.
#[derive(Debug, Clone)]
struct HttpResponseRuntime {
    /// Per-response read timeout inherited from request/client.
    read_timeout: Duration,
    /// Optional cancellation token inherited from request.
    cancellation_token: Option<CancellationToken>,
    /// Request URL used in read/cancellation error context.
    request_url: Url,
    /// First response body read failure, if the backend stream failed after
    /// being taken from this response.
    body_read_failure: BodyReadFailureState,
}

impl HttpResponseRuntime {
    fn new(
        read_timeout: Duration,
        cancellation_token: Option<CancellationToken>,
        request_url: Url,
    ) -> Self {
        Self {
            read_timeout,
            cancellation_token,
            request_url,
            body_read_failure: Arc::new(Mutex::new(None)),
        }
    }
}

/// Unified HTTP response with lazily consumed body.
#[derive(Debug)]
pub struct HttpResponse {
    /// Response metadata (status, headers, final URL, request method).
    pub meta: HttpResponseMeta,
    /// Raw backend response until consumed.
    backend: Option<reqwest::Response>,
    /// Cached full body bytes after eager or lazy read.
    buffered_body: Option<Bytes>,
    /// Runtime state inherited from request/client.
    runtime: HttpResponseRuntime,
    /// Decode and error-preview options inherited from client options.
    options: HttpResponseOptions,
}

impl HttpResponse {
    /// Creates a buffered response.
    pub fn new(
        status: StatusCode,
        headers: HeaderMap,
        body: Bytes,
        url: Url,
        method: Method,
    ) -> Self {
        Self {
            meta: HttpResponseMeta::new(status, headers, url.clone(), method),
            backend: None,
            buffered_body: Some(body),
            runtime: HttpResponseRuntime::new(Duration::from_secs(30), None, url),
            options: HttpResponseOptions::default(),
        }
    }

    /// Creates a response from backend response and request-scoped options.
    pub(crate) fn from_backend(
        meta: HttpResponseMeta,
        backend: reqwest::Response,
        read_timeout: Duration,
        cancellation_token: Option<CancellationToken>,
        request_url: Url,
        options: HttpResponseOptions,
    ) -> Self {
        Self {
            meta,
            backend: Some(backend),
            buffered_body: None,
            runtime: HttpResponseRuntime::new(read_timeout, cancellation_token, request_url),
            options,
        }
    }

    /// Returns shared response metadata.
    #[inline]
    pub fn meta(&self) -> &HttpResponseMeta {
        &self.meta
    }

    /// Returns response status code.
    #[inline]
    pub fn status(&self) -> StatusCode {
        self.meta.status
    }

    /// Returns response headers.
    #[inline]
    pub fn headers(&self) -> &HeaderMap {
        &self.meta.headers
    }

    /// Returns final response URL.
    #[inline]
    pub fn url(&self) -> &Url {
        &self.meta.url
    }

    /// Returns request URL used in response read context.
    #[inline]
    pub fn request_url(&self) -> &Url {
        &self.runtime.request_url
    }

    /// Returns whether status is success.
    #[inline]
    pub fn is_success(&self) -> bool {
        self.status().is_success()
    }

    /// Returns parsed `Retry-After` hint when status and headers provide one.
    #[inline]
    pub fn retry_after_hint(&self) -> Option<Duration> {
        self.meta.retry_after_hint()
    }

    /// Returns a previous body read failure, if any.
    ///
    /// # Returns
    /// `Some(HttpError)` when an earlier body read failed or the failure-state
    /// lock is poisoned; otherwise `None`.
    fn previous_body_read_error(&self) -> Option<HttpError> {
        match self.runtime.body_read_failure.lock() {
            Ok(guard) => guard.as_ref().map(BodyReadFailure::to_error),
            Err(_) => Some(
                HttpError::other("Response body read failure state lock poisoned")
                    .with_method(&self.meta.method)
                    .with_url(&self.runtime.request_url),
            ),
        }
    }

    /// Stores the first body read failure for later read attempts.
    ///
    /// # Parameters
    /// - `error`: Error produced while reading the response body.
    fn remember_body_read_failure(&self, error: &HttpError) {
        Self::remember_body_read_failure_state(&self.runtime.body_read_failure, error);
    }

    /// Stores the first body read failure in a shared state holder.
    ///
    /// # Parameters
    /// - `state`: Shared failure state captured by response streams.
    /// - `error`: Error produced while reading the response body.
    fn remember_body_read_failure_state(state: &BodyReadFailureState, error: &HttpError) {
        if let Ok(mut guard) = state.lock() {
            if guard.is_none() {
                *guard = Some(BodyReadFailure::from_error(error));
            }
        }
    }

    /// Returns `Ok(self)` for success statuses, otherwise maps a status error
    /// with `Retry-After` and response-body preview context.
    pub(crate) async fn into_success_or_status_error(
        self,
        message_prefix: &str,
    ) -> HttpResult<Self> {
        let status = self.status();
        if status.is_success() {
            return Ok(self);
        }
        let retry_after = self.retry_after_hint();
        let method = self.meta.method.clone();
        let url = self.request_url().clone();
        let error_preview_limit = self.options.error_response_preview_limit;
        let body_preview = self.into_error_body_preview(error_preview_limit).await?;
        let message = format!(
            "{} with status {} for {} {}; response body preview: {}",
            message_prefix, status, method, url, body_preview
        );
        let mut mapped = HttpError::status(status, message)
            .with_method(&method)
            .with_url(&url)
            .with_response_body_preview(body_preview);
        if let Some(retry_after) = retry_after {
            mapped = mapped.with_retry_after(retry_after);
        }
        Err(mapped)
    }

    /// Consumes this response and returns a bounded body preview for status errors.
    ///
    /// # Errors
    /// Returns [`HttpErrorKind::Cancelled`](crate::HttpErrorKind::Cancelled)
    /// when the request cancellation token fires while preview bytes are being
    /// read.
    pub(crate) async fn into_error_body_preview(mut self, max_bytes: usize) -> HttpResult<String> {
        let limit = max_bytes.max(1);
        if let Some(error) = self.cancelled_error_if_needed(
            "Request cancelled while reading status error response body preview",
        ) {
            return Err(error);
        }
        if let Some(body) = self.buffered_body.take() {
            let end = body.len().min(limit);
            return Ok(Self::render_error_body_preview(
                &body[..end],
                body.len() > limit,
            ));
        }
        let Some(backend) = self.backend.take() else {
            return Ok("<empty>".to_string());
        };
        Self::read_error_body_preview(
            backend,
            self.runtime.read_timeout,
            self.runtime.cancellation_token.clone(),
            self.meta.method.clone(),
            self.runtime.request_url.clone(),
            limit,
        )
        .await
    }

    /// Returns full body bytes, consuming backend stream lazily on first call.
    pub async fn bytes(&mut self) -> HttpResult<Bytes> {
        if let Some(body) = &self.buffered_body {
            return Ok(body.clone());
        }
        if let Some(error) = self.previous_body_read_error() {
            return Err(error);
        }
        let Some(mut backend) = self.backend.take() else {
            self.buffered_body = Some(Bytes::new());
            return Ok(Bytes::new());
        };

        let method = self.meta.method.clone();
        let url = self.runtime.request_url.clone();
        let read_timeout = self.runtime.read_timeout;
        let cancellation_token = self.runtime.cancellation_token.clone();
        let mut body = bytes::BytesMut::new();

        loop {
            let next = if let Some(token) = &cancellation_token {
                tokio::select! {
                    _ = token.cancelled() => {
                        let error = HttpError::cancelled("Request cancelled while reading response body")
                            .with_method(&method)
                            .with_url(&url);
                        self.remember_body_read_failure(&error);
                        return Err(error);
                    }
                    item = tokio::time::timeout(read_timeout, backend.chunk()) => item,
                }
            } else {
                tokio::time::timeout(read_timeout, backend.chunk()).await
            };

            match next {
                Ok(Ok(Some(chunk))) => body.extend_from_slice(&chunk),
                Ok(Ok(None)) => {
                    let body = body.freeze();
                    self.buffered_body = Some(body.clone());
                    return Ok(body);
                }
                Ok(Err(error)) => {
                    let error = map_response_read_error(error, method, url);
                    self.remember_body_read_failure(&error);
                    return Err(error);
                }
                Err(_) => {
                    let error = HttpError::read_timeout(format!(
                        "Read timeout after {:?} while reading response body",
                        read_timeout
                    ))
                    .with_method(&self.meta.method)
                    .with_url(&self.runtime.request_url);
                    self.remember_body_read_failure(&error);
                    return Err(error);
                }
            }
        }
    }

    /// Returns body as stream; if already buffered, returns stream backed by cached bytes.
    pub fn stream(&mut self) -> HttpResult<HttpByteStream> {
        if let Some(body) = self.buffered_body.as_ref() {
            let bytes = body.clone();
            return Ok(Box::pin(futures_stream::once(async move { Ok(bytes) })));
        }
        if let Some(error) = self.previous_body_read_error() {
            return Err(error);
        }
        if let Some(error) = self
            .cancelled_error_if_needed("Streaming response cancelled before reading response body")
        {
            return Err(error);
        }
        let Some(backend) = self.backend.take() else {
            return Ok(Box::pin(futures_stream::empty()));
        };

        let method = self.meta.method.clone();
        let url = self.runtime.request_url.clone();
        let read_timeout = self.runtime.read_timeout;
        let cancellation_token = self.runtime.cancellation_token.clone();
        let body_read_failure = self.runtime.body_read_failure.clone();
        let mut stream = backend.bytes_stream();
        let wrapped = stream! {
            loop {
                let next = if let Some(token) = &cancellation_token {
                    tokio::select! {
                        _ = token.cancelled() => {
                            let error = HttpError::cancelled("Streaming response cancelled while reading body")
                                .with_method(&method)
                                .with_url(&url);
                            Self::remember_body_read_failure_state(&body_read_failure, &error);
                            yield Err(error);
                            break;
                        }
                        item = tokio::time::timeout(read_timeout, stream.next()) => item,
                    }
                } else {
                    tokio::time::timeout(read_timeout, stream.next()).await
                };
                match next {
                    Ok(Some(Ok(bytes))) => yield Ok(bytes),
                    Ok(Some(Err(error))) => {
                        let mapped = map_response_read_error(error, method.clone(), url.clone());
                        Self::remember_body_read_failure_state(&body_read_failure, &mapped);
                        yield Err(mapped);
                        break;
                    }
                    Ok(None) => break,
                    Err(_) => {
                        let error = HttpError::read_timeout(format!(
                            "Read timeout after {:?} while streaming response",
                            read_timeout
                        ))
                        .with_method(&method)
                        .with_url(&url);
                        Self::remember_body_read_failure_state(&body_read_failure, &error);
                        yield Err(error);
                        break;
                    }
                }
            }
        };
        Ok(Box::pin(wrapped))
    }

    /// Interprets response body as UTF-8 text.
    pub async fn text(&mut self) -> HttpResult<String> {
        let body = self.bytes().await?;
        String::from_utf8(body.to_vec()).map_err(|error| {
            HttpError::decode(format!(
                "Failed to decode response body as UTF-8: {}",
                error
            ))
            .with_status(self.meta.status)
            .with_url(&self.meta.url)
        })
    }

    /// Deserializes response body as JSON.
    pub async fn json<T>(&mut self) -> HttpResult<T>
    where
        T: DeserializeOwned,
    {
        let body = self.bytes().await?;
        serde_json::from_slice(&body).map_err(|error| {
            HttpError::decode(format!("Failed to decode response JSON: {}", error))
                .with_status(self.meta.status)
                .with_url(&self.meta.url)
        })
    }

    /// Overrides the maximum allowed size (in bytes) for one SSE line on this response.
    ///
    /// Values below 1 are clamped to 1. Returns `self` so callers can chain configuration
    /// before consuming the body with [`Self::sse_events`] or [`Self::sse_chunks`]
    /// (together with [`Self::sse_json_mode`], [`Self::sse_done_marker_policy`], etc.).
    #[inline]
    pub fn sse_max_line_bytes(mut self, max_line_bytes: usize) -> Self {
        self.options.sse_max_line_bytes = max_line_bytes.max(1);
        self
    }

    /// Overrides the maximum allowed size (in bytes) for one SSE frame on this response.
    ///
    /// Values below 1 are clamped to 1. Returns `self` for chained configuration.
    #[inline]
    pub fn sse_max_frame_bytes(mut self, max_frame_bytes: usize) -> Self {
        self.options.sse_max_frame_bytes = max_frame_bytes.max(1);
        self
    }

    /// Overrides the JSON decoding mode used by [`Self::sse_chunks`] on this response.
    #[inline]
    pub fn sse_json_mode(mut self, mode: SseJsonMode) -> Self {
        self.options.sse_json_mode = mode;
        self
    }

    /// Overrides how [`Self::sse_chunks`] detects end-of-stream from trimmed `data:` payloads.
    #[inline]
    pub fn sse_done_marker_policy(mut self, policy: DoneMarkerPolicy) -> Self {
        self.options.sse_done_marker_policy = policy;
        self
    }

    /// Decodes body stream as SSE events using this response's SSE line/frame byte limits (from
    /// client defaults unless overridden via [`Self::sse_max_line_bytes`] /
    /// [`Self::sse_max_frame_bytes`]).
    pub fn sse_events(mut self) -> SseEventStream {
        let max_line_bytes = self.options.sse_max_line_bytes;
        let max_frame_bytes = self.options.sse_max_frame_bytes;
        match self.stream() {
            Ok(stream) => crate::sse::decode_events_from_stream_with_limits(
                stream,
                max_line_bytes,
                max_frame_bytes,
            ),
            Err(error) => Box::pin(futures_stream::once(async move { Err(error) })),
        }
    }

    /// Decodes SSE `data:` lines as JSON chunks using this response's SSE JSON mode, done-marker
    /// policy, and line/frame limits (see [`Self::sse_json_mode`], [`Self::sse_done_marker_policy`],
    /// [`Self::sse_max_line_bytes`], [`Self::sse_max_frame_bytes`]).
    pub fn sse_chunks<T>(mut self) -> SseChunkStream<T>
    where
        T: DeserializeOwned + Send + 'static,
    {
        let done_policy = self.options.sse_done_marker_policy.clone();
        let mode = self.options.sse_json_mode;
        let max_line_bytes = self.options.sse_max_line_bytes;
        let max_frame_bytes = self.options.sse_max_frame_bytes;
        match self.stream() {
            Ok(stream) => crate::sse::decode_json_chunks_from_stream_with_limits(
                stream,
                done_policy,
                mode,
                max_line_bytes,
                max_frame_bytes,
            ),
            Err(error) => Box::pin(futures_stream::once(async move { Err(error) })),
        }
    }

    /// Returns a buffered body reference for response logging if available.
    ///
    /// # Returns
    /// `Some(&Bytes)` when response body has already been buffered.
    pub(crate) fn buffered_body_for_logging(&self) -> Option<&Bytes> {
        self.buffered_body.as_ref()
    }

    /// Returns whether logger may safely buffer the full body for logging.
    ///
    /// # Parameters
    /// - `body_log_limit`: Configured logging body preview limit in bytes.
    ///
    /// # Returns
    /// `true` only when this response is not SSE, has an explicit `Content-Length`,
    /// and declared length is within `body_log_limit`.
    pub(crate) fn can_buffer_body_for_logging(&self, body_log_limit: usize) -> bool {
        if self.backend.is_none() {
            return false;
        }
        if self.is_sse_response() {
            return false;
        }
        self.content_length_hint()
            .is_some_and(|content_length| content_length <= body_log_limit as u64)
    }

    /// Reads bounded preview bytes from a response body for status error messages.
    ///
    /// # Errors
    /// Returns [`HttpErrorKind::Cancelled`](crate::HttpErrorKind::Cancelled)
    /// when the supplied cancellation token fires while waiting for preview
    /// bytes.
    async fn read_error_body_preview(
        mut response: reqwest::Response,
        read_timeout: Duration,
        cancellation_token: Option<CancellationToken>,
        method: Method,
        url: Url,
        max_bytes: usize,
    ) -> HttpResult<String> {
        let limit = max_bytes.max(1);
        let mut preview = Vec::new();
        let mut truncated = false;

        loop {
            let next = if let Some(token) = cancellation_token.as_ref() {
                tokio::select! {
                    _ = token.cancelled() => {
                        return Err(HttpError::cancelled(
                            "Request cancelled while reading status error response body preview",
                        )
                        .with_method(&method)
                        .with_url(&url));
                    }
                    item = tokio::time::timeout(read_timeout, response.chunk()) => item,
                }
            } else {
                tokio::time::timeout(read_timeout, response.chunk()).await
            };
            match next {
                Ok(Ok(Some(chunk))) => {
                    if preview.len() >= limit {
                        truncated = true;
                        break;
                    }
                    let remaining = limit - preview.len();
                    if chunk.len() > remaining {
                        preview.extend_from_slice(&chunk[..remaining]);
                        truncated = true;
                        break;
                    }
                    preview.extend_from_slice(&chunk);
                }
                Ok(Ok(None)) => break,
                Ok(Err(error)) => {
                    return Ok(format!(
                        "<error body unavailable: failed to read response body: {}>",
                        error
                    ));
                }
                Err(_) => {
                    return Ok(format!(
                        "<error body unavailable: read timeout after {:?}>",
                        read_timeout
                    ));
                }
            }
        }
        Ok(Self::render_error_body_preview(&preview, truncated))
    }

    /// Returns a cancellation error with response read context when cancelled.
    ///
    /// # Parameters
    /// - `message`: Cancellation message to include in the error.
    ///
    /// # Returns
    /// `Some(HttpError)` when this response has a cancelled token; otherwise
    /// `None`.
    fn cancelled_error_if_needed(&self, message: &str) -> Option<HttpError> {
        if self
            .runtime
            .cancellation_token
            .as_ref()
            .is_some_and(CancellationToken::is_cancelled)
        {
            Some(
                HttpError::cancelled(message.to_string())
                    .with_method(&self.meta.method)
                    .with_url(&self.runtime.request_url),
            )
        } else {
            None
        }
    }

    /// Returns `Content-Length` parsed from response headers when present and valid.
    fn content_length_hint(&self) -> Option<u64> {
        self.meta
            .headers
            .get(CONTENT_LENGTH)
            .and_then(|value| value.to_str().ok())
            .and_then(|value| value.parse::<u64>().ok())
    }

    /// Returns whether response content-type is SSE (`text/event-stream`).
    fn is_sse_response(&self) -> bool {
        self.meta
            .headers
            .get(CONTENT_TYPE)
            .and_then(|value| value.to_str().ok())
            .is_some_and(|content_type| {
                content_type
                    .to_ascii_lowercase()
                    .starts_with("text/event-stream")
            })
    }

    fn render_error_body_preview(bytes: &[u8], truncated: bool) -> String {
        if bytes.is_empty() {
            return "<empty>".to_string();
        }
        let suffix = if truncated { "...<truncated>" } else { "" };
        match std::str::from_utf8(bytes) {
            Ok(text) => format!("{text}{suffix}"),
            Err(_) => format!("<binary {} bytes>{suffix}", bytes.len()),
        }
    }
}