1use std::collections::HashMap;
2use std::sync::Arc;
3
4use chrono::{
5 DateTime,
6 Utc,
7};
8use serde::{
9 Deserialize,
10 Serialize,
11};
12use tokio::sync::RwLock;
13use tracing::{
14 debug,
15 info,
16};
17
18use crate::domain::{
19 DomainError,
20 DomainResult,
21};
22use crate::infrastructure::storage::{
23 ObjectMetadata,
24 StorageBackend,
25};
26
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub enum SyncDirection {
29 Push,
30 Pull,
31 Bidirectional,
32}
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
35pub enum ConflictResolution {
36 #[default]
37 LastWriteWins,
38 PreferLocal,
39 PreferRemote,
40 Skip,
41}
42
43#[derive(Debug, Clone)]
44pub struct SyncConfig {
45 pub direction: SyncDirection,
46 pub conflict_resolution: ConflictResolution,
47 pub include_prefixes: Vec<String>,
48 pub exclude_prefixes: Vec<String>,
49 pub delete_orphaned: bool,
50 pub dry_run: bool,
51}
52
53impl Default for SyncConfig {
54 fn default() -> Self {
55 Self {
56 direction: SyncDirection::Bidirectional,
57 conflict_resolution: ConflictResolution::LastWriteWins,
58 include_prefixes: vec![],
59 exclude_prefixes: vec![
60 "pipedash.db".to_string(),
61 "metrics.db".to_string(),
62 "cache/".to_string(),
63 ],
64 delete_orphaned: false,
65 dry_run: false,
66 }
67 }
68}
69
70#[derive(Debug, Clone, Serialize, Deserialize)]
71pub struct SyncFileResult {
72 pub key: String,
73 pub action: SyncAction,
74 pub success: bool,
75 pub error: Option<String>,
76}
77
78#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
79pub enum SyncAction {
80 Pushed,
81 Pulled,
82 Skipped,
83 DeletedRemote,
84 DeletedLocal,
85 NoChange,
86 Conflict,
87}
88
89#[derive(Debug, Clone, Default, Serialize, Deserialize)]
90pub struct SyncResult {
91 pub pushed: usize,
92 pub pulled: usize,
93 pub skipped: usize,
94 pub conflicts: usize,
95 pub deleted: usize,
96 pub errors: usize,
97 pub files: Vec<SyncFileResult>,
98 pub started_at: DateTime<Utc>,
99 pub completed_at: Option<DateTime<Utc>>,
100}
101
102#[derive(Debug, Clone, Default, Serialize, Deserialize)]
103pub struct SyncState {
104 pub last_sync: Option<DateTime<Utc>>,
105 pub file_checksums: HashMap<String, String>,
106 pub file_timestamps: HashMap<String, DateTime<Utc>>,
107}
108
109pub struct SyncManager {
110 local: Arc<dyn StorageBackend>,
111 remote: Option<Arc<dyn StorageBackend>>,
112 state: RwLock<SyncState>,
113 default_config: SyncConfig,
114}
115
116impl SyncManager {
117 pub fn new(local: Arc<dyn StorageBackend>) -> Self {
118 Self {
119 local,
120 remote: None,
121 state: RwLock::new(SyncState::default()),
122 default_config: SyncConfig::default(),
123 }
124 }
125
126 pub fn with_remote(local: Arc<dyn StorageBackend>, remote: Arc<dyn StorageBackend>) -> Self {
127 Self {
128 local,
129 remote: Some(remote),
130 state: RwLock::new(SyncState::default()),
131 default_config: SyncConfig::default(),
132 }
133 }
134
135 pub fn set_remote(&mut self, remote: Arc<dyn StorageBackend>) {
136 self.remote = Some(remote);
137 }
138
139 pub fn clear_remote(&mut self) {
140 self.remote = None;
141 }
142
143 pub fn has_remote(&self) -> bool {
144 self.remote.is_some()
145 }
146
147 pub fn remote_type(&self) -> Option<&str> {
148 self.remote.as_ref().map(|r| r.backend_type())
149 }
150
151 pub fn set_default_config(&mut self, config: SyncConfig) {
152 self.default_config = config;
153 }
154
155 fn should_sync_key(&self, key: &str, config: &SyncConfig) -> bool {
156 for prefix in &config.exclude_prefixes {
157 if key.starts_with(prefix) {
158 return false;
159 }
160 }
161
162 if config.include_prefixes.is_empty() {
163 return true;
164 }
165
166 for prefix in &config.include_prefixes {
167 if key.starts_with(prefix) {
168 return true;
169 }
170 }
171
172 false
173 }
174
175 fn determine_action(
176 &self, _key: &str, local_meta: Option<&ObjectMetadata>,
177 remote_meta: Option<&ObjectMetadata>, config: &SyncConfig,
178 ) -> SyncAction {
179 match (local_meta, remote_meta) {
180 (None, None) => SyncAction::NoChange,
181 (Some(_), None) => match config.direction {
182 SyncDirection::Push | SyncDirection::Bidirectional => SyncAction::Pushed,
183 SyncDirection::Pull => {
184 if config.delete_orphaned {
185 SyncAction::DeletedLocal
186 } else {
187 SyncAction::Skipped
188 }
189 }
190 },
191 (None, Some(_)) => match config.direction {
192 SyncDirection::Pull | SyncDirection::Bidirectional => SyncAction::Pulled,
193 SyncDirection::Push => {
194 if config.delete_orphaned {
195 SyncAction::DeletedRemote
196 } else {
197 SyncAction::Skipped
198 }
199 }
200 },
201 (Some(local), Some(remote)) => {
202 let local_id = local.etag.as_ref().unwrap_or(&"".to_string()).clone();
203 let remote_id = remote.etag.as_ref().unwrap_or(&"".to_string()).clone();
204
205 if !local_id.is_empty() && !remote_id.is_empty() && local_id == remote_id {
206 return SyncAction::NoChange;
207 }
208
209 match config.conflict_resolution {
210 ConflictResolution::LastWriteWins => {
211 if local.last_modified > remote.last_modified {
212 SyncAction::Pushed
213 } else if remote.last_modified > local.last_modified {
214 SyncAction::Pulled
215 } else {
216 SyncAction::NoChange
217 }
218 }
219 ConflictResolution::PreferLocal => SyncAction::Pushed,
220 ConflictResolution::PreferRemote => SyncAction::Pulled,
221 ConflictResolution::Skip => SyncAction::Conflict,
222 }
223 }
224 }
225 }
226
227 pub async fn sync(&self, config: Option<SyncConfig>) -> DomainResult<SyncResult> {
228 let config = config.unwrap_or_else(|| self.default_config.clone());
229 let mut result = SyncResult {
230 started_at: Utc::now(),
231 ..Default::default()
232 };
233
234 let remote = match &self.remote {
235 Some(r) => r,
236 None => {
237 info!("No remote storage configured, skipping sync");
238 result.completed_at = Some(Utc::now());
239 return Ok(result);
240 }
241 };
242
243 if !remote.is_available().await {
244 return Err(DomainError::NetworkError(
245 "Remote storage is not available".to_string(),
246 ));
247 }
248
249 info!(
250 "Starting sync: direction={:?}, conflict_resolution={:?}",
251 config.direction, config.conflict_resolution
252 );
253
254 let local_files = self.local.list(None).await?;
255 let remote_files = remote.list(None).await?;
256
257 let local_map: HashMap<String, ObjectMetadata> = local_files
258 .into_iter()
259 .filter(|f| self.should_sync_key(&f.key, &config))
260 .map(|f| (f.key.clone(), f))
261 .collect();
262
263 let remote_map: HashMap<String, ObjectMetadata> = remote_files
264 .into_iter()
265 .filter(|f| self.should_sync_key(&f.key, &config))
266 .map(|f| (f.key.clone(), f))
267 .collect();
268
269 let mut all_keys: Vec<String> = local_map.keys().cloned().collect();
270 for key in remote_map.keys() {
271 if !all_keys.contains(key) {
272 all_keys.push(key.clone());
273 }
274 }
275
276 for key in all_keys {
277 let local_meta = local_map.get(&key);
278 let remote_meta = remote_map.get(&key);
279 let action = self.determine_action(&key, local_meta, remote_meta, &config);
280
281 debug!("Processing {}: action={:?}", key, action);
282
283 if config.dry_run {
284 result.files.push(SyncFileResult {
285 key: key.clone(),
286 action,
287 success: true,
288 error: None,
289 });
290 match action {
291 SyncAction::Pushed => result.pushed += 1,
292 SyncAction::Pulled => result.pulled += 1,
293 SyncAction::Skipped | SyncAction::NoChange => result.skipped += 1,
294 SyncAction::Conflict => result.conflicts += 1,
295 SyncAction::DeletedLocal | SyncAction::DeletedRemote => result.deleted += 1,
296 }
297 continue;
298 }
299
300 let file_result = match action {
301 SyncAction::Pushed => self.push_file(&key, remote.as_ref()).await,
302 SyncAction::Pulled => self.pull_file(&key, remote.as_ref()).await,
303 SyncAction::DeletedRemote => self.delete_remote(&key, remote.as_ref()).await,
304 SyncAction::DeletedLocal => self.delete_local(&key).await,
305 SyncAction::NoChange | SyncAction::Skipped | SyncAction::Conflict => {
306 Ok(SyncFileResult {
307 key: key.clone(),
308 action,
309 success: true,
310 error: None,
311 })
312 }
313 };
314
315 match file_result {
316 Ok(file_res) => {
317 match file_res.action {
318 SyncAction::Pushed => result.pushed += 1,
319 SyncAction::Pulled => result.pulled += 1,
320 SyncAction::Skipped | SyncAction::NoChange => result.skipped += 1,
321 SyncAction::Conflict => result.conflicts += 1,
322 SyncAction::DeletedLocal | SyncAction::DeletedRemote => result.deleted += 1,
323 }
324 result.files.push(file_res);
325 }
326 Err(e) => {
327 result.errors += 1;
328 result.files.push(SyncFileResult {
329 key: key.clone(),
330 action,
331 success: false,
332 error: Some(e.to_string()),
333 });
334 }
335 }
336 }
337
338 {
339 let mut state = self.state.write().await;
340 state.last_sync = Some(Utc::now());
341 }
342
343 result.completed_at = Some(Utc::now());
344 info!(
345 "Sync completed: pushed={}, pulled={}, skipped={}, conflicts={}, errors={}",
346 result.pushed, result.pulled, result.skipped, result.conflicts, result.errors
347 );
348
349 Ok(result)
350 }
351
352 async fn push_file(
353 &self, key: &str, remote: &dyn StorageBackend,
354 ) -> DomainResult<SyncFileResult> {
355 debug!("Pushing file: {}", key);
356 let data = self.local.get(key).await?;
357 remote.put(key, &data, None).await?;
358
359 Ok(SyncFileResult {
360 key: key.to_string(),
361 action: SyncAction::Pushed,
362 success: true,
363 error: None,
364 })
365 }
366
367 async fn pull_file(
368 &self, key: &str, remote: &dyn StorageBackend,
369 ) -> DomainResult<SyncFileResult> {
370 debug!("Pulling file: {}", key);
371 let data = remote.get(key).await?;
372 self.local.put(key, &data, None).await?;
373
374 Ok(SyncFileResult {
375 key: key.to_string(),
376 action: SyncAction::Pulled,
377 success: true,
378 error: None,
379 })
380 }
381
382 async fn delete_remote(
383 &self, key: &str, remote: &dyn StorageBackend,
384 ) -> DomainResult<SyncFileResult> {
385 debug!("Deleting remote file: {}", key);
386 remote.delete(key).await?;
387
388 Ok(SyncFileResult {
389 key: key.to_string(),
390 action: SyncAction::DeletedRemote,
391 success: true,
392 error: None,
393 })
394 }
395
396 async fn delete_local(&self, key: &str) -> DomainResult<SyncFileResult> {
397 debug!("Deleting local file: {}", key);
398 self.local.delete(key).await?;
399
400 Ok(SyncFileResult {
401 key: key.to_string(),
402 action: SyncAction::DeletedLocal,
403 success: true,
404 error: None,
405 })
406 }
407
408 pub async fn force_push(&self) -> DomainResult<SyncResult> {
409 self.sync(Some(SyncConfig {
410 direction: SyncDirection::Push,
411 conflict_resolution: ConflictResolution::PreferLocal,
412 delete_orphaned: true,
413 ..self.default_config.clone()
414 }))
415 .await
416 }
417
418 pub async fn force_pull(&self) -> DomainResult<SyncResult> {
419 self.sync(Some(SyncConfig {
420 direction: SyncDirection::Pull,
421 conflict_resolution: ConflictResolution::PreferRemote,
422 delete_orphaned: true,
423 ..self.default_config.clone()
424 }))
425 .await
426 }
427
428 pub async fn preview(&self, config: Option<SyncConfig>) -> DomainResult<SyncResult> {
429 let mut config = config.unwrap_or_else(|| self.default_config.clone());
430 config.dry_run = true;
431 self.sync(Some(config)).await
432 }
433
434 pub async fn last_sync_time(&self) -> Option<DateTime<Utc>> {
435 self.state.read().await.last_sync
436 }
437
438 pub async fn get_state(&self) -> SyncState {
439 self.state.read().await.clone()
440 }
441}
442
443#[cfg(test)]
444mod tests {
445 use tempfile::TempDir;
446
447 use super::*;
448 use crate::infrastructure::storage::LocalStorage;
449
450 #[tokio::test]
451 async fn test_sync_manager_creation() {
452 let temp_dir = TempDir::new().unwrap();
453 let storage = Arc::new(LocalStorage::new(temp_dir.path().to_path_buf()));
454 let manager = SyncManager::new(storage);
455
456 assert!(!manager.has_remote());
457 assert!(manager.remote_type().is_none());
458 }
459
460 #[tokio::test]
461 async fn test_sync_without_remote() {
462 let temp_dir = TempDir::new().unwrap();
463 let storage = Arc::new(LocalStorage::new(temp_dir.path().to_path_buf()));
464 let manager = SyncManager::new(storage);
465
466 let result = manager.sync(None).await.unwrap();
467 assert_eq!(result.pushed, 0);
468 assert_eq!(result.pulled, 0);
469 assert_eq!(result.errors, 0);
470 }
471
472 #[tokio::test]
473 async fn test_should_sync_key() {
474 let temp_dir = TempDir::new().unwrap();
475 let storage = Arc::new(LocalStorage::new(temp_dir.path().to_path_buf()));
476 let manager = SyncManager::new(storage);
477
478 let config = SyncConfig {
479 exclude_prefixes: vec!["cache/".to_string(), ".git/".to_string()],
480 include_prefixes: vec![],
481 ..Default::default()
482 };
483
484 assert!(manager.should_sync_key("config/settings.json", &config));
485 assert!(!manager.should_sync_key("cache/temp.bin", &config));
486 assert!(!manager.should_sync_key(".git/config", &config));
487 }
488
489 #[tokio::test]
490 async fn test_sync_between_local_storages() {
491 let local_dir = TempDir::new().unwrap();
492 let remote_dir = TempDir::new().unwrap();
493
494 let local = Arc::new(LocalStorage::new(local_dir.path().to_path_buf()));
495 let remote: Arc<dyn StorageBackend> =
496 Arc::new(LocalStorage::new(remote_dir.path().to_path_buf()));
497
498 local.put("test.txt", b"hello world", None).await.unwrap();
499
500 let manager = SyncManager::with_remote(local.clone(), remote.clone());
501
502 let result = manager
503 .sync(Some(SyncConfig {
504 direction: SyncDirection::Push,
505 ..Default::default()
506 }))
507 .await
508 .unwrap();
509
510 assert_eq!(result.pushed, 1);
511 assert_eq!(result.errors, 0);
512
513 let remote_data = remote.get("test.txt").await.unwrap();
514 assert_eq!(remote_data, b"hello world");
515 }
516}