stormchaser_model/
schema_cache.rs

1use serde_json::Value;
2use std::collections::HashMap;
3use std::sync::Arc;
4use tokio::sync::RwLock;
5
6/// In-memory cache for OCI-backed CloudEvent schemas.
7#[derive(Clone)]
8pub struct SchemaCache {
9    schemas: Arc<RwLock<HashMap<String, Value>>>,
10}
11
12impl Default for SchemaCache {
13    fn default() -> Self {
14        Self::new()
15    }
16}
17
18impl SchemaCache {
19    /// Create a new empty schema cache.
20    pub fn new() -> Self {
21        Self {
22            schemas: Arc::new(RwLock::new(HashMap::new())),
23        }
24    }
25
26    /// Get a schema by its ID or Version.
27    pub async fn get(&self, schema_id: &str) -> Option<Value> {
28        let cache = self.schemas.read().await;
29        cache.get(schema_id).cloned()
30    }
31
32    /// Insert or update a schema in the cache.
33    pub async fn insert(&self, schema_id: String, schema: Value) {
34        let mut cache = self.schemas.write().await;
35        cache.insert(schema_id, schema);
36    }
37
38    /// Start a background sync process to fetch schemas from an OCI registry.
39    /// This is a permissive sync; it runs asynchronously and updates the cache.
40    pub fn start_background_sync(&self, _oci_registry_url: String) {
41        let _schemas = self.schemas.clone();
42        tokio::spawn(async move {
43            // Implementation for syncing from `oci_registry_url` goes here.
44            // For now, this is a stub that represents the background sync loop.
45            loop {
46                // TODO: Implement OCI fetch
47                // let fetched_schemas = fetch_schemas_from_oci(&oci_registry_url).await;
48                // for (id, schema) in fetched_schemas {
49                //     schemas.write().await.insert(id, schema);
50                // }
51                tokio::time::sleep(std::time::Duration::from_secs(3600)).await;
52            }
53        });
54    }
55}
56
57#[cfg(test)]
58mod tests {
59    use super::*;
60
61    #[tokio::test]
62    async fn test_schema_cache_insert_and_get() {
63        let cache = SchemaCache::default();
64        let schema_id = "test_schema_id".to_string();
65        let schema_val = serde_json::json!({"type": "object"});
66
67        // Should be empty initially
68        assert_eq!(cache.get(&schema_id).await, None);
69
70        // Insert
71        cache.insert(schema_id.clone(), schema_val.clone()).await;
72
73        // Should return the inserted value
74        assert_eq!(cache.get(&schema_id).await, Some(schema_val));
75    }
76
77    #[tokio::test]
78    async fn test_schema_cache_start_background_sync() {
79        let cache = SchemaCache::new();
80        // Since it's a mock implementation, we just verify it runs without panicking.
81        cache.start_background_sync("dummy_url".to_string());
82    }
83}
stormchaser_model/schema_cache.rs

stormchaser_model/
schema_cache.rs
