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

feltdb_client/
lib.rs
