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 source 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                let original_path = 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                    return Ok(());
270                } else {
271                    source_db.parent().unwrap().join("pipedash.db")
272                };
273
274                Self::restore_sqlite_database(source_db, &original_path).await?;
275                tracing::info!("Source database restored from: {:?}", source_db);
276            }
277        }
278
279        tracing::info!("Backup restoration complete");
280        Ok(())
281    }
282
283    pub async fn cleanup_backups(&self, backups: &BackupPaths, keep: bool) -> DomainResult<()> {
284        if keep {
285            tracing::info!("Keeping backups as requested");
286            return Ok(());
287        }
288
289        tracing::info!("Cleaning up migration backups");
290
291        if let Some(source_db) = &backups.source_db {
292            if source_db.exists() {
293                tokio::fs::remove_file(source_db).await.ok();
294                tracing::debug!("Deleted source database backup: {:?}", source_db);
295            }
296        }
297
298        if backups.source_config.exists() {
299            tokio::fs::remove_file(&backups.source_config).await.ok();
300            tracing::debug!("Deleted config backup: {:?}", backups.source_config);
301        }
302
303        if let Some(target_db) = &backups.target_db {
304            if target_db.exists() {
305                tokio::fs::remove_file(target_db).await.ok();
306                tracing::debug!("Deleted target database backup: {:?}", target_db);
307            }
308        }
309
310        self.cleanup_old_backups().await?;
311
312        Ok(())
313    }
314
315    async fn cleanup_old_backups(&self) -> DomainResult<()> {
316        use std::time::{
317            Duration,
318            SystemTime,
319        };
320
321        let retention_days = 7;
322        let cutoff = SystemTime::now() - Duration::from_secs(retention_days * 24 * 3600);
323
324        let patterns = vec!["*.backup-*", "*_backup_*.json"];
325
326        for pattern in patterns {
327            let glob_pattern = self.data_dir.join(pattern);
328            if let Ok(entries) = glob::glob(glob_pattern.to_str().unwrap()) {
329                for entry in entries.flatten() {
330                    if let Ok(metadata) = tokio::fs::metadata(&entry).await {
331                        if let Ok(modified) = metadata.modified() {
332                            if modified < cutoff {
333                                tokio::fs::remove_file(&entry).await.ok();
334                                tracing::debug!("Deleted old backup: {:?}", entry);
335                            }
336                        }
337                    }
338                }
339            }
340        }
341
342        Ok(())
343    }
344
345    pub async fn estimate_backup_size(&self, config: &PipedashConfig) -> u64 {
346        let mut total_size = 0u64;
347
348        let db_path = config.db_path();
349        if let Ok(metadata) = tokio::fs::metadata(&db_path).await {
350            total_size += metadata.len();
351        }
352
353        use crate::infrastructure::config::ConfigLoader;
354        let config_path = ConfigLoader::discover_config_path();
355        if let Ok(metadata) = tokio::fs::metadata(&config_path).await {
356            total_size += metadata.len();
357        }
358
359        total_size
360    }
361}
362
363#[cfg(test)]
364mod tests {
365    use super::*;
366
367    async fn check_wal_backup(target: bool) {
368        let dir = tempfile::tempdir().unwrap();
369        let mut config = PipedashConfig::default();
370        config.storage.data_dir = dir.path().to_str().unwrap().to_owned();
371        let pool = crate::infrastructure::database::init_database(config.db_path())
372            .await
373            .unwrap();
374        let mut writer = pool.acquire().await.unwrap();
375        sqlx::query("PRAGMA wal_autocheckpoint = 0")
376            .execute(&mut *writer)
377            .await
378            .unwrap();
379        sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)")
380            .execute(&mut *writer)
381            .await
382            .unwrap();
383        sqlx::query(
384            "INSERT INTO providers (id, name, provider_type, token_encrypted) VALUES (42, 'committed', 'jenkins', '')",
385        )
386        .execute(&mut *writer)
387        .await
388        .unwrap();
389        let mut transaction = writer.begin().await.unwrap();
390        sqlx::query(
391            "INSERT INTO providers (name, provider_type, token_encrypted) VALUES ('uncommitted', 'jenkins', '')",
392        )
393        .execute(&mut *transaction)
394        .await
395        .unwrap();
396
397        let manager = BackupManager::new(dir.path().to_path_buf());
398        let backup = if target {
399            manager.backup_target_database(&config, "wal").await
400        } else {
401            manager.backup_source_database(&config, None, "wal").await
402        }
403        .unwrap()
404        .unwrap();
405        let standalone = tempfile::tempdir().unwrap();
406        let snapshot_path = standalone.path().join("snapshot.db");
407        tokio::fs::copy(backup, &snapshot_path).await.unwrap();
408        let mut snapshot = SqliteConnection::connect_with(
409            &SqliteConnectOptions::new()
410                .filename(snapshot_path)
411                .read_only(true),
412        )
413        .await
414        .unwrap();
415        let providers: Vec<(i64, String)> =
416            sqlx::query_as("SELECT id, name FROM providers ORDER BY id")
417                .fetch_all(&mut snapshot)
418                .await
419                .unwrap();
420        assert_eq!(providers, vec![(42, "committed".to_owned())]);
421        let integrity: String = sqlx::query_scalar("PRAGMA integrity_check")
422            .fetch_one(&mut snapshot)
423            .await
424            .unwrap();
425        assert_eq!(integrity, "ok");
426        snapshot.close().await.unwrap();
427        transaction.rollback().await.unwrap();
428        drop(writer);
429        pool.close().await;
430    }
431
432    #[tokio::test]
433    async fn source_backup_includes_committed_wal_providers() {
434        check_wal_backup(false).await;
435    }
436
437    #[tokio::test]
438    async fn target_backup_includes_committed_wal_providers() {
439        check_wal_backup(true).await;
440    }
441
442    #[tokio::test]
443    async fn restore_keeps_open_wal_source_consistent() {
444        let dir = tempfile::tempdir().unwrap();
445        let mut config = PipedashConfig::default();
446        config.storage.data_dir = dir.path().to_str().unwrap().to_owned();
447        let pool = crate::infrastructure::database::init_database(config.db_path())
448            .await
449            .unwrap();
450        let mut writer = pool.acquire().await.unwrap();
451        for statement in [
452            "PRAGMA wal_autocheckpoint = 0",
453            "WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < 200) \
454             INSERT INTO providers (name, provider_type, token_encrypted) \
455             SELECT 'bulk' || i, 'jenkins', hex(randomblob(1000)) FROM n",
456            "PRAGMA wal_checkpoint(TRUNCATE)",
457            "DELETE FROM providers",
458            "INSERT INTO providers (id, name, provider_type, token_encrypted) VALUES (42, 'kept', 'jenkins', '')",
459        ] {
460            sqlx::query(statement).execute(&mut *writer).await.unwrap();
461        }
462
463        let manager = BackupManager::new(dir.path().to_path_buf());
464        let source_db = manager
465            .backup_source_database(&config, None, "restore")
466            .await
467            .unwrap();
468        sqlx::query("INSERT INTO providers (name, provider_type, token_encrypted) VALUES ('failed', 'jenkins', '')")
469            .execute(&mut *writer)
470            .await
471            .unwrap();
472        let backups = BackupPaths {
473            source_db,
474            source_config: dir.path().join("missing-config.toml.backup"),
475            target_db: None,
476            timestamp: "restore".into(),
477        };
478        manager.restore_backups(&backups).await.unwrap();
479
480        let integrity: String = sqlx::query_scalar("PRAGMA integrity_check")
481            .fetch_one(&mut *writer)
482            .await
483            .unwrap();
484        assert_eq!(integrity, "ok");
485        let providers: Vec<(i64, String)> =
486            sqlx::query_as("SELECT id, name FROM providers ORDER BY id")
487                .fetch_all(&mut *writer)
488                .await
489                .unwrap();
490        assert_eq!(providers, vec![(42, "kept".to_owned())]);
491        drop(writer);
492        pool.close().await;
493    }
494
495    #[tokio::test]
496    async fn restore_keeps_sqlite_file_after_postgres_backup() {
497        let dir = tempfile::tempdir().unwrap();
498        let config_path = dir.path().join("config.toml");
499        std::env::set_var("PIPEDASH_CONFIG_PATH", &config_path);
500        tokio::fs::write(&config_path, "[storage]\nbackend = \"postgres\"\n")
501            .await
502            .unwrap();
503        let config_backup = dir.path().join("config.toml.backup-x");
504        tokio::fs::copy(&config_path, &config_backup).await.unwrap();
505        let sqlite_path = dir.path().join("pipedash.db");
506        tokio::fs::write(&sqlite_path, b"SQLite format 3\0")
507            .await
508            .unwrap();
509        let json_backup = dir.path().join("postgres_backup_x.json");
510        tokio::fs::write(&json_backup, "{\"providers\": []}")
511            .await
512            .unwrap();
513        let backups = BackupPaths {
514            source_db: Some(json_backup),
515            source_config: config_backup,
516            target_db: None,
517            timestamp: "x".into(),
518        };
519
520        BackupManager::new(dir.path().to_path_buf())
521            .restore_backups(&backups)
522            .await
523            .unwrap();
524
525        let after = tokio::fs::read(&sqlite_path).await.unwrap();
526        assert_eq!(
527            after,
528            b"SQLite format 3\0",
529            "{}",
530            String::from_utf8_lossy(&after)
531        );
532    }
533}
pipedash_core/infrastructure/migration/backup.rs

pipedash_core/infrastructure/migration/
backup.rs
