pipedash_core/infrastructure/migration/
backup.rs

1use std::path::{
2    Path,
3    PathBuf,
4};
5
6use chrono::Utc;
7use serde::{
8    Deserialize,
9    Serialize,
10};
11use sqlx::sqlite::SqliteConnectOptions;
12use sqlx::{
13    Connection,
14    SqliteConnection,
15};
16
17use crate::domain::{
18    DomainError,
19    DomainResult,
20};
21use crate::infrastructure::config::{
22    PipedashConfig,
23    StorageBackend,
24};
25use crate::infrastructure::ConfigBackend;
26
27#[derive(Debug, Clone, Serialize, Deserialize)]
28pub struct BackupPaths {
29    pub source_db: Option<PathBuf>,
30    pub source_config: PathBuf,
31    pub target_db: Option<PathBuf>,
32    pub timestamp: String,
33}
34
35pub struct BackupManager {
36    data_dir: PathBuf,
37}
38
39impl BackupManager {
40    pub fn new(data_dir: PathBuf) -> Self {
41        Self { data_dir }
42    }
43
44    async fn backup_sqlite_database(src: &Path, dst: &Path) -> DomainResult<()> {
45        let destination = dst.to_str().ok_or_else(|| {
46            DomainError::DatabaseError(format!("Backup path is not valid UTF-8: {}", dst.display()))
47        })?;
48        let mut connection = SqliteConnection::connect_with(
49            &SqliteConnectOptions::new().filename(src).read_only(true),
50        )
51        .await
52        .map_err(|e| {
53            DomainError::DatabaseError(format!("Failed to open SQLite database for backup: {}", e))
54        })?;
55
56        sqlx::query("VACUUM INTO ?")
57            .bind(destination)
58            .execute(&mut connection)
59            .await
60            .map_err(|e| {
61                DomainError::DatabaseError(format!("Failed to backup SQLite database: {}", e))
62            })?;
63
64        connection.close().await.map_err(|e| {
65            DomainError::DatabaseError(format!("Failed to close SQLite backup connection: {}", e))
66        })
67    }
68
69    async fn restore_sqlite_database(backup: &Path, dst: &Path) -> DomainResult<()> {
70        let source = backup.to_str().ok_or_else(|| {
71            DomainError::DatabaseError(format!(
72                "Backup path is not valid UTF-8: {}",
73                backup.display()
74            ))
75        })?;
76        let restore_error = |e: sqlx::Error| {
77            DomainError::DatabaseError(format!("Failed to restore SQLite database: {}", e))
78        };
79        let mut connection = SqliteConnection::connect_with(
80            &SqliteConnectOptions::new()
81                .filename(dst)
82                .foreign_keys(false),
83        )
84        .await
85        .map_err(restore_error)?;
86
87        sqlx::query("ATTACH DATABASE ? AS backup")
88            .bind(source)
89            .execute(&mut connection)
90            .await
91            .map_err(restore_error)?;
92        let tables: Vec<String> = sqlx::query_scalar(
93            "SELECT name FROM backup.sqlite_master WHERE type = 'table' \
94             AND (name NOT LIKE 'sqlite\\_%' ESCAPE '\\' OR name = 'sqlite_sequence')",
95        )
96        .fetch_all(&mut connection)
97        .await
98        .map_err(restore_error)?;
99
100        let mut transaction = connection.begin().await.map_err(restore_error)?;
101        for table in &tables {
102            let table = table.replace('"', "\"\"");
103            let sql = format!(
104                "DELETE FROM main.\"{0}\"; INSERT INTO main.\"{0}\" SELECT * FROM backup.\"{0}\"",
105                table
106            );
107            sqlx::raw_sql(sqlx::AssertSqlSafe(sql))
108                .execute(&mut *transaction)
109                .await
110                .map_err(restore_error)?;
111        }
112        transaction.commit().await.map_err(restore_error)?;
113
114        sqlx::query("DETACH DATABASE backup")
115            .execute(&mut connection)
116            .await
117            .map_err(restore_error)?;
118        connection.close().await.map_err(restore_error)
119    }
120
121    pub async fn create_backups(
122        &self, source_config: &PipedashConfig, target_config: &PipedashConfig,
123        source_backend: Option<&dyn ConfigBackend>,
124    ) -> DomainResult<BackupPaths> {
125        let timestamp = Utc::now().format("%Y%m%d_%H%M%S").to_string();
126
127        tracing::info!("Creating migration backups with timestamp: {}", timestamp);
128
129        let source_db_backup = self
130            .backup_source_database(source_config, source_backend, &timestamp)
131            .await?;
132
133        let config_backup = self.backup_config_file(&timestamp).await?;
134
135        let target_db_backup = self
136            .backup_target_database(target_config, &timestamp)
137            .await?;
138
139        let backups = BackupPaths {
140            source_db: source_db_backup,
141            source_config: config_backup,
142            target_db: target_db_backup,
143            timestamp,
144        };
145
146        tracing::info!("Backups created successfully: {:?}", backups);
147
148        Ok(backups)
149    }
150
151    async fn backup_source_database(
152        &self, source_config: &PipedashConfig, source_backend: Option<&dyn ConfigBackend>,
153        timestamp: &str,
154    ) -> DomainResult<Option<PathBuf>> {
155        match source_config.storage.backend {
156            StorageBackend::Sqlite => {
157                let src = source_config.db_path();
158                if !src.exists() {
159                    tracing::warn!("Source database does not exist: {:?}", src);
160                    return Ok(None);
161                }
162
163                let dst = src.with_file_name(format!("pipedash.db.backup-{}", timestamp));
164                Self::backup_sqlite_database(&src, &dst).await?;
165
166                tracing::info!("SQLite database backed up to: {:?}", dst);
167                Ok(Some(dst))
168            }
169            StorageBackend::Postgres => {
170                if let Some(backend) = source_backend {
171                    let export = backend.export_all().await?;
172                    let backup_path = self
173                        .data_dir
174                        .join(format!("postgres_backup_{}.json", timestamp));
175
176                    let json = serde_json::to_string_pretty(&export).map_err(|e| {
177                        DomainError::DatabaseError(format!(
178                            "Failed to serialize PostgreSQL backup: {}",
179                            e
180                        ))
181                    })?;
182
183                    tokio::fs::write(&backup_path, json).await.map_err(|e| {
184                        DomainError::DatabaseError(format!(
185                            "Failed to write PostgreSQL backup: {}",
186                            e
187                        ))
188                    })?;
189
190                    tracing::info!("PostgreSQL database exported to: {:?}", backup_path);
191                    Ok(Some(backup_path))
192                } else {
193                    tracing::warn!("No source backend provided for PostgreSQL backup");
194                    Ok(None)
195                }
196            }
197        }
198    }
199
200    async fn backup_config_file(&self, timestamp: &str) -> DomainResult<PathBuf> {
201        use crate::infrastructure::config::ConfigLoader;
202
203        let config_path = ConfigLoader::discover_config_path();
204        if !config_path.exists() {
205            return Err(DomainError::InvalidConfig(format!(
206                "Config file not found: {:?}",
207                config_path
208            )));
209        }
210
211        let backup_path = config_path.with_file_name(format!("config.toml.backup-{}", timestamp));
212        tokio::fs::copy(&config_path, &backup_path)
213            .await
214            .map_err(|e| {
215                DomainError::DatabaseError(format!("Failed to backup config file: {}", e))
216            })?;
217
218        tracing::info!("Config file backed up to: {:?}", backup_path);
219        Ok(backup_path)
220    }
221
222    async fn backup_target_database(
223        &self, target_config: &PipedashConfig, timestamp: &str,
224    ) -> DomainResult<Option<PathBuf>> {
225        match target_config.storage.backend {
226            StorageBackend::Sqlite => {
227                let target_path = target_config.db_path();
228                if !target_path.exists() {
229                    return Ok(None);
230                }
231
232                let metadata = tokio::fs::metadata(&target_path).await.ok();
233                if let Some(meta) = metadata {
234                    if meta.len() > 0 {
235                        let backup_path = target_path
236                            .with_file_name(format!("pipedash_target.db.backup-{}", timestamp));
237                        Self::backup_sqlite_database(&target_path, &backup_path).await?;
238
239                        tracing::info!("Target database backed up to: {:?}", backup_path);
240                        return Ok(Some(backup_path));
241                    }
242                }
243                Ok(None)
244            }
245            StorageBackend::Postgres => Ok(None),
246        }
247    }
248
249    pub async fn restore_backups(&self, backups: &BackupPaths) -> DomainResult<()> {
250        tracing::warn!("Migration failed - restoring from backups");
251
252        use crate::infrastructure::config::ConfigLoader;
253        let config_path = ConfigLoader::discover_config_path();
254        if backups.source_config.exists() {
255            tokio::fs::copy(&backups.source_config, &config_path)
256                .await
257                .map_err(|e| {
258                    DomainError::DatabaseError(format!("Failed to restore config file: {}", e))
259                })?;
260            tracing::info!("Config file restored from: {:?}", backups.source_config);
261        }
262
263        if let Some(source_db) = &backups.source_db {
264            if source_db.exists() {
265                if source_db.extension().is_some_and(|ext| ext == "json") {
266                    tracing::warn!(
267                        "PostgreSQL source backup exists but cannot be automatically restored"
268                    );
269                } else {
270                    let original_path = source_db.parent().unwrap().join("pipedash.db");
271                    Self::restore_sqlite_database(source_db, &original_path).await?;
272                    tracing::info!("Source database restored from: {:?}", source_db);
273                }
274            }
275        }
276
277        if let Some(target_db) = &backups.target_db {
278            if target_db.exists() {
279                let original_path = target_db.parent().unwrap().join("pipedash.db");
280                Self::restore_sqlite_database(target_db, &original_path).await?;
281                tracing::info!("Target database restored from: {:?}", target_db);
282            }
283        }
284
285        tracing::info!("Backup restoration complete");
286        Ok(())
287    }
288
289    pub async fn cleanup_backups(&self, backups: &BackupPaths, keep: bool) -> DomainResult<()> {
290        if keep {
291            tracing::info!("Keeping backups as requested");
292            return Ok(());
293        }
294
295        tracing::info!("Cleaning up migration backups");
296
297        if let Some(source_db) = &backups.source_db {
298            if source_db.exists() {
299                tokio::fs::remove_file(source_db).await.ok();
300                tracing::debug!("Deleted source database backup: {:?}", source_db);
301            }
302        }
303
304        if backups.source_config.exists() {
305            tokio::fs::remove_file(&backups.source_config).await.ok();
306            tracing::debug!("Deleted config backup: {:?}", backups.source_config);
307        }
308
309        if let Some(target_db) = &backups.target_db {
310            if target_db.exists() {
311                tokio::fs::remove_file(target_db).await.ok();
312                tracing::debug!("Deleted target database backup: {:?}", target_db);
313            }
314        }
315
316        self.cleanup_old_backups().await?;
317
318        Ok(())
319    }
320
321    async fn cleanup_old_backups(&self) -> DomainResult<()> {
322        use std::time::{
323            Duration,
324            SystemTime,
325        };
326
327        let retention_days = 7;
328        let cutoff = SystemTime::now() - Duration::from_secs(retention_days * 24 * 3600);
329
330        let patterns = vec!["*.backup-*", "*_backup_*.json"];
331
332        for pattern in patterns {
333            let glob_pattern = self.data_dir.join(pattern);
334            if let Ok(entries) = glob::glob(glob_pattern.to_str().unwrap()) {
335                for entry in entries.flatten() {
336                    if let Ok(metadata) = tokio::fs::metadata(&entry).await {
337                        if let Ok(modified) = metadata.modified() {
338                            if modified < cutoff {
339                                tokio::fs::remove_file(&entry).await.ok();
340                                tracing::debug!("Deleted old backup: {:?}", entry);
341                            }
342                        }
343                    }
344                }
345            }
346        }
347
348        Ok(())
349    }
350
351    pub async fn estimate_backup_size(&self, config: &PipedashConfig) -> u64 {
352        let mut total_size = 0u64;
353
354        let db_path = config.db_path();
355        if let Ok(metadata) = tokio::fs::metadata(&db_path).await {
356            total_size += metadata.len();
357        }
358
359        use crate::infrastructure::config::ConfigLoader;
360        let config_path = ConfigLoader::discover_config_path();
361        if let Ok(metadata) = tokio::fs::metadata(&config_path).await {
362            total_size += metadata.len();
363        }
364
365        total_size
366    }
367}
368
369#[cfg(test)]
370mod tests {
371    use super::*;
372
373    async fn check_wal_backup(target: bool) {
374        let dir = tempfile::tempdir().unwrap();
375        let mut config = PipedashConfig::default();
376        config.storage.data_dir = dir.path().to_str().unwrap().to_owned();
377        let pool = crate::infrastructure::database::init_database(config.db_path())
378            .await
379            .unwrap();
380        let mut writer = pool.acquire().await.unwrap();
381        sqlx::query("PRAGMA wal_autocheckpoint = 0")
382            .execute(&mut *writer)
383            .await
384            .unwrap();
385        sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)")
386            .execute(&mut *writer)
387            .await
388            .unwrap();
389        sqlx::query(
390            "INSERT INTO providers (id, name, provider_type, token_encrypted) VALUES (42, 'committed', 'jenkins', '')",
391        )
392        .execute(&mut *writer)
393        .await
394        .unwrap();
395        let mut transaction = writer.begin().await.unwrap();
396        sqlx::query(
397            "INSERT INTO providers (name, provider_type, token_encrypted) VALUES ('uncommitted', 'jenkins', '')",
398        )
399        .execute(&mut *transaction)
400        .await
401        .unwrap();
402
403        let manager = BackupManager::new(dir.path().to_path_buf());
404        let backup = if target {
405            manager.backup_target_database(&config, "wal").await
406        } else {
407            manager.backup_source_database(&config, None, "wal").await
408        }
409        .unwrap()
410        .unwrap();
411        let standalone = tempfile::tempdir().unwrap();
412        let snapshot_path = standalone.path().join("snapshot.db");
413        tokio::fs::copy(backup, &snapshot_path).await.unwrap();
414        let mut snapshot = SqliteConnection::connect_with(
415            &SqliteConnectOptions::new()
416                .filename(snapshot_path)
417                .read_only(true),
418        )
419        .await
420        .unwrap();
421        let providers: Vec<(i64, String)> =
422            sqlx::query_as("SELECT id, name FROM providers ORDER BY id")
423                .fetch_all(&mut snapshot)
424                .await
425                .unwrap();
426        assert_eq!(providers, vec![(42, "committed".to_owned())]);
427        let integrity: String = sqlx::query_scalar("PRAGMA integrity_check")
428            .fetch_one(&mut snapshot)
429            .await
430            .unwrap();
431        assert_eq!(integrity, "ok");
432        snapshot.close().await.unwrap();
433        transaction.rollback().await.unwrap();
434        drop(writer);
435        pool.close().await;
436    }
437
438    #[tokio::test]
439    async fn source_backup_includes_committed_wal_providers() {
440        check_wal_backup(false).await;
441    }
442
443    #[tokio::test]
444    async fn target_backup_includes_committed_wal_providers() {
445        check_wal_backup(true).await;
446    }
447
448    #[tokio::test]
449    async fn restore_keeps_open_wal_source_consistent() {
450        let dir = tempfile::tempdir().unwrap();
451        let mut config = PipedashConfig::default();
452        config.storage.data_dir = dir.path().to_str().unwrap().to_owned();
453        let pool = crate::infrastructure::database::init_database(config.db_path())
454            .await
455            .unwrap();
456        let mut writer = pool.acquire().await.unwrap();
457        for statement in [
458            "PRAGMA wal_autocheckpoint = 0",
459            "WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < 200) \
460             INSERT INTO providers (name, provider_type, token_encrypted) \
461             SELECT 'bulk' || i, 'jenkins', hex(randomblob(1000)) FROM n",
462            "PRAGMA wal_checkpoint(TRUNCATE)",
463            "DELETE FROM providers",
464            "INSERT INTO providers (id, name, provider_type, token_encrypted) VALUES (42, 'kept', 'jenkins', '')",
465        ] {
466            sqlx::query(statement).execute(&mut *writer).await.unwrap();
467        }
468
469        let manager = BackupManager::new(dir.path().to_path_buf());
470        let source_db = manager
471            .backup_source_database(&config, None, "restore")
472            .await
473            .unwrap();
474        sqlx::query("INSERT INTO providers (name, provider_type, token_encrypted) VALUES ('failed', 'jenkins', '')")
475            .execute(&mut *writer)
476            .await
477            .unwrap();
478        let backups = BackupPaths {
479            source_db,
480            source_config: dir.path().join("missing-config.toml.backup"),
481            target_db: None,
482            timestamp: "restore".into(),
483        };
484        manager.restore_backups(&backups).await.unwrap();
485
486        let integrity: String = sqlx::query_scalar("PRAGMA integrity_check")
487            .fetch_one(&mut *writer)
488            .await
489            .unwrap();
490        assert_eq!(integrity, "ok");
491        let providers: Vec<(i64, String)> =
492            sqlx::query_as("SELECT id, name FROM providers ORDER BY id")
493                .fetch_all(&mut *writer)
494                .await
495                .unwrap();
496        assert_eq!(providers, vec![(42, "kept".to_owned())]);
497        drop(writer);
498        pool.close().await;
499    }
500
501    async fn check_target_restore(postgres_source: bool) {
502        let source_dir = tempfile::tempdir().unwrap();
503        let target_dir = tempfile::tempdir().unwrap();
504        let mut target_config = PipedashConfig::default();
505        target_config.storage.data_dir = target_dir.path().to_str().unwrap().to_owned();
506        let pool = crate::infrastructure::database::init_database(target_config.db_path())
507            .await
508            .unwrap();
509        sqlx::query("INSERT INTO providers (id, name, provider_type, token_encrypted) VALUES (7, 'original', 'jenkins', '')")
510            .execute(&pool)
511            .await
512            .unwrap();
513
514        let manager = BackupManager::new(source_dir.path().to_path_buf());
515        let target_db = manager
516            .backup_target_database(&target_config, "target")
517            .await
518            .unwrap();
519        sqlx::query("INSERT INTO providers (name, provider_type, token_encrypted) VALUES ('first', 'jenkins', '')")
520            .execute(&pool)
521            .await
522            .unwrap();
523        let source_db = if postgres_source {
524            let json_backup = source_dir.path().join("postgres_backup_target.json");
525            tokio::fs::write(&json_backup, "{\"providers\": []}")
526                .await
527                .unwrap();
528            Some(json_backup)
529        } else {
530            None
531        };
532        let backups = BackupPaths {
533            source_db,
534            source_config: source_dir.path().join("missing-config.toml.backup"),
535            target_db,
536            timestamp: "target".into(),
537        };
538        manager.restore_backups(&backups).await.unwrap();
539
540        let providers: Vec<(i64, String)> =
541            sqlx::query_as("SELECT id, name FROM providers ORDER BY id")
542                .fetch_all(&pool)
543                .await
544                .unwrap();
545        assert_eq!(providers, vec![(7, "original".to_owned())]);
546        pool.close().await;
547    }
548
549    #[tokio::test]
550    async fn restore_drops_committed_target_writes() {
551        check_target_restore(false).await;
552    }
553
554    #[tokio::test]
555    async fn restore_drops_committed_target_writes_after_postgres_source() {
556        check_target_restore(true).await;
557    }
558
559    #[tokio::test]
560    async fn restore_keeps_sqlite_file_after_postgres_backup() {
561        let dir = tempfile::tempdir().unwrap();
562        let config_path = dir.path().join("config.toml");
563        std::env::set_var("PIPEDASH_CONFIG_PATH", &config_path);
564        tokio::fs::write(&config_path, "[storage]\nbackend = \"postgres\"\n")
565            .await
566            .unwrap();
567        let config_backup = dir.path().join("config.toml.backup-x");
568        tokio::fs::copy(&config_path, &config_backup).await.unwrap();
569        let sqlite_path = dir.path().join("pipedash.db");
570        tokio::fs::write(&sqlite_path, b"SQLite format 3\0")
571            .await
572            .unwrap();
573        let json_backup = dir.path().join("postgres_backup_x.json");
574        tokio::fs::write(&json_backup, "{\"providers\": []}")
575            .await
576            .unwrap();
577        let backups = BackupPaths {
578            source_db: Some(json_backup),
579            source_config: config_backup,
580            target_db: None,
581            timestamp: "x".into(),
582        };
583
584        BackupManager::new(dir.path().to_path_buf())
585            .restore_backups(&backups)
586            .await
587            .unwrap();
588
589        let after = tokio::fs::read(&sqlite_path).await.unwrap();
590        assert_eq!(
591            after,
592            b"SQLite format 3\0",
593            "{}",
594            String::from_utf8_lossy(&after)
595        );
596    }
597}
pipedash_core/infrastructure/migration/backup.rs

pipedash_core/infrastructure/migration/
backup.rs
