confium_coordinator/coordinator/
store.rs

1//! Persistent session store — pluggable backend for session state.
2//!
3//! Sessions are serialized to JSON and stored via a [`SessionStore`]
4//! backend. On coordinator restart, sessions are loaded from the
5//! store and restored to their last-persisted state.
6//!
7//! ## OCP design
8//!
9//! New backends (Redis, PostgreSQL, S3) are added by implementing
10//! the [`SessionStore`] trait.
11
12use serde::{Deserialize, Serialize};
13use std::collections::HashMap;
14use std::path::{Path, PathBuf};
15use std::sync::Mutex;
16
17/// Serialized session snapshot — the unit of persistence.
18#[derive(Debug, Clone, Serialize, Deserialize)]
19pub struct SessionSnapshot {
20    /// Session ID.
21    pub session_id: String,
22    /// JSON-encoded session state.
23    pub state_json: String,
24    /// When the snapshot was taken.
25    pub snapshot_at: chrono::DateTime<chrono::Utc>,
26}
27
28/// Errors during store operations.
29#[derive(Debug, thiserror::Error)]
30pub enum StoreError {
31    /// I/O error.
32    #[error("io error: {0}")]
33    Io(String),
34    /// Serialization error.
35    #[error("serialization error: {0}")]
36    Serialization(String),
37}
38
39/// Trait for session persistence backends.
40pub trait SessionStore: Send + Sync {
41    /// Save a session snapshot.
42    fn save(&self, snapshot: &SessionSnapshot) -> Result<(), StoreError>;
43
44    /// Load a session snapshot by ID.
45    fn load(&self, session_id: &str) -> Result<Option<SessionSnapshot>, StoreError>;
46
47    /// Delete a session.
48    fn delete(&self, session_id: &str) -> Result<(), StoreError>;
49
50    /// List all stored session IDs.
51    fn list(&self) -> Result<Vec<String>, StoreError>;
52}
53
54/// In-memory session store (default, no persistence).
55#[derive(Default)]
56pub struct InMemorySessionStore {
57    entries: Mutex<HashMap<String, SessionSnapshot>>,
58}
59
60impl SessionStore for InMemorySessionStore {
61    fn save(&self, snapshot: &SessionSnapshot) -> Result<(), StoreError> {
62        self.entries
63            .lock()
64            .unwrap()
65            .insert(snapshot.session_id.clone(), snapshot.clone());
66        Ok(())
67    }
68
69    fn load(&self, session_id: &str) -> Result<Option<SessionSnapshot>, StoreError> {
70        Ok(self.entries.lock().unwrap().get(session_id).cloned())
71    }
72
73    fn delete(&self, session_id: &str) -> Result<(), StoreError> {
74        self.entries.lock().unwrap().remove(session_id);
75        Ok(())
76    }
77
78    fn list(&self) -> Result<Vec<String>, StoreError> {
79        Ok(self.entries.lock().unwrap().keys().cloned().collect())
80    }
81}
82
83/// File-based session store. Each session is a JSON file in a
84/// directory.
85pub struct FileSessionStore {
86    dir: PathBuf,
87}
88
89impl FileSessionStore {
90    /// Create a new file store rooted at `dir`. Creates the directory
91    /// if it doesn't exist.
92    pub fn new(dir: impl AsRef<Path>) -> Result<Self, StoreError> {
93        let dir = dir.as_ref().to_path_buf();
94        std::fs::create_dir_all(&dir).map_err(|e| StoreError::Io(e.to_string()))?;
95        Ok(Self { dir })
96    }
97
98    fn session_path(&self, session_id: &str) -> PathBuf {
99        let safe_name = session_id.replace('/', "_");
100        self.dir.join(format!("{safe_name}.json"))
101    }
102}
103
104impl SessionStore for FileSessionStore {
105    fn save(&self, snapshot: &SessionSnapshot) -> Result<(), StoreError> {
106        let path = self.session_path(&snapshot.session_id);
107        let json = serde_json::to_string_pretty(snapshot)
108            .map_err(|e| StoreError::Serialization(e.to_string()))?;
109        std::fs::write(&path, json).map_err(|e| StoreError::Io(e.to_string()))
110    }
111
112    fn load(&self, session_id: &str) -> Result<Option<SessionSnapshot>, StoreError> {
113        let path = self.session_path(session_id);
114        if !path.exists() {
115            return Ok(None);
116        }
117        let contents = std::fs::read_to_string(&path).map_err(|e| StoreError::Io(e.to_string()))?;
118        let snapshot: SessionSnapshot = serde_json::from_str(&contents)
119            .map_err(|e| StoreError::Serialization(e.to_string()))?;
120        Ok(Some(snapshot))
121    }
122
123    fn delete(&self, session_id: &str) -> Result<(), StoreError> {
124        let path = self.session_path(session_id);
125        if path.exists() {
126            std::fs::remove_file(&path).map_err(|e| StoreError::Io(e.to_string()))?;
127        }
128        Ok(())
129    }
130
131    fn list(&self) -> Result<Vec<String>, StoreError> {
132        let mut ids = Vec::new();
133        for entry in std::fs::read_dir(&self.dir).map_err(|e| StoreError::Io(e.to_string()))? {
134            let entry = entry.map_err(|e| StoreError::Io(e.to_string()))?;
135            let name = entry.file_name().to_string_lossy().to_string();
136            if let Some(id) = name.strip_suffix(".json") {
137                ids.push(id.to_string());
138            }
139        }
140        Ok(ids)
141    }
142}
143
144#[cfg(test)]
145mod tests {
146    use super::*;
147    use chrono::Utc;
148
149    fn make_snapshot(id: &str) -> SessionSnapshot {
150        SessionSnapshot {
151            session_id: id.into(),
152            state_json: r#"{"state":"pending"}"#.into(),
153            snapshot_at: Utc::now(),
154        }
155    }
156
157    // InMemorySessionStore tests
158
159    #[test]
160    fn in_memory_save_and_load() {
161        let store = InMemorySessionStore::default();
162        let snapshot = make_snapshot("s1");
163        store.save(&snapshot).unwrap();
164        let loaded = store.load("s1").unwrap().unwrap();
165        assert_eq!(loaded.session_id, "s1");
166    }
167
168    #[test]
169    fn in_memory_load_missing_returns_none() {
170        let store = InMemorySessionStore::default();
171        assert!(store.load("missing").unwrap().is_none());
172    }
173
174    #[test]
175    fn in_memory_delete_removes_entry() {
176        let store = InMemorySessionStore::default();
177        store.save(&make_snapshot("s1")).unwrap();
178        store.delete("s1").unwrap();
179        assert!(store.load("s1").unwrap().is_none());
180    }
181
182    #[test]
183    fn in_memory_list_returns_all() {
184        let store = InMemorySessionStore::default();
185        store.save(&make_snapshot("s1")).unwrap();
186        store.save(&make_snapshot("s2")).unwrap();
187        let mut ids = store.list().unwrap();
188        ids.sort();
189        assert_eq!(ids, vec!["s1", "s2"]);
190    }
191
192    #[test]
193    fn in_memory_overwrite_on_save() {
194        let store = InMemorySessionStore::default();
195        store.save(&make_snapshot("s1")).unwrap();
196        let mut updated = make_snapshot("s1");
197        updated.state_json = r#"{"state":"completed"}"#.into();
198        store.save(&updated).unwrap();
199        let loaded = store.load("s1").unwrap().unwrap();
200        assert!(loaded.state_json.contains("completed"));
201    }
202
203    // FileSessionStore tests
204
205    #[test]
206    fn file_store_creates_directory() {
207        let tmp = tempfile::tempdir().unwrap();
208        let path = tmp.path().join("nested/sessions");
209        let _store = FileSessionStore::new(&path).unwrap();
210        assert!(path.exists());
211    }
212
213    #[test]
214    fn file_store_save_and_load() {
215        let tmp = tempfile::tempdir().unwrap();
216        let store = FileSessionStore::new(tmp.path()).unwrap();
217        store.save(&make_snapshot("s1")).unwrap();
218        let loaded = store.load("s1").unwrap().unwrap();
219        assert_eq!(loaded.session_id, "s1");
220    }
221
222    #[test]
223    fn file_store_load_missing_returns_none() {
224        let tmp = tempfile::tempdir().unwrap();
225        let store = FileSessionStore::new(tmp.path()).unwrap();
226        assert!(store.load("missing").unwrap().is_none());
227    }
228
229    #[test]
230    fn file_store_delete_removes_file() {
231        let tmp = tempfile::tempdir().unwrap();
232        let store = FileSessionStore::new(tmp.path()).unwrap();
233        store.save(&make_snapshot("s1")).unwrap();
234        assert!(store.load("s1").unwrap().is_some());
235        store.delete("s1").unwrap();
236        assert!(store.load("s1").unwrap().is_none());
237    }
238
239    #[test]
240    fn file_store_list_returns_ids() {
241        let tmp = tempfile::tempdir().unwrap();
242        let store = FileSessionStore::new(tmp.path()).unwrap();
243        store.save(&make_snapshot("s1")).unwrap();
244        store.save(&make_snapshot("s2")).unwrap();
245        let mut ids = store.list().unwrap();
246        ids.sort();
247        assert_eq!(ids, vec!["s1", "s2"]);
248    }
249
250    #[test]
251    fn file_store_survives_across_instances() {
252        let tmp = tempfile::tempdir().unwrap();
253        {
254            let store = FileSessionStore::new(tmp.path()).unwrap();
255            store.save(&make_snapshot("persistent")).unwrap();
256        }
257        {
258            let store = FileSessionStore::new(tmp.path()).unwrap();
259            let loaded = store.load("persistent").unwrap().unwrap();
260            assert_eq!(loaded.session_id, "persistent");
261        }
262    }
263
264    #[test]
265    fn file_store_sanitizes_slashes() {
266        let tmp = tempfile::tempdir().unwrap();
267        let store = FileSessionStore::new(tmp.path()).unwrap();
268        store.save(&make_snapshot("session/with/slashes")).unwrap();
269        let loaded = store.load("session/with/slashes").unwrap().unwrap();
270        assert_eq!(loaded.session_id, "session/with/slashes");
271    }
272}
confium_coordinator/coordinator/store.rs

confium_coordinator/coordinator/
store.rs
