mneme/
cloud.rs

1use std::sync::Arc;
2
3use chrono::Utc;
4use serde::{Deserialize, Serialize};
5use uuid::Uuid;
6
7use crate::store::db::Database;
8use crate::sync::engine::SyncEngine;
9use crate::sync::peer::{Peer, TransportType};
10
11/// Estado de enrollment del cloud.
12#[derive(Debug, Clone, Serialize, Deserialize)]
13#[serde(rename_all = "snake_case")]
14pub enum CloudState {
15    /// No configurado.
16    Disabled,
17    /// Conectando al servidor cloud.
18    Connecting,
19    /// Enrolled y sincronizando.
20    Syncing,
21    /// Error de conexión.
22    Error(String),
23}
24
25/// Configuración del cloud para un proyecto.
26#[derive(Debug, Clone, Serialize, Deserialize)]
27pub struct CloudConfig {
28    /// Servidor cloud URL.
29    pub server_url: String,
30    /// Token de autenticación.
31    pub token: String,
32    /// Proyecto en el cloud.
33    pub project: String,
34    /// Intervalo de autosync en segundos.
35    pub auto_sync_interval: u64,
36    /// Último sync exitoso.
37    pub last_sync: Option<chrono::DateTime<Utc>>,
38    /// Estado actual.
39    pub state: CloudState,
40}
41
42/// Resultado de una operación cloud.
43#[derive(Debug, Clone, Serialize, Deserialize)]
44pub struct CloudResult {
45    pub success: bool,
46    pub message: String,
47    pub project: String,
48    pub memories_synced: u32,
49    pub conflicts_resolved: u32,
50    pub duration_ms: u64,
51}
52
53/// Orquestador de sync cloud.
54pub struct CloudOrchestrator {
55    db: Arc<Database>,
56    config: crate::config::settings::SyncConfig,
57}
58
59impl CloudOrchestrator {
60    pub fn new(db: Arc<Database>, config: crate::config::settings::SyncConfig) -> Self {
61        Self { db, config }
62    }
63
64    /// Enrolla el proyecto con un servidor cloud.
65    pub async fn enroll(
66        &self,
67        server_url: &str,
68        token: &str,
69        project: &str,
70    ) -> crate::error::Result<CloudResult> {
71        let start = std::time::Instant::now();
72        let peer_id = Uuid::new_v4();
73
74        // Save cloud peer
75        let peer = Peer {
76            id: peer_id,
77            name: format!("cloud-{}", project),
78            transport: TransportType::Http,
79            address: server_url.to_string(),
80            project: project.to_string(),
81            last_sync: None,
82            last_status: None,
83            auto_sync: true,
84            created_at: Utc::now(),
85        };
86
87        let peer_store = self.db.peers();
88        peer_store.add(&peer)?;
89
90        // Send hello to cloud server
91        let engine = SyncEngine::new(self.db.clone(), self.config.clone())?;
92        let hello = engine.build_hello(project)?;
93
94        let client = reqwest::Client::new();
95        let resp = client
96            .post(format!("{}/api/v1/sync/hello", server_url.trim_end_matches('/')))
97            .header("Authorization", format!("Bearer {}", token))
98            .json(&hello)
99            .send()
100            .await;
101
102        let duration_ms = start.elapsed().as_millis() as u64;
103
104        match resp {
105            Ok(r) if r.status().is_success() => {
106                peer_store.update_status(peer_id, "enrolled")?;
107                tracing::info!(project = %project, server = %server_url, "cloud enrollment successful");
108                Ok(CloudResult {
109                    success: true,
110                    message: format!("Enrolled with cloud server at {}", server_url),
111                    project: project.to_string(),
112                    memories_synced: hello.memory_count,
113                    conflicts_resolved: 0,
114                    duration_ms,
115                })
116            }
117            Ok(r) => {
118                let status = r.status().as_u16();
119                let body = r.text().await.unwrap_or_default();
120                peer_store.update_status(peer_id, &format!("enroll_failed_{}", status))?;
121                Err(crate::error::MnemeError::SyncFailed {
122                    peer: server_url.to_string(),
123                    message: format!("enrollment failed ({}): {}", status, body),
124                })
125            }
126            Err(e) => {
127                let msg = e.to_string();
128                peer_store.update_status(peer_id, "enroll_error")?;
129                Err(crate::error::MnemeError::SyncFailed {
130                    peer: server_url.to_string(),
131                    message: msg,
132                })
133            }
134        }
135    }
136
137    /// Ejecuta un ciclo de sync cloud completo.
138    pub async fn sync_cloud(&self, project: &str) -> crate::error::Result<CloudResult> {
139        let start = std::time::Instant::now();
140        let mut memories_synced = 0u32;
141        let mut conflicts_resolved = 0u32;
142
143        let peers = self.db.peers().list(project)?;
144        let cloud_peers: Vec<&Peer> = peers
145            .iter()
146            .filter(|p| matches!(p.transport, TransportType::Http) && p.auto_sync)
147            .collect();
148
149        if cloud_peers.is_empty() {
150            return Ok(CloudResult {
151                success: true,
152                message: "No cloud peers configured".to_string(),
153                project: project.to_string(),
154                memories_synced: 0,
155                conflicts_resolved: 0,
156                duration_ms: 0,
157            });
158        }
159
160        let engine = SyncEngine::new(self.db.clone(), self.config.clone())?;
161
162        for peer in cloud_peers {
163            match engine.sync_with_peer(peer).await {
164                Ok(result) => {
165                    self.db.peers().record_sync(&result, project)?;
166                    memories_synced += result.memories_sent + result.memories_received;
167                    conflicts_resolved += result.conflicts_resolved;
168                }
169                Err(e) => {
170                    tracing::warn!(peer = %peer.name, error = %e, "cloud sync failed");
171                }
172            }
173        }
174
175        let duration_ms = start.elapsed().as_millis() as u64;
176        Ok(CloudResult {
177            success: true,
178            message: format!("Synced {} memories", memories_synced),
179            project: project.to_string(),
180            memories_synced,
181            conflicts_resolved,
182            duration_ms,
183        })
184    }
185
186    /// Obtiene el estado del sync cloud.
187    pub fn cloud_status(&self, project: &str) -> crate::error::Result<serde_json::Value> {
188        let peers = self.db.peers().list(project)?;
189        let cloud_peers: Vec<&Peer> = peers
190            .iter()
191            .filter(|p| matches!(p.transport, TransportType::Http))
192            .collect();
193
194        // Get latest sync log entries
195        let conn = self.db.get_conn();
196        let sync_log: Vec<serde_json::Value>;
197        {
198            let conn_guard = conn.lock().map_err(|_| {
199                crate::error::MnemeError::Config("mutex poisoned".into())
200            })?;
201            let mut stmt = conn_guard
202                .prepare(
203                    "SELECT peer_id, direction, status, memories_sent, memories_received, 
204                            conflicts_resolved, duration_ms, error, finished_at
205                     FROM sync_log WHERE project = ?1
206                     ORDER BY finished_at DESC LIMIT 10",
207                )
208                .ok();
209            sync_log = if let Some(ref mut stmt) = stmt {
210                stmt.query_map(rusqlite::params![project], |row| {
211                    Ok(serde_json::json!({
212                        "peer": row.get::<_, String>(0).unwrap_or_default(),
213                        "direction": row.get::<_, String>(1).unwrap_or_default(),
214                        "status": row.get::<_, String>(2).unwrap_or_default(),
215                        "sent": row.get::<_, i32>(3).unwrap_or(0),
216                        "received": row.get::<_, i32>(4).unwrap_or(0),
217                        "conflicts": row.get::<_, i32>(5).unwrap_or(0),
218                        "duration_ms": row.get::<_, i32>(6).unwrap_or(0),
219                        "error": row.get::<_, Option<String>>(7).unwrap_or(None),
220                        "at": row.get::<_, String>(8).unwrap_or_default(),
221                    }))
222                })
223                .ok()
224                .map(|rows| rows.filter_map(|r| r.ok()).collect())
225                .unwrap_or_default()
226            } else {
227                vec![]
228            };
229        }
230
231        Ok(serde_json::json!({
232            "project": project,
233            "cloud_peers": cloud_peers.len(),
234            "peers": cloud_peers.iter().map(|p| serde_json::json!({
235                "name": p.name,
236                "address": p.address,
237                "last_sync": p.last_sync,
238                "last_status": p.last_status,
239                "auto_sync": p.auto_sync,
240            })).collect::<Vec<_>>(),
241            "recent_syncs": sync_log,
242        }))
243    }
244
245    /// Inicia el autosync background task.
246    pub fn start_autosync(
247        db: Arc<Database>,
248        config: crate::config::settings::SyncConfig,
249        interval_secs: u64,
250    ) -> tokio::task::JoinHandle<()> {
251        tokio::spawn(async move {
252            let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs));
253            loop {
254                interval.tick().await;
255                let engine = match SyncEngine::new(db.clone(), config.clone()) {
256                    Ok(e) => e,
257                    Err(_) => continue,
258                };
259                // Auto-sync all projects
260                if let Ok(projects) = db.memories().list_projects() {
261                    for proj in projects {
262                        if let Err(e) = engine.sync_auto(&proj.name).await {
263                            tracing::warn!(project = %proj.name, error = %e, "autosync failed");
264                        }
265                    }
266                }
267            }
268        })
269    }
270}
mneme/cloud.rs

mneme/
cloud.rs
