cleanlib_client/

transport.rs

//! HTTP transport — `Client` wraps reqwest with auth header injection +
//! timeout + per Client spec rev1 §2 + Rev 2 amendment §2 + §4.
//!
//! Phase 1 substrate: `fetch_verdict` is the first transport consumer
//! (CLEANLIB-30); subsequent CLEANLIB-N PRs add `scan` / `audit` /
//! `policy preview` / `fetch` verb impls reusing this `Client`.

use std::time::Duration;

use reqwest::{header, Client as ReqwestClient, Method, Url};

use crate::config::Config;
use crate::errors::{from_http, CleanLibraryError, TransportError};
use crate::types::{
    AuditResponse, PolicyPreviewRequest, PolicyPreviewResponse, ScanRequest, ScanResponse, Verdict,
};

/// Default HTTP timeout per Rev 2 amendment §4.1 — allows 1-5s GCS-catalog
/// first-request-ingest + buffer.
const DEFAULT_TIMEOUT_SECS: u64 = 30;

/// `Client` is the cleanlib-client HTTP transport handle. Cheap to clone;
/// share across verbs in the same process.
#[derive(Debug, Clone)]
pub struct Client {
    http: ReqwestClient,
    base_url: Url,
    api_key: Option<String>,
    api_version: String,
}

impl Client {
    /// Construct from a loaded `Config`. TLS required for non-localhost
    /// endpoints. Returns `TlsRequired` for `http://` on remote hosts.
    pub fn from_config(config: &Config) -> Result<Self, CleanLibraryError> {
        Self::build(&config.endpoint.url, config.auth.api_key.clone(), &config.endpoint.api_version)
    }

    /// Construct with explicit endpoint + api_key. Intended for integration
    /// tests + ad-hoc invocations (e.g., CLI `--endpoint=` flag future).
    pub fn new(endpoint: &str, api_key: Option<String>) -> Result<Self, CleanLibraryError> {
        Self::build(endpoint, api_key, "v1")
    }

    fn build(
        endpoint: &str,
        api_key: Option<String>,
        api_version: &str,
    ) -> Result<Self, CleanLibraryError> {
        let base_url = Url::parse(endpoint)
            .map_err(|e| TransportError::InvalidUrl(format!("{}: {}", endpoint, e)))?;

        let is_localhost = matches!(
            base_url.host_str(),
            Some("localhost") | Some("127.0.0.1") | Some("::1")
        );
        if base_url.scheme() != "https" && !is_localhost {
            return Err(TransportError::TlsRequired(endpoint.to_string()).into());
        }

        let http = ReqwestClient::builder()
            .timeout(Duration::from_secs(DEFAULT_TIMEOUT_SECS))
            .user_agent(concat!("cleanlib-cli/", env!("CARGO_PKG_VERSION")))
            .build()
            .map_err(TransportError::Network)?;

        Ok(Self {
            http,
            base_url,
            api_key,
            api_version: api_version.to_string(),
        })
    }

    /// Fetch a single verdict per App Rev 4 §9.3 +
    /// `GET /v1/customer/verdicts/{ecosystem}/{package}/{version}`.
    pub async fn fetch_verdict(
        &self,
        ecosystem: &str,
        package: &str,
        version: &str,
    ) -> Result<Verdict, CleanLibraryError> {
        let path = format!(
            "{}/customer/verdicts/{}/{}/{}",
            self.api_version,
            urlencode(ecosystem),
            urlencode(package),
            urlencode(version),
        );
        let url = self
            .base_url
            .join(&path)
            .map_err(|e| TransportError::InvalidUrl(format!("{}: {}", path, e)))?;

        let response = self.send(Method::GET, url).await?;
        let status = response.status();
        let headers = response.headers().clone();
        let body = response.text().await.map_err(TransportError::Network)?;

        if !status.is_success() {
            return Err(from_http(status.as_u16(), &headers, &body));
        }

        serde_json::from_str(&body)
            .map_err(|e| CleanLibraryError::Parse(format!("verdict response: {}", e)))
    }

    /// Submit a packages + optional-policy request to
    /// `POST /v1/scan` — batch-resolve verdicts for a set of packages against
    /// the customer's active policy. Used by `cleanlib scan`. The App resolves
    /// each package independently (partial-success: a per-package miss lands as
    /// `ScanResult.error`, not a whole-batch failure), so the caller derives
    /// the gating decision per `ScanResult.verdict` and aggregates the exit
    /// code (`commands::scan`).
    ///
    /// Distinct endpoint from [`Self::policy_preview`]: `/v1/scan` needs no
    /// `policy_yaml`. Routing scan through `/v1/policy/preview` produced a
    /// `422 missing field policy_yaml`, and the old `PolicyPreviewResponse`
    /// (`{decisions}`, `#[serde(default)]`) silently parsed the App's
    /// `{results}` body into an empty vec → `scan_exit_code(&[]) == 0`, a
    /// fail-open on the security gate. Both are closed here.
    pub async fn scan(&self, req: &ScanRequest) -> Result<ScanResponse, CleanLibraryError> {
        let path = format!("{}/scan", self.api_version);
        let url = self
            .base_url
            .join(&path)
            .map_err(|e| TransportError::InvalidUrl(format!("{}: {}", path, e)))?;

        let body = serde_json::to_vec(req)
            .map_err(|e| CleanLibraryError::Parse(format!("scan body: {}", e)))?;

        let response = self
            .send_with_body(Method::POST, url, body, "application/json")
            .await?;
        let status = response.status();
        let headers = response.headers().clone();
        let body = response.text().await.map_err(TransportError::Network)?;

        if !status.is_success() {
            return Err(from_http(status.as_u16(), &headers, &body));
        }

        serde_json::from_str(&body)
            .map_err(|e| CleanLibraryError::Parse(format!("scan response: {}", e)))
    }

    /// `POST /v1/policy/preview`. Used by `cleanlib policy preview` (with an
    /// explicit candidate policy override). NOTE: `cleanlib scan` uses
    /// [`Self::scan`] (`/v1/scan`), NOT this endpoint.
    ///
    /// CLEANLIB-305 DX-fix: corrected from `/v1/customer/policy/preview`
    /// (which 404s) to `/v1/policy/preview` (which the App mounts via
    /// `verbs_router`). Same class of bug as the cycle-14 `/v1/audit`
    /// fix (see `audit` below): the `/customer/` prefix is used only by
    /// `customer_verdicts_router` (`/v1/customer/verdicts/*`); the
    /// cycle-6 verb surface (`scan`, `audit`, `policy/preview`,
    /// `risk-accept`, `fetch/*`) mounts flat under `/v1`.
    pub async fn policy_preview(
        &self,
        req: &PolicyPreviewRequest,
    ) -> Result<PolicyPreviewResponse, CleanLibraryError> {
        let path = format!("{}/policy/preview", self.api_version);
        let url = self
            .base_url
            .join(&path)
            .map_err(|e| TransportError::InvalidUrl(format!("{}: {}", path, e)))?;

        let body = serde_json::to_vec(req)
            .map_err(|e| CleanLibraryError::Parse(format!("policy_preview body: {}", e)))?;

        let response = self.send_with_body(Method::POST, url, body, "application/json").await?;
        let status = response.status();
        let headers = response.headers().clone();
        let body = response.text().await.map_err(TransportError::Network)?;

        if !status.is_success() {
            return Err(from_http(status.as_u16(), &headers, &body));
        }

        serde_json::from_str(&body)
            .map_err(|e| CleanLibraryError::Parse(format!("policy_preview response: {}", e)))
    }

    /// Fetch the raw artifact bytes for `(ecosystem, package, version)` via
    /// App's unified catalog-proxy `GET /v1/fetch/{ecosystem}/{package}/{version}`
    /// (CLEANLIB-302 / CLEANLIB-368). Returns owned `Vec<u8>` — caller decides
    /// write target. Emits decision + reason headers to stderr for visibility
    /// (binary stdout stays clean).
    ///
    /// **CLEANLIB-368 route pivot**: prior cycles built per-ecosystem registry
    /// paths (`/npm/<pkg>/-/<pkg>-<ver>.tgz`, `/go/<pkg>/@v/<ver>.zip`, …)
    /// against the App's cycle-3 §C.8 nested per-eco routers. Those paths are
    /// legacy registry-mimic shapes that never surfaced the CLEANLIB-302 audit
    /// row (`gcs_hit` / `gcs_object_path` / `bytes_served`) and hard-coded
    /// pypi wheel-variant assumptions that diverge from the real serve path.
    /// The App unifies both under `verbs_router` at `/v1/fetch/*` — that is
    /// now the sole client-side route.
    pub async fn fetch_artifact(
        &self,
        ecosystem: &str,
        package: &str,
        version: &str,
    ) -> Result<Vec<u8>, CleanLibraryError> {
        let url = build_fetch_url(&self.base_url, &self.api_version, ecosystem, package, version)?;
        let response = self.send(Method::GET, url).await?;
        let status = response.status();
        let headers = response.headers().clone();

        if !status.is_success() {
            let body = response.text().await.map_err(TransportError::Network)?;
            return Err(from_http(status.as_u16(), &headers, &body));
        }

        emit_decision_headers(&headers);

        let bytes = response.bytes().await.map_err(TransportError::Network)?;
        Ok(bytes.to_vec())
    }

    /// Streaming variant of [`Self::fetch_artifact`] — writes chunks to
    /// `writer` without buffering the full body in memory. Per Client Rev 2
    /// amendment §9.4 cycle-4 §D.5 streaming substrate. Returns total bytes
    /// written. Decision + reason headers surface to stderr before stream.
    /// Hits the same unified `/v1/fetch/{ecosystem}/{package}/{version}`
    /// proxy as [`Self::fetch_artifact`] — see the CLEANLIB-368 route pivot
    /// note there.
    pub async fn fetch_artifact_stream<W>(
        &self,
        ecosystem: &str,
        package: &str,
        version: &str,
        writer: &mut W,
    ) -> Result<u64, CleanLibraryError>
    where
        W: tokio::io::AsyncWrite + Unpin,
    {
        use futures_util::StreamExt;
        use tokio::io::AsyncWriteExt;

        let url = build_fetch_url(&self.base_url, &self.api_version, ecosystem, package, version)?;
        let response = self.send(Method::GET, url).await?;
        let status = response.status();
        let headers = response.headers().clone();

        if !status.is_success() {
            let body = response.text().await.map_err(TransportError::Network)?;
            return Err(from_http(status.as_u16(), &headers, &body));
        }

        emit_decision_headers(&headers);

        let mut total: u64 = 0;
        let mut stream = response.bytes_stream();
        while let Some(chunk) = stream.next().await {
            let bytes = chunk.map_err(TransportError::Network)?;
            writer
                .write_all(&bytes)
                .await
                .map_err(|e| CleanLibraryError::Parse(format!("write artifact chunk: {}", e)))?;
            total += bytes.len() as u64;
        }
        writer
            .flush()
            .await
            .map_err(|e| CleanLibraryError::Parse(format!("flush artifact stream: {}", e)))?;
        Ok(total)
    }

    /// Query customer audit log via `GET /v1/audit` with optional filters.
    /// Caller passes already-validated filter values.
    ///
    /// Cycle-14 DX-fix: corrected from `/v1/customer/audit` (which 404s) to
    /// `/v1/audit` (which the App mounts via `verbs_router`). Verified live
    /// against cleanapp.clnstrt.dev 2026-06-05 — direct probe returns 200
    /// with `{window, records, backend_status}`.
    pub async fn audit(
        &self,
        since: Option<&str>,
        decision: Option<&str>,
        ecosystem: Option<&str>,
    ) -> Result<AuditResponse, CleanLibraryError> {
        let path = format!("{}/audit", self.api_version);
        let mut url = self
            .base_url
            .join(&path)
            .map_err(|e| TransportError::InvalidUrl(format!("{}: {}", path, e)))?;
        {
            let mut q = url.query_pairs_mut();
            if let Some(s) = since {
                q.append_pair("since", s);
            }
            if let Some(d) = decision {
                q.append_pair("decision", d);
            }
            if let Some(e) = ecosystem {
                q.append_pair("ecosystem", e);
            }
        }

        let response = self.send(Method::GET, url).await?;
        let status = response.status();
        let headers = response.headers().clone();
        let body = response.text().await.map_err(TransportError::Network)?;

        if !status.is_success() {
            return Err(from_http(status.as_u16(), &headers, &body));
        }

        serde_json::from_str(&body)
            .map_err(|e| CleanLibraryError::Parse(format!("audit response: {}", e)))
    }

        pub async fn probe_auth(&self) -> Result<(), CleanLibraryError> {
        let path = format!("{}/audit", self.api_version);
        let url = self
            .base_url
            .join(&path)
            .map_err(|e| TransportError::InvalidUrl(format!("{}: {}", path, e)))?;
        let response = self.send(Method::GET, url).await?;
        let status = response.status();
        let headers = response.headers().clone();
        let body = response.text().await.map_err(TransportError::Network)?;
        if status.is_success() {
            return Ok(());
        }
        Err(from_http(status.as_u16(), &headers, &body))
    }

    /// Low-level: send a request with body + content-type + auth header.
    async fn send_with_body(
        &self,
        method: Method,
        url: Url,
        body: Vec<u8>,
        content_type: &str,
    ) -> Result<reqwest::Response, CleanLibraryError> {
        let mut req = self.http.request(method, url).body(body);
        req = req.header(header::CONTENT_TYPE, content_type);
        if let Some(key) = &self.api_key {
            req = req.header(header::AUTHORIZATION, format!("Bearer {}", key));
        }
        req.send().await.map_err(|e| {
            if e.is_timeout() {
                CleanLibraryError::Transport(TransportError::Timeout)
            } else {
                CleanLibraryError::Transport(TransportError::Network(e))
            }
        })
    }

    /// Low-level: send a request with auth header. Used internally by verb
    /// methods; public for advanced consumers (future).
    pub async fn send(
        &self,
        method: Method,
        url: Url,
    ) -> Result<reqwest::Response, CleanLibraryError> {
        let mut req = self.http.request(method, url);
        if let Some(key) = &self.api_key {
            req = req.header(header::AUTHORIZATION, format!("Bearer {}", key));
        }
        req.send().await.map_err(|e| {
            if e.is_timeout() {
                CleanLibraryError::Transport(TransportError::Timeout)
            } else {
                CleanLibraryError::Transport(TransportError::Network(e))
            }
        })
    }

    /// Expose the base URL for diagnostics + integration tests.
    pub fn base_url(&self) -> &Url {
        &self.base_url
    }

    /// Fetch supported ecosystems from GET /health ecosystems_mounted field.
    pub async fn get_ecosystems(&self) -> Result<Vec<String>, CleanLibraryError> {
        let url = self.base_url.join("/health")
            .map_err(|e| CleanLibraryError::Transport(
                TransportError::InvalidUrl(e.to_string())
            ))?;
        let resp = self.http.get(url).send().await.map_err(|e| {
            CleanLibraryError::Transport(TransportError::Network(e))
        })?;
        let body_text = resp.text().await.map_err(|e| {
            CleanLibraryError::Transport(TransportError::Network(e))
        })?;
        let body: serde_json::Value = serde_json::from_str(&body_text)
            .map_err(|e| CleanLibraryError::Parse(e.to_string()))?;
        let ecosystems = body["ecosystems_mounted"]
            .as_array()
            .unwrap_or(&vec![])
            .iter()
            .filter_map(|v| v.as_str().map(|s| s.to_string()))
            .collect();
        Ok(ecosystems)
    }
}

/// Emit App's verdict + reason headers (when present) to stderr. Keeps
/// stdout clean for binary content + scripted consumers; visible to TTY
/// users for verdict awareness.
fn emit_decision_headers(headers: &reqwest::header::HeaderMap) {
    if let Some(decision) = headers
        .get("X-CleanLibrary-Decision")
        .and_then(|v| v.to_str().ok())
    {
        eprintln!("# decision: {}", decision);
    }
    if let Some(reason) = headers
        .get("X-CleanLibrary-Reason")
        .and_then(|v| v.to_str().ok())
    {
        eprintln!("# reason: {}", reason);
    }
}

/// Build the App-side catalog-proxy URL for a triple. Post-CLEANLIB-368 this
/// is the ONLY fetch URL shape the client emits: `GET /{api_version}/fetch/
/// {ecosystem}/{package}/{version}` — the unified verb the App mounts in
/// `verbs_router` (see `cleanlib-app/src/verbs.rs`), which returns
/// `application/octet-stream` on catalog-hit and a structured JSON 404 on
/// miss. Per CLEANLIB-302 this is the surface that emits the fetch-audit
/// `AuditRow` with `gcs_hit` / `gcs_object_path` / `bytes_served`
/// populated from the real catalog outcome — the substrate the
/// `/v1/audit` cache-hit-ratio metric consumes.
///
/// Every triple component is percent-encoded so path-embedded slashes
/// (npm-scoped `@scope/name`, go-module `github.com/…/…`) survive the
/// App's axum `Path<(String, String, String)>` extractor as a single
/// segment each.
fn build_fetch_url(
    base: &Url,
    api_version: &str,
    ecosystem: &str,
    package: &str,
    version: &str,
) -> Result<Url, TransportError> {
    let path = format!(
        "{}/fetch/{}/{}/{}",
        api_version,
        urlencode(ecosystem),
        urlencode(package),
        urlencode(version),
    );
    base.join(&path)
        .map_err(|e| TransportError::InvalidUrl(format!("{}: {}", path, e)))
}

/// URL-encode a path segment. Covers `@` + `/` + reserved chars + unicode.
/// Avoids pulling the full `url` crate as a separate dep (re-export is via
/// `reqwest::Url`).
fn urlencode(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for c in s.chars() {
        match c {
            'A'..='Z' | 'a'..='z' | '0'..='9' | '-' | '_' | '.' | '~' => out.push(c),
            _ => {
                let mut buf = [0u8; 4];
                let encoded = c.encode_utf8(&mut buf);
                for b in encoded.bytes() {
                    out.push_str(&format!("%{:02X}", b));
                }
            }
        }
    }
    out
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::{Config, EndpointConfig};

    fn cfg(url: &str) -> Config {
        let mut c = Config::default();
        c.endpoint = EndpointConfig {
            url: url.to_string(),
            api_version: "v1".to_string(),
        };
        c
    }

    #[test]
    fn refuses_remote_plaintext() {
        let err = Client::from_config(&cfg("http://cleanapp.clnstrt.dev")).unwrap_err();
        assert!(matches!(
            err,
            CleanLibraryError::Transport(TransportError::TlsRequired(_))
        ));
    }

    #[test]
    fn allows_localhost_plaintext_for_testing() {
        let client = Client::new("http://localhost:8080", None).unwrap();
        assert_eq!(client.base_url().host_str(), Some("localhost"));
    }

    #[test]
    fn allows_127_loopback_plaintext() {
        let client = Client::new("http://127.0.0.1:8080", None).unwrap();
        assert_eq!(client.base_url().host_str(), Some("127.0.0.1"));
    }

    #[test]
    fn accepts_https_endpoint() {
        let client = Client::from_config(&cfg("https://cleanapp.clnstrt.dev")).unwrap();
        assert_eq!(client.base_url().as_str(), "https://cleanapp.clnstrt.dev/");
    }

    #[test]
    fn rejects_invalid_url() {
        let err = Client::from_config(&cfg("not a url")).unwrap_err();
        assert!(matches!(
            err,
            CleanLibraryError::Transport(TransportError::InvalidUrl(_))
        ));
    }

    #[test]
    fn urlencode_npm_scoped_pkg() {
        // @scope/pkg → %40scope%2Fpkg for safe URL embedding
        assert_eq!(urlencode("@my-org/foo"), "%40my-org%2Ffoo");
    }

    #[test]
    fn urlencode_passes_simple() {
        assert_eq!(urlencode("lodash"), "lodash");
        assert_eq!(urlencode("4.17.21"), "4.17.21");
        assert_eq!(urlencode("github.com/sirupsen/logrus"), "github.com%2Fsirupsen%2Flogrus");
    }

    #[test]
    fn urlencode_handles_unicode() {
        // UTF-8 multi-byte chars get percent-encoded byte-by-byte
        assert_eq!(urlencode("é"), "%C3%A9");
    }

    fn base() -> Url {
        Url::parse("https://cleanapp.clnstrt.dev").unwrap()
    }

    // CLEANLIB-368 route pivot: every ecosystem now resolves to the App's
    // unified `/{api_version}/fetch/{ecosystem}/{package}/{version}` catalog
    // proxy. The client no longer emits registry-mimic per-ecosystem paths
    // (`/npm/…/-/….tgz`, `/go/…/@v/….zip`, `/pypi/…/…-….tar.gz`) — those
    // URL-shape assertions are retired because they are the pre-pivot bug.
    #[test]
    fn fetch_url_routes_through_app_v1_fetch_for_npm_bare() {
        let url = build_fetch_url(&base(), "v1", "npm", "lodash", "4.17.21").unwrap();
        assert_eq!(
            url.as_str(),
            "https://cleanapp.clnstrt.dev/v1/fetch/npm/lodash/4.17.21"
        );
    }

    #[test]
    fn fetch_url_npm_scoped_encodes_at_and_slash() {
        // `@my-org/foo` must survive the App's Path extractor as a single
        // `{package}` segment — so `@` → `%40` and `/` → `%2F`.
        let url = build_fetch_url(&base(), "v1", "npm", "@my-org/foo", "1.0.0").unwrap();
        assert_eq!(
            url.as_str(),
            "https://cleanapp.clnstrt.dev/v1/fetch/npm/%40my-org%2Ffoo/1.0.0"
        );
    }

    #[test]
    fn fetch_url_go_module_path_slashes_encoded() {
        // `github.com/sirupsen/logrus` is one package identifier; its
        // internal `/`s must not split into extra path segments.
        let url = build_fetch_url(&base(), "v1", "go", "github.com/sirupsen/logrus", "v1.9.0")
            .unwrap();
        assert_eq!(
            url.as_str(),
            "https://cleanapp.clnstrt.dev/v1/fetch/go/github.com%2Fsirupsen%2Flogrus/v1.9.0"
        );
    }

    #[test]
    fn fetch_url_pypi_routes_through_v1_fetch_not_registry_mimic() {
        let url = build_fetch_url(&base(), "v1", "pypi", "requests", "2.31.0").unwrap();
        assert_eq!(
            url.as_str(),
            "https://cleanapp.clnstrt.dev/v1/fetch/pypi/requests/2.31.0"
        );
        // Explicitly assert the legacy pypi shape is NOT emitted.
        assert!(!url.as_str().contains("/pypi/requests/requests-"));
    }

    #[test]
    fn fetch_url_ecosystem_client_side_unopinionated() {
        // Post-pivot the client no longer rejects any ecosystem name; the
        // App owns catalog-lookup + returns a structured 404 on miss. This
        // closes the CLEANLIB-368 gap that had maven/crates/nuget/rubygems
        // failing at the client with `Phase 1 Tier A` before even hitting
        // the wire.
        for eco in ["maven", "crates", "nuget", "rubygems", "composer"] {
            let url = build_fetch_url(&base(), "v1", eco, "somepkg", "1.0.0").unwrap();
            assert_eq!(
                url.as_str(),
                format!("https://cleanapp.clnstrt.dev/v1/fetch/{}/somepkg/1.0.0", eco)
            );
        }
    }

    #[test]
    fn fetch_url_maven_coordinates_group_id_encoded() {
        // Maven coordinates use `group:artifact` (no `/`), but the `:` is a
        // reserved char in URL segments — urlencode covers it.
        let url = build_fetch_url(&base(), "v1", "maven", "junit:junit", "4.13.2").unwrap();
        assert_eq!(
            url.as_str(),
            "https://cleanapp.clnstrt.dev/v1/fetch/maven/junit%3Ajunit/4.13.2"
        );
    }

    #[test]
    fn fetch_url_honors_configured_api_version() {
        // Future-proof: if the config carries `api_version = "v2"`, the
        // client picks it up — no hard-coded `v1` in transport.
        let url = build_fetch_url(&base(), "v2", "npm", "lodash", "4.17.21").unwrap();
        assert_eq!(
            url.as_str(),
            "https://cleanapp.clnstrt.dev/v2/fetch/npm/lodash/4.17.21"
        );
    }
}