hyphae_client/
lib.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Bounded asynchronous Rust client for the public Hyphae HTTP API.
4//!
5//! This crate depends only on public wire contracts. It does not import the
6//! server, engine, storage, query, or retrieval implementations.
7
8use std::time::Duration;
9
10use hyphae_contracts::v1::{
11    CapabilitiesV1, CommitReceiptV1, DefineLexicalIndexRequestV1, DefineVectorSpaceRequestV1,
12    DeleteRequestV1, DeleteVectorsRequestV1, ErrorV1, ExactRetrievalRequestV1,
13    ExactRetrievalResponseV1, GetRequestV1, GetResponseV1, HealthV1, HybridRetrievalRequestV1,
14    HybridRetrievalResponseV1, LexicalRetrievalRequestV1, LexicalRetrievalResponseV1, ProofV1,
15    PutRequestV1, PutVectorsRequestV1, QueryRequestV1, QueryResponseV1, RetrievalProofV1,
16};
17use reqwest::{
18    Method, StatusCode, Url,
19    header::{self, HeaderMap, HeaderValue},
20};
21use serde::{Serialize, de::DeserializeOwned};
22use thiserror::Error;
23
24const DEFAULT_RESPONSE_BYTES: usize = 32 * 1024 * 1024;
25const DEFAULT_WITNESS_BYTES: usize = 512 * 1024 * 1024;
26const DEFAULT_TIMEOUT: Duration = Duration::from_secs(60);
27
28/// Successful typed API value and its response correlation identifier.
29#[derive(Clone, Debug, Eq, PartialEq)]
30pub struct ApiResponse<T> {
31    /// Decoded public contract value.
32    pub value: T,
33    /// UUID copied from `X-Request-Id`.
34    pub request_id: String,
35}
36
37/// Stable server-declared error received from `/v1`.
38#[derive(Clone, Debug, Error, Eq, PartialEq)]
39#[error("Hyphae API returned HTTP {status} {code} (request {request_id})")]
40pub struct ApiFailure {
41    /// HTTP status code.
42    pub status: u16,
43    /// Stable machine-readable error code.
44    pub code: String,
45    /// Bounded server diagnostic.
46    pub message: String,
47    /// UUID matching the response header.
48    pub request_id: String,
49}
50
51/// Invalid client configuration detected before any network operation.
52#[derive(Clone, Debug, Error, Eq, PartialEq)]
53pub enum ClientConfigError {
54    /// Base URL is not syntactically valid.
55    #[error("invalid Hyphae base URL")]
56    InvalidBaseUrl,
57    /// Only explicit HTTP and HTTPS origins are supported.
58    #[error("Hyphae base URL must use http or https")]
59    UnsupportedScheme,
60    /// Base URL must be one origin without credentials, query, fragment, or path prefix.
61    #[error("Hyphae base URL must be an origin without credentials, query, fragment, or path")]
62    NonOriginBaseUrl,
63    /// Bearer secret cannot be represented safely in one header.
64    #[error("invalid bearer token for an HTTP authorization header")]
65    InvalidBearerToken,
66    /// Every local client bound must be positive.
67    #[error("client timeout and response limits must be nonzero")]
68    ZeroLimit,
69    /// Underlying HTTP client construction failed.
70    #[error("failed to construct HTTP client")]
71    HttpClient,
72}
73
74/// Transport, bound, envelope, or declared API failure.
75#[derive(Debug, Error)]
76pub enum ClientError {
77    /// HTTP transport failed before a complete bounded response arrived.
78    #[error("Hyphae HTTP transport failed: {0}")]
79    Transport(#[from] reqwest::Error),
80    /// Response declared or exceeded a local byte bound.
81    #[error("Hyphae response exceeded local limit {maximum} bytes")]
82    ResponseTooLarge {
83        /// Configured maximum.
84        maximum: usize,
85    },
86    /// A JSON operation returned a non-JSON media type.
87    #[error("Hyphae response did not use a JSON content type")]
88    InvalidContentType,
89    /// JSON response did not match the public versioned contract.
90    #[error("Hyphae response violated the version 1 contract: {0}")]
91    InvalidJson(#[from] serde_json::Error),
92    /// The request ID header was missing, duplicated, or malformed.
93    #[error("Hyphae response has no single valid X-Request-Id header")]
94    InvalidRequestId,
95    /// Error envelope and header correlation identifiers disagree.
96    #[error("Hyphae error envelope request ID differs from its response header")]
97    RequestIdMismatch,
98    /// Server returned a non-success response that conformed to `ErrorV1`.
99    #[error(transparent)]
100    Api(#[from] ApiFailure),
101    /// A successful operation returned an unexpected HTTP status.
102    #[error("Hyphae returned unexpected success status {0}")]
103    UnexpectedStatus(u16),
104    /// Proof witness reference was not canonical for its proof identity.
105    #[error("proof contains a noncanonical witness reference")]
106    InvalidWitnessReference,
107    /// Downloaded witness digest header differs from the proof.
108    #[error("downloaded witness digest header differs from the proof")]
109    WitnessDigestMismatch,
110    /// Downloaded witness length differs from the proof.
111    #[error("downloaded witness length differs from the proof")]
112    WitnessLengthMismatch,
113}
114
115/// Builder for a bounded public API client.
116#[derive(Clone, Debug)]
117#[must_use = "a client builder has no effect until build is called"]
118pub struct ClientBuilder {
119    base_url: Url,
120    bearer_token: Option<HeaderValue>,
121    timeout: Duration,
122    response_bytes: usize,
123    witness_bytes: usize,
124}
125
126impl ClientBuilder {
127    /// Parses one root HTTP(S) origin.
128    ///
129    /// # Errors
130    ///
131    /// Rejects malformed URLs, non-HTTP schemes, credentials, query strings,
132    /// fragments, and non-root paths.
133    pub fn new(base_url: &str) -> Result<Self, ClientConfigError> {
134        let mut base_url = Url::parse(base_url).map_err(|_| ClientConfigError::InvalidBaseUrl)?;
135        if !matches!(base_url.scheme(), "http" | "https") {
136            return Err(ClientConfigError::UnsupportedScheme);
137        }
138        if !base_url.username().is_empty()
139            || base_url.password().is_some()
140            || base_url.query().is_some()
141            || base_url.fragment().is_some()
142            || !matches!(base_url.path(), "" | "/")
143        {
144            return Err(ClientConfigError::NonOriginBaseUrl);
145        }
146        base_url.set_path("/");
147        Ok(Self {
148            base_url,
149            bearer_token: None,
150            timeout: DEFAULT_TIMEOUT,
151            response_bytes: DEFAULT_RESPONSE_BYTES,
152            witness_bytes: DEFAULT_WITNESS_BYTES,
153        })
154    }
155
156    /// Configures an opaque bearer token without retaining a second copy.
157    ///
158    /// # Errors
159    ///
160    /// Rejects values that cannot be represented in one HTTP header.
161    pub fn bearer_token(mut self, token: &str) -> Result<Self, ClientConfigError> {
162        if token.is_empty() {
163            return Err(ClientConfigError::InvalidBearerToken);
164        }
165        let mut value = HeaderValue::from_str(&format!("Bearer {token}"))
166            .map_err(|_| ClientConfigError::InvalidBearerToken)?;
167        value.set_sensitive(true);
168        self.bearer_token = Some(value);
169        Ok(self)
170    }
171
172    /// Sets the complete request/response deadline.
173    pub fn timeout(mut self, timeout: Duration) -> Self {
174        self.timeout = timeout;
175        self
176    }
177
178    /// Sets the maximum complete JSON response bytes.
179    pub fn response_bytes(mut self, maximum: usize) -> Self {
180        self.response_bytes = maximum;
181        self
182    }
183
184    /// Sets the maximum complete snapshot witness bytes.
185    pub fn witness_bytes(mut self, maximum: usize) -> Self {
186        self.witness_bytes = maximum;
187        self
188    }
189
190    /// Constructs the reusable client.
191    ///
192    /// # Errors
193    ///
194    /// Rejects zero limits or an underlying HTTP client configuration error.
195    pub fn build(self) -> Result<HyphaeClient, ClientConfigError> {
196        if self.timeout.is_zero() || self.response_bytes == 0 || self.witness_bytes == 0 {
197            return Err(ClientConfigError::ZeroLimit);
198        }
199        let http = reqwest::Client::builder()
200            .timeout(self.timeout)
201            .build()
202            .map_err(|_| ClientConfigError::HttpClient)?;
203        Ok(HyphaeClient {
204            http,
205            base_url: self.base_url,
206            bearer_token: self.bearer_token,
207            response_bytes: self.response_bytes,
208            witness_bytes: self.witness_bytes,
209        })
210    }
211}
212
213/// Reusable bounded client for Hyphae HTTP API version 1.
214#[derive(Clone, Debug)]
215pub struct HyphaeClient {
216    http: reqwest::Client,
217    base_url: Url,
218    bearer_token: Option<HeaderValue>,
219    response_bytes: usize,
220    witness_bytes: usize,
221}
222
223impl HyphaeClient {
224    /// Starts a client builder for one root origin.
225    ///
226    /// # Errors
227    ///
228    /// Returns a base-URL validation error.
229    pub fn builder(base_url: &str) -> Result<ClientBuilder, ClientConfigError> {
230        ClientBuilder::new(base_url)
231    }
232
233    /// Reports server capabilities and effective limits.
234    ///
235    /// # Errors
236    ///
237    /// Returns a transport, bound, contract, correlation, or API error.
238    pub async fn capabilities(&self) -> Result<ApiResponse<CapabilitiesV1>, ClientError> {
239        self.get_json("v1/capabilities", false).await
240    }
241
242    /// Reports process liveness.
243    ///
244    /// # Errors
245    ///
246    /// Returns a transport, bound, contract, correlation, or API error.
247    pub async fn liveness(&self) -> Result<ApiResponse<HealthV1>, ClientError> {
248        self.get_json("v1/health/live", false).await
249    }
250
251    /// Reports engine readiness.
252    ///
253    /// # Errors
254    ///
255    /// Returns a transport, bound, contract, correlation, or API error.
256    pub async fn readiness(&self) -> Result<ApiResponse<HealthV1>, ClientError> {
257        self.get_json("v1/health/ready", false).await
258    }
259
260    /// Atomically stores a structured-record batch.
261    ///
262    /// # Errors
263    ///
264    /// Returns a transport, bound, contract, correlation, or API error.
265    pub async fn put(
266        &self,
267        request: &PutRequestV1,
268    ) -> Result<ApiResponse<CommitReceiptV1>, ClientError> {
269        self.post_json("v1/kv/put", request).await
270    }
271
272    /// Atomically deletes a binary-key batch.
273    ///
274    /// # Errors
275    ///
276    /// Returns a transport, bound, contract, correlation, or API error.
277    pub async fn delete(
278        &self,
279        request: &DeleteRequestV1,
280    ) -> Result<ApiResponse<CommitReceiptV1>, ClientError> {
281        self.post_json("v1/kv/delete", request).await
282    }
283
284    /// Gets proven key presence or absence.
285    ///
286    /// # Errors
287    ///
288    /// Returns a transport, bound, contract, correlation, or API error.
289    pub async fn get(
290        &self,
291        request: &GetRequestV1,
292    ) -> Result<ApiResponse<GetResponseV1>, ClientError> {
293        self.post_json("v1/kv/get", request).await
294    }
295
296    /// Executes a deterministic proof-bearing structured query.
297    ///
298    /// # Errors
299    ///
300    /// Returns a transport, bound, contract, correlation, or API error.
301    pub async fn query(
302        &self,
303        request: &QueryRequestV1,
304    ) -> Result<ApiResponse<QueryResponseV1>, ClientError> {
305        self.post_json("v1/query", request).await
306    }
307
308    /// Defines or exactly reuses one immutable durable vector space.
309    ///
310    /// # Errors
311    ///
312    /// Returns a transport, bound, contract, correlation, or API error.
313    pub async fn define_vector_space(
314        &self,
315        request: &DefineVectorSpaceRequestV1,
316    ) -> Result<ApiResponse<CommitReceiptV1>, ClientError> {
317        self.post_json("v1/vector-spaces/define", request).await
318    }
319
320    /// Atomically stores one durable vector batch.
321    ///
322    /// # Errors
323    ///
324    /// Returns a transport, bound, contract, correlation, or API error.
325    pub async fn put_vectors(
326        &self,
327        request: &PutVectorsRequestV1,
328    ) -> Result<ApiResponse<CommitReceiptV1>, ClientError> {
329        self.post_json("v1/vectors/put", request).await
330    }
331
332    /// Atomically deletes one durable vector batch.
333    ///
334    /// # Errors
335    ///
336    /// Returns a transport, bound, contract, correlation, or API error.
337    pub async fn delete_vectors(
338        &self,
339        request: &DeleteVectorsRequestV1,
340    ) -> Result<ApiResponse<CommitReceiptV1>, ClientError> {
341        self.post_json("v1/vectors/delete", request).await
342    }
343
344    /// Executes proof-bearing exact durable vector retrieval.
345    ///
346    /// # Errors
347    ///
348    /// Returns a transport, bound, contract, correlation, or API error.
349    pub async fn retrieve_exact(
350        &self,
351        request: &ExactRetrievalRequestV1,
352    ) -> Result<ApiResponse<ExactRetrievalResponseV1>, ClientError> {
353        self.post_json("v1/retrieve/exact", request).await
354    }
355
356    /// Defines or exactly reuses one immutable provider-free lexical index.
357    ///
358    /// # Errors
359    ///
360    /// Returns a transport, bound, contract, correlation, or API error.
361    pub async fn define_lexical_index(
362        &self,
363        request: &DefineLexicalIndexRequestV1,
364    ) -> Result<ApiResponse<CommitReceiptV1>, ClientError> {
365        self.post_json("v1/lexical-indexes/define", request).await
366    }
367
368    /// Executes proof-bearing provider-free lexical retrieval.
369    ///
370    /// # Errors
371    ///
372    /// Returns a transport, bound, contract, correlation, or API error.
373    pub async fn retrieve_lexical(
374        &self,
375        request: &LexicalRetrievalRequestV1,
376    ) -> Result<ApiResponse<LexicalRetrievalResponseV1>, ClientError> {
377        self.post_json("v1/retrieve/lexical", request).await
378    }
379
380    /// Executes proof-bearing deterministic hybrid retrieval.
381    ///
382    /// # Errors
383    ///
384    /// Returns a transport, bound, contract, correlation, or API error.
385    pub async fn retrieve_hybrid(
386        &self,
387        request: &HybridRetrievalRequestV1,
388    ) -> Result<ApiResponse<HybridRetrievalResponseV1>, ClientError> {
389        self.post_json("v1/retrieve/hybrid", request).await
390    }
391
392    /// Downloads the exact snapshot witness referenced by a proof.
393    ///
394    /// # Errors
395    ///
396    /// Rejects a noncanonical reference, transport/bound/API failure, or a
397    /// digest header that disagrees with the proof.
398    pub async fn download_witness(
399        &self,
400        proof: &ProofV1,
401    ) -> Result<ApiResponse<Vec<u8>>, ClientError> {
402        self.download_witness_parts(
403            proof.checkpoint_sequence,
404            &proof.snapshot_digest,
405            &proof.witness.path,
406            proof.witness.file_bytes,
407        )
408        .await
409    }
410
411    /// Downloads the exact snapshot witness referenced by a retrieval proof.
412    ///
413    /// # Errors
414    ///
415    /// Rejects a noncanonical reference, transport/bound/API failure, or a
416    /// digest header that disagrees with the proof.
417    pub async fn download_retrieval_witness(
418        &self,
419        proof: &RetrievalProofV1,
420    ) -> Result<ApiResponse<Vec<u8>>, ClientError> {
421        self.download_witness_parts(
422            proof.checkpoint_sequence,
423            &proof.snapshot_digest,
424            &proof.witness.path,
425            proof.witness.file_bytes,
426        )
427        .await
428    }
429
430    async fn download_witness_parts(
431        &self,
432        checkpoint_sequence: u64,
433        snapshot_digest: &str,
434        witness_path: &str,
435        witness_bytes: u64,
436    ) -> Result<ApiResponse<Vec<u8>>, ClientError> {
437        let expected_path = format!("/v1/witnesses/{checkpoint_sequence}/{snapshot_digest}");
438        if witness_path != expected_path {
439            return Err(ClientError::InvalidWitnessReference);
440        }
441        if witness_bytes > u64::try_from(self.witness_bytes).unwrap_or(u64::MAX) {
442            return Err(ClientError::ResponseTooLarge {
443                maximum: self.witness_bytes,
444            });
445        }
446        let response = self
447            .request(Method::GET, expected_path.trim_start_matches('/'), true)
448            .send()
449            .await?;
450        if !response.status().is_success() {
451            return Err(self.decode_api_failure(response).await?);
452        }
453        if response.status() != StatusCode::OK {
454            return Err(ClientError::UnexpectedStatus(response.status().as_u16()));
455        }
456        let request_id = request_id(response.headers())?;
457        let expected_digest = format!("blake3={snapshot_digest}");
458        if single_header(response.headers(), "digest") != Some(expected_digest.as_str()) {
459            return Err(ClientError::WitnessDigestMismatch);
460        }
461        let value = read_bounded(response, self.witness_bytes).await?;
462        if u64::try_from(value.len()) != Ok(witness_bytes) {
463            return Err(ClientError::WitnessLengthMismatch);
464        }
465        Ok(ApiResponse { value, request_id })
466    }
467
468    async fn get_json<T: DeserializeOwned>(
469        &self,
470        path: &str,
471        authenticated: bool,
472    ) -> Result<ApiResponse<T>, ClientError> {
473        let response = self
474            .request(Method::GET, path, authenticated)
475            .send()
476            .await?;
477        self.decode_json(response).await
478    }
479
480    async fn post_json<RequestBody: Serialize, ResponseBody: DeserializeOwned>(
481        &self,
482        path: &str,
483        request: &RequestBody,
484    ) -> Result<ApiResponse<ResponseBody>, ClientError> {
485        let response = self
486            .request(Method::POST, path, true)
487            .json(request)
488            .send()
489            .await?;
490        self.decode_json(response).await
491    }
492
493    fn request(&self, method: Method, path: &str, authenticated: bool) -> reqwest::RequestBuilder {
494        let mut request = self.http.request(method, self.endpoint(path));
495        if authenticated && let Some(token) = &self.bearer_token {
496            request = request.header(header::AUTHORIZATION, token.clone());
497        }
498        request
499    }
500
501    fn endpoint(&self, path: &str) -> Url {
502        let mut endpoint = self.base_url.clone();
503        endpoint.set_path(&format!("/{}", path.trim_start_matches('/')));
504        endpoint
505    }
506
507    async fn decode_json<T: DeserializeOwned>(
508        &self,
509        response: reqwest::Response,
510    ) -> Result<ApiResponse<T>, ClientError> {
511        if !response.status().is_success() {
512            return Err(self.decode_api_failure(response).await?);
513        }
514        if response.status() != StatusCode::OK {
515            return Err(ClientError::UnexpectedStatus(response.status().as_u16()));
516        }
517        require_json(response.headers())?;
518        let request_id = request_id(response.headers())?;
519        let encoded = read_bounded(response, self.response_bytes).await?;
520        Ok(ApiResponse {
521            value: serde_json::from_slice(&encoded)?,
522            request_id,
523        })
524    }
525
526    async fn decode_api_failure(
527        &self,
528        response: reqwest::Response,
529    ) -> Result<ClientError, ClientError> {
530        let status = response.status().as_u16();
531        require_json(response.headers())?;
532        let header_request_id = request_id(response.headers())?;
533        let encoded = read_bounded(response, self.response_bytes).await?;
534        let envelope: ErrorV1 = serde_json::from_slice(&encoded)?;
535        if envelope.request_id != header_request_id {
536            return Err(ClientError::RequestIdMismatch);
537        }
538        Ok(ClientError::Api(ApiFailure {
539            status,
540            code: envelope.code,
541            message: envelope.message,
542            request_id: envelope.request_id,
543        }))
544    }
545}
546
547async fn read_bounded(
548    mut response: reqwest::Response,
549    maximum: usize,
550) -> Result<Vec<u8>, ClientError> {
551    if response
552        .content_length()
553        .is_some_and(|length| length > u64::try_from(maximum).unwrap_or(u64::MAX))
554    {
555        return Err(ClientError::ResponseTooLarge { maximum });
556    }
557    let mut encoded = Vec::new();
558    while let Some(chunk) = response.chunk().await? {
559        let next_length = encoded
560            .len()
561            .checked_add(chunk.len())
562            .ok_or(ClientError::ResponseTooLarge { maximum })?;
563        if next_length > maximum {
564            return Err(ClientError::ResponseTooLarge { maximum });
565        }
566        encoded.extend_from_slice(&chunk);
567    }
568    Ok(encoded)
569}
570
571fn require_json(headers: &HeaderMap) -> Result<(), ClientError> {
572    let content_type = single_header(headers, header::CONTENT_TYPE.as_str())
573        .ok_or(ClientError::InvalidContentType)?;
574    let media_type = content_type
575        .split(';')
576        .next()
577        .unwrap_or_default()
578        .trim()
579        .to_ascii_lowercase();
580    if media_type == "application/json"
581        || (media_type.starts_with("application/") && media_type.ends_with("+json"))
582    {
583        Ok(())
584    } else {
585        Err(ClientError::InvalidContentType)
586    }
587}
588
589fn request_id(headers: &HeaderMap) -> Result<String, ClientError> {
590    single_header(headers, "x-request-id")
591        .map(ToOwned::to_owned)
592        .ok_or(ClientError::InvalidRequestId)
593}
594
595fn single_header<'headers>(headers: &'headers HeaderMap, name: &str) -> Option<&'headers str> {
596    let mut values = headers.get_all(name).iter();
597    let value = values.next()?;
598    if values.next().is_some() {
599        return None;
600    }
601    value.to_str().ok()
602}
603
604#[cfg(test)]
605mod tests {
606    use std::time::Duration;
607
608    use super::{ClientBuilder, ClientConfigError};
609
610    #[test]
611    fn builder_accepts_only_bounded_root_http_origins() -> Result<(), ClientConfigError> {
612        ClientBuilder::new("http://127.0.0.1:8787")?.build()?;
613        assert!(matches!(
614            ClientBuilder::new("file:///tmp/hyphae"),
615            Err(ClientConfigError::UnsupportedScheme)
616        ));
617        assert!(matches!(
618            ClientBuilder::new("https://user@example.test/"),
619            Err(ClientConfigError::NonOriginBaseUrl)
620        ));
621        assert!(matches!(
622            ClientBuilder::new("https://example.test/prefix"),
623            Err(ClientConfigError::NonOriginBaseUrl)
624        ));
625        Ok(())
626    }
627
628    #[test]
629    fn builder_rejects_invalid_secrets_and_zero_limits() -> Result<(), ClientConfigError> {
630        assert!(matches!(
631            ClientBuilder::new("http://localhost")?.bearer_token("bad\nsecret"),
632            Err(ClientConfigError::InvalidBearerToken)
633        ));
634        assert!(matches!(
635            ClientBuilder::new("http://localhost")?
636                .timeout(Duration::ZERO)
637                .build(),
638            Err(ClientConfigError::ZeroLimit)
639        ));
640        Ok(())
641    }
642}
hyphae_client/lib.rs

hyphae_client/
lib.rs
