mecha10-core 0.6.2

//! `Context`'s Redis-backed implementation of [`super::SchemaRegistryExt`].

use super::{MessageSchema, SchemaRegistryExt};
use crate::context::Context;
use crate::error::{Mecha10Error, Result};

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")]
        {
            crate::redis_list_util::collect_from_redis_set(&MessageSchema::redis_list_key(), |name| async move {
                self.get_latest_schema(&name).await
            })
            .await
        }

        #[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()))
        }
    }
}