Voiced/src-tauri/src/integrations/erpnext/client.rs

//! HTTP client for a Frappe/ERPNext site. All requests run here, in Rust: a webview `fetch` from
//! `tauri://localhost` would be blocked by CORS.
//!
//! TLS is rustls with the `ring` provider (installed once, process-wide). Roots come from the OS through
//! rustls-platform-verifier; an optional extra CA PEM is merged in for self-hosted sites.

use super::config::{ErpnextConfig, Secret};
use super::errors::{ErpError, ErrorKind};
use reqwest::header::{HeaderMap, HeaderValue, ACCEPT, AUTHORIZATION, RETRY_AFTER};
use reqwest::{Method, Url};
use serde_json::Value;
use std::sync::Once;
use std::time::Duration;

static CRYPTO_PROVIDER: Once = Once::new();

/// Installs the ring crypto provider for rustls. Idempotent; must run before the first client is built.
pub fn ensure_crypto_provider() {
    CRYPTO_PROVIDER.call_once(|| {
        // Err means another provider is already installed, which is fine for our purposes.
        let _ = rustls::crypto::ring::default_provider().install_default();
    });
}

#[derive(Debug, Clone)]
pub struct ClientOptions {
    pub connect_timeout: Duration,
    pub read_timeout: Duration,
    pub total_timeout: Duration,
    /// Retries after the first attempt.
    pub max_retries: u32,
    pub backoff_base: Duration,
    pub max_backoff: Duration,
}

impl Default for ClientOptions {
    fn default() -> Self {
        ClientOptions {
            connect_timeout: Duration::from_secs(10),
            read_timeout: Duration::from_secs(30),
            total_timeout: Duration::from_secs(60),
            max_retries: 3,
            backoff_base: Duration::from_millis(500),
            max_backoff: Duration::from_secs(8),
        }
    }
}

/// Plain http is only for a local development site; anything else must use https.
pub fn is_local_dev_host(host: &str) -> bool {
    let host = host.trim_end_matches('.').to_ascii_lowercase();
    host == "localhost"
        || host == "127.0.0.1"
        || host.ends_with(".localhost")
        || host.ends_with(".test")
}

/// Normalises what the user typed into `scheme://host[:port][/prefix]` without a trailing slash.
/// A missing scheme means https, except for local development hosts, which get http.
pub fn normalize_base_url(raw: &str) -> Result<String, ErpError> {
    let raw = raw.trim();
    if raw.is_empty() {
        return Err(ErpError::config("The ERPNext address is empty."));
    }
    let with_scheme = if raw.contains("://") {
        raw.to_string()
    } else {
        let host = raw.split(['/', ':']).next().unwrap_or("");
        let scheme = if is_local_dev_host(host) {
            "http"
        } else {
            "https"
        };
        format!("{scheme}://{raw}")
    };
    let mut url = Url::parse(&with_scheme)
        .map_err(|e| ErpError::config(format!("The ERPNext address is not valid: {e}")))?;
    let host = url
        .host_str()
        .ok_or_else(|| ErpError::config("The ERPNext address has no host name."))?
        .to_string();
    match url.scheme() {
        "https" => {}
        "http" => {
            if !is_local_dev_host(&host) {
                return Err(ErpError::config(
                    "Use an https:// address. Plain http is only allowed for localhost, 127.0.0.1, *.localhost and *.test.",
                ));
            }
        }
        other => {
            return Err(ErpError::config(format!(
                "Unsupported address scheme \"{other}\"; use https://."
            )))
        }
    }
    if !url.username().is_empty() || url.password().is_some() {
        return Err(ErpError::config(
            "Do not put credentials in the address; use the API key and secret fields.",
        ));
    }
    url.set_query(None);
    url.set_fragment(None);
    Ok(url.as_str().trim_end_matches('/').to_string())
}

/// A file to upload through a multipart POST.
pub struct Upload<'a> {
    pub file_name: &'a str,
    pub mime: &'a str,
    pub bytes: &'a [u8],
    pub fields: &'a [(&'a str, String)],
}

enum Payload<'a> {
    None,
    Json(&'a Value),
    Upload(&'a Upload<'a>),
}

pub struct ErpClient {
    http: reqwest::Client,
    base: Url,
    opts: ClientOptions,
    /// Strings scrubbed from every error message.
    secrets: Vec<String>,
}

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

impl ErpClient {
    pub fn new(
        base_url: &str,
        api_key: &str,
        api_secret: &Secret,
        extra_ca_pem: &str,
        opts: ClientOptions,
    ) -> Result<Self, ErpError> {
        ensure_crypto_provider();
        let base = Url::parse(&normalize_base_url(base_url)?)
            .map_err(|e| ErpError::config(e.to_string()))?;
        if api_key.trim().is_empty() || !api_secret.is_set() {
            return Err(ErpError::config("Enter the API key and API secret."));
        }
        let mut auth =
            HeaderValue::from_str(&format!("token {}:{}", api_key.trim(), api_secret.expose()))
                .map_err(|_| {
                    ErpError::config(
                        "The API key or secret contains characters that cannot be sent.",
                    )
                })?;
        auth.set_sensitive(true);

        let mut headers = HeaderMap::new();
        headers.insert(AUTHORIZATION, auth);
        // Frappe answers `Accept: text/*` with HTML; always ask for JSON.
        headers.insert(ACCEPT, HeaderValue::from_static("application/json"));

        let mut builder = reqwest::Client::builder()
            .default_headers(headers)
            .user_agent(concat!("Voiced/", env!("CARGO_PKG_VERSION")))
            .connect_timeout(opts.connect_timeout)
            .read_timeout(opts.read_timeout)
            .timeout(opts.total_timeout)
            // A redirect drops the body or the auth header; surface it instead of guessing.
            .redirect(reqwest::redirect::Policy::none());
        if !extra_ca_pem.trim().is_empty() {
            let certs = reqwest::Certificate::from_pem_bundle(extra_ca_pem.trim().as_bytes())
                .map_err(|_| ErpError::config("The extra CA certificate is not valid PEM."))?;
            if certs.is_empty() {
                return Err(ErpError::config(
                    "The extra CA certificate field holds no certificate.",
                ));
            }
            builder = builder.tls_certs_merge(certs);
        }
        let http = builder.build().map_err(|e| {
            ErpError::config(format!(
                "Could not set up the HTTP client: {}",
                error_chain(&e)
            ))
        })?;
        Ok(ErpClient {
            http,
            base,
            opts,
            secrets: vec![api_secret.expose().to_string(), api_key.trim().to_string()],
        })
    }

    pub fn from_config(cfg: &ErpnextConfig) -> Result<Self, ErpError> {
        Self::new(
            &cfg.base_url,
            &cfg.api_key,
            &cfg.api_secret,
            &cfg.extra_ca_pem,
            ClientOptions::default(),
        )
    }

    pub fn base_url(&self) -> &str {
        self.base.as_str().trim_end_matches('/')
    }

    fn url(&self, segments: &[&str]) -> Url {
        let mut url = self.base.clone();
        if let Ok(mut path) = url.path_segments_mut() {
            path.pop_if_empty().extend(segments);
        }
        url
    }

    /// GET, retried on 429, 5xx and timeouts.
    pub async fn get(
        &self,
        segments: &[&str],
        query: &[(&str, String)],
    ) -> Result<Value, ErpError> {
        self.send(Method::GET, segments, query, Payload::None, true)
            .await
    }

    /// POST with a JSON body. 429 is always retried (the server refused before doing anything). 5xx and
    /// timeouts are retried only when `idempotent` is true, because the server may have processed the
    /// request: a mirrored-name Sales Invoice is idempotent (a repeat is a 409), a series-named one is not.
    pub async fn post(
        &self,
        segments: &[&str],
        body: &Value,
        idempotent: bool,
    ) -> Result<Value, ErpError> {
        self.send(Method::POST, segments, &[], Payload::Json(body), idempotent)
            .await
    }

    /// Multipart POST carrying one file part named `file` (the name Frappe's `upload_file` reads; the file
    /// name travels in the part's Content-Disposition) plus plain text fields. Never retried after a 5xx or
    /// timeout, because an upload that may have landed would attach the file twice.
    pub async fn post_file(
        &self,
        segments: &[&str],
        upload: &Upload<'_>,
    ) -> Result<Value, ErpError> {
        self.send(Method::POST, segments, &[], Payload::Upload(upload), false)
            .await
    }

    async fn send(
        &self,
        method: Method,
        segments: &[&str],
        query: &[(&str, String)],
        payload: Payload<'_>,
        retry_ambiguous: bool,
    ) -> Result<Value, ErpError> {
        let mut attempt: u32 = 0;
        loop {
            let mut req = self.http.request(method.clone(), self.url(segments));
            if !query.is_empty() {
                req = req.query(query);
            }
            match &payload {
                Payload::None => {}
                Payload::Json(b) => req = req.json(*b),
                Payload::Upload(u) => {
                    let part = reqwest::multipart::Part::bytes(u.bytes.to_vec())
                        .file_name(u.file_name.to_string())
                        .mime_str(u.mime)
                        .map_err(|e| ErpError::config(format!("Bad upload type: {e}")))?;
                    let mut form = reqwest::multipart::Form::new();
                    for (k, v) in u.fields {
                        form = form.text(k.to_string(), v.clone());
                    }
                    req = req.multipart(form.part("file", part));
                }
            }
            let can_retry = attempt < self.opts.max_retries;
            match req.send().await {
                Ok(resp) => {
                    let status = resp.status();
                    let retry_after = resp
                        .headers()
                        .get(RETRY_AFTER)
                        .and_then(|v| v.to_str().ok())
                        .and_then(|v| v.trim().parse::<u64>().ok())
                        .map(Duration::from_secs);
                    let location = resp
                        .headers()
                        .get(reqwest::header::LOCATION)
                        .and_then(|v| v.to_str().ok())
                        .map(str::to_string);
                    let text = match resp.text().await {
                        Ok(t) => t,
                        Err(e) => {
                            if e.is_timeout() && retry_ambiguous && can_retry {
                                self.backoff(attempt, None).await;
                                attempt += 1;
                                continue;
                            }
                            return Err(self.map_transport(&e));
                        }
                    };
                    if status.is_success() {
                        return self.parse_success(&text);
                    }
                    let retryable =
                        status.as_u16() == 429 || (status.is_server_error() && retry_ambiguous);
                    if retryable && can_retry {
                        self.backoff(attempt, retry_after).await;
                        attempt += 1;
                        continue;
                    }
                    if status.is_redirection() {
                        let to = location.unwrap_or_else(|| "another address".to_string());
                        return Err(self.scrub(ErpError::protocol(format!(
                            "The server redirected the request to {to}. Use the final address (usually https://) in the ERPNext address field."
                        ))));
                    }
                    return Err(self.scrub(ErpError::from_response(status.as_u16(), &text)));
                }
                Err(e) => {
                    if e.is_timeout() && retry_ambiguous && can_retry {
                        self.backoff(attempt, None).await;
                        attempt += 1;
                        continue;
                    }
                    return Err(self.map_transport(&e));
                }
            }
        }
    }

    async fn backoff(&self, attempt: u32, retry_after: Option<Duration>) {
        let exp = self
            .opts
            .backoff_base
            .saturating_mul(1u32 << attempt.min(10));
        let delay = retry_after.unwrap_or(exp).min(self.opts.max_backoff);
        tokio::time::sleep(delay).await;
    }

    fn parse_success(&self, text: &str) -> Result<Value, ErpError> {
        if text.trim().is_empty() {
            return Ok(Value::Null);
        }
        serde_json::from_str(text).map_err(|_| {
            ErpError::protocol(
                "The server did not return JSON. Check the address: it should be the root of your ERPNext site.",
            )
        })
    }

    fn scrub(&self, err: ErpError) -> ErpError {
        let secrets: Vec<&str> = self.secrets.iter().map(String::as_str).collect();
        err.redacted(&secrets)
    }

    fn map_transport(&self, e: &reqwest::Error) -> ErpError {
        let chain = error_chain(e);
        let err = if e.is_timeout() {
            ErpError::new(
                ErrorKind::Timeout,
                "The ERPNext server did not answer in time.",
            )
        } else if e.is_connect() {
            let lower = chain.to_ascii_lowercase();
            let hint = if lower.contains("certificate") || lower.contains("unknownissuer") {
                " The certificate is not trusted: paste the site's CA certificate into the extra CA field."
            } else {
                ""
            };
            ErpError::new(
                ErrorKind::Network,
                format!("Could not connect to the ERPNext server ({chain}).{hint}"),
            )
        } else {
            ErpError::new(
                ErrorKind::Network,
                format!("The request to ERPNext failed ({chain})."),
            )
        };
        self.scrub(err)
    }
}

/// `a: b: c` for an error and its sources.
fn error_chain(e: &dyn std::error::Error) -> String {
    let mut parts = vec![e.to_string()];
    let mut source = e.source();
    while let Some(s) = source {
        let text = s.to_string();
        if parts.last() != Some(&text) {
            parts.push(text);
        }
        source = s.source();
    }
    parts.join(": ")
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::integrations::erpnext::testutil::{client, fast_opts, reply, serve, Reply};
    use serde_json::json;
    use std::sync::atomic::Ordering;

    // ---- URL rules ----

    #[test]
    fn https_is_accepted_and_normalised() {
        assert_eq!(
            normalize_base_url(" https://Erp.Example.com/ ").unwrap(),
            "https://erp.example.com"
        );
        assert_eq!(
            normalize_base_url("erp.example.com").unwrap(),
            "https://erp.example.com"
        );
        assert_eq!(
            normalize_base_url("https://erp.example.com:8443/sub/?a=1#x").unwrap(),
            "https://erp.example.com:8443/sub"
        );
    }

    #[test]
    fn http_is_only_for_local_development_hosts() {
        for ok in [
            "http://localhost:8000",
            "http://127.0.0.1:8080",
            "http://mysite.localhost:8000",
            "http://erp.test",
            "localhost:8000",
            "127.0.0.1:8000",
        ] {
            let url = normalize_base_url(ok).unwrap_or_else(|e| panic!("{ok}: {e}"));
            assert!(url.starts_with("http://"), "{ok} -> {url}");
        }
        for bad in [
            "http://erp.example.com",
            "http://192.168.1.10:8000",
            "http://localhost.evil.com",
            "http://notlocalhost",
        ] {
            assert!(normalize_base_url(bad).is_err(), "{bad} must be refused");
        }
    }

    #[test]
    fn other_schemes_credentials_and_blanks_are_refused() {
        for bad in [
            "ftp://erp.example.com",
            "file:///etc/passwd",
            "https://user:pw@erp.example.com",
            "",
            "   ",
            "https://",
        ] {
            assert!(normalize_base_url(bad).is_err(), "{bad:?} must be refused");
        }
    }

    #[tokio::test]
    async fn sends_token_auth_and_a_json_accept_header() {
        let mock = serve(vec![reply(200, json!({ "message": "Sample@example.com" }))]).await;
        let value = client(&mock)
            .get(&["api", "method", "frappe.auth.get_logged_user"], &[])
            .await
            .unwrap();
        assert_eq!(value["message"], "Sample@example.com");
        let raw = mock.requests.lock().unwrap()[0].to_ascii_lowercase();
        assert!(
            raw.starts_with("get /api/method/frappe.auth.get_logged_user "),
            "{raw}"
        );
        assert!(
            raw.contains("authorization: token thekey:thesecret"),
            "{raw}"
        );
        assert!(raw.contains("accept: application/json"), "{raw}");
    }

    #[tokio::test]
    async fn path_segments_are_percent_encoded_once() {
        let mock = serve(vec![reply(200, json!({ "data": [] }))]).await;
        client(&mock)
            .get(
                &["api", "resource", "Sales Invoice", "INV/2026-001"],
                &[("limit_page_length", "0".into())],
            )
            .await
            .unwrap();
        let raw = mock.requests.lock().unwrap()[0].clone();
        assert!(
            raw.starts_with(
                "GET /api/resource/Sales%20Invoice/INV%2F2026-001?limit_page_length=0 "
            ),
            "{raw}"
        );
    }

    #[tokio::test]
    async fn retries_429_then_succeeds() {
        let mut limited = reply(429, json!({ "message": "slow down" }));
        limited.headers.push(("Retry-After", "0".into()));
        let mock = serve(vec![limited, reply(200, json!({ "message": "ok" }))]).await;
        let value = client(&mock)
            .get(&["api", "method", "ping"], &[])
            .await
            .unwrap();
        assert_eq!(value["message"], "ok");
        assert_eq!(mock.hits.load(Ordering::SeqCst), 2);
    }

    #[tokio::test]
    async fn persistent_500_exhausts_the_retries() {
        let mock = serve(vec![reply(
            500,
            json!({ "exception": "frappe.exceptions.ValidationError: boom" }),
        )])
        .await;
        let err = client(&mock)
            .get(&["api", "method", "ping"], &[])
            .await
            .unwrap_err();
        assert_eq!(err.kind, ErrorKind::Server);
        assert_eq!(err.message, "boom");
        assert_eq!(mock.hits.load(Ordering::SeqCst), 4, "1 attempt + 3 retries");
    }

    #[tokio::test]
    async fn validation_errors_are_not_retried() {
        let mock = serve(vec![reply(
            417,
            json!({ "_error_message": "Rate is required" }),
        )])
        .await;
        let err = client(&mock)
            .post(
                &["api", "resource", "Sales Invoice"],
                &json!({ "a": 1 }),
                true,
            )
            .await
            .unwrap_err();
        assert_eq!(err.kind, ErrorKind::Validation);
        assert_eq!(err.message, "Rate is required");
        assert_eq!(mock.hits.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn timeouts_are_retried_for_reads_but_not_for_ambiguous_writes() {
        let slow = Reply {
            delay_ms: 2_000,
            ..reply(200, json!({}))
        };
        let mock = serve(vec![slow]).await;
        let err = client(&mock)
            .get(&["api", "method", "ping"], &[])
            .await
            .unwrap_err();
        assert_eq!(err.kind, ErrorKind::Timeout);
        assert_eq!(mock.hits.load(Ordering::SeqCst), 4);

        let slow = Reply {
            delay_ms: 2_000,
            ..reply(200, json!({}))
        };
        let mock = serve(vec![slow]).await;
        let err = client(&mock)
            .post(&["api", "resource", "Sales Invoice"], &json!({}), false)
            .await
            .unwrap_err();
        assert_eq!(err.kind, ErrorKind::Timeout);
        assert_eq!(
            mock.hits.load(Ordering::SeqCst),
            1,
            "a write that may have landed is not repeated"
        );
    }

    #[tokio::test]
    async fn non_idempotent_write_retries_429_but_not_500() {
        let mock = serve(vec![
            reply(429, json!({})),
            reply(200, json!({ "data": { "name": "X" } })),
        ])
        .await;
        let value = client(&mock)
            .post(&["api", "resource", "Sales Invoice"], &json!({}), false)
            .await
            .unwrap();
        assert_eq!(value["data"]["name"], "X");
        assert_eq!(mock.hits.load(Ordering::SeqCst), 2);

        let mock = serve(vec![reply(500, json!({}))]).await;
        let err = client(&mock)
            .post(&["api", "resource", "Sales Invoice"], &json!({}), false)
            .await
            .unwrap_err();
        assert_eq!(err.kind, ErrorKind::Server);
        assert_eq!(mock.hits.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn redirects_are_reported_not_followed() {
        let mut moved = reply(301, json!({}));
        moved
            .headers
            .push(("Location", "https://erp.example.com/".into()));
        let mock = serve(vec![moved]).await;
        let err = client(&mock)
            .get(&["api", "method", "ping"], &[])
            .await
            .unwrap_err();
        assert_eq!(err.kind, ErrorKind::Protocol);
        assert!(
            err.message.contains("https://erp.example.com/"),
            "{}",
            err.message
        );
        assert_eq!(mock.hits.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn html_success_page_is_a_protocol_error() {
        let mut page = reply(200, json!({}));
        page.body = "<html>login</html>".into();
        let mock = serve(vec![page]).await;
        let err = client(&mock)
            .get(&["api", "method", "ping"], &[])
            .await
            .unwrap_err();
        assert_eq!(err.kind, ErrorKind::Protocol);
    }

    #[tokio::test]
    async fn refused_connection_is_a_network_error_without_the_secret() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = listener.local_addr().unwrap().port();
        drop(listener);
        let c = ErpClient::new(
            &format!("http://127.0.0.1:{port}"),
            "thekey",
            &Secret::new("thesecret"),
            "",
            fast_opts(),
        )
        .unwrap();
        let err = c.get(&["api", "method", "ping"], &[]).await.unwrap_err();
        assert_eq!(err.kind, ErrorKind::Network);
        assert!(!err.to_string().contains("thesecret"));
        assert!(!format!("{c:?}").contains("thesecret"));
    }

    #[test]
    fn credentials_are_required_and_a_bad_ca_is_refused() {
        assert!(ErpClient::new(
            "https://erp.example.com",
            "",
            &Secret::new("s"),
            "",
            fast_opts()
        )
        .is_err());
        assert!(ErpClient::new(
            "https://erp.example.com",
            "k",
            &Secret::default(),
            "",
            fast_opts()
        )
        .is_err());
        let err = ErpClient::new(
            "https://erp.example.com",
            "k",
            &Secret::new("s"),
            "not a certificate",
            fast_opts(),
        )
        .unwrap_err();
        assert_eq!(err.kind, ErrorKind::Config);
    }
}