feltdb-client 0.4.1

//! # feltdb-client
//!
//! A native async Rust client for a FeltDB server (`feltdb-server`): the Rust counterpart of
//! `@feltdb/core`'s server runtime. It owns no state: every read and write is a request to the
//! authority, which alone decides what commits.
//!
//! ```no_run
//! use feltdb_client::{Client, Operation, Transaction};
//! use serde_json::json;
//!
//! # async fn run() -> feltdb_client::Result<()> {
//! let client = Client::builder("http://127.0.0.1:7700")
//!     .token(std::env::var("FELTDB_TOKEN").unwrap_or_default())
//!     .build()?;
//!
//! // Several records, one atomic commit. The id makes a retry safe: resubmitting it after an
//! // uncertain outcome is recognized as a duplicate, never applied twice.
//! let receipt = client
//!     .commit(
//!         &Transaction::new("ship-o-1")
//!             .operation(Operation::put("orders", "o-1", json!({ "status": "shipped" })))
//!             .operation(Operation::create("audit", "a-1", json!({ "order": "o-1" }))),
//!     )
//!     .await?;
//! assert!(!receipt.duplicate);
//!
//! let order = client.get("orders", "o-1").await?;
//! # let _ = order;
//! # Ok(())
//! # }
//! ```
//!
//! ## Two surfaces, as on the server
//!
//! * **The authority** ([`Client`]): records, atomic transactions with preconditions and read-basis
//!   fences, compare-and-set, the bounded query, the committed revision and the event stream. This
//!   is the direct authority surface: the routes `@feltdb/core` uses without an `applicationId`.
//! * **An application** ([`Application`], from [`Client::application`]): the canonical `/v1`
//!   Service API, scoped to one application and environment and authorized against its contract:
//!   contract discovery, canonical queries, transactions, read bases, events and replay.
//!
//! * **The control plane** ([`Control`], from [`Client::control`]): tenants, applications, the
//!   draft → revision → promotion lifecycle of an application's contract, and online backup; and
//!   the server's [`Client::health`].
//!
//! Anything else the server exposes is reachable with [`Client::request`].
//!
//! ## Outcomes
//!
//! A refusal is an [`Error::Api`] carrying the server's stable `code` and, for a refused
//! transaction, the structured [`PreconditionFailure`]. A write that was sent but not answered is
//! [`Error::is_outcome_unknown`]: resubmit the same transaction id to learn what happened.
//!
//! ## Wire types
//!
//! Request and response documents that the server defines in `feltdb` are used directly, so the
//! two sides cannot disagree about a field: [`ReadSet`], [`ReadBasis`], [`PreconditionFailure`] and
//! the canonical query and transaction types in [`wire`]. `feltdb` itself is re-exported at the
//! exact version this client is built against.

mod application;
mod authority;
mod control;
mod error;
mod events;

use std::{
    sync::{Arc, RwLock},
    time::Duration,
};

use serde::{de::DeserializeOwned, Serialize};
use serde_json::Value;

pub use application::{
    AppCommit, AppTransaction, Application, ApplicationInfo, ReplayPage, StateVersion,
};
pub use authority::{
    BoundedQuery, CasConflict, CasOutcome, CasRequest, CommitReceipt, Condition, Direction,
    Operation, Precondition, QueryPage, Record, Revision, Transaction,
};
pub use control::{ApplicationRecord, Control, Health, Promotion, Tenant};
pub use error::{ApiError, Error, Result};
pub use events::{EventStream, ServerEvent};
pub use feltdb;
pub use feltdb::{
    PreconditionFailure, ReadBasis, ReadCondition, ReadPredicate, ReadRecord, ReadSet,
    TransactionStatus,
};
pub use reqwest::Method;

/// The canonical application contract's wire types, from `feltdb::state_contract`.
pub mod wire {
    pub use feltdb::state_contract::{
        AggregateDefinition, AggregateFunction, CanonicalQuery, QueryFilter, QueryOrder,
        QueryReference, QueryResult, SortDirection, TransactionOperation, TransactionOperationKind,
        TransactionPrecondition, TransactionResult,
    };
}

/// The protocol version this client speaks; the server refuses one it does not.
const PROTOCOL_VERSION: &str = "1";
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);

/// Which authority a request is addressed to, for servers that host several tenants.
#[derive(Debug, Clone, PartialEq, Eq)]
enum AuthorityScope {
    Tenant {
        tenant_id: String,
        environment_id: Option<String>,
    },
    Platform,
}

/// Configures a [`Client`].
#[derive(Debug, Clone)]
pub struct ClientBuilder {
    url: String,
    token: Option<String>,
    scope: Option<AuthorityScope>,
    timeout: Duration,
    root_certificates: Vec<Vec<u8>>,
}

impl ClientBuilder {
    /// Authenticate with a bearer token: an API key from `feltdb-server keys create`, or an
    /// application session's access token.
    pub fn token(mut self, token: impl Into<String>) -> Self {
        let token = token.into();
        self.token = (!token.is_empty()).then_some(token);
        self
    }

    /// Bind every request to one tenant. A request the server resolves to another tenant is refused
    /// with `AUTHORITY_SCOPE_MISMATCH` instead of being served.
    pub fn tenant(mut self, tenant_id: impl Into<String>) -> Self {
        self.scope = Some(AuthorityScope::Tenant {
            tenant_id: tenant_id.into(),
            environment_id: None,
        });
        self
    }

    /// Bind every request to one tenant and one of its environments.
    pub fn tenant_environment(
        mut self,
        tenant_id: impl Into<String>,
        environment_id: impl Into<String>,
    ) -> Self {
        self.scope = Some(AuthorityScope::Tenant {
            tenant_id: tenant_id.into(),
            environment_id: Some(environment_id.into()),
        });
        self
    }

    /// Address the platform authority rather than a tenant's.
    pub fn platform(mut self) -> Self {
        self.scope = Some(AuthorityScope::Platform);
        self
    }

    /// How long one request may take before it is abandoned (30 seconds by default). Event streams
    /// are exempt: they stay open until closed.
    pub fn timeout(mut self, timeout: Duration) -> Self {
        self.timeout = timeout;
        self
    }

    /// Trust a PEM certificate authority in addition to the bundled public roots, for a server
    /// whose certificate a private CA issued. Certificate verification is never disabled. A PEM
    /// that does not parse is refused by [`ClientBuilder::build`].
    pub fn add_root_certificate_pem(mut self, pem: impl Into<Vec<u8>>) -> Self {
        self.root_certificates.push(pem.into());
        self
    }

    pub fn build(self) -> Result<Client> {
        let url = self.url.trim_end_matches('/').to_owned();
        let parsed = reqwest::Url::parse(&url)
            .map_err(|error| Error::InvalidRequest(format!("server URL {url:?}: {error}")))?;
        if !matches!(parsed.scheme(), "http" | "https") {
            return Err(Error::InvalidRequest(format!(
                "server URL {url:?} must be http or https"
            )));
        }
        let mut http = reqwest::Client::builder();
        for pem in &self.root_certificates {
            // `Certificate::from_pem` accepts anything and adds nothing for input that holds no
            // certificate; parse the bundle so a wrong file is refused here, not at a handshake.
            let certificates = reqwest::Certificate::from_pem_bundle(pem)
                .ok()
                .filter(|certificates| !certificates.is_empty())
                .ok_or_else(|| {
                    Error::InvalidRequest("root certificate is not a PEM certificate".into())
                })?;
            for certificate in certificates {
                http = http.add_root_certificate(certificate);
            }
        }
        let http = http
            .build()
            .map_err(|source| Error::transport(source, false))?;
        Ok(Client {
            inner: Arc::new(Inner {
                http,
                url,
                token: RwLock::new(self.token),
                scope: self.scope,
                timeout: self.timeout,
            }),
        })
    }
}

struct Inner {
    http: reqwest::Client,
    url: String,
    token: RwLock<Option<String>>,
    scope: Option<AuthorityScope>,
    timeout: Duration,
}

/// A connection to one FeltDB server. Cheap to clone; clones share the connection pool and token.
#[derive(Clone)]
pub struct Client {
    inner: Arc<Inner>,
}

impl std::fmt::Debug for Client {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Client")
            .field("url", &self.inner.url)
            .field("scope", &self.inner.scope)
            .finish_non_exhaustive()
    }
}

impl Client {
    /// Start configuring a client for the server at `url`, e.g. `http://127.0.0.1:7700`.
    pub fn builder(url: impl Into<String>) -> ClientBuilder {
        ClientBuilder {
            url: url.into(),
            token: None,
            scope: None,
            timeout: DEFAULT_TIMEOUT,
            root_certificates: Vec::new(),
        }
    }

    /// The server URL, without a trailing slash.
    pub fn url(&self) -> &str {
        &self.inner.url
    }

    /// Replace the bearer token (or clear it with an empty string) for this client and its clones.
    pub fn set_token(&self, token: impl Into<String>) {
        let token = token.into();
        *self
            .inner
            .token
            .write()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) =
            (!token.is_empty()).then_some(token);
    }

    /// A handle on one application's canonical Service API, in its `production` environment
    /// unless [`Application::environment`] says otherwise.
    pub fn application(&self, application_id: impl Into<String>) -> Application {
        Application::new(self.clone(), application_id.into())
    }

    /// Any server route, for what this client does not name. `path` starts with `/` and may carry
    /// a query string. An empty success (`204`) is `Value::Null`.
    pub async fn request<B: Serialize + ?Sized>(
        &self,
        method: Method,
        path: &str,
        body: Option<&B>,
    ) -> Result<Value> {
        self.send(method, path, body).await
    }

    /// [`Client::request`] with extra headers, e.g. a route's `if-version` precondition.
    pub async fn request_with_headers<B: Serialize + ?Sized>(
        &self,
        method: Method,
        path: &str,
        headers: &[(&str, &str)],
        body: Option<&B>,
    ) -> Result<Value> {
        self.send_with_headers(method, path, headers, body).await
    }

    async fn send_with_headers<T: DeserializeOwned, B: Serialize + ?Sized>(
        &self,
        method: Method,
        path: &str,
        headers: &[(&str, &str)],
        body: Option<&B>,
    ) -> Result<T> {
        let mutating = !matches!(method, Method::GET | Method::HEAD | Method::OPTIONS);
        let mut builder = self.builder_for(method, path).timeout(self.inner.timeout);
        for (name, value) in headers {
            builder = builder.header(*name, *value);
        }
        if let Some(body) = body {
            builder = builder.json(body);
        }
        let response = self.respond(builder, mutating).await?;
        decode(response, mutating).await
    }

    fn builder_for(&self, method: Method, path: &str) -> reqwest::RequestBuilder {
        let mut builder = self
            .inner
            .http
            .request(method, format!("{}{path}", self.inner.url))
            .header("FeltDB-Protocol", PROTOCOL_VERSION);
        if let Some(token) = self
            .inner
            .token
            .read()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .as_deref()
        {
            builder = builder.bearer_auth(token);
        }
        match &self.inner.scope {
            Some(AuthorityScope::Tenant {
                tenant_id,
                environment_id,
            }) => {
                builder = builder.header("FeltDB-Tenant-Id", tenant_id);
                if let Some(environment_id) = environment_id {
                    builder = builder.header("FeltDB-Environment-Id", environment_id);
                }
            }
            Some(AuthorityScope::Platform) => {
                builder = builder.header("FeltDB-Authority-Mode", "platform");
            }
            None => {}
        }
        builder
    }

    /// Send a request and return the raw response, mapping a refusal to [`Error::Api`].
    async fn exchange<B: Serialize + ?Sized>(
        &self,
        method: Method,
        path: &str,
        body: Option<&B>,
    ) -> Result<reqwest::Response> {
        let mutating = !matches!(method, Method::GET | Method::HEAD | Method::OPTIONS);
        let mut builder = self.builder_for(method, path).timeout(self.inner.timeout);
        if let Some(body) = body {
            builder = builder.json(body);
        }
        self.respond(builder, mutating).await
    }

    async fn respond(
        &self,
        builder: reqwest::RequestBuilder,
        mutating: bool,
    ) -> Result<reqwest::Response> {
        let response = builder
            .send()
            .await
            .map_err(|source| Error::transport(source, mutating))?;
        if response.status().is_success() {
            return Ok(response);
        }
        let status = response.status().as_u16();
        let text = response
            .text()
            .await
            .map_err(|source| Error::transport(source, mutating))?;
        Err(Error::Api(Box::new(ApiError::from_body(status, &text))))
    }

    async fn send<T: DeserializeOwned, B: Serialize + ?Sized>(
        &self,
        method: Method,
        path: &str,
        body: Option<&B>,
    ) -> Result<T> {
        let mutating = !matches!(method, Method::GET | Method::HEAD | Method::OPTIONS);
        let response = self.exchange(method, path, body).await?;
        decode(response, mutating).await
    }

    async fn open_events(&self, path: &str, cursor: Option<&str>) -> Result<EventStream> {
        let mut builder = self.builder_for(Method::GET, path);
        if let Some(cursor) = cursor {
            builder = builder.header("Last-Event-ID", cursor);
        }
        let response = builder
            .send()
            .await
            .map_err(|source| Error::transport(source, false))?;
        if !response.status().is_success() {
            let status = response.status().as_u16();
            let text = response
                .text()
                .await
                .map_err(|source| Error::transport(source, false))?;
            return Err(Error::Api(Box::new(ApiError::from_body(status, &text))));
        }
        Ok(EventStream::new(response, cursor.map(str::to_owned)))
    }
}

async fn decode<T: DeserializeOwned>(response: reqwest::Response, mutating: bool) -> Result<T> {
    let status = response.status().as_u16();
    let bytes = response
        .bytes()
        .await
        .map_err(|source| Error::transport(source, mutating))?;
    let bytes: &[u8] = if bytes.is_empty() { b"null" } else { &bytes };
    serde_json::from_slice(bytes).map_err(|source| Error::Decode { status, source })
}

/// Percent-encode one path segment or query value, for paths passed to [`Client::request`].
/// Identifiers such as revision ids may contain `/` and `:`, which must not reach the path raw.
pub fn path_segment(value: &str) -> String {
    encode(value)
}

fn encode(value: &str) -> String {
    let mut out = String::with_capacity(value.len());
    for byte in value.bytes() {
        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') {
            out.push(byte as char);
        } else {
            out.push_str(&format!("%{byte:02X}"));
        }
    }
    out
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn path_segments_are_encoded() {
        assert_eq!(encode("o-1"), "o-1");
        assert_eq!(encode("a/b c:é"), "a%2Fb%20c%3A%C3%A9");
    }

    #[test]
    fn a_url_that_is_not_http_is_refused() {
        assert!(matches!(
            Client::builder("ftp://example.com").build(),
            Err(Error::InvalidRequest(_))
        ));
        assert_eq!(
            Client::builder("http://127.0.0.1:7700/")
                .build()
                .unwrap()
                .url(),
            "http://127.0.0.1:7700"
        );
    }
}