feltdb-client 0.6.1

//! The canonical application Service API (`/v1`), scoped to one application and environment.

use std::{
    collections::BTreeSet,
    sync::{Arc, Mutex},
};

use feltdb::{
    state_contract::{
        AuthorizationContext, CanonicalQuery, QueryResult, TransactionOperation,
        TransactionOperationKind, TransactionPrecondition, TransactionRequest, TransactionResult,
    },
    ReadBasis, ReadSet,
};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};

use crate::{encode, Client, Error, EventStream, Method, Result};

/// The active contract of an application environment, as discovery reports it.
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct ApplicationInfo {
    pub application_id: String,
    #[serde(default)]
    pub application_name: Option<String>,
    pub environment: String,
    /// The contract revision every canonical request is evaluated against.
    pub revision_id: String,
    #[serde(default)]
    pub version: Option<u64>,
    pub contract_hash: String,
    #[serde(default)]
    pub flow_hash: Option<String>,
    #[serde(default)]
    pub dsl_version: Option<u32>,
    /// The contract snapshot: schema, indexes, policies, workflows, agents and capabilities.
    #[serde(default)]
    pub snapshot: Value,
}

/// A canonical transaction: operations and preconditions the application contract validates and
/// authorizes, committed atomically.
#[derive(Debug, Clone)]
pub struct AppTransaction {
    transaction_id: String,
    operations: Vec<TransactionOperation>,
    preconditions: Vec<TransactionPrecondition>,
    fences: Vec<ReadBasis>,
}

impl AppTransaction {
    /// A transaction with this identity; resubmitting it after an uncertain outcome is safe.
    pub fn new(transaction_id: impl Into<String>) -> Self {
        Self {
            transaction_id: transaction_id.into(),
            operations: Vec::new(),
            preconditions: Vec::new(),
            fences: Vec::new(),
        }
    }

    /// A transaction with a fresh random identity.
    pub fn generated() -> Self {
        Self::new(format!("tx-{}", uuid::Uuid::new_v4()))
    }

    pub fn id(&self) -> &str {
        &self.transaction_id
    }

    fn push(
        self,
        kind: TransactionOperationKind,
        collection: String,
        id: String,
        value: Value,
    ) -> Self {
        self.operation(TransactionOperation {
            kind,
            collection,
            id,
            value,
            if_version: None,
            if_version_token: None,
        })
    }

    /// Create a record that must not exist.
    pub fn insert(
        self,
        collection: impl Into<String>,
        id: impl Into<String>,
        value: Value,
    ) -> Self {
        self.push(
            TransactionOperationKind::Insert,
            collection.into(),
            id.into(),
            value,
        )
    }

    /// Write a record.
    pub fn update(
        self,
        collection: impl Into<String>,
        id: impl Into<String>,
        value: Value,
    ) -> Self {
        self.push(
            TransactionOperationKind::Update,
            collection.into(),
            id.into(),
            value,
        )
    }

    pub fn delete(self, collection: impl Into<String>, id: impl Into<String>) -> Self {
        self.push(
            TransactionOperationKind::Delete,
            collection.into(),
            id.into(),
            Value::Null,
        )
    }

    /// Merge `value` into a record only if its `__version` is still `version`. Refused with
    /// `PRECONDITION_FAILED` otherwise; nothing in the transaction is written.
    pub fn update_if_version(
        self,
        collection: impl Into<String>,
        id: impl Into<String>,
        value: Value,
        version: u64,
    ) -> Self {
        self.operation(TransactionOperation {
            kind: TransactionOperationKind::Update,
            collection: collection.into(),
            id: id.into(),
            value,
            if_version: Some(version),
            if_version_token: None,
        })
    }

    /// [`AppTransaction::update_if_version`] fenced on a version token
    /// (`fv1:<history>:<n>`, from [`version_token`](crate::version_token)),
    /// which a restore of the authority can never make valid again.
    pub fn update_if_token(
        self,
        collection: impl Into<String>,
        id: impl Into<String>,
        value: Value,
        token: impl Into<String>,
    ) -> Self {
        self.operation(TransactionOperation {
            kind: TransactionOperationKind::Update,
            collection: collection.into(),
            id: id.into(),
            value,
            if_version: None,
            if_version_token: Some(token.into()),
        })
    }

    /// Delete a record only if its `__version` is still `version`. With an
    /// [`AppTransaction::insert`] of the same id after it, an atomic, fenced replace.
    pub fn delete_if_version(
        self,
        collection: impl Into<String>,
        id: impl Into<String>,
        version: u64,
    ) -> Self {
        self.operation(TransactionOperation {
            kind: TransactionOperationKind::Delete,
            collection: collection.into(),
            id: id.into(),
            value: Value::Null,
            if_version: Some(version),
            if_version_token: None,
        })
    }

    /// [`AppTransaction::delete_if_version`] fenced on a version token.
    pub fn delete_if_token(
        self,
        collection: impl Into<String>,
        id: impl Into<String>,
        token: impl Into<String>,
    ) -> Self {
        self.operation(TransactionOperation {
            kind: TransactionOperationKind::Delete,
            collection: collection.into(),
            id: id.into(),
            value: Value::Null,
            if_version: None,
            if_version_token: Some(token.into()),
        })
    }

    /// Any operation of the canonical contract, as the wire document states it.
    pub fn operation(mut self, operation: TransactionOperation) -> Self {
        self.operations.push(operation);
        self
    }

    /// Any precondition of the canonical contract, e.g. a record's `__version`:
    ///
    /// ```
    /// # use feltdb_client::{AppTransaction, wire::TransactionPrecondition};
    /// let tx = AppTransaction::new("tx-1").precondition(TransactionPrecondition::Record {
    ///     collection: "Order".into(),
    ///     id: "o-1".into(),
    ///     require_absent: false,
    ///     expected_version: Some(3),
    ///     expected_version_token: None,
    ///     expected_epoch: None,
    ///     expected_lease_id: None,
    /// });
    /// # let _ = tx;
    /// ```
    pub fn precondition(mut self, precondition: TransactionPrecondition) -> Self {
        self.preconditions.push(precondition);
        self
    }

    /// Commit only if the record's `__version` is still `version`.
    pub fn expect_version(
        self,
        collection: impl Into<String>,
        id: impl Into<String>,
        version: u64,
    ) -> Self {
        self.precondition(TransactionPrecondition::Record {
            collection: collection.into(),
            id: id.into(),
            require_absent: false,
            expected_version: Some(version),
            expected_version_token: None,
            expected_epoch: None,
            expected_lease_id: None,
        })
    }

    /// [`AppTransaction::expect_version`] with a version token.
    pub fn expect_token(
        self,
        collection: impl Into<String>,
        id: impl Into<String>,
        token: impl Into<String>,
    ) -> Self {
        self.precondition(TransactionPrecondition::Record {
            collection: collection.into(),
            id: id.into(),
            require_absent: false,
            expected_version: None,
            expected_version_token: Some(token.into()),
            expected_epoch: None,
            expected_lease_id: None,
        })
    }

    /// Commit only if the record does not exist.
    pub fn expect_absent(self, collection: impl Into<String>, id: impl Into<String>) -> Self {
        self.precondition(TransactionPrecondition::Record {
            collection: collection.into(),
            id: id.into(),
            require_absent: true,
            expected_version: None,
            expected_version_token: None,
            expected_epoch: None,
            expected_lease_id: None,
        })
    }

    /// Commit only if the state `basis` covers is unchanged since [`Application::read_basis`].
    pub fn fence(mut self, basis: ReadBasis) -> Self {
        self.fences.push(basis);
        self
    }
}

/// What a canonical transaction committed.
#[derive(Debug, Clone, Deserialize)]
pub struct AppCommit {
    #[serde(flatten)]
    pub result: TransactionResult,
    pub revision_id: String,
    #[serde(default)]
    pub operation_ids: Vec<String>,
}

/// The application's committed state version: [`Application::state_version`].
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct StateVersion {
    pub application_id: String,
    pub revision_id: String,
    pub schema_version: u64,
    /// Advances inside the commit critical section, so it totally orders commits.
    pub state_version: u64,
    /// The state namespace the version counts. Versions of two namespaces are not comparable.
    #[serde(default)]
    pub state_namespace: Option<String>,
    /// Per-operation versions, for a causal read.
    #[serde(default)]
    pub causal_cursor: Value,
}

/// One page of committed history.
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct ReplayPage {
    pub contract_version: String,
    pub cursor_format: String,
    pub application_id: String,
    pub environment: String,
    /// Committed events in order; each carries its own resumable cursor in `metadata.cursor`.
    pub events: Vec<Value>,
    /// Where the next page starts. Bound to this authority, tenant and application.
    pub next_cursor: String,
}

/// One application environment on a FeltDB server. Cheap to clone.
#[derive(Debug, Clone)]
pub struct Application {
    client: Client,
    application_id: String,
    environment: String,
    /// The active revision, cached after a successful discovery. A failure is never cached.
    revision: Arc<Mutex<Option<String>>>,
}

#[derive(Serialize)]
struct Envelope<'a, T: Serialize> {
    application_id: &'a str,
    revision_id: &'a str,
    environment: &'a str,
    #[serde(flatten)]
    payload: T,
}

impl Application {
    pub(crate) fn new(client: Client, application_id: String) -> Self {
        Self {
            client,
            application_id,
            environment: "production".into(),
            revision: Arc::default(),
        }
    }

    /// The same application in another environment.
    pub fn environment(mut self, environment: impl Into<String>) -> Self {
        self.environment = environment.into();
        self.revision = Arc::default();
        self
    }

    pub fn id(&self) -> &str {
        &self.application_id
    }

    pub fn environment_name(&self) -> &str {
        &self.environment
    }

    fn scoped(&self, path: &str) -> String {
        let separator = if path.contains('?') { '&' } else { '?' };
        format!(
            "/v1{path}{separator}application_id={}&environment={}",
            encode(&self.application_id),
            encode(&self.environment)
        )
    }

    /// Discover the environment's active contract, and remember its revision for later requests.
    pub async fn discover(&self) -> Result<ApplicationInfo> {
        let info: ApplicationInfo = self
            .client
            .send::<_, ()>(Method::GET, &self.scoped("/application"), None)
            .await?;
        *self
            .revision
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(info.revision_id.clone());
        Ok(info)
    }

    /// The revision canonical requests are evaluated against: cached after the first discovery.
    /// Call [`Application::discover`] again after a promotion to move to the new revision.
    pub async fn revision_id(&self) -> Result<String> {
        let cached = self
            .revision
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clone();
        match cached {
            Some(revision) => Ok(revision),
            None => Ok(self.discover().await?.revision_id),
        }
    }

    async fn canonical<T: serde::de::DeserializeOwned>(
        &self,
        path: &str,
        payload: impl Serialize,
    ) -> Result<T> {
        let revision_id = self.revision_id().await?;
        let body = Envelope {
            application_id: &self.application_id,
            revision_id: &revision_id,
            environment: &self.environment,
            payload,
        };
        self.client
            .send(Method::POST, &self.scoped(path), Some(&body))
            .await
    }

    /// The committed state version of the environment's state, at its active revision.
    pub async fn state_version(&self) -> Result<StateVersion> {
        let revision_id = self.revision_id().await?;
        let path = self.scoped(&format!(
            "/state/version?revision_id={}",
            encode(&revision_id)
        ));
        self.client.send::<_, ()>(Method::GET, &path, None).await
    }

    /// Run a canonical query, authorized by the contract's read policy for the collection.
    pub async fn query(&self, query: &CanonicalQuery) -> Result<QueryResult> {
        self.canonical("/query", json!({ "query": query })).await
    }

    /// Commit a canonical transaction.
    pub async fn commit(&self, transaction: &AppTransaction) -> Result<AppCommit> {
        if transaction.operations.is_empty() {
            return Err(Error::InvalidRequest(
                "a transaction requires at least one operation".into(),
            ));
        }
        let revision_id = self.revision_id().await?;
        // Tenant, revision, schema version and authorization are the authority's to assign: it
        // overwrites them from the authenticated principal and the active contract. They are sent
        // because the document requires them, never as a claim.
        let tenant_id = match &self.client.inner.scope {
            Some(crate::AuthorityScope::Tenant { tenant_id, .. }) => tenant_id.clone(),
            _ => String::new(),
        };
        let request = TransactionRequest {
            transaction_id: Some(transaction.transaction_id.clone()),
            tenant_id: tenant_id.clone(),
            application_id: self.application_id.clone(),
            revision_id: revision_id.clone(),
            schema_version: 0,
            state_namespace: None,
            causal_parent: None,
            authorization: AuthorizationContext {
                subject: String::new(),
                tenant_id,
                application_id: self.application_id.clone(),
                revision_id,
                capabilities: BTreeSet::new(),
                readable_collections: BTreeSet::new(),
                writable_collections: BTreeSet::new(),
                field_projections: Default::default(),
            },
            operations: transaction.operations.clone(),
            preconditions: transaction.preconditions.clone(),
            fences: transaction.fences.clone(),
        };
        self.canonical("/transactions", json!({ "transaction": request }))
            .await
    }

    /// Capture a read basis over the application's state, under the caller's read authorization.
    pub async fn read_basis(&self, reads: &ReadSet) -> Result<ReadBasis> {
        self.canonical("/transactions/read-basis", json!({ "reads": reads }))
            .await
    }

    /// Committed changes to this application as they happen, from `cursor` or from now.
    pub async fn events(&self, cursor: Option<&str>) -> Result<EventStream> {
        self.client
            .open_events(&self.scoped("/events"), cursor)
            .await
    }

    /// A bounded page of committed history from `cursor` (the start when `None`), at most `limit`
    /// events (the server caps it at 256). Replay, not the live stream, is the authority on history.
    pub async fn replay(&self, cursor: Option<&str>, limit: Option<usize>) -> Result<ReplayPage> {
        let mut path = self.scoped("/public/events/replay");
        if let Some(cursor) = cursor {
            path.push_str(&format!("&cursor={}", encode(cursor)));
        }
        if let Some(limit) = limit {
            path.push_str(&format!("&limit={limit}"));
        }
        self.client.send::<_, ()>(Method::GET, &path, None).await
    }

    /// The complete mutation chain of one record.
    pub async fn provenance(&self, collection: &str, id: &str) -> Result<Value> {
        let path = self.scoped(&format!(
            "/control-plane/provenance/{}/{}",
            encode(collection),
            encode(id)
        ));
        self.client.send::<_, ()>(Method::GET, &path, None).await
    }

    /// Sign an end user in to this application; the session's access token becomes this client's
    /// bearer token.
    pub async fn sign_in(&self, email: &str, password: &str) -> Result<Value> {
        let body =
            json!({ "application_id": self.application_id, "email": email, "password": password });
        self.authenticate("/v1/auth/signin", &body).await
    }

    /// Register an end user of this application and sign them in.
    pub async fn sign_up(
        &self,
        email: &str,
        password: &str,
        display_name: Option<&str>,
    ) -> Result<Value> {
        let body = json!({
            "application_id": self.application_id,
            "email": email,
            "password": password,
            "display_name": display_name,
        });
        self.authenticate("/v1/auth/signup", &body).await
    }

    async fn authenticate(&self, path: &str, body: &Value) -> Result<Value> {
        let session: Value = self.client.send(Method::POST, path, Some(body)).await?;
        if let Some(token) = session.get("access_token").and_then(Value::as_str) {
            self.client.set_token(token);
        }
        Ok(session)
    }

    /// The current session's principal.
    pub async fn session(&self) -> Result<Value> {
        self.client
            .send::<_, ()>(Method::GET, "/v1/auth/session", None)
            .await
    }

    /// End the current session and clear this client's token.
    pub async fn sign_out(&self) -> Result<()> {
        self.client
            .exchange::<()>(Method::POST, "/v1/auth/signout", None)
            .await?;
        self.client.set_token("");
        Ok(())
    }
}

impl Client {
    /// The server's versioned capability declaration.
    pub async fn capabilities(&self) -> Result<Value> {
        self.send::<_, ()>(Method::GET, "/v1/control-plane/capabilities", None)
            .await
    }
}

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

    #[test]
    fn scoped_paths_carry_the_application_and_environment() {
        let app = Client::builder("http://localhost:1")
            .build()
            .unwrap()
            .application("app 1")
            .environment("staging");
        assert_eq!(
            app.scoped("/application"),
            "/v1/application?application_id=app%201&environment=staging"
        );
        assert_eq!(
            app.scoped("/x?y=1"),
            "/v1/x?y=1&application_id=app%201&environment=staging"
        );
    }
}