feltdb_client/
lib.rs

1//! # feltdb-client
2//!
3//! A native async Rust client for a FeltDB server (`feltdb-server`): the Rust counterpart of
4//! `@feltdb/core`'s server runtime. It owns no state: every read and write is a request to the
5//! authority, which alone decides what commits.
6//!
7//! ```no_run
8//! use feltdb_client::{Client, Operation, Transaction};
9//! use serde_json::json;
10//!
11//! # async fn run() -> feltdb_client::Result<()> {
12//! let client = Client::builder("http://127.0.0.1:7700")
13//!     .token(std::env::var("FELTDB_TOKEN").unwrap_or_default())
14//!     .build()?;
15//!
16//! // Several records, one atomic commit. The id makes a retry safe: resubmitting it after an
17//! // uncertain outcome is recognized as a duplicate, never applied twice.
18//! let receipt = client
19//!     .commit(
20//!         &Transaction::new("ship-o-1")
21//!             .operation(Operation::put("orders", "o-1", json!({ "status": "shipped" })))
22//!             .operation(Operation::create("audit", "a-1", json!({ "order": "o-1" }))),
23//!     )
24//!     .await?;
25//! assert!(!receipt.duplicate);
26//!
27//! let order = client.get("orders", "o-1").await?;
28//! # let _ = order;
29//! # Ok(())
30//! # }
31//! ```
32//!
33//! ## Two surfaces, as on the server
34//!
35//! * **The authority** ([`Client`]): records, atomic transactions with preconditions and read-basis
36//!   fences, compare-and-set, the bounded query, the committed revision and the event stream. This
37//!   is the direct authority surface: the routes `@feltdb/core` uses without an `applicationId`.
38//! * **An application** ([`Application`], from [`Client::application`]): the canonical `/v1`
39//!   Service API, scoped to one application and environment and authorized against its contract:
40//!   contract discovery, canonical queries, transactions, read bases, events and replay.
41//!
42//! Anything else the server exposes is reachable with [`Client::request`].
43//!
44//! ## Outcomes
45//!
46//! A refusal is an [`Error::Api`] carrying the server's stable `code` and, for a refused
47//! transaction, the structured [`PreconditionFailure`]. A write that was sent but not answered is
48//! [`Error::is_outcome_unknown`]: resubmit the same transaction id to learn what happened.
49//!
50//! ## Wire types
51//!
52//! Request and response documents that the server defines in `feltdb` are used directly, so the
53//! two sides cannot disagree about a field: [`ReadSet`], [`ReadBasis`], [`PreconditionFailure`] and
54//! the canonical query and transaction types in [`wire`]. `feltdb` itself is re-exported at the
55//! exact version this client is built against.
56
57mod application;
58mod authority;
59mod error;
60mod events;
61
62use std::{
63    sync::{Arc, RwLock},
64    time::Duration,
65};
66
67use serde::{de::DeserializeOwned, Serialize};
68use serde_json::Value;
69
70pub use application::{AppCommit, AppTransaction, Application, ApplicationInfo, ReplayPage};
71pub use authority::{
72    BoundedQuery, CasConflict, CasOutcome, CasRequest, CommitReceipt, Condition, Direction,
73    Operation, Precondition, QueryPage, Record, Revision, Transaction,
74};
75pub use error::{ApiError, Error, Result};
76pub use events::{EventStream, ServerEvent};
77pub use feltdb;
78pub use feltdb::{
79    PreconditionFailure, ReadBasis, ReadCondition, ReadPredicate, ReadRecord, ReadSet,
80    TransactionStatus,
81};
82pub use reqwest::Method;
83
84/// The canonical application contract's wire types, from `feltdb::state_contract`.
85pub mod wire {
86    pub use feltdb::state_contract::{
87        AggregateDefinition, AggregateFunction, CanonicalQuery, QueryFilter, QueryOrder,
88        QueryReference, QueryResult, SortDirection, TransactionOperation, TransactionOperationKind,
89        TransactionPrecondition, TransactionResult,
90    };
91}
92
93/// The protocol version this client speaks; the server refuses one it does not.
94const PROTOCOL_VERSION: &str = "1";
95const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
96
97/// Which authority a request is addressed to, for servers that host several tenants.
98#[derive(Debug, Clone, PartialEq, Eq)]
99enum AuthorityScope {
100    Tenant {
101        tenant_id: String,
102        environment_id: Option<String>,
103    },
104    Platform,
105}
106
107/// Configures a [`Client`].
108#[derive(Debug, Clone)]
109pub struct ClientBuilder {
110    url: String,
111    token: Option<String>,
112    scope: Option<AuthorityScope>,
113    timeout: Duration,
114}
115
116impl ClientBuilder {
117    /// Authenticate with a bearer token: an API key from `feltdb-server keys create`, or an
118    /// application session's access token.
119    pub fn token(mut self, token: impl Into<String>) -> Self {
120        let token = token.into();
121        self.token = (!token.is_empty()).then_some(token);
122        self
123    }
124
125    /// Bind every request to one tenant. A request the server resolves to another tenant is refused
126    /// with `AUTHORITY_SCOPE_MISMATCH` instead of being served.
127    pub fn tenant(mut self, tenant_id: impl Into<String>) -> Self {
128        self.scope = Some(AuthorityScope::Tenant {
129            tenant_id: tenant_id.into(),
130            environment_id: None,
131        });
132        self
133    }
134
135    /// Bind every request to one tenant and one of its environments.
136    pub fn tenant_environment(
137        mut self,
138        tenant_id: impl Into<String>,
139        environment_id: impl Into<String>,
140    ) -> Self {
141        self.scope = Some(AuthorityScope::Tenant {
142            tenant_id: tenant_id.into(),
143            environment_id: Some(environment_id.into()),
144        });
145        self
146    }
147
148    /// Address the platform authority rather than a tenant's.
149    pub fn platform(mut self) -> Self {
150        self.scope = Some(AuthorityScope::Platform);
151        self
152    }
153
154    /// How long one request may take before it is abandoned (30 seconds by default). Event streams
155    /// are exempt: they stay open until closed.
156    pub fn timeout(mut self, timeout: Duration) -> Self {
157        self.timeout = timeout;
158        self
159    }
160
161    pub fn build(self) -> Result<Client> {
162        let url = self.url.trim_end_matches('/').to_owned();
163        let parsed = reqwest::Url::parse(&url)
164            .map_err(|error| Error::InvalidRequest(format!("server URL {url:?}: {error}")))?;
165        if !matches!(parsed.scheme(), "http" | "https") {
166            return Err(Error::InvalidRequest(format!(
167                "server URL {url:?} must be http or https"
168            )));
169        }
170        let http = reqwest::Client::builder()
171            .build()
172            .map_err(|source| Error::transport(source, false))?;
173        Ok(Client {
174            inner: Arc::new(Inner {
175                http,
176                url,
177                token: RwLock::new(self.token),
178                scope: self.scope,
179                timeout: self.timeout,
180            }),
181        })
182    }
183}
184
185struct Inner {
186    http: reqwest::Client,
187    url: String,
188    token: RwLock<Option<String>>,
189    scope: Option<AuthorityScope>,
190    timeout: Duration,
191}
192
193/// A connection to one FeltDB server. Cheap to clone; clones share the connection pool and token.
194#[derive(Clone)]
195pub struct Client {
196    inner: Arc<Inner>,
197}
198
199impl std::fmt::Debug for Client {
200    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
201        f.debug_struct("Client")
202            .field("url", &self.inner.url)
203            .field("scope", &self.inner.scope)
204            .finish_non_exhaustive()
205    }
206}
207
208impl Client {
209    /// Start configuring a client for the server at `url`, e.g. `http://127.0.0.1:7700`.
210    pub fn builder(url: impl Into<String>) -> ClientBuilder {
211        ClientBuilder {
212            url: url.into(),
213            token: None,
214            scope: None,
215            timeout: DEFAULT_TIMEOUT,
216        }
217    }
218
219    /// The server URL, without a trailing slash.
220    pub fn url(&self) -> &str {
221        &self.inner.url
222    }
223
224    /// Replace the bearer token (or clear it with an empty string) for this client and its clones.
225    pub fn set_token(&self, token: impl Into<String>) {
226        let token = token.into();
227        *self
228            .inner
229            .token
230            .write()
231            .unwrap_or_else(|poisoned| poisoned.into_inner()) =
232            (!token.is_empty()).then_some(token);
233    }
234
235    /// A handle on one application's canonical Service API, in its `production` environment
236    /// unless [`Application::environment`] says otherwise.
237    pub fn application(&self, application_id: impl Into<String>) -> Application {
238        Application::new(self.clone(), application_id.into())
239    }
240
241    /// Any server route, for what this client does not name. `path` starts with `/` and may carry
242    /// a query string. An empty success (`204`) is `Value::Null`.
243    pub async fn request<B: Serialize + ?Sized>(
244        &self,
245        method: Method,
246        path: &str,
247        body: Option<&B>,
248    ) -> Result<Value> {
249        self.send(method, path, body).await
250    }
251
252    /// [`Client::request`] with extra headers, e.g. a route's `if-version` precondition.
253    pub async fn request_with_headers<B: Serialize + ?Sized>(
254        &self,
255        method: Method,
256        path: &str,
257        headers: &[(&str, &str)],
258        body: Option<&B>,
259    ) -> Result<Value> {
260        let mutating = !matches!(method, Method::GET | Method::HEAD | Method::OPTIONS);
261        let mut builder = self.builder_for(method, path).timeout(self.inner.timeout);
262        for (name, value) in headers {
263            builder = builder.header(*name, *value);
264        }
265        if let Some(body) = body {
266            builder = builder.json(body);
267        }
268        let response = self.respond(builder, mutating).await?;
269        decode(response, mutating).await
270    }
271
272    fn builder_for(&self, method: Method, path: &str) -> reqwest::RequestBuilder {
273        let mut builder = self
274            .inner
275            .http
276            .request(method, format!("{}{path}", self.inner.url))
277            .header("FeltDB-Protocol", PROTOCOL_VERSION);
278        if let Some(token) = self
279            .inner
280            .token
281            .read()
282            .unwrap_or_else(|poisoned| poisoned.into_inner())
283            .as_deref()
284        {
285            builder = builder.bearer_auth(token);
286        }
287        match &self.inner.scope {
288            Some(AuthorityScope::Tenant {
289                tenant_id,
290                environment_id,
291            }) => {
292                builder = builder.header("FeltDB-Tenant-Id", tenant_id);
293                if let Some(environment_id) = environment_id {
294                    builder = builder.header("FeltDB-Environment-Id", environment_id);
295                }
296            }
297            Some(AuthorityScope::Platform) => {
298                builder = builder.header("FeltDB-Authority-Mode", "platform");
299            }
300            None => {}
301        }
302        builder
303    }
304
305    /// Send a request and return the raw response, mapping a refusal to [`Error::Api`].
306    async fn exchange<B: Serialize + ?Sized>(
307        &self,
308        method: Method,
309        path: &str,
310        body: Option<&B>,
311    ) -> Result<reqwest::Response> {
312        let mutating = !matches!(method, Method::GET | Method::HEAD | Method::OPTIONS);
313        let mut builder = self.builder_for(method, path).timeout(self.inner.timeout);
314        if let Some(body) = body {
315            builder = builder.json(body);
316        }
317        self.respond(builder, mutating).await
318    }
319
320    async fn respond(
321        &self,
322        builder: reqwest::RequestBuilder,
323        mutating: bool,
324    ) -> Result<reqwest::Response> {
325        let response = builder
326            .send()
327            .await
328            .map_err(|source| Error::transport(source, mutating))?;
329        if response.status().is_success() {
330            return Ok(response);
331        }
332        let status = response.status().as_u16();
333        let text = response
334            .text()
335            .await
336            .map_err(|source| Error::transport(source, mutating))?;
337        Err(Error::Api(Box::new(ApiError::from_body(status, &text))))
338    }
339
340    async fn send<T: DeserializeOwned, B: Serialize + ?Sized>(
341        &self,
342        method: Method,
343        path: &str,
344        body: Option<&B>,
345    ) -> Result<T> {
346        let mutating = !matches!(method, Method::GET | Method::HEAD | Method::OPTIONS);
347        let response = self.exchange(method, path, body).await?;
348        decode(response, mutating).await
349    }
350
351    async fn open_events(&self, path: &str, cursor: Option<&str>) -> Result<EventStream> {
352        let mut builder = self.builder_for(Method::GET, path);
353        if let Some(cursor) = cursor {
354            builder = builder.header("Last-Event-ID", cursor);
355        }
356        let response = builder
357            .send()
358            .await
359            .map_err(|source| Error::transport(source, false))?;
360        if !response.status().is_success() {
361            let status = response.status().as_u16();
362            let text = response
363                .text()
364                .await
365                .map_err(|source| Error::transport(source, false))?;
366            return Err(Error::Api(Box::new(ApiError::from_body(status, &text))));
367        }
368        Ok(EventStream::new(response, cursor.map(str::to_owned)))
369    }
370}
371
372async fn decode<T: DeserializeOwned>(response: reqwest::Response, mutating: bool) -> Result<T> {
373    let status = response.status().as_u16();
374    let bytes = response
375        .bytes()
376        .await
377        .map_err(|source| Error::transport(source, mutating))?;
378    let bytes: &[u8] = if bytes.is_empty() { b"null" } else { &bytes };
379    serde_json::from_slice(bytes).map_err(|source| Error::Decode { status, source })
380}
381
382/// Percent-encode one path segment or query value, for paths passed to [`Client::request`].
383/// Identifiers such as revision ids may contain `/` and `:`, which must not reach the path raw.
384pub fn path_segment(value: &str) -> String {
385    encode(value)
386}
387
388fn encode(value: &str) -> String {
389    let mut out = String::with_capacity(value.len());
390    for byte in value.bytes() {
391        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') {
392            out.push(byte as char);
393        } else {
394            out.push_str(&format!("%{byte:02X}"));
395        }
396    }
397    out
398}
399
400#[cfg(test)]
401mod tests {
402    use super::*;
403
404    #[test]
405    fn path_segments_are_encoded() {
406        assert_eq!(encode("o-1"), "o-1");
407        assert_eq!(encode("a/b c:é"), "a%2Fb%20c%3A%C3%A9");
408    }
409
410    #[test]
411    fn a_url_that_is_not_http_is_refused() {
412        assert!(matches!(
413            Client::builder("ftp://example.com").build(),
414            Err(Error::InvalidRequest(_))
415        ));
416        assert_eq!(
417            Client::builder("http://127.0.0.1:7700/")
418                .build()
419                .unwrap()
420                .url(),
421            "http://127.0.0.1:7700"
422        );
423    }
424}
feltdb_client/lib.rs

feltdb_client/
lib.rs
