alkhttp 0.4.1

//! Shared HTTP client: `reqwest_middleware::ClientWithMiddleware` with a
//! retry stack (an idempotency-gated `RetryGateMiddleware` in front of
//! RetryTransientMiddleware + the inlined RetryAfterMiddleware),
//! connection pooling, keep-alive, TLS, and rebuild-and-swap hot-reload.
//!
//! Credential injection happens per-request (from
//! `OperationContext.capabilities`), not at client construction — the
//! client is shared across all operations, the credentials are per-call.
//!
//! # Redirect policy (FWD-03)
//!
//! The default reqwest redirect policy (`limited(10)`) ships custom
//! credential headers across hosts: its cross-host scrub list covers only
//! `Authorization`/`Cookie`/`Proxy-Authorization`/`WWW-Authenticate`, so an
//! `HttpAuthScheme::ApiKey { header_name }` header (and any
//! `default_headers` entry) rides a 302 to an attacker-controlled host
//! intact. The shared client therefore installs an explicit same-host
//! redirect policy: a redirect is followed only when the target's scheme,
//! host and port all match the URL it came from. Otherwise the redirect
//! response is surfaced to the caller untouched and no request — with or
//! without credentials — is sent to another host.
//!
//! # Retry policy (FWD-04)
//!
//! Retries apply only to idempotent methods (GET/HEAD/PUT/DELETE/
//! OPTIONS); POST, PATCH, CONNECT and TRACE bypass the retry middleware
//! entirely (`RetryGateMiddleware`), so a non-idempotent request can
//! never be re-sent and can never duplicate upstream side effects. The
//! backoff is jittered and bounded by a wall-clock budget
//! ([`HttpClientConfig::max_total_retry_duration`]), not just an attempt
//! count.
//!
//! # Timeout / Retry-After caps (FWD-05)
//!
//! `HttpClientConfig::default()` carries a 30 s overall request timeout
//! (anchored to the gateway's 30 s deadline), a 10 s connect timeout and
//! a 30 s read timeout. The overall timeout bounds the request/response
//! forwarding halves (Once-op forwards); it never covers a streaming
//! send — subscriptions are unbounded in *time* by contract (alkcall
//! ADR-021 sets `deadline: None` for the streaming dispatch), so the
//! derived client exposed by
//! [`SharedHttpClient::stream_client`] is built from the same config
//! minus the total timeout while keeping the connect + read timeouts,
//! whose read timeout is the upstream-staleness guard (FWD-15). The
//! byte-wise bound for streaming responses is
//! [`HttpClientConfig::stream_total_byte_cap`], enforced by the SSE
//! forwarding handler (FWD-14). Retry-After deadlines are clamped to
//! [`HttpClientConfig::retry_after_ceiling`] (default 300 s) — a hostile
//! backend cannot park the client on a 10-year deadline.
//!
//! # Blocking reads (FWD-09)
//!
//! The hot-reload path [`SharedHttpClient::reload`] reads CA bundle /
//! client-identity PEM files via `tokio::fs` (`spawn_blocking`-backed),
//! so a rebuild never blocks an async worker. One-shot construction
//! (`SharedHttpClient::new`) performs the same small reads synchronously
//! at assembly time only.

use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, SystemTime};

use arc_swap::ArcSwap;
use http::Extensions;
use reqwest::ClientBuilder;
use reqwest_middleware::ClientWithMiddleware;
use reqwest_retry::policies::ExponentialBackoff;
use reqwest_retry::RetryTransientMiddleware;
use reqwest_retry::{Jitter, RetryDecision, RetryPolicy};
use thiserror::Error;

use super::retry_after::{anchor_budget, RetryAfterMiddleware};

/// Maximum number of URLs tracked by the inlined `Retry-After`
/// middleware (LRU-bounded; see `retry_after.rs`).
const DEFAULT_RETRY_AFTER_CAPACITY: usize = 256;

/// Default overall request timeout. Anchored to the gateway's 30 s
/// deadline: a forwarded call must fail before its caller does.
const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);

/// Default TCP connect timeout — fails fast on unreachable upstreams.
const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);

/// Default idle-read timeout between body bytes.
const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(30);

/// Default wall-clock budget across all retry attempts of a single
/// request (in addition to the retry count).
const DEFAULT_MAX_TOTAL_RETRY_DURATION: Duration = Duration::from_secs(10);

/// Default ceiling applied to any `Retry-After` deadline handed us by an
/// upstream (seconds and HTTP-date forms alike).
const DEFAULT_RETRY_AFTER_CEILING: Duration = Duration::from_secs(300);

/// Default total streamed-bytes cap for one streaming (SSE)
/// subscription forward: 1 GiB. Subscriptions are unbounded in *time*
/// by contract (alkcall ADR-021), so the only agent of a hostile
/// upstream is the total number of bytes it may push into envelope
/// allocation per subscription; past this cap the forward terminates
/// with a single terminal error envelope (FWD-14).
const DEFAULT_STREAM_TOTAL_BYTE_CAP: u64 = 1024 * 1024 * 1024;

/// Default retry count: attempts beyond the first failure of an
/// idempotent request.
const DEFAULT_MAX_RETRIES: u32 = 3;

/// Default lower bound of the retry backoff interval.
const RETRY_BACKOFF_MIN_INTERVAL: Duration = Duration::from_millis(100);

/// Default upper bound of the retry backoff interval.
const RETRY_BACKOFF_MAX_INTERVAL: Duration = Duration::from_secs(2);

/// A mutual-TLS identity presented to upstreams: paths to the
/// PEM-encoded client certificate and its private key. Both files are
/// read at client-construction time (see `HttpClientBuildError` for
/// the failure shapes) and combined into a single reqwest
/// `Identity`.
#[derive(Debug, Clone)]
pub struct ClientCertConfig {
    /// Path to the PEM-encoded client certificate chain.
    pub cert_pem: PathBuf,
    /// Path to the PEM-encoded (PKCS#8) private key for `cert_pem`.
    pub key_pem: PathBuf,
}

/// Retry backoff shape for the shared outbound client. The public
/// surface is plain scalars (HY-06) — the concrete
/// `reqwest_retry::ExponentialBackoff` policy is built internally from
/// these at client-construction time, keeping the upstream concrete
/// type out of this crate's API.
#[derive(Debug, Clone)]
pub struct RetryConfig {
    /// Maximum number of retries after the initial attempt
    /// (idempotent-method requests only — see `RetryGateMiddleware`).
    pub max_retries: u32,
    /// Lower bound of the jittered exponential backoff interval.
    pub initial_backoff: Duration,
    /// Upper bound of the jittered exponential backoff interval.
    pub max_retry_interval: Duration,
}

impl Default for RetryConfig {
    fn default() -> Self {
        Self {
            max_retries: DEFAULT_MAX_RETRIES,
            initial_backoff: RETRY_BACKOFF_MIN_INTERVAL,
            max_retry_interval: RETRY_BACKOFF_MAX_INTERVAL,
        }
    }
}

/// Policy knobs for the shared outbound client
/// (`SharedHttpClient`). Defaults satisfy the review-001 request-policy
/// findings (FWD-03/04/05): same-host-only redirects, idempotent-only
/// retries with a wall-clock budget, 30 s request + 10 s connect
/// timeouts, and a 300 s `Retry-After` ceiling. Streaming (SSE)
/// forwards ride a derived client with the total request timeout
/// removed (FWD-15) and enforce a total streamed-bytes cap (FWD-14).
#[derive(Debug, Clone)]
pub struct HttpClientConfig {
    /// Idle connections kept per host; `None` uses the reqwest default.
    pub pool_max_idle_per_host: Option<usize>,
    /// Overall per-request timeout (default 30 s, off with `None`).
    /// Anchored to the gateway's 30 s Once-op deadline: a forwarded
    /// call must fail before its caller does. Applies only to
    /// request/response forwards — streaming sends ride
    /// [`SharedHttpClient::stream_client`], which is built without it
    /// (FWD-15).
    pub request_timeout: Option<Duration>,
    /// TCP connect timeout (default 10 s, off with `None`).
    pub connect_timeout: Option<Duration>,
    /// Idle timeout between body bytes (default 30 s, off with `None`).
    /// The stall guard: it bounds upstream *staleness*, not stream
    /// lifetime.
    pub read_timeout: Option<Duration>,
    /// Total bytes a single streaming (SSE) forward may pull from its
    /// upstream before the forward terminates with one terminal error
    /// envelope (default 1 GiB, 0 = uncapped — un-recommended outside
    /// tests). Bounded *bytes* per subscription; unbounded *time* is the
    /// contract (alkcall ADR-021) (FWD-14).
    pub stream_total_byte_cap: u64,
    /// Retry backoff shape; only idempotent methods are ever retried
    /// (see `RetryGateMiddleware`).
    pub retry: RetryConfig,
    /// Wall-clock budget all retry attempts of one request must fit in.
    pub max_total_retry_duration: Duration,
    /// Ceiling for `Retry-After` values parsed from upstream responses.
    pub retry_after_ceiling: Duration,
    /// Extra root certificates for upstream TLS verification.
    pub ca_bundle: Option<PathBuf>,
    /// Mutual-TLS identity presented to upstreams.
    pub client_cert: Option<ClientCertConfig>,
}

impl Default for HttpClientConfig {
    fn default() -> Self {
        Self {
            pool_max_idle_per_host: None,
            request_timeout: Some(DEFAULT_REQUEST_TIMEOUT),
            connect_timeout: Some(DEFAULT_CONNECT_TIMEOUT),
            read_timeout: Some(DEFAULT_READ_TIMEOUT),
            stream_total_byte_cap: DEFAULT_STREAM_TOTAL_BYTE_CAP,
            retry: RetryConfig::default(),
            max_total_retry_duration: DEFAULT_MAX_TOTAL_RETRY_DURATION,
            retry_after_ceiling: DEFAULT_RETRY_AFTER_CEILING,
            ca_bundle: None,
            client_cert: None,
        }
    }
}

/// Why a [`SharedHttpClient`] could not be built: PEM file reads fail
/// with `CaBundleRead`/`ClientCertRead`, PEM parsing fails with
/// `CaBundleParse`/`ClientCertParse` (both carry the offending path),
/// and the underlying reqwest builder fails with `Build`.
#[derive(Debug, Error)]
pub enum HttpClientBuildError {
    /// The configured `ca_bundle` path could not be read.
    #[error("failed to read CA bundle from {path}: {source}")]
    CaBundleRead {
        /// The unreadable path, for the caller's diagnostics.
        path: PathBuf,
        /// The underlying I/O error.
        #[source]
        source: std::io::Error,
    },
    /// The CA bundle file exists but is not valid PEM.
    #[error("failed to parse CA bundle at {path}: {source}")]
    CaBundleParse {
        /// The unparseable path, for the caller's diagnostics.
        path: PathBuf,
        /// The underlying reqwest parse error.
        #[source]
        source: reqwest::Error,
    },
    /// A client-certificate PEM path (`cert_pem` or `key_pem`) could
    /// not be read.
    #[error("failed to read client cert from {path}: {source}")]
    ClientCertRead {
        /// The unreadable path.
        path: PathBuf,
        /// The underlying I/O error.
        #[source]
        source: std::io::Error,
    },
    /// The client cert/key files exist but do not form a valid
    /// reqwest `Identity`.
    #[error("failed to parse client cert at {path}: {source}")]
    ClientCertParse {
        /// The path of the identity whose parse failed.
        path: PathBuf,
        /// The underlying reqwest parse error.
        #[source]
        source: reqwest::Error,
    },
    /// The reqwest client itself failed to build (e.g. TLS backend
    /// initialization).
    #[error("failed to build reqwest client: {0}")]
    Build(reqwest::Error),
}

/// A hot-reloadable, middleware-stacked outbound HTTP client shared by
/// consumer adapters. Clone-cheap (`ArcSwap` inner); callers reach the
/// current stack through [`SharedHttpClient::client`].
///
/// Streaming (SSE) subscription forwards reach for
/// [`SharedHttpClient::stream_client`]: a client derived from the same
/// config minus the total request timeout (connect + read timeouts
/// intact; FWD-15) — a healthy subscription longer than 30 s must
/// survive, and
/// reqwest 0.13's per-request override can lengthen but never clear a
/// client-level total timeout. Both rebuild-and-swap together so a
/// reload can never pair one's config with the other's transport
/// (FWD-12).
pub struct SharedHttpClient {
    inner: ArcSwap<SharedHttpInner>,
}

/// Joint holder for both derived clients and the config so a reload
/// swaps all three in one atomic `ArcSwap::store` — a reader can never
/// observe the new config paired with the previous clients, nor the
/// request client paired with the stale stream client (FWD-12, FWD-15).
#[derive(Clone)]
struct SharedHttpInner {
    client: Arc<ClientWithMiddleware>,
    stream_client: Arc<ClientWithMiddleware>,
    config: Arc<HttpClientConfig>,
}

impl std::fmt::Debug for SharedHttpClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SharedHttpClient")
            .field("config", &self.inner.load().config)
            .finish_non_exhaustive()
    }
}

impl SharedHttpClient {
    /// Build a shared client. Blocks only on the (small) PEM reads when
    /// `ca_bundle`/`client_cert` are configured — this is one-shot
    /// construction wiring at assembly time, never a per-request or
    /// hot-reload path; use [`SharedHttpClient::reload`] for async
    /// rebuilds.
    pub fn new(config: HttpClientConfig) -> Result<Self, HttpClientBuildError> {
        let (client, stream_client) = build_clients_sync(&config)?;
        Ok(Self {
            inner: ArcSwap::from_pointee(SharedHttpInner {
                client: Arc::new(client),
                stream_client: Arc::new(stream_client),
                config: Arc::new(config),
            }),
        })
    }

    /// The current middleware-stacked client for request/response
    /// forwarding. Every call loads the latest stack — after a
    /// [`reload`](Self::reload), new requests ride the rebuilt client.
    pub fn client(&self) -> Arc<ClientWithMiddleware> {
        Arc::clone(&self.inner.load().client)
    }

    /// The current middleware-stacked client for streaming (SSE)
    /// subscription forwards: built from the same config with the total
    /// request timeout removed and the connect + read timeouts retained
    /// (FWD-15 — a subscription is unbounded in *time* by contract,
    /// alkcall ADR-021; the read timeout remains the stall guard).
    /// Swapped atomically with [`client`](Self::client) (FWD-12).
    pub fn stream_client(&self) -> Arc<ClientWithMiddleware> {
        Arc::clone(&self.inner.load().stream_client)
    }

    /// The config the current clients were built from (swapped together
    /// with them; FWD-12).
    pub fn config(&self) -> Arc<HttpClientConfig> {
        Arc::clone(&self.inner.load().config)
    }

    /// Rebuild the underlying clients and swap them in for new callers
    /// (in-flight requests complete on the previous clients). PEM reads
    /// use `tokio::fs`, so this is safe to call from async contexts
    /// without blocking a worker. Clients and config swap together in a
    /// single atomic store (FWD-12).
    pub async fn reload(&self, config: HttpClientConfig) -> Result<(), HttpClientBuildError> {
        let (client, stream_client) = build_clients(&config).await?;
        self.inner.store(Arc::new(SharedHttpInner {
            client: Arc::new(client),
            stream_client: Arc::new(stream_client),
            config: Arc::new(config),
        }));
        Ok(())
    }
}

fn same_host_redirect_policy() -> reqwest::redirect::Policy {
    reqwest::redirect::Policy::custom(|attempt| {
        if attempt.previous().len() >= MAX_REDIRECT_HOPS {
            return attempt.error("too many redirects");
        }
        let previous = match attempt.previous().last() {
            Some(url) => url.clone(),
            None => return attempt.follow(),
        };
        let next = attempt.url();
        let scheme_matches = previous.scheme() == next.scheme();
        let host_matches = previous.host_str() == next.host_str();
        let port_matches = previous.port_or_known_default() == next.port_or_known_default();
        if scheme_matches && host_matches && port_matches {
            attempt.follow()
        } else {
            attempt.stop()
        }
    })
}

const MAX_REDIRECT_HOPS: usize = 10;

fn is_idempotent(method: &reqwest::Method) -> bool {
    matches!(
        *method,
        reqwest::Method::GET
            | reqwest::Method::HEAD
            | reqwest::Method::PUT
            | reqwest::Method::DELETE
            | reqwest::Method::OPTIONS
    )
}

struct RetryGateMiddleware {
    retry: Arc<RetryTransientMiddleware<TotalRetryBudget<ExponentialBackoff>>>,
    max_total_retry_duration: Duration,
}

impl RetryGateMiddleware {
    fn new(retry: &RetryConfig, max_total_retry_duration: Duration) -> Self {
        let policy = ExponentialBackoff::builder()
            .retry_bounds(retry.initial_backoff, retry.max_retry_interval)
            .jitter(Jitter::Bounded)
            .base(2)
            .build_with_max_retries(retry.max_retries);
        Self {
            retry: Arc::new(RetryTransientMiddleware::new_with_policy(
                TotalRetryBudget {
                    budget: max_total_retry_duration,
                    inner: policy,
                },
            )),
            max_total_retry_duration,
        }
    }
}

#[async_trait::async_trait]
impl reqwest_middleware::Middleware for RetryGateMiddleware {
    async fn handle(
        &self,
        req: reqwest::Request,
        extensions: &mut Extensions,
        next: reqwest_middleware::Next<'_>,
    ) -> reqwest_middleware::Result<reqwest::Response> {
        if is_idempotent(req.method()) {
            anchor_budget(extensions, self.max_total_retry_duration);
            self.retry.handle(req, extensions, next).await
        } else {
            next.run(req, extensions).await
        }
    }
}

struct TotalRetryBudget<P> {
    budget: Duration,
    inner: P,
}

impl<P: RetryPolicy> RetryPolicy for TotalRetryBudget<P> {
    fn should_retry(&self, request_start_time: SystemTime, n_past_retries: u32) -> RetryDecision {
        let elapsed = SystemTime::now()
            .duration_since(request_start_time)
            .unwrap_or_default();
        if elapsed >= self.budget {
            return RetryDecision::DoNotRetry;
        }
        match self.inner.should_retry(request_start_time, n_past_retries) {
            RetryDecision::DoNotRetry => RetryDecision::DoNotRetry,
            RetryDecision::Retry { execute_after } => {
                let hard_stop = request_start_time
                    .checked_add(self.budget)
                    .unwrap_or(execute_after);
                RetryDecision::Retry {
                    execute_after: execute_after.min(hard_stop),
                }
            }
        }
    }
}

/// Builds the pair (request client, streaming client) from one config:
/// identical stacks except the streaming client carries no total
/// request timeout (FWD-15).
async fn build_clients(
    config: &HttpClientConfig,
) -> Result<(ClientWithMiddleware, ClientWithMiddleware), HttpClientBuildError> {
    let pems = read_pems(config).await?;
    let client = build_client_with_pems(config, pems.clone(), false)?;
    let stream_client = build_client_with_pems(config, pems, true)?;
    Ok((client, stream_client))
}

fn build_clients_sync(
    config: &HttpClientConfig,
) -> Result<(ClientWithMiddleware, ClientWithMiddleware), HttpClientBuildError> {
    let pems = read_pems_sync(config)?;
    let client = build_client_with_pems(config, pems.clone(), false)?;
    let stream_client = build_client_with_pems(config, pems, true)?;
    Ok((client, stream_client))
}

#[derive(Clone)]
struct ClientPems {
    ca: Option<Vec<u8>>,
    client: Option<(Vec<u8>, Vec<u8>)>,
}

async fn read_pems(config: &HttpClientConfig) -> Result<ClientPems, HttpClientBuildError> {
    let ca = match &config.ca_bundle {
        Some(path) => Some(tokio::fs::read(path).await.map_err(|source| {
            HttpClientBuildError::CaBundleRead {
                path: path.clone(),
                source,
            }
        })?),
        None => None,
    };
    let client = match &config.client_cert {
        Some(cfg) => {
            let cert_pem = tokio::fs::read(&cfg.cert_pem).await.map_err(|source| {
                HttpClientBuildError::ClientCertRead {
                    path: cfg.cert_pem.clone(),
                    source,
                }
            })?;
            let key_pem = tokio::fs::read(&cfg.key_pem).await.map_err(|source| {
                HttpClientBuildError::ClientCertRead {
                    path: cfg.key_pem.clone(),
                    source,
                }
            })?;
            Some((cert_pem, key_pem))
        }
        None => None,
    };
    Ok(ClientPems { ca, client })
}

fn read_pems_sync(config: &HttpClientConfig) -> Result<ClientPems, HttpClientBuildError> {
    let ca = match &config.ca_bundle {
        Some(path) => {
            Some(
                std::fs::read(path).map_err(|source| HttpClientBuildError::CaBundleRead {
                    path: path.clone(),
                    source,
                })?,
            )
        }
        None => None,
    };
    let client = match &config.client_cert {
        Some(cfg) => {
            let cert_pem = std::fs::read(&cfg.cert_pem).map_err(|source| {
                HttpClientBuildError::ClientCertRead {
                    path: cfg.cert_pem.clone(),
                    source,
                }
            })?;
            let key_pem = std::fs::read(&cfg.key_pem).map_err(|source| {
                HttpClientBuildError::ClientCertRead {
                    path: cfg.key_pem.clone(),
                    source,
                }
            })?;
            Some((cert_pem, key_pem))
        }
        None => None,
    };
    Ok(ClientPems { ca, client })
}

fn build_client_with_pems(
    config: &HttpClientConfig,
    pems: ClientPems,
    streaming: bool,
) -> Result<ClientWithMiddleware, HttpClientBuildError> {
    let ClientPems {
        ca: ca_pem,
        client: client_pems,
    } = pems;
    let mut builder = ClientBuilder::new();
    builder = builder.redirect(same_host_redirect_policy());
    if let Some(pool_max_idle) = config.pool_max_idle_per_host {
        builder = builder.pool_max_idle_per_host(pool_max_idle);
    }
    if !streaming {
        if let Some(timeout) = config.request_timeout {
            builder = builder.timeout(timeout);
        }
    }
    if let Some(timeout) = config.connect_timeout {
        builder = builder.connect_timeout(timeout);
    }
    if let Some(timeout) = config.read_timeout {
        builder = builder.read_timeout(timeout);
    }
    if let Some(pem) = &ca_pem {
        let certs = reqwest::Certificate::from_pem_bundle(pem).map_err(|source| {
            HttpClientBuildError::CaBundleParse {
                path: config
                    .ca_bundle
                    .clone()
                    .unwrap_or_else(|| PathBuf::from("<ca-bundle>")),
                source,
            }
        })?;
        for cert in certs {
            builder = builder.add_root_certificate(cert);
        }
    }
    if let Some((cert_pem, key_pem)) = &client_pems {
        let identity = reqwest::Identity::from_pem(concat_pem(cert_pem, key_pem).as_slice())
            .map_err(|source| HttpClientBuildError::ClientCertParse {
                path: config
                    .client_cert
                    .as_ref()
                    .map(|cfg| cfg.cert_pem.clone())
                    .unwrap_or_else(|| PathBuf::from("<client-cert>")),
                source,
            })?;
        builder = builder.identity(identity);
    }
    let reqwest_client = builder.build().map_err(HttpClientBuildError::Build)?;
    let retry_after = RetryAfterMiddleware::with_capacity_ceiling_and_budget(
        DEFAULT_RETRY_AFTER_CAPACITY,
        config.retry_after_ceiling,
        config.max_total_retry_duration,
    );
    let client = reqwest_middleware::ClientBuilder::new(reqwest_client)
        .with(RetryGateMiddleware::new(
            &config.retry,
            config.max_total_retry_duration,
        ))
        .with(retry_after)
        .build();
    Ok(client)
}

fn concat_pem(cert: &[u8], key: &[u8]) -> Vec<u8> {
    let mut combined = Vec::with_capacity(cert.len() + key.len() + 1);
    combined.extend_from_slice(cert);
    if !cert.is_empty() && cert.last() != Some(&b'\n') {
        combined.push(b'\n');
    }
    combined.extend_from_slice(key);
    combined
}