mecha10_core/schema/
registry_impl.rs

1//! `Context`'s Redis-backed implementation of [`super::SchemaRegistryExt`].
2
3use super::{MessageSchema, SchemaRegistryExt};
4use crate::context::Context;
5use crate::error::{Mecha10Error, Result};
6
7impl SchemaRegistryExt for Context {
8    async fn register_schema(&self, schema: &MessageSchema) -> Result<()> {
9        #[cfg(feature = "messaging")]
10        {
11            use redis::AsyncCommands;
12
13            let redis_url = Context::get_redis_url()?;
14            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
15                message: format!("Failed to connect to Redis: {}", e),
16                suggestion: "Ensure Redis is running".to_string(),
17            })?;
18
19            let mut conn =
20                client
21                    .get_multiplexed_async_connection()
22                    .await
23                    .map_err(|e| Mecha10Error::MessagingError {
24                        message: format!("Failed to get Redis connection: {}", e),
25                        suggestion: "Ensure Redis is running".to_string(),
26                    })?;
27
28            // Serialize schema
29            let schema_json = serde_json::to_string(&schema)
30                .map_err(|e| Mecha10Error::Other(format!("Failed to serialize schema: {}", e)))?;
31
32            // Store schema with version
33            let key = schema.redis_key();
34            conn.set::<_, _, ()>(&key, &schema_json)
35                .await
36                .map_err(|e| Mecha10Error::MessagingError {
37                    message: format!("Failed to register schema: {}", e),
38                    suggestion: "Check Redis connection".to_string(),
39                })?;
40
41            // Update latest version pointer
42            let latest_key = MessageSchema::redis_latest_key(&schema.name);
43            conn.set::<_, _, ()>(&latest_key, schema.version)
44                .await
45                .map_err(|e| Mecha10Error::MessagingError {
46                    message: format!("Failed to update latest version: {}", e),
47                    suggestion: "Check Redis connection".to_string(),
48                })?;
49
50            // Add to schema list
51            let list_key = MessageSchema::redis_list_key();
52            conn.sadd::<_, _, ()>(&list_key, &schema.name)
53                .await
54                .map_err(|e| Mecha10Error::MessagingError {
55                    message: format!("Failed to update schema list: {}", e),
56                    suggestion: "Check Redis connection".to_string(),
57                })?;
58
59            Ok(())
60        }
61
62        #[cfg(not(feature = "messaging"))]
63        {
64            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
65        }
66    }
67
68    async fn get_schema(&self, name: &str, version: u32) -> Result<Option<MessageSchema>> {
69        #[cfg(feature = "messaging")]
70        {
71            use redis::AsyncCommands;
72
73            let redis_url = Context::get_redis_url()?;
74            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
75                message: format!("Failed to connect to Redis: {}", e),
76                suggestion: "Ensure Redis is running".to_string(),
77            })?;
78
79            let mut conn =
80                client
81                    .get_multiplexed_async_connection()
82                    .await
83                    .map_err(|e| Mecha10Error::MessagingError {
84                        message: format!("Failed to get Redis connection: {}", e),
85                        suggestion: "Ensure Redis is running".to_string(),
86                    })?;
87
88            let key = format!("mecha10:schemas:{}:{}", name, version);
89            let json: Option<String> = conn.get(&key).await.map_err(|e| Mecha10Error::MessagingError {
90                message: format!("Failed to get schema: {}", e),
91                suggestion: "Check Redis connection".to_string(),
92            })?;
93
94            if let Some(json) = json {
95                let schema = serde_json::from_str::<MessageSchema>(&json)
96                    .map_err(|e| Mecha10Error::Other(format!("Failed to parse schema: {}", e)))?;
97                Ok(Some(schema))
98            } else {
99                Ok(None)
100            }
101        }
102
103        #[cfg(not(feature = "messaging"))]
104        {
105            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
106        }
107    }
108
109    async fn get_latest_schema(&self, name: &str) -> Result<Option<MessageSchema>> {
110        #[cfg(feature = "messaging")]
111        {
112            use redis::AsyncCommands;
113
114            let redis_url = Context::get_redis_url()?;
115            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
116                message: format!("Failed to connect to Redis: {}", e),
117                suggestion: "Ensure Redis is running".to_string(),
118            })?;
119
120            let mut conn =
121                client
122                    .get_multiplexed_async_connection()
123                    .await
124                    .map_err(|e| Mecha10Error::MessagingError {
125                        message: format!("Failed to get Redis connection: {}", e),
126                        suggestion: "Ensure Redis is running".to_string(),
127                    })?;
128
129            // Get latest version number
130            let latest_key = MessageSchema::redis_latest_key(name);
131            let version: Option<u32> = conn.get(&latest_key).await.map_err(|e| Mecha10Error::MessagingError {
132                message: format!("Failed to get latest version: {}", e),
133                suggestion: "Check Redis connection".to_string(),
134            })?;
135
136            if let Some(version) = version {
137                self.get_schema(name, version).await
138            } else {
139                Ok(None)
140            }
141        }
142
143        #[cfg(not(feature = "messaging"))]
144        {
145            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
146        }
147    }
148
149    async fn list_schemas(&self) -> Result<Vec<MessageSchema>> {
150        #[cfg(feature = "messaging")]
151        {
152            crate::redis_list_util::collect_from_redis_set(&MessageSchema::redis_list_key(), |name| async move {
153                self.get_latest_schema(&name).await
154            })
155            .await
156        }
157
158        #[cfg(not(feature = "messaging"))]
159        {
160            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
161        }
162    }
163
164    async fn get_schema_versions(&self, name: &str) -> Result<Vec<MessageSchema>> {
165        #[cfg(feature = "messaging")]
166        {
167            use redis::AsyncCommands;
168
169            let redis_url = Context::get_redis_url()?;
170            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
171                message: format!("Failed to connect to Redis: {}", e),
172                suggestion: "Ensure Redis is running".to_string(),
173            })?;
174
175            let mut conn =
176                client
177                    .get_multiplexed_async_connection()
178                    .await
179                    .map_err(|e| Mecha10Error::MessagingError {
180                        message: format!("Failed to get Redis connection: {}", e),
181                        suggestion: "Ensure Redis is running".to_string(),
182                    })?;
183
184            // Scan for all versions of this schema
185            let pattern = format!("mecha10:schemas:{}:*", name);
186            let keys: Vec<String> = conn.keys(&pattern).await.map_err(|e| Mecha10Error::MessagingError {
187                message: format!("Failed to scan for schema versions: {}", e),
188                suggestion: "Check Redis connection".to_string(),
189            })?;
190
191            let mut schemas = Vec::new();
192            for key in keys {
193                // Skip the 'latest' key
194                if key.ends_with(":latest") {
195                    continue;
196                }
197
198                let json: String = conn.get(&key).await.map_err(|e| Mecha10Error::MessagingError {
199                    message: format!("Failed to get schema: {}", e),
200                    suggestion: "Check Redis connection".to_string(),
201                })?;
202
203                if let Ok(schema) = serde_json::from_str::<MessageSchema>(&json) {
204                    schemas.push(schema);
205                }
206            }
207
208            // Sort by version
209            schemas.sort_by_key(|s| s.version);
210
211            Ok(schemas)
212        }
213
214        #[cfg(not(feature = "messaging"))]
215        {
216            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
217        }
218    }
219
220    async fn delete_schema(&self, name: &str, version: Option<u32>) -> Result<()> {
221        #[cfg(feature = "messaging")]
222        {
223            use redis::AsyncCommands;
224
225            let redis_url = Context::get_redis_url()?;
226            let client = redis::Client::open(redis_url.as_str()).map_err(|e| Mecha10Error::MessagingError {
227                message: format!("Failed to connect to Redis: {}", e),
228                suggestion: "Ensure Redis is running".to_string(),
229            })?;
230
231            let mut conn =
232                client
233                    .get_multiplexed_async_connection()
234                    .await
235                    .map_err(|e| Mecha10Error::MessagingError {
236                        message: format!("Failed to get Redis connection: {}", e),
237                        suggestion: "Ensure Redis is running".to_string(),
238                    })?;
239
240            if let Some(version) = version {
241                // Delete specific version
242                let key = format!("mecha10:schemas:{}:{}", name, version);
243                conn.del::<_, ()>(&key)
244                    .await
245                    .map_err(|e| Mecha10Error::MessagingError {
246                        message: format!("Failed to delete schema: {}", e),
247                        suggestion: "Check Redis connection".to_string(),
248                    })?;
249            } else {
250                // Delete all versions
251                let pattern = format!("mecha10:schemas:{}:*", name);
252                let keys: Vec<String> = conn.keys(&pattern).await.map_err(|e| Mecha10Error::MessagingError {
253                    message: format!("Failed to scan for schemas: {}", e),
254                    suggestion: "Check Redis connection".to_string(),
255                })?;
256
257                for key in keys {
258                    conn.del::<_, ()>(&key)
259                        .await
260                        .map_err(|e| Mecha10Error::MessagingError {
261                            message: format!("Failed to delete schema: {}", e),
262                            suggestion: "Check Redis connection".to_string(),
263                        })?;
264                }
265
266                // Remove from schema list
267                let list_key = MessageSchema::redis_list_key();
268                conn.srem::<_, _, ()>(&list_key, name)
269                    .await
270                    .map_err(|e| Mecha10Error::MessagingError {
271                        message: format!("Failed to update schema list: {}", e),
272                        suggestion: "Check Redis connection".to_string(),
273                    })?;
274            }
275
276            Ok(())
277        }
278
279        #[cfg(not(feature = "messaging"))]
280        {
281            Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
282        }
283    }
284}
mecha10_core/schema/registry_impl.rs

mecha10_core/schema/
registry_impl.rs
