hibp_sync_client/
client.rs

1use std::time::Duration;
2
3use bytes::{BufMut as _, Bytes, BytesMut};
4use chrono::{DateTime, Utc};
5use compact_str::CompactString;
6use futures_util::StreamExt;
7use http_body_util::{BodyExt as _, Empty};
8use hyper_util::client::legacy::connect::HttpConnector;
9use hyper_util::rt::TokioExecutor;
10use serde::Deserialize;
11use tokio::io::AsyncRead;
12
13use crate::error::Error;
14
15const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
16const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
17const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
18
19static STATUS_PATH: &[u8] = b"/v1/status";
20static CHANGED_PATH: &[u8] = b"/v1/changed";
21static SEGMENT_PREFIX: &[u8] = b"/v1/segment?segment=";
22static OF_PARAM: &[u8] = b"&of=";
23static SINCE_PARAM: &[u8] = b"&since=";
24
25// Max path+query: "/v1/segment?segment=254&of=255&since=2026-01-01T00:00:00Z" = 57 bytes
26const SEGMENT_BUF_CAP: usize = 64;
27
28pub struct Client {
29    scheme: http::uri::Scheme,
30    authority: http::uri::Authority,
31    http_client: hyper_util::client::legacy::Client<HttpConnector, Empty<Bytes>>,
32}
33
34#[derive(Deserialize)]
35pub struct Status {
36    pub last_updated: Option<DateTime<Utc>>,
37}
38
39#[derive(Deserialize)]
40pub struct Changed {
41    pub last_updated: Option<DateTime<Utc>>,
42    pub prev_last_updated: Option<DateTime<Utc>>,
43    pub prefixes: Vec<CompactString>,
44}
45
46impl Client {
47    pub fn new(base_url: &http::Uri) -> Result<Self, Error> {
48        let scheme = base_url.scheme().ok_or(Error::InvalidServerUrl("missing scheme"))?.clone();
49        let authority = base_url
50            .authority()
51            .ok_or(Error::InvalidServerUrl("missing authority"))?
52            .clone();
53        let mut connector = HttpConnector::new();
54        connector.set_connect_timeout(Some(CONNECT_TIMEOUT));
55        connector.set_keepalive(Some(KEEPALIVE_INTERVAL));
56        let http_client =
57            hyper_util::client::legacy::Client::builder(TokioExecutor::new()).build(connector);
58        Ok(Self { scheme, authority, http_client })
59    }
60
61    fn uri(&self, pq: Bytes) -> http::Uri {
62        let mut parts = http::uri::Parts::default();
63        parts.scheme = Some(self.scheme.clone());
64        parts.authority = Some(self.authority.clone());
65        parts.path_and_query = Some(
66            http::uri::PathAndQuery::from_maybe_shared(pq)
67                .expect("caller ensures valid path-and-query bytes"),
68        );
69        http::Uri::from_parts(parts).expect("Uri from all parts is infallible")
70    }
71
72    async fn get(&self, pq: Bytes) -> Result<Bytes, Error> {
73        let req = hyper::Request::get(self.uri(pq)).body(Empty::new())?;
74        let resp = tokio::time::timeout(REQUEST_TIMEOUT, self.http_client.request(req))
75            .await
76            .map_err(|_| Error::Timeout("response headers"))?
77            .map_err(Error::Request)?;
78        if !resp.status().is_success() {
79            return Err(Error::HttpStatus(resp.status()));
80        }
81        let body = tokio::time::timeout(REQUEST_TIMEOUT, resp.into_body().collect())
82            .await
83            .map_err(|_| Error::Timeout("response body"))?
84            .map_err(Error::ResponseRead)?;
85        Ok(body.to_bytes())
86    }
87
88    #[tracing::instrument(skip_all)]
89    pub async fn status(&self) -> Result<Status, Error> {
90        let bytes = self.get(Bytes::from_static(STATUS_PATH)).await?;
91        Ok(serde_json::from_slice(&bytes)?)
92    }
93
94    #[tracing::instrument(skip_all)]
95    pub async fn changed(&self) -> Result<Changed, Error> {
96        let bytes = self.get(Bytes::from_static(CHANGED_PATH)).await?;
97        Ok(serde_json::from_slice(&bytes)?)
98    }
99
100    #[tracing::instrument(skip(self))]
101    pub async fn segment_stream(
102        &self,
103        segment: u8,
104        of: u8,
105        since: Option<&str>,
106    ) -> Result<impl AsyncRead + Unpin + Send + 'static, Error> {
107        let since_bytes = since.map_or(&[][..], str::as_bytes);
108        let mut buf = BytesMut::with_capacity(SEGMENT_BUF_CAP + since_bytes.len());
109        buf.extend_from_slice(SEGMENT_PREFIX);
110        write_decimal(&mut buf, segment);
111        buf.extend_from_slice(OF_PARAM);
112        write_decimal(&mut buf, of);
113        if !since_bytes.is_empty() {
114            buf.extend_from_slice(SINCE_PARAM);
115            buf.extend_from_slice(since_bytes);
116        }
117
118        let req = hyper::Request::get(self.uri(buf.freeze())).body(Empty::new())?;
119        tracing::debug!("sending request");
120        let resp = tokio::time::timeout(REQUEST_TIMEOUT, self.http_client.request(req))
121            .await
122            .map_err(|_| Error::Timeout("segment response headers"))?
123            .map_err(Error::Request)?;
124        tracing::debug!(status = %resp.status(), "response headers received");
125
126        if resp.status() == http::StatusCode::SERVICE_UNAVAILABLE {
127            return Err(Error::ServerBusy);
128        }
129        if !resp.status().is_success() {
130            return Err(Error::HttpStatus(resp.status()));
131        }
132
133        let body_stream = resp
134            .into_body()
135            .into_data_stream()
136            .map(|res| res.map_err(std::io::Error::other));
137        let reader = tokio_util::io::StreamReader::new(body_stream);
138        let decoder = async_compression::tokio::bufread::ZstdDecoder::new(reader);
139        Ok(decoder)
140    }
141}
142
143fn write_decimal(buf: &mut BytesMut, mut n: u8) {
144    let start = buf.len();
145    let mut count = 0;
146
147    loop {
148        buf.put_u8(b'0' + (n % 10));
149        count += 1;
150
151        n /= 10;
152        if n == 0 {
153            break;
154        }
155    }
156
157    buf[start..start + count].reverse();
158}
159
160#[cfg(test)]
161mod tests {
162    use super::*;
163
164    #[test]
165    fn segment_uri_no_since() {
166        let client = Client::new(&"http://127.0.0.1:8765".parse().unwrap()).unwrap();
167        let mut buf = BytesMut::with_capacity(SEGMENT_BUF_CAP);
168        buf.extend_from_slice(SEGMENT_PREFIX);
169        write_decimal(&mut buf, 3);
170        buf.extend_from_slice(OF_PARAM);
171        write_decimal(&mut buf, 16);
172        let uri = client.uri(buf.freeze());
173        assert_eq!(
174            uri.to_string(),
175            "http://127.0.0.1:8765/v1/segment?segment=3&of=16"
176        );
177    }
178
179    #[test]
180    fn segment_uri_with_since() {
181        let client = Client::new(&"http://127.0.0.1:8765".parse().unwrap()).unwrap();
182        let mut buf = BytesMut::with_capacity(SEGMENT_BUF_CAP);
183        buf.extend_from_slice(SEGMENT_PREFIX);
184        write_decimal(&mut buf, 0);
185        buf.extend_from_slice(OF_PARAM);
186        write_decimal(&mut buf, 1);
187        buf.extend_from_slice(SINCE_PARAM);
188        buf.extend_from_slice(b"2026-01-01T00:00:00Z");
189        let uri = client.uri(buf.freeze());
190        assert_eq!(
191            uri.to_string(),
192            "http://127.0.0.1:8765/v1/segment?segment=0&of=1&since=2026-01-01T00:00:00Z"
193        );
194    }
195
196    #[test]
197    fn new_rejects_missing_scheme() {
198        assert!(Client::new(&http::Uri::from_static("/no-scheme")).is_err());
199    }
200}
hibp_sync_client/client.rs

hibp_sync_client/
client.rs
