nodedb_lite/nodedb/
array.rs

1//! Public array-engine methods on `NodeDbLite`.
2//!
3//! The array engine is locked via the `tokio::sync::Mutex` held in `NodeDbLite`.
4
5use nodedb_array::query::slice::DimRange;
6use nodedb_array::schema::ArraySchema;
7use nodedb_array::tile::cell_payload::CellPayload;
8use nodedb_array::types::cell_value::value::CellValue;
9use nodedb_array::types::coord::value::CoordValue;
10#[cfg(not(target_arch = "wasm32"))]
11use nodedb_types::OPEN_UPPER;
12use nodedb_types::error::{NodeDbError, NodeDbResult};
13
14use crate::engine::array::engine::CellWrite;
15use crate::storage::engine::StorageEngine;
16
17use super::core::NodeDbLite;
18
19impl<S: StorageEngine> NodeDbLite<S> {
20    /// Create a new ND sparse array with the given schema.
21    ///
22    /// Returns an error if an array named `name` already exists.
23    pub async fn create_array(&self, name: &str, schema: ArraySchema) -> NodeDbResult<()> {
24        // Clone before the engine call to avoid a partial-move of `schema`.
25        let schema_for_crdt = schema.clone();
26
27        self.array_state
28            .lock()
29            .await
30            // This direct API takes no retention policy; one is only expressed
31            // through `CREATE ARRAY ... WITH (audit_retain_ms = ...)`, which
32            // reaches the engine through the SQL lowering.
33            .create_array(&self.storage, name, schema, None, None)
34            .await
35            .map_err(NodeDbError::storage)?;
36
37        // Register the schema CRDT so subsequent emit_* calls can find schema_hlc.
38        #[cfg(not(target_arch = "wasm32"))]
39        self.array_schemas
40            .put_schema(name, &schema_for_crdt)
41            .await
42            .map_err(NodeDbError::storage)?;
43        // On wasm, schema registration is skipped (sync not available).
44        #[cfg(target_arch = "wasm32")]
45        let _ = schema_for_crdt;
46
47        Ok(())
48    }
49
50    /// Write a cell into array `name` at `coord`.
51    ///
52    /// `system_from_ms` is the bitemporal system time (typically `now()`).
53    /// `valid_from_ms` / `valid_until_ms` are the valid-time bounds
54    /// (`OPEN_UPPER` = open-ended, no expiry).
55    pub async fn array_put_cell(
56        &self,
57        name: &str,
58        coord: Vec<CoordValue>,
59        attrs: Vec<CellValue>,
60        system_from_ms: i64,
61        valid_from_ms: i64,
62        valid_until_ms: i64,
63    ) -> NodeDbResult<()> {
64        // Clone before the engine call to keep copies for emit.
65        let coord_emit = coord.clone();
66        let attrs_emit = attrs.clone();
67
68        self.array_state
69            .lock()
70            .await
71            .put_cell(
72                &self.storage,
73                name,
74                CellWrite {
75                    coord,
76                    attrs,
77                    system_from_ms,
78                    valid_from_ms,
79                    valid_until_ms,
80                },
81            )
82            .await
83            .map_err(NodeDbError::storage)?;
84
85        // Emit op after engine succeeds (non-wasm only — sync not available on wasm).
86        #[cfg(not(target_arch = "wasm32"))]
87        if let Err(e) = self
88            .array_outbound
89            .emit_put(name, coord_emit, attrs_emit, valid_from_ms, valid_until_ms)
90            .await
91        {
92            tracing::error!(
93                array = name,
94                "array_put_cell: emit failed after engine write — op-log gap: {e}"
95            );
96            return Err(NodeDbError::storage(e));
97        }
98        #[cfg(target_arch = "wasm32")]
99        {
100            let _ = (coord_emit, attrs_emit, valid_from_ms, valid_until_ms);
101        }
102
103        Ok(())
104    }
105
106    /// Slice query: return all live cells whose coordinates fall within
107    /// `ranges` at or before `as_of_system_ms` (defaults to `i64::MAX` for
108    /// the current live snapshot).
109    pub async fn array_slice(
110        &self,
111        name: &str,
112        ranges: Vec<Option<DimRange>>,
113        as_of_system_ms: Option<i64>,
114    ) -> NodeDbResult<Vec<CellPayload>> {
115        let sys = as_of_system_ms.unwrap_or(i64::MAX);
116        self.array_state
117            .lock()
118            .await
119            .slice(&self.storage, name, ranges, sys)
120            .await
121            .map_err(NodeDbError::storage)
122    }
123
124    /// Read the most recent live payload for `coord` at or before
125    /// `as_of_system_ms`. Returns `None` if not found, tombstoned, or erased.
126    pub async fn array_read_coord(
127        &self,
128        name: &str,
129        coord: &[CoordValue],
130        as_of_system_ms: Option<i64>,
131    ) -> NodeDbResult<Option<CellPayload>> {
132        let sys = as_of_system_ms.unwrap_or(i64::MAX);
133        self.array_state
134            .lock()
135            .await
136            .read_coord(&self.storage, name, coord, sys)
137            .await
138            .map_err(NodeDbError::storage)
139    }
140
141    /// Soft-delete a cell by writing a tombstone version at `system_from_ms`.
142    ///
143    /// The cell is still visible AS-OF any system time < `system_from_ms`.
144    pub async fn array_delete_cell(
145        &self,
146        name: &str,
147        coord: Vec<CoordValue>,
148        system_from_ms: i64,
149    ) -> NodeDbResult<()> {
150        // Clone before the engine call for emit.
151        let coord_emit = coord.clone();
152
153        self.array_state
154            .lock()
155            .await
156            .delete_cell(name, coord, system_from_ms)
157            .map_err(NodeDbError::storage)?;
158
159        // valid_from_ms / valid_until_ms: the current API does not expose
160        // valid-time arguments on delete. Defaults of 0 / OPEN_UPPER are used
161        // here. A future API revision will widen this to carry the full bitemporal envelope.
162        #[cfg(not(target_arch = "wasm32"))]
163        if let Err(e) = self
164            .array_outbound
165            .emit_delete(name, coord_emit, 0, OPEN_UPPER)
166            .await
167        {
168            tracing::error!(
169                array = name,
170                "array_delete_cell: emit failed after engine write — op-log gap: {e}"
171            );
172            return Err(NodeDbError::storage(e));
173        }
174        #[cfg(target_arch = "wasm32")]
175        let _ = coord_emit;
176
177        Ok(())
178    }
179
180    /// GDPR erasure: write the `0xFE` sentinel and flush to disk.
181    ///
182    /// After this call `array_read_coord` returns `None` for the erased
183    /// coordinate at any `system_as_of >= system_from_ms`, and the raw
184    /// payload bytes are not present on disk.
185    pub async fn array_gdpr_erase_cell(
186        &self,
187        name: &str,
188        coord: Vec<CoordValue>,
189        system_from_ms: i64,
190    ) -> NodeDbResult<()> {
191        // Clone before the engine call for emit.
192        let coord_emit = coord.clone();
193
194        self.array_state
195            .lock()
196            .await
197            .gdpr_erase_cell(&self.storage, name, coord, system_from_ms)
198            .await
199            .map_err(NodeDbError::storage)?;
200
201        // valid_from_ms / valid_until_ms: same defaulting as array_delete_cell.
202        // A future API revision will widen this to carry the full bitemporal envelope.
203        #[cfg(not(target_arch = "wasm32"))]
204        if let Err(e) = self
205            .array_outbound
206            .emit_erase(name, coord_emit, 0, OPEN_UPPER)
207            .await
208        {
209            tracing::error!(
210                array = name,
211                "array_gdpr_erase_cell: emit failed after engine write — op-log gap: {e}"
212            );
213            return Err(NodeDbError::storage(e));
214        }
215        #[cfg(target_arch = "wasm32")]
216        let _ = coord_emit;
217
218        Ok(())
219    }
220
221    /// Flush any pending memtable data for `name` to a durable segment.
222    pub async fn array_flush(&self, name: &str) -> NodeDbResult<()> {
223        self.array_state
224            .lock()
225            .await
226            .flush(&self.storage, name)
227            .await
228            .map_err(NodeDbError::storage)
229    }
230
231    /// Return the current schema HLC for `name` from the local schema registry.
232    ///
233    /// Used by tests that need to construct `ArrayOpHeader::schema_hlc` values
234    /// matching the locally registered schema.
235    #[cfg(not(target_arch = "wasm32"))]
236    pub fn array_schema_hlc(&self, name: &str) -> Option<nodedb_array::sync::hlc::Hlc> {
237        self.array_schemas.schema_hlc(name)
238    }
239}
nodedb_lite/nodedb/array.rs

nodedb_lite/nodedb/
array.rs
