path: root/src/client.rs

//! The HTTP client.
//!
//! Retry policy: 429 and the transient 5xx in [`RETRIABLE_STATUS`], plus
//! network failures, are retried with exponentially backed-off, jittered waits.
//! 400 and 404 are deliberately absent from that set — both are definitive
//! answers about the request, so retrying returns the same thing more slowly
//! while burning someone else's quota.

use std::time::{Duration, Instant};

use serde_json::{Map, Value};

use crate::csv;
use crate::error::{Error, Result};
use crate::model::{Address, Faixa, Health};
use crate::retry::{jitter, parse_retry_after};

/// Where the public API lives.
pub const DEFAULT_URL: &str = "https://cep.is";

/// How many CEPs the batch endpoint accepts in one request.
pub const MAX_BATCH: usize = 500;

/// The statuses worth trying again: rate limiting and the transient 5xx.
pub const RETRIABLE_STATUS: [u16; 5] = [429, 500, 502, 503, 504];

const ACCEPT: &str = "application/json, text/csv;q=0.9, */*;q=0.1";

/// Return the CEP's 8 digits, accepting the mask.
///
/// `"01001-000"`, `"01001000"` and `" 01001-000 "` all become `"01001000"`.
///
/// Fails with [`Error::InvalidCep`] unless exactly 8 digits remain. This runs
/// **before** any request: a 7-digit CEP is the caller's bug, and spending a
/// round trip for the server to say so wastes both sides' time.
///
/// ```
/// assert_eq!(cepis::normalize_cep("01001-000").unwrap(), "01001000");
/// assert!(cepis::normalize_cep("123").is_err());
/// ```
pub fn normalize_cep(value: &str) -> Result<String> {
    let cleaned: String = value
        .trim()
        .chars()
        .filter(|c| !matches!(c, '-' | '.' | ' '))
        .collect();

    if cleaned.len() == 8 && cleaned.bytes().all(|b| b.is_ascii_digit()) {
        Ok(cleaned)
    } else {
        Err(Error::invalid_cep(format!(
            "CEP must be 8 digits, got {value:?}"
        )))
    }
}

/// Builder for a [`Client`].
///
/// Every value has a default; set only what differs. See [`Client`] for what
/// each one means.
#[derive(Debug, Clone)]
pub struct ClientBuilder {
    base_url: String,
    timeout: Duration,
    max_attempts: u32,
    backoff_base: Duration,
    backoff_max: Duration,
    total_timeout: Option<Duration>,
    user_agent: String,
}

impl Default for ClientBuilder {
    fn default() -> Self {
        ClientBuilder {
            base_url: DEFAULT_URL.to_owned(),
            timeout: Duration::from_secs(10),
            max_attempts: 3,
            backoff_base: Duration::from_millis(500),
            backoff_max: Duration::from_secs(8),
            total_timeout: None,
            user_agent: format!("cepis-rust/{} (+https://cep.is)", crate::VERSION),
        }
    }
}

impl ClientBuilder {
    /// Where the API lives.
    ///
    /// Point it at a local instance (the API repo brings one up on
    /// `http://localhost:8080` with `just up`) or at a test server. It is the
    /// only setting that usually changes between environments. A trailing slash
    /// is stripped, so it cannot produce doubled separators that some proxies
    /// normalise and others 404.
    pub fn base_url(mut self, value: impl Into<String>) -> Self {
        self.base_url = value.into();
        self
    }

    /// How long to wait **per socket operation** (connect, and each read).
    ///
    /// This is not a cap on the whole call — that cap is
    /// [`total_timeout`](ClientBuilder::total_timeout).
    pub fn timeout(mut self, value: Duration) -> Self {
        self.timeout = value;
        self
    }

    /// How many attempts in total, counting the first.
    ///
    /// `1` disables retrying. The default of `3` covers the common case (one
    /// transient failure) without turning a real outage into a long wait.
    pub fn max_attempts(mut self, value: u32) -> Self {
        self.max_attempts = value;
        self
    }

    /// Base of the exponential wait.
    ///
    /// The actual wait is drawn from `[0, base * 2^(n-1)]` — *full jitter*. The
    /// randomness is not decoration: without it, N clients that fail together
    /// come back together and reproduce the spike that took the service down.
    pub fn backoff_base(mut self, value: Duration) -> Self {
        self.backoff_base = value;
        self
    }

    /// Cap on each individual wait.
    ///
    /// It applies to a server-sent `Retry-After` too, or a `Retry-After: 3600`
    /// from a misconfigured proxy would park the thread for an hour.
    pub fn backoff_max(mut self, value: Duration) -> Self {
        self.backoff_max = value;
        self
    }

    /// Cap on the whole call, attempts and waits included.
    ///
    /// `None` leaves [`max_attempts`](ClientBuilder::max_attempts) as the only
    /// bound. Set it in code serving a web request: without it the worst case
    /// is `max_attempts * timeout` plus the sum of the backoffs.
    pub fn total_timeout(mut self, value: Option<Duration>) -> Self {
        self.total_timeout = value;
        self
    }

    /// Override the `User-Agent`.
    ///
    /// The default identifies the library and its version, which is what lets
    /// the server diagnose an old client misbehaving.
    pub fn user_agent(mut self, value: impl Into<String>) -> Self {
        self.user_agent = value.into();
        self
    }

    /// Build the client.
    ///
    /// # Panics
    ///
    /// If `max_attempts` is 0 or `timeout` is zero — both are configuration
    /// mistakes that can only produce a client which never completes a call.
    pub fn build(self) -> Client {
        assert!(self.max_attempts >= 1, "max_attempts must be >= 1");
        assert!(!self.timeout.is_zero(), "timeout must be > 0");

        let config = ureq::Agent::config_builder()
            .http_status_as_error(false)
            .user_agent(self.user_agent.as_str())
            .accept(ACCEPT)
            .accept_encoding("identity")
            .timeout_connect(Some(self.timeout))
            .timeout_send_request(Some(self.timeout))
            .timeout_recv_response(Some(self.timeout))
            .timeout_recv_body(Some(self.timeout))
            .build();

        Client {
            base_url: self.base_url.trim_end_matches('/').to_owned(),
            timeout: self.timeout,
            max_attempts: self.max_attempts,
            backoff_base: self.backoff_base,
            backoff_max: self.backoff_max,
            total_timeout: self.total_timeout,
            user_agent: self.user_agent,
            agent: ureq::Agent::new_with_config(config),
        }
    }
}

/// Client for the cep.is API.
///
/// Build one and reuse it: the configuration is immutable, no method keeps
/// state between calls, and the underlying connection pool is shared — so a
/// second lookup on the same host skips the TLS handshake.
///
/// ```no_run
/// use std::time::Duration;
///
/// let client = cepis::Client::builder()
///     .timeout(Duration::from_secs(5))
///     .total_timeout(Some(Duration::from_secs(15)))
///     .build();
///
/// let address = client.lookup("01001000")?;
/// println!("{} {}", address.localidade, address.uf);
/// # Ok::<_, cepis::Error>(())
/// ```
#[derive(Debug, Clone)]
pub struct Client {
    base_url: String,
    timeout: Duration,
    max_attempts: u32,
    backoff_base: Duration,
    backoff_max: Duration,
    total_timeout: Option<Duration>,
    user_agent: String,
    agent: ureq::Agent,
}

impl Default for Client {
    fn default() -> Self {
        Client::new()
    }
}

impl Client {
    /// A client with the default configuration, pointing at <https://cep.is>.
    pub fn new() -> Client {
        ClientBuilder::default().build()
    }

    /// Start configuring a client.
    pub fn builder() -> ClientBuilder {
        ClientBuilder::default()
    }

    /// The API URL in use, without a trailing slash.
    pub fn base_url(&self) -> &str {
        &self.base_url
    }

    /// The `User-Agent` sent with every request.
    pub fn user_agent(&self) -> &str {
        &self.user_agent
    }

    /// The per-socket-operation timeout.
    pub fn timeout(&self) -> Duration {
        self.timeout
    }

    /// How many attempts each call makes at most, counting the first.
    pub fn max_attempts(&self) -> u32 {
        self.max_attempts
    }

    /// The cap on the whole call, if one is set.
    pub fn total_timeout(&self) -> Option<Duration> {
        self.total_timeout
    }

    /// Look up a single CEP.
    ///
    /// The CEP may be masked or not. This is the only way to obtain a
    /// coordinate — see [`lookup_many`](Client::lookup_many) for why.
    ///
    /// A 400 from the API is surfaced as [`Error::InvalidCep`] carrying the
    /// API's own message: the format already passed local validation, so the
    /// server knows something this client does not.
    ///
    /// # Errors
    ///
    /// - [`Error::InvalidCep`] — not 8 digits (no request is made).
    /// - [`Error::NotFound`] — well-formed, but not in the database. It carries
    ///   a [`Faixa`] naming the municipality the CEP would belong to, whenever
    ///   the API can place it in one.
    /// - [`Error::RateLimited`] — 429 after every attempt.
    /// - [`Error::Server`] — 5xx after every attempt.
    /// - [`Error::Network`] — the request never completed.
    /// - [`Error::InvalidResponse`] — the body could not be read as JSON.
    pub fn lookup(&self, cep: &str) -> Result<Address> {
        let number = normalize_cep(cep)?;
        let (status, body) = self.request(&format!("/{number}"), None, &[200])?;

        if status == 404 {
            return Err(Error::NotFound {
                cep: number,
                faixa: faixa(&body),
            });
        }
        if status == 400 {
            return Err(Error::invalid_cep(error_message(
                &body,
                &format!("CEP {number} rejected"),
            )));
        }

        Ok(Address::from_json(json_object(&body)?))
    }

    /// Look up many CEPs in one batch.
    ///
    /// Returns a `Vec` of **the same length and in the same order** as the
    /// input. A CEP that does not exist becomes `None` at its position — it
    /// does not vanish from the list, and it is not an error. That is the right
    /// difference from [`lookup`](Client::lookup): in a batch of a thousand,
    /// one bad CEP must not cost the other nine hundred and ninety-nine.
    ///
    /// A malformed CEP also becomes `None`, for the same reason. If you need to
    /// know *which* one was malformed, validate first with
    /// [`normalize_cep`].
    ///
    /// Input is chunked into blocks of [`MAX_BATCH`] automatically. Do not loop
    /// over [`lookup`](Client::lookup) instead: 500 calls pay 500 round trips,
    /// the batch pays one.
    ///
    /// # Warning
    ///
    /// The batch carries **no coordinate**. The endpoint behind it is CSV, and
    /// the CSV has no `geo` column. Every [`Address`] returned here has
    /// `geo: None`, including for CEPs that do have one. Need `geo`? Use
    /// [`lookup`](Client::lookup), one at a time.
    ///
    /// # Errors
    ///
    /// Transport failures only — [`Error::RateLimited`], [`Error::Server`],
    /// [`Error::Network`], [`Error::InvalidResponse`]. A bad CEP never fails a
    /// batch.
    pub fn lookup_many(&self, ceps: &[impl AsRef<str>]) -> Result<Vec<Option<Address>>> {
        if ceps.is_empty() {
            return Ok(Vec::new());
        }

        let mut result = Vec::with_capacity(ceps.len());
        for chunk in ceps.chunks(MAX_BATCH) {
            result.extend(self.lookup_chunk(chunk)?);
        }
        Ok(result)
    }

    /// Query `GET /health`: does the API answer, and with which database.
    ///
    /// An API reporting itself degraded is not an error — that arrives in
    /// [`Health::status`], and checking health is exactly when you want the
    /// diagnosis rather than a failure. Network failures still fail.
    pub fn health(&self) -> Result<Health> {
        let (_, body) = self.request("/health", None, &[200, 503])?;
        Ok(Health::from_json(json_object(&body)?))
    }

    /// One chunk of at most [`MAX_BATCH`] CEPs, in a single request.
    ///
    /// A blank CSV row means "not found": the API fills only the `cep` column
    /// and leaves the rest empty, precisely so it does not misalign whoever is
    /// consuming the batch in a spreadsheet.
    ///
    /// The API guarantees one row per requested CEP. Should that ever fail to
    /// hold — a proxy truncating, a cut response — the result is padded to the
    /// input length rather than returned short, because a shifted list silently
    /// pairs the wrong address with the wrong CEP.
    fn lookup_chunk(&self, ceps: &[impl AsRef<str>]) -> Result<Vec<Option<Address>>> {
        let joined = ceps
            .iter()
            .map(|c| c.as_ref())
            .collect::<Vec<_>>()
            .join(",");
        let (_, body) = self.request("/csv", Some(("ceps", joined)), &[200])?;

        let text = String::from_utf8_lossy(&body);
        let records = csv::parse(&text);
        let mut rows: Vec<Option<Address>> = Vec::with_capacity(ceps.len());

        if let Some((header, data)) = records.split_first() {
            for record in data {
                let mut object = Map::new();
                for (column, value) in header.iter().zip(record.iter()) {
                    object.insert(column.clone(), Value::String(value.clone()));
                }
                let known = object
                    .get("localidade")
                    .and_then(Value::as_str)
                    .is_some_and(|v| !v.trim().is_empty());
                rows.push(known.then(|| Address::from_json(Value::Object(object))));
            }
        }

        rows.resize(ceps.len(), None);
        Ok(rows)
    }

    /// Perform the GET, with retries. Returns `(status, body)`.
    ///
    /// *Definitive* error statuses (400, 404) come back normally, for the
    /// caller to turn into the right error — only transient failures become
    /// errors here, and only once the attempts are exhausted.
    fn request(
        &self,
        path: &str,
        query: Option<(&str, String)>,
        accepted: &[u16],
    ) -> Result<(u16, Vec<u8>)> {
        let url = format!("{}{}", self.base_url, path);
        let deadline = self.total_timeout.map(|cap| Instant::now() + cap);

        let mut last_status: Option<u16> = None;
        let mut last_retry_after: Option<Duration> = None;
        let mut last_network: Option<ureq::Error> = None;

        for attempt in 1..=self.max_attempts {
            let remaining = match deadline {
                Some(d) => {
                    let left = d.saturating_duration_since(Instant::now());
                    if left.is_zero() {
                        break;
                    }
                    Some(left)
                }
                None => None,
            };

            let mut request = self.agent.get(&url);
            if let Some((key, value)) = &query {
                request = request.query(key, value);
            }
            if let Some(left) = remaining {
                request = request.config().timeout_global(Some(left)).build();
            }

            match request.call() {
                Ok(mut response) => {
                    let status = response.status().as_u16();
                    let retry_after = response
                        .headers()
                        .get("retry-after")
                        .and_then(|v| v.to_str().ok())
                        .and_then(parse_retry_after);

                    match response.body_mut().read_to_vec() {
                        Ok(body) => {
                            if accepted.contains(&status) || !RETRIABLE_STATUS.contains(&status) {
                                return Ok((status, body));
                            }
                            last_status = Some(status);
                            last_retry_after = retry_after;
                            last_network = None;
                        }
                        Err(err) => {
                            last_status = None;
                            last_retry_after = None;
                            last_network = Some(err);
                        }
                    }
                }
                Err(ureq::Error::BadUri(detail)) => {
                    return Err(Error::network(
                        format!("{url} is not a usable URL: {detail}"),
                        None,
                    ));
                }
                Err(err) => {
                    last_status = None;
                    last_retry_after = None;
                    last_network = Some(err);
                }
            }

            if attempt == self.max_attempts {
                break;
            }

            let mut wait = self.backoff(attempt, last_retry_after);
            if let Some(d) = deadline {
                let left = d.saturating_duration_since(Instant::now());
                if left.is_zero() {
                    break;
                }
                wait = wait.min(left);
            }
            std::thread::sleep(wait);
        }

        let attempts = self.max_attempts;
        match (last_status, last_network) {
            (Some(429), _) => Err(Error::RateLimited {
                message: format!("rate limit reached at {url} (after {attempts} attempt(s))"),
                retry_after: last_retry_after,
            }),
            (Some(status), _) => Err(Error::Server {
                message: format!(
                    "the API answered {status} at {url} (after {attempts} attempt(s))"
                ),
                status,
            }),
            (None, source) => {
                let detail = source
                    .as_ref()
                    .map(|e| e.to_string())
                    .unwrap_or_else(|| "no attempt was made within the total timeout".to_owned());
                Err(Error::network(
                    format!("could not reach {url} (after {attempts} attempt(s)): {detail}"),
                    source.map(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>),
                ))
            }
        }
    }

    /// How long to wait before the next attempt.
    ///
    /// The server's `Retry-After` wins over the local calculation when present:
    /// it knows when it will be back, we are guessing. It still goes through
    /// the `backoff_max` cap. Otherwise the wait is drawn from `[0, ceiling]`
    /// rather than being the ceiling.
    fn backoff(&self, attempt: u32, retry_after: Option<Duration>) -> Duration {
        if let Some(asked) = retry_after {
            return asked.min(self.backoff_max);
        }
        let ceiling = self
            .backoff_base
            .saturating_mul(1u32 << (attempt - 1).min(31))
            .min(self.backoff_max);
        jitter(ceiling)
    }
}

/// Decode a JSON object body, or fail with [`Error::InvalidResponse`].
fn json_object(body: &[u8]) -> Result<Value> {
    let value: Value = serde_json::from_slice(body).map_err(|err| {
        Error::invalid_response(format!(
            "response is not valid JSON — something between you and the API is \
             answering instead of it: {}: {err}",
            preview(body)
        ))
    })?;

    if value.is_object() {
        Ok(value)
    } else {
        Err(Error::invalid_response(format!(
            "expected a JSON object, got {}",
            preview(body)
        )))
    }
}

/// Pull `mensagem` out of an error body, falling back to a default.
fn error_message(body: &[u8], fallback: &str) -> String {
    serde_json::from_slice::<Value>(body)
        .ok()
        .as_ref()
        .and_then(|v| v.get("mensagem"))
        .and_then(Value::as_str)
        .filter(|m| !m.is_empty())
        .map(str::to_owned)
        .unwrap_or_else(|| fallback.to_owned())
}

/// Pull the `faixa` block out of a 404 body, if there is a readable one.
///
/// Every step is allowed to fail into `None`: a body that is not JSON, a
/// missing or null `faixa`, an object that names no municipality. The caller is
/// on its way to returning [`Error::NotFound`] either way, and an unparseable
/// extra field must not turn a correct not-found into an
/// [`Error::InvalidResponse`].
fn faixa(body: &[u8]) -> Option<Box<Faixa>> {
    serde_json::from_slice::<Value>(body)
        .ok()
        .as_ref()
        .and_then(|v| v.get("faixa"))
        .and_then(Faixa::from_json)
        .map(Box::new)
}

fn preview(body: &[u8]) -> String {
    let text = String::from_utf8_lossy(body);
    let cut = text
        .char_indices()
        .nth(120)
        .map(|(i, _)| i)
        .unwrap_or(text.len());
    format!("{:?}", &text[..cut])
}