synapto_interface/
storage.rs

1#![doc = include_str!("storage.md")]
2
3use async_trait::async_trait;
4
5/// A marker trait for safe DB connection pooling
6pub trait StorageProviderPool: Send + Sync + 'static {}
7
8#[derive(Default)]
9pub struct StorageRegistry {
10    map: tokio::sync::Mutex<
11        std::collections::HashMap<
12            std::any::TypeId,
13            std::sync::Arc<dyn std::any::Any + Send + Sync>,
14        >,
15    >,
16}
17
18impl StorageRegistry {
19    /// Lazily initializes a global shared resource. If the resource already exists,
20    /// it is returned immediately. This allows multiple plugins to safely share a
21    /// single connection pool without requiring manual initialization in main.rs.
22    pub async fn get_or_init<T: StorageProviderPool, F, Fut, E>(
23        &self,
24        init: F,
25    ) -> Result<std::sync::Arc<T>, E>
26    where
27        F: FnOnce() -> Fut,
28        Fut: std::future::Future<Output = Result<T, E>>,
29    {
30        let mut map = self.map.lock().await;
31        let type_id = std::any::TypeId::of::<T>();
32
33        if let Some(resource) = map.get(&type_id) {
34            // This can only fail if the TypeId of T doesn't match the Arc's inner type,
35            // which is impossible since we keyed the HashMap by TypeId::of::<T>().
36            return Ok(resource
37                .clone()
38                .downcast::<T>()
39                .unwrap_or_else(|_| unreachable!("TypeId mismatch in StorageRegistry")));
40        }
41
42        let resource = std::sync::Arc::new(init().await?);
43        map.insert(type_id, resource.clone());
44        Ok(resource)
45    }
46}
47
48pub trait StorageConfigResolver: Send + Sync + 'static {
49    fn resolve_config(
50        &self,
51        crate_name: &str,
52        storage_type_name: &str,
53    ) -> Option<serde_json::Value>;
54}
55
56/// Opaque handle encapsulating storage connection pooling and configuration resolution.
57#[derive(Clone)]
58pub struct StorageHandle {
59    registry: std::sync::Arc<StorageRegistry>,
60    resolver: std::sync::Arc<dyn StorageConfigResolver>,
61}
62
63impl std::fmt::Debug for StorageHandle {
64    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
65        f.debug_struct("StorageHandle").finish_non_exhaustive()
66    }
67}
68
69struct DefaultStorageConfigResolver;
70
71impl StorageConfigResolver for DefaultStorageConfigResolver {
72    fn resolve_config(
73        &self,
74        _crate_name: &str,
75        _storage_type_name: &str,
76    ) -> Option<serde_json::Value> {
77        None
78    }
79}
80
81impl Default for StorageHandle {
82    fn default() -> Self {
83        Self {
84            registry: std::sync::Arc::new(StorageRegistry::default()),
85            resolver: std::sync::Arc::new(DefaultStorageConfigResolver),
86        }
87    }
88}
89
90impl StorageHandle {
91    pub fn new(resolver: std::sync::Arc<dyn StorageConfigResolver>) -> Self {
92        Self {
93            registry: std::sync::Arc::new(StorageRegistry::default()),
94            resolver,
95        }
96    }
97
98    pub fn with_registry(
99        registry: std::sync::Arc<StorageRegistry>,
100        resolver: std::sync::Arc<dyn StorageConfigResolver>,
101    ) -> Self {
102        Self { registry, resolver }
103    }
104
105    /// Resolves configuration and establishes a scoped storage connection.
106    pub async fn connect_store<S: StorageConnection>(
107        &self,
108        plugin_namespace: &str,
109    ) -> Result<std::sync::Arc<S>, String> {
110        let full_path = std::any::type_name::<S>();
111        let crate_name = full_path
112            .split("::")
113            .next()
114            .unwrap_or("")
115            .to_string()
116            .replace('-', "_");
117        let base_path = full_path.split('<').next().unwrap_or(full_path);
118        let storage_type_name = base_path.split("::").last().unwrap_or("").to_string();
119
120        let config_val = self
121            .resolver
122            .resolve_config(&crate_name, &storage_type_name)
123            .unwrap_or_else(|| serde_json::json!({}));
124
125        let config: S::Config = serde_json::from_value(config_val).map_err(|e| {
126            format!(
127                "Failed to parse config for storage '{}::{}': {}",
128                crate_name, storage_type_name, e
129            )
130        })?;
131
132        let store = S::connect(config, self, plugin_namespace).await?;
133        Ok(std::sync::Arc::new(store))
134    }
135
136    /// Accesses or initializes a pooled storage resource safely.
137    pub async fn get_or_init_pool<T: StorageProviderPool, F, Fut>(
138        &self,
139        init: F,
140    ) -> Result<std::sync::Arc<T>, String>
141    where
142        F: FnOnce() -> Fut,
143        Fut: std::future::Future<Output = Result<T, String>>,
144    {
145        self.registry.get_or_init(init).await
146    }
147}
148
149/// The entry point for a generic storage adapter.
150/// It guarantees that plugins can seamlessly initialize their underlying connection
151/// using the shared StorageHandle without requiring manual setup in main.rs.
152#[async_trait]
153pub trait StorageConnection: Send + Sync + Sized + 'static {
154    type Config: serde::de::DeserializeOwned + Send + Sync;
155
156    async fn connect(
157        config: Self::Config,
158        storage_handle: &StorageHandle,
159        plugin_namespace: &str,
160    ) -> Result<Self, String>;
161}
162
163use serde::{Deserialize, Serialize, de::DeserializeOwned};
164
165#[derive(Debug, Clone, Serialize, Deserialize, Default)]
166pub struct EmptyStorageConfig {}
167
168#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
169pub enum SortOrder {
170    #[default]
171    Ascending,
172    Descending,
173}
174
175#[async_trait]
176pub trait RecordStore: Send + Sync + 'static {
177    /// Inserts or updates an individual record.
178    /// If the key is a Timestamp or ULID, time-based ordering is natively maintained.
179    async fn upsert_record<T>(&self, collection: &str, key: &str, value: T) -> Result<(), String>
180    where
181        T: Serialize + Send + Sync + 'static;
182
183    /// Retrieves records guaranteed to be sorted by their key.
184    /// Allows pagination to avoid loading the entire history into RAM.
185    async fn get_ordered_records<T>(
186        &self,
187        collection: &str,
188        limit: Option<usize>,
189        order: SortOrder,
190    ) -> Result<Vec<(String, T)>, String>
191    where
192        T: DeserializeOwned + Send + Sync + 'static;
193
194    /// Deletes a specific record.
195    async fn delete_record(&self, collection: &str, key: &str) -> Result<(), String>;
196
197    /// Atomically deletes all records with a key smaller than `cutoff_key`.
198    /// This natively delegates sliding-window "VecDeque::pop_front()" operations to the DB.
199    async fn trim_records_before(&self, collection: &str, cutoff_key: &str) -> Result<(), String>;
200}
201
202/// For storing and retrieving items by a unique string ID.
203#[async_trait]
204pub trait KeyValueStore: Send + Sync + 'static {
205    async fn set<T>(&self, collection: &str, key: &str, value: T) -> Result<(), String>
206    where
207        T: Serialize + Send + Sync + 'static;
208
209    async fn get<T>(&self, collection: &str, key: &str) -> Result<Option<T>, String>
210    where
211        T: DeserializeOwned + Send + Sync + 'static;
212
213    async fn delete(&self, collection: &str, key: &str) -> Result<(), String>;
214
215    async fn get_all<T>(&self, collection: &str) -> Result<Vec<T>, String>
216    where
217        T: DeserializeOwned + Send + Sync + 'static;
218}
219
220/// Trait for storing, retrieving, and deleting raw binary files.
221#[async_trait]
222pub trait FileStore: Send + Sync + 'static {
223    /// Saves raw bytes under the specified collection and unique file identifier.
224    async fn save_file(
225        &self,
226        collection: &str,
227        file_id: &str,
228        content: Vec<u8>,
229    ) -> Result<(), String>;
230
231    /// Retrieves raw bytes by its identifier.
232    async fn get_file(&self, collection: &str, file_id: &str) -> Result<Option<Vec<u8>>, String>;
233
234    /// Deletes a file.
235    async fn delete_file(&self, collection: &str, file_id: &str) -> Result<(), String>;
236}
237#[async_trait]
238pub trait VectorStore: Send + Sync + 'static {
239    /// Ensures a collection is ready for vector operations.
240    ///
241    /// This method is typically called during the application boot sequence or when a service starts up.
242    /// It should be idempotent.
243    ///
244    /// Depending on the underlying database, this method might:
245    /// - Define a schema or table if it doesn't exist.
246    /// - Create necessary vector search indexes (e.g., M-Tree, HNSW).
247    /// - Do absolutely nothing if the database manages indexing transparently (e.g., Firestore).
248    ///
249    /// By default, this does nothing and returns `Ok(())`. Storage providers that require
250    /// explicit schema or index definition must override this implementation.
251    async fn setup_collection(&self, _collection: &str, _dimension: u32) -> Result<(), String> {
252        Ok(())
253    }
254
255    async fn insert_vectors<T>(&self, collection: &str, records: Vec<T>) -> Result<(), String>
256    where
257        T: Serialize + Send + Sync + 'static;
258
259    async fn search_vectors<T>(
260        &self,
261        collection: &str,
262        vector: Vec<f32>,
263        limit: u32,
264    ) -> Result<Vec<T>, String>
265    where
266        T: DeserializeOwned + Send + Sync + 'static;
267
268    async fn delete_vectors(
269        &self,
270        collection: &str,
271        filter_field: &str,
272        filter_value: &str,
273    ) -> Result<(), String>;
274}
synapto_interface/storage.rs

synapto_interface/
storage.rs
