fastapi-core 0.3.0

//! HTTP request types.

use std::any::{Any, TypeId};
use std::collections::HashMap;
use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::sync::Mutex;

// Re-export Method from fastapi-types
pub use fastapi_types::Method;

use asupersync::stream::Stream;

/// Error yielded by streaming request bodies.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RequestBodyStreamError {
    /// The body exceeded the configured maximum size.
    TooLarge { received: usize, max: usize },
    /// The connection closed before the full body was read.
    ConnectionClosed,
    /// An I/O error occurred while reading the body.
    Io(String),
}

impl fmt::Display for RequestBodyStreamError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::TooLarge { received, max } => write!(
                f,
                "request body too large: received {received} bytes (max {max})"
            ),
            Self::ConnectionClosed => write!(f, "connection closed while reading request body"),
            Self::Io(e) => write!(f, "I/O error while reading request body: {e}"),
        }
    }
}

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

/// Streamed request body type (yields chunks or a streaming error).
pub type RequestBodyStream =
    Pin<Box<dyn Stream<Item = Result<Vec<u8>, RequestBodyStreamError>> + Send>>;

/// HTTP protocol version.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum HttpVersion {
    /// HTTP/1.0
    Http10,
    /// HTTP/1.1
    Http11,
    /// HTTP/2
    Http2,
}

impl HttpVersion {
    /// Parse an HTTP version string like "HTTP/1.1".
    #[must_use]
    pub fn parse(s: &str) -> Option<Self> {
        match s {
            "HTTP/1.0" => Some(Self::Http10),
            "HTTP/1.1" => Some(Self::Http11),
            "HTTP/2" | "HTTP/2.0" => Some(Self::Http2),
            _ => None,
        }
    }
}

/// Connection metadata supplied by the server (or test harness).
///
/// The core request type does not inherently know whether it arrived over TLS;
/// servers should insert this as an extension when that information is available.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ConnectionInfo {
    /// True if the connection is using TLS (HTTPS).
    pub is_tls: bool,
}

impl ConnectionInfo {
    /// Plain HTTP connection.
    #[allow(dead_code)]
    pub const HTTP: Self = Self { is_tls: false };
    /// HTTPS connection (TLS).
    #[allow(dead_code)]
    pub const HTTPS: Self = Self { is_tls: true };
}

/// HTTP headers collection.
#[derive(Debug, Default)]
pub struct Headers {
    inner: HashMap<String, Vec<u8>>,
}

impl Headers {
    /// Create empty headers.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Get a header value by name (case-insensitive).
    #[must_use]
    pub fn get(&self, name: &str) -> Option<&[u8]> {
        self.inner
            .get(&name.to_ascii_lowercase())
            .map(Vec::as_slice)
    }

    /// Returns true if a header is present (case-insensitive).
    #[must_use]
    pub fn contains(&self, name: &str) -> bool {
        self.inner.contains_key(&name.to_ascii_lowercase())
    }

    /// Insert a header.
    pub fn insert(&mut self, name: impl Into<String>, value: impl Into<Vec<u8>>) {
        self.inner
            .insert(name.into().to_ascii_lowercase(), value.into());
    }

    /// Insert a header from borrowed name/value slices.
    ///
    /// This is a convenience for parsers that already have `&str`/`&[u8]` and want
    /// to avoid constructing intermediate owned buffers.
    pub fn insert_from_slice(&mut self, name: &str, value: &[u8]) {
        self.inner.insert(name.to_ascii_lowercase(), value.to_vec());
    }

    /// Remove a header value by name (case-insensitive).
    pub fn remove(&mut self, name: &str) -> Option<Vec<u8>> {
        self.inner.remove(&name.to_ascii_lowercase())
    }

    /// Iterate over all headers as (name, value) pairs.
    pub fn iter(&self) -> impl Iterator<Item = (&str, &[u8])> {
        self.inner
            .iter()
            .map(|(name, value)| (name.as_str(), value.as_slice()))
    }

    /// Returns the number of headers.
    #[must_use]
    pub fn len(&self) -> usize {
        self.inner.len()
    }

    /// Returns true if there are no headers.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }
}

/// Request body.
pub enum Body {
    /// Empty body.
    Empty,
    /// Bytes body.
    Bytes(Vec<u8>),
    /// Streaming body, optionally with a known content length.
    Stream {
        /// Streamed chunks.
        stream: Mutex<RequestBodyStream>,
        /// Known content length, if available.
        content_length: Option<usize>,
    },
}

impl fmt::Debug for Body {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Empty => f.debug_tuple("Empty").finish(),
            Self::Bytes(b) => f.debug_tuple("Bytes").field(b).finish(),
            Self::Stream { content_length, .. } => f
                .debug_struct("Stream")
                .field("content_length", content_length)
                .finish(),
        }
    }
}

impl Body {
    /// Create a streaming body.
    #[must_use]
    pub fn streaming<S>(stream: S) -> Self
    where
        S: Stream<Item = Result<Vec<u8>, RequestBodyStreamError>> + Send + 'static,
    {
        Self::Stream {
            stream: Mutex::new(Box::pin(stream)),
            content_length: None,
        }
    }

    /// Create a streaming body with a known content length.
    #[must_use]
    pub fn streaming_with_size<S>(stream: S, content_length: usize) -> Self
    where
        S: Stream<Item = Result<Vec<u8>, RequestBodyStreamError>> + Send + 'static,
    {
        Self::Stream {
            stream: Mutex::new(Box::pin(stream)),
            content_length: Some(content_length),
        }
    }

    /// Get body as bytes, consuming it.
    ///
    /// Note: streaming bodies cannot be synchronously collected; this returns
    /// an empty vector for `Body::Stream`.
    #[must_use]
    pub fn into_bytes(self) -> Vec<u8> {
        match self {
            Self::Empty => Vec::new(),
            Self::Bytes(b) => b,
            Self::Stream { .. } => Vec::new(),
        }
    }

    /// Check if body is empty.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        matches!(self, Self::Empty)
            || matches!(self, Self::Bytes(b) if b.is_empty())
            || matches!(
                self,
                Self::Stream {
                    content_length: Some(0),
                    ..
                }
            )
    }

    /// Take ownership of the inner stream, if this is a streaming body.
    pub fn into_stream(self) -> Option<(RequestBodyStream, Option<usize>)> {
        match self {
            Self::Stream {
                stream,
                content_length,
            } => Some((
                stream.into_inner().unwrap_or_else(|e| e.into_inner()),
                content_length,
            )),
            _ => None,
        }
    }
}

/// Request-scoped background tasks to execute after the response is sent.
///
/// This is inspired by FastAPI's `BackgroundTasks`. Handlers can enqueue work
/// that is executed by the server after the main response completes.
pub type BackgroundTasksInner = Mutex<Vec<Pin<Box<dyn Future<Output = ()> + Send>>>>;

pub struct BackgroundTasks {
    tasks: BackgroundTasksInner,
}

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

impl BackgroundTasks {
    /// Create an empty background task set.
    #[must_use]
    pub fn new() -> Self {
        Self {
            tasks: Mutex::new(Vec::new()),
        }
    }

    /// Add a synchronous task to run after the response is written.
    ///
    /// This matches FastAPI's `BackgroundTasks.add_task(...)` UX: you enqueue work
    /// and the server runs it after the response is sent.
    pub fn add<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        self.add_async(async move { f() });
    }

    /// Add an async task (future) to run after the response is written.
    pub fn add_async<Fut>(&self, fut: Fut)
    where
        Fut: Future<Output = ()> + Send + 'static,
    {
        let mut guard = self
            .tasks
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        guard.push(Box::pin(fut));
    }

    /// Execute all background tasks sequentially.
    pub async fn execute_all(self) {
        let tasks = self
            .tasks
            .into_inner()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        for t in tasks {
            t.await;
        }
    }
}

impl Default for BackgroundTasks {
    fn default() -> Self {
        Self::new()
    }
}

/// HTTP request.
#[derive(Debug)]
pub struct Request {
    method: Method,
    version: HttpVersion,
    path: String,
    query: Option<String>,
    headers: Headers,
    body: Body,
    // Extensions for middleware/extractors
    #[allow(dead_code)] // Used in future implementation
    extensions: HashMap<std::any::TypeId, Box<dyn std::any::Any + Send + Sync>>,
}

impl Request {
    /// Create a new request.
    #[must_use]
    pub fn new(method: Method, path: impl Into<String>) -> Self {
        Self {
            method,
            version: HttpVersion::Http11,
            path: path.into(),
            query: None,
            headers: Headers::new(),
            body: Body::Empty,
            extensions: HashMap::new(),
        }
    }

    /// Create a new request with an explicit HTTP version.
    #[must_use]
    pub fn with_version(method: Method, path: impl Into<String>, version: HttpVersion) -> Self {
        let mut req = Self::new(method, path);
        req.version = version;
        req
    }

    /// Get the HTTP method.
    #[must_use]
    pub fn method(&self) -> Method {
        self.method
    }

    /// Get the HTTP version.
    #[must_use]
    pub fn version(&self) -> HttpVersion {
        self.version
    }

    /// Set the HTTP version.
    pub fn set_version(&mut self, version: HttpVersion) {
        self.version = version;
    }

    /// Get the request path.
    #[must_use]
    pub fn path(&self) -> &str {
        &self.path
    }

    /// Get the query string.
    #[must_use]
    pub fn query(&self) -> Option<&str> {
        self.query.as_deref()
    }

    /// Get the headers.
    #[must_use]
    pub fn headers(&self) -> &Headers {
        &self.headers
    }

    /// Get mutable headers.
    pub fn headers_mut(&mut self) -> &mut Headers {
        &mut self.headers
    }

    /// Get the body.
    #[must_use]
    pub fn body(&self) -> &Body {
        &self.body
    }

    /// Take the body, replacing with Empty.
    pub fn take_body(&mut self) -> Body {
        std::mem::replace(&mut self.body, Body::Empty)
    }

    /// Set the body.
    pub fn set_body(&mut self, body: Body) {
        self.body = body;
    }

    /// Set the query string.
    pub fn set_query(&mut self, query: Option<String>) {
        self.query = query;
    }

    /// Insert a typed extension value.
    pub fn insert_extension<T: Any + Send + Sync>(&mut self, value: T) {
        self.extensions.insert(TypeId::of::<T>(), Box::new(value));
    }

    /// Get a typed extension value.
    #[must_use]
    pub fn get_extension<T: Any + Send + Sync>(&self) -> Option<&T> {
        self.extensions
            .get(&TypeId::of::<T>())
            .and_then(|boxed| boxed.downcast_ref::<T>())
    }

    /// Get a mutable typed extension value.
    pub fn get_extension_mut<T: Any + Send + Sync>(&mut self) -> Option<&mut T> {
        self.extensions
            .get_mut(&TypeId::of::<T>())
            .and_then(|boxed| boxed.downcast_mut::<T>())
    }

    /// Remove and return a typed extension value.
    pub fn take_extension<T: Any + Send + Sync>(&mut self) -> Option<T> {
        self.extensions
            .remove(&TypeId::of::<T>())
            .and_then(|boxed| boxed.downcast::<T>().ok())
            .map(|boxed| *boxed)
    }

    /// Access (and lazily create) the request-scoped background tasks container.
    pub fn background_tasks(&mut self) -> &BackgroundTasks {
        if !self
            .extensions
            .contains_key(&TypeId::of::<BackgroundTasks>())
        {
            self.insert_extension(BackgroundTasks::new());
        }
        self.get_extension::<BackgroundTasks>()
            .expect("BackgroundTasks extension should exist")
    }
}