mneme/sync/
engine.rs

1use std::collections::HashMap;
2use std::sync::Arc;
3
4use chrono::Utc;
5use rusqlite::params;
6use uuid::Uuid;
7
8use crate::config::settings::SyncConfig;
9use crate::error::{MnemeError, Result};
10use crate::store::db::Database;
11use crate::sync::crdt;
12use crate::sync::peer::{Peer, TransportType};
13use crate::sync::protocol::{
14    ApplyStats, ExportStats, MemoryChangeset, SyncDirection, SyncHello, SyncRequest, SyncResponse,
15    SyncResult, SyncStatus,
16};
17use crate::sync::transport::file::FileTransport;
18use crate::sync::transport::http::HttpTransport;
19
20/// Motor de sincronizacion CRDT.
21pub struct SyncEngine {
22    db: Arc<Database>,
23    config: SyncConfig,
24}
25
26impl SyncEngine {
27    /// Crea un nuevo SyncEngine.
28    pub fn new(db: Arc<Database>, config: SyncConfig) -> Result<Self> {
29        if !config.enabled {
30            return Err(MnemeError::SyncDisabled);
31        }
32        Ok(Self { db, config })
33    }
34
35    /// Sincroniza con un peer especifico.
36    pub async fn sync_with_peer(&self, peer: &Peer) -> Result<SyncResult> {
37        let start = std::time::Instant::now();
38        let project = peer.project.clone();
39
40        let result = match peer.transport {
41            TransportType::Http => self.sync_http(peer, &project).await,
42            TransportType::File => self.sync_file(peer, &project).await,
43            TransportType::Ssh => Err(MnemeError::UnsupportedTransport("ssh".to_string())),
44        };
45
46        let duration_ms = start.elapsed().as_millis() as u64;
47
48        match result {
49            Ok(mut r) => {
50                r.duration_ms = duration_ms;
51                let peer_store = self.db.peers();
52                peer_store.update_status(peer.id, "ok")?;
53                Ok(r)
54            }
55            Err(e) => {
56                let peer_store = self.db.peers();
57                let _ = peer_store.update_status(peer.id, "error");
58                Ok(SyncResult {
59                    peer_name: peer.name.clone(),
60                    direction: SyncDirection::Bidirectional,
61                    memories_sent: 0,
62                    memories_received: 0,
63                    conflicts_resolved: 0,
64                    duration_ms,
65                    status: SyncStatus::Error,
66                    error: Some(e.to_string()),
67                })
68            }
69        }
70    }
71
72    /// Sincroniza automaticamente todos los peers auto_sync de un proyecto.
73    pub async fn sync_auto(&self, project: &str) -> Result<Vec<SyncResult>> {
74        let peers = self.db.peers().list(project)?;
75        let mut results = Vec::new();
76
77        for peer in peers {
78            if peer.auto_sync {
79                match self.sync_with_peer(&peer).await {
80                    Ok(r) => results.push(r),
81                    Err(e) => {
82                        tracing::warn!("auto sync failed for {}: {}", peer.name, e);
83                        results.push(SyncResult {
84                            peer_name: peer.name.clone(),
85                            direction: SyncDirection::Bidirectional,
86                            memories_sent: 0,
87                            memories_received: 0,
88                            conflicts_resolved: 0,
89                            duration_ms: 0,
90                            status: SyncStatus::Error,
91                            error: Some(e.to_string()),
92                        });
93                    }
94                }
95            }
96        }
97
98        Ok(results)
99    }
100
101    /// Exporta un proyecto a archivo.
102    pub fn export_project(
103        &self,
104        project: &str,
105        output: Option<std::path::PathBuf>,
106    ) -> Result<ExportStats> {
107        let conn_arc = self.db.get_conn();
108        let conn = conn_arc
109            .lock()
110            .map_err(|_| MnemeError::Config("mutex poisoned".into()))?;
111
112        let mut stmt = conn.prepare(
113            "SELECT automerge_id, doc_bytes FROM sync_state
114             JOIN memories ON sync_state.memory_id = memories.id
115             WHERE memories.project = ?1 AND sync_state.is_tombstone = 0",
116        )?;
117
118        let rows = stmt.query_map(params![project], |row| {
119            Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?))
120        })?;
121
122        let mut changes = Vec::new();
123        for row in rows {
124            let (automerge_id, doc_bytes) = row?;
125            changes.push(MemoryChangeset {
126                automerge_id,
127                payload: doc_bytes,
128                is_full_doc: true,
129            });
130        }
131
132        let dir = output.unwrap_or_else(|| {
133            let mut path =
134                std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from("."));
135            path.push("sync_exports");
136            path
137        });
138
139        let transport = FileTransport::new(dir)?;
140        let (_, stats) = transport.export(project, &changes)?;
141
142        tracing::info!(
143            "exported project {}: {} memories",
144            project,
145            stats.memories_exported
146        );
147        Ok(ExportStats {
148            memories_exported: stats.memories_exported,
149            bytes_written: stats.bytes_written,
150        })
151    }
152
153    /// Construye mensaje de saludo.
154    pub fn build_hello(&self, project: &str) -> Result<SyncHello> {
155        let conn_arc = self.db.get_conn();
156        let conn = conn_arc
157            .lock()
158            .map_err(|_| MnemeError::Config("mutex poisoned".into()))?;
159
160        let count: u32 = conn.query_row(
161            "SELECT COUNT(*) FROM memories WHERE project = ?1 AND deleted_at IS NULL",
162            params![project],
163            |row| row.get(0),
164        )?;
165
166        let mut heads = HashMap::new();
167        let mut stmt = conn.prepare(
168            "SELECT automerge_id, doc_bytes FROM sync_state
169             JOIN memories ON sync_state.memory_id = memories.id
170             WHERE memories.project = ?1",
171        )?;
172        let rows = stmt.query_map(params![project], |row| {
173            Ok((row.get::<_, String>(0)?, row.get::<_, Vec<u8>>(1)?))
174        })?;
175
176        for row in rows {
177            let (automerge_id, doc_bytes) = row?;
178            match crdt::get_heads(&doc_bytes) {
179                Ok(h) => {
180                    heads.insert(automerge_id, h);
181                }
182                Err(e) => {
183                    tracing::warn!("failed to get heads: {}", e);
184                }
185            }
186        }
187
188        let peer_id = if self.config.peer_id.is_empty() {
189            Uuid::new_v4()
190        } else {
191            Uuid::parse_str(&self.config.peer_id).unwrap_or_else(|_| Uuid::new_v4())
192        };
193
194        Ok(SyncHello {
195            peer_id,
196            peer_name: self.config.peer_name.clone(),
197            project: project.to_string(),
198            mneme_version: env!("CARGO_PKG_VERSION").to_string(),
199            memory_count: count,
200            heads,
201        })
202    }
203
204    /// Aplica una respuesta de sync entrante.
205    pub fn apply_response(&self, response: &SyncResponse) -> Result<ApplyStats> {
206        let mut applied = 0u32;
207        let mut conflicts = 0u32;
208
209        for change in &response.changes {
210            if let Err(e) = self.apply_changeset(change) {
211                tracing::warn!("failed to apply changeset: {}", e);
212                conflicts += 1;
213            } else {
214                applied += 1;
215            }
216        }
217
218        Ok(ApplyStats {
219            memories_applied: applied,
220            conflicts_resolved: conflicts,
221        })
222    }
223
224    /// Construye respuesta a una solicitud de sync (placeholder).
225    pub fn build_response(&self, request: &SyncRequest) -> Result<SyncResponse> {
226        Ok(SyncResponse {
227            project: request.project.clone(),
228            changes: Vec::new(),
229            tombstones: Vec::new(),
230        })
231    }
232
233    async fn sync_http(&self, peer: &Peer, project: &str) -> Result<SyncResult> {
234        let transport = HttpTransport::new(peer.address.clone())?;
235        let hello = self.build_hello(project)?;
236        let _remote_hello = transport.hello(&hello).await?;
237
238        let request = SyncRequest {
239            project: project.to_string(),
240            have: hello.heads,
241        };
242
243        let response = transport.pull(&request).await?;
244        let apply_stats = self.apply_response(&response)?;
245
246        let push_response = self.build_response(&request)?;
247        transport.push(&push_response).await?;
248
249        Ok(SyncResult {
250            peer_name: peer.name.clone(),
251            direction: SyncDirection::Bidirectional,
252            memories_sent: push_response.changes.len() as u32,
253            memories_received: response.changes.len() as u32,
254            conflicts_resolved: apply_stats.conflicts_resolved,
255            duration_ms: 0,
256            status: SyncStatus::Ok,
257            error: None,
258        })
259    }
260
261    async fn sync_file(&self, peer: &Peer, project: &str) -> Result<SyncResult> {
262        let dir = std::path::PathBuf::from(&peer.address);
263        let transport = FileTransport::new(dir)?;
264        let (changes, _) = transport.import_pending(project)?;
265
266        let mut applied = 0u32;
267        let mut conflicts = 0u32;
268
269        for change in changes {
270            if let Err(e) = self.apply_changeset(&change) {
271                tracing::warn!("failed to apply file changeset: {}", e);
272                conflicts += 1;
273            } else {
274                applied += 1;
275            }
276        }
277
278        Ok(SyncResult {
279            peer_name: peer.name.clone(),
280            direction: SyncDirection::Pull,
281            memories_sent: 0,
282            memories_received: applied + conflicts,
283            conflicts_resolved: conflicts,
284            duration_ms: 0,
285            status: if conflicts > 0 {
286                SyncStatus::Partial
287            } else {
288                SyncStatus::Ok
289            },
290            error: None,
291        })
292    }
293
294    fn apply_changeset(&self, change: &MemoryChangeset) -> Result<()> {
295        let conn_arc = self.db.get_conn();
296        let conn = conn_arc
297            .lock()
298            .map_err(|_| MnemeError::Config("mutex poisoned".into()))?;
299
300        if change.is_full_doc {
301            let memory = crdt::doc_to_memory(&change.payload)?;
302            let exists: bool = conn
303                .query_row(
304                    "SELECT 1 FROM sync_state WHERE automerge_id = ?1",
305                    params![&change.automerge_id],
306                    |_| Ok(true),
307                )
308                .unwrap_or(false);
309
310            if exists {
311                let existing_bytes: Vec<u8> = conn.query_row(
312                    "SELECT doc_bytes FROM sync_state WHERE automerge_id = ?1",
313                    params![&change.automerge_id],
314                    |row| row.get(0),
315                )?;
316                let merged = crdt::merge_docs(&existing_bytes, &change.payload)?;
317                conn.execute(
318                    "UPDATE sync_state SET doc_bytes = ?1 WHERE automerge_id = ?2",
319                    params![merged, &change.automerge_id],
320                )?;
321            } else {
322                conn.execute(
323                    "INSERT INTO sync_state (memory_id, automerge_id, doc_bytes, last_synced, is_tombstone)
324                     VALUES (?1, ?2, ?3, ?4, 0)
325                     ON CONFLICT(automerge_id) DO UPDATE SET
326                         doc_bytes = excluded.doc_bytes,
327                         last_synced = excluded.last_synced",
328                    params![
329                        memory.id.to_string(),
330                        &change.automerge_id,
331                        &change.payload,
332                        Utc::now().to_rfc3339(),
333                    ],
334                )?;
335            }
336        }
337
338        Ok(())
339    }
340
341    /// Crea o actualiza estado sync para una memoria existente.
342    pub fn ensure_sync_state(&self, memory_id: Uuid, automerge_id: &str) -> Result<()> {
343        let conn_arc = self.db.get_conn();
344        let conn = conn_arc
345            .lock()
346            .map_err(|_| MnemeError::Config("mutex poisoned".into()))?;
347
348        let exists: bool = conn
349            .query_row(
350                "SELECT 1 FROM sync_state WHERE memory_id = ?1",
351                params![memory_id.to_string()],
352                |_| Ok(true),
353            )
354            .unwrap_or(false);
355
356        if !exists {
357            let memory = match self.db.memories().get(memory_id)? {
358                Some(m) => m,
359                None => return Err(MnemeError::NotFound(memory_id)),
360            };
361
362            let mut doc = crdt::memory_to_doc(&memory)?;
363            let doc_bytes = crdt::doc_to_bytes(&mut doc)?;
364
365            conn.execute(
366                "INSERT INTO sync_state (memory_id, automerge_id, doc_bytes, last_synced, is_tombstone)
367                 VALUES (?1, ?2, ?3, ?4, 0)
368                 ON CONFLICT(memory_id) DO UPDATE SET
369                     automerge_id = excluded.automerge_id,
370                     doc_bytes = excluded.doc_bytes,
371                     last_synced = excluded.last_synced",
372                params![
373                    memory_id.to_string(),
374                    automerge_id,
375                    doc_bytes,
376                    Utc::now().to_rfc3339(),
377                ],
378            )?;
379        }
380
381        Ok(())
382    }
383}
mneme/sync/engine.rs

mneme/sync/
engine.rs
