openlatch-client 0.3.3

//! Runtime egress health: what the route currently is, and whether it works.
//!
//! [`EgressConfig`] answers "where would a request go"; this module answers "did
//! it get there". The two are deliberately separate — a configuration that parses
//! perfectly can still be pointed at a proxy that died an hour ago, and that is
//! the failure an operator actually hits.
//!
//! # Shape
//!
//! [`EgressState`] follows the two precedents already in the tree rather than
//! inventing a third:
//!
//! | Part | Precedent | Why |
//! |---|---|---|
//! | Counters as `Arc<Atomic*>`, mutated in place | `cloud::CloudState` | A counter bump must not clone a struct |
//! | The resolved view behind an `ArcSwap` | `core::policy::PolicyHandle` | Readers get one coherent view, lock-free |
//!
//! Consumers call [`EgressState::record_ok`] / [`EgressState::record_failure`]
//! **at their own send sites**, exactly as the cloud worker already calls
//! `record_successful_forwards` / `record_drops`. There is no outcome channel:
//! a `reqwest::Client` has no post-send hook to attach one to, a second
//! supervised task would have to be kept alive to drain it, and the in-place
//! call is one line at each site.
//!
//! # Status derivation
//!
//! | Condition | Status |
//! |---|---|
//! | `consecutive_failures >= `[`FAILURE_THRESHOLD`] | `failed` |
//! | otherwise, the configuration resolved with warnings | `degraded` |
//! | otherwise | `ok` |
//!
//! Two consecutive failures, never one: a single timeout on a laptop that just
//! woke from sleep is not an outage, and treating it as one would fire the
//! remedy — and, once plan 02 lands, a self-heal pass — against a perfectly
//! healthy proxy. `unknown` never occurs inside the daemon; it is what the CLI
//! renders when there is no daemon to ask.

use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU32, Ordering};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

use arc_swap::ArcSwap;
use tokio::sync::Notify;

use crate::core::error::{ERR_EGRESS_TLS_FAILED, ERR_EGRESS_UNREACHABLE, ERR_PROXY_UNREACHABLE};

use super::config::{EgressConfig, ProxyAuth, ProxyMode, ProxySource};
use super::no_proxy::NoProxyMatcher;
use super::tls::{self, CaSource};

/// Consecutive failures required before the aggregate reads `failed`.
///
/// The contract's number, and load-bearing: below it, one blip on a flaky link
/// would publish an outage.
pub const FAILURE_THRESHOLD: u32 = 2;

/// Seconds of silence after which an idle health probe is worth issuing.
///
/// Recorded outcomes from real traffic answer the same question for free, so
/// the probe only exists to cover a host that is not sending anything.
pub const IDLE_PROBE_SECS: i64 = 60;

/// Unix epoch seconds. Never fails: a clock before 1970 reports 0, which is the
/// same "never" sentinel an unset timestamp carries.
fn now_secs() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

// ---------------------------------------------------------------------------
// Enums
// ---------------------------------------------------------------------------

/// The aggregate egress health. Mirrors the frozen `egress.status` enum.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EgressStatus {
    /// Reaching the outside world, with nothing to report.
    Ok,
    /// Working, but the configuration resolved with something worth saying.
    Degraded,
    /// [`FAILURE_THRESHOLD`] consecutive failures or more.
    Failed,
    /// **CLI-only.** What `openlatch proxy status` renders when the daemon is
    /// not answering; a running daemon never publishes it.
    Unknown,
}

impl EgressStatus {
    /// The wire string.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Ok => "ok",
            Self::Degraded => "degraded",
            Self::Failed => "failed",
            Self::Unknown => "unknown",
        }
    }
}

/// The scheme actually presented to the proxy. Mirrors the frozen `auth_scheme`
/// values, which are narrower than [`ProxyAuth`]: `auto` is a *policy*, not a
/// scheme, so it resolves here to what the configuration can actually send.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuthScheme {
    /// Nothing is presented.
    None,
    /// Preemptive HTTP Basic.
    Basic,
    /// Kerberos/SPNEGO.
    Negotiate,
}

impl AuthScheme {
    /// The wire string.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::None => "none",
            Self::Basic => "basic",
            Self::Negotiate => "negotiate",
        }
    }

    /// What `cfg` will present.
    ///
    /// `auto` with a username resolves to `basic`, because that is literally
    /// what the factory embeds: reqwest treats a 407 as terminal, so Basic is
    /// sent on the first request or never. `auto` with no username resolves to
    /// `none` — nothing is presented until a scheme is configured.
    pub fn of(cfg: &EgressConfig) -> Self {
        match cfg.auth {
            ProxyAuth::Negotiate => Self::Negotiate,
            ProxyAuth::Basic => Self::Basic,
            ProxyAuth::None => Self::None,
            ProxyAuth::Auto if cfg.username.is_some() => Self::Basic,
            ProxyAuth::Auto => Self::None,
        }
    }
}

/// The kind of proxy in the path. Mirrors the frozen `proxy_type` enum.
///
/// **Shape, never address.** This is the most an outside reader ever learns about an
/// enterprise's egress topology: the `proxy_configured` telemetry event and the
/// `proxytype` envelope attribute both carry it, and neither carries a host or a port
/// (D-15, D-29).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProxyType {
    /// An `http://` proxy.
    Http,
    /// An `https://` proxy — TLS to the proxy itself.
    Https,
    /// A SOCKS5 proxy (`socks5://` or `socks5h://`).
    Socks5,
    /// The route came from a PAC file, **whatever scheme the file returned**.
    Pac,
    /// No proxy is in the path.
    Direct,
}

impl ProxyType {
    /// The wire string.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Http => "http",
            Self::Https => "https",
            Self::Socks5 => "socks5",
            Self::Pac => "pac",
            Self::Direct => "direct",
        }
    }

    /// Derive the type from a resolved route.
    ///
    /// PAC wins over the scheme by contract: a PAC file answering `PROXY host:8080` still
    /// produced a *PAC-sourced* route, and calling it `http` would lose the one fact an
    /// operator debugging a PAC deployment needs.
    fn derive(proxy_in_use: bool, source: Option<ProxySource>, url: Option<&str>) -> Self {
        if !proxy_in_use {
            return Self::Direct;
        }
        if matches!(source, Some(ProxySource::Pac | ProxySource::Wpad)) {
            return Self::Pac;
        }
        match url.and_then(|u| u.split_once("://")).map(|(s, _)| s) {
            Some("https") => Self::Https,
            Some("socks5" | "socks5h") => Self::Socks5,
            // Anything reaching here already passed `validate_url`, which admits only the
            // four schemes above plus `http`.
            _ => Self::Http,
        }
    }
}

/// The latest transport failure, ready to render.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LastError {
    /// The `OL-122x` code.
    pub code: &'static str,
    /// The failure text, with any userinfo masked.
    pub message: String,
}

// ---------------------------------------------------------------------------
// Snapshot
// ---------------------------------------------------------------------------

/// The coherent resolved view of egress, swapped wholesale.
///
/// Swapped rather than mutated field-by-field so a reader never sees a proxy URL
/// from one route beside the source of another — which is exactly what plan 02's
/// self-heal would produce if the fields moved independently.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EgressSnapshot {
    /// See [`EgressStatus`].
    pub status: EgressStatus,
    /// Whether a proxy is actually in the path.
    pub proxy_in_use: bool,
    /// The masked proxy URL — **the only URL form stored anywhere in this
    /// module**. An enterprise's proxy address is internal topology, so it
    /// reaches the authenticated `/admin/egress/status` and nothing else.
    pub proxy_url_masked: Option<String>,
    /// See [`ProxySource`].
    pub source: Option<ProxySource>,
    /// See [`AuthScheme`].
    pub auth_scheme: AuthScheme,
    /// See [`CaSource`].
    pub ca_source: CaSource,
    /// `Some(true)` once a successful handshake was verified against a merged
    /// `ca_bundle`; `None` until an observation exists. See
    /// [`tls::interception_verdict`] for why a native-store success stays
    /// `None` rather than becoming `Some(false)`.
    pub tls_intercepted: Option<bool>,
    /// The observed leaf issuer DN, when one has been read.
    pub tls_issuer: Option<String>,
}

impl EgressSnapshot {
    /// The boot view of `cfg`, before any request has been made.
    ///
    /// `ok` when the configuration parsed clean, `degraded` from the very first
    /// second when it produced warnings — an operator whose `no_proxy` entry was
    /// dropped should not have to wait for traffic to learn that.
    pub fn from_config(cfg: &EgressConfig) -> Self {
        Self {
            status: if cfg.warnings.is_empty() {
                EgressStatus::Ok
            } else {
                EgressStatus::Degraded
            },
            proxy_in_use: cfg.has_proxy(),
            proxy_url_masked: masked_route(cfg),
            source: cfg.source,
            auth_scheme: AuthScheme::of(cfg),
            ca_source: tls::ca_source(cfg),
            tls_intercepted: None,
            tls_issuer: None,
        }
    }

    /// The frozen `proxy_type` value for this route.
    ///
    /// Read off the masked URL rather than the raw one on purpose: the scheme is the only
    /// part of it this answer needs, and the masked form is the only URL this module keeps.
    pub fn proxy_type(&self) -> ProxyType {
        ProxyType::derive(
            self.proxy_in_use,
            self.source,
            self.proxy_url_masked.as_deref(),
        )
    }
}

/// The active route with its password blanked, or `None` when nothing is proxied.
///
/// `EgressConfig::url` never carries userinfo — it is stripped at resolve time —
/// so the username is re-attached here from the field that holds it, and the
/// password is never in scope to leak in the first place.
fn masked_route(cfg: &EgressConfig) -> Option<String> {
    if cfg.mode == ProxyMode::Direct {
        return None;
    }
    let url = cfg.url.as_deref()?;
    let Some(user) = cfg
        .username
        .as_deref()
        .filter(|_| cfg.auth != ProxyAuth::None)
    else {
        return Some(mask_text(url));
    };
    match url.split_once("://") {
        Some((scheme, rest)) => Some(format!("{scheme}://{user}:*****@{rest}")),
        None => Some(mask_text(url)),
    }
}

// ---------------------------------------------------------------------------
// Masking
// ---------------------------------------------------------------------------

/// Blank the password in every URL embedded in `text`.
///
/// Applied to anything derived from a transport error before it is stored:
/// reqwest's `Display` carries the URL it was given, and the factory embeds
/// Basic credentials in the proxy URL, so an unmasked error message is a
/// password in a log line.
pub fn mask_text(text: &str) -> String {
    let mut out = String::with_capacity(text.len());
    let mut rest = text;
    while let Some(i) = rest.find("://") {
        let (head, tail) = rest.split_at(i + 3);
        out.push_str(head);
        let end = tail
            .find(|c: char| c.is_whitespace() || matches!(c, '/' | '?' | '#' | '"' | ',' | ')'))
            .unwrap_or(tail.len());
        let (authority, after) = tail.split_at(end);
        out.push_str(&mask_authority(authority));
        rest = after;
    }
    out.push_str(rest);
    out
}

/// Blank the password in one `user:pass@host:port` authority.
fn mask_authority(authority: &str) -> String {
    let Some((userinfo, host)) = authority.rsplit_once('@') else {
        return authority.to_string();
    };
    match userinfo.split_once(':') {
        Some((user, _)) => format!("{user}:*****@{host}"),
        // A bare username is not a secret, and blanking it would make the
        // rendered URL useless for telling two proxy identities apart.
        None => authority.to_string(),
    }
}

/// Which `OL-122x` a transport failure is.
///
/// The distinction that matters to the operator is *whose* fault the hop was:
/// with a proxy in the path a connect failure is the proxy's (`OL-1221`),
/// without one it is the platform's or the network's (`OL-1220`). A verification
/// failure is neither, and gets the interception remedy (`OL-1224`).
pub fn transport_error_code(
    err: &(dyn std::error::Error + 'static),
    proxied: bool,
) -> &'static str {
    if is_tls_failure(err) {
        return ERR_EGRESS_TLS_FAILED;
    }
    if proxied {
        ERR_PROXY_UNREACHABLE
    } else {
        ERR_EGRESS_UNREACHABLE
    }
}

/// Walk the source chain looking for a certificate rejection.
///
/// rustls does not expose a typed "this was a trust failure" discriminant through
/// reqwest's opaque `Error`, so the chain's own text is the only signal available.
/// It steers a *remedy line*, never a trust decision — a miss costs the operator a
/// less specific suggestion and nothing else.
fn is_tls_failure(err: &(dyn std::error::Error + 'static)) -> bool {
    const MARKERS: &[&str] = &[
        "certificate",
        "unknownissuer",
        "handshake",
        "self-signed",
        "notvalidforname",
    ];
    let mut cursor = Some(err);
    while let Some(e) = cursor {
        let text = e.to_string().to_ascii_lowercase();
        if MARKERS.iter().any(|m| text.contains(m)) {
            return true;
        }
        cursor = e.source();
    }
    false
}

// ---------------------------------------------------------------------------
// EgressState
// ---------------------------------------------------------------------------

/// Shared, cheaply cloned runtime egress health.
///
/// Every field is behind an `Arc`, so a clone is a handful of refcount bumps and
/// every holder observes the same state — the `CloudState` model.
#[derive(Debug, Clone)]
pub struct EgressState {
    snapshot: Arc<ArcSwap<EgressSnapshot>>,
    consecutive_failures: Arc<AtomicU32>,
    /// Epoch seconds of the last success; 0 = never.
    last_ok_at: Arc<AtomicI64>,
    /// Epoch seconds of the last outcome of any kind; 0 = never. Separate from
    /// `last_ok_at` because the idle probe exists to break *silence*, and a
    /// stream of failures is not silence.
    last_activity_at: Arc<AtomicI64>,
    last_error: Arc<ArcSwap<Option<LastError>>>,
    probing: Arc<AtomicBool>,
    failure_notify: Arc<Notify>,
    /// Rendered configuration warnings, fixed at construction. Drives the
    /// `degraded` derivation and is what `doctor` prints.
    warnings: Arc<[String]>,
}

impl EgressState {
    /// Build the state for a resolved configuration.
    pub fn new(cfg: &EgressConfig) -> Self {
        let warnings: Arc<[String]> = cfg.warnings.iter().map(|w| w.to_string()).collect();
        Self {
            snapshot: Arc::new(ArcSwap::from_pointee(EgressSnapshot::from_config(cfg))),
            consecutive_failures: Arc::new(AtomicU32::new(0)),
            last_ok_at: Arc::new(AtomicI64::new(0)),
            last_activity_at: Arc::new(AtomicI64::new(0)),
            last_error: Arc::new(ArcSwap::from_pointee(None)),
            probing: Arc::new(AtomicBool::new(false)),
            failure_notify: Arc::new(Notify::new()),
            warnings,
        }
    }

    /// The current coherent view.
    pub fn snapshot(&self) -> Arc<EgressSnapshot> {
        self.snapshot.load_full()
    }

    /// The derived aggregate status. See the module docs for the ladder.
    pub fn status(&self) -> EgressStatus {
        if self.consecutive_failures() >= FAILURE_THRESHOLD {
            EgressStatus::Failed
        } else if !self.warnings.is_empty() {
            EgressStatus::Degraded
        } else {
            EgressStatus::Ok
        }
    }

    /// The configuration warnings, already rendered.
    pub fn warnings(&self) -> &[String] {
        &self.warnings
    }

    /// Consecutive failures since the last success.
    pub fn consecutive_failures(&self) -> u32 {
        self.consecutive_failures.load(Ordering::Relaxed)
    }

    /// Epoch seconds of the last success, or `None` if there has never been one.
    pub fn last_ok_at(&self) -> Option<i64> {
        match self.last_ok_at.load(Ordering::Relaxed) {
            0 => None,
            secs => Some(secs),
        }
    }

    /// The latest transport failure. Deliberately **not** cleared on recovery:
    /// the contract calls it the *latest* error, and `status` alongside it is
    /// what says whether it is still happening.
    pub fn last_error(&self) -> Option<LastError> {
        self.last_error.load_full().as_ref().clone()
    }

    /// True while an idle probe or a self-heal pass is in flight.
    pub fn is_probing(&self) -> bool {
        self.probing.load(Ordering::Relaxed)
    }

    /// Seconds since the last recorded outcome of any kind. `i64::MAX` when
    /// nothing has ever been recorded, so a freshly started daemon counts as
    /// idle and probes once rather than waiting a minute to start counting.
    pub fn idle_secs(&self) -> i64 {
        match self.last_activity_at.load(Ordering::Relaxed) {
            0 => i64::MAX,
            at => now_secs().saturating_sub(at),
        }
    }

    /// Whether the host has been quiet long enough that a probe would tell us
    /// something real traffic has not already answered.
    pub fn is_idle(&self) -> bool {
        self.idle_secs() >= IDLE_PROBE_SECS
    }

    /// Mark a probe in flight until the returned guard drops.
    pub fn probe_guard(&self) -> ProbeGuard {
        self.probing.store(true, Ordering::Relaxed);
        ProbeGuard {
            flag: self.probing.clone(),
        }
    }

    /// The wake signal the `egress-monitor` task parks on.
    pub fn failure_notify(&self) -> &Notify {
        &self.failure_notify
    }

    /// Record a request that reached its destination.
    ///
    /// "Reached" means the transport worked, not that the server was happy: a
    /// 401 or a 500 both prove the route is fine, and only the route is this
    /// module's business.
    pub fn record_ok(&self) {
        let now = now_secs();
        self.consecutive_failures.store(0, Ordering::Relaxed);
        self.last_ok_at.store(now, Ordering::Relaxed);
        self.last_activity_at.store(now, Ordering::Relaxed);
        self.publish_status();
    }

    /// Record a request that never got out.
    ///
    /// `message` must already be masked — [`mask_text`] is the helper for that,
    /// and [`EgressReporter`] applies it for every consumer.
    pub fn record_failure(&self, code: &'static str, message: impl Into<String>) {
        self.consecutive_failures
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| n.checked_add(1))
            .ok();
        self.last_activity_at.store(now_secs(), Ordering::Relaxed);
        self.last_error.store(Arc::new(Some(LastError {
            code,
            message: message.into(),
        })));
        self.publish_status();
        // Wakes the monitor so a transition is logged (and, from plan 02, a
        // self-heal pass is considered) without waiting for the next tick.
        self.failure_notify.notify_one();
    }

    /// Fold a TLS observation into the snapshot.
    ///
    /// Both halves are optional because they become available independently: the
    /// interception verdict follows from which trust source verified the
    /// handshake, while the issuer DN needs the peer certificate itself. Passing
    /// `None` for either leaves what is already recorded untouched.
    pub fn record_tls_observation(&self, issuer: Option<String>, intercepted: Option<bool>) {
        if issuer.is_none() && intercepted.is_none() {
            return;
        }
        let current = self.snapshot.load();
        if current.tls_issuer == issuer && current.tls_intercepted == intercepted {
            return;
        }
        let mut next = (**current).clone();
        if let Some(issuer) = issuer {
            next.tls_issuer = Some(issuer);
        }
        if intercepted.is_some() {
            next.tls_intercepted = intercepted;
        }
        self.snapshot.store(Arc::new(next));
    }

    /// Recompute the published status from the counters and swap it in if it
    /// moved. Called on every recorded outcome and on every monitor tick.
    ///
    /// Two threads racing here both derive from the same atomics, so whichever
    /// swap lands last still publishes a value consistent with them.
    pub fn publish_status(&self) {
        let want = self.status();
        let current = self.snapshot.load();
        if current.status == want {
            return;
        }
        let mut next = (**current).clone();
        next.status = want;
        self.snapshot.store(Arc::new(next));
    }

    /// Replace the resolved view wholesale. Plan 02's self-heal publishes a
    /// rebuilt route through here; nothing in this plan calls it except tests.
    pub fn publish_snapshot(&self, snapshot: EgressSnapshot) {
        self.snapshot.store(Arc::new(snapshot));
    }
}

/// Clears the `probing` flag when it drops, so an early return or a panic in the
/// probe cannot leave the endpoint claiming a probe is in flight forever.
pub struct ProbeGuard {
    flag: Arc<AtomicBool>,
}

impl Drop for ProbeGuard {
    fn drop(&mut self) {
        self.flag.store(false, Ordering::Relaxed);
    }
}

// ---------------------------------------------------------------------------
// EgressReporter
// ---------------------------------------------------------------------------

/// A consumer's route, plus where its outcomes go.
///
/// One type rather than passing the config and the state side by side: every
/// consumer needs both, and pairing them is what stops a call site from
/// reporting an outcome against a route it did not actually use.
///
/// `state` is `None` for the many hermetic constructions (tests, benches, the
/// CLI's one-shot probes) that have a route but no shared health to update.
#[derive(Debug, Clone)]
pub struct EgressReporter {
    cfg: Arc<EgressConfig>,
    state: Option<EgressState>,
    /// Cached from `cfg` so the hot path does not re-derive it per request.
    proxied: bool,
    ca_source: CaSource,
    bypass: NoProxyMatcher,
}

impl EgressReporter {
    /// A route that reports its outcomes into `state`.
    pub fn recording(cfg: &EgressConfig, state: EgressState) -> Self {
        Self::build(cfg, Some(state))
    }

    /// A route that reports nowhere.
    pub fn silent(cfg: &EgressConfig) -> Self {
        Self::build(cfg, None)
    }

    /// A direct, non-reporting route. The hermetic default for tests and benches.
    pub fn direct() -> Self {
        Self::silent(&EgressConfig::direct())
    }

    fn build(cfg: &EgressConfig, state: Option<EgressState>) -> Self {
        Self {
            proxied: cfg.has_proxy(),
            ca_source: tls::ca_source(cfg),
            bypass: cfg.no_proxy.clone(),
            cfg: Arc::new(cfg.clone()),
            state,
        }
    }

    /// The route, for building clients.
    pub fn config(&self) -> &EgressConfig {
        &self.cfg
    }

    /// The health handle, when this route reports.
    pub fn state(&self) -> Option<&EgressState> {
        self.state.as_ref()
    }

    /// Whether an outcome against `url` says anything about the egress route.
    ///
    /// A bypassed destination — every loopback address, and anything the
    /// operator listed in `no_proxy` — never travels the route, so its success
    /// must not reset a failure streak and its failure must not create one. This
    /// is what keeps a test's loopback mock, and the hook's own POST to the
    /// daemon, out of the numbers.
    pub fn covers(&self, url: &str) -> bool {
        let Some((host, port)) = split_destination(url) else {
            return false;
        };
        !self.bypass.matches(&host, port)
    }

    /// Record a request to `url` that reached its destination.
    pub fn record_ok(&self, url: &str) {
        let Some(state) = self.state.as_ref() else {
            return;
        };
        if !self.covers(url) {
            return;
        }
        state.record_ok();
        // A completed TLS handshake is the only moment interception is knowable,
        // and it is knowable from *which trust source verified it* — never from
        // the leaf alone. Plain HTTP proves nothing either way.
        if url.starts_with("https://") {
            state.record_tls_observation(None, tls::interception_verdict(self.ca_source));
        }
    }

    /// Record a request to `url` that never got out, classifying the error.
    pub fn record_failure(&self, url: &str, err: &(dyn std::error::Error + 'static)) {
        self.record_failure_with(
            url,
            transport_error_code(err, self.proxied),
            mask_text(&err.to_string()),
        );
    }

    /// Record a failure whose code the caller already knows.
    pub fn record_failure_with(&self, url: &str, code: &'static str, message: String) {
        let Some(state) = self.state.as_ref() else {
            return;
        };
        if !self.covers(url) {
            return;
        }
        state.record_failure(code, message);
    }
}

impl From<&EgressConfig> for EgressReporter {
    fn from(cfg: &EgressConfig) -> Self {
        Self::silent(cfg)
    }
}

/// Pull `(host, port)` out of an absolute URL, defaulting the port by scheme.
///
/// Hand-rolled rather than routed through `reqwest::Url`: the answer feeds a
/// bypass decision on a path that must not allocate a parser per request, and
/// the only shapes reaching it are the URLs this crate itself constructs.
fn split_destination(url: &str) -> Option<(String, u16)> {
    let (scheme, rest) = url.split_once("://")?;
    let authority = rest
        .split(['/', '?', '#'])
        .next()
        .filter(|a| !a.is_empty())?;
    // Userinfo is never part of the destination.
    let authority = authority.rsplit_once('@').map_or(authority, |(_, h)| h);
    let default_port = match scheme {
        "https" | "wss" => 443,
        "http" | "ws" => 80,
        _ => return None,
    };
    if let Some(rest) = authority.strip_prefix('[') {
        let (host, tail) = rest.split_once(']')?;
        let port = tail
            .strip_prefix(':')
            .and_then(|p| p.parse().ok())
            .unwrap_or(default_port);
        return Some((host.to_ascii_lowercase(), port));
    }
    match authority.rsplit_once(':') {
        Some((host, port)) => Some((
            host.to_ascii_lowercase(),
            port.parse().unwrap_or(default_port),
        )),
        None => Some((authority.to_ascii_lowercase(), default_port)),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::core::egress::config::{EgressWarning, ProxyToml};
    use std::collections::HashMap;

    struct MapEnv(HashMap<String, String>);
    impl super::super::config::EnvSource for MapEnv {
        fn var(&self, key: &str) -> Option<String> {
            self.0.get(key).cloned()
        }
    }

    fn proxied_cfg() -> EgressConfig {
        let toml = ProxyToml {
            mode: Some("manual".into()),
            url: Some("http://proxy.corp:8080".into()),
            username: Some("alice".into()),
            auth: Some("basic".into()),
            source: Some("windows".into()),
            ..Default::default()
        };
        EgressConfig::resolve(Some(&toml), &MapEnv(HashMap::new()), 7443, 7444).expect("resolve")
    }

    // --- status derivation ---------------------------------------------------

    #[test]
    fn a_clean_config_boots_ok() {
        let state = EgressState::new(&EgressConfig::direct());
        assert_eq!(state.status(), EgressStatus::Ok);
        assert_eq!(state.snapshot().status, EgressStatus::Ok);
        assert_eq!(state.last_ok_at(), None);
        assert!(state.last_error().is_none());
    }

    #[test]
    fn warnings_make_it_degraded_from_boot() {
        let mut cfg = EgressConfig::direct();
        cfg.warnings
            .push(EgressWarning::UnsupportedNoProxyEntry("!!".into()));
        let state = EgressState::new(&cfg);
        assert_eq!(state.status(), EgressStatus::Degraded);
        assert_eq!(
            state.snapshot().status,
            EgressStatus::Degraded,
            "the boot snapshot must already say degraded, not wait for traffic"
        );
        assert_eq!(state.warnings().len(), 1);
    }

    #[test]
    fn one_failure_is_not_an_outage() {
        // The whole reason FAILURE_THRESHOLD exists: a single timeout on a
        // laptop that just woke up must not fire the remedy.
        let state = EgressState::new(&EgressConfig::direct());
        state.record_failure(ERR_PROXY_UNREACHABLE, "timed out");
        assert_eq!(state.consecutive_failures(), 1);
        assert_eq!(state.status(), EgressStatus::Ok);
        assert_eq!(state.snapshot().status, EgressStatus::Ok);
    }

    #[test]
    fn two_failures_are() {
        let state = EgressState::new(&EgressConfig::direct());
        state.record_failure(ERR_PROXY_UNREACHABLE, "timed out");
        state.record_failure(ERR_PROXY_UNREACHABLE, "timed out");
        assert_eq!(state.status(), EgressStatus::Failed);
        assert_eq!(
            state.snapshot().status,
            EgressStatus::Failed,
            "the swap must happen on the recording call, not only on a monitor tick"
        );
        let err = state.last_error().expect("an error must be recorded");
        assert_eq!(err.code, ERR_PROXY_UNREACHABLE);
    }

    #[test]
    fn a_success_resets_the_streak_but_keeps_the_latest_error() {
        let state = EgressState::new(&EgressConfig::direct());
        state.record_failure(ERR_PROXY_UNREACHABLE, "timed out");
        state.record_failure(ERR_PROXY_UNREACHABLE, "timed out");
        state.record_ok();
        assert_eq!(state.consecutive_failures(), 0);
        assert_eq!(state.status(), EgressStatus::Ok);
        assert_eq!(state.snapshot().status, EgressStatus::Ok);
        assert!(
            state.last_error().is_some(),
            "last_error is the LATEST error, not the current one"
        );
        assert!(state.last_ok_at().is_some());
    }

    #[test]
    fn a_degraded_config_never_reports_ok_however_well_traffic_goes() {
        let mut cfg = EgressConfig::direct();
        cfg.warnings.push(EgressWarning::EnvCaseMismatch {
            lower: "https_proxy".into(),
            upper: "HTTPS_PROXY".into(),
        });
        let state = EgressState::new(&cfg);
        state.record_ok();
        assert_eq!(state.status(), EgressStatus::Degraded);
    }

    // --- snapshot ------------------------------------------------------------

    #[test]
    fn the_boot_snapshot_masks_the_url_and_carries_the_provenance() {
        let state = EgressState::new(&proxied_cfg());
        let snap = state.snapshot();
        assert!(snap.proxy_in_use);
        assert_eq!(
            snap.proxy_url_masked.as_deref(),
            Some("http://alice:*****@proxy.corp:8080")
        );
        assert_eq!(snap.source, Some(ProxySource::Windows));
        assert_eq!(snap.auth_scheme, AuthScheme::Basic);
        assert_eq!(snap.ca_source, CaSource::Native);
        assert_eq!(snap.tls_intercepted, None);
    }

    #[test]
    fn direct_stores_no_url_at_all() {
        let snap = EgressSnapshot::from_config(&EgressConfig::direct());
        assert!(!snap.proxy_in_use);
        assert_eq!(snap.proxy_url_masked, None);
        assert_eq!(snap.source, None);
    }

    #[test]
    fn a_snapshot_swap_is_visible_to_every_holder() {
        let a = EgressState::new(&EgressConfig::direct());
        let b = a.clone();
        let mut next = (*a.snapshot()).clone();
        next.proxy_in_use = true;
        next.proxy_url_masked = Some("http://proxy.corp:8080".into());
        a.publish_snapshot(next);
        assert!(
            b.snapshot().proxy_in_use,
            "a clone must observe the same ArcSwap, not a copy of it"
        );
    }

    #[test]
    fn a_tls_observation_lands_on_the_snapshot() {
        let state = EgressState::new(&EgressConfig::direct());
        state.record_tls_observation(Some("CN=Zscaler Root CA".into()), Some(true));
        let snap = state.snapshot();
        assert_eq!(snap.tls_intercepted, Some(true));
        assert_eq!(snap.tls_issuer.as_deref(), Some("CN=Zscaler Root CA"));
        // A no-op call must not clobber what is already known.
        state.record_tls_observation(None, None);
        assert_eq!(
            state.snapshot().tls_issuer.as_deref(),
            Some("CN=Zscaler Root CA")
        );
    }

    // --- notify --------------------------------------------------------------

    #[tokio::test]
    async fn a_failure_wakes_the_monitor() {
        let state = EgressState::new(&EgressConfig::direct());
        let waiter = state.clone();
        let task = tokio::spawn(async move { waiter.failure_notify().notified().await });
        // Give the task a moment to park on the notify before firing it.
        tokio::task::yield_now().await;
        state.record_failure(ERR_PROXY_UNREACHABLE, "boom");
        tokio::time::timeout(std::time::Duration::from_secs(2), task)
            .await
            .expect("record_failure must wake the monitor")
            .expect("waiter task");
    }

    // --- idle ----------------------------------------------------------------

    #[test]
    fn a_host_with_no_outcomes_yet_counts_as_idle() {
        let state = EgressState::new(&EgressConfig::direct());
        assert!(state.is_idle());
    }

    #[test]
    fn a_recent_outcome_suppresses_the_probe() {
        let state = EgressState::new(&EgressConfig::direct());
        state.record_ok();
        assert!(
            !state.is_idle(),
            "traffic already answered the question the probe would ask"
        );
        // A failure is an answer too — a stream of failures is not silence.
        let failing = EgressState::new(&EgressConfig::direct());
        failing.record_failure(ERR_PROXY_UNREACHABLE, "boom");
        assert!(!failing.is_idle());
    }

    #[test]
    fn the_probe_guard_clears_itself() {
        let state = EgressState::new(&EgressConfig::direct());
        assert!(!state.is_probing());
        {
            let _guard = state.probe_guard();
            assert!(state.is_probing());
        }
        assert!(!state.is_probing());
    }

    // --- masking -------------------------------------------------------------

    #[test]
    fn a_password_never_survives_masking() {
        assert_eq!(
            mask_text("http://alice:s3cr3t@proxy.corp:8080"),
            "http://alice:*****@proxy.corp:8080"
        );
        assert_eq!(
            mask_text("error sending request for url (https://u:p@host/path)"),
            "error sending request for url (https://u:*****@host/path)"
        );
        // A bare username is not a secret and stays readable.
        assert_eq!(
            mask_text("http://alice@proxy:8080"),
            "http://alice@proxy:8080"
        );
        // Nothing to mask must round-trip byte-identically.
        assert_eq!(
            mask_text("connect timed out to https://app.openlatch.ai/api/v1/health"),
            "connect timed out to https://app.openlatch.ai/api/v1/health"
        );
    }

    #[test]
    fn masking_is_idempotent() {
        let once = mask_text("http://alice:s3cr3t@proxy.corp:8080");
        assert_eq!(mask_text(&once), once);
    }

    // --- error classification ------------------------------------------------

    #[derive(Debug)]
    struct Plain(&'static str);
    impl std::fmt::Display for Plain {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            f.write_str(self.0)
        }
    }
    impl std::error::Error for Plain {}

    #[test]
    fn a_connect_failure_names_whoever_was_in_the_path() {
        let err = Plain("connection refused");
        assert_eq!(transport_error_code(&err, true), ERR_PROXY_UNREACHABLE);
        assert_eq!(transport_error_code(&err, false), ERR_EGRESS_UNREACHABLE);
    }

    #[test]
    fn a_verification_failure_gets_the_interception_remedy() {
        let err = Plain("invalid peer certificate: UnknownIssuer");
        assert_eq!(transport_error_code(&err, true), ERR_EGRESS_TLS_FAILED);
        assert_eq!(transport_error_code(&err, false), ERR_EGRESS_TLS_FAILED);
    }

    // --- reporter ------------------------------------------------------------

    #[test]
    fn loopback_outcomes_are_never_recorded() {
        // The hazard this closes: a hermetic test's loopback mock, or the hook's
        // own POST to the daemon, resetting a real failure streak.
        let state = EgressState::new(&EgressConfig::direct());
        let reporter = EgressReporter::recording(&EgressConfig::direct(), state.clone());
        assert!(!reporter.covers("http://127.0.0.1:9099/api/v1/health"));
        assert!(!reporter.covers("http://localhost:1234/x"));
        state.record_failure(ERR_PROXY_UNREACHABLE, "boom");
        state.record_failure(ERR_PROXY_UNREACHABLE, "boom");
        reporter.record_ok("http://127.0.0.1:9099/api/v1/health");
        assert_eq!(
            state.status(),
            EgressStatus::Failed,
            "a loopback success must not clear a real outage"
        );
    }

    #[test]
    fn a_real_destination_is_covered_and_recorded() {
        let state = EgressState::new(&EgressConfig::direct());
        let reporter = EgressReporter::recording(&EgressConfig::direct(), state.clone());
        assert!(reporter.covers("https://app.openlatch.ai/api/v1/health"));
        reporter.record_failure(
            "https://app.openlatch.ai/api/v1/health",
            &Plain("connection refused"),
        );
        assert_eq!(state.consecutive_failures(), 1);
        assert_eq!(
            state.last_error().expect("error").code,
            ERR_EGRESS_UNREACHABLE
        );
    }

    #[test]
    fn a_no_proxy_destination_is_not_covered() {
        let toml = ProxyToml {
            mode: Some("manual".into()),
            url: Some("http://proxy.corp:8080".into()),
            no_proxy: Some("internal.corp".into()),
            ..Default::default()
        };
        let cfg =
            EgressConfig::resolve(Some(&toml), &MapEnv(HashMap::new()), 7443, 7444).expect("cfg");
        let reporter = EgressReporter::silent(&cfg);
        assert!(!reporter.covers("https://api.internal.corp/health"));
        assert!(reporter.covers("https://app.openlatch.ai/api/v1/health"));
    }

    #[test]
    fn a_silent_reporter_records_nothing() {
        let reporter = EgressReporter::direct();
        assert!(reporter.state().is_none());
        // Must not panic, must not need a state.
        reporter.record_ok("https://app.openlatch.ai/api/v1/health");
        reporter.record_failure("https://app.openlatch.ai/", &Plain("nope"));
    }

    #[test]
    fn a_proxied_reporter_classifies_failures_as_the_proxys() {
        let cfg = proxied_cfg();
        let state = EgressState::new(&cfg);
        let reporter = EgressReporter::recording(&cfg, state.clone());
        reporter.record_failure("https://app.openlatch.ai/api/v1/health", &Plain("refused"));
        assert_eq!(
            state.last_error().expect("error").code,
            ERR_PROXY_UNREACHABLE
        );
    }

    #[test]
    fn a_custom_bundle_success_records_interception() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("ca.pem");
        let issued = rcgen::generate_simple_self_signed(vec!["ca.test".to_string()]).expect("cert");
        std::fs::write(&path, issued.cert.pem()).expect("write");
        let mut cfg = EgressConfig::direct();
        cfg.ca_bundle = Some(path);

        let state = EgressState::new(&cfg);
        let reporter = EgressReporter::recording(&cfg, state.clone());
        reporter.record_ok("https://app.openlatch.ai/api/v1/health");
        assert_eq!(state.snapshot().tls_intercepted, Some(true));
    }

    #[test]
    fn a_native_store_success_claims_nothing_about_interception() {
        let cfg = EgressConfig::direct();
        let state = EgressState::new(&cfg);
        let reporter = EgressReporter::recording(&cfg, state.clone());
        reporter.record_ok("https://app.openlatch.ai/api/v1/health");
        assert_eq!(
            state.snapshot().tls_intercepted,
            None,
            "an OS-installed private CA is indistinguishable from a public one here"
        );
    }

    // --- destination parsing -------------------------------------------------

    #[test]
    fn destinations_split_with_their_default_ports() {
        assert_eq!(
            split_destination("https://app.openlatch.ai/api/v1/health"),
            Some(("app.openlatch.ai".into(), 443))
        );
        assert_eq!(
            split_destination("http://proxy.corp:8080"),
            Some(("proxy.corp".into(), 8080))
        );
        assert_eq!(
            split_destination("http://[::1]:7443/health"),
            Some(("::1".into(), 7443))
        );
        assert_eq!(
            split_destination("https://u:p@host.example/x"),
            Some(("host.example".into(), 443))
        );
        assert_eq!(split_destination("ftp://host"), None);
        assert_eq!(split_destination("not a url"), None);
    }
}