vectorizer_sdk/

http_transport.rs

//! HTTP transport implementation using reqwest

use std::time::Duration;

use async_trait::async_trait;
use reqwest::header::{CONTENT_TYPE, HeaderMap, HeaderValue};
use reqwest::{Client, ClientBuilder};
use serde_json::Value;

use crate::error::{Result, VectorizerError};
use crate::transport::{Protocol, Transport};

/// Maximum number of times an HTTP 429 will be retried before the
/// error is surfaced to the caller (issue #263).
const RETRY_AFTER_MAX_ATTEMPTS: u32 = 3;
/// Cap on the `Retry-After` header value the client is willing to
/// honor. A misconfigured server can't pin the client into a long
/// sleep beyond this.
const RETRY_AFTER_MAX_SECS: u64 = 30;
/// Floor on the parsed `Retry-After` value when the header is missing
/// or zero, so we don't busy-loop the server.
const RETRY_AFTER_DEFAULT_SECS: u64 = 1;

/// HTTP transport client
pub struct HttpTransport {
    client: Client,
    base_url: String,
}

impl HttpTransport {
    /// Create a new HTTP transport.
    ///
    /// The `api_key` argument carries either a raw Vectorizer API key
    /// (created via `POST /auth/keys`) or a JWT minted by `POST /auth/login`.
    /// The transport sniffs the shape — three dot-separated base64url
    /// segments → JWT, sent as `Authorization: Bearer <token>`; otherwise
    /// sent as `X-API-Key: <key>`. The server's auth middleware treats
    /// Bearer-wrapped strings as JWTs and never falls back to the API-key
    /// validator, so sending a raw API key under `Authorization: Bearer`
    /// silently 401s. This sniff keeps the public method signature
    /// unchanged while routing each credential down the path the server
    /// actually accepts.
    pub fn new(base_url: &str, api_key: Option<&str>, timeout_secs: u64) -> Result<Self> {
        validate_base_url_scheme(base_url)?;

        let mut headers = HeaderMap::new();
        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));

        if let Some(key) = api_key {
            let (header_name, header_value) = if looks_like_jwt(key) {
                ("Authorization", format!("Bearer {key}"))
            } else {
                ("X-API-Key", key.to_string())
            };
            headers.insert(
                header_name,
                HeaderValue::from_str(&header_value).map_err(|e| {
                    VectorizerError::configuration(format!("Invalid auth credential: {e}"))
                })?,
            );
        }

        let client = ClientBuilder::new()
            .timeout(std::time::Duration::from_secs(timeout_secs))
            .default_headers(headers)
            .build()
            .map_err(|e| {
                VectorizerError::configuration(format!("Failed to create HTTP client: {e}"))
            })?;

        Ok(Self {
            client,
            base_url: base_url.to_string(),
        })
    }
}

/// Reject a base URL this transport cannot dial, at construction.
///
/// Without this, a `vectorizer://` URL builds a client fine and fails at the
/// first request, deep inside reqwest:
///
/// ```text
/// Network error: HTTP request failed:
/// builder error for url (vectorizer://127.0.0.1:15503/auth/login)
/// ```
///
/// which names neither the scheme nor the client that would have worked. The
/// RPC side already handles the mirror-image mistake well — `connect_url` on
/// an `http://` URL points the caller at `VectorizerClient` — so this closes
/// the asymmetry (issue #392).
///
/// Deliberately hand-rolled rather than delegating to
/// `rpc::endpoint::parse_endpoint`: that parser maps a scheme-less
/// `host:port` to *RPC*, and this transport accepts scheme-less base URLs and
/// hands them to reqwest. Routing them through it would reject `localhost:15002`,
/// a form that works today. The only question here is whether reqwest can
/// dial the scheme.
fn validate_base_url_scheme(base_url: &str) -> Result<()> {
    // No `://` means no scheme — `localhost:15002` and bare hosts are
    // accepted, unchanged.
    let Some((scheme, _)) = base_url.split_once("://") else {
        return Ok(());
    };

    if scheme.eq_ignore_ascii_case("http") || scheme.eq_ignore_ascii_case("https") {
        return Ok(());
    }

    if scheme.eq_ignore_ascii_case("vectorizer") {
        return Err(VectorizerError::configuration(format!(
            "VectorizerClient cannot dial RPC URL '{base_url}'; \
             `vectorizer://` is the RPC transport — use \
             `vectorizer_sdk::rpc::RpcClient::connect_url` instead, \
             or pass an `http(s)://` URL"
        )));
    }

    Err(VectorizerError::configuration(format!(
        "unsupported base URL scheme `{scheme}://` in '{base_url}'; \
         VectorizerClient speaks HTTP — pass an `http(s)://` URL"
    )))
}

/// Cheap JWT shape sniff. A JWT is three base64url-encoded segments
/// separated by `.`; every segment must be non-empty. Raw API keys
/// generated by `POST /auth/keys` are a single 32-char alphanumeric
/// string, so they fail this check and get routed to `X-API-Key`.
fn looks_like_jwt(token: &str) -> bool {
    let mut parts = token.split('.');
    let Some(header) = parts.next() else {
        return false;
    };
    let Some(payload) = parts.next() else {
        return false;
    };
    let Some(signature) = parts.next() else {
        return false;
    };
    if parts.next().is_some() {
        return false;
    }
    !header.is_empty() && !payload.is_empty() && !signature.is_empty()
}

impl HttpTransport {
    /// Make a generic request. Honors `Retry-After` on HTTP 429
    /// responses (issue #263): the client sleeps for the header's
    /// value (capped) and retries up to [`RETRY_AFTER_MAX_ATTEMPTS`]
    /// times before surfacing a `RateLimit` error.
    async fn request(&self, method: &str, path: &str, body: Option<&Value>) -> Result<String> {
        let url = format!("{}{}", self.base_url, path);
        let mut attempts_remaining = RETRY_AFTER_MAX_ATTEMPTS;

        loop {
            let mut request = match method {
                "GET" => self.client.get(&url),
                "POST" => self.client.post(&url),
                "PUT" => self.client.put(&url),
                "DELETE" => self.client.delete(&url),
                "PATCH" => self.client.patch(&url),
                _ => {
                    return Err(VectorizerError::configuration(format!(
                        "Unsupported HTTP method: {method}"
                    )));
                }
            };

            if let Some(data) = body {
                request = request.json(data);
            }

            let response = request
                .send()
                .await
                .map_err(|e| VectorizerError::network(format!("HTTP request failed: {e}")))?;

            if response.status().as_u16() == 429 {
                let retry_after = parse_retry_after_secs(
                    response
                        .headers()
                        .get(reqwest::header::RETRY_AFTER)
                        .and_then(|v| v.to_str().ok()),
                );
                let body_text = response
                    .text()
                    .await
                    .unwrap_or_else(|_| "Unknown error".to_string());

                if attempts_remaining == 0 {
                    return Err(VectorizerError::rate_limit(format!(
                        "HTTP 429 after {RETRY_AFTER_MAX_ATTEMPTS} retries: {body_text}",
                    )));
                }

                tracing::info!(
                    "Vectorizer 429 — sleeping {retry_after:?} before retry \
                     (remaining attempts={attempts_remaining})",
                );
                attempts_remaining -= 1;
                tokio::time::sleep(retry_after).await;
                continue;
            }

            if !response.status().is_success() {
                let status = response.status();
                let error_text = response
                    .text()
                    .await
                    .unwrap_or_else(|_| "Unknown error".to_string());
                return Err(VectorizerError::server(format!(
                    "HTTP {status}: {error_text}"
                )));
            }

            return response
                .text()
                .await
                .map_err(|e| VectorizerError::network(format!("Failed to read response: {e}")));
        }
    }
}

/// Parse a `Retry-After` header value (seconds form only). Returns a
/// sensible default when missing/unparseable; caps the value so a
/// misconfigured server can't pin the client into a long sleep.
///
/// Public for test consumption only; not part of the stable SDK API.
#[doc(hidden)]
pub fn parse_retry_after_secs(value: Option<&str>) -> Duration {
    let raw = match value {
        Some(v) => v.trim(),
        None => return Duration::from_secs(RETRY_AFTER_DEFAULT_SECS),
    };
    let secs = raw.parse::<u64>().unwrap_or(RETRY_AFTER_DEFAULT_SECS);
    let secs = if secs == 0 {
        RETRY_AFTER_DEFAULT_SECS
    } else {
        secs.min(RETRY_AFTER_MAX_SECS)
    };
    Duration::from_secs(secs)
}

#[async_trait]
impl Transport for HttpTransport {
    async fn get(&self, path: &str) -> Result<String> {
        self.request("GET", path, None).await
    }

    async fn post(&self, path: &str, data: Option<&Value>) -> Result<String> {
        self.request("POST", path, data).await
    }

    async fn put(&self, path: &str, data: Option<&Value>) -> Result<String> {
        self.request("PUT", path, data).await
    }

    async fn delete(&self, path: &str) -> Result<String> {
        self.request("DELETE", path, None).await
    }

    async fn patch(&self, path: &str, data: Option<&Value>) -> Result<String> {
        self.request("PATCH", path, data).await
    }

    fn protocol(&self) -> Protocol {
        Protocol::Http
    }
}

impl HttpTransport {
    /// Upload a file using multipart/form-data (not part of Transport trait)
    pub async fn post_multipart(
        &self,
        path: &str,
        file_bytes: Vec<u8>,
        filename: &str,
        form_fields: std::collections::HashMap<String, String>,
    ) -> Result<String> {
        let url = format!("{}{}", self.base_url, path);

        // Create multipart form
        let mut form = reqwest::multipart::Form::new();

        // Add file
        let file_part = reqwest::multipart::Part::bytes(file_bytes).file_name(filename.to_string());
        form = form.part("file", file_part);

        // Add other form fields
        for (key, value) in form_fields {
            form = form.text(key, value);
        }

        let response = self
            .client
            .post(&url)
            .multipart(form)
            .send()
            .await
            .map_err(|e| VectorizerError::network(format!("File upload failed: {e}")))?;

        if !response.status().is_success() {
            let status = response.status();
            let error_text = response
                .text()
                .await
                .unwrap_or_else(|_| "Unknown error".to_string());
            return Err(VectorizerError::server(format!(
                "HTTP {status}: {error_text}"
            )));
        }

        response
            .text()
            .await
            .map_err(|e| VectorizerError::network(format!("Failed to read response: {e}")))
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Build a transport the way `VectorizerClient` does, and report the
    /// error message on failure.
    fn try_new(base_url: &str) -> std::result::Result<(), String> {
        HttpTransport::new(base_url, None, 30)
            .map(|_| ())
            .map_err(|e| e.to_string())
    }

    #[test]
    fn rpc_scheme_is_rejected_and_names_the_right_client() {
        let err = try_new("vectorizer://127.0.0.1:15503")
            .expect_err("the REST transport must not accept an RPC URL");

        // The whole point of the fix: the message has to name the scheme AND
        // the client that would have worked. The old failure was reqwest's
        // "builder error for url (...)", which named neither.
        assert!(
            err.contains("RpcClient"),
            "the error must point at the RPC client: {err}"
        );
        assert!(
            err.contains("vectorizer://"),
            "the error must name the offending scheme: {err}"
        );
        assert!(
            err.contains("127.0.0.1:15503"),
            "the error must quote the URL passed in: {err}"
        );
    }

    #[test]
    fn rpc_scheme_is_rejected_case_insensitively() {
        // Schemes are case-insensitive per RFC 3986; a caller shouting the
        // scheme deserves the same guidance.
        let err = try_new("VECTORIZER://host:15503").expect_err("uppercase scheme must be caught");
        assert!(err.contains("RpcClient"), "{err}");
    }

    #[test]
    fn other_schemes_are_rejected_generically() {
        let err = try_new("umicp://host:15004").expect_err("only http(s) is dialable here");
        assert!(
            err.contains("umicp"),
            "the message must name what was passed: {err}"
        );
        assert!(
            !err.contains("RpcClient"),
            "an unrelated scheme must not be misrouted to the RPC client: {err}"
        );
        assert!(
            err.contains("http"),
            "the message must say what is wanted: {err}"
        );
    }

    #[test]
    fn http_and_https_are_accepted() {
        try_new("http://localhost:15002").expect("http is the normal case");
        try_new("https://vectorizer.example.com").expect("https is the normal case");
        try_new("HTTPS://vectorizer.example.com").expect("scheme case must not matter");
    }

    #[test]
    fn scheme_less_base_urls_keep_working() {
        // Callers pass these today and reqwest handles them. This is also why
        // the guard does not delegate to `rpc::endpoint::parse_endpoint`,
        // which would classify them as RPC endpoints and reject them.
        try_new("localhost:15002").expect("scheme-less host:port must still build");
        try_new("localhost").expect("bare host must still build");
    }
}