brain_db_sdk/http/
client.rs

1//! [`BrainHttpClient`] — the async HTTP tier of the Brain SDK.
2//!
3//! Talks JSON to the Brain HTTP edge (`brain-edge` self-hosted, or the Arc cloud
4//! gateway). Same verb surface and field names as the Python and TypeScript HTTP
5//! clients. For the native wire protocol (streaming, transactions, typed-graph
6//! management), use [`crate::BrainClient`] instead.
7
8use std::time::Duration;
9
10use serde::de::DeserializeOwned;
11use serde::Serialize;
12
13use super::error::BrainHttpError;
14use super::retry::HttpRetryPolicy;
15use super::types::{
16    Capabilities, CreateEntityInput, CreateEntityResult, EncodeInput, EncodeResult, EntityDetail,
17    ForgetInput, ForgetResult, GetRelationQuery, GetStatementQuery, GraphFetchQuery, GraphPage,
18    LinkInput, LinkResult, ListEntitiesQuery, ListEntitiesResult, ListRelationsQuery,
19    ListRelationsResult, ListStatementsQuery, ListStatementsResult, MemoryInspect, MemoryListPage,
20    MemoryListQuery, PlanInput, PlanResult, ReasonInput, ReasonResult, RecallInput, RecallResult,
21    RelationDetail, ResolveEntityInput, ResolveEntityResult, Schema, SchemaGetQuery,
22    SchemaReplaceInput, SchemaReplaceResult, SchemaUploadInput, SchemaUploadResult,
23    SchemaValidateInput, SchemaValidateResult, StatementDetail, TraverseInput, TraverseResult,
24    UnlinkInput, UnlinkResult, Whoami,
25};
26
27/// The default edge base URL (self-host).
28pub const DEFAULT_BASE_URL: &str = "http://127.0.0.1:8080";
29
30/// The default per-request timeout: a hung Brain must not hang the caller.
31pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
32
33/// Build the underlying `reqwest` client with a request timeout. Falls back to
34/// the default client if the builder rejects the config (never happens for a
35/// plain timeout, but we avoid a panic either way).
36fn build_http(timeout: Duration) -> reqwest::Client {
37    reqwest::Client::builder()
38        .timeout(timeout)
39        .build()
40        .unwrap_or_else(|_| reqwest::Client::new())
41}
42
43/// Percent-encode one path segment.
44///
45/// Ids are caller data. An id carrying `/`, `?`, `#` — or the two dots of a
46/// parent reference — would otherwise rewrite the route instead of addressing a
47/// row, so everything outside the unreserved set is escaped. `.` is escaped
48/// too, which `..` needs and which no id is harmed by.
49fn segment(value: &str) -> String {
50    let mut out = String::with_capacity(value.len());
51    for &b in value.as_bytes() {
52        match b {
53            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'~' => out.push(b as char),
54            // `write!` into the buffer directly; `format!` would allocate a
55            // String per escaped byte.
56            _ => {
57                use std::fmt::Write as _;
58                let _ = write!(out, "%{b:02X}");
59            }
60        }
61    }
62    out
63}
64
65/// Parse a `Retry-After` header expressed as integer seconds. HTTP-date form is
66/// ignored (returns `None`), leaving the computed backoff in force.
67fn parse_retry_after(headers: &reqwest::header::HeaderMap) -> Option<Duration> {
68    headers
69        .get(reqwest::header::RETRY_AFTER)
70        .and_then(|v| v.to_str().ok())
71        .and_then(|s| s.trim().parse::<u64>().ok())
72        .map(Duration::from_secs)
73}
74
75/// An async HTTP client for the Brain edge.
76#[derive(Clone, Debug)]
77pub struct BrainHttpClient {
78    base: String,
79    api_key: String,
80    http: reqwest::Client,
81    retry: HttpRetryPolicy,
82}
83
84impl BrainHttpClient {
85    /// Authenticate as `api_key` against `base_url`. Uses the
86    /// [`DEFAULT_TIMEOUT`] request timeout and the default [`HttpRetryPolicy`].
87    ///
88    /// Key first, URL second — matching the Python and TypeScript clients,
89    /// which both take the key as the required argument and the URL as the
90    /// optional one. Both parameters are `impl Into<String>`, so a swapped call
91    /// compiles cleanly and sends the base URL as the bearer token; keeping the
92    /// three SDKs in one order is the only thing that makes that mistake hard.
93    /// [`Self::localhost`] covers the common case in one argument.
94    #[must_use]
95    pub fn new(api_key: impl Into<String>, base_url: impl Into<String>) -> Self {
96        let base = base_url.into().trim_end_matches('/').to_string();
97        Self {
98            base,
99            api_key: api_key.into(),
100            http: build_http(DEFAULT_TIMEOUT),
101            retry: HttpRetryPolicy::default(),
102        }
103    }
104
105    /// Build a client against [`DEFAULT_BASE_URL`].
106    #[must_use]
107    pub fn localhost(api_key: impl Into<String>) -> Self {
108        Self::new(api_key, DEFAULT_BASE_URL)
109    }
110
111    /// Override the per-request timeout (default [`DEFAULT_TIMEOUT`]). A timeout
112    /// surfaces as a transport [`BrainHttpError`], not a panic.
113    #[must_use]
114    pub fn with_timeout(mut self, timeout: Duration) -> Self {
115        self.http = build_http(timeout);
116        self
117    }
118
119    /// Override the retry policy applied to idempotent verbs (`encode`,
120    /// `whoami`, `capabilities`). Pass [`HttpRetryPolicy::none`] to disable.
121    #[must_use]
122    pub fn with_retry_policy(mut self, policy: HttpRetryPolicy) -> Self {
123        self.retry = policy;
124        self
125    }
126
127    /// Store a memory. Idempotent (stable server-side request id) — retried.
128    pub async fn encode(&self, input: &EncodeInput) -> Result<EncodeResult, BrainHttpError> {
129        self.send(reqwest::Method::POST, "/v1/memories", Some(input), true)
130            .await
131    }
132
133    /// Recall memories for a cue.
134    // A PURE READ IS RETRYABLE, whatever its method.
135    // `recall`, `plan`, `reason` and `traverse` are POSTs only because their filters need a body; none of them changes anything, so a transient 503/429 should be absorbed rather than raised at the caller.
136    // `validate_schema` already worked this way — it is a dry run — and the other four were left behind.
137    // What stays single-shot is what CREATES state: link, create_entity, resolve_entity (it can create), and the schema upload/replace.
138    pub async fn recall(&self, input: &RecallInput) -> Result<RecallResult, BrainHttpError> {
139        self.send(reqwest::Method::POST, "/v1/recall", Some(input), true)
140            .await
141    }
142
143    /// Forget a memory.
144    pub async fn forget(&self, input: &ForgetInput) -> Result<ForgetResult, BrainHttpError> {
145        // A DELETE whose result type MODELS the repeat is safe to retry: `forget` answers `was_already_forgotten` and `unlink` answers `removed: false` when there was nothing there.
146        // Repeating one reaches the same end state, so a transient 503/429 on a cleanup path should be absorbed rather than raised at the caller.
147        self.send(reqwest::Method::DELETE, "/v1/memories", Some(input), true)
148            .await
149    }
150
151    /// Create/overwrite a directed edge between two memories.
152    pub async fn link(&self, input: &LinkInput) -> Result<LinkResult, BrainHttpError> {
153        self.send(reqwest::Method::POST, "/v1/links", Some(input), false)
154            .await
155    }
156
157    /// Remove a directed edge (idempotent).
158    pub async fn unlink(&self, input: &UnlinkInput) -> Result<UnlinkResult, BrainHttpError> {
159        // Idempotent for the same reason `forget` is: `removed: false` is the
160        // documented answer when the edge was not there.
161        self.send(reqwest::Method::DELETE, "/v1/links", Some(input), true)
162            .await
163    }
164
165    /// Plan a path from a start state to a goal state.
166    pub async fn plan(&self, input: &PlanInput) -> Result<PlanResult, BrainHttpError> {
167        self.send(reqwest::Method::POST, "/v1/plan", Some(input), true)
168            .await
169    }
170
171    /// Infer over the graph from an observation.
172    pub async fn reason(&self, input: &ReasonInput) -> Result<ReasonResult, BrainHttpError> {
173        self.send(reqwest::Method::POST, "/v1/reason", Some(input), true)
174            .await
175    }
176
177    /// The identity Brain resolves from the credential. Idempotent GET — retried.
178    pub async fn whoami(&self) -> Result<Whoami, BrainHttpError> {
179        self.send::<(), _>(reqwest::Method::GET, "/v1/whoami", None, true)
180            .await
181    }
182
183    /// What the connected shard supports. Idempotent GET — retried.
184    pub async fn capabilities(&self) -> Result<Capabilities, BrainHttpError> {
185        self.send::<(), _>(reqwest::Method::GET, "/v1/capabilities", None, true)
186            .await
187    }
188
189    // --- reads ---------------------------------------------------------------
190    //
191    // Every read below is idempotent and therefore retried, and every one of
192    // them takes its filters as a query struct whose unset fields are simply
193    // not sent, so the edge's own defaults stay in force. Ids are interpolated
194    // through `segment`, so an id carrying a slash or a query marker addresses
195    // a row rather than reshaping the route.
196
197    /// A page of the space's memories. This is not a recall: no cue, no
198    /// ranking, just the timeline in a stable order plus a cursor.
199    pub async fn memory_list(
200        &self,
201        query: &MemoryListQuery,
202    ) -> Result<MemoryListPage, BrainHttpError> {
203        self.send_with_query::<(), _, _>(
204            reqwest::Method::GET,
205            "/v1/memories",
206            Some(query),
207            None,
208            true,
209        )
210        .await
211    }
212
213    /// Everything the encode pipeline produced for one memory: the vector, the
214    /// persisted row, the extracted graph.
215    pub async fn memory_inspect(&self, memory_id: &str) -> Result<MemoryInspect, BrainHttpError> {
216        let id = segment(memory_id);
217        self.send::<(), _>(
218            reqwest::Method::GET,
219            &format!("/v1/memories/{id}/inspect"),
220            None,
221            true,
222        )
223        .await
224    }
225
226    /// Entities in the typed graph, filtered by the query.
227    pub async fn list_entities(
228        &self,
229        query: &ListEntitiesQuery,
230    ) -> Result<ListEntitiesResult, BrainHttpError> {
231        self.send_with_query::<(), _, _>(
232            reqwest::Method::GET,
233            "/v1/entities",
234            Some(query),
235            None,
236            true,
237        )
238        .await
239    }
240
241    /// One entity by id.
242    pub async fn get_entity(&self, entity_id: &str) -> Result<EntityDetail, BrainHttpError> {
243        let id = segment(entity_id);
244        self.send::<(), _>(
245            reqwest::Method::GET,
246            &format!("/v1/entities/{id}"),
247            None,
248            true,
249        )
250        .await
251    }
252
253    /// The relations incident to one entity.
254    pub async fn list_relations(
255        &self,
256        entity_id: &str,
257        query: &ListRelationsQuery,
258    ) -> Result<ListRelationsResult, BrainHttpError> {
259        let id = segment(entity_id);
260        self.send_with_query::<(), _, _>(
261            reqwest::Method::GET,
262            &format!("/v1/entities/{id}/relations"),
263            Some(query),
264            None,
265            true,
266        )
267        .await
268    }
269
270    /// One page of the space's typed graph, as a node/edge set.
271    pub async fn graph_fetch(&self, query: &GraphFetchQuery) -> Result<GraphPage, BrainHttpError> {
272        self.send_with_query::<(), _, _>(reqwest::Method::GET, "/v1/graph", Some(query), None, true)
273            .await
274    }
275
276    /// One relation by id.
277    pub async fn get_relation(
278        &self,
279        relation_id: &str,
280        query: &GetRelationQuery,
281    ) -> Result<RelationDetail, BrainHttpError> {
282        let id = segment(relation_id);
283        self.send_with_query::<(), _, _>(
284            reqwest::Method::GET,
285            &format!("/v1/relations/{id}"),
286            Some(query),
287            None,
288            true,
289        )
290        .await
291    }
292
293    /// Statements in the typed graph, filtered by the query.
294    pub async fn list_statements(
295        &self,
296        query: &ListStatementsQuery,
297    ) -> Result<ListStatementsResult, BrainHttpError> {
298        self.send_with_query::<(), _, _>(
299            reqwest::Method::GET,
300            "/v1/statements",
301            Some(query),
302            None,
303            true,
304        )
305        .await
306    }
307
308    /// One statement by id.
309    pub async fn get_statement(
310        &self,
311        statement_id: &str,
312        query: &GetStatementQuery,
313    ) -> Result<StatementDetail, BrainHttpError> {
314        let id = segment(statement_id);
315        self.send_with_query::<(), _, _>(
316            reqwest::Method::GET,
317            &format!("/v1/statements/{id}"),
318            Some(query),
319            None,
320            true,
321        )
322        .await
323    }
324
325    /// The active schema document for a namespace, or a specific version of it.
326    pub async fn get_schema(&self, query: &SchemaGetQuery) -> Result<Schema, BrainHttpError> {
327        self.send_with_query::<(), _, _>(
328            reqwest::Method::GET,
329            "/v1/schema",
330            Some(query),
331            None,
332            true,
333        )
334        .await
335    }
336
337    // --- writes --------------------------------------------------------------
338    //
339    // None of these is retried. A write that fails in flight may or may not
340    // have landed, and a second attempt would either duplicate it or race the
341    // first. `encode` above is the single exception, because the edge derives a
342    // stable request id for it and dedupes server-side.
343    //
344    // `validate_schema` is the other kind of exception: a write-shaped route
345    // that changes nothing — the edge parses the document and answers — so it
346    // is safe to repeat and is retried like a read.
347    //
348    // `replace_schema` is the destructive one. It drops every declared row in
349    // the namespace before the new document lands, so it is never retried under
350    // any policy, and its confirmation flag has no default to fall back on.
351
352    /// Check a schema document and report what an upload would do. A dry run
353    /// that changes nothing, so it is idempotent and retried.
354    pub async fn validate_schema(
355        &self,
356        input: &SchemaValidateInput,
357    ) -> Result<SchemaValidateResult, BrainHttpError> {
358        self.send(
359            reqwest::Method::POST,
360            "/v1/schema/validate",
361            Some(input),
362            true,
363        )
364        .await
365    }
366
367    /// Upload a schema document as the namespace's next version.
368    pub async fn upload_schema(
369        &self,
370        input: &SchemaUploadInput,
371    ) -> Result<SchemaUploadResult, BrainHttpError> {
372        self.send(reqwest::Method::POST, "/v1/schema", Some(input), false)
373            .await
374    }
375
376    /// Replace the namespace's schema, dropping every declared row first.
377    /// Irreversible, and never retried: a second attempt would drop what the
378    /// first one just wrote.
379    pub async fn replace_schema(
380        &self,
381        input: &SchemaReplaceInput,
382    ) -> Result<SchemaReplaceResult, BrainHttpError> {
383        self.send(reqwest::Method::PUT, "/v1/schema", Some(input), false)
384            .await
385    }
386
387    // Entity writes. `create_entity` and `resolve_entity` both mint entities;
388    // the difference is that `resolve_entity` first tries to match the name
389    // against what is already there, and creates only when `allow_create` says
390    // it may. `traverse_relations` is a read in every sense except its verb —
391    // the walk it describes does not fit in a query string, which is why the
392    // edge takes a body for it — and it is left unretried all the same, so that
393    // "reads are retried, writes are not" stays a statement about the verb
394    // rather than one about intent.
395
396    /// Create an entity in the typed graph.
397    pub async fn create_entity(
398        &self,
399        input: &CreateEntityInput,
400    ) -> Result<CreateEntityResult, BrainHttpError> {
401        self.send(reqwest::Method::POST, "/v1/entities", Some(input), false)
402            .await
403    }
404
405    /// Resolve a candidate name against the typed graph, optionally creating
406    /// the entity when nothing matches.
407    pub async fn resolve_entity(
408        &self,
409        input: &ResolveEntityInput,
410    ) -> Result<ResolveEntityResult, BrainHttpError> {
411        self.send(
412            reqwest::Method::POST,
413            "/v1/entities/resolve",
414            Some(input),
415            false,
416        )
417        .await
418    }
419
420    /// Walk the relation graph outward from one entity.
421    pub async fn traverse_relations(
422        &self,
423        entity_id: &str,
424        input: &TraverseInput,
425    ) -> Result<TraverseResult, BrainHttpError> {
426        let id = segment(entity_id);
427        self.send(
428            reqwest::Method::POST,
429            &format!("/v1/entities/{id}/traverse"),
430            Some(input),
431            true,
432        )
433        .await
434    }
435
436    /// Send with the retry policy applied when `idempotent`. Non-idempotent
437    /// verbs (`retry == false`) run exactly once.
438    async fn send<B, R>(
439        &self,
440        method: reqwest::Method,
441        path: &str,
442        body: Option<&B>,
443        idempotent: bool,
444    ) -> Result<R, BrainHttpError>
445    where
446        B: Serialize + ?Sized,
447        R: DeserializeOwned,
448    {
449        self.send_with_query::<B, (), R>(method, path, None, body, idempotent)
450            .await
451    }
452
453    /// [`Self::send`] with a query struct appended to the path. Its `None`
454    /// fields skip (see `types`), so a default-constructed query sends no query
455    /// string at all rather than a row of empty parameters.
456    async fn send_with_query<B, Q, R>(
457        &self,
458        method: reqwest::Method,
459        path: &str,
460        query: Option<&Q>,
461        body: Option<&B>,
462        idempotent: bool,
463    ) -> Result<R, BrainHttpError>
464    where
465        B: Serialize + ?Sized,
466        Q: Serialize + ?Sized,
467        R: DeserializeOwned,
468    {
469        let mut attempt = 1u32;
470        loop {
471            match self
472                .send_once::<B, Q, R>(method.clone(), path, query, body)
473                .await
474            {
475                Ok(value) => return Ok(value),
476                Err((err, retry_after)) => {
477                    if idempotent && self.retry.should_retry(attempt, &err) {
478                        tokio::time::sleep(self.retry.backoff(attempt, retry_after)).await;
479                        attempt += 1;
480                    } else {
481                        return Err(err);
482                    }
483                }
484            }
485        }
486    }
487
488    /// One request attempt. On failure returns the error plus an optional
489    /// server `Retry-After` hint for the retry scheduler.
490    async fn send_once<B, Q, R>(
491        &self,
492        method: reqwest::Method,
493        path: &str,
494        query: Option<&Q>,
495        body: Option<&B>,
496    ) -> Result<R, (BrainHttpError, Option<Duration>)>
497    where
498        B: Serialize + ?Sized,
499        Q: Serialize + ?Sized,
500        R: DeserializeOwned,
501    {
502        let url = format!("{}{}", self.base, path);
503        let mut rb = self.http.request(method, &url).bearer_auth(&self.api_key);
504        if let Some(q) = query {
505            rb = rb.query(q);
506        }
507        if let Some(b) = body {
508            rb = rb.json(b);
509        }
510        let resp = rb.send().await.map_err(|e| {
511            (
512                BrainHttpError::transport(format!("request to {url} failed: {e}")),
513                None,
514            )
515        })?;
516        let status = resp.status();
517        let retry_after = parse_retry_after(resp.headers());
518        let text = resp.text().await.map_err(|e| {
519            (
520                BrainHttpError::transport(format!("read body from {url}: {e}")),
521                None,
522            )
523        })?;
524        if !status.is_success() {
525            return Err((
526                BrainHttpError::from_body(status.as_u16(), &text),
527                retry_after,
528            ));
529        }
530        serde_json::from_str(&text).map_err(|e| {
531            (
532                BrainHttpError::transport(format!("decode {url} response: {e}")),
533                None,
534            )
535        })
536    }
537}
brain_db_sdk/http/client.rs

brain_db_sdk/http/
client.rs
