mecha10_core/

schema.rs

//! Schema Registry
//!
//! Provides versioned message schema storage and retrieval.
//! Enables schema evolution and documentation of message types.
//!
//! # Architecture
//!
//! - Stores schemas in Redis with versioning
//! - Each schema has a name and version number
//! - Schemas stored as JSON Schema format
//! - Supports listing all schemas and retrieving specific versions
//!
//! # Example
//!
//! ```rust
//! use mecha10::prelude::*;
//! use mecha10::schema::{SchemaRegistryExt, MessageSchema};
//!
//! # async fn example(ctx: &Context) -> Result<()> {
//! // Register a schema
//! let schema = MessageSchema::new("LaserScan", 1)
//!     .with_description("2D laser scan data")
//!     .with_field("timestamp", "integer", true)
//!     .with_field("ranges", "array", true)
//!     .with_field("intensities", "array", false);
//!
//! ctx.register_schema(&schema).await?;
//!
//! // Get latest schema
//! let schema = ctx.get_latest_schema("LaserScan").await?;
//!
//! // Get specific version
//! let schema_v1 = ctx.get_schema("LaserScan", 1).await?;
//!
//! // List all schemas
//! let all_schemas = ctx.list_schemas().await?;
//!
//! # Ok(())
//! # }
//! ```

use crate::context::Context;
use crate::error::{Mecha10Error, Result};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// Message schema definition
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageSchema {
    /// Schema name (typically the message type name)
    pub name: String,

    /// Schema version (incremental)
    pub version: u32,

    /// Human-readable description
    #[serde(default)]
    pub description: String,

    /// Schema fields definition
    pub fields: Vec<SchemaField>,

    /// Additional metadata
    #[serde(default)]
    pub metadata: HashMap<String, String>,

    /// When this schema was registered (Unix timestamp)
    #[serde(default)]
    pub registered_at: u64,
}

/// Schema field definition
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaField {
    /// Field name
    pub name: String,

    /// Field type (string, integer, number, boolean, array, object)
    pub field_type: String,

    /// Whether this field is required
    pub required: bool,

    /// Field description
    #[serde(default)]
    pub description: String,

    /// For array types, the item type
    #[serde(skip_serializing_if = "Option::is_none")]
    pub items: Option<String>,
}

impl MessageSchema {
    /// Create a new schema
    pub fn new(name: impl Into<String>, version: u32) -> Self {
        Self {
            name: name.into(),
            version,
            description: String::new(),
            fields: Vec::new(),
            metadata: HashMap::new(),
            registered_at: current_timestamp(),
        }
    }

    /// Set description
    pub fn with_description(mut self, description: impl Into<String>) -> Self {
        self.description = description.into();
        self
    }

    /// Add a field
    pub fn with_field(mut self, name: impl Into<String>, field_type: impl Into<String>, required: bool) -> Self {
        self.fields.push(SchemaField {
            name: name.into(),
            field_type: field_type.into(),
            required,
            description: String::new(),
            items: None,
        });
        self
    }

    /// Add an array field with item type
    pub fn with_array_field(mut self, name: impl Into<String>, items: impl Into<String>, required: bool) -> Self {
        self.fields.push(SchemaField {
            name: name.into(),
            field_type: "array".to_string(),
            required,
            description: String::new(),
            items: Some(items.into()),
        });
        self
    }

    /// Add metadata
    pub fn with_metadata(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.metadata.insert(key.into(), value.into());
        self
    }

    /// Get the schema key for Redis
    fn redis_key(&self) -> String {
        format!("mecha10:schemas:{}:{}", self.name, self.version)
    }

    /// Get the latest version key for Redis
    fn redis_latest_key(name: &str) -> String {
        format!("mecha10:schemas:{}:latest", name)
    }

    /// Get the schema list key for Redis
    fn redis_list_key() -> String {
        "mecha10:schemas:list".to_string()
    }
}

/// Schema registry extension trait for Context
pub trait SchemaRegistryExt {
    /// Register a schema in the registry
    ///
    /// # Arguments
    ///
    /// * `schema` - Schema to register
    ///
    /// # Example
    ///
    /// ```rust
    /// use mecha10::prelude::*;
    /// use mecha10::schema::{SchemaRegistryExt, MessageSchema};
    ///
    /// # async fn example(ctx: &Context) -> Result<()> {
    /// let schema = MessageSchema::new("LaserScan", 1)
    ///     .with_description("2D laser scan data")
    ///     .with_field("timestamp", "integer", true)
    ///     .with_field("ranges", "array", true);
    ///
    /// ctx.register_schema(&schema).await?;
    /// # Ok(())
    /// # }
    /// ```
    fn register_schema(&self, schema: &MessageSchema) -> impl std::future::Future<Output = Result<()>> + Send;

    /// Get a specific version of a schema
    ///
    /// # Arguments
    ///
    /// * `name` - Schema name
    /// * `version` - Schema version
    fn get_schema(
        &self,
        name: &str,
        version: u32,
    ) -> impl std::future::Future<Output = Result<Option<MessageSchema>>> + Send;

    /// Get the latest version of a schema
    ///
    /// # Arguments
    ///
    /// * `name` - Schema name
    fn get_latest_schema(&self, name: &str) -> impl std::future::Future<Output = Result<Option<MessageSchema>>> + Send;

    /// List all registered schemas (returns latest version of each)
    fn list_schemas(&self) -> impl std::future::Future<Output = Result<Vec<MessageSchema>>> + Send;

    /// Get all versions of a schema
    ///
    /// # Arguments
    ///
    /// * `name` - Schema name
    fn get_schema_versions(&self, name: &str) -> impl std::future::Future<Output = Result<Vec<MessageSchema>>> + Send;

    /// Delete a schema version
    ///
    /// # Arguments
    ///
    /// * `name` - Schema name
    /// * `version` - Schema version (if None, deletes all versions)
    fn delete_schema(&self, name: &str, version: Option<u32>) -> impl std::future::Future<Output = Result<()>> + Send;
}

impl SchemaRegistryExt for Context {
    async fn register_schema(&self, schema: &MessageSchema) -> Result<()> {
        #[cfg(feature = "messaging")]
        {
            use redis::AsyncCommands;

            let redis_url = Context::get_redis_url()?;
            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
                message: format!("Failed to connect to Redis: {}", e),
                suggestion: "Ensure Redis is running".to_string(),
            })?;

            let mut conn =
                client
                    .get_multiplexed_async_connection()
                    .await
                    .map_err(|e| Mecha10Error::MessagingError {
                        message: format!("Failed to get Redis connection: {}", e),
                        suggestion: "Ensure Redis is running".to_string(),
                    })?;

            // Serialize schema
            let schema_json = serde_json::to_string(&schema)
                .map_err(|e| Mecha10Error::Other(format!("Failed to serialize schema: {}", e)))?;

            // Store schema with version
            let key = schema.redis_key();
            conn.set::<_, _, ()>(&key, &schema_json)
                .await
                .map_err(|e| Mecha10Error::MessagingError {
                    message: format!("Failed to register schema: {}", e),
                    suggestion: "Check Redis connection".to_string(),
                })?;

            // Update latest version pointer
            let latest_key = MessageSchema::redis_latest_key(&schema.name);
            conn.set::<_, _, ()>(&latest_key, schema.version)
                .await
                .map_err(|e| Mecha10Error::MessagingError {
                    message: format!("Failed to update latest version: {}", e),
                    suggestion: "Check Redis connection".to_string(),
                })?;

            // Add to schema list
            let list_key = MessageSchema::redis_list_key();
            conn.sadd::<_, _, ()>(&list_key, &schema.name)
                .await
                .map_err(|e| Mecha10Error::MessagingError {
                    message: format!("Failed to update schema list: {}", e),
                    suggestion: "Check Redis connection".to_string(),
                })?;

            Ok(())
        }

        #[cfg(not(feature = "messaging"))]
        {
            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
        }
    }

    async fn get_schema(&self, name: &str, version: u32) -> Result<Option<MessageSchema>> {
        #[cfg(feature = "messaging")]
        {
            use redis::AsyncCommands;

            let redis_url = Context::get_redis_url()?;
            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
                message: format!("Failed to connect to Redis: {}", e),
                suggestion: "Ensure Redis is running".to_string(),
            })?;

            let mut conn =
                client
                    .get_multiplexed_async_connection()
                    .await
                    .map_err(|e| Mecha10Error::MessagingError {
                        message: format!("Failed to get Redis connection: {}", e),
                        suggestion: "Ensure Redis is running".to_string(),
                    })?;

            let key = format!("mecha10:schemas:{}:{}", name, version);
            let json: Option<String> = conn.get(&key).await.map_err(|e| Mecha10Error::MessagingError {
                message: format!("Failed to get schema: {}", e),
                suggestion: "Check Redis connection".to_string(),
            })?;

            if let Some(json) = json {
                let schema = serde_json::from_str::<MessageSchema>(&json)
                    .map_err(|e| Mecha10Error::Other(format!("Failed to parse schema: {}", e)))?;
                Ok(Some(schema))
            } else {
                Ok(None)
            }
        }

        #[cfg(not(feature = "messaging"))]
        {
            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
        }
    }

    async fn get_latest_schema(&self, name: &str) -> Result<Option<MessageSchema>> {
        #[cfg(feature = "messaging")]
        {
            use redis::AsyncCommands;

            let redis_url = Context::get_redis_url()?;
            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
                message: format!("Failed to connect to Redis: {}", e),
                suggestion: "Ensure Redis is running".to_string(),
            })?;

            let mut conn =
                client
                    .get_multiplexed_async_connection()
                    .await
                    .map_err(|e| Mecha10Error::MessagingError {
                        message: format!("Failed to get Redis connection: {}", e),
                        suggestion: "Ensure Redis is running".to_string(),
                    })?;

            // Get latest version number
            let latest_key = MessageSchema::redis_latest_key(name);
            let version: Option<u32> = conn.get(&latest_key).await.map_err(|e| Mecha10Error::MessagingError {
                message: format!("Failed to get latest version: {}", e),
                suggestion: "Check Redis connection".to_string(),
            })?;

            if let Some(version) = version {
                self.get_schema(name, version).await
            } else {
                Ok(None)
            }
        }

        #[cfg(not(feature = "messaging"))]
        {
            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
        }
    }

    async fn list_schemas(&self) -> Result<Vec<MessageSchema>> {
        #[cfg(feature = "messaging")]
        {
            use redis::AsyncCommands;

            let redis_url = Context::get_redis_url()?;
            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
                message: format!("Failed to connect to Redis: {}", e),
                suggestion: "Ensure Redis is running".to_string(),
            })?;

            let mut conn =
                client
                    .get_multiplexed_async_connection()
                    .await
                    .map_err(|e| Mecha10Error::MessagingError {
                        message: format!("Failed to get Redis connection: {}", e),
                        suggestion: "Ensure Redis is running".to_string(),
                    })?;

            // Get all schema names
            let list_key = MessageSchema::redis_list_key();
            let names: Vec<String> = conn
                .smembers(&list_key)
                .await
                .map_err(|e| Mecha10Error::MessagingError {
                    message: format!("Failed to get schema list: {}", e),
                    suggestion: "Check Redis connection".to_string(),
                })?;

            // Get latest version of each schema
            let mut schemas = Vec::new();
            for name in names {
                if let Some(schema) = self.get_latest_schema(&name).await? {
                    schemas.push(schema);
                }
            }

            Ok(schemas)
        }

        #[cfg(not(feature = "messaging"))]
        {
            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
        }
    }

    async fn get_schema_versions(&self, name: &str) -> Result<Vec<MessageSchema>> {
        #[cfg(feature = "messaging")]
        {
            use redis::AsyncCommands;

            let redis_url = Context::get_redis_url()?;
            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
                message: format!("Failed to connect to Redis: {}", e),
                suggestion: "Ensure Redis is running".to_string(),
            })?;

            let mut conn =
                client
                    .get_multiplexed_async_connection()
                    .await
                    .map_err(|e| Mecha10Error::MessagingError {
                        message: format!("Failed to get Redis connection: {}", e),
                        suggestion: "Ensure Redis is running".to_string(),
                    })?;

            // Scan for all versions of this schema
            let pattern = format!("mecha10:schemas:{}:*", name);
            let keys: Vec<String> = conn.keys(&pattern).await.map_err(|e| Mecha10Error::MessagingError {
                message: format!("Failed to scan for schema versions: {}", e),
                suggestion: "Check Redis connection".to_string(),
            })?;

            let mut schemas = Vec::new();
            for key in keys {
                // Skip the 'latest' key
                if key.ends_with(":latest") {
                    continue;
                }

                let json: String = conn.get(&key).await.map_err(|e| Mecha10Error::MessagingError {
                    message: format!("Failed to get schema: {}", e),
                    suggestion: "Check Redis connection".to_string(),
                })?;

                if let Ok(schema) = serde_json::from_str::<MessageSchema>(&json) {
                    schemas.push(schema);
                }
            }

            // Sort by version
            schemas.sort_by_key(|s| s.version);

            Ok(schemas)
        }

        #[cfg(not(feature = "messaging"))]
        {
            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
        }
    }

    async fn delete_schema(&self, name: &str, version: Option<u32>) -> Result<()> {
        #[cfg(feature = "messaging")]
        {
            use redis::AsyncCommands;

            let redis_url = Context::get_redis_url()?;
            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
                message: format!("Failed to connect to Redis: {}", e),
                suggestion: "Ensure Redis is running".to_string(),
            })?;

            let mut conn =
                client
                    .get_multiplexed_async_connection()
                    .await
                    .map_err(|e| Mecha10Error::MessagingError {
                        message: format!("Failed to get Redis connection: {}", e),
                        suggestion: "Ensure Redis is running".to_string(),
                    })?;

            if let Some(version) = version {
                // Delete specific version
                let key = format!("mecha10:schemas:{}:{}", name, version);
                conn.del::<_, ()>(&key)
                    .await
                    .map_err(|e| Mecha10Error::MessagingError {
                        message: format!("Failed to delete schema: {}", e),
                        suggestion: "Check Redis connection".to_string(),
                    })?;
            } else {
                // Delete all versions
                let pattern = format!("mecha10:schemas:{}:*", name);
                let keys: Vec<String> = conn.keys(&pattern).await.map_err(|e| Mecha10Error::MessagingError {
                    message: format!("Failed to scan for schemas: {}", e),
                    suggestion: "Check Redis connection".to_string(),
                })?;

                for key in keys {
                    conn.del::<_, ()>(&key)
                        .await
                        .map_err(|e| Mecha10Error::MessagingError {
                            message: format!("Failed to delete schema: {}", e),
                            suggestion: "Check Redis connection".to_string(),
                        })?;
                }

                // Remove from schema list
                let list_key = MessageSchema::redis_list_key();
                conn.srem::<_, _, ()>(&list_key, name)
                    .await
                    .map_err(|e| Mecha10Error::MessagingError {
                        message: format!("Failed to update schema list: {}", e),
                        suggestion: "Check Redis connection".to_string(),
                    })?;
            }

            Ok(())
        }

        #[cfg(not(feature = "messaging"))]
        {
            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
        }
    }
}

/// Get current Unix timestamp in seconds
fn current_timestamp() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_secs()
}

// ============================================================================
// Message Versioning and Compatibility
// ============================================================================

/// Schema compatibility mode
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
pub enum CompatibilityMode {
    /// No compatibility checking
    None,

    /// New schema can read data written by old schema (backward compatible)
    #[default]
    Backward,

    /// Old schema can read data written by new schema (forward compatible)
    Forward,

    /// Both backward and forward compatible
    Full,

    /// No schema changes allowed
    Strict,
}

/// Schema evolution operation
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum SchemaEvolution {
    /// Field added (name, type, optional/required, default_value)
    FieldAdded {
        name: String,
        field_type: String,
        required: bool,
        default_value: Option<serde_json::Value>,
    },

    /// Field removed
    FieldRemoved { name: String },

    /// Field made optional (was required)
    FieldMadeOptional { name: String },

    /// Field made required (was optional) - requires default value
    FieldMadeRequired {
        name: String,
        default_value: serde_json::Value,
    },

    /// Field type changed
    FieldTypeChanged {
        name: String,
        old_type: String,
        new_type: String,
    },

    /// Field renamed
    FieldRenamed { old_name: String, new_name: String },
}

/// Schema compatibility check result
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CompatibilityCheck {
    /// Whether schemas are compatible
    pub compatible: bool,

    /// List of incompatibilities found
    pub incompatibilities: Vec<String>,

    /// List of evolutions detected
    pub evolutions: Vec<SchemaEvolution>,

    /// Warnings (non-breaking but notable changes)
    pub warnings: Vec<String>,
}

impl CompatibilityCheck {
    /// Check if two schemas are compatible
    pub fn check(old: &MessageSchema, new: &MessageSchema, mode: CompatibilityMode) -> Self {
        let mut incompatibilities = Vec::new();
        let mut evolutions = Vec::new();
        let mut warnings = Vec::new();

        // Build field maps for easier comparison
        let old_fields: HashMap<String, &SchemaField> = old.fields.iter().map(|f| (f.name.clone(), f)).collect();
        let new_fields: HashMap<String, &SchemaField> = new.fields.iter().map(|f| (f.name.clone(), f)).collect();

        match mode {
            CompatibilityMode::None => {
                // No checking required
                return Self {
                    compatible: true,
                    incompatibilities,
                    evolutions,
                    warnings,
                };
            }
            CompatibilityMode::Strict => {
                // No changes allowed
                if old.fields.len() != new.fields.len() {
                    incompatibilities.push("Field count changed".to_string());
                }
                for (name, old_field) in &old_fields {
                    if let Some(new_field) = new_fields.get(name) {
                        if old_field.field_type != new_field.field_type {
                            incompatibilities.push(format!("Field '{}' type changed", name));
                        }
                        if old_field.required != new_field.required {
                            incompatibilities.push(format!("Field '{}' required status changed", name));
                        }
                    } else {
                        incompatibilities.push(format!("Field '{}' removed", name));
                    }
                }
                for name in new_fields.keys() {
                    if !old_fields.contains_key(name) {
                        incompatibilities.push(format!("Field '{}' added", name));
                    }
                }
            }
            CompatibilityMode::Backward => {
                // New schema can read old data
                // - Can add optional fields
                // - Can make required fields optional
                // - Cannot remove required fields
                // - Cannot add required fields
                // - Cannot change field types

                for (name, old_field) in &old_fields {
                    if let Some(new_field) = new_fields.get(name) {
                        // Field exists in both
                        if old_field.field_type != new_field.field_type {
                            incompatibilities.push(format!(
                                "Field '{}' type changed from {} to {}",
                                name, old_field.field_type, new_field.field_type
                            ));
                            evolutions.push(SchemaEvolution::FieldTypeChanged {
                                name: name.clone(),
                                old_type: old_field.field_type.clone(),
                                new_type: new_field.field_type.clone(),
                            });
                        }

                        if old_field.required && !new_field.required {
                            // Field made optional - OK for backward compatibility
                            evolutions.push(SchemaEvolution::FieldMadeOptional { name: name.clone() });
                        } else if !old_field.required && new_field.required {
                            // Field made required - breaks backward compatibility
                            incompatibilities.push(format!("Field '{}' made required (was optional)", name));
                            evolutions.push(SchemaEvolution::FieldMadeRequired {
                                name: name.clone(),
                                default_value: serde_json::Value::Null,
                            });
                        }
                    } else {
                        // Field removed
                        if old_field.required {
                            incompatibilities.push(format!("Required field '{}' removed", name));
                        } else {
                            warnings.push(format!("Optional field '{}' removed", name));
                        }
                        evolutions.push(SchemaEvolution::FieldRemoved { name: name.clone() });
                    }
                }

                // Check for added fields
                for (name, new_field) in &new_fields {
                    if !old_fields.contains_key(name) {
                        if new_field.required {
                            incompatibilities.push(format!("Required field '{}' added", name));
                        } else {
                            warnings.push(format!("Optional field '{}' added", name));
                        }
                        evolutions.push(SchemaEvolution::FieldAdded {
                            name: name.clone(),
                            field_type: new_field.field_type.clone(),
                            required: new_field.required,
                            default_value: None,
                        });
                    }
                }
            }
            CompatibilityMode::Forward => {
                // Old schema can read new data
                // - Can remove optional fields
                // - Can make optional fields required
                // - Cannot add required fields
                // - Cannot remove required fields
                // - Cannot change field types

                for (name, old_field) in &old_fields {
                    if let Some(new_field) = new_fields.get(name) {
                        if old_field.field_type != new_field.field_type {
                            incompatibilities.push(format!("Field '{}' type changed", name));
                            evolutions.push(SchemaEvolution::FieldTypeChanged {
                                name: name.clone(),
                                old_type: old_field.field_type.clone(),
                                new_type: new_field.field_type.clone(),
                            });
                        }
                    } else if old_field.required {
                        incompatibilities.push(format!("Required field '{}' removed", name));
                        evolutions.push(SchemaEvolution::FieldRemoved { name: name.clone() });
                    }
                }

                for (name, new_field) in &new_fields {
                    if !old_fields.contains_key(name) && new_field.required {
                        incompatibilities.push(format!("Required field '{}' added", name));
                        evolutions.push(SchemaEvolution::FieldAdded {
                            name: name.clone(),
                            field_type: new_field.field_type.clone(),
                            required: new_field.required,
                            default_value: None,
                        });
                    }
                }
            }
            CompatibilityMode::Full => {
                // Both backward and forward compatible
                // - Can only add/remove optional fields
                // - Cannot change field types
                // - Cannot change required status

                for (name, old_field) in &old_fields {
                    if let Some(new_field) = new_fields.get(name) {
                        if old_field.field_type != new_field.field_type {
                            incompatibilities.push(format!("Field '{}' type changed", name));
                        }
                        if old_field.required != new_field.required {
                            incompatibilities.push(format!("Field '{}' required status changed", name));
                        }
                    } else if old_field.required {
                        incompatibilities.push(format!("Required field '{}' removed", name));
                    }
                }

                for (name, new_field) in &new_fields {
                    if !old_fields.contains_key(name) && new_field.required {
                        incompatibilities.push(format!("Required field '{}' added", name));
                    }
                }
            }
        }

        Self {
            compatible: incompatibilities.is_empty(),
            incompatibilities,
            evolutions,
            warnings,
        }
    }
}

/// Message version metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageVersionInfo {
    /// Schema name
    pub name: String,

    /// Current version being used
    pub current_version: u32,

    /// Latest available version
    pub latest_version: u32,

    /// Whether current version is deprecated
    pub deprecated: bool,

    /// Deprecation message if applicable
    pub deprecation_message: Option<String>,

    /// Minimum supported version
    pub min_supported_version: u32,

    /// Compatibility mode for this message type
    pub compatibility_mode: CompatibilityMode,
}

/// Extension trait for versioned schema operations
pub trait SchemaVersioningExt: SchemaRegistryExt {
    /// Check compatibility between two schema versions
    fn check_compatibility(
        &self,
        name: &str,
        old_version: u32,
        new_version: u32,
        mode: CompatibilityMode,
    ) -> impl std::future::Future<Output = Result<CompatibilityCheck>> + Send;

    /// Register a new schema version with compatibility check
    fn register_versioned_schema(
        &self,
        schema: &MessageSchema,
        mode: CompatibilityMode,
    ) -> impl std::future::Future<Output = Result<CompatibilityCheck>> + Send;

    /// Get version info for a message type
    fn get_version_info(
        &self,
        name: &str,
    ) -> impl std::future::Future<Output = Result<Option<MessageVersionInfo>>> + Send;

    /// Mark a schema version as deprecated
    fn deprecate_version(
        &self,
        name: &str,
        version: u32,
        message: String,
    ) -> impl std::future::Future<Output = Result<()>> + Send;

    /// Check if using a deprecated version
    fn is_version_deprecated(&self, name: &str, version: u32)
        -> impl std::future::Future<Output = Result<bool>> + Send;
}

impl SchemaVersioningExt for Context {
    async fn check_compatibility(
        &self,
        name: &str,
        old_version: u32,
        new_version: u32,
        mode: CompatibilityMode,
    ) -> Result<CompatibilityCheck> {
        let old_schema = self
            .get_schema(name, old_version)
            .await?
            .ok_or_else(|| Mecha10Error::Other(format!("Schema {} v{} not found", name, old_version)))?;

        let new_schema = self
            .get_schema(name, new_version)
            .await?
            .ok_or_else(|| Mecha10Error::Other(format!("Schema {} v{} not found", name, new_version)))?;

        Ok(CompatibilityCheck::check(&old_schema, &new_schema, mode))
    }

    async fn register_versioned_schema(
        &self,
        schema: &MessageSchema,
        mode: CompatibilityMode,
    ) -> Result<CompatibilityCheck> {
        // Get previous version if exists
        if schema.version > 1 {
            let prev_version = schema.version - 1;
            let check = self
                .check_compatibility(&schema.name, prev_version, schema.version, mode)
                .await?;

            if !check.compatible {
                return Err(Mecha10Error::Other(format!(
                    "Schema compatibility check failed: {:?}",
                    check.incompatibilities
                )));
            }

            // Register the schema
            self.register_schema(schema).await?;

            Ok(check)
        } else {
            // First version - no compatibility check needed
            self.register_schema(schema).await?;

            Ok(CompatibilityCheck {
                compatible: true,
                incompatibilities: Vec::new(),
                evolutions: Vec::new(),
                warnings: Vec::new(),
            })
        }
    }

    async fn get_version_info(&self, name: &str) -> Result<Option<MessageVersionInfo>> {
        let latest = self.get_latest_schema(name).await?;

        if let Some(latest_schema) = latest {
            // Check for deprecation metadata
            let deprecated = latest_schema.metadata.contains_key("deprecated");
            let deprecation_message = latest_schema.metadata.get("deprecation_message").cloned();
            let min_supported = latest_schema
                .metadata
                .get("min_supported_version")
                .and_then(|v| v.parse().ok())
                .unwrap_or(1);

            let compatibility_mode = latest_schema
                .metadata
                .get("compatibility_mode")
                .and_then(|m| serde_json::from_str(m).ok())
                .unwrap_or(CompatibilityMode::Backward);

            Ok(Some(MessageVersionInfo {
                name: name.to_string(),
                current_version: latest_schema.version,
                latest_version: latest_schema.version,
                deprecated,
                deprecation_message,
                min_supported_version: min_supported,
                compatibility_mode,
            }))
        } else {
            Ok(None)
        }
    }

    async fn deprecate_version(&self, name: &str, version: u32, message: String) -> Result<()> {
        let mut schema = self
            .get_schema(name, version)
            .await?
            .ok_or_else(|| Mecha10Error::Other(format!("Schema {} v{} not found", name, version)))?;

        schema.metadata.insert("deprecated".to_string(), "true".to_string());
        schema.metadata.insert("deprecation_message".to_string(), message);

        self.register_schema(&schema).await
    }

    async fn is_version_deprecated(&self, name: &str, version: u32) -> Result<bool> {
        let schema = self
            .get_schema(name, version)
            .await?
            .ok_or_else(|| Mecha10Error::Other(format!("Schema {} v{} not found", name, version)))?;

        Ok(schema.metadata.contains_key("deprecated"))
    }
}