//! 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, 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 { 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, 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) -> 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) -> 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) -> 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, 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 . 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 { 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. /// - [`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
{ let number = normalize_cep(cep)?; let (status, body) = self.request(&format!("/{number}"), None, &[200])?; if status == 404 { return Err(Error::NotFound { cep: number }); } 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]) -> Result>> { 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 { 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]) -> Result>> { let joined = ceps .iter() .map(|c| c.as_ref()) .collect::>() .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> = 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)> { let url = format!("{}{}", self.base_url, path); let deadline = self.total_timeout.map(|cap| Instant::now() + cap); let mut last_status: Option = None; let mut last_retry_after: Option = None; let mut last_network: Option = 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), )) } } } /// 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 { 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 { 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::(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()) } 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]) }