Skip to main content

pipedash_core/infrastructure/
migration.rs

1pub mod backup;
2pub mod validator;
3
4use std::collections::{
5    BTreeMap,
6    HashMap,
7};
8use std::sync::Arc;
9
10pub use backup::{
11    BackupManager,
12    BackupPaths,
13};
14use chrono::{
15    DateTime,
16    Utc,
17};
18use serde::{
19    Deserialize,
20    Serialize,
21};
22pub use validator::{
23    PreMigrationValidationReport,
24    PreMigrationValidator,
25};
26
27use crate::domain::{
28    DeviceAlert,
29    DomainError,
30    DomainResult,
31    MetricsQuery,
32    NotificationPreferences,
33    OnceWatch,
34    ProviderConfig,
35};
36use crate::event::EventBus;
37use crate::infrastructure::config::{
38    PipedashConfig,
39    StorageConfig,
40    StorageManager,
41};
42use crate::infrastructure::token_store::TokenStore;
43use crate::infrastructure::{
44    database::{
45        DatabasePool,
46        DeviceAlertRepository,
47        DevicePushRepository,
48        DeviceRepository,
49        MetricsRepository,
50        PipelineViewRepository,
51        PushRepository,
52        RecordMetrics,
53        Repository,
54    },
55    ConfigBackend,
56    StorageBackend,
57};
58
59#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
60pub enum MigrationStep {
61    ValidateTarget,
62    MigrateTokens,
63    MigrateConfigs,
64    MigrateCache,
65    VerifyMigration,
66    UpdateConfig,
67}
68
69impl MigrationStep {
70    pub fn description(&self) -> &str {
71        match self {
72            MigrationStep::ValidateTarget => "Validating target configuration",
73            MigrationStep::MigrateTokens => "Migrating tokens",
74            MigrationStep::MigrateConfigs => "Migrating provider configurations",
75            MigrationStep::MigrateCache => "Migrating cached data",
76            MigrationStep::VerifyMigration => "Verifying migration",
77            MigrationStep::UpdateConfig => "Updating storage configuration",
78        }
79    }
80}
81
82#[derive(Debug, Clone, Serialize, Deserialize)]
83pub struct MigrationPlan {
84    pub from: PipedashConfig,
85    pub to: PipedashConfig,
86    pub steps: Vec<MigrationStep>,
87    pub migrate_tokens: bool,
88    pub migrate_configs: bool,
89    pub migrate_cache: bool,
90    pub backend_changed: bool,
91    pub data_dir_changed: bool,
92    pub created_at: DateTime<Utc>,
93}
94
95impl MigrationPlan {
96    pub fn summary(&self) -> String {
97        format!(
98            "Migrate from {} to {} ({} steps)",
99            self.from.storage.summary(),
100            self.to.storage.summary(),
101            self.steps.len()
102        )
103    }
104}
105
106#[derive(Debug, Clone, Serialize, Deserialize)]
107pub struct MigrationProgress {
108    pub current_step: MigrationStep,
109    pub step_index: usize,
110    pub total_steps: usize,
111    pub message: String,
112}
113
114#[derive(Debug, Clone, Serialize, Deserialize)]
115pub struct MigrationResult {
116    pub success: bool,
117    pub steps_completed: Vec<MigrationStep>,
118    pub errors: Vec<String>,
119    pub duration_ms: u64,
120    pub stats: MigrationStats,
121    #[serde(default)]
122    pub provider_id_mapping: std::collections::HashMap<i64, i64>,
123    #[serde(skip_serializing_if = "Option::is_none")]
124    pub backups: Option<BackupPaths>,
125}
126
127impl MigrationResult {
128    pub fn summary(&self) -> String {
129        if self.success {
130            format!(
131                "Migration successful: {} providers, {} cache entries migrated in {}ms",
132                self.stats.providers_migrated, self.stats.cache_entries_migrated, self.duration_ms
133            )
134        } else {
135            format!(
136                "Migration failed after {} steps: {}",
137                self.steps_completed.len(),
138                self.errors.join(", ")
139            )
140        }
141    }
142}
143
144#[derive(Debug, Clone, Default, Serialize, Deserialize)]
145pub struct MigrationStats {
146    pub providers_migrated: usize,
147    pub tokens_migrated: usize,
148    pub cache_entries_migrated: usize,
149    pub metrics_migrated: usize,
150    pub permissions_migrated: usize,
151    pub providers_cleaned: usize,
152    pub tokens_cleaned: usize,
153}
154
155#[derive(Debug, Clone, Serialize, Deserialize)]
156pub struct MigrationOptions {
157    pub migrate_tokens: bool,
158    pub migrate_cache: bool,
159    pub token_password: Option<String>,
160    pub dry_run: bool,
161    #[serde(default)]
162    pub allow_non_empty_target: bool,
163    #[serde(default)]
164    pub clean_target: bool,
165    #[serde(default)]
166    pub keep_backups: bool,
167    #[serde(default = "default_max_retries")]
168    pub max_retries: u32,
169}
170
171fn default_max_retries() -> u32 {
172    3
173}
174
175impl Default for MigrationOptions {
176    fn default() -> Self {
177        Self {
178            migrate_tokens: false,
179            migrate_cache: false,
180            token_password: None,
181            dry_run: false,
182            allow_non_empty_target: false,
183            clean_target: false,
184            keep_backups: false,
185            max_retries: 3,
186        }
187    }
188}
189
190#[derive(Debug, Clone, Serialize, Deserialize)]
191pub struct ValidationReport {
192    pub errors: Vec<String>,
193    pub warnings: Vec<String>,
194    pub recommendations: Vec<String>,
195}
196
197impl ValidationReport {
198    pub fn is_valid(&self) -> bool {
199        self.errors.is_empty()
200    }
201}
202
203pub const VAULT_PASSWORD_ENV_VAR: &str = "PIPEDASH_VAULT_PASSWORD";
204
205fn get_vault_password(options: &MigrationOptions) -> DomainResult<String> {
206    if let Ok(password) = std::env::var(VAULT_PASSWORD_ENV_VAR) {
207        if !password.is_empty() {
208            return Ok(password);
209        }
210    }
211
212    options.token_password.clone().ok_or_else(|| {
213        DomainError::InvalidConfig(format!(
214            "Vault password required for token migration. Set {} environment variable or provide token_password in options.",
215            VAULT_PASSWORD_ENV_VAR
216        ))
217    })
218}
219
220fn remap_pipeline_id(pipeline_id: &str, provider_ids: &HashMap<i64, i64>) -> Option<String> {
221    let mut parts = pipeline_id.splitn(3, "__");
222    let (Some(provider_type), Some(provider_id), Some(rest)) =
223        (parts.next(), parts.next(), parts.next())
224    else {
225        return None;
226    };
227    let new_id = provider_ids.get(&provider_id.parse().ok()?)?;
228
229    Some(format!("{provider_type}__{new_id}__{rest}"))
230}
231
232fn remap_preferences(
233    mut preferences: NotificationPreferences, providers: &HashMap<i64, i64>,
234) -> NotificationPreferences {
235    preferences.muted_providers = preferences
236        .muted_providers
237        .iter()
238        .filter_map(|id| providers.get(id).copied())
239        .collect();
240    let pipelines = |ids: &[String]| -> Vec<String> {
241        ids.iter()
242            .filter_map(|id| remap_pipeline_id(id, providers))
243            .collect()
244    };
245    preferences.muted_pipelines = pipelines(&preferences.muted_pipelines);
246    preferences.watched_pipelines = pipelines(&preferences.watched_pipelines);
247    preferences.watch_once = preferences
248        .watch_once
249        .into_iter()
250        .filter_map(|once| {
251            Some(OnceWatch {
252                pipeline_id: remap_pipeline_id(&once.pipeline_id, providers)?,
253                ..once
254            })
255        })
256        .collect();
257
258    preferences
259}
260
261fn remap_alert(mut alert: DeviceAlert, providers: &HashMap<i64, i64>) -> Option<DeviceAlert> {
262    if let Some(provider_id) = alert.provider_id {
263        alert.provider_id = Some(*providers.get(&provider_id)?);
264    }
265    alert.provider_ids = alert
266        .provider_ids
267        .iter()
268        .filter_map(|id| providers.get(id).copied())
269        .collect();
270    if let Some(pipeline_id) = alert.pipeline_id.take() {
271        let next = remap_pipeline_id(&pipeline_id, providers)?;
272        if alert.key == format!("pipeline:{pipeline_id}") {
273            alert.key = format!("pipeline:{next}");
274        }
275        alert.pipeline_id = Some(next);
276    }
277    for prefix in ["connection:", "instance:"] {
278        if let Some(id) = alert.key.strip_prefix(prefix) {
279            let next = providers.get(&id.parse().ok()?)?;
280            alert.key = format!("{prefix}{next}");
281            break;
282        }
283    }
284
285    Some(alert)
286}
287
288fn same_database(from: &PipedashConfig, to: &PipedashConfig) -> bool {
289    from.storage.backend == to.storage.backend
290        && if to.storage.backend.is_sqlite() {
291            from.db_path() == to.db_path()
292        } else {
293            from.storage.postgres.connection_string == to.storage.postgres.connection_string
294        }
295}
296
297pub struct MigrationOrchestrator {
298    source_config: PipedashConfig,
299    source_token_store: Arc<dyn TokenStore>,
300    source_config_backend: Arc<dyn ConfigBackend>,
301    source_cache_backend: Arc<dyn StorageBackend>,
302    source_metrics: MetricsRepository,
303    source_pool: DatabasePool,
304    event_bus: Option<Arc<dyn EventBus>>,
305    target_token_store: Option<Arc<dyn TokenStore>>,
306}
307
308impl MigrationOrchestrator {
309    pub async fn from_manager(
310        manager: &StorageManager, event_bus: Option<Arc<dyn EventBus>>,
311        target_token_store: Option<Arc<dyn TokenStore>>,
312    ) -> Self {
313        Self {
314            source_config: manager.config().clone(),
315            source_token_store: manager.token_store().await,
316            source_config_backend: manager.config_backend(),
317            source_cache_backend: manager.cache_backend(),
318            source_metrics: MetricsRepository::new_from_pool(manager.database_pool()),
319            source_pool: manager.database_pool(),
320            event_bus,
321            target_token_store,
322        }
323    }
324
325    pub fn plan_migration(
326        &self, mut target_config: PipedashConfig, options: &MigrationOptions,
327    ) -> DomainResult<MigrationPlan> {
328        use crate::infrastructure::config::StorageBackend as ConfigStorageBackend;
329
330        let source_data_dir = self.source_config.data_dir();
331        let target_data_dir = target_config.data_dir();
332        let data_dir_changed = source_data_dir != target_data_dir;
333
334        let source_storage = &self.source_config.storage;
335        let target_storage = &mut target_config.storage;
336
337        if data_dir_changed {
338            tracing::info!(
339                "[Migration Planning] Data directory change detected: {:?} → {:?}",
340                source_data_dir,
341                target_data_dir
342            );
343        }
344
345        let backend_changed = source_storage.backend != target_storage.backend;
346
347        let token_backend_changing = {
348            let source_uses_keyring = source_storage.backend.is_sqlite()
349                && std::env::var(VAULT_PASSWORD_ENV_VAR).is_err();
350            let target_uses_encrypted = target_storage.backend.is_sqlite()
351                && (options.token_password.is_some()
352                    || std::env::var(VAULT_PASSWORD_ENV_VAR).is_ok());
353
354            source_uses_keyring && target_uses_encrypted
355        };
356
357        if token_backend_changing {
358            tracing::info!(
359                "[Migration Planning] Token backend change detected: keyring → encrypted storage"
360            );
361        }
362
363        let needs_data_migration = backend_changed || data_dir_changed || token_backend_changing;
364
365        if target_storage.backend == ConfigStorageBackend::Postgres {
366            if target_storage.data_dir.starts_with("postgresql://")
367                || target_storage.data_dir.starts_with("postgres://")
368            {
369                tracing::warn!(
370                    "Target data_dir contains PostgreSQL connection string: {}. \
371                     Resetting to empty (will use default) - connection string should be in postgres.connection_string field.",
372                    target_storage.data_dir
373                );
374                target_storage.data_dir = String::new(); // Use default data_dir
375            } else if target_storage.data_dir.is_empty() {
376                tracing::debug!("Target uses PostgreSQL - data_dir is empty (will use default for cache/local files)");
377            }
378        }
379
380        let mut steps = Vec::new();
381
382        steps.push(MigrationStep::ValidateTarget);
383
384        if needs_data_migration {
385            steps.push(MigrationStep::MigrateConfigs);
386        }
387
388        if needs_data_migration && options.migrate_tokens {
389            steps.push(MigrationStep::MigrateTokens);
390        }
391
392        if needs_data_migration && options.migrate_cache {
393            steps.push(MigrationStep::MigrateCache);
394        }
395
396        if !steps.is_empty() {
397            steps.push(MigrationStep::VerifyMigration);
398            steps.push(MigrationStep::UpdateConfig);
399        }
400
401        Ok(MigrationPlan {
402            from: self.source_config.clone(),
403            to: target_config,
404            steps,
405            migrate_tokens: needs_data_migration && options.migrate_tokens,
406            migrate_configs: needs_data_migration,
407            migrate_cache: needs_data_migration && options.migrate_cache,
408            backend_changed,
409            data_dir_changed,
410            created_at: Utc::now(),
411        })
412    }
413
414    pub async fn validate_target_config(
415        &self, config: &StorageConfig,
416    ) -> DomainResult<ValidationReport> {
417        use crate::infrastructure::config::StorageBackend as ConfigStorageBackend;
418
419        let mut report = ValidationReport {
420            errors: Vec::new(),
421            warnings: Vec::new(),
422            recommendations: Vec::new(),
423        };
424
425        if config.backend == ConfigStorageBackend::Postgres {
426            if config.postgres.connection_string.is_empty() {
427                report.errors.push(
428                    "PostgreSQL connection string required when using postgres backend".into(),
429                );
430            }
431
432            let has_settings_password = config.vault_password.is_some();
433            let has_env_password = std::env::var(VAULT_PASSWORD_ENV_VAR).is_ok();
434
435            if !has_settings_password && !has_env_password {
436                report.warnings.push(format!(
437                    "Vault password not provided. Set {} environment variable or provide password during setup.",
438                    VAULT_PASSWORD_ENV_VAR
439                ));
440            }
441        }
442
443        Ok(report)
444    }
445
446    pub async fn execute_migration(
447        &self, plan: MigrationPlan, options: MigrationOptions, is_desktop: bool,
448    ) -> MigrationResult {
449        let start_time = std::time::Instant::now();
450        let mut result = MigrationResult {
451            success: false,
452            steps_completed: Vec::new(),
453            errors: Vec::new(),
454            duration_ms: 0,
455            stats: MigrationStats::default(),
456            provider_id_mapping: std::collections::HashMap::new(),
457            backups: None,
458        };
459
460        if options.dry_run {
461            result.success = true;
462            result.duration_ms = start_time.elapsed().as_millis() as u64;
463            return result;
464        }
465
466        let backup_manager = BackupManager::new(plan.from.data_dir());
467        let backups = match backup_manager
468            .create_backups(&plan.from, &plan.to, Some(&*self.source_config_backend))
469            .await
470        {
471            Ok(backups) => backups,
472            Err(e) => {
473                result.errors.push(format!(
474                    "Failed to create backups: {}\n\n\
475                    WARNING: Migration aborted - backup creation is required for safe migration.\n\
476                    INFO: Your data is safe - no modifications were performed.\n\n\
477                    Please ensure:\n\
478                    1. Sufficient disk space (need ~2x source database size)\n\
479                    2. Write permissions to data directory\n\
480                    3. Source database is not locked by another process",
481                    e
482                ));
483                result.duration_ms = start_time.elapsed().as_millis() as u64;
484                return result;
485            }
486        };
487        result.backups = Some(backups.clone());
488
489        let target_manager = if let Some(token_store) = &self.target_token_store {
490            match StorageManager::with_token_store(plan.to.clone(), token_store.clone(), is_desktop)
491                .await
492            {
493                Ok(m) => m,
494                Err(e) => {
495                    result.errors.push(format!(
496                        "Failed to create target manager with token store: {}\n\n\
497                        INFO: Migration aborted before any changes were made.\n\
498                        INFO: Your data is safe - no modifications were performed.",
499                        e
500                    ));
501                    backup_manager.cleanup_backups(&backups, false).await.ok();
502                    result.duration_ms = start_time.elapsed().as_millis() as u64;
503                    return result;
504                }
505            }
506        } else {
507            match StorageManager::from_config(plan.to.clone(), false).await {
508                Ok(m) => m,
509                Err(e) => {
510                    result.errors.push(format!(
511                        "Failed to create target manager: {}\n\n\
512                        INFO: Migration aborted before any changes were made.\n\
513                        INFO: Your data is safe - no modifications were performed.",
514                        e
515                    ));
516                    backup_manager.cleanup_backups(&backups, false).await.ok();
517                    result.duration_ms = start_time.elapsed().as_millis() as u64;
518                    return result;
519                }
520            }
521        };
522
523        for (idx, step) in plan.steps.iter().enumerate() {
524            self.emit_progress(*step, idx, plan.steps.len()).await;
525
526            let step_result = self
527                .execute_step(*step, &plan, &options, &target_manager, &mut result)
528                .await;
529
530            match step_result {
531                Ok(()) => {
532                    result.steps_completed.push(*step);
533                }
534                Err(e) => {
535                    let step_name = format!("{:?}", step);
536
537                    tracing::error!(
538                        "Migration failed at step {} - restoring from backups",
539                        step_name
540                    );
541
542                    let recovery_instructions = match backup_manager.restore_backups(&backups).await
543                    {
544                        Ok(()) => {
545                            tracing::info!("Successfully restored from backups");
546                            format!(
547                                "✓ Your data has been restored from backups.\n\
548                                 ✓ Config file: {}\n\
549                                 ✓ Database: {}\n\n\
550                                 The migration failed but your original data is intact.\n\
551                                 Please review the error below and try again after fixing the issue:\n\n\
552                                 Error: {}",
553                                backups.source_config.display(),
554                                backups.source_db.as_ref().map(|p| p.display().to_string()).unwrap_or_else(|| "N/A".to_string()),
555                                e
556                            )
557                        }
558                        Err(restore_err) => {
559                            format!(
560                                "CRITICAL ERROR: Migration failed AND backup restoration failed!\n\n\
561                                 Original error: {}\n\
562                                 Restore error: {}\n\n\
563                                 MANUAL RECOVERY REQUIRED:\n\
564                                 1. Restore config file from: {}\n\
565                                 2. Restore database from: {}\n\
566                                 3. Review logs for details\n\
567                                 4. Contact support if needed",
568                                e,
569                                restore_err,
570                                backups.source_config.display(),
571                                backups.source_db.as_ref().map(|p| p.display().to_string()).unwrap_or_else(|| "N/A".to_string())
572                            )
573                        }
574                    };
575
576                    result.errors.push(format!(
577                        "Migration failed at step '{}':\n{}\n\n{}",
578                        step_name, e, recovery_instructions
579                    ));
580
581                    self.rollback(&result).await;
582                    result.duration_ms = start_time.elapsed().as_millis() as u64;
583                    return result;
584                }
585            }
586        }
587
588        result.success = true;
589        result.duration_ms = start_time.elapsed().as_millis() as u64;
590
591        if let Err(e) = backup_manager
592            .cleanup_backups(&backups, options.keep_backups)
593            .await
594        {
595            tracing::warn!("Failed to cleanup backups: {}", e);
596        }
597
598        result
599    }
600
601    async fn execute_step(
602        &self, step: MigrationStep, plan: &MigrationPlan, options: &MigrationOptions,
603        target_manager: &StorageManager, result: &mut MigrationResult,
604    ) -> DomainResult<()> {
605        match step {
606            MigrationStep::ValidateTarget => {
607                if !target_manager.cache_backend().is_available().await {
608                    return Err(DomainError::NetworkError(
609                        "Target cache backend not available".into(),
610                    ));
611                }
612
613                target_manager.config_backend().list_providers().await?;
614            }
615
616            MigrationStep::MigrateTokens => {
617                tracing::info!("[Migration] Starting token migration...");
618                let password = get_vault_password(options)?;
619                tracing::info!(
620                    "[Migration] Vault password retrieved: {} chars",
621                    password.len()
622                );
623
624                tracing::info!("[Migration] Exporting tokens from source token store...");
625                let encrypted_blob = self.source_token_store.export_encrypted(&password).await?;
626                tracing::info!(
627                    "[Migration] Exported encrypted blob: {} bytes",
628                    encrypted_blob.len()
629                );
630
631                if !result.provider_id_mapping.is_empty() {
632                    tracing::info!(
633                        "[Migration] Using provider ID remapping path ({} mappings)",
634                        result.provider_id_mapping.len()
635                    );
636
637                    let source_tokens = self.source_token_store.get_all_tokens().await?;
638                    tracing::info!(
639                        "[Migration] Retrieved {} tokens from source",
640                        source_tokens.len()
641                    );
642
643                    let mut migrated_count = 0;
644                    for (old_id, token) in source_tokens {
645                        if let Some(&new_id) = result.provider_id_mapping.get(&old_id) {
646                            tracing::debug!(
647                                "[Migration] Storing token for provider {} → {} (token length: {})",
648                                old_id,
649                                new_id,
650                                token.len()
651                            );
652                            target_manager
653                                .token_store()
654                                .await
655                                .store_token(new_id, &token)
656                                .await?;
657                            migrated_count += 1;
658                            tracing::info!(
659                                "[Migration] Successfully stored token for provider {} (remapped to {})",
660                                old_id,
661                                new_id
662                            );
663                        } else {
664                            let source_providers =
665                                self.source_config_backend.list_providers().await?;
666                            return Err(DomainError::DataConsistency(format!(
667                                "Cannot migrate token for provider ID {} - provider not found in remapping.\n\n\
668                                This indicates data corruption or inconsistency.\n\n\
669                                Source provider IDs: {:?}\n\
670                                Remapping table: {:?}\n\n\
671                                This token would be lost if migration continued.\n\
672                                Please investigate why this provider's configuration wasn't migrated.",
673                                old_id,
674                                source_providers.iter().filter_map(|p| p.id).collect::<Vec<_>>(),
675                                result.provider_id_mapping
676                            )));
677                        }
678                    }
679                    tracing::info!(
680                        "[Migration] Migrated {} tokens via ID remapping",
681                        migrated_count
682                    );
683                } else {
684                    tracing::info!("[Migration] Using direct import path (no ID remapping)");
685                    tracing::info!(
686                        "[Migration] Importing encrypted blob ({} bytes) to target token store...",
687                        encrypted_blob.len()
688                    );
689                    target_manager
690                        .token_store()
691                        .await
692                        .import_encrypted(&encrypted_blob, &password)
693                        .await?;
694                    tracing::info!("[Migration] Direct import completed");
695                }
696
697                let count: usize = target_manager
698                    .token_store()
699                    .await
700                    .get_all_tokens()
701                    .await?
702                    .len();
703                tracing::info!(
704                    "[Migration] Verification: {} tokens now in target database",
705                    count
706                );
707                result.stats.tokens_migrated = count;
708            }
709
710            MigrationStep::MigrateConfigs => {
711                tracing::info!(
712                    "[Migration] Exporting providers from source database: {:?}",
713                    self.source_config.db_path()
714                );
715                let export = self.source_config_backend.export_all().await?;
716                result.stats.providers_migrated = export.providers.len();
717                result.stats.permissions_migrated = export.permissions.len();
718
719                tracing::info!(
720                    "[Migration] Importing {} providers to target database: {:?}",
721                    export.providers.len(),
722                    plan.to.db_path()
723                );
724                target_manager.config_backend().import_all(&export).await?;
725
726                let source_providers = self.source_config_backend.list_providers().await?;
727                let target_providers: Vec<ProviderConfig> =
728                    target_manager.config_backend().list_providers().await?;
729
730                for source_provider in &source_providers {
731                    if let Some(source_id) = source_provider.id {
732                        if let Some(target_provider) = target_providers
733                            .iter()
734                            .find(|p| p.name == source_provider.name)
735                        {
736                            if let Some(target_id) = target_provider.id {
737                                result.provider_id_mapping.insert(source_id, target_id);
738                            }
739                        }
740                    }
741                }
742
743                if !same_database(&plan.from, &plan.to) {
744                    let mut metrics_config = self.source_metrics.export_config().await?;
745                    metrics_config.pipeline_configs.retain_mut(|config| {
746                        match remap_pipeline_id(&config.pipeline_id, &result.provider_id_mapping) {
747                            Some(pipeline_id) => {
748                                config.pipeline_id = pipeline_id;
749                                true
750                            }
751                            None => false,
752                        }
753                    });
754                    MetricsRepository::new_from_pool(target_manager.database_pool())
755                        .import_config(&metrics_config)
756                        .await?;
757                }
758
759                if !same_database(&plan.from, &plan.to) {
760                    let subscriptions =
761                        PushRepository::new(self.source_pool.clone()).list().await?;
762                    let target_push = PushRepository::new(target_manager.database_pool());
763                    target_push.import(&subscriptions).await?;
764                    if options.clean_target {
765                        for stale in target_push.list().await? {
766                            if !subscriptions.iter().any(|s| s.id == stale.id) {
767                                target_push.delete(&stale.id).await?;
768                            }
769                        }
770                    }
771                    tracing::info!(
772                        "[Migration] Migrated {} push subscriptions",
773                        subscriptions.len()
774                    );
775                }
776
777                if !same_database(&plan.from, &plan.to) {
778                    let source_devices = DeviceRepository::new(self.source_pool.clone());
779                    let devices = source_devices.list().await?;
780                    let target_devices = DeviceRepository::new(target_manager.database_pool());
781                    target_devices.import(&devices).await?;
782                    for (id, preferences) in source_devices.notification_preferences().await? {
783                        let preferences =
784                            remap_preferences(preferences, &result.provider_id_mapping);
785                        target_devices
786                            .set_notification_preferences(&id, &preferences)
787                            .await?;
788                    }
789                    if options.clean_target {
790                        for stale in target_devices.list().await? {
791                            if !devices.iter().any(|d| d.id == stale.id) {
792                                target_devices.delete(&stale.id).await?;
793                            }
794                        }
795                    }
796                    tracing::info!("[Migration] Migrated {} paired devices", devices.len());
797                }
798
799                if !same_database(&plan.from, &plan.to) {
800                    let pushes = DevicePushRepository::new(self.source_pool.clone())
801                        .list()
802                        .await?;
803                    let target_pushes = DevicePushRepository::new(target_manager.database_pool());
804                    target_pushes.import(&pushes).await?;
805                    if options.clean_target {
806                        for stale in target_pushes.list().await? {
807                            if !pushes.iter().any(|p| p.device_id == stale.device_id) {
808                                target_pushes.delete(&stale.device_id).await?;
809                            }
810                        }
811                    }
812                    tracing::info!(
813                        "[Migration] Migrated {} device push registrations",
814                        pushes.len()
815                    );
816                }
817
818                if !same_database(&plan.from, &plan.to) {
819                    let mapping = &result.provider_id_mapping;
820                    let mut export = PipelineViewRepository::new(self.source_pool.clone())
821                        .export()
822                        .await?;
823                    let remap = |pipeline_id: &mut String, provider_id: &mut i64| match (
824                        remap_pipeline_id(pipeline_id, mapping),
825                        mapping.get(provider_id),
826                    ) {
827                        (Some(next), Some(target)) => {
828                            *pipeline_id = next;
829                            *provider_id = *target;
830                            true
831                        }
832                        _ => false,
833                    };
834                    export
835                        .members
836                        .retain_mut(|m| remap(&mut m.pipeline_id, &mut m.provider_id));
837                    for view in &mut export.views {
838                        view.rules.provider_ids = view
839                            .rules
840                            .provider_ids
841                            .iter()
842                            .filter_map(|id| mapping.get(id).copied())
843                            .collect();
844                    }
845                    PipelineViewRepository::new(target_manager.database_pool())
846                        .import(&export, options.clean_target)
847                        .await?;
848                    tracing::info!("[Migration] Migrated {} pipeline views", export.views.len());
849                }
850
851                if options.clean_target {
852                    let source_names: std::collections::HashSet<String> =
853                        source_providers.iter().map(|p| p.name.clone()).collect();
854
855                    let orphaned: Vec<_> = target_providers
856                        .iter()
857                        .filter(|p| !source_names.contains(&p.name))
858                        .collect();
859
860                    for orphan in &orphaned {
861                        if let Some(id) = orphan.id {
862                            tracing::info!(
863                                "[Migration] Removing orphaned provider: {} (id={})",
864                                orphan.name,
865                                id
866                            );
867                            target_manager.config_backend().delete_provider(id).await?;
868                            let _ = target_manager.token_store().await.delete_token(id).await;
869                            result.stats.tokens_cleaned += 1;
870                        }
871                    }
872
873                    if !orphaned.is_empty() {
874                        result.stats.providers_cleaned = orphaned.len();
875                        tracing::info!(
876                            "[Migration] Cleaned {} orphaned providers from target",
877                            orphaned.len()
878                        );
879                    }
880                }
881
882                if !same_database(&plan.from, &plan.to) {
883                    let alerts: Vec<DeviceAlert> =
884                        DeviceAlertRepository::new(self.source_pool.clone())
885                            .list()
886                            .await?
887                            .into_iter()
888                            .filter_map(|alert| remap_alert(alert, &result.provider_id_mapping))
889                            .collect();
890                    let target_alerts = DeviceAlertRepository::new(target_manager.database_pool());
891                    if options.clean_target {
892                        target_alerts.delete_all().await?;
893                        target_alerts.import(&alerts).await?;
894                    } else {
895                        target_alerts.merge(&alerts).await?;
896                    }
897                    tracing::info!("[Migration] Migrated {} device alerts", alerts.len());
898                }
899            }
900
901            MigrationStep::MigrateCache => {
902                let objects = self.source_cache_backend.list(None).await?;
903                tracing::info!(
904                    "[Migration] Migrating {} cache entries from {:?} to {:?}",
905                    objects.len(),
906                    self.source_config.data_dir().join("cache"),
907                    plan.to.data_dir().join("cache")
908                );
909                for obj in objects {
910                    let data = self.source_cache_backend.get(&obj.key).await?;
911                    target_manager
912                        .cache_backend()
913                        .put(&obj.key, &data, obj.content_type.as_deref())
914                        .await?;
915                    result.stats.cache_entries_migrated += 1;
916                }
917
918                if !same_database(&plan.from, &plan.to) {
919                    let source_repo = Repository::new(
920                        self.source_config_backend.clone(),
921                        self.source_pool.clone(),
922                        self.source_token_store.clone(),
923                    );
924                    let target_repo = Repository::new(
925                        target_manager.config_backend(),
926                        target_manager.database_pool(),
927                        target_manager.token_store().await,
928                    );
929                    for source_id in source_repo.list_run_history_pipeline_ids().await? {
930                        let Some(target_id) =
931                            remap_pipeline_id(&source_id, &result.provider_id_mapping)
932                        else {
933                            continue;
934                        };
935                        let mut runs = source_repo
936                            .get_cached_run_history(&source_id, i64::MAX as usize)
937                            .await?;
938                        for run in &mut runs {
939                            run.pipeline_id.clone_from(&target_id);
940                        }
941                        target_repo.cache_run_history(&target_id, &runs).await?;
942                        result.stats.cache_entries_migrated += runs.len();
943                    }
944
945                    let target_metrics =
946                        MetricsRepository::new_from_pool(target_manager.database_pool());
947                    let pipeline_query = |pipeline_id: &str| MetricsQuery {
948                        pipeline_id: Some(pipeline_id.to_string()),
949                        limit: None,
950                        ..Default::default()
951                    };
952                    let mut pipelines = 0;
953                    for source_id in self.source_metrics.list_pipeline_ids().await? {
954                        let Some(target_id) =
955                            remap_pipeline_id(&source_id, &result.provider_id_mapping)
956                        else {
957                            continue;
958                        };
959                        let last_run = self
960                            .source_metrics
961                            .get_last_processed_run(&source_id)
962                            .await?;
963                        let mut runs: BTreeMap<i64, RecordMetrics> = BTreeMap::new();
964                        for entry in self
965                            .source_metrics
966                            .query_metrics(pipeline_query(&source_id))
967                            .await?
968                        {
969                            let slot = entry.metric_type as usize;
970                            let record =
971                                runs.entry(entry.run_number)
972                                    .or_insert_with(|| RecordMetrics {
973                                        run_number: entry.run_number,
974                                        run_hash: entry.run_hash.clone().unwrap_or_default(),
975                                        entries: [None, None, None],
976                                    });
977                            record.entries[slot] = Some(entry);
978                        }
979                        let records: Vec<_> = runs.into_values().collect();
980                        target_metrics
981                            .refresh_record_metrics(&target_id, &records)
982                            .await?;
983                        result.stats.metrics_migrated += records
984                            .iter()
985                            .map(|record| record.entries.iter().flatten().count())
986                            .sum::<usize>();
987                        pipelines += 1;
988
989                        let Some(run) = last_run else {
990                            continue;
991                        };
992                        if target_metrics
993                            .get_last_processed_run(&target_id)
994                            .await?
995                            .is_none_or(|target_run| target_run < run)
996                        {
997                            target_metrics
998                                .update_last_processed_run(&target_id, run)
999                                .await?;
1000                        }
1001                    }
1002                    tracing::info!(
1003                        "[Migration] Migrated {} metric entries for {} pipelines",
1004                        result.stats.metrics_migrated,
1005                        pipelines
1006                    );
1007                }
1008            }
1009
1010            MigrationStep::VerifyMigration => {
1011                let source_providers = self.source_config_backend.list_providers().await?;
1012
1013                if plan.migrate_configs && !source_providers.is_empty() {
1014                    let target_providers: Vec<ProviderConfig> =
1015                        target_manager.config_backend().list_providers().await?;
1016
1017                    for source_provider in &source_providers {
1018                        let found = target_providers
1019                            .iter()
1020                            .any(|t| t.name == source_provider.name);
1021                        if !found {
1022                            return Err(DomainError::DatabaseError(format!(
1023                                "Provider '{}' from source was not found in target after migration",
1024                                source_provider.name
1025                            )));
1026                        }
1027                    }
1028
1029                    if options.clean_target {
1030                        if source_providers.len() != target_providers.len() {
1031                            return Err(DomainError::DatabaseError(format!(
1032                                "Provider count mismatch after cleanup: source={}, target={}",
1033                                source_providers.len(),
1034                                target_providers.len()
1035                            )));
1036                        }
1037                        tracing::info!(
1038                            "[Migration] Verified exact match: {} providers",
1039                            source_providers.len()
1040                        );
1041                    } else if target_providers.len() > source_providers.len() {
1042                        tracing::info!(
1043                            "[Migration] Merge complete: {} migrated, {} pre-existing kept (total: {})",
1044                            source_providers.len(),
1045                            target_providers.len() - source_providers.len(),
1046                            target_providers.len()
1047                        );
1048                    } else {
1049                        tracing::info!(
1050                            "[Migration] Verified {} providers migrated successfully",
1051                            source_providers.len()
1052                        );
1053                    }
1054                } else if plan.migrate_configs {
1055                    tracing::info!("Skipping provider verification: source database is empty");
1056                }
1057
1058                if plan.migrate_tokens {
1059                    let source_tokens = self.source_token_store.get_all_tokens().await?;
1060
1061                    if !source_tokens.is_empty() {
1062                        let target_tokens: HashMap<i64, String> =
1063                            target_manager.token_store().await.get_all_tokens().await?;
1064
1065                        if options.clean_target {
1066                            if source_tokens.len() != target_tokens.len() {
1067                                return Err(DomainError::DatabaseError(format!(
1068                                    "Token count mismatch after cleanup: source={}, target={}",
1069                                    source_tokens.len(),
1070                                    target_tokens.len()
1071                                )));
1072                            }
1073                            tracing::info!(
1074                                "[Migration] Verified exact match: {} tokens",
1075                                source_tokens.len()
1076                            );
1077                        } else {
1078                            if target_tokens.len() < source_tokens.len() {
1079                                return Err(DomainError::DatabaseError(format!(
1080                                    "Token count too low after migration: source={}, target={}",
1081                                    source_tokens.len(),
1082                                    target_tokens.len()
1083                                )));
1084                            }
1085                            tracing::info!(
1086                                "[Migration] Token migration complete: {} migrated (target has {} total)",
1087                                source_tokens.len(),
1088                                target_tokens.len()
1089                            );
1090                        }
1091                    } else {
1092                        tracing::info!("Skipping token verification: source has no tokens");
1093                    }
1094                }
1095            }
1096
1097            MigrationStep::UpdateConfig => {
1098                use crate::infrastructure::config::ConfigLoader;
1099
1100                let config_path = ConfigLoader::discover_config_path();
1101
1102                tracing::info!(
1103                    "Saving migrated config to: {} (data_dir: {:?}, backend: {})",
1104                    config_path.display(),
1105                    plan.to.data_dir(),
1106                    plan.to.storage.backend
1107                );
1108
1109                ConfigLoader::save(&plan.to, &config_path).map_err(|e| {
1110                    DomainError::DatabaseError(format!("Failed to save config: {}", e))
1111                })?;
1112            }
1113        }
1114
1115        Ok(())
1116    }
1117
1118    async fn emit_progress(&self, step: MigrationStep, step_index: usize, total_steps: usize) {
1119        if let Some(ref event_bus) = self.event_bus {
1120            event_bus
1121                .emit(crate::event::CoreEvent::MigrationProgress {
1122                    step: format!("{:?}", step),
1123                    step_index,
1124                    total_steps,
1125                    message: step.description().to_string(),
1126                })
1127                .await;
1128        }
1129    }
1130
1131    async fn rollback(&self, _result: &MigrationResult) {
1132        tracing::warn!("Migration failed. Manual intervention may be required.");
1133    }
1134}
1135
1136#[cfg(test)]
1137mod tests {
1138    use tempfile::TempDir;
1139
1140    use super::*;
1141    use crate::{
1142        infrastructure::token_store::MemoryTokenStore,
1143        test_env::EnvGuard,
1144    };
1145
1146    async fn create_test_manager(temp_dir: &TempDir) -> StorageManager {
1147        let mut config = PipedashConfig::default();
1148        config.storage.data_dir = temp_dir.path().to_string_lossy().to_string();
1149
1150        let token_store: Arc<dyn TokenStore> = Arc::new(MemoryTokenStore::new());
1151        StorageManager::with_token_store(config, token_store, true)
1152            .await
1153            .unwrap()
1154    }
1155
1156    #[tokio::test]
1157    async fn test_plan_migration_no_changes() {
1158        let temp_dir = TempDir::new().unwrap();
1159        let manager = create_test_manager(&temp_dir).await;
1160
1161        let orchestrator = MigrationOrchestrator::from_manager(&manager, None, None).await;
1162
1163        let target_config = manager.config().clone();
1164        let plan = orchestrator
1165            .plan_migration(target_config, &MigrationOptions::default())
1166            .unwrap();
1167
1168        assert!(!plan.steps.contains(&MigrationStep::MigrateConfigs));
1169        assert!(!plan.steps.contains(&MigrationStep::MigrateTokens));
1170    }
1171
1172    #[tokio::test]
1173    async fn test_migration_dry_run() {
1174        let temp_dir = TempDir::new().unwrap();
1175        let manager = create_test_manager(&temp_dir).await;
1176
1177        let orchestrator = MigrationOrchestrator::from_manager(&manager, None, None).await;
1178
1179        let target_config = manager.config().clone();
1180        let plan = MigrationPlan {
1181            from: manager.config().clone(),
1182            to: target_config,
1183            steps: vec![MigrationStep::ValidateTarget],
1184            migrate_tokens: false,
1185            migrate_configs: false,
1186            migrate_cache: false,
1187            backend_changed: false,
1188            data_dir_changed: false,
1189            created_at: Utc::now(),
1190        };
1191
1192        let result = orchestrator
1193            .execute_migration(
1194                plan,
1195                MigrationOptions {
1196                    dry_run: true,
1197                    ..Default::default()
1198                },
1199                true, // is_desktop
1200            )
1201            .await;
1202
1203        assert!(result.success);
1204    }
1205
1206    #[test]
1207    fn test_migration_step_descriptions() {
1208        assert!(!MigrationStep::ValidateTarget.description().is_empty());
1209        assert!(!MigrationStep::MigrateTokens.description().is_empty());
1210        assert!(!MigrationStep::MigrateConfigs.description().is_empty());
1211    }
1212
1213    #[test]
1214    fn test_migration_result_summary() {
1215        let result = MigrationResult {
1216            success: true,
1217            steps_completed: vec![MigrationStep::MigrateConfigs],
1218            errors: vec![],
1219            duration_ms: 100,
1220            stats: MigrationStats {
1221                providers_migrated: 5,
1222                tokens_migrated: 3,
1223                cache_entries_migrated: 10,
1224                metrics_migrated: 0,
1225                permissions_migrated: 2,
1226                providers_cleaned: 0,
1227                tokens_cleaned: 0,
1228            },
1229            provider_id_mapping: std::collections::HashMap::new(),
1230            backups: None,
1231        };
1232
1233        let summary = result.summary();
1234        assert!(summary.contains("5 providers"));
1235        assert!(summary.contains("10 cache entries"));
1236    }
1237
1238    #[tokio::test]
1239    async fn migrate_tokens_after_a_provider_was_deleted() {
1240        let source_dir = TempDir::new().unwrap();
1241        let target_dir = TempDir::new().unwrap();
1242        let config_path = source_dir.path().join("config.toml");
1243        std::fs::write(&config_path, "[storage]\nbackend = \"sqlite\"\n").unwrap();
1244        let _env = EnvGuard::set(&[("PIPEDASH_CONFIG_PATH", config_path.to_str())]);
1245
1246        let mut source = PipedashConfig::default();
1247        source.storage.data_dir = source_dir.path().to_string_lossy().to_string();
1248        source.storage.vault_password = Some("test-password".to_string());
1249        let mut target = source.clone();
1250        target.storage.data_dir = target_dir.path().to_string_lossy().to_string();
1251
1252        let manager = StorageManager::from_config(source, false).await.unwrap();
1253        let backend = manager.config_backend();
1254        let tokens = manager.token_store().await;
1255        let mut ids = Vec::new();
1256        for name in ["a", "b", "c"] {
1257            let provider = ProviderConfig {
1258                id: None,
1259                name: name.to_string(),
1260                provider_type: "github".to_string(),
1261                token: String::new(),
1262                config: HashMap::new(),
1263                refresh_interval: 30,
1264                version: None,
1265            };
1266            ids.push(backend.create_provider(&provider).await.unwrap());
1267        }
1268        backend.delete_provider(ids[1]).await.unwrap();
1269        tokens.store_token(ids[0], "tok-a").await.unwrap();
1270        tokens.store_token(ids[2], "tok-c").await.unwrap();
1271
1272        let orchestrator = MigrationOrchestrator::from_manager(&manager, None, None).await;
1273        let options = MigrationOptions {
1274            migrate_tokens: true,
1275            token_password: Some("test-password".to_string()),
1276            ..Default::default()
1277        };
1278        let plan = orchestrator
1279            .plan_migration(target.clone(), &options)
1280            .unwrap();
1281        let result = orchestrator.execute_migration(plan, options, false).await;
1282        assert!(result.success, "{:?}", result.errors);
1283
1284        let target_manager = StorageManager::from_config(target, false).await.unwrap();
1285        let mut migrated = HashMap::new();
1286        for provider in target_manager
1287            .config_backend()
1288            .list_providers()
1289            .await
1290            .unwrap()
1291        {
1292            let token = target_manager
1293                .token_store()
1294                .await
1295                .get_token(provider.id.unwrap())
1296                .await
1297                .unwrap();
1298            migrated.insert(provider.name, token);
1299        }
1300        assert_eq!(
1301            migrated,
1302            HashMap::from([
1303                ("a".to_string(), "tok-a".to_string()),
1304                ("c".to_string(), "tok-c".to_string()),
1305            ])
1306        );
1307    }
1308
1309    #[tokio::test]
1310    async fn migration_keeps_archived_runs_with_remapped_ids_after_restart() {
1311        use crate::domain::PipelineRun;
1312        use crate::infrastructure::database::Repository;
1313
1314        let source_dir = TempDir::new().unwrap();
1315        let target_dir = TempDir::new().unwrap();
1316        let config_path = source_dir.path().join("config.toml");
1317        std::fs::write(&config_path, "[storage]\nbackend = \"sqlite\"\n").unwrap();
1318        let _env = EnvGuard::set(&[("PIPEDASH_CONFIG_PATH", config_path.to_str())]);
1319
1320        let mut source = PipedashConfig::default();
1321        source.storage.data_dir = source_dir.path().to_string_lossy().to_string();
1322        source.storage.vault_password = Some("test-password".to_string());
1323        let mut target = source.clone();
1324        target.storage.data_dir = target_dir.path().to_string_lossy().to_string();
1325
1326        let manager = StorageManager::from_config(source, false).await.unwrap();
1327        let backend = manager.config_backend();
1328        let mut ids = Vec::new();
1329        for name in ["deleted", "flux"] {
1330            ids.push(
1331                backend
1332                    .create_provider(&ProviderConfig {
1333                        id: None,
1334                        name: name.to_string(),
1335                        provider_type: "fluxcd".to_string(),
1336                        token: String::new(),
1337                        config: HashMap::new(),
1338                        refresh_interval: 30,
1339                        version: None,
1340                    })
1341                    .await
1342                    .unwrap(),
1343            );
1344        }
1345        backend.delete_provider(ids[0]).await.unwrap();
1346        let repo = Repository::new(
1347            backend,
1348            manager.database_pool(),
1349            manager.token_store().await,
1350        );
1351        let source_pipeline = format!("fluxcd__{}__HelmRelease__apps__web", ids[1]);
1352        let runs: Vec<PipelineRun> = (1..=105)
1353            .rev()
1354            .map(|n| {
1355                serde_json::from_value(serde_json::json!({
1356                    "id": format!("release-{n}"),
1357                    "pipeline_id": source_pipeline,
1358                    "run_number": n,
1359                    "status": "success",
1360                    "started_at": "2026-10-20T00:00:00Z",
1361                    "logs_url": "",
1362                    "metadata": {
1363                        "uid": "resource-1",
1364                        "record_source": "status.history",
1365                        "release_version": n,
1366                        "archived": true
1367                    }
1368                }))
1369                .unwrap()
1370            })
1371            .collect();
1372        repo.cache_run_history(&source_pipeline, &runs)
1373            .await
1374            .unwrap();
1375        let other_pipeline = format!("fluxcd__{}__Kustomization__apps__api", ids[1]);
1376        let mut other = runs[0].clone();
1377        other.pipeline_id = other_pipeline.clone();
1378        repo.cache_run_history(&other_pipeline, &[other.clone()])
1379            .await
1380            .unwrap();
1381        let orphan_pipeline = format!("fluxcd__{}__HelmRelease__apps__web", ids[0]);
1382        let mut orphan = runs[0].clone();
1383        orphan.id = "deleted-provider-run".to_string();
1384        orphan.pipeline_id = orphan_pipeline;
1385        repo.cache_run_history(&orphan.pipeline_id, &[orphan.clone()])
1386            .await
1387            .unwrap();
1388
1389        let orchestrator = MigrationOrchestrator::from_manager(&manager, None, None).await;
1390        let options = MigrationOptions {
1391            migrate_cache: true,
1392            ..Default::default()
1393        };
1394        let plan = orchestrator
1395            .plan_migration(target.clone(), &options)
1396            .unwrap();
1397        let result = orchestrator.execute_migration(plan, options, false).await;
1398        assert!(result.success, "{:?}", result.errors);
1399        let target_id = result.provider_id_mapping[&ids[1]];
1400        assert_ne!(target_id, ids[1]);
1401        assert_eq!(result.stats.cache_entries_migrated, 106);
1402        assert_eq!(
1403            repo.get_cached_run_history(&source_pipeline, 200)
1404                .await
1405                .unwrap(),
1406            runs
1407        );
1408        drop(orchestrator);
1409        drop(repo);
1410        drop(manager);
1411
1412        let reopened = StorageManager::from_config(target, false).await.unwrap();
1413        let target_repo = Repository::new(
1414            reopened.config_backend(),
1415            reopened.database_pool(),
1416            reopened.token_store().await,
1417        );
1418        let target_pipeline = format!("fluxcd__{target_id}__HelmRelease__apps__web");
1419        let mut expected = runs;
1420        for run in &mut expected {
1421            run.pipeline_id = target_pipeline.clone();
1422        }
1423        assert_eq!(
1424            target_repo
1425                .get_cached_run_history(&target_pipeline, 200)
1426                .await
1427                .unwrap(),
1428            expected
1429        );
1430        other.pipeline_id = format!("fluxcd__{target_id}__Kustomization__apps__api");
1431        assert_eq!(
1432            target_repo
1433                .get_cached_run(&other.pipeline_id, other.run_number)
1434                .await
1435                .unwrap(),
1436            Some(other)
1437        );
1438        assert!(target_repo
1439            .get_cached_run_history(&source_pipeline, 200)
1440            .await
1441            .unwrap()
1442            .is_empty());
1443    }
1444
1445    #[tokio::test]
1446    async fn migration_keeps_pipeline_metrics() {
1447        use crate::domain::{
1448            MetricEntry,
1449            MetricType,
1450        };
1451        use crate::infrastructure::database::RecordMetrics;
1452
1453        let source_dir = TempDir::new().unwrap();
1454        let target_dir = TempDir::new().unwrap();
1455        let config_path = source_dir.path().join("config.toml");
1456        std::fs::write(&config_path, "[storage]\nbackend = \"sqlite\"\n").unwrap();
1457        let _env = EnvGuard::set(&[("PIPEDASH_CONFIG_PATH", config_path.to_str())]);
1458
1459        let mut source = PipedashConfig::default();
1460        source.storage.data_dir = source_dir.path().to_string_lossy().to_string();
1461        source.storage.vault_password = Some("test-password".to_string());
1462        let mut target = source.clone();
1463        target.storage.data_dir = target_dir.path().to_string_lossy().to_string();
1464
1465        let manager = StorageManager::from_config(source, false).await.unwrap();
1466        let backend = manager.config_backend();
1467        let mut ids = Vec::new();
1468        for name in ["deleted", "gh"] {
1469            let provider = ProviderConfig {
1470                id: None,
1471                name: name.to_string(),
1472                provider_type: "github".to_string(),
1473                token: String::new(),
1474                config: HashMap::new(),
1475                refresh_interval: 30,
1476                version: None,
1477            };
1478            ids.push(backend.create_provider(&provider).await.unwrap());
1479        }
1480        backend.delete_provider(ids[0]).await.unwrap();
1481        let orphan_pipeline = format!("github__{}__o__r__1", ids[0]);
1482        let source_pipeline = format!("github__{}__o__r__1", ids[1]);
1483        let expired_pipeline = format!("github__{}__o__r__2", ids[1]);
1484
1485        let metrics = MetricsRepository::new_from_pool(manager.database_pool());
1486        metrics.update_global_config(true, 30).await.unwrap();
1487        metrics
1488            .upsert_pipeline_config(&source_pipeline, true, 14)
1489            .await
1490            .unwrap();
1491        metrics
1492            .upsert_pipeline_config(&orphan_pipeline, false, 3)
1493            .await
1494            .unwrap();
1495        let entry = |pipeline_id: &str, n: i64| MetricEntry {
1496            id: 0,
1497            pipeline_id: pipeline_id.to_string(),
1498            run_number: n,
1499            timestamp: Utc::now(),
1500            metric_type: MetricType::RunDuration,
1501            value: 60.0 * n as f64,
1502            metadata: None,
1503            created_at: Utc::now(),
1504            run_hash: Some(format!("hash-{n}")),
1505        };
1506        let mut entries: Vec<MetricEntry> = (1..=5).map(|n| entry(&source_pipeline, n)).collect();
1507        entries.push(entry(&orphan_pipeline, 1));
1508        metrics.insert_metrics_batch(entries).await.unwrap();
1509        metrics
1510            .update_last_processed_run(&source_pipeline, 5)
1511            .await
1512            .unwrap();
1513        metrics
1514            .update_last_processed_run(&expired_pipeline, 9)
1515            .await
1516            .unwrap();
1517
1518        let orchestrator = MigrationOrchestrator::from_manager(&manager, None, None).await;
1519        let options = MigrationOptions {
1520            migrate_tokens: true,
1521            migrate_cache: true,
1522            token_password: Some("test-password".to_string()),
1523            ..Default::default()
1524        };
1525        let plan = orchestrator
1526            .plan_migration(target.clone(), &options)
1527            .unwrap();
1528        let migrate = || orchestrator.execute_migration(plan.clone(), options.clone(), false);
1529        let result = migrate().await;
1530        assert!(result.success, "{:?}", result.errors);
1531        assert_eq!(result.stats.metrics_migrated, 5);
1532        let refreshed = MetricEntry {
1533            run_hash: Some("hash-4b".to_string()),
1534            value: 999.0,
1535            ..entry(&source_pipeline, 4)
1536        };
1537        metrics
1538            .refresh_record_metrics(
1539                &source_pipeline,
1540                &[RecordMetrics {
1541                    run_number: 4,
1542                    run_hash: "hash-4b".to_string(),
1543                    entries: [Some(refreshed), None, None],
1544                }],
1545            )
1546            .await
1547            .unwrap();
1548        let result = migrate().await;
1549        assert!(result.success, "{:?}", result.errors);
1550
1551        let target_manager = StorageManager::from_config(target, false).await.unwrap();
1552        let target_id = target_manager
1553            .config_backend()
1554            .list_providers()
1555            .await
1556            .unwrap()[0]
1557            .id
1558            .unwrap();
1559        assert_ne!(target_id, ids[1]);
1560        let target_pipeline = format!("github__{target_id}__o__r__1");
1561        let target_metrics = MetricsRepository::new_from_pool(target_manager.database_pool());
1562
1563        let copied = target_metrics
1564            .query_metrics(MetricsQuery {
1565                pipeline_id: Some(target_pipeline.clone()),
1566                ..Default::default()
1567            })
1568            .await
1569            .unwrap();
1570        assert_eq!(copied.len(), 5);
1571        let run_4: Vec<_> = copied
1572            .into_iter()
1573            .filter(|entry| entry.run_number == 4)
1574            .map(|entry| (entry.run_hash, entry.value))
1575            .collect();
1576        assert_eq!(run_4, [(Some("hash-4b".to_string()), 999.0)]);
1577        let config = target_metrics
1578            .get_pipeline_config(&target_pipeline)
1579            .await
1580            .unwrap()
1581            .unwrap();
1582        assert!(config.enabled);
1583        assert_eq!(config.retention_days, 14);
1584        let global = target_metrics.get_global_config().await.unwrap();
1585        assert!(global.enabled);
1586        assert_eq!(global.default_retention_days, 30);
1587        assert_eq!(
1588            target_metrics
1589                .get_last_processed_run(&target_pipeline)
1590                .await
1591                .unwrap(),
1592            Some(5)
1593        );
1594        assert_eq!(
1595            target_metrics
1596                .get_last_processed_run(&format!("github__{target_id}__o__r__2"))
1597                .await
1598                .unwrap(),
1599            Some(9)
1600        );
1601        assert_eq!(
1602            target_metrics
1603                .export_config()
1604                .await
1605                .unwrap()
1606                .pipeline_configs
1607                .len(),
1608            1
1609        );
1610        assert_eq!(
1611            target_metrics
1612                .calculate_storage_stats()
1613                .await
1614                .unwrap()
1615                .total_metrics_count,
1616            5
1617        );
1618    }
1619
1620    #[tokio::test]
1621    async fn push_subscriptions_follow_a_storage_migration_without_the_cache() {
1622        let source_dir = TempDir::new().unwrap();
1623        let target_dir = TempDir::new().unwrap();
1624        let config_path = source_dir.path().join("config.toml");
1625        std::fs::write(&config_path, "[storage]\nbackend = \"sqlite\"\n").unwrap();
1626        let _env = EnvGuard::set(&[("PIPEDASH_CONFIG_PATH", config_path.to_str())]);
1627
1628        let mut source = PipedashConfig::default();
1629        source.storage.data_dir = source_dir.path().to_string_lossy().to_string();
1630        source.storage.vault_password = Some("test-password".to_string());
1631        let mut target = source.clone();
1632        target.storage.data_dir = target_dir.path().to_string_lossy().to_string();
1633
1634        let manager = StorageManager::from_config(source, false).await.unwrap();
1635        let push = PushRepository::new(manager.database_pool());
1636        push.register("relay-a", "iPhone", "ios").await.unwrap();
1637        let paired = push.register("relay-b", "iPad", "ios").await.unwrap();
1638        push.mark_success(&paired.id).await.unwrap();
1639        let expected = push.list().await.unwrap();
1640
1641        let orchestrator = MigrationOrchestrator::from_manager(&manager, None, None).await;
1642        let options = MigrationOptions {
1643            migrate_tokens: false,
1644            migrate_cache: false,
1645            ..Default::default()
1646        };
1647        let plan = orchestrator
1648            .plan_migration(target.clone(), &options)
1649            .unwrap();
1650        let result = orchestrator.execute_migration(plan, options, false).await;
1651        assert!(result.success, "{:?}", result.errors);
1652
1653        let target_manager = StorageManager::from_config(target, false).await.unwrap();
1654        let migrated = PushRepository::new(target_manager.database_pool())
1655            .list()
1656            .await
1657            .unwrap();
1658        assert_eq!(migrated, expected);
1659    }
1660
1661    #[tokio::test]
1662    async fn device_alerts_and_push_registrations_follow_a_storage_migration() {
1663        use crate::domain::{
1664            AlertKind,
1665            NewDeviceAlert,
1666        };
1667
1668        let source_dir = TempDir::new().unwrap();
1669        let target_dir = TempDir::new().unwrap();
1670        let config_path = source_dir.path().join("config.toml");
1671        std::fs::write(&config_path, "[storage]\nbackend = \"sqlite\"\n").unwrap();
1672        let _env = EnvGuard::set(&[("PIPEDASH_CONFIG_PATH", config_path.to_str())]);
1673
1674        let mut source = PipedashConfig::default();
1675        source.storage.data_dir = source_dir.path().to_string_lossy().to_string();
1676        source.storage.vault_password = Some("test-password".to_string());
1677        let mut target = source.clone();
1678        target.storage.data_dir = target_dir.path().to_string_lossy().to_string();
1679
1680        let manager = StorageManager::from_config(source, false).await.unwrap();
1681        let backend = manager.config_backend();
1682        let mut ids = Vec::new();
1683        for name in ["deleted", "gh"] {
1684            let provider = ProviderConfig {
1685                id: None,
1686                name: name.to_string(),
1687                provider_type: "github".to_string(),
1688                token: String::new(),
1689                config: HashMap::new(),
1690                refresh_interval: 30,
1691                version: None,
1692            };
1693            ids.push(backend.create_provider(&provider).await.unwrap());
1694        }
1695        backend.delete_provider(ids[0]).await.unwrap();
1696        let (gone, provider) = (ids[0], ids[1]);
1697        let pipeline = format!("github__{provider}__o__r__1");
1698        let new_alert =
1699            |key: &str, kind: AlertKind, pipeline_id: Option<&str>, provider_id: i64| {
1700                NewDeviceAlert {
1701                    key: key.to_string(),
1702                    kind,
1703                    title: "title".to_string(),
1704                    body: "body".to_string(),
1705                    pipeline_id: pipeline_id.map(str::to_string),
1706                    provider_id: Some(provider_id),
1707                    provider_ids: vec![provider_id],
1708                    finish: None,
1709                }
1710            };
1711        let alerts = DeviceAlertRepository::new(manager.database_pool());
1712        for _ in 0..3 {
1713            alerts
1714                .insert(&new_alert("x", AlertKind::Outage, None, gone))
1715                .await
1716                .unwrap();
1717        }
1718        alerts.delete_all().await.unwrap();
1719        for new in [
1720            new_alert(
1721                &format!("pipeline:{pipeline}"),
1722                AlertKind::Failed,
1723                Some(&pipeline),
1724                provider,
1725            ),
1726            NewDeviceAlert {
1727                provider_ids: vec![provider, gone],
1728                ..new_alert(
1729                    &format!("connection:{provider}"),
1730                    AlertKind::Outage,
1731                    None,
1732                    provider,
1733                )
1734            },
1735            new_alert("connection:99", AlertKind::Outage, None, gone),
1736        ] {
1737            alerts.insert(&new).await.unwrap();
1738        }
1739        let pushes = DevicePushRepository::new(manager.database_pool());
1740        pushes.set("device-1", "sub-1").await.unwrap();
1741        pushes.mark_success("device-1").await.unwrap();
1742        let expected_pushes = pushes.list().await.unwrap();
1743        let (device, _) = crate::domain::PairedDevice::issue("Pixel", "android");
1744        let devices = DeviceRepository::new(manager.database_pool());
1745        devices.insert(&device).await.unwrap();
1746        devices
1747            .set_notification_preferences(
1748                &device.id,
1749                &NotificationPreferences {
1750                    outage: false,
1751                    muted_providers: vec![provider, gone],
1752                    muted_pipelines: vec![pipeline.clone()],
1753                    watched_pipelines: vec![pipeline.clone(), format!("github__{gone}__x")],
1754                    watch_once: vec![OnceWatch {
1755                        pipeline_id: pipeline.clone(),
1756                        since: chrono::DateTime::UNIX_EPOCH,
1757                        fired_at: Some(chrono::DateTime::UNIX_EPOCH),
1758                    }],
1759                    ..NotificationPreferences::default()
1760                },
1761            )
1762            .await
1763            .unwrap();
1764
1765        let target_manager = StorageManager::from_config(target.clone(), false)
1766            .await
1767            .unwrap();
1768        DeviceAlertRepository::new(target_manager.database_pool())
1769            .insert(&new_alert("stale", AlertKind::Failed, None, 1))
1770            .await
1771            .unwrap();
1772        DevicePushRepository::new(target_manager.database_pool())
1773            .set("device-1", "stale-sub")
1774            .await
1775            .unwrap();
1776        drop(target_manager);
1777
1778        let orchestrator = MigrationOrchestrator::from_manager(&manager, None, None).await;
1779        let options = MigrationOptions {
1780            migrate_tokens: false,
1781            migrate_cache: false,
1782            clean_target: true,
1783            ..Default::default()
1784        };
1785        let plan = orchestrator
1786            .plan_migration(target.clone(), &options)
1787            .unwrap();
1788        let result = orchestrator.execute_migration(plan, options, false).await;
1789        assert!(result.success, "{:?}", result.errors);
1790
1791        let target_manager = StorageManager::from_config(target, false).await.unwrap();
1792        let target_provider = target_manager
1793            .config_backend()
1794            .list_providers()
1795            .await
1796            .unwrap()
1797            .into_iter()
1798            .find(|p| p.name == "gh")
1799            .and_then(|p| p.id)
1800            .unwrap();
1801        let migrated = DeviceAlertRepository::new(target_manager.database_pool());
1802        let moved = migrated.list().await.unwrap();
1803        assert_eq!(
1804            moved
1805                .iter()
1806                .map(|a| (a.seq, a.key.as_str(), a.provider_id))
1807                .collect::<Vec<_>>(),
1808            vec![
1809                (
1810                    4,
1811                    format!("pipeline:github__{target_provider}__o__r__1").as_str(),
1812                    Some(target_provider)
1813                ),
1814                (
1815                    5,
1816                    format!("connection:{target_provider}").as_str(),
1817                    Some(target_provider)
1818                ),
1819            ]
1820        );
1821        assert_eq!(
1822            moved[0].pipeline_id.as_deref(),
1823            Some(format!("github__{target_provider}__o__r__1").as_str())
1824        );
1825        assert_eq!(moved[0].provider_ids, vec![target_provider]);
1826        assert_eq!(moved[1].provider_ids, vec![target_provider]);
1827        assert_ne!(target_provider, provider);
1828        assert_eq!(
1829            DeviceRepository::new(target_manager.database_pool())
1830                .notification_preferences()
1831                .await
1832                .unwrap()[&device.id],
1833            NotificationPreferences {
1834                outage: false,
1835                muted_providers: vec![target_provider],
1836                muted_pipelines: vec![format!("github__{target_provider}__o__r__1")],
1837                watched_pipelines: vec![format!("github__{target_provider}__o__r__1")],
1838                watch_once: vec![OnceWatch {
1839                    pipeline_id: format!("github__{target_provider}__o__r__1"),
1840                    since: chrono::DateTime::UNIX_EPOCH,
1841                    fired_at: Some(chrono::DateTime::UNIX_EPOCH),
1842                }],
1843                ..NotificationPreferences::default()
1844            }
1845        );
1846        assert_eq!(
1847            DevicePushRepository::new(target_manager.database_pool())
1848                .list()
1849                .await
1850                .unwrap(),
1851            expected_pushes
1852        );
1853        let next = migrated
1854            .insert(&new_alert("c", AlertKind::Failed, None, target_provider))
1855            .await
1856            .unwrap();
1857        assert_eq!(next.seq, 6);
1858    }
1859
1860    #[tokio::test]
1861    async fn a_storage_migration_keeps_the_target_alerts_unless_the_target_is_cleaned() {
1862        use crate::domain::{
1863            AlertKind,
1864            NewDeviceAlert,
1865        };
1866
1867        let source_dir = TempDir::new().unwrap();
1868        let target_dir = TempDir::new().unwrap();
1869        let config_path = source_dir.path().join("config.toml");
1870        std::fs::write(&config_path, "[storage]\nbackend = \"sqlite\"\n").unwrap();
1871        let _env = EnvGuard::set(&[("PIPEDASH_CONFIG_PATH", config_path.to_str())]);
1872
1873        let mut source = PipedashConfig::default();
1874        source.storage.data_dir = source_dir.path().to_string_lossy().to_string();
1875        source.storage.vault_password = Some("test-password".to_string());
1876        let mut target = source.clone();
1877        target.storage.data_dir = target_dir.path().to_string_lossy().to_string();
1878
1879        let new_alert = |key: &str| NewDeviceAlert {
1880            key: key.to_string(),
1881            kind: AlertKind::Failed,
1882            title: "title".to_string(),
1883            body: "body".to_string(),
1884            pipeline_id: None,
1885            provider_id: None,
1886            provider_ids: Vec::new(),
1887            finish: None,
1888        };
1889        let manager = StorageManager::from_config(source, false).await.unwrap();
1890        let source_alerts = DeviceAlertRepository::new(manager.database_pool());
1891        for key in ["source-a", "source-b"] {
1892            source_alerts.insert(&new_alert(key)).await.unwrap();
1893        }
1894
1895        let target_manager = StorageManager::from_config(target.clone(), false)
1896            .await
1897            .unwrap();
1898        let target_alerts = DeviceAlertRepository::new(target_manager.database_pool());
1899        let mut target_ids = Vec::new();
1900        for key in ["target-a", "target-b", "target-c"] {
1901            target_ids.push(target_alerts.insert(&new_alert(key)).await.unwrap().id);
1902        }
1903        drop(target_manager);
1904
1905        let orchestrator = MigrationOrchestrator::from_manager(&manager, None, None).await;
1906        let options = MigrationOptions {
1907            migrate_tokens: false,
1908            migrate_cache: false,
1909            ..Default::default()
1910        };
1911        let plan = orchestrator
1912            .plan_migration(target.clone(), &options)
1913            .unwrap();
1914        let result = orchestrator.execute_migration(plan, options, false).await;
1915        assert!(result.success, "{:?}", result.errors);
1916
1917        let target_manager = StorageManager::from_config(target, false).await.unwrap();
1918        let migrated = DeviceAlertRepository::new(target_manager.database_pool());
1919        let merged = migrated.list().await.unwrap();
1920        assert_eq!(
1921            merged
1922                .iter()
1923                .map(|a| (a.seq, a.key.as_str()))
1924                .collect::<Vec<_>>(),
1925            vec![
1926                (1, "source-a"),
1927                (2, "source-b"),
1928                (3, "target-c"),
1929                (4, "target-a"),
1930                (5, "target-b"),
1931            ]
1932        );
1933        for id in &target_ids {
1934            assert!(migrated.get(id).await.unwrap().is_some());
1935        }
1936        assert_eq!(migrated.insert(&new_alert("next")).await.unwrap().seq, 6);
1937    }
1938
1939    #[tokio::test]
1940    async fn views_and_pins_survive_a_restart_and_a_storage_migration() {
1941        use crate::domain::{
1942            PinKind,
1943            SectionMode,
1944            ViewDisplay,
1945            ViewKind,
1946            ViewLayout,
1947            ViewRules,
1948        };
1949        use crate::infrastructure::database::ViewExport;
1950
1951        let source_dir = TempDir::new().unwrap();
1952        let target_dir = TempDir::new().unwrap();
1953        let config_path = source_dir.path().join("config.toml");
1954        std::fs::write(&config_path, "[storage]\nbackend = \"sqlite\"\n").unwrap();
1955        let _env = EnvGuard::set(&[("PIPEDASH_CONFIG_PATH", config_path.to_str())]);
1956
1957        let mut source = PipedashConfig::default();
1958        source.storage.data_dir = source_dir.path().to_string_lossy().to_string();
1959        source.storage.vault_password = Some("test-password".to_string());
1960        let mut target = source.clone();
1961        target.storage.data_dir = target_dir.path().to_string_lossy().to_string();
1962
1963        let manager = StorageManager::from_config(source.clone(), false)
1964            .await
1965            .unwrap();
1966        let backend = manager.config_backend();
1967        let mut ids = Vec::new();
1968        for name in ["deleted", "gh"] {
1969            let provider = ProviderConfig {
1970                id: None,
1971                name: name.to_string(),
1972                provider_type: "github".to_string(),
1973                token: String::new(),
1974                config: HashMap::new(),
1975                refresh_interval: 30,
1976                version: None,
1977            };
1978            ids.push(backend.create_provider(&provider).await.unwrap());
1979        }
1980        backend.delete_provider(ids[0]).await.unwrap();
1981        let provider = ids[1];
1982        let pipeline = |n: u32| format!("github__{provider}__o__r__{n}");
1983
1984        let repo = PipelineViewRepository::new(manager.database_pool());
1985        let rules = ViewRules {
1986            provider_ids: vec![provider],
1987            repositories: vec!["o/*".to_string()],
1988            name_contains: Some("deploy".to_string()),
1989            statuses: Vec::new(),
1990        };
1991        let cards = ViewDisplay {
1992            layout: ViewLayout::Cards,
1993            sections: SectionMode::Provider,
1994            ..ViewDisplay::default()
1995        };
1996        let release = repo
1997            .insert_custom("Release", None, &rules, &cards)
1998            .await
1999            .unwrap();
2000        let prod = repo
2001            .insert_custom(
2002                "Prod",
2003                Some(release),
2004                &ViewRules::default(),
2005                &ViewDisplay::default(),
2006            )
2007            .await
2008            .unwrap();
2009        let eu = repo
2010            .insert_custom(
2011                "EU",
2012                Some(prod),
2013                &ViewRules::default(),
2014                &ViewDisplay::default(),
2015            )
2016            .await
2017            .unwrap();
2018        repo.upsert_member(eu, provider, &pipeline(1), false)
2019            .await
2020            .unwrap();
2021        repo.upsert_member(release, provider, &pipeline(2), true)
2022            .await
2023            .unwrap();
2024        let pinned = repo
2025            .list_views()
2026            .await
2027            .unwrap()
2028            .into_iter()
2029            .find(|v| v.kind == ViewKind::Pinned)
2030            .unwrap();
2031        repo.update_view(
2032            pinned.id,
2033            &pinned.name,
2034            None,
2035            &pinned.rules,
2036            &ViewDisplay {
2037                layout: ViewLayout::Table,
2038                ..pinned.display
2039            },
2040        )
2041        .await
2042        .unwrap();
2043        repo.insert_pin(PinKind::View, &eu.to_string())
2044            .await
2045            .unwrap();
2046        repo.insert_pin(PinKind::Repository, "o/r").await.unwrap();
2047        repo.insert_pin(PinKind::View, &release.to_string())
2048            .await
2049            .unwrap();
2050
2051        let snapshot = |export: ViewExport, provider: i64| {
2052            let pid = |id: &str| id.replace(&format!("__{provider}__"), "__P__");
2053            let view_name = |id: i64| {
2054                export
2055                    .views
2056                    .iter()
2057                    .find(|v| v.id == id)
2058                    .unwrap()
2059                    .name
2060                    .clone()
2061            };
2062            let views: Vec<_> = export
2063                .views
2064                .iter()
2065                .map(|v| {
2066                    let mut rules = v.rules.clone();
2067                    rules.provider_ids = rules
2068                        .provider_ids
2069                        .iter()
2070                        .map(|id| i64::from(*id == provider))
2071                        .collect();
2072                    (
2073                        v.kind,
2074                        v.name.clone(),
2075                        v.parent_id.map(view_name),
2076                        v.position,
2077                        rules,
2078                        v.display,
2079                    )
2080                })
2081                .collect();
2082            let members: Vec<_> = export
2083                .members
2084                .iter()
2085                .map(|m| {
2086                    (
2087                        view_name(m.view_id),
2088                        pid(&m.pipeline_id),
2089                        m.excluded,
2090                        m.provider_id == provider,
2091                    )
2092                })
2093                .collect();
2094            let pins: Vec<_> = export
2095                .pins
2096                .iter()
2097                .map(|p| match p.kind {
2098                    PinKind::View => view_name(p.reference.parse().unwrap()),
2099                    PinKind::Repository => p.reference.clone(),
2100                })
2101                .collect();
2102            format!("{views:?}\n{members:?}\n{pins:?}")
2103        };
2104        let before = snapshot(repo.export().await.unwrap(), provider);
2105        drop(repo);
2106        drop(manager);
2107
2108        let manager = StorageManager::from_config(source, false).await.unwrap();
2109        let reopened = PipelineViewRepository::new(manager.database_pool())
2110            .export()
2111            .await
2112            .unwrap();
2113        assert_eq!(snapshot(reopened, provider), before);
2114
2115        let orchestrator = MigrationOrchestrator::from_manager(&manager, None, None).await;
2116        let options = MigrationOptions {
2117            migrate_tokens: false,
2118            migrate_cache: false,
2119            ..Default::default()
2120        };
2121        let plan = orchestrator
2122            .plan_migration(target.clone(), &options)
2123            .unwrap();
2124        let result = orchestrator.execute_migration(plan, options, false).await;
2125        assert!(result.success, "{:?}", result.errors);
2126
2127        let target_manager = StorageManager::from_config(target, false).await.unwrap();
2128        let target_provider = target_manager
2129            .config_backend()
2130            .list_providers()
2131            .await
2132            .unwrap()[0]
2133            .id
2134            .unwrap();
2135        assert_ne!(target_provider, provider);
2136        let migrated = PipelineViewRepository::new(target_manager.database_pool())
2137            .export()
2138            .await
2139            .unwrap();
2140        assert_eq!(snapshot(migrated, target_provider), before);
2141    }
2142}