mneme/sync/
peer.rs

1use std::str::FromStr;
2use std::sync::{Arc, Mutex};
3
4use chrono::{DateTime, Utc};
5use rusqlite::{params, Connection};
6use serde::{Deserialize, Serialize};
7use uuid::Uuid;
8
9use crate::error::{MnemeError, Result};
10
11/// Tipo de transporte para sincronizacion.
12#[derive(Debug, Clone, Serialize, Deserialize)]
13#[serde(rename_all = "lowercase")]
14pub enum TransportType {
15    Http,
16    Ssh,
17    File,
18}
19
20impl std::fmt::Display for TransportType {
21    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
22        let s = match self {
23            TransportType::Http => "http",
24            TransportType::Ssh => "ssh",
25            TransportType::File => "file",
26        };
27        write!(f, "{}", s)
28    }
29}
30
31impl std::str::FromStr for TransportType {
32    type Err = MnemeError;
33
34    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
35        match s.to_lowercase().as_str() {
36            "http" => Ok(TransportType::Http),
37            "ssh" => Ok(TransportType::Ssh),
38            "file" => Ok(TransportType::File),
39            other => Err(MnemeError::UnsupportedTransport(other.to_string())),
40        }
41    }
42}
43
44/// Representa un peer de sincronizacion.
45#[derive(Debug, Clone, Serialize, Deserialize)]
46pub struct Peer {
47    /// ID unico del peer.
48    pub id: Uuid,
49    /// Nombre descriptivo del peer.
50    pub name: String,
51    /// Tipo de transporte (http, ssh, file).
52    pub transport: TransportType,
53    /// Direccion del peer (URL o ruta).
54    pub address: String,
55    /// Proyecto asociado.
56    pub project: String,
57    /// Fecha del ultimo sync.
58    pub last_sync: Option<DateTime<Utc>>,
59    /// Estado del ultimo sync.
60    pub last_status: Option<String>,
61    /// Habilitar auto-sync.
62    pub auto_sync: bool,
63    /// Fecha de creacion.
64    pub created_at: DateTime<Utc>,
65}
66
67/// Store para operaciones de peers.
68pub struct PeerStore {
69    conn: Arc<Mutex<Connection>>,
70}
71
72impl PeerStore {
73    /// Crea un nuevo PeerStore.
74    pub fn new(conn: Arc<Mutex<Connection>>) -> Self {
75        Self { conn }
76    }
77
78    /// Agrega un nuevo peer.
79    pub fn add(&self, peer: &Peer) -> Result<()> {
80        let conn = self
81            .conn
82            .lock()
83            .map_err(|_| MnemeError::Config("mutex poisoned".into()))?;
84        conn.execute(
85            "INSERT INTO sync_peers (id, name, transport, address, project, last_sync, last_status, auto_sync, created_at)
86             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
87             ON CONFLICT(id) DO UPDATE SET
88                 name = excluded.name,
89                 transport = excluded.transport,
90                 address = excluded.address,
91                 project = excluded.project,
92                 auto_sync = excluded.auto_sync",
93            params![
94                peer.id.to_string(),
95                &peer.name,
96                peer.transport.to_string(),
97                &peer.address,
98                &peer.project,
99                peer.last_sync.map(|d| d.to_rfc3339()),
100                peer.last_status.as_deref(),
101                peer.auto_sync as i32,
102                peer.created_at.to_rfc3339(),
103            ],
104        )?;
105        tracing::info!("added peer: {} ({})", peer.id, peer.name);
106        Ok(())
107    }
108
109    /// Elimina un peer por ID.
110    pub fn remove(&self, id: Uuid) -> Result<()> {
111        let conn = self
112            .conn
113            .lock()
114            .map_err(|_| MnemeError::Config("mutex poisoned".into()))?;
115        let rows = conn.execute(
116            "DELETE FROM sync_peers WHERE id = ?1",
117            params![id.to_string()],
118        )?;
119        if rows == 0 {
120            return Err(MnemeError::PeerNotFound(id));
121        }
122        tracing::info!("removed peer: {}", id);
123        Ok(())
124    }
125
126    /// Lista todos los peers de un proyecto.
127    pub fn list(&self, project: &str) -> Result<Vec<Peer>> {
128        let conn = self
129            .conn
130            .lock()
131            .map_err(|_| MnemeError::Config("mutex poisoned".into()))?;
132        let mut stmt = conn.prepare(
133            "SELECT id, name, transport, address, project, last_sync, last_status, auto_sync, created_at
134             FROM sync_peers WHERE project = ?1 ORDER BY created_at DESC",
135        )?;
136        let rows = stmt.query_map(params![project], |row| {
137            Ok(Peer {
138                id: Uuid::parse_str(&row.get::<_, String>(0)?).map_err(|e| {
139                    rusqlite::Error::FromSqlConversionFailure(
140                        0,
141                        rusqlite::types::Type::Text,
142                        Box::new(e),
143                    )
144                })?,
145                name: row.get(1)?,
146                transport: TransportType::from_str(&row.get::<_, String>(2)?).map_err(|e| {
147                    rusqlite::Error::FromSqlConversionFailure(
148                        2,
149                        rusqlite::types::Type::Text,
150                        Box::new(e),
151                    )
152                })?,
153                address: row.get(3)?,
154                project: row.get(4)?,
155                last_sync: row
156                    .get::<_, Option<String>>(5)?
157                    .and_then(|s| DateTime::parse_from_rfc3339(&s).ok())
158                    .map(|d| d.with_timezone(&Utc)),
159                last_status: row.get(6)?,
160                auto_sync: row.get::<_, i32>(7)? != 0,
161                created_at: DateTime::parse_from_rfc3339(&row.get::<_, String>(8)?)
162                    .map_err(|e| {
163                        rusqlite::Error::FromSqlConversionFailure(
164                            8,
165                            rusqlite::types::Type::Text,
166                            Box::new(e),
167                        )
168                    })?
169                    .with_timezone(&Utc),
170            })
171        })?;
172
173        let mut peers = Vec::new();
174        for row in rows {
175            peers.push(row?);
176        }
177        Ok(peers)
178    }
179
180    /// Obtiene un peer por ID.
181    pub fn get(&self, id: Uuid) -> Result<Option<Peer>> {
182        let conn = self
183            .conn
184            .lock()
185            .map_err(|_| MnemeError::Config("mutex poisoned".into()))?;
186        let mut stmt = conn.prepare(
187            "SELECT id, name, transport, address, project, last_sync, last_status, auto_sync, created_at
188             FROM sync_peers WHERE id = ?1",
189        )?;
190        let result = stmt.query_row(params![id.to_string()], |row| {
191            Ok(Peer {
192                id: Uuid::parse_str(&row.get::<_, String>(0)?).map_err(|e| {
193                    rusqlite::Error::FromSqlConversionFailure(
194                        0,
195                        rusqlite::types::Type::Text,
196                        Box::new(e),
197                    )
198                })?,
199                name: row.get(1)?,
200                transport: TransportType::from_str(&row.get::<_, String>(2)?).map_err(|e| {
201                    rusqlite::Error::FromSqlConversionFailure(
202                        2,
203                        rusqlite::types::Type::Text,
204                        Box::new(e),
205                    )
206                })?,
207                address: row.get(3)?,
208                project: row.get(4)?,
209                last_sync: row
210                    .get::<_, Option<String>>(5)?
211                    .and_then(|s| DateTime::parse_from_rfc3339(&s).ok())
212                    .map(|d| d.with_timezone(&Utc)),
213                last_status: row.get(6)?,
214                auto_sync: row.get::<_, i32>(7)? != 0,
215                created_at: DateTime::parse_from_rfc3339(&row.get::<_, String>(8)?)
216                    .map_err(|e| {
217                        rusqlite::Error::FromSqlConversionFailure(
218                            8,
219                            rusqlite::types::Type::Text,
220                            Box::new(e),
221                        )
222                    })?
223                    .with_timezone(&Utc),
224            })
225        });
226
227        match result {
228            Ok(peer) => Ok(Some(peer)),
229            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
230            Err(e) => Err(e.into()),
231        }
232    }
233
234    /// Actualiza el estado del ultimo sync de un peer.
235    pub fn update_status(&self, id: Uuid, status: &str) -> Result<()> {
236        let conn = self
237            .conn
238            .lock()
239            .map_err(|_| MnemeError::Config("mutex poisoned".into()))?;
240        let rows = conn.execute(
241            "UPDATE sync_peers SET last_sync = ?1, last_status = ?2 WHERE id = ?3",
242            params![Utc::now().to_rfc3339(), status, id.to_string()],
243        )?;
244        if rows == 0 {
245            return Err(MnemeError::PeerNotFound(id));
246        }
247        tracing::info!("updated peer status: {} -> {}", id, status);
248        Ok(())
249    }
250
251    /// Registra el resultado de un sync en el log.
252    pub fn record_sync(
253        &self,
254        result: &crate::sync::protocol::SyncResult,
255        project: &str,
256    ) -> Result<()> {
257        let conn = self
258            .conn
259            .lock()
260            .map_err(|_| MnemeError::Config("mutex poisoned".into()))?;
261        let id = Uuid::new_v4();
262        let direction = match result.direction {
263            crate::sync::protocol::SyncDirection::Push => "push",
264            crate::sync::protocol::SyncDirection::Pull => "pull",
265            crate::sync::protocol::SyncDirection::Bidirectional => "bidirectional",
266        };
267        let status = match result.status {
268            crate::sync::protocol::SyncStatus::Ok => "ok",
269            crate::sync::protocol::SyncStatus::Error => "error",
270            crate::sync::protocol::SyncStatus::Partial => "partial",
271        };
272        conn.execute(
273            "INSERT INTO sync_log (id, peer_id, direction, project, memories_sent, memories_received, conflicts_resolved, duration_ms, status, error, started_at, finished_at)
274             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
275            params![
276                id.to_string(),
277                result.peer_name,
278                direction,
279                project,
280                result.memories_sent as i32,
281                result.memories_received as i32,
282                result.conflicts_resolved as i32,
283                result.duration_ms as i32,
284                status,
285                result.error.as_deref(),
286                Utc::now().to_rfc3339(),
287                Utc::now().to_rfc3339(),
288            ],
289        )?;
290        Ok(())
291    }
292}
mneme/sync/peer.rs

mneme/sync/
peer.rs
