noema-actix-webapi 0.1.0

use std::sync::Arc;
use std::sync::OnceLock;
use std::time::Duration;

use noema::core::{Container, Injectable};
use serde::Serialize;
use serde::de::DeserializeOwned;
use tokio::sync::Semaphore;

use crate::config::HttpClientConfig;

static SETTINGS: OnceLock<HttpClientConfig> = OnceLock::new();

pub(crate) fn install(cfg: HttpClientConfig) {
    let _ = SETTINGS.set(cfg);
}

fn settings() -> HttpClientConfig {
    SETTINGS.get().cloned().unwrap_or_default()
}

/// Outbound HTTP. Default impl is one process-wide pooled `reqwest::Client`.
#[async_trait::async_trait]
pub trait HttpClient: Send + Sync {
    async fn send(&self, request: OutboundRequest) -> Result<OutboundResponse, HttpClientError>;
}

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

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

#[derive(Debug, Clone)]
pub struct OutboundRequest {
    pub method: Method,
    pub url: String,
    pub headers: Vec<(String, String)>,
    pub body: Option<Vec<u8>>,
}

impl OutboundRequest {
    pub fn get(url: impl Into<String>) -> Self {
        Self::new(Method::Get, url)
    }

    pub fn post(url: impl Into<String>) -> Self {
        Self::new(Method::Post, url)
    }

    pub fn put(url: impl Into<String>) -> Self {
        Self::new(Method::Put, url)
    }

    pub fn patch(url: impl Into<String>) -> Self {
        Self::new(Method::Patch, url)
    }

    pub fn delete(url: impl Into<String>) -> Self {
        Self::new(Method::Delete, url)
    }

    pub fn head(url: impl Into<String>) -> Self {
        Self::new(Method::Head, url)
    }

    fn new(method: Method, url: impl Into<String>) -> Self {
        Self {
            method,
            url: url.into(),
            headers: Vec::new(),
            body: None,
        }
    }

    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
        self.headers.push((name.into(), value.into()));
        self
    }

    pub fn json<T: Serialize>(mut self, value: &T) -> Result<Self, HttpClientError> {
        self.body = Some(serde_json::to_vec(value).map_err(HttpClientError::json)?);
        self.headers
            .push(("Content-Type".into(), "application/json".into()));
        Ok(self)
    }

    pub fn body(mut self, bytes: impl Into<Vec<u8>>) -> Self {
        self.body = Some(bytes.into());
        self
    }
}

#[derive(Debug, Clone)]
pub struct OutboundResponse {
    status: u16,
    headers: Vec<(String, String)>,
    body: Vec<u8>,
}

impl OutboundResponse {
    pub fn new(status: u16, headers: Vec<(String, String)>, body: Vec<u8>) -> Self {
        Self {
            status,
            headers,
            body,
        }
    }

    pub fn status(&self) -> u16 {
        self.status
    }

    pub fn headers(&self) -> &[(String, String)] {
        &self.headers
    }

    pub fn body(&self) -> &[u8] {
        &self.body
    }

    pub fn is_success(&self) -> bool {
        (200..300).contains(&self.status)
    }

    pub fn text(&self) -> Result<String, HttpClientError> {
        String::from_utf8(self.body.clone()).map_err(HttpClientError::body)
    }

    pub fn json<T: DeserializeOwned>(&self) -> Result<T, HttpClientError> {
        serde_json::from_slice(&self.body).map_err(HttpClientError::json)
    }
}

#[derive(Debug)]
pub struct HttpClientError {
    message: String,
}

impl HttpClientError {
    fn request(err: reqwest::Error) -> Self {
        Self {
            message: err.to_string(),
        }
    }

    fn json(err: serde_json::Error) -> Self {
        Self {
            message: err.to_string(),
        }
    }

    fn body(err: std::string::FromUtf8Error) -> Self {
        Self {
            message: err.to_string(),
        }
    }

    fn header(name: &str, err: reqwest::header::InvalidHeaderName) -> Self {
        Self {
            message: format!("{name}: {err}"),
        }
    }

    fn header_value(name: &str, err: reqwest::header::InvalidHeaderValue) -> Self {
        Self {
            message: format!("{name}: {err}"),
        }
    }

    fn busy() -> Self {
        Self {
            message: "outbound http at capacity".into(),
        }
    }

    pub fn is_busy(&self) -> bool {
        self.message == "outbound http at capacity"
    }
}

impl std::fmt::Display for HttpClientError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.message)
    }
}

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

/// Default [`HttpClient`]. One `reqwest::Client` (pool per host) plus an in-flight cap.
/// Fail-fast when the cap is full. Does not retry or rebuild the client.
#[derive(Clone)]
pub struct ReqwestHttpClient {
    inner: reqwest::Client,
    in_flight: Option<Arc<Semaphore>>,
}

impl ReqwestHttpClient {
    pub fn from_config(cfg: &HttpClientConfig) -> Self {
        let mut builder = reqwest::Client::builder()
            .timeout(Duration::from_millis(u64::from(cfg.timeout_ms)))
            .connect_timeout(Duration::from_millis(u64::from(cfg.connect_timeout_ms)))
            .pool_max_idle_per_host(cfg.pool_max_idle_per_host as usize)
            .user_agent(&cfg.user_agent);
        if let Some(secs) = cfg.pool_idle_timeout_secs {
            builder = builder.pool_idle_timeout(Duration::from_secs(u64::from(secs)));
        }
        let in_flight = if cfg.max_in_flight == 0 {
            None
        } else {
            Some(Arc::new(Semaphore::new(cfg.max_in_flight as usize)))
        };
        Self {
            inner: builder.build().expect("reqwest client"),
            in_flight,
        }
    }
}

#[async_trait::async_trait]
impl HttpClient for ReqwestHttpClient {
    async fn send(&self, request: OutboundRequest) -> Result<OutboundResponse, HttpClientError> {
        let _permit = if let Some(sem) = &self.in_flight {
            Some(sem.try_acquire().map_err(|_| HttpClientError::busy())?)
        } else {
            None
        };
        let mut req = self.inner.request(request.method.reqwest(), &request.url);
        for (name, value) in &request.headers {
            let header_name = reqwest::header::HeaderName::from_bytes(name.as_bytes())
                .map_err(|e| HttpClientError::header(name, e))?;
            let header_value = reqwest::header::HeaderValue::from_str(value)
                .map_err(|e| HttpClientError::header_value(name, e))?;
            req = req.header(header_name, header_value);
        }
        if let Some(body) = request.body {
            req = req.body(body);
        }
        let response = req.send().await.map_err(HttpClientError::request)?;
        let status = response.status().as_u16();
        let headers = response
            .headers()
            .iter()
            .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
            .collect();
        let body = response.bytes().await.map_err(HttpClientError::request)?;
        Ok(OutboundResponse::new(status, headers, body.to_vec()))
    }
}

impl Injectable<Container> for ReqwestHttpClient {
    fn inject(_: &Container) -> Self {
        Self::from_config(&settings())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::TcpListener;

    struct Fake {
        status: u16,
        body: Vec<u8>,
    }

    #[async_trait::async_trait]
    impl HttpClient for Fake {
        async fn send(
            &self,
            _request: OutboundRequest,
        ) -> Result<OutboundResponse, HttpClientError> {
            Ok(OutboundResponse::new(
                self.status,
                Vec::new(),
                self.body.clone(),
            ))
        }
    }

    #[test]
    fn handler_accepts_injected_http_client() {
        let client: Arc<dyn HttpClient + Send + Sync> = Arc::new(Fake {
            status: 200,
            body: b"ok".to_vec(),
        });
        let _ = client;
    }

    #[test]
    fn http_client_resolve_is_singleton() {
        let a = noema::resolve::<dyn HttpClient + Send + Sync>();
        let b = noema::resolve::<dyn HttpClient + Send + Sync>();
        assert!(std::sync::Arc::ptr_eq(&a, &b));
    }

    async fn serve_pong() -> String {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            loop {
                let Ok((mut socket, _)) = listener.accept().await else {
                    break;
                };
                tokio::spawn(async move {
                    let mut buf = vec![0u8; 4096];
                    loop {
                        match socket.read(&mut buf).await {
                            Ok(0) | Err(_) => break,
                            Ok(_) => {
                                let _ = socket
                                    .write_all(
                                        b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: keep-alive\r\n\r\npong",
                                    )
                                    .await;
                            }
                        }
                    }
                });
            }
        });
        format!("http://{addr}/ping")
    }

    #[actix_web::test]
    async fn get_hits_local_server() {
        let url = serve_pong().await;
        let client = ReqwestHttpClient::from_config(&HttpClientConfig::default());
        let first = client.send(OutboundRequest::get(&url)).await.unwrap();
        assert_eq!(first.status(), 200);
        assert_eq!(first.text().unwrap(), "pong");
        let second = client.send(OutboundRequest::get(&url)).await.unwrap();
        assert_eq!(second.status(), 200);
        assert_eq!(second.body(), b"pong");
    }

    #[test]
    fn json_roundtrip_on_response() {
        let resp = OutboundResponse::new(201, Vec::new(), br#"{"n":1}"#.to_vec());
        let v: serde_json::Value = resp.json().unwrap();
        assert_eq!(v["n"], 1);
        assert!(resp.is_success());
    }

    #[actix_web::test]
    async fn send_fails_fast_when_in_flight_full() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let entered = Arc::new(tokio::sync::Notify::new());
        let release = Arc::new(tokio::sync::Notify::new());
        let entered_s = Arc::clone(&entered);
        let release_s = Arc::clone(&release);
        tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            let mut buf = vec![0u8; 4096];
            let _ = socket.read(&mut buf).await;
            entered_s.notify_one();
            release_s.notified().await;
            let _ = socket
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
                .await;
        });
        let url = format!("http://{addr}/slow");
        let cfg = HttpClientConfig {
            max_in_flight: 1,
            timeout_ms: 5_000,
            ..HttpClientConfig::default()
        };
        let client = ReqwestHttpClient::from_config(&cfg);
        let waiting = client.clone();
        let url_first = url.clone();
        let first =
            actix_web::rt::spawn(
                async move { waiting.send(OutboundRequest::get(url_first)).await },
            );
        entered.notified().await;
        let err = client.send(OutboundRequest::get(&url)).await.unwrap_err();
        assert!(err.is_busy(), "{err}");
        release.notify_one();
        first.await.unwrap().unwrap();
    }
}