diff options
Diffstat (limited to 'src/client.rs')
| -rw-r--r-- | src/client.rs | 559 |
1 files changed, 559 insertions, 0 deletions
diff --git a/src/client.rs b/src/client.rs new file mode 100644 index 0000000..8d73f06 --- /dev/null +++ b/src/client.rs @@ -0,0 +1,559 @@ +//! 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<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. + /// - [`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 }); + } + 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()) +} + +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]) +} |