apollo-http-client 0.3.0

use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::time::Instant;
use tokio_util::sync::CancellationToken;

use bytes::Bytes;
use http_body_util::combinators::BoxBody;
use hyper::body::Incoming;
use tower::BoxError;

pub(crate) use apollo_http_shared::method::normalize_method;
pub(crate) use apollo_http_shared::protocol::{ProtocolVersionCell, version_str};

use crate::error::HttpClientError;

pub(crate) mod alpn;
pub(crate) mod h1;
pub(crate) mod h2;
pub(crate) mod handshake;
#[cfg(unix)]
pub(crate) mod unix;

/// Identifies a connection pool by the destination it serves.
#[derive(Clone, Debug, Hash, Eq, PartialEq)]
pub(crate) struct Origin {
    pub(crate) scheme: http::uri::Scheme,
    pub(crate) authority: http::uri::Authority,
}

impl Origin {
    /// `server.address` for telemetry: the URI host for TCP connections, or the
    /// decoded socket path for `unix://`. Falls back to [`INVALID_SOCKET_PATH`]
    /// when the authority cannot be decoded or Unix sockets are not supported.
    pub(crate) fn address(&self) -> String {
        if self.scheme.as_str() == "unix" {
            #[cfg(unix)]
            return crate::protocol::unix::authority_to_str(&self.authority)
                .unwrap_or_else(|_| INVALID_SOCKET_PATH.to_owned());
            #[cfg(not(unix))]
            return INVALID_SOCKET_PATH.to_owned();
        }
        self.authority.host().to_owned()
    }

    /// `server.port` for telemetry: the URI port, defaulting to 443 for HTTPS
    /// and 80 for HTTP. Returns `None` for `unix://` connections.
    pub(crate) fn port(&self) -> Option<i64> {
        if self.scheme.as_str() == "unix" {
            return None;
        }
        let port = self.authority.port_u16().map(i64::from).unwrap_or_else(|| {
            if self.scheme == http::uri::Scheme::HTTPS {
                443
            } else {
                80
            }
        });
        Some(port)
    }
}

/// Placeholder recorded as `server.address` when a `unix://` socket path cannot
/// be decoded from the URI authority — either because the hex is malformed or
/// because the build has no Unix decoder compiled in.
pub(crate) const INVALID_SOCKET_PATH: &str = "<invalid socket path>";

/// HTTP request body type for [`crate::HttpClient`].
///
/// Use `http_body_util` to construct bodies and call `.map_err(…).boxed()` to convert:
///
/// ```rust
/// use apollo_http_client::HttpBody;
/// use bytes::Bytes;
/// use http_body_util::{BodyExt, Empty, Full};
///
/// // Empty body (e.g. for GET requests)
/// let _: HttpBody = Empty::<Bytes>::new()
///     .map_err(|e: std::convert::Infallible| match e {})
///     .boxed();
///
/// // Fixed-size body (e.g. for POST)
/// let _: HttpBody = Full::new(Bytes::from_static(b"hello"))
///     .map_err(|e: std::convert::Infallible| match e {})
///     .boxed();
/// ```
pub type HttpBody = BoxBody<Bytes, BoxError>;

// ---- Host sender types ------------------------------------------------------

/// The future returned when a [`HostSender`] dispatches a request.
pub(crate) type SendFuture = Pin<
    Box<dyn Future<Output = Result<http::Response<Incoming>, HttpClientError>> + Send + 'static>,
>;

/// Sends HTTP requests to a specific host via its connection pool.
///
/// One `HostSender` exists per `(scheme, authority)` pair, created on first use and cached
/// for the lifetime of the client.
pub(crate) type HostSender =
    Arc<dyn Fn(http::Request<HttpBody>, ProtocolVersionCell) -> SendFuture + Send + Sync>;

// ---- Pool eviction helpers --------------------------------------------------

/// Spawns a background task that calls `evict` every `interval` to prune stale
/// connections from the pool.
///
/// Returns a guard that, when dropped, cancels the task promptly rather than waiting
/// for the next interval. The task also exits automatically when the pool is dropped.
pub(super) fn spawn_eviction_task<C>(
    cache: &Arc<Mutex<C>>,
    interval: Duration,
    mut evict: impl FnMut(&mut C, Instant) + Send + 'static,
) -> tokio_util::sync::DropGuard
where
    C: Send + 'static,
{
    let cancel = CancellationToken::new();
    tokio::spawn({
        let weak = Arc::downgrade(cache);
        let cancel = cancel.clone();
        async move {
            loop {
                tokio::select! {
                    _ = cancel.cancelled() => break,
                    _ = tokio::time::sleep(interval) => {}
                }
                let Some(arc) = weak.upgrade() else { break };
                evict(&mut arc.lock().unwrap(), Instant::now());
            }
        }
    });
    cancel.drop_guard()
}

// ---- TCP connect helper -----------------------------------------------------

/// Builds a synthetic URI from `key` and dispatches it through `connector`
/// under a connect timeout.
///
/// Maps timeouts to [`HttpClientError::ConnectionTimeout`] and any underlying
/// connector error to [`HttpClientError::Transport`]. The "/" path is required
/// to build a syntactically valid URI; the connector only uses scheme + authority.
pub(crate) async fn connect_with_timeout<S, IO>(
    mut connector: S,
    origin: Origin,
    timeout: std::time::Duration,
) -> Result<IO, HttpClientError>
where
    S: tower::Service<http::Uri, Response = IO>,
    S::Error: Into<BoxError>,
{
    let uri = http::Uri::builder()
        .scheme(origin.scheme)
        .authority(origin.authority)
        .path_and_query("/")
        .build()
        .map_err(|source| HttpClientError::InvalidUri { source })?;
    tokio::time::timeout(timeout, connector.call(uri))
        .await
        .map_err(|_| HttpClientError::ConnectionTimeout)?
        .map_err(|e| HttpClientError::Transport { source: e.into() })
}

// ---- Error helper -----------------------------------------------------------

/// Maps a [`BoxError`] emerging from a pool connector into [`HttpClientError`],
/// preserving any structured variant that was boxed into it (e.g.
/// [`HttpClientError::ProxyTunnel`], [`HttpClientError::ConnectionTimeout`]).
/// Errors that did not originate as an `HttpClientError` are wrapped as
/// [`HttpClientError::Transport`].
pub(crate) fn into_http_client_error(e: BoxError) -> HttpClientError {
    match e.downcast::<HttpClientError>() {
        Ok(http_err) => *http_err,
        Err(other) => HttpClientError::Transport { source: other },
    }
}

// ---- Pool retain helper -----------------------------------------------------

/// Returns whether a connection should be retained in the pool.
///
/// An idle connection is kept if its age is within `idle_timeout` and the running
/// `idle_count` is within `max_idle`. Active connections (no `idle_since`) are always kept.
pub(crate) fn retain_predicate(
    idle_since: Option<Instant>,
    idle_timeout: Duration,
    now: Instant,
    idle_count: &mut usize,
    max_idle: usize,
) -> bool {
    match idle_since {
        Some(t) => {
            if now.duration_since(t) >= idle_timeout {
                return false;
            }
            *idle_count += 1;
            *idle_count <= max_idle
        }
        None => true,
    }
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use super::{HttpClientError, into_http_client_error, retain_predicate};

    #[test]
    fn into_http_client_error_unwraps_boxed_proxy_tunnel() {
        let original = HttpClientError::ProxyTunnel {
            status: http::StatusCode::PROXY_AUTHENTICATION_REQUIRED,
        };
        let boxed: tower::BoxError = Box::new(original);

        let err = into_http_client_error(boxed);
        assert!(
            matches!(
                err,
                HttpClientError::ProxyTunnel {
                    status: http::StatusCode::PROXY_AUTHENTICATION_REQUIRED,
                }
            ),
            "expected ProxyTunnel, got {err}"
        );
    }

    #[test]
    fn into_http_client_error_unwraps_boxed_connection_timeout() {
        let boxed: tower::BoxError = Box::new(HttpClientError::ConnectionTimeout);

        let err = into_http_client_error(boxed);
        assert!(
            matches!(err, HttpClientError::ConnectionTimeout),
            "expected ConnectionTimeout, got {err}"
        );
    }

    #[test]
    fn into_http_client_error_wraps_unstructured_box_error_as_transport() {
        let boxed: tower::BoxError = "io error".into();

        let err = into_http_client_error(boxed);
        assert!(
            matches!(err, HttpClientError::Transport { .. }),
            "expected Transport, got {err}"
        );
    }

    /// An active connection (idle_since = None) is always retained and does not
    /// count toward the idle limit.
    #[tokio::test(start_paused = true)]
    async fn active_connection_always_retained() {
        let now = tokio::time::Instant::now();
        let mut count = 0;
        assert!(retain_predicate(
            None,
            Duration::from_secs(1),
            now,
            &mut count,
            1
        ));
        assert_eq!(count, 0);
    }

    /// An idle connection younger than the timeout is retained.
    #[tokio::test(start_paused = true)]
    async fn idle_within_timeout_retained() {
        let idle_since = tokio::time::Instant::now();
        tokio::time::advance(Duration::from_secs(4)).await;
        let now = tokio::time::Instant::now();
        let mut count = 0;
        assert!(retain_predicate(
            Some(idle_since),
            Duration::from_secs(5),
            now,
            &mut count,
            1
        ));
        assert_eq!(count, 1);
    }

    /// An idle connection that has exceeded the timeout is evicted.
    #[tokio::test(start_paused = true)]
    async fn idle_past_timeout_evicted() {
        let idle_since = tokio::time::Instant::now();
        tokio::time::advance(Duration::from_secs(6)).await;
        let now = tokio::time::Instant::now();
        let mut count = 0;
        assert!(!retain_predicate(
            Some(idle_since),
            Duration::from_secs(5),
            now,
            &mut count,
            1
        ));
    }

    /// The idle count limit evicts connections beyond max_idle, even if they are
    /// within the timeout.
    #[tokio::test(start_paused = true)]
    async fn idle_count_limit_evicts_excess() {
        let idle_since = tokio::time::Instant::now();
        let now = tokio::time::Instant::now();
        let mut count = 0;
        assert!(retain_predicate(
            Some(idle_since),
            Duration::from_secs(60),
            now,
            &mut count,
            2
        ));
        assert!(retain_predicate(
            Some(idle_since),
            Duration::from_secs(60),
            now,
            &mut count,
            2
        ));
        assert!(!retain_predicate(
            Some(idle_since),
            Duration::from_secs(60),
            now,
            &mut count,
            2
        ));
    }
}