hibp_sync_client/
client.rs

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

hibp_sync_client/
client.rs
