feltdb_client/
authority.rs

1//! The direct authority surface: records, transactions, compare-and-set, the bounded query, the
2//! committed revision and the event stream.
3
4use feltdb::{ReadBasis, ReadSet, SemanticQueryClause, TransactionStatus};
5use serde::{ser::SerializeMap, Deserialize, Serialize, Serializer};
6use serde_json::{Map, Value};
7
8use crate::{encode, Client, Error, EventStream, Method, Result};
9
10/// A stored record.
11#[derive(Debug, Clone, PartialEq, Deserialize)]
12pub struct Record {
13    /// The record id within its collection.
14    pub id: String,
15    /// The storage key, `collection:id`.
16    pub key: String,
17    /// The document, including its `id` and, after a fenced write, its `__version`.
18    pub value: Value,
19}
20
21/// The committed revision of the authority, and the identity of the store that counted it.
22///
23/// Two revisions are comparable only when their scopes are equal: behind a load balancer or after
24/// a failover the URL stays the same while the store, and so the counter, changes.
25#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
26pub struct Revision {
27    pub revision: u64,
28    pub scope: String,
29}
30
31/// One write inside a [`Transaction`].
32#[derive(Debug, Clone, PartialEq, Serialize)]
33#[serde(rename_all = "camelCase")]
34pub struct Operation {
35    collection: String,
36    id: String,
37    /// `None` deletes the record.
38    #[serde(skip_serializing_if = "Option::is_none")]
39    value: Option<Value>,
40    #[serde(skip_serializing_if = "std::ops::Not::not")]
41    require_absent: bool,
42    #[serde(skip_serializing_if = "Option::is_none")]
43    expected_version: Option<u64>,
44    #[serde(skip_serializing_if = "Option::is_none")]
45    expected_epoch: Option<u64>,
46    #[serde(skip_serializing_if = "Option::is_none")]
47    expected_lease_id: Option<String>,
48}
49
50impl Operation {
51    fn new(collection: impl Into<String>, id: impl Into<String>, value: Option<Value>) -> Self {
52        Self {
53            collection: collection.into(),
54            id: id.into(),
55            value,
56            require_absent: false,
57            expected_version: None,
58            expected_epoch: None,
59            expected_lease_id: None,
60        }
61    }
62
63    /// Write `value` (a JSON object) as the record, whether or not it exists.
64    pub fn put(collection: impl Into<String>, id: impl Into<String>, value: Value) -> Self {
65        Self::new(collection, id, Some(value))
66    }
67
68    /// Create the record; the whole transaction is refused if it already exists. The authority
69    /// assigns the created document `__version` 1.
70    pub fn create(collection: impl Into<String>, id: impl Into<String>, value: Value) -> Self {
71        Self {
72            require_absent: true,
73            ..Self::new(collection, id, Some(value))
74        }
75    }
76
77    /// Delete the record.
78    pub fn delete(collection: impl Into<String>, id: impl Into<String>) -> Self {
79        Self::new(collection, id, None)
80    }
81
82    /// Commit only if the record's `__version` is still `version`.
83    pub fn expect_version(mut self, version: u64) -> Self {
84        self.expected_version = Some(version);
85        self
86    }
87
88    /// Commit only if the record's authority epoch is still `epoch`.
89    pub fn expect_epoch(mut self, epoch: u64) -> Self {
90        self.expected_epoch = Some(epoch);
91        self
92    }
93
94    /// Commit only while `lease_id` is an unexpired lease on the record.
95    pub fn expect_lease(mut self, lease_id: impl Into<String>) -> Self {
96        self.expected_lease_id = Some(lease_id.into());
97        self
98    }
99}
100
101/// A condition on a record the transaction need not write, evaluated inside the commit.
102#[derive(Debug, Clone, PartialEq, Serialize)]
103#[serde(rename_all = "camelCase")]
104pub struct Precondition {
105    collection: String,
106    id: String,
107    #[serde(skip_serializing_if = "std::ops::Not::not")]
108    require_absent: bool,
109    #[serde(skip_serializing_if = "Option::is_none")]
110    expected_version: Option<u64>,
111    #[serde(skip_serializing_if = "Option::is_none")]
112    expected_epoch: Option<u64>,
113    #[serde(skip_serializing_if = "Option::is_none")]
114    expected_lease_id: Option<String>,
115}
116
117impl Precondition {
118    fn new(collection: impl Into<String>, id: impl Into<String>) -> Self {
119        Self {
120            collection: collection.into(),
121            id: id.into(),
122            require_absent: false,
123            expected_version: None,
124            expected_epoch: None,
125            expected_lease_id: None,
126        }
127    }
128
129    /// The record must not exist.
130    pub fn absent(collection: impl Into<String>, id: impl Into<String>) -> Self {
131        Self {
132            require_absent: true,
133            ..Self::new(collection, id)
134        }
135    }
136
137    /// The record's `__version` must be `version`.
138    pub fn version(collection: impl Into<String>, id: impl Into<String>, version: u64) -> Self {
139        Self {
140            expected_version: Some(version),
141            ..Self::new(collection, id)
142        }
143    }
144
145    /// The record's authority epoch must be `epoch`.
146    pub fn epoch(collection: impl Into<String>, id: impl Into<String>, epoch: u64) -> Self {
147        Self {
148            expected_epoch: Some(epoch),
149            ..Self::new(collection, id)
150        }
151    }
152
153    /// `lease_id` must be an unexpired lease on the record.
154    pub fn lease(
155        collection: impl Into<String>,
156        id: impl Into<String>,
157        lease_id: impl Into<String>,
158    ) -> Self {
159        Self {
160            expected_lease_id: Some(lease_id.into()),
161            ..Self::new(collection, id)
162        }
163    }
164}
165
166/// Several writes committed as one durable record, or not at all.
167///
168/// The id is the transaction's identity. Resubmitting the same transaction with the same id after
169/// an uncertain outcome returns the original commit with `duplicate: true`; the same id with
170/// different content is refused with `IDEMPOTENCY_CONFLICT`.
171#[derive(Debug, Clone, PartialEq, Serialize)]
172#[serde(rename_all = "camelCase")]
173pub struct Transaction {
174    transaction_id: String,
175    operations: Vec<Operation>,
176    #[serde(skip_serializing_if = "Vec::is_empty")]
177    preconditions: Vec<Precondition>,
178    #[serde(skip_serializing_if = "Vec::is_empty")]
179    fences: Vec<ReadBasis>,
180}
181
182impl Transaction {
183    /// A transaction with this identity.
184    pub fn new(transaction_id: impl Into<String>) -> Self {
185        Self {
186            transaction_id: transaction_id.into(),
187            operations: Vec::new(),
188            preconditions: Vec::new(),
189            fences: Vec::new(),
190        }
191    }
192
193    /// A transaction with a fresh random identity. Keep [`Transaction::id`] if you may need to
194    /// resubmit it.
195    pub fn generated() -> Self {
196        Self::new(format!("tx-{}", uuid::Uuid::new_v4()))
197    }
198
199    pub fn id(&self) -> &str {
200        &self.transaction_id
201    }
202
203    pub fn operation(mut self, operation: Operation) -> Self {
204        self.operations.push(operation);
205        self
206    }
207
208    pub fn precondition(mut self, precondition: Precondition) -> Self {
209        self.preconditions.push(precondition);
210        self
211    }
212
213    /// Commit only if the state `basis` covers is unchanged since [`Client::read_basis`] captured it.
214    pub fn fence(mut self, basis: ReadBasis) -> Self {
215        self.fences.push(basis);
216        self
217    }
218
219    fn validate(&self) -> Result<()> {
220        if self.transaction_id.trim().is_empty() {
221            return Err(Error::InvalidRequest("a transaction id is required".into()));
222        }
223        if self.operations.is_empty() {
224            return Err(Error::InvalidRequest(
225                "a transaction requires at least one operation".into(),
226            ));
227        }
228        for operation in &self.operations {
229            if operation
230                .value
231                .as_ref()
232                .is_some_and(|value| !value.is_object())
233            {
234                return Err(Error::InvalidRequest(format!(
235                    "the value written to {}/{} must be a JSON object",
236                    operation.collection, operation.id
237                )));
238            }
239        }
240        Ok(())
241    }
242}
243
244/// What the authority committed.
245#[derive(Debug, Clone, PartialEq, Deserialize)]
246#[serde(rename_all = "camelCase")]
247pub struct CommitReceipt {
248    pub transaction_id: String,
249    pub base_revision: u64,
250    pub commit_revision: u64,
251    pub status: TransactionStatus,
252    /// True when this id had already committed: nothing was applied again.
253    pub duplicate: bool,
254    pub operations: usize,
255    pub state_before: u64,
256    pub state_after: u64,
257    pub revision_id: String,
258    #[serde(default)]
259    pub operation_ids: Vec<String>,
260}
261
262/// A conditional update of one record.
263#[derive(Debug, Clone, PartialEq, Serialize)]
264#[serde(rename_all = "camelCase")]
265pub struct CasRequest {
266    expected_version: u64,
267    #[serde(skip_serializing_if = "Option::is_none")]
268    expected_epoch: Option<u64>,
269    #[serde(skip_serializing_if = "Option::is_none")]
270    expected_lease_id: Option<String>,
271    #[serde(skip_serializing_if = "std::ops::Not::not")]
272    reject_if_diverged: bool,
273    value: Value,
274}
275
276impl CasRequest {
277    /// Replace the record with `value` if its `__version` is `expected_version`. A record that has
278    /// never been through a fenced write is version 1.
279    pub fn new(expected_version: u64, value: Value) -> Self {
280        Self {
281            expected_version,
282            expected_epoch: None,
283            expected_lease_id: None,
284            reject_if_diverged: false,
285            value,
286        }
287    }
288
289    pub fn expect_epoch(mut self, epoch: u64) -> Self {
290        self.expected_epoch = Some(epoch);
291        self
292    }
293
294    pub fn expect_lease(mut self, lease_id: impl Into<String>) -> Self {
295        self.expected_lease_id = Some(lease_id.into());
296        self
297    }
298
299    /// Also refuse when the record has diverged across replicas.
300    pub fn reject_if_diverged(mut self) -> Self {
301        self.reject_if_diverged = true;
302        self
303    }
304}
305
306/// Which predicate refused a compare-and-set.
307#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
308pub enum CasConflict {
309    #[serde(rename = "VERSION_CONFLICT")]
310    Version,
311    #[serde(rename = "AUTHORITY_CONFLICT")]
312    Authority,
313    #[serde(rename = "LEASE_CONFLICT")]
314    Lease,
315    #[serde(rename = "REPLICATION_CONFLICT")]
316    Replication,
317}
318
319/// How a compare-and-set ended. A lost race is a value, not an error.
320#[derive(Debug, Clone, PartialEq)]
321pub enum CasOutcome {
322    Updated {
323        current_version: u64,
324        current_epoch: u64,
325        /// The document as committed.
326        item: Value,
327    },
328    /// Nothing was written; this is what the authority holds.
329    Conflict {
330        conflict: CasConflict,
331        current_version: u64,
332        current_epoch: u64,
333    },
334    NotFound,
335}
336
337#[derive(Deserialize)]
338#[serde(rename_all = "camelCase")]
339struct CasBody {
340    updated: bool,
341    #[serde(default)]
342    code: Option<String>,
343    #[serde(default)]
344    current_version: u64,
345    #[serde(default)]
346    current_epoch: u64,
347    #[serde(default)]
348    item: Value,
349}
350
351/// A comparison operator of the bounded query.
352#[derive(Debug, Clone, Copy, PartialEq, Eq)]
353enum Operator {
354    Eq,
355    Neq,
356    Lt,
357    Lte,
358    Gt,
359    Gte,
360}
361
362/// One `where` condition of a [`BoundedQuery`].
363#[derive(Debug, Clone, PartialEq)]
364pub struct Condition {
365    field: String,
366    operator: Operator,
367    value: Value,
368}
369
370impl Serialize for Condition {
371    fn serialize<S: Serializer>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error> {
372        let operator = match self.operator {
373            Operator::Eq => "eq",
374            Operator::Neq => "neq",
375            Operator::Lt => "lt",
376            Operator::Lte => "lte",
377            Operator::Gt => "gt",
378            Operator::Gte => "gte",
379        };
380        let mut map = serializer.serialize_map(Some(2))?;
381        map.serialize_entry("field", &self.field)?;
382        map.serialize_entry(operator, &self.value)?;
383        map.end()
384    }
385}
386
387/// Sort direction.
388#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
389#[serde(rename_all = "lowercase")]
390pub enum Direction {
391    Asc,
392    Desc,
393}
394
395#[derive(Debug, Clone, PartialEq, Serialize)]
396struct Order {
397    field: String,
398    direction: Direction,
399}
400
401/// The bounded authority query: a page of at most `limit` records in a total order, with an opaque
402/// cursor for the next page. It never reads more than it returns plus what its plan requires.
403#[derive(Debug, Clone, PartialEq, Serialize)]
404#[serde(rename_all = "camelCase")]
405pub struct BoundedQuery {
406    collection: String,
407    #[serde(rename = "where", skip_serializing_if = "Vec::is_empty")]
408    conditions: Vec<Condition>,
409    order_by: Vec<Order>,
410    limit: usize,
411    #[serde(skip_serializing_if = "Option::is_none")]
412    cursor: Option<String>,
413    #[serde(skip_serializing_if = "Option::is_none")]
414    semantic: Option<SemanticQueryClause>,
415}
416
417impl BoundedQuery {
418    /// A query over `collection` returning at most `limit` records. Add at least one
419    /// [`BoundedQuery::order_by`]: the server requires a total order.
420    pub fn new(collection: impl Into<String>, limit: usize) -> Self {
421        Self {
422            collection: collection.into(),
423            conditions: Vec::new(),
424            order_by: Vec::new(),
425            limit,
426            cursor: None,
427            semantic: None,
428        }
429    }
430
431    fn condition(mut self, field: impl Into<String>, operator: Operator, value: Value) -> Self {
432        self.conditions.push(Condition {
433            field: field.into(),
434            operator,
435            value,
436        });
437        self
438    }
439
440    pub fn eq(self, field: impl Into<String>, value: Value) -> Self {
441        self.condition(field, Operator::Eq, value)
442    }
443    pub fn neq(self, field: impl Into<String>, value: Value) -> Self {
444        self.condition(field, Operator::Neq, value)
445    }
446    pub fn lt(self, field: impl Into<String>, value: Value) -> Self {
447        self.condition(field, Operator::Lt, value)
448    }
449    pub fn lte(self, field: impl Into<String>, value: Value) -> Self {
450        self.condition(field, Operator::Lte, value)
451    }
452    pub fn gt(self, field: impl Into<String>, value: Value) -> Self {
453        self.condition(field, Operator::Gt, value)
454    }
455    pub fn gte(self, field: impl Into<String>, value: Value) -> Self {
456        self.condition(field, Operator::Gte, value)
457    }
458
459    pub fn order_by(mut self, field: impl Into<String>, direction: Direction) -> Self {
460        self.order_by.push(Order {
461            field: field.into(),
462            direction,
463        });
464        self
465    }
466
467    /// Continue from a previous page's [`QueryPage::next_cursor`].
468    pub fn after(mut self, cursor: impl Into<String>) -> Self {
469        self.cursor = Some(cursor.into());
470        self
471    }
472
473    /// Compose a semantic clause over the page this query produces (experimental; see
474    /// `docs/reference/semantic-query.md`). The page is never enlarged or refilled.
475    pub fn semantic(mut self, clause: SemanticQueryClause) -> Self {
476        self.semantic = Some(clause);
477        self
478    }
479}
480
481/// One page of a [`BoundedQuery`].
482#[derive(Debug, Clone, PartialEq, Deserialize)]
483#[serde(rename_all = "camelCase")]
484pub struct QueryPage {
485    pub records: Vec<Value>,
486    #[serde(default)]
487    pub next_cursor: Option<String>,
488    pub exhausted: bool,
489    /// Present exactly when the query carried a semantic clause.
490    #[serde(default)]
491    pub semantic: Option<Value>,
492}
493
494impl Client {
495    /// The record `collection/id`, or `None` if it does not exist.
496    pub async fn get(&self, collection: &str, id: &str) -> Result<Option<Record>> {
497        match self
498            .send::<Record, ()>(Method::GET, &record_path(collection, id), None)
499            .await
500        {
501            Ok(record) => Ok(Some(record)),
502            Err(Error::Api(error)) if error.status == 404 => Ok(None),
503            Err(error) => Err(error),
504        }
505    }
506
507    /// Every record of `collection`. This reads the whole collection: on a request path use
508    /// [`Client::query`], which is bounded.
509    pub async fn list(&self, collection: &str) -> Result<Vec<Record>> {
510        self.send::<_, ()>(Method::GET, &collection_path(collection), None)
511            .await
512    }
513
514    /// Create a record only if `id` is free; refused with HTTP 409 if it exists. `value` must be a
515    /// JSON object. Without an id, the server assigns one.
516    pub async fn create(&self, collection: &str, id: Option<&str>, value: Value) -> Result<Record> {
517        let Value::Object(mut fields) = value else {
518            return Err(Error::InvalidRequest(
519                "a record must be a JSON object".into(),
520            ));
521        };
522        if let Some(id) = id {
523            fields.insert("id".into(), Value::String(id.to_owned()));
524        }
525        self.send(Method::POST, &collection_path(collection), Some(&fields))
526            .await
527    }
528
529    /// Merge `changes` into an existing record (unconditionally). Refused with HTTP 404 if the
530    /// record does not exist; use a [`Transaction`] or [`Client::compare_and_set`] to fence it.
531    pub async fn update(
532        &self,
533        collection: &str,
534        id: &str,
535        changes: Map<String, Value>,
536    ) -> Result<Record> {
537        self.send(Method::PATCH, &record_path(collection, id), Some(&changes))
538            .await
539    }
540
541    /// Delete an existing record. Refused with HTTP 404 if it does not exist.
542    pub async fn delete(&self, collection: &str, id: &str) -> Result<()> {
543        self.exchange::<()>(Method::DELETE, &record_path(collection, id), None)
544            .await
545            .map(drop)
546    }
547
548    /// Replace a record if it still has the version (and epoch and lease) you read.
549    pub async fn compare_and_set(
550        &self,
551        collection: &str,
552        id: &str,
553        request: &CasRequest,
554    ) -> Result<CasOutcome> {
555        let path = format!("{}/cas", record_path(collection, id));
556        let body: CasBody = match self.send(Method::POST, &path, Some(request)).await {
557            Ok(body) => body,
558            // A conflict is an answer, carried in the error body; anything else stays an error.
559            Err(Error::Api(error)) if error.status == 409 || error.status == 404 => {
560                if error.status == 404 && error.code == "NOT_FOUND" {
561                    return Ok(CasOutcome::NotFound);
562                }
563                match serde_json::from_value::<CasBody>(error.body.clone()) {
564                    Ok(body) if body.code.is_some() => body,
565                    _ => return Err(Error::Api(error)),
566                }
567            }
568            Err(error) => return Err(error),
569        };
570        if body.updated {
571            return Ok(CasOutcome::Updated {
572                current_version: body.current_version,
573                current_epoch: body.current_epoch,
574                item: body.item,
575            });
576        }
577        let code = body.code.unwrap_or_default();
578        let conflict = serde_json::from_value(Value::String(code.clone())).map_err(|_| {
579            Error::InvalidRequest(format!("unrecognized compare-and-set refusal {code:?}"))
580        })?;
581        Ok(CasOutcome::Conflict {
582            conflict,
583            current_version: body.current_version,
584            current_epoch: body.current_epoch,
585        })
586    }
587
588    /// Commit a transaction atomically. A refusal applied nothing; a failed precondition or fence
589    /// is reported in [`Error::precondition_failure`].
590    pub async fn commit(&self, transaction: &Transaction) -> Result<CommitReceipt> {
591        transaction.validate()?;
592        self.send(Method::POST, "/transactions", Some(transaction))
593            .await
594    }
595
596    /// Capture what the authority holds for `reads`, to fence a later transaction on it.
597    pub async fn read_basis(&self, reads: &ReadSet) -> Result<ReadBasis> {
598        self.send(Method::POST, "/transactions/read-basis", Some(reads))
599            .await
600    }
601
602    /// Run one page of a bounded query.
603    pub async fn query(&self, query: &BoundedQuery) -> Result<QueryPage> {
604        self.send(Method::POST, "/query", Some(query)).await
605    }
606
607    /// The authority's committed revision.
608    pub async fn revision(&self) -> Result<Revision> {
609        self.send::<_, ()>(Method::GET, "/revision", None).await
610    }
611
612    /// Committed changes as they happen, from `cursor` (a previous event's id) or from now.
613    pub async fn events(&self, cursor: Option<&str>) -> Result<EventStream> {
614        self.open_events("/events", cursor).await
615    }
616}
617
618fn collection_path(collection: &str) -> String {
619    format!("/collections/{}", encode(collection))
620}
621
622fn record_path(collection: &str, id: &str) -> String {
623    format!("/collections/{}/{}", encode(collection), encode(id))
624}
625
626#[cfg(test)]
627mod tests {
628    use super::*;
629    use serde_json::json;
630
631    #[test]
632    fn a_transaction_serializes_as_the_authority_reads_it() {
633        let transaction = Transaction::new("tx-1")
634            .operation(Operation::create("orders", "o-1", json!({ "n": 1 })))
635            .operation(Operation::put("orders", "o-2", json!({ "n": 2 })).expect_version(3))
636            .operation(Operation::delete("orders", "o-3"))
637            .precondition(Precondition::absent("locks", "l-1"));
638        assert_eq!(
639            serde_json::to_value(&transaction).unwrap(),
640            json!({
641                "transactionId": "tx-1",
642                "operations": [
643                    { "collection": "orders", "id": "o-1", "value": { "n": 1 }, "requireAbsent": true },
644                    { "collection": "orders", "id": "o-2", "value": { "n": 2 }, "expectedVersion": 3 },
645                    { "collection": "orders", "id": "o-3" }
646                ],
647                "preconditions": [{ "collection": "locks", "id": "l-1", "requireAbsent": true }]
648            })
649        );
650    }
651
652    #[test]
653    fn a_transaction_the_server_would_refuse_is_refused_here() {
654        assert!(Transaction::new("tx").validate().is_err());
655        assert!(Transaction::new("tx")
656            .operation(Operation::put("c", "i", json!([1])))
657            .validate()
658            .is_err());
659    }
660
661    #[test]
662    fn a_bounded_query_serializes_its_conditions_by_operator() {
663        let query = BoundedQuery::new("orders", 10)
664            .eq("status", json!("open"))
665            .gte("total", json!(5))
666            .order_by("id", Direction::Asc)
667            .after("cursor-1");
668        assert_eq!(
669            serde_json::to_value(&query).unwrap(),
670            json!({
671                "collection": "orders",
672                "where": [{ "field": "status", "eq": "open" }, { "field": "total", "gte": 5 }],
673                "orderBy": [{ "field": "id", "direction": "asc" }],
674                "limit": 10,
675                "cursor": "cursor-1"
676            })
677        );
678    }
679}
feltdb_client/authority.rs

feltdb_client/
authority.rs
