grafeo_engine/catalog/

mod.rs

//! Schema metadata - what labels, properties, and indexes exist.
//!
//! The catalog is the "dictionary" of your database. When you write `(:Person)`,
//! the catalog maps "Person" to an internal LabelId. This indirection keeps
//! storage compact while names stay readable.
//!
//! | What it tracks | Why it matters |
//! | -------------- | -------------- |
//! | Labels | Maps "Person" → LabelId for efficient storage |
//! | Property keys | Maps "name" → PropertyKeyId |
//! | Edge types | Maps "KNOWS" → EdgeTypeId |
//! | Indexes | Which properties are indexed for fast lookups |

use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};

use parking_lot::RwLock;

use grafeo_common::types::{EdgeTypeId, IndexId, LabelId, PropertyKeyId, Value};

/// The database's schema dictionary - maps names to compact internal IDs.
///
/// You rarely interact with this directly. The query processor uses it to
/// resolve names like "Person" and "name" to internal IDs.
pub struct Catalog {
    /// Label name-to-ID mappings.
    labels: LabelCatalog,
    /// Property key name-to-ID mappings.
    property_keys: PropertyCatalog,
    /// Edge type name-to-ID mappings.
    edge_types: EdgeTypeCatalog,
    /// Index definitions.
    indexes: IndexCatalog,
    /// Optional schema constraints.
    schema: Option<SchemaCatalog>,
}

impl Catalog {
    /// Creates a new empty catalog with schema support enabled.
    #[must_use]
    pub fn new() -> Self {
        Self {
            labels: LabelCatalog::new(),
            property_keys: PropertyCatalog::new(),
            edge_types: EdgeTypeCatalog::new(),
            indexes: IndexCatalog::new(),
            schema: Some(SchemaCatalog::new()),
        }
    }

    /// Creates a new catalog with schema constraints enabled.
    ///
    /// This is now equivalent to `new()` since schema is always enabled.
    #[must_use]
    pub fn with_schema() -> Self {
        Self::new()
    }

    // === Label Operations ===

    /// Gets or creates a label ID for the given label name.
    pub fn get_or_create_label(&self, name: &str) -> LabelId {
        self.labels.get_or_create(name)
    }

    /// Gets the label ID for a label name, if it exists.
    #[must_use]
    pub fn get_label_id(&self, name: &str) -> Option<LabelId> {
        self.labels.get_id(name)
    }

    /// Gets the label name for a label ID, if it exists.
    #[must_use]
    pub fn get_label_name(&self, id: LabelId) -> Option<Arc<str>> {
        self.labels.get_name(id)
    }

    /// Returns the number of distinct labels.
    #[must_use]
    pub fn label_count(&self) -> usize {
        self.labels.count()
    }

    /// Returns all label names.
    #[must_use]
    pub fn all_labels(&self) -> Vec<Arc<str>> {
        self.labels.all_names()
    }

    // === Property Key Operations ===

    /// Gets or creates a property key ID for the given property key name.
    pub fn get_or_create_property_key(&self, name: &str) -> PropertyKeyId {
        self.property_keys.get_or_create(name)
    }

    /// Gets the property key ID for a property key name, if it exists.
    #[must_use]
    pub fn get_property_key_id(&self, name: &str) -> Option<PropertyKeyId> {
        self.property_keys.get_id(name)
    }

    /// Gets the property key name for a property key ID, if it exists.
    #[must_use]
    pub fn get_property_key_name(&self, id: PropertyKeyId) -> Option<Arc<str>> {
        self.property_keys.get_name(id)
    }

    /// Returns the number of distinct property keys.
    #[must_use]
    pub fn property_key_count(&self) -> usize {
        self.property_keys.count()
    }

    /// Returns all property key names.
    #[must_use]
    pub fn all_property_keys(&self) -> Vec<Arc<str>> {
        self.property_keys.all_names()
    }

    // === Edge Type Operations ===

    /// Gets or creates an edge type ID for the given edge type name.
    pub fn get_or_create_edge_type(&self, name: &str) -> EdgeTypeId {
        self.edge_types.get_or_create(name)
    }

    /// Gets the edge type ID for an edge type name, if it exists.
    #[must_use]
    pub fn get_edge_type_id(&self, name: &str) -> Option<EdgeTypeId> {
        self.edge_types.get_id(name)
    }

    /// Gets the edge type name for an edge type ID, if it exists.
    #[must_use]
    pub fn get_edge_type_name(&self, id: EdgeTypeId) -> Option<Arc<str>> {
        self.edge_types.get_name(id)
    }

    /// Returns the number of distinct edge types.
    #[must_use]
    pub fn edge_type_count(&self) -> usize {
        self.edge_types.count()
    }

    /// Returns all edge type names.
    #[must_use]
    pub fn all_edge_types(&self) -> Vec<Arc<str>> {
        self.edge_types.all_names()
    }

    // === Index Operations ===

    /// Creates a new index on a label and property key.
    pub fn create_index(
        &self,
        label: LabelId,
        property_key: PropertyKeyId,
        index_type: IndexType,
    ) -> IndexId {
        self.indexes.create(label, property_key, index_type)
    }

    /// Drops an index by ID.
    pub fn drop_index(&self, id: IndexId) -> bool {
        self.indexes.drop(id)
    }

    /// Gets the index definition for an index ID.
    #[must_use]
    pub fn get_index(&self, id: IndexId) -> Option<IndexDefinition> {
        self.indexes.get(id)
    }

    /// Finds indexes for a given label.
    #[must_use]
    pub fn indexes_for_label(&self, label: LabelId) -> Vec<IndexId> {
        self.indexes.for_label(label)
    }

    /// Finds indexes for a given label and property key.
    #[must_use]
    pub fn indexes_for_label_property(
        &self,
        label: LabelId,
        property_key: PropertyKeyId,
    ) -> Vec<IndexId> {
        self.indexes.for_label_property(label, property_key)
    }

    /// Returns all index definitions.
    #[must_use]
    pub fn all_indexes(&self) -> Vec<IndexDefinition> {
        self.indexes.all()
    }

    /// Returns the number of indexes.
    #[must_use]
    pub fn index_count(&self) -> usize {
        self.indexes.count()
    }

    // === Schema Operations ===

    /// Returns whether schema constraints are enabled.
    #[must_use]
    pub fn has_schema(&self) -> bool {
        self.schema.is_some()
    }

    /// Adds a uniqueness constraint.
    ///
    /// Returns an error if schema is not enabled or constraint already exists.
    pub fn add_unique_constraint(
        &self,
        label: LabelId,
        property_key: PropertyKeyId,
    ) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.add_unique_constraint(label, property_key),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Adds a required property constraint (NOT NULL).
    ///
    /// Returns an error if schema is not enabled or constraint already exists.
    pub fn add_required_property(
        &self,
        label: LabelId,
        property_key: PropertyKeyId,
    ) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.add_required_property(label, property_key),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Checks if a property is required for a label.
    #[must_use]
    pub fn is_property_required(&self, label: LabelId, property_key: PropertyKeyId) -> bool {
        self.schema
            .as_ref()
            .is_some_and(|s| s.is_property_required(label, property_key))
    }

    /// Checks if a property must be unique for a label.
    #[must_use]
    pub fn is_property_unique(&self, label: LabelId, property_key: PropertyKeyId) -> bool {
        self.schema
            .as_ref()
            .is_some_and(|s| s.is_property_unique(label, property_key))
    }

    // === Type Definition Operations ===

    /// Returns a reference to the schema catalog.
    #[must_use]
    pub fn schema(&self) -> Option<&SchemaCatalog> {
        self.schema.as_ref()
    }

    /// Registers a node type definition.
    pub fn register_node_type(&self, def: NodeTypeDefinition) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.register_node_type(def),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Registers or replaces a node type definition.
    pub fn register_or_replace_node_type(&self, def: NodeTypeDefinition) {
        if let Some(schema) = &self.schema {
            schema.register_or_replace_node_type(def);
        }
    }

    /// Drops a node type definition.
    pub fn drop_node_type(&self, name: &str) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.drop_node_type(name),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Gets a node type definition by name.
    #[must_use]
    pub fn get_node_type(&self, name: &str) -> Option<NodeTypeDefinition> {
        self.schema.as_ref().and_then(|s| s.get_node_type(name))
    }

    /// Returns all registered node type names.
    #[must_use]
    pub fn all_node_type_names(&self) -> Vec<String> {
        self.schema
            .as_ref()
            .map(SchemaCatalog::all_node_types)
            .unwrap_or_default()
    }

    /// Returns all registered edge type definition names.
    #[must_use]
    pub fn all_edge_type_names(&self) -> Vec<String> {
        self.schema
            .as_ref()
            .map(SchemaCatalog::all_edge_types)
            .unwrap_or_default()
    }

    /// Registers an edge type definition.
    pub fn register_edge_type_def(&self, def: EdgeTypeDefinition) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.register_edge_type(def),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Registers or replaces an edge type definition.
    pub fn register_or_replace_edge_type_def(&self, def: EdgeTypeDefinition) {
        if let Some(schema) = &self.schema {
            schema.register_or_replace_edge_type(def);
        }
    }

    /// Drops an edge type definition.
    pub fn drop_edge_type_def(&self, name: &str) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.drop_edge_type(name),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Gets an edge type definition by name.
    #[must_use]
    pub fn get_edge_type_def(&self, name: &str) -> Option<EdgeTypeDefinition> {
        self.schema.as_ref().and_then(|s| s.get_edge_type(name))
    }

    /// Registers a graph type definition.
    pub fn register_graph_type(&self, def: GraphTypeDefinition) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.register_graph_type(def),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Drops a graph type definition.
    pub fn drop_graph_type(&self, name: &str) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.drop_graph_type(name),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Registers a schema namespace.
    pub fn register_schema_namespace(&self, name: String) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.register_schema(name),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Drops a schema namespace.
    pub fn drop_schema_namespace(&self, name: &str) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.drop_schema(name),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Adds a property to a node type.
    pub fn alter_node_type_add_property(
        &self,
        type_name: &str,
        property: TypedProperty,
    ) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.alter_node_type_add_property(type_name, property),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Drops a property from a node type.
    pub fn alter_node_type_drop_property(
        &self,
        type_name: &str,
        property_name: &str,
    ) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.alter_node_type_drop_property(type_name, property_name),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Adds a property to an edge type.
    pub fn alter_edge_type_add_property(
        &self,
        type_name: &str,
        property: TypedProperty,
    ) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.alter_edge_type_add_property(type_name, property),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Drops a property from an edge type.
    pub fn alter_edge_type_drop_property(
        &self,
        type_name: &str,
        property_name: &str,
    ) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.alter_edge_type_drop_property(type_name, property_name),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Adds a node type to a graph type.
    pub fn alter_graph_type_add_node_type(
        &self,
        graph_type_name: &str,
        node_type: String,
    ) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.alter_graph_type_add_node_type(graph_type_name, node_type),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Drops a node type from a graph type.
    pub fn alter_graph_type_drop_node_type(
        &self,
        graph_type_name: &str,
        node_type: &str,
    ) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.alter_graph_type_drop_node_type(graph_type_name, node_type),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Adds an edge type to a graph type.
    pub fn alter_graph_type_add_edge_type(
        &self,
        graph_type_name: &str,
        edge_type: String,
    ) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.alter_graph_type_add_edge_type(graph_type_name, edge_type),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Drops an edge type from a graph type.
    pub fn alter_graph_type_drop_edge_type(
        &self,
        graph_type_name: &str,
        edge_type: &str,
    ) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.alter_graph_type_drop_edge_type(graph_type_name, edge_type),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Binds a graph instance to a graph type.
    pub fn bind_graph_type(
        &self,
        graph_name: &str,
        graph_type: String,
    ) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => {
                // Verify the graph type exists
                if schema.get_graph_type(&graph_type).is_none() {
                    return Err(CatalogError::TypeNotFound(graph_type));
                }
                schema
                    .graph_type_bindings
                    .write()
                    .insert(graph_name.to_string(), graph_type);
                Ok(())
            }
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Gets the graph type binding for a graph instance.
    pub fn get_graph_type_binding(&self, graph_name: &str) -> Option<String> {
        self.schema
            .as_ref()?
            .graph_type_bindings
            .read()
            .get(graph_name)
            .cloned()
    }

    /// Registers a stored procedure.
    pub fn register_procedure(&self, def: ProcedureDefinition) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.register_procedure(def),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Replaces or creates a stored procedure.
    pub fn replace_procedure(&self, def: ProcedureDefinition) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => {
                schema.replace_procedure(def);
                Ok(())
            }
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Drops a stored procedure.
    pub fn drop_procedure(&self, name: &str) -> Result<(), CatalogError> {
        match &self.schema {
            Some(schema) => schema.drop_procedure(name),
            None => Err(CatalogError::SchemaNotEnabled),
        }
    }

    /// Gets a stored procedure by name.
    pub fn get_procedure(&self, name: &str) -> Option<ProcedureDefinition> {
        self.schema.as_ref()?.get_procedure(name)
    }
}

impl Default for Catalog {
    fn default() -> Self {
        Self::new()
    }
}

// === Label Catalog ===

/// Bidirectional mapping between label names and IDs.
struct LabelCatalog {
    name_to_id: RwLock<HashMap<Arc<str>, LabelId>>,
    id_to_name: RwLock<Vec<Arc<str>>>,
    next_id: AtomicU32,
}

impl LabelCatalog {
    fn new() -> Self {
        Self {
            name_to_id: RwLock::new(HashMap::new()),
            id_to_name: RwLock::new(Vec::new()),
            next_id: AtomicU32::new(0),
        }
    }

    fn get_or_create(&self, name: &str) -> LabelId {
        // Fast path: check if already exists
        {
            let name_to_id = self.name_to_id.read();
            if let Some(&id) = name_to_id.get(name) {
                return id;
            }
        }

        // Slow path: create new entry
        let mut name_to_id = self.name_to_id.write();
        let mut id_to_name = self.id_to_name.write();

        // Double-check after acquiring write lock
        if let Some(&id) = name_to_id.get(name) {
            return id;
        }

        let id = LabelId::new(self.next_id.fetch_add(1, Ordering::Relaxed));
        let name: Arc<str> = name.into();
        name_to_id.insert(Arc::clone(&name), id);
        id_to_name.push(name);
        id
    }

    fn get_id(&self, name: &str) -> Option<LabelId> {
        self.name_to_id.read().get(name).copied()
    }

    fn get_name(&self, id: LabelId) -> Option<Arc<str>> {
        self.id_to_name.read().get(id.as_u32() as usize).cloned()
    }

    fn count(&self) -> usize {
        self.id_to_name.read().len()
    }

    fn all_names(&self) -> Vec<Arc<str>> {
        self.id_to_name.read().clone()
    }
}

// === Property Catalog ===

/// Bidirectional mapping between property key names and IDs.
struct PropertyCatalog {
    name_to_id: RwLock<HashMap<Arc<str>, PropertyKeyId>>,
    id_to_name: RwLock<Vec<Arc<str>>>,
    next_id: AtomicU32,
}

impl PropertyCatalog {
    fn new() -> Self {
        Self {
            name_to_id: RwLock::new(HashMap::new()),
            id_to_name: RwLock::new(Vec::new()),
            next_id: AtomicU32::new(0),
        }
    }

    fn get_or_create(&self, name: &str) -> PropertyKeyId {
        // Fast path: check if already exists
        {
            let name_to_id = self.name_to_id.read();
            if let Some(&id) = name_to_id.get(name) {
                return id;
            }
        }

        // Slow path: create new entry
        let mut name_to_id = self.name_to_id.write();
        let mut id_to_name = self.id_to_name.write();

        // Double-check after acquiring write lock
        if let Some(&id) = name_to_id.get(name) {
            return id;
        }

        let id = PropertyKeyId::new(self.next_id.fetch_add(1, Ordering::Relaxed));
        let name: Arc<str> = name.into();
        name_to_id.insert(Arc::clone(&name), id);
        id_to_name.push(name);
        id
    }

    fn get_id(&self, name: &str) -> Option<PropertyKeyId> {
        self.name_to_id.read().get(name).copied()
    }

    fn get_name(&self, id: PropertyKeyId) -> Option<Arc<str>> {
        self.id_to_name.read().get(id.as_u32() as usize).cloned()
    }

    fn count(&self) -> usize {
        self.id_to_name.read().len()
    }

    fn all_names(&self) -> Vec<Arc<str>> {
        self.id_to_name.read().clone()
    }
}

// === Edge Type Catalog ===

/// Bidirectional mapping between edge type names and IDs.
struct EdgeTypeCatalog {
    name_to_id: RwLock<HashMap<Arc<str>, EdgeTypeId>>,
    id_to_name: RwLock<Vec<Arc<str>>>,
    next_id: AtomicU32,
}

impl EdgeTypeCatalog {
    fn new() -> Self {
        Self {
            name_to_id: RwLock::new(HashMap::new()),
            id_to_name: RwLock::new(Vec::new()),
            next_id: AtomicU32::new(0),
        }
    }

    fn get_or_create(&self, name: &str) -> EdgeTypeId {
        // Fast path: check if already exists
        {
            let name_to_id = self.name_to_id.read();
            if let Some(&id) = name_to_id.get(name) {
                return id;
            }
        }

        // Slow path: create new entry
        let mut name_to_id = self.name_to_id.write();
        let mut id_to_name = self.id_to_name.write();

        // Double-check after acquiring write lock
        if let Some(&id) = name_to_id.get(name) {
            return id;
        }

        let id = EdgeTypeId::new(self.next_id.fetch_add(1, Ordering::Relaxed));
        let name: Arc<str> = name.into();
        name_to_id.insert(Arc::clone(&name), id);
        id_to_name.push(name);
        id
    }

    fn get_id(&self, name: &str) -> Option<EdgeTypeId> {
        self.name_to_id.read().get(name).copied()
    }

    fn get_name(&self, id: EdgeTypeId) -> Option<Arc<str>> {
        self.id_to_name.read().get(id.as_u32() as usize).cloned()
    }

    fn count(&self) -> usize {
        self.id_to_name.read().len()
    }

    fn all_names(&self) -> Vec<Arc<str>> {
        self.id_to_name.read().clone()
    }
}

// === Index Catalog ===

/// Type of index.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IndexType {
    /// Hash index for equality lookups.
    Hash,
    /// BTree index for range queries.
    BTree,
    /// Full-text index for text search.
    FullText,
}

/// Index definition.
#[derive(Debug, Clone)]
pub struct IndexDefinition {
    /// The index ID.
    pub id: IndexId,
    /// The label this index applies to.
    pub label: LabelId,
    /// The property key being indexed.
    pub property_key: PropertyKeyId,
    /// The type of index.
    pub index_type: IndexType,
}

/// Manages index definitions.
struct IndexCatalog {
    indexes: RwLock<HashMap<IndexId, IndexDefinition>>,
    label_indexes: RwLock<HashMap<LabelId, Vec<IndexId>>>,
    label_property_indexes: RwLock<HashMap<(LabelId, PropertyKeyId), Vec<IndexId>>>,
    next_id: AtomicU32,
}

impl IndexCatalog {
    fn new() -> Self {
        Self {
            indexes: RwLock::new(HashMap::new()),
            label_indexes: RwLock::new(HashMap::new()),
            label_property_indexes: RwLock::new(HashMap::new()),
            next_id: AtomicU32::new(0),
        }
    }

    fn create(
        &self,
        label: LabelId,
        property_key: PropertyKeyId,
        index_type: IndexType,
    ) -> IndexId {
        let id = IndexId::new(self.next_id.fetch_add(1, Ordering::Relaxed));
        let definition = IndexDefinition {
            id,
            label,
            property_key,
            index_type,
        };

        let mut indexes = self.indexes.write();
        let mut label_indexes = self.label_indexes.write();
        let mut label_property_indexes = self.label_property_indexes.write();

        indexes.insert(id, definition);
        label_indexes.entry(label).or_default().push(id);
        label_property_indexes
            .entry((label, property_key))
            .or_default()
            .push(id);

        id
    }

    fn drop(&self, id: IndexId) -> bool {
        let mut indexes = self.indexes.write();
        let mut label_indexes = self.label_indexes.write();
        let mut label_property_indexes = self.label_property_indexes.write();

        if let Some(definition) = indexes.remove(&id) {
            // Remove from label index
            if let Some(ids) = label_indexes.get_mut(&definition.label) {
                ids.retain(|&i| i != id);
            }
            // Remove from label-property index
            if let Some(ids) =
                label_property_indexes.get_mut(&(definition.label, definition.property_key))
            {
                ids.retain(|&i| i != id);
            }
            true
        } else {
            false
        }
    }

    fn get(&self, id: IndexId) -> Option<IndexDefinition> {
        self.indexes.read().get(&id).cloned()
    }

    fn for_label(&self, label: LabelId) -> Vec<IndexId> {
        self.label_indexes
            .read()
            .get(&label)
            .cloned()
            .unwrap_or_default()
    }

    fn for_label_property(&self, label: LabelId, property_key: PropertyKeyId) -> Vec<IndexId> {
        self.label_property_indexes
            .read()
            .get(&(label, property_key))
            .cloned()
            .unwrap_or_default()
    }

    fn count(&self) -> usize {
        self.indexes.read().len()
    }

    fn all(&self) -> Vec<IndexDefinition> {
        self.indexes.read().values().cloned().collect()
    }
}

// === Type Definitions ===

/// Data type for a typed property in a node or edge type definition.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PropertyDataType {
    /// UTF-8 string.
    String,
    /// 64-bit signed integer.
    Int64,
    /// 64-bit floating point.
    Float64,
    /// Boolean.
    Bool,
    /// Calendar date.
    Date,
    /// Time of day.
    Time,
    /// Timestamp (date + time).
    Timestamp,
    /// Duration / interval.
    Duration,
    /// Ordered list of values (untyped).
    List,
    /// Typed list: `LIST<element_type>` (ISO sec 4.16.9).
    ListTyped(Box<PropertyDataType>),
    /// Key-value map.
    Map,
    /// Raw bytes.
    Bytes,
    /// Node reference type (ISO sec 4.15.1).
    Node,
    /// Edge reference type (ISO sec 4.15.1).
    Edge,
    /// Any type (no enforcement).
    Any,
}

impl PropertyDataType {
    /// Parses a type name string (case-insensitive) into a `PropertyDataType`.
    #[must_use]
    pub fn from_type_name(name: &str) -> Self {
        let upper = name.to_uppercase();
        // Handle parameterized LIST<element_type>
        if let Some(inner) = upper
            .strip_prefix("LIST<")
            .and_then(|s| s.strip_suffix('>'))
        {
            return Self::ListTyped(Box::new(Self::from_type_name(inner)));
        }
        match upper.as_str() {
            "STRING" | "VARCHAR" | "TEXT" => Self::String,
            "INT" | "INT64" | "INTEGER" | "BIGINT" => Self::Int64,
            "FLOAT" | "FLOAT64" | "DOUBLE" | "REAL" => Self::Float64,
            "BOOL" | "BOOLEAN" => Self::Bool,
            "DATE" => Self::Date,
            "TIME" => Self::Time,
            "TIMESTAMP" | "DATETIME" => Self::Timestamp,
            "DURATION" | "INTERVAL" => Self::Duration,
            "LIST" | "ARRAY" => Self::List,
            "MAP" | "RECORD" => Self::Map,
            "BYTES" | "BINARY" | "BLOB" => Self::Bytes,
            "NODE" => Self::Node,
            "EDGE" | "RELATIONSHIP" => Self::Edge,
            _ => Self::Any,
        }
    }

    /// Checks whether a value conforms to this type.
    #[must_use]
    pub fn matches(&self, value: &Value) -> bool {
        match (self, value) {
            (Self::Any, _) | (_, Value::Null) => true,
            (Self::String, Value::String(_)) => true,
            (Self::Int64, Value::Int64(_)) => true,
            (Self::Float64, Value::Float64(_)) => true,
            (Self::Bool, Value::Bool(_)) => true,
            (Self::Date, Value::Date(_)) => true,
            (Self::Time, Value::Time(_)) => true,
            (Self::Timestamp, Value::Timestamp(_)) => true,
            (Self::Duration, Value::Duration(_)) => true,
            (Self::List, Value::List(_)) => true,
            (Self::ListTyped(elem_type), Value::List(items)) => {
                items.iter().all(|item| elem_type.matches(item))
            }
            (Self::Bytes, Value::Bytes(_)) => true,
            // Node/Edge reference types match Map values (graph elements are
            // represented as maps with _id, _labels/_type, and properties)
            (Self::Node | Self::Edge, Value::Map(_)) => true,
            _ => false,
        }
    }
}

impl std::fmt::Display for PropertyDataType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::String => write!(f, "STRING"),
            Self::Int64 => write!(f, "INT64"),
            Self::Float64 => write!(f, "FLOAT64"),
            Self::Bool => write!(f, "BOOLEAN"),
            Self::Date => write!(f, "DATE"),
            Self::Time => write!(f, "TIME"),
            Self::Timestamp => write!(f, "TIMESTAMP"),
            Self::Duration => write!(f, "DURATION"),
            Self::List => write!(f, "LIST"),
            Self::ListTyped(elem) => write!(f, "LIST<{elem}>"),
            Self::Map => write!(f, "MAP"),
            Self::Bytes => write!(f, "BYTES"),
            Self::Node => write!(f, "NODE"),
            Self::Edge => write!(f, "EDGE"),
            Self::Any => write!(f, "ANY"),
        }
    }
}

/// A typed property within a node or edge type definition.
#[derive(Debug, Clone)]
pub struct TypedProperty {
    /// Property name.
    pub name: String,
    /// Expected data type.
    pub data_type: PropertyDataType,
    /// Whether NULL values are allowed.
    pub nullable: bool,
    /// Default value (used when property is not explicitly set).
    pub default_value: Option<Value>,
}

/// A constraint on a node or edge type.
#[derive(Debug, Clone)]
pub enum TypeConstraint {
    /// Primary key (implies UNIQUE + NOT NULL).
    PrimaryKey(Vec<String>),
    /// Uniqueness constraint on one or more properties.
    Unique(Vec<String>),
    /// NOT NULL constraint on a single property.
    NotNull(String),
    /// CHECK constraint with a named expression string.
    Check {
        /// Optional constraint name.
        name: Option<String>,
        /// Expression (stored as string for now).
        expression: String,
    },
}

/// Definition of a node type (label schema).
#[derive(Debug, Clone)]
pub struct NodeTypeDefinition {
    /// Type name (corresponds to a label).
    pub name: String,
    /// Typed property definitions.
    pub properties: Vec<TypedProperty>,
    /// Type-level constraints.
    pub constraints: Vec<TypeConstraint>,
}

/// Definition of an edge type (relationship type schema).
#[derive(Debug, Clone)]
pub struct EdgeTypeDefinition {
    /// Type name (corresponds to an edge type / relationship type).
    pub name: String,
    /// Typed property definitions.
    pub properties: Vec<TypedProperty>,
    /// Type-level constraints.
    pub constraints: Vec<TypeConstraint>,
}

/// Definition of a graph type (constrains which node/edge types a graph allows).
#[derive(Debug, Clone)]
pub struct GraphTypeDefinition {
    /// Graph type name.
    pub name: String,
    /// Allowed node types (empty = open).
    pub allowed_node_types: Vec<String>,
    /// Allowed edge types (empty = open).
    pub allowed_edge_types: Vec<String>,
    /// Whether unlisted types are permitted.
    pub open: bool,
}

/// Definition of a stored procedure.
#[derive(Debug, Clone)]
pub struct ProcedureDefinition {
    /// Procedure name.
    pub name: String,
    /// Parameter definitions: (name, type).
    pub params: Vec<(String, String)>,
    /// Return column definitions: (name, type).
    pub returns: Vec<(String, String)>,
    /// Raw GQL query body.
    pub body: String,
}

// === Schema Catalog ===

/// Schema constraints and type definitions.
pub struct SchemaCatalog {
    /// Properties that must be unique for a given label.
    unique_constraints: RwLock<HashSet<(LabelId, PropertyKeyId)>>,
    /// Properties that are required (NOT NULL) for a given label.
    required_properties: RwLock<HashSet<(LabelId, PropertyKeyId)>>,
    /// Registered node type definitions.
    node_types: RwLock<HashMap<String, NodeTypeDefinition>>,
    /// Registered edge type definitions.
    edge_types: RwLock<HashMap<String, EdgeTypeDefinition>>,
    /// Registered graph type definitions.
    graph_types: RwLock<HashMap<String, GraphTypeDefinition>>,
    /// Schema namespaces.
    schemas: RwLock<Vec<String>>,
    /// Graph instance to graph type bindings.
    graph_type_bindings: RwLock<HashMap<String, String>>,
    /// Stored procedure definitions.
    procedures: RwLock<HashMap<String, ProcedureDefinition>>,
}

impl SchemaCatalog {
    fn new() -> Self {
        Self {
            unique_constraints: RwLock::new(HashSet::new()),
            required_properties: RwLock::new(HashSet::new()),
            node_types: RwLock::new(HashMap::new()),
            edge_types: RwLock::new(HashMap::new()),
            graph_types: RwLock::new(HashMap::new()),
            schemas: RwLock::new(Vec::new()),
            graph_type_bindings: RwLock::new(HashMap::new()),
            procedures: RwLock::new(HashMap::new()),
        }
    }

    // --- Node type operations ---

    /// Registers a new node type definition.
    pub fn register_node_type(&self, def: NodeTypeDefinition) -> Result<(), CatalogError> {
        let mut types = self.node_types.write();
        if types.contains_key(&def.name) {
            return Err(CatalogError::TypeAlreadyExists(def.name));
        }
        types.insert(def.name.clone(), def);
        Ok(())
    }

    /// Registers or replaces a node type definition.
    pub fn register_or_replace_node_type(&self, def: NodeTypeDefinition) {
        self.node_types.write().insert(def.name.clone(), def);
    }

    /// Drops a node type definition by name.
    pub fn drop_node_type(&self, name: &str) -> Result<(), CatalogError> {
        let mut types = self.node_types.write();
        if types.remove(name).is_none() {
            return Err(CatalogError::TypeNotFound(name.to_string()));
        }
        Ok(())
    }

    /// Gets a node type definition by name.
    #[must_use]
    pub fn get_node_type(&self, name: &str) -> Option<NodeTypeDefinition> {
        self.node_types.read().get(name).cloned()
    }

    /// Returns all registered node type names.
    #[must_use]
    pub fn all_node_types(&self) -> Vec<String> {
        self.node_types.read().keys().cloned().collect()
    }

    // --- Edge type operations ---

    /// Registers a new edge type definition.
    pub fn register_edge_type(&self, def: EdgeTypeDefinition) -> Result<(), CatalogError> {
        let mut types = self.edge_types.write();
        if types.contains_key(&def.name) {
            return Err(CatalogError::TypeAlreadyExists(def.name));
        }
        types.insert(def.name.clone(), def);
        Ok(())
    }

    /// Registers or replaces an edge type definition.
    pub fn register_or_replace_edge_type(&self, def: EdgeTypeDefinition) {
        self.edge_types.write().insert(def.name.clone(), def);
    }

    /// Drops an edge type definition by name.
    pub fn drop_edge_type(&self, name: &str) -> Result<(), CatalogError> {
        let mut types = self.edge_types.write();
        if types.remove(name).is_none() {
            return Err(CatalogError::TypeNotFound(name.to_string()));
        }
        Ok(())
    }

    /// Gets an edge type definition by name.
    #[must_use]
    pub fn get_edge_type(&self, name: &str) -> Option<EdgeTypeDefinition> {
        self.edge_types.read().get(name).cloned()
    }

    /// Returns all registered edge type names.
    #[must_use]
    pub fn all_edge_types(&self) -> Vec<String> {
        self.edge_types.read().keys().cloned().collect()
    }

    // --- Graph type operations ---

    /// Registers a new graph type definition.
    pub fn register_graph_type(&self, def: GraphTypeDefinition) -> Result<(), CatalogError> {
        let mut types = self.graph_types.write();
        if types.contains_key(&def.name) {
            return Err(CatalogError::TypeAlreadyExists(def.name));
        }
        types.insert(def.name.clone(), def);
        Ok(())
    }

    /// Drops a graph type definition by name.
    pub fn drop_graph_type(&self, name: &str) -> Result<(), CatalogError> {
        let mut types = self.graph_types.write();
        if types.remove(name).is_none() {
            return Err(CatalogError::TypeNotFound(name.to_string()));
        }
        Ok(())
    }

    /// Gets a graph type definition by name.
    #[must_use]
    pub fn get_graph_type(&self, name: &str) -> Option<GraphTypeDefinition> {
        self.graph_types.read().get(name).cloned()
    }

    // --- Schema namespace operations ---

    /// Registers a schema namespace.
    pub fn register_schema(&self, name: String) -> Result<(), CatalogError> {
        let mut schemas = self.schemas.write();
        if schemas.contains(&name) {
            return Err(CatalogError::SchemaAlreadyExists(name));
        }
        schemas.push(name);
        Ok(())
    }

    /// Drops a schema namespace.
    pub fn drop_schema(&self, name: &str) -> Result<(), CatalogError> {
        let mut schemas = self.schemas.write();
        if let Some(pos) = schemas.iter().position(|s| s == name) {
            schemas.remove(pos);
            Ok(())
        } else {
            Err(CatalogError::SchemaNotFound(name.to_string()))
        }
    }

    // --- ALTER operations ---

    /// Adds a property to an existing node type.
    pub fn alter_node_type_add_property(
        &self,
        type_name: &str,
        property: TypedProperty,
    ) -> Result<(), CatalogError> {
        let mut types = self.node_types.write();
        let def = types
            .get_mut(type_name)
            .ok_or_else(|| CatalogError::TypeNotFound(type_name.to_string()))?;
        if def.properties.iter().any(|p| p.name == property.name) {
            return Err(CatalogError::TypeAlreadyExists(format!(
                "property {} on {}",
                property.name, type_name
            )));
        }
        def.properties.push(property);
        Ok(())
    }

    /// Drops a property from an existing node type.
    pub fn alter_node_type_drop_property(
        &self,
        type_name: &str,
        property_name: &str,
    ) -> Result<(), CatalogError> {
        let mut types = self.node_types.write();
        let def = types
            .get_mut(type_name)
            .ok_or_else(|| CatalogError::TypeNotFound(type_name.to_string()))?;
        let len_before = def.properties.len();
        def.properties.retain(|p| p.name != property_name);
        if def.properties.len() == len_before {
            return Err(CatalogError::TypeNotFound(format!(
                "property {} on {}",
                property_name, type_name
            )));
        }
        Ok(())
    }

    /// Adds a property to an existing edge type.
    pub fn alter_edge_type_add_property(
        &self,
        type_name: &str,
        property: TypedProperty,
    ) -> Result<(), CatalogError> {
        let mut types = self.edge_types.write();
        let def = types
            .get_mut(type_name)
            .ok_or_else(|| CatalogError::TypeNotFound(type_name.to_string()))?;
        if def.properties.iter().any(|p| p.name == property.name) {
            return Err(CatalogError::TypeAlreadyExists(format!(
                "property {} on {}",
                property.name, type_name
            )));
        }
        def.properties.push(property);
        Ok(())
    }

    /// Drops a property from an existing edge type.
    pub fn alter_edge_type_drop_property(
        &self,
        type_name: &str,
        property_name: &str,
    ) -> Result<(), CatalogError> {
        let mut types = self.edge_types.write();
        let def = types
            .get_mut(type_name)
            .ok_or_else(|| CatalogError::TypeNotFound(type_name.to_string()))?;
        let len_before = def.properties.len();
        def.properties.retain(|p| p.name != property_name);
        if def.properties.len() == len_before {
            return Err(CatalogError::TypeNotFound(format!(
                "property {} on {}",
                property_name, type_name
            )));
        }
        Ok(())
    }

    /// Adds a node type to a graph type.
    pub fn alter_graph_type_add_node_type(
        &self,
        graph_type_name: &str,
        node_type: String,
    ) -> Result<(), CatalogError> {
        let mut types = self.graph_types.write();
        let def = types
            .get_mut(graph_type_name)
            .ok_or_else(|| CatalogError::TypeNotFound(graph_type_name.to_string()))?;
        if !def.allowed_node_types.contains(&node_type) {
            def.allowed_node_types.push(node_type);
        }
        Ok(())
    }

    /// Drops a node type from a graph type.
    pub fn alter_graph_type_drop_node_type(
        &self,
        graph_type_name: &str,
        node_type: &str,
    ) -> Result<(), CatalogError> {
        let mut types = self.graph_types.write();
        let def = types
            .get_mut(graph_type_name)
            .ok_or_else(|| CatalogError::TypeNotFound(graph_type_name.to_string()))?;
        def.allowed_node_types.retain(|t| t != node_type);
        Ok(())
    }

    /// Adds an edge type to a graph type.
    pub fn alter_graph_type_add_edge_type(
        &self,
        graph_type_name: &str,
        edge_type: String,
    ) -> Result<(), CatalogError> {
        let mut types = self.graph_types.write();
        let def = types
            .get_mut(graph_type_name)
            .ok_or_else(|| CatalogError::TypeNotFound(graph_type_name.to_string()))?;
        if !def.allowed_edge_types.contains(&edge_type) {
            def.allowed_edge_types.push(edge_type);
        }
        Ok(())
    }

    /// Drops an edge type from a graph type.
    pub fn alter_graph_type_drop_edge_type(
        &self,
        graph_type_name: &str,
        edge_type: &str,
    ) -> Result<(), CatalogError> {
        let mut types = self.graph_types.write();
        let def = types
            .get_mut(graph_type_name)
            .ok_or_else(|| CatalogError::TypeNotFound(graph_type_name.to_string()))?;
        def.allowed_edge_types.retain(|t| t != edge_type);
        Ok(())
    }

    // --- Procedure operations ---

    /// Registers a stored procedure.
    pub fn register_procedure(&self, def: ProcedureDefinition) -> Result<(), CatalogError> {
        let mut procs = self.procedures.write();
        if procs.contains_key(&def.name) {
            return Err(CatalogError::TypeAlreadyExists(def.name.clone()));
        }
        procs.insert(def.name.clone(), def);
        Ok(())
    }

    /// Replaces or creates a stored procedure.
    pub fn replace_procedure(&self, def: ProcedureDefinition) {
        self.procedures.write().insert(def.name.clone(), def);
    }

    /// Drops a stored procedure.
    pub fn drop_procedure(&self, name: &str) -> Result<(), CatalogError> {
        let mut procs = self.procedures.write();
        if procs.remove(name).is_none() {
            return Err(CatalogError::TypeNotFound(name.to_string()));
        }
        Ok(())
    }

    /// Gets a stored procedure by name.
    pub fn get_procedure(&self, name: &str) -> Option<ProcedureDefinition> {
        self.procedures.read().get(name).cloned()
    }

    fn add_unique_constraint(
        &self,
        label: LabelId,
        property_key: PropertyKeyId,
    ) -> Result<(), CatalogError> {
        let mut constraints = self.unique_constraints.write();
        let key = (label, property_key);
        if !constraints.insert(key) {
            return Err(CatalogError::ConstraintAlreadyExists);
        }
        Ok(())
    }

    fn add_required_property(
        &self,
        label: LabelId,
        property_key: PropertyKeyId,
    ) -> Result<(), CatalogError> {
        let mut required = self.required_properties.write();
        let key = (label, property_key);
        if !required.insert(key) {
            return Err(CatalogError::ConstraintAlreadyExists);
        }
        Ok(())
    }

    fn is_property_required(&self, label: LabelId, property_key: PropertyKeyId) -> bool {
        self.required_properties
            .read()
            .contains(&(label, property_key))
    }

    fn is_property_unique(&self, label: LabelId, property_key: PropertyKeyId) -> bool {
        self.unique_constraints
            .read()
            .contains(&(label, property_key))
    }
}

// === Errors ===

/// Catalog-related errors.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CatalogError {
    /// Schema constraints are not enabled.
    SchemaNotEnabled,
    /// The constraint already exists.
    ConstraintAlreadyExists,
    /// The label does not exist.
    LabelNotFound(String),
    /// The property key does not exist.
    PropertyKeyNotFound(String),
    /// The edge type does not exist.
    EdgeTypeNotFound(String),
    /// The index does not exist.
    IndexNotFound(IndexId),
    /// A type with this name already exists.
    TypeAlreadyExists(String),
    /// No type with this name exists.
    TypeNotFound(String),
    /// A schema with this name already exists.
    SchemaAlreadyExists(String),
    /// No schema with this name exists.
    SchemaNotFound(String),
}

impl std::fmt::Display for CatalogError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::SchemaNotEnabled => write!(f, "Schema constraints are not enabled"),
            Self::ConstraintAlreadyExists => write!(f, "Constraint already exists"),
            Self::LabelNotFound(name) => write!(f, "Label not found: {name}"),
            Self::PropertyKeyNotFound(name) => write!(f, "Property key not found: {name}"),
            Self::EdgeTypeNotFound(name) => write!(f, "Edge type not found: {name}"),
            Self::IndexNotFound(id) => write!(f, "Index not found: {id}"),
            Self::TypeAlreadyExists(name) => write!(f, "Type already exists: {name}"),
            Self::TypeNotFound(name) => write!(f, "Type not found: {name}"),
            Self::SchemaAlreadyExists(name) => write!(f, "Schema already exists: {name}"),
            Self::SchemaNotFound(name) => write!(f, "Schema not found: {name}"),
        }
    }
}

impl std::error::Error for CatalogError {}

// === Constraint Validator ===

use grafeo_core::execution::operators::ConstraintValidator;
use grafeo_core::execution::operators::OperatorError;

/// Validates schema constraints during mutation operations using the Catalog.
///
/// Checks type definitions, NOT NULL constraints, and UNIQUE constraints
/// against registered node/edge type definitions.
pub struct CatalogConstraintValidator {
    catalog: Arc<Catalog>,
}

impl CatalogConstraintValidator {
    /// Creates a new validator wrapping the given catalog.
    pub fn new(catalog: Arc<Catalog>) -> Self {
        Self { catalog }
    }
}

impl ConstraintValidator for CatalogConstraintValidator {
    fn validate_node_property(
        &self,
        labels: &[String],
        key: &str,
        value: &Value,
    ) -> Result<(), OperatorError> {
        for label in labels {
            if let Some(type_def) = self.catalog.get_node_type(label)
                && let Some(typed_prop) = type_def.properties.iter().find(|p| p.name == key)
            {
                // Check NOT NULL
                if !typed_prop.nullable && *value == Value::Null {
                    return Err(OperatorError::ConstraintViolation(format!(
                        "property '{key}' on :{label} is NOT NULL, cannot set to null"
                    )));
                }
                // Check type compatibility
                if *value != Value::Null && !typed_prop.data_type.matches(value) {
                    return Err(OperatorError::ConstraintViolation(format!(
                        "property '{key}' on :{label} expects {:?}, got {:?}",
                        typed_prop.data_type, value
                    )));
                }
            }
        }
        Ok(())
    }

    fn validate_node_complete(
        &self,
        labels: &[String],
        properties: &[(String, Value)],
    ) -> Result<(), OperatorError> {
        let prop_names: std::collections::HashSet<&str> =
            properties.iter().map(|(n, _)| n.as_str()).collect();

        for label in labels {
            if let Some(type_def) = self.catalog.get_node_type(label) {
                // Check that all NOT NULL properties are present
                for typed_prop in &type_def.properties {
                    if !typed_prop.nullable
                        && typed_prop.default_value.is_none()
                        && !prop_names.contains(typed_prop.name.as_str())
                    {
                        return Err(OperatorError::ConstraintViolation(format!(
                            "missing required property '{}' on :{label}",
                            typed_prop.name
                        )));
                    }
                }
                // Check type-level constraints
                for constraint in &type_def.constraints {
                    match constraint {
                        TypeConstraint::NotNull(prop_name) => {
                            if !prop_names.contains(prop_name.as_str()) {
                                return Err(OperatorError::ConstraintViolation(format!(
                                    "missing required property '{prop_name}' on :{label} (NOT NULL constraint)"
                                )));
                            }
                        }
                        TypeConstraint::PrimaryKey(key_props) => {
                            for pk in key_props {
                                if !prop_names.contains(pk.as_str()) {
                                    return Err(OperatorError::ConstraintViolation(format!(
                                        "missing primary key property '{pk}' on :{label}"
                                    )));
                                }
                            }
                        }
                        _ => {}
                    }
                }
            }
        }
        Ok(())
    }

    fn check_unique_node_property(
        &self,
        labels: &[String],
        key: &str,
        value: &Value,
    ) -> Result<(), OperatorError> {
        // Skip uniqueness check for NULL values (NULLs are never duplicates)
        if *value == Value::Null {
            return Ok(());
        }
        for label in labels {
            if let Some(type_def) = self.catalog.get_node_type(label) {
                for constraint in &type_def.constraints {
                    let is_unique = match constraint {
                        TypeConstraint::Unique(props) => props.iter().any(|p| p == key),
                        TypeConstraint::PrimaryKey(props) => props.iter().any(|p| p == key),
                        _ => false,
                    };
                    if is_unique {
                        // Use the catalog's label/property ID-based check if available
                        let label_id = self.catalog.get_or_create_label(label);
                        let prop_id = self.catalog.get_or_create_property_key(key);
                        if self.catalog.is_property_unique(label_id, prop_id) {
                            // The constraint is registered, enforcement depends on
                            // the property index which is checked at the store level.
                            // For now, we mark the constraint as active.
                            // Full duplicate detection requires index lookup, which
                            // is done when the property index exists.
                        }
                    }
                }
            }
        }
        Ok(())
    }

    fn validate_edge_property(
        &self,
        edge_type: &str,
        key: &str,
        value: &Value,
    ) -> Result<(), OperatorError> {
        if let Some(type_def) = self.catalog.get_edge_type_def(edge_type)
            && let Some(typed_prop) = type_def.properties.iter().find(|p| p.name == key)
        {
            // Check NOT NULL
            if !typed_prop.nullable && *value == Value::Null {
                return Err(OperatorError::ConstraintViolation(format!(
                    "property '{key}' on :{edge_type} is NOT NULL, cannot set to null"
                )));
            }
            // Check type compatibility
            if *value != Value::Null && !typed_prop.data_type.matches(value) {
                return Err(OperatorError::ConstraintViolation(format!(
                    "property '{key}' on :{edge_type} expects {:?}, got {:?}",
                    typed_prop.data_type, value
                )));
            }
        }
        Ok(())
    }

    fn validate_edge_complete(
        &self,
        edge_type: &str,
        properties: &[(String, Value)],
    ) -> Result<(), OperatorError> {
        if let Some(type_def) = self.catalog.get_edge_type_def(edge_type) {
            let prop_names: std::collections::HashSet<&str> =
                properties.iter().map(|(n, _)| n.as_str()).collect();

            for typed_prop in &type_def.properties {
                if !typed_prop.nullable
                    && typed_prop.default_value.is_none()
                    && !prop_names.contains(typed_prop.name.as_str())
                {
                    return Err(OperatorError::ConstraintViolation(format!(
                        "missing required property '{}' on :{edge_type}",
                        typed_prop.name
                    )));
                }
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::thread;

    #[test]
    fn test_catalog_labels() {
        let catalog = Catalog::new();

        // Get or create labels
        let person_id = catalog.get_or_create_label("Person");
        let company_id = catalog.get_or_create_label("Company");

        // IDs should be different
        assert_ne!(person_id, company_id);

        // Getting the same label should return the same ID
        assert_eq!(catalog.get_or_create_label("Person"), person_id);

        // Should be able to look up by name
        assert_eq!(catalog.get_label_id("Person"), Some(person_id));
        assert_eq!(catalog.get_label_id("Company"), Some(company_id));
        assert_eq!(catalog.get_label_id("Unknown"), None);

        // Should be able to look up by ID
        assert_eq!(catalog.get_label_name(person_id).as_deref(), Some("Person"));
        assert_eq!(
            catalog.get_label_name(company_id).as_deref(),
            Some("Company")
        );

        // Count should be correct
        assert_eq!(catalog.label_count(), 2);
    }

    #[test]
    fn test_catalog_property_keys() {
        let catalog = Catalog::new();

        let name_id = catalog.get_or_create_property_key("name");
        let age_id = catalog.get_or_create_property_key("age");

        assert_ne!(name_id, age_id);
        assert_eq!(catalog.get_or_create_property_key("name"), name_id);
        assert_eq!(catalog.get_property_key_id("name"), Some(name_id));
        assert_eq!(
            catalog.get_property_key_name(name_id).as_deref(),
            Some("name")
        );
        assert_eq!(catalog.property_key_count(), 2);
    }

    #[test]
    fn test_catalog_edge_types() {
        let catalog = Catalog::new();

        let knows_id = catalog.get_or_create_edge_type("KNOWS");
        let works_at_id = catalog.get_or_create_edge_type("WORKS_AT");

        assert_ne!(knows_id, works_at_id);
        assert_eq!(catalog.get_or_create_edge_type("KNOWS"), knows_id);
        assert_eq!(catalog.get_edge_type_id("KNOWS"), Some(knows_id));
        assert_eq!(
            catalog.get_edge_type_name(knows_id).as_deref(),
            Some("KNOWS")
        );
        assert_eq!(catalog.edge_type_count(), 2);
    }

    #[test]
    fn test_catalog_indexes() {
        let catalog = Catalog::new();

        let person_id = catalog.get_or_create_label("Person");
        let name_id = catalog.get_or_create_property_key("name");
        let age_id = catalog.get_or_create_property_key("age");

        // Create indexes
        let idx1 = catalog.create_index(person_id, name_id, IndexType::Hash);
        let idx2 = catalog.create_index(person_id, age_id, IndexType::BTree);

        assert_ne!(idx1, idx2);
        assert_eq!(catalog.index_count(), 2);

        // Look up by label
        let label_indexes = catalog.indexes_for_label(person_id);
        assert_eq!(label_indexes.len(), 2);
        assert!(label_indexes.contains(&idx1));
        assert!(label_indexes.contains(&idx2));

        // Look up by label and property
        let name_indexes = catalog.indexes_for_label_property(person_id, name_id);
        assert_eq!(name_indexes.len(), 1);
        assert_eq!(name_indexes[0], idx1);

        // Get definition
        let def = catalog.get_index(idx1).unwrap();
        assert_eq!(def.label, person_id);
        assert_eq!(def.property_key, name_id);
        assert_eq!(def.index_type, IndexType::Hash);

        // Drop index
        assert!(catalog.drop_index(idx1));
        assert_eq!(catalog.index_count(), 1);
        assert!(catalog.get_index(idx1).is_none());
        assert_eq!(catalog.indexes_for_label(person_id).len(), 1);
    }

    #[test]
    fn test_catalog_schema_constraints() {
        let catalog = Catalog::with_schema();

        let person_id = catalog.get_or_create_label("Person");
        let email_id = catalog.get_or_create_property_key("email");
        let name_id = catalog.get_or_create_property_key("name");

        // Add constraints
        assert!(catalog.add_unique_constraint(person_id, email_id).is_ok());
        assert!(catalog.add_required_property(person_id, name_id).is_ok());

        // Check constraints
        assert!(catalog.is_property_unique(person_id, email_id));
        assert!(!catalog.is_property_unique(person_id, name_id));
        assert!(catalog.is_property_required(person_id, name_id));
        assert!(!catalog.is_property_required(person_id, email_id));

        // Duplicate constraint should fail
        assert_eq!(
            catalog.add_unique_constraint(person_id, email_id),
            Err(CatalogError::ConstraintAlreadyExists)
        );
    }

    #[test]
    fn test_catalog_schema_always_enabled() {
        // Catalog::new() always enables schema
        let catalog = Catalog::new();
        assert!(catalog.has_schema());

        let person_id = catalog.get_or_create_label("Person");
        let email_id = catalog.get_or_create_property_key("email");

        // Should succeed with schema enabled
        assert_eq!(catalog.add_unique_constraint(person_id, email_id), Ok(()));
    }

    // === Additional tests for comprehensive coverage ===

    #[test]
    fn test_catalog_default() {
        let catalog = Catalog::default();
        assert!(catalog.has_schema());
        assert_eq!(catalog.label_count(), 0);
        assert_eq!(catalog.property_key_count(), 0);
        assert_eq!(catalog.edge_type_count(), 0);
        assert_eq!(catalog.index_count(), 0);
    }

    #[test]
    fn test_catalog_all_labels() {
        let catalog = Catalog::new();

        catalog.get_or_create_label("Person");
        catalog.get_or_create_label("Company");
        catalog.get_or_create_label("Product");

        let all = catalog.all_labels();
        assert_eq!(all.len(), 3);
        assert!(all.iter().any(|l| l.as_ref() == "Person"));
        assert!(all.iter().any(|l| l.as_ref() == "Company"));
        assert!(all.iter().any(|l| l.as_ref() == "Product"));
    }

    #[test]
    fn test_catalog_all_property_keys() {
        let catalog = Catalog::new();

        catalog.get_or_create_property_key("name");
        catalog.get_or_create_property_key("age");
        catalog.get_or_create_property_key("email");

        let all = catalog.all_property_keys();
        assert_eq!(all.len(), 3);
        assert!(all.iter().any(|k| k.as_ref() == "name"));
        assert!(all.iter().any(|k| k.as_ref() == "age"));
        assert!(all.iter().any(|k| k.as_ref() == "email"));
    }

    #[test]
    fn test_catalog_all_edge_types() {
        let catalog = Catalog::new();

        catalog.get_or_create_edge_type("KNOWS");
        catalog.get_or_create_edge_type("WORKS_AT");
        catalog.get_or_create_edge_type("LIVES_IN");

        let all = catalog.all_edge_types();
        assert_eq!(all.len(), 3);
        assert!(all.iter().any(|t| t.as_ref() == "KNOWS"));
        assert!(all.iter().any(|t| t.as_ref() == "WORKS_AT"));
        assert!(all.iter().any(|t| t.as_ref() == "LIVES_IN"));
    }

    #[test]
    fn test_catalog_invalid_id_lookup() {
        let catalog = Catalog::new();

        // Create one label to ensure IDs are allocated
        let _ = catalog.get_or_create_label("Person");

        // Try to look up non-existent IDs
        let invalid_label = LabelId::new(999);
        let invalid_property = PropertyKeyId::new(999);
        let invalid_edge_type = EdgeTypeId::new(999);
        let invalid_index = IndexId::new(999);

        assert!(catalog.get_label_name(invalid_label).is_none());
        assert!(catalog.get_property_key_name(invalid_property).is_none());
        assert!(catalog.get_edge_type_name(invalid_edge_type).is_none());
        assert!(catalog.get_index(invalid_index).is_none());
    }

    #[test]
    fn test_catalog_drop_nonexistent_index() {
        let catalog = Catalog::new();
        let invalid_index = IndexId::new(999);
        assert!(!catalog.drop_index(invalid_index));
    }

    #[test]
    fn test_catalog_indexes_for_nonexistent_label() {
        let catalog = Catalog::new();
        let invalid_label = LabelId::new(999);
        let invalid_property = PropertyKeyId::new(999);

        assert!(catalog.indexes_for_label(invalid_label).is_empty());
        assert!(
            catalog
                .indexes_for_label_property(invalid_label, invalid_property)
                .is_empty()
        );
    }

    #[test]
    fn test_catalog_multiple_indexes_same_property() {
        let catalog = Catalog::new();

        let person_id = catalog.get_or_create_label("Person");
        let name_id = catalog.get_or_create_property_key("name");

        // Create multiple indexes on the same property with different types
        let hash_idx = catalog.create_index(person_id, name_id, IndexType::Hash);
        let btree_idx = catalog.create_index(person_id, name_id, IndexType::BTree);
        let fulltext_idx = catalog.create_index(person_id, name_id, IndexType::FullText);

        assert_eq!(catalog.index_count(), 3);

        let indexes = catalog.indexes_for_label_property(person_id, name_id);
        assert_eq!(indexes.len(), 3);
        assert!(indexes.contains(&hash_idx));
        assert!(indexes.contains(&btree_idx));
        assert!(indexes.contains(&fulltext_idx));

        // Verify each has the correct type
        assert_eq!(
            catalog.get_index(hash_idx).unwrap().index_type,
            IndexType::Hash
        );
        assert_eq!(
            catalog.get_index(btree_idx).unwrap().index_type,
            IndexType::BTree
        );
        assert_eq!(
            catalog.get_index(fulltext_idx).unwrap().index_type,
            IndexType::FullText
        );
    }

    #[test]
    fn test_catalog_schema_required_property_duplicate() {
        let catalog = Catalog::with_schema();

        let person_id = catalog.get_or_create_label("Person");
        let name_id = catalog.get_or_create_property_key("name");

        // First should succeed
        assert!(catalog.add_required_property(person_id, name_id).is_ok());

        // Duplicate should fail
        assert_eq!(
            catalog.add_required_property(person_id, name_id),
            Err(CatalogError::ConstraintAlreadyExists)
        );
    }

    #[test]
    fn test_catalog_schema_check_without_constraints() {
        let catalog = Catalog::new();

        let person_id = catalog.get_or_create_label("Person");
        let name_id = catalog.get_or_create_property_key("name");

        // Without schema enabled, these should return false
        assert!(!catalog.is_property_unique(person_id, name_id));
        assert!(!catalog.is_property_required(person_id, name_id));
    }

    #[test]
    fn test_catalog_has_schema() {
        // Both new() and with_schema() enable schema by default
        let catalog = Catalog::new();
        assert!(catalog.has_schema());

        let with_schema = Catalog::with_schema();
        assert!(with_schema.has_schema());
    }

    #[test]
    fn test_catalog_error_display() {
        assert_eq!(
            CatalogError::SchemaNotEnabled.to_string(),
            "Schema constraints are not enabled"
        );
        assert_eq!(
            CatalogError::ConstraintAlreadyExists.to_string(),
            "Constraint already exists"
        );
        assert_eq!(
            CatalogError::LabelNotFound("Person".to_string()).to_string(),
            "Label not found: Person"
        );
        assert_eq!(
            CatalogError::PropertyKeyNotFound("name".to_string()).to_string(),
            "Property key not found: name"
        );
        assert_eq!(
            CatalogError::EdgeTypeNotFound("KNOWS".to_string()).to_string(),
            "Edge type not found: KNOWS"
        );
        let idx = IndexId::new(42);
        assert!(CatalogError::IndexNotFound(idx).to_string().contains("42"));
    }

    #[test]
    fn test_catalog_concurrent_label_creation() {
        use std::sync::Arc;

        let catalog = Arc::new(Catalog::new());
        let mut handles = vec![];

        // Spawn multiple threads trying to create the same labels
        for i in 0..10 {
            let catalog = Arc::clone(&catalog);
            handles.push(thread::spawn(move || {
                let label_name = format!("Label{}", i % 3); // Only 3 unique labels
                catalog.get_or_create_label(&label_name)
            }));
        }

        let mut ids: Vec<LabelId> = handles.into_iter().map(|h| h.join().unwrap()).collect();
        ids.sort_by_key(|id| id.as_u32());
        ids.dedup();

        // Should only have 3 unique label IDs
        assert_eq!(ids.len(), 3);
        assert_eq!(catalog.label_count(), 3);
    }

    #[test]
    fn test_catalog_concurrent_property_key_creation() {
        use std::sync::Arc;

        let catalog = Arc::new(Catalog::new());
        let mut handles = vec![];

        for i in 0..10 {
            let catalog = Arc::clone(&catalog);
            handles.push(thread::spawn(move || {
                let key_name = format!("key{}", i % 4);
                catalog.get_or_create_property_key(&key_name)
            }));
        }

        let mut ids: Vec<PropertyKeyId> = handles.into_iter().map(|h| h.join().unwrap()).collect();
        ids.sort_by_key(|id| id.as_u32());
        ids.dedup();

        assert_eq!(ids.len(), 4);
        assert_eq!(catalog.property_key_count(), 4);
    }

    #[test]
    fn test_catalog_concurrent_index_operations() {
        use std::sync::Arc;

        let catalog = Arc::new(Catalog::new());
        let label = catalog.get_or_create_label("Node");

        let mut handles = vec![];

        // Create indexes concurrently
        for i in 0..5 {
            let catalog = Arc::clone(&catalog);
            handles.push(thread::spawn(move || {
                let prop = PropertyKeyId::new(i);
                catalog.create_index(label, prop, IndexType::Hash)
            }));
        }

        let ids: Vec<IndexId> = handles.into_iter().map(|h| h.join().unwrap()).collect();
        assert_eq!(ids.len(), 5);
        assert_eq!(catalog.index_count(), 5);
    }

    #[test]
    fn test_catalog_special_characters_in_names() {
        let catalog = Catalog::new();

        // Test with various special characters
        let label1 = catalog.get_or_create_label("Label With Spaces");
        let label2 = catalog.get_or_create_label("Label-With-Dashes");
        let label3 = catalog.get_or_create_label("Label_With_Underscores");
        let label4 = catalog.get_or_create_label("LabelWithUnicode\u{00E9}");

        assert_ne!(label1, label2);
        assert_ne!(label2, label3);
        assert_ne!(label3, label4);

        assert_eq!(
            catalog.get_label_name(label1).as_deref(),
            Some("Label With Spaces")
        );
        assert_eq!(
            catalog.get_label_name(label4).as_deref(),
            Some("LabelWithUnicode\u{00E9}")
        );
    }

    #[test]
    fn test_catalog_empty_names() {
        let catalog = Catalog::new();

        // Empty names should be valid (edge case)
        let empty_label = catalog.get_or_create_label("");
        let empty_prop = catalog.get_or_create_property_key("");
        let empty_edge = catalog.get_or_create_edge_type("");

        assert_eq!(catalog.get_label_name(empty_label).as_deref(), Some(""));
        assert_eq!(
            catalog.get_property_key_name(empty_prop).as_deref(),
            Some("")
        );
        assert_eq!(catalog.get_edge_type_name(empty_edge).as_deref(), Some(""));

        // Calling again should return same ID
        assert_eq!(catalog.get_or_create_label(""), empty_label);
    }

    #[test]
    fn test_catalog_large_number_of_entries() {
        let catalog = Catalog::new();

        // Create many labels
        for i in 0..1000 {
            catalog.get_or_create_label(&format!("Label{}", i));
        }

        assert_eq!(catalog.label_count(), 1000);

        // Verify we can retrieve them all
        let all = catalog.all_labels();
        assert_eq!(all.len(), 1000);

        // Verify a specific one
        let id = catalog.get_label_id("Label500").unwrap();
        assert_eq!(catalog.get_label_name(id).as_deref(), Some("Label500"));
    }

    #[test]
    fn test_index_definition_debug() {
        let def = IndexDefinition {
            id: IndexId::new(1),
            label: LabelId::new(2),
            property_key: PropertyKeyId::new(3),
            index_type: IndexType::Hash,
        };

        // Should be able to debug print
        let debug_str = format!("{:?}", def);
        assert!(debug_str.contains("IndexDefinition"));
        assert!(debug_str.contains("Hash"));
    }

    #[test]
    fn test_index_type_equality() {
        assert_eq!(IndexType::Hash, IndexType::Hash);
        assert_ne!(IndexType::Hash, IndexType::BTree);
        assert_ne!(IndexType::BTree, IndexType::FullText);

        // Clone
        let t = IndexType::Hash;
        let t2 = t;
        assert_eq!(t, t2);
    }

    #[test]
    fn test_catalog_error_equality() {
        assert_eq!(
            CatalogError::SchemaNotEnabled,
            CatalogError::SchemaNotEnabled
        );
        assert_eq!(
            CatalogError::ConstraintAlreadyExists,
            CatalogError::ConstraintAlreadyExists
        );
        assert_eq!(
            CatalogError::LabelNotFound("X".to_string()),
            CatalogError::LabelNotFound("X".to_string())
        );
        assert_ne!(
            CatalogError::LabelNotFound("X".to_string()),
            CatalogError::LabelNotFound("Y".to_string())
        );
    }
}