nodedb_lite/nodedb/collection/
ddl.rs

1//! Collection DDL: create, drop, and list collections with metadata.
2
3use nodedb_types::error::{NodeDbError, NodeDbResult};
4
5use super::super::{LockExt, NodeDbLite};
6use crate::storage::engine::StorageEngine;
7
8/// Collection metadata stored in the KV store.
9#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
10pub struct CollectionMeta {
11    pub name: String,
12    pub collection_type: String,
13    pub created_at_ms: u64,
14    pub fields: Vec<(String, String)>,
15    /// Optional JSON-serialized engine config (e.g., `KvConfig` for KV collections,
16    /// `StrictSchema` for strict collections). Empty for schemaless document collections.
17    #[serde(default)]
18    pub config_json: Option<String>,
19    /// Optional JSON-serialized full `CollectionDescriptor` (from
20    /// `nodedb_types::sync::wire::CollectionDescriptor`). Set for collections
21    /// materialized from an inbound sync schema announcement so the SQL catalog
22    /// can surface the real engine, bitemporal flag, and column schema, and so a
23    /// future emit path can reconstruct the descriptor losslessly. `None` for
24    /// locally-created collections.
25    #[serde(default)]
26    pub descriptor_json: Option<String>,
27    /// Whether the collection tracks system-time + valid-time versions.
28    #[serde(default)]
29    pub bitemporal: bool,
30    /// Whether the collection was declared with `WITH (crdt=true)`, i.e. its
31    /// DML is routed through the CRDT (Loro) path rather than the plain
32    /// document path. Announced to Origin in the collection descriptor so the
33    /// peer applies the same routing. A declared property, not an inferred one:
34    /// Lite's schemaless store happens to be Loro-backed, but that alone does
35    /// NOT make a collection CRDT-declared.
36    #[serde(default)]
37    pub crdt: bool,
38}
39
40impl<S: StorageEngine> NodeDbLite<S> {
41    /// Create a collection with optional schema.
42    ///
43    /// If the collection already exists, returns Ok (idempotent).
44    /// Schema is advisory — documents are schemaless by default.
45    pub async fn create_collection(
46        &self,
47        name: &str,
48        fields: &[(String, String)],
49    ) -> NodeDbResult<()> {
50        let guard = self.fts_state.admit_mutation().await;
51        let result = self
52            .create_collection_admitted(guard.permit(), name, fields)
53            .await;
54        guard.finish(result)
55    }
56
57    pub(crate) async fn create_collection_admitted(
58        &self,
59        _permit: &crate::engine::fts::coordinator::TextMutationPermit,
60        name: &str,
61        fields: &[(String, String)],
62    ) -> NodeDbResult<()> {
63        let meta = CollectionMeta {
64            name: name.to_string(),
65            collection_type: "document".to_string(),
66            created_at_ms: crate::runtime::now_millis(),
67            fields: fields.to_vec(),
68            config_json: None,
69            descriptor_json: None,
70            bitemporal: false,
71            crdt: false,
72        };
73        let key = format!("collection:{name}");
74        let bytes = sonic_rs::to_vec(&meta).map_err(|e| NodeDbError::storage(e.to_string()))?;
75        self.storage
76            .put(nodedb_types::Namespace::Meta, key.as_bytes(), &bytes)
77            .await?;
78        Ok(())
79    }
80
81    /// Create a KV collection with typed schema and optional TTL.
82    ///
83    /// Stores the `KvConfig` as JSON in the collection metadata so that the
84    /// KV engine can reconstruct the schema on startup.
85    pub async fn create_kv_collection(
86        &self,
87        name: &str,
88        config: &nodedb_types::KvConfig,
89    ) -> NodeDbResult<()> {
90        let guard = self.fts_state.admit_mutation().await;
91        let result = self
92            .create_kv_collection_admitted(guard.permit(), name, config)
93            .await;
94        guard.finish(result)
95    }
96
97    pub(crate) async fn create_kv_collection_admitted(
98        &self,
99        _permit: &crate::engine::fts::coordinator::TextMutationPermit,
100        name: &str,
101        config: &nodedb_types::KvConfig,
102    ) -> NodeDbResult<()> {
103        let fields: Vec<(String, String)> = config
104            .schema
105            .columns
106            .iter()
107            .map(|c| (c.name.clone(), c.column_type.to_string()))
108            .collect();
109
110        let config_json =
111            sonic_rs::to_string(config).map_err(|e| NodeDbError::storage(e.to_string()))?;
112
113        let meta = CollectionMeta {
114            name: name.to_string(),
115            collection_type: "kv".to_string(),
116            created_at_ms: crate::runtime::now_millis(),
117            fields,
118            config_json: Some(config_json),
119            descriptor_json: None,
120            bitemporal: false,
121            crdt: false,
122        };
123        let key = format!("collection:{name}");
124        let bytes = sonic_rs::to_vec(&meta).map_err(|e| NodeDbError::storage(e.to_string()))?;
125        self.storage
126            .put(nodedb_types::Namespace::Meta, key.as_bytes(), &bytes)
127            .await?;
128        Ok(())
129    }
130
131    /// Synthesize a base document `CollectionMeta` for an implicit CRDT-only
132    /// collection — one that exists in CRDT state (created by a document upsert,
133    /// a vector insert, or an FTS write, with no explicit `create_collection`)
134    /// and therefore has no persisted `collection:` meta. Returns `None` if the
135    /// collection is not present in CRDT state (or is an internal `__` name).
136    ///
137    /// Mirrors the synthetic entry [`list_collections`](Self::list_collections)
138    /// already produces for implicit collections, so the outbound sync announce
139    /// path can register such collections on Origin with a base document
140    /// descriptor before their overlay (vector/FTS) or document data arrives.
141    /// Equivalent to the meta an explicit `create_collection` would persist
142    /// (`collection_type = "document"`, no `descriptor_json`).
143    #[cfg(not(target_arch = "wasm32"))]
144    pub(crate) fn implicit_collection_meta(&self, name: &str) -> Option<CollectionMeta> {
145        if name.starts_with("__") {
146            return None;
147        }
148        let crdt = self.crdt.lock_or_recover();
149        if !crdt.collection_names().iter().any(|n| n == name) {
150            return None;
151        }
152        Some(CollectionMeta {
153            name: name.to_string(),
154            collection_type: "document".to_string(),
155            created_at_ms: 0,
156            fields: Vec::new(),
157            config_json: None,
158            descriptor_json: None,
159            bitemporal: false,
160            crdt: false,
161        })
162    }
163
164    /// Drop a collection — deletes all documents and metadata.
165    ///
166    /// Uses `clear_collection` for single-batch deletion (one Loro delta
167    /// for all document removals). Also removes the secondary and text
168    /// indexes.
169    pub async fn drop_collection(&self, name: &str) -> NodeDbResult<()> {
170        let guard = self.fts_state.admit_mutation().await;
171        let result = async {
172            self.fts_state
173                .persist_collection_tombstone(&*self.storage, name, guard.permit())
174                .await
175                .map_err(NodeDbError::from)?;
176            // Batch-delete all documents in one delta.
177            {
178                let mut crdt = self.crdt.lock_or_recover();
179                crdt.clear_collection(name).map_err(NodeDbError::storage)?;
180            }
181            crate::query::kv_ops::indexes::kv_drop_collection(&self.query_engine, name)
182                .await
183                .map_err(NodeDbError::storage)?;
184
185            // Remove text index for this collection.
186            {
187                let mut fts = self.fts_state.manager.lock_or_recover();
188                fts.drop_collection(name);
189            }
190
191            // Delete collection metadata from the KV store.
192            let key = format!("collection:{name}");
193            self.storage
194                .delete(nodedb_types::Namespace::Meta, key.as_bytes())
195                .await?;
196            Ok(())
197        }
198        .await;
199        guard.finish(result)
200    }
201
202    /// Whether `name` is a collection any engine or the catalog knows: a
203    /// CRDT (schemaless) collection, a strict or columnar collection, or one
204    /// with persisted metadata.
205    pub(crate) async fn collection_known(&self, name: &str) -> NodeDbResult<bool> {
206        if self
207            .crdt
208            .lock_or_recover()
209            .collection_names()
210            .iter()
211            .any(|n| n == name)
212            || self.strict.schema(name).is_some()
213            || self.columnar.schema(name).is_some()
214        {
215            return Ok(true);
216        }
217        let key = format!("collection:{name}");
218        Ok(self
219            .storage
220            .get(nodedb_types::Namespace::Meta, key.as_bytes())
221            .await?
222            .is_some())
223    }
224
225    /// List all collections.
226    pub async fn list_collections(&self) -> NodeDbResult<Vec<CollectionMeta>> {
227        let pairs = self
228            .storage
229            .scan_prefix(nodedb_types::Namespace::Meta, b"collection:")
230            .await?;
231        let mut result = Vec::new();
232        for (_, value) in &pairs {
233            if let Ok(meta) = sonic_rs::from_slice::<CollectionMeta>(value) {
234                result.push(meta);
235            }
236        }
237        // Also include implicit collections (from CRDT state without explicit DDL).
238        let crdt = self.crdt.lock_or_recover();
239        let crdt_names = crdt.collection_names();
240        let explicit: std::collections::HashSet<String> =
241            result.iter().map(|m| m.name.clone()).collect();
242        for name in crdt_names {
243            if !name.starts_with("__") && !explicit.contains(&name) {
244                result.push(CollectionMeta {
245                    name,
246                    collection_type: "document".to_string(),
247                    created_at_ms: 0,
248                    fields: Vec::new(),
249                    config_json: None,
250                    descriptor_json: None,
251                    bitemporal: false,
252                    crdt: false,
253                });
254            }
255        }
256        Ok(result)
257    }
258}
259
260/// Load all explicitly-persisted collection metadata as a name→meta map.
261///
262/// Unlike [`NodeDbLite::list_collections`], this does NOT merge implicit CRDT
263/// collections — it returns only the metas durably written under the
264/// `collection:` prefix (via `create_collection`, `create_kv_collection`, or
265/// inbound schema sync). The SQL catalog uses this snapshot to surface the real
266/// engine, bitemporal flag, and columns for DDL/synced collections, while
267/// implicit CRDT-only collections still fall through to engine-based detection.
268/// Free-function form so callers that hold only an `&S` (e.g. the SQL query
269/// engine building its catalog) can reuse the same scan-and-decode logic.
270pub(crate) async fn load_persisted_collection_metas<S: StorageEngine>(
271    storage: &S,
272) -> NodeDbResult<std::collections::HashMap<String, CollectionMeta>> {
273    let pairs = storage
274        .scan_prefix(nodedb_types::Namespace::Meta, b"collection:")
275        .await?;
276    let mut map = std::collections::HashMap::with_capacity(pairs.len());
277    for (_, value) in &pairs {
278        if let Ok(meta) = sonic_rs::from_slice::<CollectionMeta>(value) {
279            map.insert(meta.name.clone(), meta);
280        }
281    }
282    Ok(map)
283}
nodedb_lite/nodedb/collection/ddl.rs

nodedb_lite/nodedb/collection/
ddl.rs
