Implement Schema Management Service with Migration Infrastructure #106

// In create_node() method
pub async fn create_node(&self, params: CreateNodeParams) -> Result<Node, ServiceError> {
    // ... existing validation ...

    // NEW: Add schema version tracking
    let mut properties = params.properties.unwrap_or_else(|| json!({}));

    // Get current schema version for this node type
    if let Some(schema) = self.get_schema_for_type(&params.node_type).await? {
        if let Some(obj) = properties.as_object_mut() {
            obj.insert(
                "_schema_version".to_string(),
                json!(schema.version)
            );
        }
    } else {
        // No schema defined - default to version 1
        if let Some(obj) = properties.as_object_mut() {
            obj.insert("_schema_version".to_string(), json!(1));
        }
    }

    // ... rest of create logic ...
}

// Add new helper method to NodeService
impl NodeService {
    /// Get schema definition for a given node type
    async fn get_schema_for_type(&self, node_type: &str) -> Result<Option<SchemaDefinition>, ServiceError> {
        // Schema lookup convention: id = type_name, node_type = "schema"
        let schema_node = self.get_node(node_type).await?;

        if let Some(node) = schema_node {
            if node.node_type == "schema" {
                let schema: SchemaDefinition = serde_json::from_value(node.properties)?;
                return Ok(Some(schema));
            }
        }

        Ok(None)
    }
}

// In get_node() method - add backfill logic
pub async fn get_node(&self, id: &str) -> Result<Option<Node>, ServiceError> {
    let mut node = self.db.get_node(id).await?;

    if let Some(ref mut node) = node {
        // NEW: Backfill nodes without version
        if let Some(props) = node.properties.as_object() {
            if !props.contains_key("_schema_version") {
                // No version - backfill with v1
                if let Some(obj) = node.properties.as_object_mut() {
                    obj.insert("_schema_version".to_string(), json!(1));

                    // Persist the backfilled version
                    self.db.update_node(node).await?;
                }
            }
        }
    }

    Ok(node)
}

#[tokio::test]
async fn test_new_nodes_get_schema_version() {
    let service = setup_test_service().await;

    let node = service.create_node(CreateNodeParams {
        node_type: "task".to_string(),
        content: "Test task".to_string(),
        properties: Some(json!({"status": "OPEN"})),
        ..Default::default()
    }).await.unwrap();

    assert!(node.properties["_schema_version"].is_number());
    assert_eq!(node.properties["_schema_version"], 1);
}

#[tokio::test]
async fn test_existing_nodes_backfilled_with_version() {
    let service = setup_test_service().await;

    // Manually create node without version (simulating old data)
    let node_id = service.db.execute(
        "INSERT INTO nodes (id, node_type, content, properties) VALUES (?, ?, ?, ?)",
        params!["test-node", "task", "Old task", json!({"status": "OPEN"})]
    ).await.unwrap();

    // Fetch the node (should trigger backfill)
    let node = service.get_node("test-node").await.unwrap().unwrap();

    // Verify version was backfilled
    assert_eq!(node.properties["_schema_version"], 1);
}

use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum ProtectionLevel {
    Core,   // Cannot modify/delete
    User,   // Fully modifiable
    System, // Auto-managed, read-only
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaField {
    pub name: String,
    #[serde(rename = "type")]
    pub field_type: String,
    pub protection: ProtectionLevel,

    // Enum-specific
    #[serde(skip_serializing_if = "Option::is_none")]
    pub core_values: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub user_values: Option<Vec<String>>,

    pub indexed: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub required: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub extensible: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub default: Option<serde_json::Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub item_type: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaDefinition {
    pub is_core: bool,
    pub version: u32,
    pub description: String,
    pub fields: Vec<SchemaField>,
}

impl SchemaDefinition {
    /// Get all enum values (core + user)
    pub fn get_enum_values(&self, field_name: &str) -> Option<Vec<String>> {
        let field = self.fields.iter().find(|f| f.name == field_name)?;

        let mut values = Vec::new();
        if let Some(core_vals) = &field.core_values {
            values.extend(core_vals.clone());
        }
        if let Some(user_vals) = &field.user_values {
            values.extend(user_vals.clone());
        }

        Some(values)
    }

    /// Check if a field can be deleted
    pub fn can_delete_field(&self, field_name: &str) -> bool {
        if let Some(field) = self.fields.iter().find(|f| f.name == field_name) {
            field.protection == ProtectionLevel::User
        } else {
            false
        }
    }
}

use crate::models::node::Node;
use crate::models::schema::{SchemaDefinition, SchemaField, ProtectionLevel};
use crate::services::node_service::NodeService;
use anyhow::{anyhow, Result};
use serde_json::json;
use std::sync::Arc;

pub struct SchemaService {
    node_service: Arc<NodeService>,
}

impl SchemaService {
    pub fn new(node_service: Arc<NodeService>) -> Self {
        Self { node_service }
    }

    /// Get schema definition by schema ID (convention: id = type_name)
    pub async fn get_schema(&self, schema_id: &str) -> Result<SchemaDefinition> {
        let node = self.node_service
            .get_node(schema_id)
            .await?
            .ok_or_else(|| anyhow!("Schema '{}' not found", schema_id))?;

        if node.node_type != "schema" {
            return Err(anyhow!("Node '{}' is not a schema (type: {})", schema_id, node.node_type));
        }

        let schema: SchemaDefinition = serde_json::from_value(node.properties)?;
        Ok(schema)
    }

    /// Add a new field to a schema (only user-protected fields allowed)
    pub async fn add_field(&self, schema_id: &str, field: SchemaField) -> Result<()> {
        // Protection: Only user fields can be added
        if field.protection != ProtectionLevel::User {
            return Err(anyhow!(
                "Can only add user-protected fields. Field '{}' has protection: {:?}",
                field.name, field.protection
            ));
        }

        let mut schema = self.get_schema(schema_id).await?;

        // Check if field already exists
        if schema.fields.iter().any(|f| f.name == field.name) {
            return Err(anyhow!("Field '{}' already exists in schema '{}'", field.name, schema_id));
        }

        // Add field and bump version
        schema.fields.push(field);
        schema.version += 1;

        // Update schema node
        self.update_schema(schema_id, schema).await?;

        Ok(())
    }

    /// Remove a field from schema (only user-protected fields can be removed)
    pub async fn remove_field(&self, schema_id: &str, field_name: &str) -> Result<()> {
        let mut schema = self.get_schema(schema_id).await?;

        // Find the field
        let field_idx = schema.fields.iter()
            .position(|f| f.name == field_name)
            .ok_or_else(|| anyhow!("Field '{}' not found in schema '{}'", field_name, schema_id))?;

        let field = &schema.fields[field_idx];

        // Protection: Only user fields can be removed
        if field.protection != ProtectionLevel::User {
            return Err(anyhow!(
                "Cannot remove field '{}' with protection level {:?}. Only user fields can be removed.",
                field_name, field.protection
            ));
        }

        // Remove field and bump version
        schema.fields.remove(field_idx);
        schema.version += 1;

        // Update schema node
        self.update_schema(schema_id, schema).await?;

        Ok(())
    }

    /// Extend an enum field with a new value (adds to user_values only)
    pub async fn extend_enum_field(&self, schema_id: &str, field_name: &str, new_value: String) -> Result<()> {
        let mut schema = self.get_schema(schema_id).await?;

        // Find the field
        let field = schema.fields.iter_mut()
            .find(|f| f.name == field_name)
            .ok_or_else(|| anyhow!("Field '{}' not found in schema '{}'", field_name, schema_id))?;

        // Verify it's an enum field
        if field.field_type != "enum" {
            return Err(anyhow!("Field '{}' is not an enum (type: {})", field_name, field.field_type));
        }

        // Verify it's extensible
        if !field.extensible.unwrap_or(false) {
            return Err(anyhow!("Enum field '{}' is not extensible", field_name));
        }

        // Check if value already exists (in core or user values)
        let all_values = schema.get_enum_values(field_name).unwrap_or_default();
        if all_values.contains(&new_value) {
            return Err(anyhow!("Value '{}' already exists in enum '{}'", new_value, field_name));
        }

        // Add to user_values
        field.user_values.get_or_insert_with(Vec::new).push(new_value);
        schema.version += 1;

        // Update schema node
        self.update_schema(schema_id, schema).await?;

        Ok(())
    }

    /// Remove a value from enum field (only from user_values, cannot remove core_values)
    pub async fn remove_enum_value(&self, schema_id: &str, field_name: &str, value: &str) -> Result<()> {
        let mut schema = self.get_schema(schema_id).await?;

        // Find the field
        let field = schema.fields.iter_mut()
            .find(|f| f.name == field_name)
            .ok_or_else(|| anyhow!("Field '{}' not found in schema '{}'", field_name, schema_id))?;

        // Check if value is in core_values (cannot remove)
        if let Some(core_vals) = &field.core_values {
            if core_vals.contains(&value.to_string()) {
                return Err(anyhow!(
                    "Cannot remove core value '{}' from enum '{}'. Only user values can be removed.",
                    value, field_name
                ));
            }
        }

        // Remove from user_values
        if let Some(user_vals) = &mut field.user_values {
            if let Some(idx) = user_vals.iter().position(|v| v == value) {
                user_vals.remove(idx);
                schema.version += 1;

                // Update schema node
                self.update_schema(schema_id, schema).await?;
                return Ok(());
            }
        }

        Err(anyhow!("Value '{}' not found in user values of enum '{}'", value, field_name))
    }

    /// Update schema node with new definition
    async fn update_schema(&self, schema_id: &str, schema: SchemaDefinition) -> Result<()> {
        let properties = serde_json::to_value(&schema)?;

        self.node_service.update_node(UpdateNodeParams {
            id: schema_id.to_string(),
            properties: Some(properties),
            ..Default::default()
        }).await?;

        Ok(())
    }
}