use crate::models::*; use crate::search::MemoryIndex; use crate::store::Store; use std::collections::HashMap; use std::path::PathBuf; use std::sync::{Arc, RwLock}; pub struct MemoryState { pub base_dir: PathBuf, pub graph: Store, pub search_index: RwLock, pub ledger: Store>, pub sticky: Store>, pub tasks: Store>, pub snippets: Store>, pub adrs: Store>, pub prefs: Store>, pub error_fixes: Store>, pub pinned_files: Store>, pub session_summaries: Store>, pub handoff_memos: Store>, pub env_fingerprints: Store>, pub env_requirements: Store>, pub milestones: Store>, pub environments: Store>, pub pr_checklists: Store>, pub tech_debts: Store>, pub gates: Store>, pub context_workspaces: Store>, pub recent_activities: Store>, pub activity_tx: tokio::sync::broadcast::Sender, } impl MemoryState { pub fn new(base_dir_str: &str) -> Self { let base = std::path::PathBuf::from(base_dir_str); std::fs::create_dir_all(&base).expect("Failed to create store dir"); let db = crate::db::init_redb(&base); Self { graph: Store::new("knowledge_graph_master", db.clone()), base_dir: base.clone(), search_index: RwLock::new(match crate::search::MemoryIndex::new(&base) { Ok(idx) => idx, Err(e) => { let log_path = dirs::home_dir() .unwrap_or_default() .join(".gemini/mcp_memory/daemon_error.log"); let _ = std::fs::write(&log_path, format!("Failed to create MemoryIndex: {}\n", e)); std::process::exit(1); } }), ledger: Store::new("audit_ledger", db.clone()), sticky: Store::new("sticky_notes", db.clone()), tasks: Store::new("tasks", db.clone()), snippets: Store::new("snippets", db.clone()), adrs: Store::new("adrs", db.clone()), prefs: Store::new("preferences", db.clone()), error_fixes: Store::new("error_fixes", db.clone()), pinned_files: Store::new("pinned_files", db.clone()), session_summaries: Store::new("session_summaries", db.clone()), handoff_memos: Store::new("handoff_memos", db.clone()), env_fingerprints: Store::new("env_fingerprints", db.clone()), env_requirements: Store::new("env_requirements", db.clone()), milestones: Store::new("milestones", db.clone()), environments: Store::new("environments", db.clone()), pr_checklists: Store::new("pr_checklists", db.clone()), tech_debts: Store::new("tech_debts", db.clone()), gates: Store::new("gates", db.clone()), context_workspaces: Store::new("context_workspaces", db.clone()), recent_activities: Store::new("recent_activities", db.clone()), activity_tx: tokio::sync::broadcast::channel(100).0, } } pub fn deduplicate(input: &mut Vec) { let mut keys = std::collections::HashSet::new(); input.retain(|entry| keys.insert(entry.clone())); } pub fn broadcast_activity(&self, message: &str) { let time = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_millis() as u64; let item = serde_json::json!({ "time": time, "message": message }); self.recent_activities.modify(|activities| { activities.push_back(item.clone()); if activities.len() > 100 { activities.pop_front(); } }); let payload = serde_json::json!({ "jsonrpc": "2.0", "method": "notifications/activity", "params": item }) .to_string(); let _ = self.activity_tx.send(payload); } pub fn read_graph(&self, f: F) -> R where F: FnOnce(&KnowledgeGraph) -> R, { self.graph.read_with(f) } pub fn modify_graph(&self, update_fn: F) { self.graph.modify(update_fn); } pub async fn rebuild_index(self: &Arc) { if let Ok(new_idx) = MemoryIndex::new(&self.base_dir) { let state = Arc::clone(self); let idx = new_idx.clone(); tokio::task::spawn_blocking(move || { let entities: Vec<_> = state .graph .read_with(|g| g.entities.values().cloned().collect()); for e in entities { idx.add_entity_sync(&e); } let tasks = state.tasks.read_with(|t| t.clone()); for t in tasks { idx.add_task_sync(&t); } let snippets = state.snippets.read_with(|s| s.clone()); for s in snippets { idx.add_snippet_sync(&s); } let adrs = state.adrs.read_with(|a| a.clone()); for a in adrs { idx.add_adr_sync(&a); } }) .await .unwrap_or_else(|e| { tracing::error!("Failed to join tantivy index rebuild thread: {}", e); }); let _ = new_idx.commit().await; if let Ok(mut w) = self.search_index.write() { *w = new_idx; } } } } #[cfg(test)] mod tests { use super::*; use tempfile::tempdir; #[tokio::test] async fn test_memory_state_initialization() { let dir = tempdir().unwrap(); let state = MemoryState::new(dir.path().to_str().unwrap()); // Ensure state fields are properly initialized assert_eq!(state.base_dir, dir.path()); // Write a test value state.tasks.modify(|tasks| { tasks.push(Task { id: "123".to_string(), title: "Test Task".to_string(), description: "".to_string(), status: "active".to_string(), created_at: 0, updated_at: 0, dependencies: vec![], acceptance_criteria: vec![], git_branch: None, parent_id: None, }); }); // Ensure it is saved state.tasks.read_with(|tasks| { assert_eq!(tasks.len(), 1); assert_eq!(tasks[0].id, "123"); }); // Test rebuild index let arc_state = Arc::new(state); arc_state.rebuild_index().await; // Check search index initialization let idx = arc_state.search_index.read().unwrap(); // Just verify we can read it without panic assert!(idx.search("Test", None).is_ok()); } }