mecha10_core/schema/
registry_impl.rs1use 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 mut conn = crate::redis_list_util::get_redis_connection().await?;
14
15 let schema_json = serde_json::to_string(&schema)
16 .map_err(|e| Mecha10Error::Other(format!("Failed to serialize schema: {}", e)))?;
17
18 let key = schema.redis_key();
19 conn.set::<_, _, ()>(&key, &schema_json)
20 .await
21 .map_err(|e| Mecha10Error::MessagingError {
22 message: format!("Failed to register schema: {}", e),
23 suggestion: "Check Redis connection".to_string(),
24 })?;
25
26 let latest_key = MessageSchema::redis_latest_key(&schema.name);
27 conn.set::<_, _, ()>(&latest_key, schema.version)
28 .await
29 .map_err(|e| Mecha10Error::MessagingError {
30 message: format!("Failed to update latest version: {}", e),
31 suggestion: "Check Redis connection".to_string(),
32 })?;
33
34 let list_key = MessageSchema::redis_list_key();
35 conn.sadd::<_, _, ()>(&list_key, &schema.name)
36 .await
37 .map_err(|e| Mecha10Error::MessagingError {
38 message: format!("Failed to update schema list: {}", e),
39 suggestion: "Check Redis connection".to_string(),
40 })?;
41
42 Ok(())
43 }
44
45 #[cfg(not(feature = "messaging"))]
46 {
47 Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
48 }
49 }
50
51 async fn get_schema(&self, name: &str, version: u32) -> Result<Option<MessageSchema>> {
52 #[cfg(feature = "messaging")]
53 {
54 use redis::AsyncCommands;
55
56 let mut conn = crate::redis_list_util::get_redis_connection().await?;
57
58 let key = format!("mecha10:schemas:{}:{}", name, version);
59 let json: Option<String> = conn.get(&key).await.map_err(|e| Mecha10Error::MessagingError {
60 message: format!("Failed to get schema: {}", e),
61 suggestion: "Check Redis connection".to_string(),
62 })?;
63
64 if let Some(json) = json {
65 let schema = serde_json::from_str::<MessageSchema>(&json)
66 .map_err(|e| Mecha10Error::Other(format!("Failed to parse schema: {}", e)))?;
67 Ok(Some(schema))
68 } else {
69 Ok(None)
70 }
71 }
72
73 #[cfg(not(feature = "messaging"))]
74 {
75 Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
76 }
77 }
78
79 async fn get_latest_schema(&self, name: &str) -> Result<Option<MessageSchema>> {
80 #[cfg(feature = "messaging")]
81 {
82 let latest_key = MessageSchema::redis_latest_key(name);
83 let version: Option<u32> =
84 crate::redis_list_util::get_optional_from_redis(&latest_key, "Failed to get latest version").await?;
85
86 if let Some(version) = version {
87 self.get_schema(name, version).await
88 } else {
89 Ok(None)
90 }
91 }
92
93 #[cfg(not(feature = "messaging"))]
94 {
95 Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
96 }
97 }
98
99 async fn list_schemas(&self) -> Result<Vec<MessageSchema>> {
100 #[cfg(feature = "messaging")]
101 {
102 crate::redis_list_util::collect_from_redis_set(&MessageSchema::redis_list_key(), |name| async move {
103 self.get_latest_schema(&name).await
104 })
105 .await
106 }
107
108 #[cfg(not(feature = "messaging"))]
109 {
110 Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
111 }
112 }
113
114 async fn get_schema_versions(&self, name: &str) -> Result<Vec<MessageSchema>> {
115 #[cfg(feature = "messaging")]
116 {
117 use redis::AsyncCommands;
118
119 let mut conn = crate::redis_list_util::get_redis_connection().await?;
120
121 let pattern = format!("mecha10:schemas:{}:*", name);
122 let keys: Vec<String> = conn.keys(&pattern).await.map_err(|e| Mecha10Error::MessagingError {
123 message: format!("Failed to scan for schema versions: {}", e),
124 suggestion: "Check Redis connection".to_string(),
125 })?;
126
127 let mut schemas = Vec::new();
128 for key in keys {
129 if key.ends_with(":latest") {
130 continue;
131 }
132
133 let json: String = conn.get(&key).await.map_err(|e| Mecha10Error::MessagingError {
134 message: format!("Failed to get schema: {}", e),
135 suggestion: "Check Redis connection".to_string(),
136 })?;
137
138 if let Ok(schema) = serde_json::from_str::<MessageSchema>(&json) {
139 schemas.push(schema);
140 }
141 }
142
143 schemas.sort_by_key(|s| s.version);
144
145 Ok(schemas)
146 }
147
148 #[cfg(not(feature = "messaging"))]
149 {
150 Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
151 }
152 }
153
154 async fn delete_schema(&self, name: &str, version: Option<u32>) -> Result<()> {
155 #[cfg(feature = "messaging")]
156 {
157 use redis::AsyncCommands;
158
159 let mut conn = crate::redis_list_util::get_redis_connection().await?;
160
161 if let Some(version) = version {
162 let key = format!("mecha10:schemas:{}:{}", name, version);
163 conn.del::<_, ()>(&key)
164 .await
165 .map_err(|e| Mecha10Error::MessagingError {
166 message: format!("Failed to delete schema: {}", e),
167 suggestion: "Check Redis connection".to_string(),
168 })?;
169 } else {
170 let pattern = format!("mecha10:schemas:{}:*", name);
171 let keys: Vec<String> = conn.keys(&pattern).await.map_err(|e| Mecha10Error::MessagingError {
172 message: format!("Failed to scan for schemas: {}", e),
173 suggestion: "Check Redis connection".to_string(),
174 })?;
175
176 for key in keys {
177 conn.del::<_, ()>(&key)
178 .await
179 .map_err(|e| Mecha10Error::MessagingError {
180 message: format!("Failed to delete schema: {}", e),
181 suggestion: "Check Redis connection".to_string(),
182 })?;
183 }
184
185 let list_key = MessageSchema::redis_list_key();
186 conn.srem::<_, _, ()>(&list_key, name)
187 .await
188 .map_err(|e| Mecha10Error::MessagingError {
189 message: format!("Failed to update schema list: {}", e),
190 suggestion: "Check Redis connection".to_string(),
191 })?;
192 }
193
194 Ok(())
195 }
196
197 #[cfg(not(feature = "messaging"))]
198 {
199 Err(Mecha10Error::Other("Messaging feature not enabled".to_string()))
200 }
201 }
202}