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

feltdb_client/
lib.rs
