feltdb-client 0.2.2

//! The direct authority surface: records, transactions, compare-and-set, the bounded query, the
//! committed revision and the event stream.

use feltdb::{ReadBasis, ReadSet, SemanticQueryClause, TransactionStatus};
use serde::{ser::SerializeMap, Deserialize, Serialize, Serializer};
use serde_json::{Map, Value};

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

/// A stored record.
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct Record {
    /// The record id within its collection.
    pub id: String,
    /// The storage key, `collection:id`.
    pub key: String,
    /// The document, including its `id` and, after a fenced write, its `__version`.
    pub value: Value,
}

/// The committed revision of the authority, and the identity of the store that counted it.
///
/// Two revisions are comparable only when their scopes are equal: behind a load balancer or after
/// a failover the URL stays the same while the store, and so the counter, changes.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
pub struct Revision {
    pub revision: u64,
    pub scope: String,
}

/// One write inside a [`Transaction`].
#[derive(Debug, Clone, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Operation {
    collection: String,
    id: String,
    /// `None` deletes the record.
    #[serde(skip_serializing_if = "Option::is_none")]
    value: Option<Value>,
    #[serde(skip_serializing_if = "std::ops::Not::not")]
    require_absent: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    expected_version: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    expected_epoch: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    expected_lease_id: Option<String>,
}

impl Operation {
    fn new(collection: impl Into<String>, id: impl Into<String>, value: Option<Value>) -> Self {
        Self {
            collection: collection.into(),
            id: id.into(),
            value,
            require_absent: false,
            expected_version: None,
            expected_epoch: None,
            expected_lease_id: None,
        }
    }

    /// Write `value` (a JSON object) as the record, whether or not it exists.
    pub fn put(collection: impl Into<String>, id: impl Into<String>, value: Value) -> Self {
        Self::new(collection, id, Some(value))
    }

    /// Create the record; the whole transaction is refused if it already exists. The authority
    /// assigns the created document `__version` 1.
    pub fn create(collection: impl Into<String>, id: impl Into<String>, value: Value) -> Self {
        Self {
            require_absent: true,
            ..Self::new(collection, id, Some(value))
        }
    }

    /// Delete the record.
    pub fn delete(collection: impl Into<String>, id: impl Into<String>) -> Self {
        Self::new(collection, id, None)
    }

    /// Commit only if the record's `__version` is still `version`.
    pub fn expect_version(mut self, version: u64) -> Self {
        self.expected_version = Some(version);
        self
    }

    /// Commit only if the record's authority epoch is still `epoch`.
    pub fn expect_epoch(mut self, epoch: u64) -> Self {
        self.expected_epoch = Some(epoch);
        self
    }

    /// Commit only while `lease_id` is an unexpired lease on the record.
    pub fn expect_lease(mut self, lease_id: impl Into<String>) -> Self {
        self.expected_lease_id = Some(lease_id.into());
        self
    }
}

/// A condition on a record the transaction need not write, evaluated inside the commit.
#[derive(Debug, Clone, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Precondition {
    collection: String,
    id: String,
    #[serde(skip_serializing_if = "std::ops::Not::not")]
    require_absent: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    expected_version: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    expected_epoch: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    expected_lease_id: Option<String>,
}

impl Precondition {
    fn new(collection: impl Into<String>, id: impl Into<String>) -> Self {
        Self {
            collection: collection.into(),
            id: id.into(),
            require_absent: false,
            expected_version: None,
            expected_epoch: None,
            expected_lease_id: None,
        }
    }

    /// The record must not exist.
    pub fn absent(collection: impl Into<String>, id: impl Into<String>) -> Self {
        Self {
            require_absent: true,
            ..Self::new(collection, id)
        }
    }

    /// The record's `__version` must be `version`.
    pub fn version(collection: impl Into<String>, id: impl Into<String>, version: u64) -> Self {
        Self {
            expected_version: Some(version),
            ..Self::new(collection, id)
        }
    }

    /// The record's authority epoch must be `epoch`.
    pub fn epoch(collection: impl Into<String>, id: impl Into<String>, epoch: u64) -> Self {
        Self {
            expected_epoch: Some(epoch),
            ..Self::new(collection, id)
        }
    }

    /// `lease_id` must be an unexpired lease on the record.
    pub fn lease(
        collection: impl Into<String>,
        id: impl Into<String>,
        lease_id: impl Into<String>,
    ) -> Self {
        Self {
            expected_lease_id: Some(lease_id.into()),
            ..Self::new(collection, id)
        }
    }
}

/// Several writes committed as one durable record, or not at all.
///
/// The id is the transaction's identity. Resubmitting the same transaction with the same id after
/// an uncertain outcome returns the original commit with `duplicate: true`; the same id with
/// different content is refused with `IDEMPOTENCY_CONFLICT`.
#[derive(Debug, Clone, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Transaction {
    transaction_id: String,
    operations: Vec<Operation>,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    preconditions: Vec<Precondition>,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    fences: Vec<ReadBasis>,
}

impl Transaction {
    /// A transaction with this identity.
    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. Keep [`Transaction::id`] if you may need to
    /// resubmit it.
    pub fn generated() -> Self {
        Self::new(format!("tx-{}", uuid::Uuid::new_v4()))
    }

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

    pub fn operation(mut self, operation: Operation) -> Self {
        self.operations.push(operation);
        self
    }

    pub fn precondition(mut self, precondition: Precondition) -> Self {
        self.preconditions.push(precondition);
        self
    }

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

    fn validate(&self) -> Result<()> {
        if self.transaction_id.trim().is_empty() {
            return Err(Error::InvalidRequest("a transaction id is required".into()));
        }
        if self.operations.is_empty() {
            return Err(Error::InvalidRequest(
                "a transaction requires at least one operation".into(),
            ));
        }
        for operation in &self.operations {
            if operation
                .value
                .as_ref()
                .is_some_and(|value| !value.is_object())
            {
                return Err(Error::InvalidRequest(format!(
                    "the value written to {}/{} must be a JSON object",
                    operation.collection, operation.id
                )));
            }
        }
        Ok(())
    }
}

/// What the authority committed.
#[derive(Debug, Clone, PartialEq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CommitReceipt {
    pub transaction_id: String,
    pub base_revision: u64,
    pub commit_revision: u64,
    pub status: TransactionStatus,
    /// True when this id had already committed: nothing was applied again.
    pub duplicate: bool,
    pub operations: usize,
    pub state_before: u64,
    pub state_after: u64,
    pub revision_id: String,
    #[serde(default)]
    pub operation_ids: Vec<String>,
}

/// A conditional update of one record.
#[derive(Debug, Clone, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct CasRequest {
    expected_version: u64,
    #[serde(skip_serializing_if = "Option::is_none")]
    expected_epoch: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    expected_lease_id: Option<String>,
    #[serde(skip_serializing_if = "std::ops::Not::not")]
    reject_if_diverged: bool,
    value: Value,
}

impl CasRequest {
    /// Replace the record with `value` if its `__version` is `expected_version`. A record that has
    /// never been through a fenced write is version 1.
    pub fn new(expected_version: u64, value: Value) -> Self {
        Self {
            expected_version,
            expected_epoch: None,
            expected_lease_id: None,
            reject_if_diverged: false,
            value,
        }
    }

    pub fn expect_epoch(mut self, epoch: u64) -> Self {
        self.expected_epoch = Some(epoch);
        self
    }

    pub fn expect_lease(mut self, lease_id: impl Into<String>) -> Self {
        self.expected_lease_id = Some(lease_id.into());
        self
    }

    /// Also refuse when the record has diverged across replicas.
    pub fn reject_if_diverged(mut self) -> Self {
        self.reject_if_diverged = true;
        self
    }
}

/// Which predicate refused a compare-and-set.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
pub enum CasConflict {
    #[serde(rename = "VERSION_CONFLICT")]
    Version,
    #[serde(rename = "AUTHORITY_CONFLICT")]
    Authority,
    #[serde(rename = "LEASE_CONFLICT")]
    Lease,
    #[serde(rename = "REPLICATION_CONFLICT")]
    Replication,
}

/// How a compare-and-set ended. A lost race is a value, not an error.
#[derive(Debug, Clone, PartialEq)]
pub enum CasOutcome {
    Updated {
        current_version: u64,
        current_epoch: u64,
        /// The document as committed.
        item: Value,
    },
    /// Nothing was written; this is what the authority holds.
    Conflict {
        conflict: CasConflict,
        current_version: u64,
        current_epoch: u64,
    },
    NotFound,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct CasBody {
    updated: bool,
    #[serde(default)]
    code: Option<String>,
    #[serde(default)]
    current_version: u64,
    #[serde(default)]
    current_epoch: u64,
    #[serde(default)]
    item: Value,
}

/// A comparison operator of the bounded query.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Operator {
    Eq,
    Neq,
    Lt,
    Lte,
    Gt,
    Gte,
}

/// One `where` condition of a [`BoundedQuery`].
#[derive(Debug, Clone, PartialEq)]
pub struct Condition {
    field: String,
    operator: Operator,
    value: Value,
}

impl Serialize for Condition {
    fn serialize<S: Serializer>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error> {
        let operator = match self.operator {
            Operator::Eq => "eq",
            Operator::Neq => "neq",
            Operator::Lt => "lt",
            Operator::Lte => "lte",
            Operator::Gt => "gt",
            Operator::Gte => "gte",
        };
        let mut map = serializer.serialize_map(Some(2))?;
        map.serialize_entry("field", &self.field)?;
        map.serialize_entry(operator, &self.value)?;
        map.end()
    }
}

/// Sort direction.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum Direction {
    Asc,
    Desc,
}

#[derive(Debug, Clone, PartialEq, Serialize)]
struct Order {
    field: String,
    direction: Direction,
}

/// The bounded authority query: a page of at most `limit` records in a total order, with an opaque
/// cursor for the next page. It never reads more than it returns plus what its plan requires.
#[derive(Debug, Clone, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct BoundedQuery {
    collection: String,
    #[serde(rename = "where", skip_serializing_if = "Vec::is_empty")]
    conditions: Vec<Condition>,
    order_by: Vec<Order>,
    limit: usize,
    #[serde(skip_serializing_if = "Option::is_none")]
    cursor: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    semantic: Option<SemanticQueryClause>,
}

impl BoundedQuery {
    /// A query over `collection` returning at most `limit` records. Add at least one
    /// [`BoundedQuery::order_by`]: the server requires a total order.
    pub fn new(collection: impl Into<String>, limit: usize) -> Self {
        Self {
            collection: collection.into(),
            conditions: Vec::new(),
            order_by: Vec::new(),
            limit,
            cursor: None,
            semantic: None,
        }
    }

    fn condition(mut self, field: impl Into<String>, operator: Operator, value: Value) -> Self {
        self.conditions.push(Condition {
            field: field.into(),
            operator,
            value,
        });
        self
    }

    pub fn eq(self, field: impl Into<String>, value: Value) -> Self {
        self.condition(field, Operator::Eq, value)
    }
    pub fn neq(self, field: impl Into<String>, value: Value) -> Self {
        self.condition(field, Operator::Neq, value)
    }
    pub fn lt(self, field: impl Into<String>, value: Value) -> Self {
        self.condition(field, Operator::Lt, value)
    }
    pub fn lte(self, field: impl Into<String>, value: Value) -> Self {
        self.condition(field, Operator::Lte, value)
    }
    pub fn gt(self, field: impl Into<String>, value: Value) -> Self {
        self.condition(field, Operator::Gt, value)
    }
    pub fn gte(self, field: impl Into<String>, value: Value) -> Self {
        self.condition(field, Operator::Gte, value)
    }

    pub fn order_by(mut self, field: impl Into<String>, direction: Direction) -> Self {
        self.order_by.push(Order {
            field: field.into(),
            direction,
        });
        self
    }

    /// Continue from a previous page's [`QueryPage::next_cursor`].
    pub fn after(mut self, cursor: impl Into<String>) -> Self {
        self.cursor = Some(cursor.into());
        self
    }

    /// Compose a semantic clause over the page this query produces (experimental; see
    /// `docs/reference/semantic-query.md`). The page is never enlarged or refilled.
    pub fn semantic(mut self, clause: SemanticQueryClause) -> Self {
        self.semantic = Some(clause);
        self
    }
}

/// One page of a [`BoundedQuery`].
#[derive(Debug, Clone, PartialEq, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct QueryPage {
    pub records: Vec<Value>,
    #[serde(default)]
    pub next_cursor: Option<String>,
    pub exhausted: bool,
    /// Present exactly when the query carried a semantic clause.
    #[serde(default)]
    pub semantic: Option<Value>,
}

impl Client {
    /// The record `collection/id`, or `None` if it does not exist.
    pub async fn get(&self, collection: &str, id: &str) -> Result<Option<Record>> {
        match self
            .send::<Record, ()>(Method::GET, &record_path(collection, id), None)
            .await
        {
            Ok(record) => Ok(Some(record)),
            Err(Error::Api(error)) if error.status == 404 => Ok(None),
            Err(error) => Err(error),
        }
    }

    /// Every record of `collection`. This reads the whole collection: on a request path use
    /// [`Client::query`], which is bounded.
    pub async fn list(&self, collection: &str) -> Result<Vec<Record>> {
        self.send::<_, ()>(Method::GET, &collection_path(collection), None)
            .await
    }

    /// Create a record only if `id` is free; refused with HTTP 409 if it exists. `value` must be a
    /// JSON object. Without an id, the server assigns one.
    pub async fn create(&self, collection: &str, id: Option<&str>, value: Value) -> Result<Record> {
        let Value::Object(mut fields) = value else {
            return Err(Error::InvalidRequest(
                "a record must be a JSON object".into(),
            ));
        };
        if let Some(id) = id {
            fields.insert("id".into(), Value::String(id.to_owned()));
        }
        self.send(Method::POST, &collection_path(collection), Some(&fields))
            .await
    }

    /// Merge `changes` into an existing record (unconditionally). Refused with HTTP 404 if the
    /// record does not exist; use a [`Transaction`] or [`Client::compare_and_set`] to fence it.
    pub async fn update(
        &self,
        collection: &str,
        id: &str,
        changes: Map<String, Value>,
    ) -> Result<Record> {
        self.send(Method::PATCH, &record_path(collection, id), Some(&changes))
            .await
    }

    /// Delete an existing record. Refused with HTTP 404 if it does not exist.
    pub async fn delete(&self, collection: &str, id: &str) -> Result<()> {
        self.exchange::<()>(Method::DELETE, &record_path(collection, id), None)
            .await
            .map(drop)
    }

    /// Replace a record if it still has the version (and epoch and lease) you read.
    pub async fn compare_and_set(
        &self,
        collection: &str,
        id: &str,
        request: &CasRequest,
    ) -> Result<CasOutcome> {
        let path = format!("{}/cas", record_path(collection, id));
        let body: CasBody = match self.send(Method::POST, &path, Some(request)).await {
            Ok(body) => body,
            // A conflict is an answer, carried in the error body; anything else stays an error.
            Err(Error::Api(error)) if error.status == 409 || error.status == 404 => {
                if error.status == 404 && error.code == "NOT_FOUND" {
                    return Ok(CasOutcome::NotFound);
                }
                match serde_json::from_value::<CasBody>(error.body.clone()) {
                    Ok(body) if body.code.is_some() => body,
                    _ => return Err(Error::Api(error)),
                }
            }
            Err(error) => return Err(error),
        };
        if body.updated {
            return Ok(CasOutcome::Updated {
                current_version: body.current_version,
                current_epoch: body.current_epoch,
                item: body.item,
            });
        }
        let code = body.code.unwrap_or_default();
        let conflict = serde_json::from_value(Value::String(code.clone())).map_err(|_| {
            Error::InvalidRequest(format!("unrecognized compare-and-set refusal {code:?}"))
        })?;
        Ok(CasOutcome::Conflict {
            conflict,
            current_version: body.current_version,
            current_epoch: body.current_epoch,
        })
    }

    /// Commit a transaction atomically. A refusal applied nothing; a failed precondition or fence
    /// is reported in [`Error::precondition_failure`].
    pub async fn commit(&self, transaction: &Transaction) -> Result<CommitReceipt> {
        transaction.validate()?;
        self.send(Method::POST, "/transactions", Some(transaction))
            .await
    }

    /// Capture what the authority holds for `reads`, to fence a later transaction on it.
    pub async fn read_basis(&self, reads: &ReadSet) -> Result<ReadBasis> {
        self.send(Method::POST, "/transactions/read-basis", Some(reads))
            .await
    }

    /// Run one page of a bounded query.
    pub async fn query(&self, query: &BoundedQuery) -> Result<QueryPage> {
        self.send(Method::POST, "/query", Some(query)).await
    }

    /// The authority's committed revision.
    pub async fn revision(&self) -> Result<Revision> {
        self.send::<_, ()>(Method::GET, "/revision", None).await
    }

    /// Committed changes as they happen, from `cursor` (a previous event's id) or from now.
    pub async fn events(&self, cursor: Option<&str>) -> Result<EventStream> {
        self.open_events("/events", cursor).await
    }
}

fn collection_path(collection: &str) -> String {
    format!("/collections/{}", encode(collection))
}

fn record_path(collection: &str, id: &str) -> String {
    format!("/collections/{}/{}", encode(collection), encode(id))
}

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

    #[test]
    fn a_transaction_serializes_as_the_authority_reads_it() {
        let transaction = Transaction::new("tx-1")
            .operation(Operation::create("orders", "o-1", json!({ "n": 1 })))
            .operation(Operation::put("orders", "o-2", json!({ "n": 2 })).expect_version(3))
            .operation(Operation::delete("orders", "o-3"))
            .precondition(Precondition::absent("locks", "l-1"));
        assert_eq!(
            serde_json::to_value(&transaction).unwrap(),
            json!({
                "transactionId": "tx-1",
                "operations": [
                    { "collection": "orders", "id": "o-1", "value": { "n": 1 }, "requireAbsent": true },
                    { "collection": "orders", "id": "o-2", "value": { "n": 2 }, "expectedVersion": 3 },
                    { "collection": "orders", "id": "o-3" }
                ],
                "preconditions": [{ "collection": "locks", "id": "l-1", "requireAbsent": true }]
            })
        );
    }

    #[test]
    fn a_transaction_the_server_would_refuse_is_refused_here() {
        assert!(Transaction::new("tx").validate().is_err());
        assert!(Transaction::new("tx")
            .operation(Operation::put("c", "i", json!([1])))
            .validate()
            .is_err());
    }

    #[test]
    fn a_bounded_query_serializes_its_conditions_by_operator() {
        let query = BoundedQuery::new("orders", 10)
            .eq("status", json!("open"))
            .gte("total", json!(5))
            .order_by("id", Direction::Asc)
            .after("cursor-1");
        assert_eq!(
            serde_json::to_value(&query).unwrap(),
            json!({
                "collection": "orders",
                "where": [{ "field": "status", "eq": "open" }, { "field": "total", "gte": 5 }],
                "orderBy": [{ "field": "id", "direction": "asc" }],
                "limit": 10,
                "cursor": "cursor-1"
            })
        );
    }
}