apollo-http-client 0.3.0

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use apollo_opentelemetry::metrics::Clock;
use apollo_opentelemetry::tower::ServiceBuilderExt as _;
use hyper_util::client::legacy::connect::HttpConnector;
use rustls::ClientConfig;
use tower::ServiceBuilder;
use tower::util::BoxCloneSyncService;

use crate::client::{HttpClient, HttpClientDispatch, HttpClientState};
use crate::config::{HttpClientConfig, Protocol};
use crate::error::HttpClientError;
use crate::metrics::HttpMetrics;
use crate::protocol::h1::H1Connector;
use crate::protocol::h2::H2Connector;
use crate::protocol::{HostSender, Origin};
use crate::spans::HttpSpans;
use crate::tls::{build_https_connector, build_tls_config, with_protocol_alpn};

/// Builder for [`HttpClient`].
///
/// Obtain via [`HttpClient::builder`].
pub struct HttpClientBuilder {
    config: HttpClientConfig,
    tls_override: Option<Arc<ClientConfig>>,
}

impl HttpClientBuilder {
    pub(crate) fn new(config: HttpClientConfig) -> Self {
        Self {
            config,
            tls_override: None,
        }
    }

    /// Injects a pre-built TLS configuration, bypassing the YAML-driven TLS
    /// settings in [`HttpClientConfig`].
    ///
    /// The `protocol` field on the supplied config still governs ALPN
    /// negotiation — any ALPN protocols on the injected `ClientConfig` are
    /// overwritten to match.
    pub fn with_tls_config(mut self, tls: Arc<ClientConfig>) -> Self {
        self.tls_override = Some(tls);
        self
    }

    /// Builds the [`HttpClient`].
    ///
    /// Returns an error if the TLS connector cannot be constructed.
    pub fn build(self) -> Result<HttpClient, HttpClientError> {
        self.build_with_clock(Clock::new())
    }

    pub(crate) fn build_with_clock(self, clock: Clock) -> Result<HttpClient, HttpClientError> {
        let HttpClientBuilder {
            config,
            tls_override,
        } = self;

        // Re-run validation defensively so callers that bypass
        // `apollo_configuration::parse_yaml` still get the same parse-time
        // guarantees (well-formed proxy URL, supported scheme, …). `parse_yaml`
        // already calls this; the cost is negligible and the result is a single
        // source of truth for what a valid `HttpClientConfig` looks like.
        apollo_configuration::validate(config.clone())
            .map_err(|source| HttpClientError::InvalidConfig { source })?;

        let mut http = HttpConnector::new();
        http.enforce_http(false);
        http.set_nodelay(config.tcp.nodelay);
        http.set_keepalive(Some(*config.tcp.keepalive));

        let tls_config: Arc<ClientConfig> = match tls_override {
            Some(ref base) => Arc::new(with_protocol_alpn(base, config.protocol)),
            None => build_tls_config(config.protocol, &config.tls)?,
        };

        let connect_timeout = *config.connect_timeout;

        let configured_protocol_version = match config.protocol {
            Protocol::Http2 => http::Version::HTTP_2,
            Protocol::Http1 | Protocol::Alpn => http::Version::HTTP_11,
        };
        let metrics = HttpMetrics::new(
            clock,
            config.telemetry.metrics.request_body_size,
            config.telemetry.metrics.response_body_size,
            config.telemetry.metrics.url_scheme,
        );
        let conn_metrics = metrics.connection_metrics();
        let h1_max_idle = config.http1.pool_max_idle_per_host.get();
        let h1_idle_timeout = *config.http1.pool_idle_timeout;
        let h2_idle_timeout = *config.http2.connection_idle_timeout;

        #[cfg(unix)]
        let unix_config = Arc::new(crate::protocol::unix::UnixSenderConfig {
            protocol: config.protocol,
            connect_timeout,
            h1_max_idle,
            h1_idle_timeout,
            h2_idle_timeout,
            keep_alive_interval: *config.http2.keep_alive_interval,
            keep_alive_timeout: *config.http2.keep_alive_timeout,
            keep_alive_while_idle: config.http2.keep_alive_while_idle,
            conn_metrics: conn_metrics.clone(),
        });

        let tcp_make_sender: Box<dyn Fn(Origin) -> HostSender + Send + Sync> =
            if let Some(proxy) = &config.proxy {
                let proxy_url = proxy.url.unredact();
                let proxy_host = proxy_url
                    .host_str()
                    .expect("validate above rejects URLs with no host")
                    .to_string();
                let proxy_port = proxy_url.port_or_known_default().expect(
                    "validate above rejects non-http(s) schemes, both of which have default ports",
                );
                // Tag connection metrics with the proxy as `network.peer.*` so
                // operators can tell from telemetry that traffic was proxied.
                let conn_metrics = conn_metrics.with_peer(proxy_host.as_str(), proxy_port);
                let proxy_config = Arc::new(crate::proxy::ProxySenderConfig {
                    proxy_host,
                    proxy_port,
                    proxy_auth: crate::proxy::proxy_authorization(proxy_url),
                    tls_config,
                    protocol: config.protocol,
                    connect_timeout,
                    h1_max_idle,
                    h1_idle_timeout,
                    h2_idle_timeout,
                    keep_alive_interval: *config.http2.keep_alive_interval,
                    keep_alive_timeout: *config.http2.keep_alive_timeout,
                    keep_alive_while_idle: config.http2.keep_alive_while_idle,
                    conn_metrics,
                });
                Box::new(move |key| crate::proxy::new_sender(&proxy_config, key))
            } else {
                let https = build_https_connector(http, tls_config, config.protocol);
                let h1_connector =
                    H1Connector::new(https.clone(), conn_metrics.clone(), connect_timeout);
                let h2_connector = H2Connector::new(
                    https.clone(),
                    conn_metrics.clone(),
                    connect_timeout,
                    *config.http2.keep_alive_interval,
                    *config.http2.keep_alive_timeout,
                    config.http2.keep_alive_while_idle,
                );
                match config.protocol {
                    Protocol::Http2 => Box::new(move |key| {
                        crate::protocol::h2::new_sender(
                            h2_connector.clone(),
                            h2_idle_timeout,
                            key,
                            |_| {},
                        )
                    }),
                    Protocol::Http1 => Box::new(move |key| {
                        crate::protocol::h1::new_sender(
                            h1_connector.clone(),
                            h1_max_idle,
                            h1_idle_timeout,
                            key,
                            |_req| {},
                        )
                    }),
                    Protocol::Alpn => Box::new(move |key| {
                        crate::protocol::alpn::new_sender(
                            h1_connector.clone(),
                            h1_max_idle,
                            h1_idle_timeout,
                            h2_connector.clone(),
                            h2_idle_timeout,
                            key,
                        )
                    }),
                }
            };

        // Route `unix://` keys through the Unix socket connector; everything else
        // goes through the TCP/TLS path built above. Proxy configuration applies
        // only to the TCP path.
        let make_sender: Box<dyn Fn(Origin) -> HostSender + Send + Sync> = Box::new(move |key| {
            if key.scheme.as_str() == "unix" {
                #[cfg(unix)]
                {
                    crate::protocol::unix::new_sender(&unix_config, key)
                }
                #[cfg(not(unix))]
                {
                    crate::client::unsupported_scheme_sender()
                }
            } else {
                tcp_make_sender(key)
            }
        });

        let spans = HttpSpans::new(config.telemetry.spans.clone());
        let dispatch = HttpClientDispatch {
            state: Arc::new(Mutex::new(HttpClientState {
                pools: HashMap::new(),
                make_sender,
            })),
            metrics,
            spans: spans.clone(),
            configured_protocol_version,
        };
        let service = ServiceBuilder::new()
            .traced(
                apollo_opentelemetry::default_instrumentation_scope!(),
                spans,
            )
            .http_client_propagation()
            .service(dispatch);
        let inner = BoxCloneSyncService::new(service);

        Ok(HttpClient::from_service(inner))
    }
}