feat(catalog): add StoredCollection <-> CollectionDescriptor conversion

Provides the bidirectional mapping between the catalog's StoredCollection
record and the engine-agnostic sync wire descriptor, so CREATE COLLECTION
and CRDT sync produce and consume the same descriptor shape.

Author: farhan-syah

nodedb/src/control/security/catalog/collection_descriptor_convert.rs

@@ -0,0 +1,202 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Bidirectional conversions between [`StoredCollection`] (catalog record)
+//! and [`CollectionDescriptor`] (engine-agnostic sync wire type), so
+//! `CREATE COLLECTION` and CRDT sync produce the same descriptor shape.
+//!
+//! `From<&StoredCollection> for CollectionDescriptor` is the emit side: it
+//! reads only the 11 fields the descriptor carries, dropping ownership,
+//! timestamps, and enforcement/index/constraint state that never travels
+//! over sync. [`stored_from_descriptor`] is the receive side: it starts
+//! from [`StoredCollection::new`] (which sets sane enforcement defaults
+//! and a fresh `created_at`) and overlays only the descriptor-carried
+//! fields.
+
+use nodedb_types::sync::wire::CollectionDescriptor;
+
+use super::collection::StoredCollection;
+
+impl From<&StoredCollection> for CollectionDescriptor {
+    fn from(stored: &StoredCollection) -> Self {
+        Self {
+            tenant_id: stored.tenant_id,
+            database_id: stored.database_id,
+            name: stored.name.clone(),
+            collection_type: stored.collection_type.clone(),
+            bitemporal: stored.bitemporal,
+            fields: stored.fields.clone(),
+            primary: stored.primary,
+            vector_primary: stored.vector_primary.clone(),
+            partition_strategy: stored.partition_strategy.clone(),
+            declared_primary_key: stored.declared_primary_key.clone(),
+            descriptor_version: stored.descriptor_version,
+        }
+    }
+}
+
+/// Materialize a [`StoredCollection`] from a synced [`CollectionDescriptor`].
+///
+/// `owner` is assigned to the receiving peer's identity — the descriptor
+/// carries no owner, since ownership is a local catalog concept, not a
+/// sync-wire one. All fields not carried by the descriptor (field_defs,
+/// event_defs, indexes, constraints, `is_active`, etc.) are left at the
+/// [`StoredCollection::new`] defaults.
+// Consumed by the sync receive handler that materializes announced
+// collections into the catalog; only the round-trip tests call it today.
+#[allow(dead_code)]
+pub(crate) fn stored_from_descriptor(
+    descriptor: &CollectionDescriptor,
+    owner: &str,
+) -> StoredCollection {
+    let mut stored = StoredCollection::new(descriptor.tenant_id, &descriptor.name, owner);
+    stored.database_id = descriptor.database_id;
+    stored.collection_type = descriptor.collection_type.clone();
+    stored.bitemporal = descriptor.bitemporal;
+    stored.fields = descriptor.fields.clone();
+    stored.primary = descriptor.primary;
+    stored.vector_primary = descriptor.vector_primary.clone();
+    stored.partition_strategy = descriptor.partition_strategy.clone();
+    stored.declared_primary_key = descriptor.declared_primary_key.clone();
+    stored.descriptor_version = descriptor.descriptor_version;
+    stored
+}
+
+#[cfg(test)]
+mod tests {
+    use nodedb_types::CollectionType;
+    use nodedb_types::collection_config::{PartitionStrategy, PrimaryEngine, VectorPrimaryConfig};
+    use nodedb_types::columnar::{ColumnDef, ColumnType, StrictSchema};
+    use nodedb_types::kv::KvConfig;
+
+    use super::*;
+
+    fn assert_mapped_fields_match(stored: &StoredCollection, back: &StoredCollection) {
+        assert_eq!(back.name, stored.name);
+        assert_eq!(back.tenant_id, stored.tenant_id);
+        assert_eq!(back.database_id, stored.database_id);
+        assert_eq!(back.collection_type, stored.collection_type);
+        assert_eq!(back.bitemporal, stored.bitemporal);
+        assert_eq!(back.fields, stored.fields);
+        assert_eq!(back.primary, stored.primary);
+        assert_eq!(back.vector_primary, stored.vector_primary);
+        assert_eq!(back.partition_strategy, stored.partition_strategy);
+        assert_eq!(back.declared_primary_key, stored.declared_primary_key);
+        assert_eq!(back.descriptor_version, stored.descriptor_version);
+    }
+
+    fn base_stored(name: &str, collection_type: CollectionType) -> StoredCollection {
+        let mut stored = StoredCollection::new(7, name, "alice");
+        stored.partition_strategy =
+            PartitionStrategy::default_for_collection_type(&collection_type);
+        stored.collection_type = collection_type;
+        stored.bitemporal = true;
+        stored.declared_primary_key = Some("id".to_string());
+        stored.descriptor_version = 5;
+        stored
+    }
+
+    #[test]
+    fn document_schemaless_round_trips() {
+        let stored = base_stored("users", CollectionType::document());
+        let descriptor = CollectionDescriptor::from(&stored);
+        let back = stored_from_descriptor(&descriptor, "sync");
+        assert_mapped_fields_match(&stored, &back);
+    }
+
+    #[test]
+    fn document_strict_round_trips() {
+        let schema = StrictSchema {
+            columns: vec![
+                ColumnDef::required("name", ColumnType::String),
+                ColumnDef::nullable("bio", ColumnType::String),
+            ],
+            version: 1,
+            dropped_columns: Vec::new(),
+            bitemporal: false,
+        };
+        let stored = base_stored("people", CollectionType::strict(schema));
+        let descriptor = CollectionDescriptor::from(&stored);
+        let back = stored_from_descriptor(&descriptor, "sync");
+        assert_mapped_fields_match(&stored, &back);
+    }
+
+    #[test]
+    fn key_value_round_trips() {
+        let schema = StrictSchema {
+            columns: vec![ColumnDef::required("id", ColumnType::Int64).with_primary_key()],
+            version: 1,
+            dropped_columns: Vec::new(),
+            bitemporal: false,
+        };
+        let stored = base_stored("sessions", CollectionType::kv(schema));
+        let descriptor = CollectionDescriptor::from(&stored);
+        let back = stored_from_descriptor(&descriptor, "sync");
+        assert_mapped_fields_match(&stored, &back);
+        // Sanity: the KvConfig payload itself survived intact.
+        match (&stored.collection_type, &back.collection_type) {
+            (CollectionType::KeyValue(a), CollectionType::KeyValue(b)) => {
+                let a: &KvConfig = a;
+                let b: &KvConfig = b;
+                assert_eq!(a.schema, b.schema);
+            }
+            _ => panic!("expected KeyValue collection type"),
+        }
+    }
+
+    #[test]
+    fn columnar_plain_round_trips() {
+        let stored = base_stored("events", CollectionType::columnar());
+        let descriptor = CollectionDescriptor::from(&stored);
+        let back = stored_from_descriptor(&descriptor, "sync");
+        assert_mapped_fields_match(&stored, &back);
+    }
+
+    #[test]
+    fn timeseries_round_trips() {
+        let stored = base_stored("metrics", CollectionType::timeseries("ts", "1m"));
+        let descriptor = CollectionDescriptor::from(&stored);
+        let back = stored_from_descriptor(&descriptor, "sync");
+        assert_mapped_fields_match(&stored, &back);
+    }
+
+    #[test]
+    fn spatial_round_trips() {
+        let stored = base_stored("places", CollectionType::spatial("geom"));
+        let descriptor = CollectionDescriptor::from(&stored);
+        let back = stored_from_descriptor(&descriptor, "sync");
+        assert_mapped_fields_match(&stored, &back);
+    }
+
+    #[test]
+    fn vector_primary_round_trips() {
+        let mut stored = base_stored("embeddings", CollectionType::document());
+        stored.primary = PrimaryEngine::Vector;
+        stored.vector_primary = Some(VectorPrimaryConfig {
+            vector_field: "emb".to_string(),
+            dim: 768,
+            ..VectorPrimaryConfig::default()
+        });
+        let descriptor = CollectionDescriptor::from(&stored);
+        let back = stored_from_descriptor(&descriptor, "sync");
+        assert_mapped_fields_match(&stored, &back);
+    }
+
+    #[test]
+    fn bitemporal_flag_round_trips() {
+        let mut stored = base_stored("audit_log", CollectionType::document());
+        stored.bitemporal = true;
+        let descriptor = CollectionDescriptor::from(&stored);
+        assert!(descriptor.bitemporal);
+        let back = stored_from_descriptor(&descriptor, "sync");
+        assert!(back.bitemporal);
+    }
+
+    #[test]
+    fn owner_assigned_on_receive() {
+        let stored = base_stored("owned", CollectionType::document());
+        assert_eq!(stored.owner, "alice");
+        let descriptor = CollectionDescriptor::from(&stored);
+        let back = stored_from_descriptor(&descriptor, "sync");
+        assert_eq!(back.owner, "sync");
+    }
+}

nodedb/src/control/security/catalog/mod.rs

@@ -13,6 +13,7 @@ pub mod checkpoints;
 pub mod clone_catalog;
 pub mod collection;
 pub mod collection_constraints;
+pub mod collection_descriptor_convert;
 pub mod collections;
 pub mod column_stats;
 pub mod constraint_translate;