hidane_core/store/
mod.rs

1//! The storage boundary (ADR 0002).
2//!
3//! Everything above this module (gRPC, REST, Listen, transactions) talks to [`Store`]; the
4//! engine behind it can change without touching them. [`MemoryStore`] is the engine for now.
5//!
6//! The model:
7//!
8//! - Documents are **versioned by commit time** (microseconds, unique and increasing per
9//!   database). A read names a [`ReadTime`] and sees the newest version at or before it, which
10//!   gives consistent snapshots for queries, transactions and Listen.
11//! - A commit runs a closure against a [`WriteBatch`]: the commit time is fixed before the
12//!   closure runs (server timestamps need it) and reads inside the batch see the batch's own
13//!   writes. The closure's error aborts the commit with nothing applied.
14//! - A commit returns every changed document with its old and new version, so change
15//!   notification costs O(changed documents), never O(stored documents).
16
17mod memory;
18
19use std::{fmt, ops::ControlFlow, sync::Arc};
20
21use hidane_proto::google::firestore::v1::{ArrayValue, MapValue, Value, value::ValueType};
22use hidane_proto::ordered::Fields;
23use prost::Message as _;
24use prost_types::Timestamp;
25
26pub use memory::MemoryStore;
27
28use crate::path::ResourcePath;
29
30/// A point in a database's history, in microseconds since the Unix epoch. Commit times are
31/// read times; reading at a commit time sees that commit.
32#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
33pub struct ReadTime(pub i64);
34
35impl ReadTime {
36    pub const MAX: Self = Self(i64::MAX);
37
38    /// Truncates to microseconds, as Firestore does.
39    pub fn from_timestamp(ts: &Timestamp) -> Self {
40        Self(ts.seconds * 1_000_000 + i64::from(ts.nanos) / 1_000)
41    }
42
43    pub fn to_timestamp(self) -> Timestamp {
44        #[allow(clippy::cast_possible_truncation)]
45        Timestamp {
46            seconds: self.0.div_euclid(1_000_000),
47            nanos: (self.0.rem_euclid(1_000_000) * 1_000) as i32,
48        }
49    }
50}
51
52/// A stored document version.
53///
54/// Fields are kept protobuf-encoded: a document then costs about its wire size instead of the
55/// decoded tree (a two-field document is ~30 bytes encoded, over 600 decoded, because every
56/// map node reserves room for eleven entries). Readers decode with [`StoredDocument::fields`].
57#[derive(Debug, Clone, PartialEq)]
58pub struct StoredDocument {
59    /// Shared by every version of the same document.
60    pub path: Arc<ResourcePath>,
61    encoded_fields: Box<[u8]>,
62    pub create_time: Timestamp,
63    pub update_time: Timestamp,
64}
65
66impl StoredDocument {
67    pub fn new(
68        path: Arc<ResourcePath>,
69        fields: Fields,
70        create_time: Timestamp,
71        update_time: Timestamp,
72    ) -> Self {
73        Self {
74            path,
75            encoded_fields: MapValue { fields }.encode_to_vec().into_boxed_slice(),
76            create_time,
77            update_time,
78        }
79    }
80
81    /// The document's fields, decoded.
82    pub fn fields(&self) -> Fields {
83        MapValue::decode(&*self.encoded_fields)
84            .expect("fields were encoded by StoredDocument::new")
85            .fields
86    }
87
88    /// Size of the stored field encoding, in bytes.
89    pub fn encoded_len(&self) -> usize {
90        self.encoded_fields.len()
91    }
92
93    /// Whether the document holds exactly `fields`, bit for bit: unlike `==` on the decoded
94    /// fields, a NaN equals itself and `-0.0` differs from `0.0`. The order of the fields does
95    /// not matter.
96    pub fn has_fields(&self, fields: &Fields) -> bool {
97        let canonical = |fields: &Fields| {
98            MapValue {
99                fields: canonical(fields),
100            }
101            .encode_to_vec()
102        };
103        fields.len() == self.fields().len() && canonical(fields) == canonical(&self.fields())
104    }
105}
106
107/// `fields` with every map, nested ones included, in UTF-8 key order.
108fn canonical(fields: &Fields) -> Fields {
109    fields
110        .sorted()
111        .into_iter()
112        .map(|(key, value)| (key.clone(), canonical_value(value)))
113        .collect()
114}
115
116fn canonical_value(value: &Value) -> Value {
117    let value_type = match &value.value_type {
118        Some(ValueType::MapValue(map)) => ValueType::MapValue(MapValue {
119            fields: canonical(&map.fields),
120        }),
121        Some(ValueType::ArrayValue(array)) => ValueType::ArrayValue(ArrayValue {
122            values: array.values.iter().map(canonical_value).collect(),
123        }),
124        _ => return value.clone(),
125    };
126    Value {
127        value_type: Some(value_type),
128    }
129}
130
131/// One document touched by a commit.
132#[derive(Debug, Clone)]
133pub struct Change {
134    pub path: Arc<ResourcePath>,
135    pub before: Option<Arc<StoredDocument>>,
136    pub after: Option<Arc<StoredDocument>>,
137}
138
139/// The result of a successful commit.
140#[derive(Debug, Clone)]
141pub struct Commit {
142    pub commit_time: ReadTime,
143    pub changes: Vec<Change>,
144}
145
146/// Why a commit was refused. Maps one to one onto gRPC status codes.
147#[derive(Debug, Clone, PartialEq, Eq)]
148pub enum StoreError {
149    InvalidArgument(String),
150    NotFound(String),
151    AlreadyExists(String),
152    FailedPrecondition(String),
153    Aborted(String),
154}
155
156impl fmt::Display for StoreError {
157    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
158        match self {
159            Self::InvalidArgument(m)
160            | Self::NotFound(m)
161            | Self::AlreadyExists(m)
162            | Self::FailedPrecondition(m)
163            | Self::Aborted(m) => f.write_str(m),
164        }
165    }
166}
167
168impl std::error::Error for StoreError {}
169
170/// The writes of one commit, with read-your-writes visibility.
171pub trait WriteBatch {
172    /// The time this commit will be stamped with (and the documents' `update_time`).
173    fn commit_time(&self) -> ReadTime;
174
175    /// The document as it will be if the commit succeeds now: the batch's own writes first,
176    /// then the latest committed version.
177    fn get(&self, path: &ResourcePath) -> Option<Arc<StoredDocument>>;
178
179    /// Replaces the document's fields. Keeps `create_time` if the document exists.
180    fn set(&mut self, path: &ResourcePath, fields: Fields);
181
182    /// Deletes the document. Deleting a missing document is not an error.
183    fn delete(&mut self, path: &ResourcePath);
184}
185
186/// Called for each document of a scan, in `__name__` order; return `Break` to stop early.
187pub type Visit<'a> = dyn FnMut(&Arc<StoredDocument>) -> ControlFlow<()> + 'a;
188
189/// An entry of a collection listing.
190#[derive(Debug)]
191pub enum ListItem<'a> {
192    Document(&'a Arc<StoredDocument>),
193    /// No document exists at this path, but documents exist below it (in subcollections).
194    Missing(&'a ResourcePath),
195}
196
197/// Options of [`Store::list_collection`].
198#[derive(Debug, Default, Clone, Copy)]
199pub struct ListOptions<'a> {
200    /// Resume after this document path (exclusive), for paging.
201    pub after: Option<&'a ResourcePath>,
202    /// Also report missing documents that have descendants (`ListDocuments.show_missing`).
203    pub include_missing: bool,
204}
205
206/// The storage engine. `database` is the full database name,
207/// `projects/{project}/databases/{database}`; databases are created on first write.
208pub trait Store: Send + Sync {
209    /// A read time that sees every commit made so far.
210    fn latest_read_time(&self, database: &str) -> ReadTime;
211
212    /// The oldest read time still served; older versions may have been dropped.
213    fn earliest_read_time(&self, database: &str) -> ReadTime;
214
215    fn get(&self, database: &str, path: &ResourcePath, at: ReadTime)
216    -> Option<Arc<StoredDocument>>;
217
218    /// The documents directly in `collection` (not in its subcollections).
219    fn scan_collection(
220        &self,
221        database: &str,
222        collection: &ResourcePath,
223        at: ReadTime,
224        visit: &mut Visit<'_>,
225    ) {
226        self.list_collection(
227            database,
228            collection,
229            at,
230            ListOptions::default(),
231            &mut |item| match item {
232                ListItem::Document(doc) => visit(doc),
233                ListItem::Missing(_) => ControlFlow::Continue(()),
234            },
235        );
236    }
237
238    /// Like [`Store::scan_collection`], with paging and missing documents.
239    fn list_collection(
240        &self,
241        database: &str,
242        collection: &ResourcePath,
243        at: ReadTime,
244        options: ListOptions<'_>,
245        visit: &mut dyn FnMut(ListItem<'_>) -> ControlFlow<()>,
246    );
247
248    /// Every document under `parent` (a document path, or the root) whose collection ID is
249    /// `collection_id`, at any depth, ordered by full path.
250    fn scan_collection_group(
251        &self,
252        database: &str,
253        parent: &ResourcePath,
254        collection_id: &str,
255        at: ReadTime,
256        visit: &mut Visit<'_>,
257    );
258
259    /// Every document below `parent` (a document path, or the root), at any depth, ordered by
260    /// path. `parent` itself is not included.
261    fn scan_descendants(
262        &self,
263        database: &str,
264        parent: &ResourcePath,
265        at: ReadTime,
266        visit: &mut Visit<'_>,
267    );
268
269    /// IDs of the collections directly under `parent` that contain at least one document at
270    /// any depth, in UTF-8 order.
271    fn list_collection_ids(
272        &self,
273        database: &str,
274        parent: &ResourcePath,
275        at: ReadTime,
276    ) -> Vec<String>;
277
278    /// Runs `write` as one atomic commit.
279    fn commit(
280        &self,
281        database: &str,
282        write: &mut dyn FnMut(&mut dyn WriteBatch) -> Result<(), StoreError>,
283    ) -> Result<Commit, StoreError>;
284
285    /// Drops every document of `database` and every older version, at a time after its last
286    /// commit, which it returns: later commits are later still, so listeners can order the
287    /// clear among them.
288    fn clear_database(&self, database: &str) -> ReadTime;
289
290    /// The databases that hold or held documents.
291    fn database_names(&self) -> Vec<String>;
292
293    /// Drops every database.
294    fn clear(&self);
295}
hidane_core/store/mod.rs

hidane_core/store/
mod.rs
