linger-openai-sdk 0.1.1

use crate::error::{HeaderMap, IntoHeaderPair, LingerError};
use bytes::Bytes;
use futures_core::Stream;
use futures_util::StreamExt;
use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;

/// EN: Boxed byte stream used by response bodies.
/// 中文：响应体使用的装箱字节流。
pub type BodyStream = Pin<Box<dyn Stream<Item = Result<Bytes, LingerError>> + Send>>;

/// EN: Runtime-neutral HTTP request body.
/// 中文：运行时无关的 HTTP 请求体。
pub enum HttpRequestBody {
    /// EN: In-memory byte body.
    /// 中文：内存字节请求体。
    Bytes(Bytes),
    /// EN: Incremental byte stream body.
    /// 中文：增量字节流请求体。
    Stream(BodyStream),
}

impl HttpRequestBody {
    /// EN: Returns true when this body is a stream.
    /// 中文：当请求体为流时返回 true。
    pub fn is_stream(&self) -> bool {
        matches!(self, Self::Stream(_))
    }

    /// EN: Returns byte body contents without buffering streams.
    /// 中文：返回字节请求体内容，但不会缓冲流式请求体。
    pub fn as_bytes(&self) -> Option<&[u8]> {
        match self {
            Self::Bytes(bytes) => Some(bytes),
            Self::Stream(_) => None,
        }
    }

    /// EN: Converts the body into an incremental byte stream.
    /// 中文：将请求体转换为增量字节流。
    pub fn into_stream(self) -> BodyStream {
        match self {
            Self::Bytes(bytes) => {
                Box::pin(futures_util::stream::once(async move { Ok(bytes) })) as BodyStream
            }
            Self::Stream(stream) => stream,
        }
    }
}

impl fmt::Debug for HttpRequestBody {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Bytes(bytes) => write!(f, "<{} bytes>", bytes.len()),
            Self::Stream(_) => f.write_str("<stream>"),
        }
    }
}

/// EN: HTTP method supported by SDK requests.
/// 中文：SDK 请求支持的 HTTP 方法。
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum HttpMethod {
    /// EN: HTTP GET.
    /// 中文：HTTP GET。
    Get,
    /// EN: HTTP POST.
    /// 中文：HTTP POST。
    Post,
    /// EN: HTTP DELETE.
    /// 中文：HTTP DELETE。
    Delete,
}

/// EN: Runtime-neutral HTTP request passed to transports.
/// 中文：传给传输层的运行时无关 HTTP 请求。
pub struct HttpRequest {
    method: HttpMethod,
    url: String,
    path: String,
    headers: HeaderMap,
    body: Option<HttpRequestBody>,
}

impl HttpRequest {
    /// EN: Creates an SDK HTTP request.
    /// 中文：创建 SDK HTTP 请求。
    pub fn new(method: HttpMethod, base_url: impl AsRef<str>, path: impl Into<String>) -> Self {
        let path = path.into();
        let url = format!("{}{}", base_url.as_ref().trim_end_matches('/'), path);
        Self {
            method,
            url,
            path,
            headers: HeaderMap::new(),
            body: None,
        }
    }

    /// EN: Returns the HTTP method.
    /// 中文：返回 HTTP 方法。
    pub fn method(&self) -> HttpMethod {
        self.method
    }

    /// EN: Returns the full URL.
    /// 中文：返回完整 URL。
    pub fn url(&self) -> &str {
        &self.url
    }

    /// EN: Returns the API path.
    /// 中文：返回 API 路径。
    pub fn path(&self) -> &str {
        &self.path
    }

    /// EN: Adds or replaces a header.
    /// 中文：添加或替换请求头。
    pub fn insert_header(&mut self, name: impl Into<String>, value: impl Into<String>) {
        self.headers.insert(name, value);
    }

    /// EN: Returns a header value by case-insensitive name.
    /// 中文：按大小写不敏感名称返回请求头值。
    pub fn header(&self, name: &str) -> Option<&str> {
        self.headers.get(name)
    }

    /// EN: Returns all request headers.
    /// 中文：返回所有请求头。
    pub fn headers(&self) -> &HeaderMap {
        &self.headers
    }

    /// EN: Sets the request body.
    /// 中文：设置请求体。
    pub fn set_body(&mut self, body: impl Into<Bytes>) {
        self.body = Some(HttpRequestBody::Bytes(body.into()));
    }

    /// EN: Sets an incremental request body stream.
    /// 中文：设置增量请求体流。
    pub fn set_body_stream<S>(&mut self, body: S)
    where
        S: Stream<Item = Result<Bytes, LingerError>> + Send + 'static,
    {
        self.body = Some(HttpRequestBody::Stream(Box::pin(body)));
    }

    /// EN: Returns the request body bytes, when present.
    /// 中文：返回请求体字节，如存在。
    pub fn body(&self) -> Option<&[u8]> {
        self.body.as_ref().and_then(HttpRequestBody::as_bytes)
    }

    /// EN: Returns true when the request body is streamed.
    /// 中文：当请求体为流式传输时返回 true。
    pub fn body_is_stream(&self) -> bool {
        self.body.as_ref().is_some_and(HttpRequestBody::is_stream)
    }

    /// EN: Consumes the request and returns its body.
    /// 中文：消耗请求并返回请求体。
    pub fn into_body(self) -> Option<HttpRequestBody> {
        self.body
    }
}

impl fmt::Debug for HttpRequest {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("HttpRequest")
            .field("method", &self.method)
            .field("url", &self.url)
            .field("path", &self.path)
            .field("headers", &self.headers)
            .field("body", &self.body.as_ref())
            .finish()
    }
}

enum HttpBody {
    Bytes(Bytes),
    Stream(BodyStream),
}

/// EN: Runtime-neutral HTTP response returned by transports.
/// 中文：传输层返回的运行时无关 HTTP 响应。
pub struct HttpResponse {
    status: u16,
    headers: HeaderMap,
    body: HttpBody,
}

impl HttpResponse {
    /// EN: Creates a response with an in-memory byte body.
    /// 中文：创建包含内存字节响应体的响应。
    pub fn from_bytes<I, P>(status: u16, headers: I, body: impl Into<Bytes>) -> Self
    where
        I: IntoIterator<Item = P>,
        P: IntoHeaderPair,
    {
        Self {
            status,
            headers: HeaderMap::from_pairs(headers),
            body: HttpBody::Bytes(body.into()),
        }
    }

    /// EN: Creates a response with an incremental byte stream body.
    /// 中文：创建包含增量字节流响应体的响应。
    pub fn from_stream<I, P, S>(status: u16, headers: I, body: S) -> Self
    where
        I: IntoIterator<Item = P>,
        P: IntoHeaderPair,
        S: Stream<Item = Result<Bytes, LingerError>> + Send + 'static,
    {
        Self {
            status,
            headers: HeaderMap::from_pairs(headers),
            body: HttpBody::Stream(Box::pin(body)),
        }
    }

    /// EN: Returns the HTTP status.
    /// 中文：返回 HTTP 状态码。
    pub fn status(&self) -> u16 {
        self.status
    }

    /// EN: Returns response headers.
    /// 中文：返回响应头。
    pub fn headers(&self) -> &HeaderMap {
        &self.headers
    }

    /// EN: Consumes the response and returns headers.
    /// 中文：消耗响应并返回响应头。
    pub fn into_parts(self) -> (u16, HeaderMap, BodyStream) {
        let body = match self.body {
            HttpBody::Bytes(bytes) => {
                Box::pin(futures_util::stream::once(async move { Ok(bytes) })) as BodyStream
            }
            HttpBody::Stream(stream) => stream,
        };
        (self.status, self.headers, body)
    }

    /// EN: Consumes the response and returns an incremental body stream.
    /// 中文：消耗响应并返回增量响应体流。
    pub fn into_body_stream(self) -> BodyStream {
        self.into_parts().2
    }

    /// EN: Consumes the response and collects the full body.
    /// 中文：消耗响应并收集完整响应体。
    pub async fn into_bytes(self) -> Result<(u16, HeaderMap, Bytes), LingerError> {
        let (status, headers, mut stream) = self.into_parts();
        let mut body = Vec::new();
        while let Some(chunk) = stream.next().await {
            body.extend_from_slice(&chunk?);
        }
        Ok((status, headers, Bytes::from(body)))
    }
}

/// EN: Runtime-neutral HTTP transport boundary.
/// 中文：运行时无关的 HTTP 传输边界。
pub trait Transport: Send + Sync {
    /// EN: Sends a request and asynchronously returns a response.
    /// 中文：发送请求并异步返回响应。
    fn send(
        &self,
        request: HttpRequest,
    ) -> Pin<Box<dyn Future<Output = Result<HttpResponse, LingerError>> + Send + '_>>;
}

/// EN: Cloneable shared transport handle.
/// 中文：可克隆的共享传输句柄。
#[derive(Clone)]
pub struct SharedTransport {
    inner: Arc<dyn Transport>,
}

impl SharedTransport {
    /// EN: Wraps a concrete transport.
    /// 中文：包装具体传输实现。
    pub fn new<T>(transport: T) -> Self
    where
        T: Transport + 'static,
    {
        Self {
            inner: Arc::new(transport),
        }
    }

    /// EN: Sends a request through the shared transport.
    /// 中文：通过共享传输发送请求。
    pub fn send(
        &self,
        request: HttpRequest,
    ) -> Pin<Box<dyn Future<Output = Result<HttpResponse, LingerError>> + Send + '_>> {
        self.inner.send(request)
    }
}

impl fmt::Debug for SharedTransport {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SharedTransport").finish_non_exhaustive()
    }
}

/// EN: Reqwest-backed default HTTP transport.
/// 中文：基于 reqwest 的默认 HTTP 传输实现。
#[cfg(feature = "reqwest-transport")]
#[derive(Clone, Debug)]
pub struct ReqwestTransport {
    client: reqwest::Client,
}

#[cfg(feature = "reqwest-transport")]
impl Default for ReqwestTransport {
    fn default() -> Self {
        Self {
            client: reqwest::Client::new(),
        }
    }
}

#[cfg(feature = "reqwest-transport")]
impl ReqwestTransport {
    /// EN: Creates a reqwest transport from an existing client.
    /// 中文：通过已有 reqwest 客户端创建传输实现。
    pub fn new(client: reqwest::Client) -> Self {
        Self { client }
    }
}

#[cfg(feature = "reqwest-transport")]
impl Transport for ReqwestTransport {
    fn send(
        &self,
        request: HttpRequest,
    ) -> Pin<Box<dyn Future<Output = Result<HttpResponse, LingerError>> + Send + '_>> {
        Box::pin(async move {
            let method = match request.method {
                HttpMethod::Get => reqwest::Method::GET,
                HttpMethod::Post => reqwest::Method::POST,
                HttpMethod::Delete => reqwest::Method::DELETE,
            };
            let mut builder = self.client.request(method, request.url);
            for (name, value) in request.headers.iter() {
                builder = builder.header(name, value);
            }
            if let Some(body) = request.body {
                builder = match body {
                    HttpRequestBody::Bytes(bytes) => builder.body(bytes),
                    HttpRequestBody::Stream(stream) => {
                        builder.body(reqwest::Body::wrap_stream(stream))
                    }
                };
            }
            let response = builder
                .send()
                .await
                .map_err(|error| LingerError::transport(error.to_string()))?;
            let status = response.status().as_u16();
            let headers =
                HeaderMap::from_pairs(response.headers().iter().filter_map(|(name, value)| {
                    value.to_str().ok().map(|value| (name.as_str(), value))
                }));
            let stream = response
                .bytes_stream()
                .map(|chunk| chunk.map_err(|error| LingerError::transport(error.to_string())));
            Ok(HttpResponse::from_stream(status, headers.iter(), stream))
        })
    }
}