elasticctl-core 0.1.3

//! HTTP transport, including URL construction, headers, retries, and error
//! classification.

use crate::auth::Credential;
use crate::config::Profile;
use crate::error::{Error, ErrorKind, Result};
use reqwest::{Client, Method, Response, StatusCode};
use serde_json::Value;
use std::collections::BTreeMap;
use std::time::Duration;

/// Version of the public API this client targets.
const API_VERSION: &str = "2023-10-31";
const MAX_ATTEMPTS: u32 = 3;

/// Response headers retained past the transport boundary.
///
/// These headers are allowlisted because recorded fixtures are public.
/// Capturing all headers could record cookies, rate-limit counters, or future
/// proxy headers that do not belong in the repository.
///
/// The capability probe reads `x-found-handling-cluster`. The other two show
/// which Cloud headers the recorded response contained.
const CAPTURED_HEADERS: [&str; 3] = [
    "x-found-handling-cluster",
    "x-found-handling-instance",
    "x-elastic-product",
];

/// A response body and its captured headers.
///
/// Hosted and self-managed stacks return the same `/api/status` body. An
/// edge-proxy header distinguishes them.
#[derive(Debug, Clone)]
pub struct Responded {
    pub body: Value,
    pub headers: BTreeMap<String, String>,
}

impl Responded {
    /// Look up a header case-insensitively.
    ///
    /// The Elastic proxy varies the casing of `x-found-handling-cluster` by
    /// endpoint.
    pub fn header(&self, name: &str) -> Option<&str> {
        self.headers.get(&name.to_ascii_lowercase()).map(|s| &**s)
    }
}

/// Percent-encode a query value while leaving URL-safe characters unchanged.
///
/// The API client and fixture recorder share this encoder so they produce the
/// same scoped-filter URL.
pub fn urlencode(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for b in s.bytes() {
        match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                out.push(b as char)
            }
            _ => out.push_str(&format!("%{b:02X}")),
        }
    }
    out
}

pub struct Transport {
    client: Client,
    base: String,
    /// Elasticsearch host. Cloud deployments use a different host from
    /// Kibana; otherwise this uses the Kibana host.
    es_base: String,
    space: String,
    auth_header: String,
    debug: bool,
}

impl Transport {
    pub fn new(profile: &Profile) -> Result<Transport> {
        Self::with_debug(profile, false)
    }

    /// Build a transport with HTTP request logging enabled or disabled.
    ///
    /// Keeping `debug` as a `bool` prevents CLI `clap` types entering `-core`.
    pub fn with_debug(profile: &Profile, debug: bool) -> Result<Transport> {
        let credential = Credential::from_profile(profile)?;
        let client = Client::builder()
            .timeout(Duration::from_secs(profile.timeout_secs))
            .danger_accept_invalid_certs(!profile.verify)
            .build()
            .map_err(|e| Error::new(ErrorKind::Connection, format!("building HTTP client: {e}")))?;

        let base = profile.kibana_url.trim_end_matches('/').to_string();
        let es_base = profile
            .es_url
            .as_deref()
            .unwrap_or(&profile.kibana_url)
            .trim_end_matches('/')
            .to_string();

        Ok(Transport {
            client,
            base,
            es_base,
            space: profile.space.clone(),
            auth_header: credential.header_value(),
            debug,
        })
    }

    /// Log one request or response line to stderr.
    ///
    /// Logs include only the method, URL, and status. They exclude
    /// authorization headers, bodies, and query-string credentials.
    fn debug_log(&self, method: &Method, url: &str, status: u16, attempt: u32) {
        if !self.debug {
            return;
        }
        if attempt > 1 {
            eprintln!(
                "[debug] {} {url} -> {status} (attempt {attempt})",
                method.as_str()
            );
        } else {
            eprintln!("[debug] {} {url} -> {status}", method.as_str());
        }
    }

    /// Log the request before sending it so timeouts produce debug output.
    fn debug_request(&self, method: &Method, url: &str, attempt: u32) {
        if !self.debug {
            return;
        }
        if attempt > 1 {
            eprintln!("[debug] -> {} {url} (attempt {attempt})", method.as_str());
        } else {
            eprintln!("[debug] -> {} {url}", method.as_str());
        }
    }

    /// Log a timeout or connection failure in the response-line format.
    fn debug_failure(&self, method: &Method, url: &str, what: &str) {
        if !self.debug {
            return;
        }
        eprintln!("[debug] {} {url} -> {what}", method.as_str());
    }

    /// Prefix non-default spaces with `/s/<name>`.
    ///
    /// Kibana serves the default space at the bare path.
    pub fn space_path(space: &str, path: &str) -> String {
        if space.is_empty() || space == "default" {
            path.to_string()
        } else {
            format!("/s/{space}{path}")
        }
    }

    fn url(&self, path: &str) -> String {
        format!("{}{}", self.base, Self::space_path(&self.space, path))
    }

    async fn send(&self, method: Method, path: &str, body: Option<&Value>) -> Result<Response> {
        let url = self.url(path);
        let mut attempt = 0;

        loop {
            attempt += 1;
            let mut req = self
                .client
                .request(method.clone(), &url)
                .header("Authorization", &self.auth_header)
                .header("elastic-api-version", API_VERSION);

            // Kibana rejects any state-changing request without this header.
            if method != Method::GET {
                req = req.header("kbn-xsrf", "true");
            }
            if let Some(b) = body {
                req = req.json(b);
            }

            self.debug_request(&method, &url, attempt);
            let result = req.send().await;

            let response = match result {
                Ok(r) => r,
                Err(e) if e.is_timeout() => {
                    self.debug_failure(&method, &url, "timeout");
                    return Err(Error::new(
                        ErrorKind::Timeout,
                        format!("request timed out: {e}"),
                    ));
                }
                Err(e) => {
                    self.debug_failure(&method, &url, "connection error");
                    return Err(Error::new(
                        ErrorKind::Connection,
                        format!("request failed: {e}"),
                    ));
                }
            };

            let status = response.status();
            self.debug_log(&method, &url, status.as_u16(), attempt);
            if status.is_success() {
                return Ok(response);
            }

            // Retry transient failures only. Retrying a 4xx repeats the same
            // caller error.
            let transient = status == StatusCode::TOO_MANY_REQUESTS || status.is_server_error();
            if transient && attempt < MAX_ATTEMPTS {
                let backoff = Duration::from_millis(200 * 2u64.pow(attempt - 1));
                tokio::time::sleep(backoff).await;
                continue;
            }

            let code = status.as_u16();
            let text = response.text().await.unwrap_or_default();
            return Err(Error::from_response_body(code, &text));
        }
    }

    async fn send_json(&self, method: Method, path: &str, body: Option<&Value>) -> Result<Value> {
        let response = self.send(method, path, body).await?;
        let text = response
            .text()
            .await
            .map_err(|e| Error::new(ErrorKind::Http, format!("reading response body: {e}")))?;
        if text.trim().is_empty() {
            return Ok(Value::Null);
        }
        serde_json::from_str(&text)
            .map_err(|e| Error::new(ErrorKind::Http, format!("parsing response JSON: {e}")))
    }

    pub async fn get(&self, path: &str) -> Result<Value> {
        self.send_json(Method::GET, path, None).await
    }

    /// GET a body with its captured headers.
    ///
    /// This is separate from `get` because only the capability probe needs
    /// headers.
    pub async fn get_with_headers(&self, path: &str) -> Result<Responded> {
        let response = self.send(Method::GET, path, None).await?;

        let mut headers = BTreeMap::new();
        for name in CAPTURED_HEADERS {
            if let Some(value) = response.headers().get(name)
                && let Ok(text) = value.to_str()
            {
                headers.insert(name.to_string(), text.to_string());
            }
        }

        let text = response
            .text()
            .await
            .map_err(|e| Error::new(ErrorKind::Http, format!("reading response body: {e}")))?;
        let body = if text.trim().is_empty() {
            Value::Null
        } else {
            serde_json::from_str(&text)
                .map_err(|e| Error::new(ErrorKind::Http, format!("parsing response JSON: {e}")))?
        };

        Ok(Responded { body, headers })
    }

    pub async fn post(&self, path: &str, body: Option<&Value>) -> Result<Value> {
        self.send_json(Method::POST, path, body).await
    }

    pub async fn put(&self, path: &str, body: &Value) -> Result<Value> {
        self.send_json(Method::PUT, path, Some(body)).await
    }

    pub async fn patch(&self, path: &str, body: &Value) -> Result<Value> {
        self.send_json(Method::PATCH, path, Some(body)).await
    }

    pub async fn delete(&self, path: &str) -> Result<Value> {
        self.send_json(Method::DELETE, path, None).await
    }

    /// GET Elasticsearch without a Kibana space prefix.
    ///
    /// Cloud deployments use a different Elasticsearch host.
    pub async fn get_absolute_es(&self, path: &str) -> Result<Value> {
        let url = format!("{}{}", self.es_base, path);
        self.debug_request(&Method::GET, &url, 1);
        let response = self
            .client
            .get(&url)
            .header("Authorization", &self.auth_header)
            .send()
            .await
            .map_err(|e| {
                self.debug_failure(&Method::GET, &url, "connection error");
                Error::new(ErrorKind::Connection, format!("request failed: {e}"))
            })?;
        let status = response.status().as_u16();
        self.debug_log(&Method::GET, &url, status, 1);
        let text = response.text().await.unwrap_or_default();
        if !(200..300).contains(&status) {
            return Err(Error::from_response_body(status, &text));
        }
        serde_json::from_str(&text)
            .map_err(|e| Error::new(ErrorKind::Http, format!("parsing response JSON: {e}")))
    }

    /// POST JSON to Elasticsearch without a Kibana space prefix or `kbn-xsrf`
    /// header.
    pub async fn post_absolute_es(&self, path: &str, body: &Value) -> Result<Value> {
        self.send_absolute_es(Method::POST, path, Some(body)).await
    }

    /// DELETE from Elasticsearch.
    ///
    /// The fixture recorder uses this to remove its scratch index.
    pub async fn delete_absolute_es(&self, path: &str) -> Result<Value> {
        self.send_absolute_es(Method::DELETE, path, None).await
    }

    async fn send_absolute_es(
        &self,
        method: Method,
        path: &str,
        body: Option<&Value>,
    ) -> Result<Value> {
        let url = format!("{}{}", self.es_base, path);
        let mut req = self
            .client
            .request(method.clone(), &url)
            .header("Authorization", &self.auth_header);
        if let Some(b) = body {
            req = req.json(b);
        }

        self.debug_request(&method, &url, 1);
        let response = req.send().await.map_err(|e| {
            self.debug_failure(&method, &url, "connection error");
            Error::new(ErrorKind::Connection, format!("request failed: {e}"))
        })?;

        let status = response.status().as_u16();
        self.debug_log(&method, &url, status, 1);
        let text = response.text().await.unwrap_or_default();
        if !(200..300).contains(&status) {
            return Err(Error::from_response_body(status, &text));
        }
        if text.trim().is_empty() {
            return Ok(Value::Null);
        }
        serde_json::from_str(&text)
            .map_err(|e| Error::new(ErrorKind::Http, format!("parsing response JSON: {e}")))
    }

    /// POST and return the raw body for NDJSON endpoints.
    pub async fn post_text(&self, path: &str, body: Option<&Value>) -> Result<String> {
        let response = self.send(Method::POST, path, body).await?;
        response
            .text()
            .await
            .map_err(|e| Error::new(ErrorKind::Http, format!("reading response body: {e}")))
    }

    /// Upload a multipart NDJSON file for Kibana rule import.
    pub async fn post_multipart_ndjson(&self, path: &str, ndjson: &str) -> Result<Value> {
        let url = self.url(path);
        let part = reqwest::multipart::Part::text(ndjson.to_string())
            .file_name("rules.ndjson")
            .mime_str("application/octet-stream")
            .map_err(|e| Error::new(ErrorKind::Error, format!("building upload: {e}")))?;
        let form = reqwest::multipart::Form::new().part("file", part);

        self.debug_request(&Method::POST, &url, 1);
        let response = self
            .client
            .post(&url)
            .header("Authorization", &self.auth_header)
            .header("elastic-api-version", API_VERSION)
            .header("kbn-xsrf", "true")
            .multipart(form)
            .send()
            .await
            .map_err(|e| {
                self.debug_failure(&Method::POST, &url, "connection error");
                Error::new(ErrorKind::Connection, format!("upload failed: {e}"))
            })?;

        let status = response.status().as_u16();
        self.debug_log(&Method::POST, &url, status, 1);
        let text = response
            .text()
            .await
            .map_err(|e| Error::new(ErrorKind::Http, format!("reading response body: {e}")))?;
        if !(200..300).contains(&status) {
            return Err(Error::from_response_body(status, &text));
        }
        serde_json::from_str(&text)
            .map_err(|e| Error::new(ErrorKind::Http, format!("parsing response JSON: {e}")))
    }
}