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 DomainError,
29 DomainResult,
30 MetricsQuery,
31 ProviderConfig,
32};
33use crate::event::EventBus;
34use crate::infrastructure::config::{
35 PipedashConfig,
36 StorageConfig,
37 StorageManager,
38};
39use crate::infrastructure::token_store::TokenStore;
40use crate::infrastructure::{
41 database::{
42 MetricsRepository,
43 RecordMetrics,
44 },
45 ConfigBackend,
46 StorageBackend,
47};
48
49#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
50pub enum MigrationStep {
51 ValidateTarget,
52 MigrateTokens,
53 MigrateConfigs,
54 MigrateCache,
55 VerifyMigration,
56 UpdateConfig,
57}
58
59impl MigrationStep {
60 pub fn description(&self) -> &str {
61 match self {
62 MigrationStep::ValidateTarget => "Validating target configuration",
63 MigrationStep::MigrateTokens => "Migrating tokens",
64 MigrationStep::MigrateConfigs => "Migrating provider configurations",
65 MigrationStep::MigrateCache => "Migrating cached data",
66 MigrationStep::VerifyMigration => "Verifying migration",
67 MigrationStep::UpdateConfig => "Updating storage configuration",
68 }
69 }
70}
71
72#[derive(Debug, Clone, Serialize, Deserialize)]
73pub struct MigrationPlan {
74 pub from: PipedashConfig,
75 pub to: PipedashConfig,
76 pub steps: Vec<MigrationStep>,
77 pub migrate_tokens: bool,
78 pub migrate_configs: bool,
79 pub migrate_cache: bool,
80 pub backend_changed: bool,
81 pub data_dir_changed: bool,
82 pub created_at: DateTime<Utc>,
83}
84
85impl MigrationPlan {
86 pub fn summary(&self) -> String {
87 format!(
88 "Migrate from {} to {} ({} steps)",
89 self.from.storage.summary(),
90 self.to.storage.summary(),
91 self.steps.len()
92 )
93 }
94}
95
96#[derive(Debug, Clone, Serialize, Deserialize)]
97pub struct MigrationProgress {
98 pub current_step: MigrationStep,
99 pub step_index: usize,
100 pub total_steps: usize,
101 pub message: String,
102}
103
104#[derive(Debug, Clone, Serialize, Deserialize)]
105pub struct MigrationResult {
106 pub success: bool,
107 pub steps_completed: Vec<MigrationStep>,
108 pub errors: Vec<String>,
109 pub duration_ms: u64,
110 pub stats: MigrationStats,
111 #[serde(default)]
112 pub provider_id_mapping: std::collections::HashMap<i64, i64>,
113 #[serde(skip_serializing_if = "Option::is_none")]
114 pub backups: Option<BackupPaths>,
115}
116
117impl MigrationResult {
118 pub fn summary(&self) -> String {
119 if self.success {
120 format!(
121 "Migration successful: {} providers, {} cache entries migrated in {}ms",
122 self.stats.providers_migrated, self.stats.cache_entries_migrated, self.duration_ms
123 )
124 } else {
125 format!(
126 "Migration failed after {} steps: {}",
127 self.steps_completed.len(),
128 self.errors.join(", ")
129 )
130 }
131 }
132}
133
134#[derive(Debug, Clone, Default, Serialize, Deserialize)]
135pub struct MigrationStats {
136 pub providers_migrated: usize,
137 pub tokens_migrated: usize,
138 pub cache_entries_migrated: usize,
139 pub metrics_migrated: usize,
140 pub permissions_migrated: usize,
141 pub providers_cleaned: usize,
142 pub tokens_cleaned: usize,
143}
144
145#[derive(Debug, Clone, Serialize, Deserialize)]
146pub struct MigrationOptions {
147 pub migrate_tokens: bool,
148 pub migrate_cache: bool,
149 pub token_password: Option<String>,
150 pub dry_run: bool,
151 #[serde(default)]
152 pub allow_non_empty_target: bool,
153 #[serde(default)]
154 pub clean_target: bool,
155 #[serde(default)]
156 pub keep_backups: bool,
157 #[serde(default = "default_max_retries")]
158 pub max_retries: u32,
159}
160
161fn default_max_retries() -> u32 {
162 3
163}
164
165impl Default for MigrationOptions {
166 fn default() -> Self {
167 Self {
168 migrate_tokens: false,
169 migrate_cache: false,
170 token_password: None,
171 dry_run: false,
172 allow_non_empty_target: false,
173 clean_target: false,
174 keep_backups: false,
175 max_retries: 3,
176 }
177 }
178}
179
180#[derive(Debug, Clone, Serialize, Deserialize)]
181pub struct ValidationReport {
182 pub errors: Vec<String>,
183 pub warnings: Vec<String>,
184 pub recommendations: Vec<String>,
185}
186
187impl ValidationReport {
188 pub fn is_valid(&self) -> bool {
189 self.errors.is_empty()
190 }
191}
192
193pub const VAULT_PASSWORD_ENV_VAR: &str = "PIPEDASH_VAULT_PASSWORD";
194
195fn get_vault_password(options: &MigrationOptions) -> DomainResult<String> {
196 if let Ok(password) = std::env::var(VAULT_PASSWORD_ENV_VAR) {
197 if !password.is_empty() {
198 return Ok(password);
199 }
200 }
201
202 options.token_password.clone().ok_or_else(|| {
203 DomainError::InvalidConfig(format!(
204 "Vault password required for token migration. Set {} environment variable or provide token_password in options.",
205 VAULT_PASSWORD_ENV_VAR
206 ))
207 })
208}
209
210fn remap_pipeline_id(pipeline_id: &str, provider_ids: &HashMap<i64, i64>) -> Option<String> {
211 let mut parts = pipeline_id.splitn(3, "__");
212 let (Some(provider_type), Some(provider_id), Some(rest)) =
213 (parts.next(), parts.next(), parts.next())
214 else {
215 return None;
216 };
217 let new_id = provider_ids.get(&provider_id.parse().ok()?)?;
218
219 Some(format!("{provider_type}__{new_id}__{rest}"))
220}
221
222fn same_database(from: &PipedashConfig, to: &PipedashConfig) -> bool {
223 from.storage.backend == to.storage.backend
224 && if to.storage.backend.is_sqlite() {
225 from.db_path() == to.db_path()
226 } else {
227 from.storage.postgres.connection_string == to.storage.postgres.connection_string
228 }
229}
230
231pub struct MigrationOrchestrator {
232 source_config: PipedashConfig,
233 source_token_store: Arc<dyn TokenStore>,
234 source_config_backend: Arc<dyn ConfigBackend>,
235 source_cache_backend: Arc<dyn StorageBackend>,
236 source_metrics: MetricsRepository,
237 event_bus: Option<Arc<dyn EventBus>>,
238 target_token_store: Option<Arc<dyn TokenStore>>,
239}
240
241impl MigrationOrchestrator {
242 pub async fn from_manager(
243 manager: &StorageManager, event_bus: Option<Arc<dyn EventBus>>,
244 target_token_store: Option<Arc<dyn TokenStore>>,
245 ) -> Self {
246 Self {
247 source_config: manager.config().clone(),
248 source_token_store: manager.token_store().await,
249 source_config_backend: manager.config_backend(),
250 source_cache_backend: manager.cache_backend(),
251 source_metrics: MetricsRepository::new_from_pool(manager.database_pool()),
252 event_bus,
253 target_token_store,
254 }
255 }
256
257 pub fn plan_migration(
258 &self, mut target_config: PipedashConfig, options: &MigrationOptions,
259 ) -> DomainResult<MigrationPlan> {
260 use crate::infrastructure::config::StorageBackend as ConfigStorageBackend;
261
262 let source_data_dir = self.source_config.data_dir();
263 let target_data_dir = target_config.data_dir();
264 let data_dir_changed = source_data_dir != target_data_dir;
265
266 let source_storage = &self.source_config.storage;
267 let target_storage = &mut target_config.storage;
268
269 if data_dir_changed {
270 tracing::info!(
271 "[Migration Planning] Data directory change detected: {:?} → {:?}",
272 source_data_dir,
273 target_data_dir
274 );
275 }
276
277 let backend_changed = source_storage.backend != target_storage.backend;
278
279 let token_backend_changing = {
280 let source_uses_keyring = source_storage.backend.is_sqlite()
281 && std::env::var(VAULT_PASSWORD_ENV_VAR).is_err();
282 let target_uses_encrypted = target_storage.backend.is_sqlite()
283 && (options.token_password.is_some()
284 || std::env::var(VAULT_PASSWORD_ENV_VAR).is_ok());
285
286 source_uses_keyring && target_uses_encrypted
287 };
288
289 if token_backend_changing {
290 tracing::info!(
291 "[Migration Planning] Token backend change detected: keyring → encrypted storage"
292 );
293 }
294
295 let needs_data_migration = backend_changed || data_dir_changed || token_backend_changing;
296
297 if target_storage.backend == ConfigStorageBackend::Postgres {
298 if target_storage.data_dir.starts_with("postgresql://")
299 || target_storage.data_dir.starts_with("postgres://")
300 {
301 tracing::warn!(
302 "Target data_dir contains PostgreSQL connection string: {}. \
303 Resetting to empty (will use default) - connection string should be in postgres.connection_string field.",
304 target_storage.data_dir
305 );
306 target_storage.data_dir = String::new(); } else if target_storage.data_dir.is_empty() {
308 tracing::debug!("Target uses PostgreSQL - data_dir is empty (will use default for cache/local files)");
309 }
310 }
311
312 let mut steps = Vec::new();
313
314 steps.push(MigrationStep::ValidateTarget);
315
316 if needs_data_migration {
317 steps.push(MigrationStep::MigrateConfigs);
318 }
319
320 if needs_data_migration && options.migrate_tokens {
321 steps.push(MigrationStep::MigrateTokens);
322 }
323
324 if needs_data_migration && options.migrate_cache {
325 steps.push(MigrationStep::MigrateCache);
326 }
327
328 if !steps.is_empty() {
329 steps.push(MigrationStep::VerifyMigration);
330 steps.push(MigrationStep::UpdateConfig);
331 }
332
333 Ok(MigrationPlan {
334 from: self.source_config.clone(),
335 to: target_config,
336 steps,
337 migrate_tokens: needs_data_migration && options.migrate_tokens,
338 migrate_configs: needs_data_migration,
339 migrate_cache: needs_data_migration && options.migrate_cache,
340 backend_changed,
341 data_dir_changed,
342 created_at: Utc::now(),
343 })
344 }
345
346 pub async fn validate_target_config(
347 &self, config: &StorageConfig,
348 ) -> DomainResult<ValidationReport> {
349 use crate::infrastructure::config::StorageBackend as ConfigStorageBackend;
350
351 let mut report = ValidationReport {
352 errors: Vec::new(),
353 warnings: Vec::new(),
354 recommendations: Vec::new(),
355 };
356
357 if config.backend == ConfigStorageBackend::Postgres {
358 if config.postgres.connection_string.is_empty() {
359 report.errors.push(
360 "PostgreSQL connection string required when using postgres backend".into(),
361 );
362 }
363
364 let has_settings_password = config.vault_password.is_some();
365 let has_env_password = std::env::var(VAULT_PASSWORD_ENV_VAR).is_ok();
366
367 if !has_settings_password && !has_env_password {
368 report.warnings.push(format!(
369 "Vault password not provided. Set {} environment variable or provide password during setup.",
370 VAULT_PASSWORD_ENV_VAR
371 ));
372 }
373 }
374
375 Ok(report)
376 }
377
378 pub async fn execute_migration(
379 &self, plan: MigrationPlan, options: MigrationOptions, is_desktop: bool,
380 ) -> MigrationResult {
381 let start_time = std::time::Instant::now();
382 let mut result = MigrationResult {
383 success: false,
384 steps_completed: Vec::new(),
385 errors: Vec::new(),
386 duration_ms: 0,
387 stats: MigrationStats::default(),
388 provider_id_mapping: std::collections::HashMap::new(),
389 backups: None,
390 };
391
392 if options.dry_run {
393 result.success = true;
394 result.duration_ms = start_time.elapsed().as_millis() as u64;
395 return result;
396 }
397
398 let backup_manager = BackupManager::new(plan.from.data_dir());
399 let backups = match backup_manager
400 .create_backups(&plan.from, &plan.to, Some(&*self.source_config_backend))
401 .await
402 {
403 Ok(backups) => backups,
404 Err(e) => {
405 result.errors.push(format!(
406 "Failed to create backups: {}\n\n\
407 WARNING: Migration aborted - backup creation is required for safe migration.\n\
408 INFO: Your data is safe - no modifications were performed.\n\n\
409 Please ensure:\n\
410 1. Sufficient disk space (need ~2x source database size)\n\
411 2. Write permissions to data directory\n\
412 3. Source database is not locked by another process",
413 e
414 ));
415 result.duration_ms = start_time.elapsed().as_millis() as u64;
416 return result;
417 }
418 };
419 result.backups = Some(backups.clone());
420
421 let target_manager = if let Some(token_store) = &self.target_token_store {
422 match StorageManager::with_token_store(plan.to.clone(), token_store.clone(), is_desktop)
423 .await
424 {
425 Ok(m) => m,
426 Err(e) => {
427 result.errors.push(format!(
428 "Failed to create target manager with token store: {}\n\n\
429 INFO: Migration aborted before any changes were made.\n\
430 INFO: Your data is safe - no modifications were performed.",
431 e
432 ));
433 backup_manager.cleanup_backups(&backups, false).await.ok();
434 result.duration_ms = start_time.elapsed().as_millis() as u64;
435 return result;
436 }
437 }
438 } else {
439 match StorageManager::from_config(plan.to.clone(), false).await {
440 Ok(m) => m,
441 Err(e) => {
442 result.errors.push(format!(
443 "Failed to create target manager: {}\n\n\
444 INFO: Migration aborted before any changes were made.\n\
445 INFO: Your data is safe - no modifications were performed.",
446 e
447 ));
448 backup_manager.cleanup_backups(&backups, false).await.ok();
449 result.duration_ms = start_time.elapsed().as_millis() as u64;
450 return result;
451 }
452 }
453 };
454
455 for (idx, step) in plan.steps.iter().enumerate() {
456 self.emit_progress(*step, idx, plan.steps.len()).await;
457
458 let step_result = self
459 .execute_step(*step, &plan, &options, &target_manager, &mut result)
460 .await;
461
462 match step_result {
463 Ok(()) => {
464 result.steps_completed.push(*step);
465 }
466 Err(e) => {
467 let step_name = format!("{:?}", step);
468
469 tracing::error!(
470 "Migration failed at step {} - restoring from backups",
471 step_name
472 );
473
474 let recovery_instructions = match backup_manager.restore_backups(&backups).await
475 {
476 Ok(()) => {
477 tracing::info!("Successfully restored from backups");
478 format!(
479 "✓ Your data has been restored from backups.\n\
480 ✓ Config file: {}\n\
481 ✓ Database: {}\n\n\
482 The migration failed but your original data is intact.\n\
483 Please review the error below and try again after fixing the issue:\n\n\
484 Error: {}",
485 backups.source_config.display(),
486 backups.source_db.as_ref().map(|p| p.display().to_string()).unwrap_or_else(|| "N/A".to_string()),
487 e
488 )
489 }
490 Err(restore_err) => {
491 format!(
492 "CRITICAL ERROR: Migration failed AND backup restoration failed!\n\n\
493 Original error: {}\n\
494 Restore error: {}\n\n\
495 MANUAL RECOVERY REQUIRED:\n\
496 1. Restore config file from: {}\n\
497 2. Restore database from: {}\n\
498 3. Review logs for details\n\
499 4. Contact support if needed",
500 e,
501 restore_err,
502 backups.source_config.display(),
503 backups.source_db.as_ref().map(|p| p.display().to_string()).unwrap_or_else(|| "N/A".to_string())
504 )
505 }
506 };
507
508 result.errors.push(format!(
509 "Migration failed at step '{}':\n{}\n\n{}",
510 step_name, e, recovery_instructions
511 ));
512
513 self.rollback(&result).await;
514 result.duration_ms = start_time.elapsed().as_millis() as u64;
515 return result;
516 }
517 }
518 }
519
520 result.success = true;
521 result.duration_ms = start_time.elapsed().as_millis() as u64;
522
523 if let Err(e) = backup_manager
524 .cleanup_backups(&backups, options.keep_backups)
525 .await
526 {
527 tracing::warn!("Failed to cleanup backups: {}", e);
528 }
529
530 result
531 }
532
533 async fn execute_step(
534 &self, step: MigrationStep, plan: &MigrationPlan, options: &MigrationOptions,
535 target_manager: &StorageManager, result: &mut MigrationResult,
536 ) -> DomainResult<()> {
537 match step {
538 MigrationStep::ValidateTarget => {
539 if !target_manager.cache_backend().is_available().await {
540 return Err(DomainError::NetworkError(
541 "Target cache backend not available".into(),
542 ));
543 }
544
545 target_manager.config_backend().list_providers().await?;
546 }
547
548 MigrationStep::MigrateTokens => {
549 tracing::info!("[Migration] Starting token migration...");
550 let password = get_vault_password(options)?;
551 tracing::info!(
552 "[Migration] Vault password retrieved: {} chars",
553 password.len()
554 );
555
556 tracing::info!("[Migration] Exporting tokens from source token store...");
557 let encrypted_blob = self.source_token_store.export_encrypted(&password).await?;
558 tracing::info!(
559 "[Migration] Exported encrypted blob: {} bytes",
560 encrypted_blob.len()
561 );
562
563 if !result.provider_id_mapping.is_empty() {
564 tracing::info!(
565 "[Migration] Using provider ID remapping path ({} mappings)",
566 result.provider_id_mapping.len()
567 );
568
569 let source_tokens = self.source_token_store.get_all_tokens().await?;
570 tracing::info!(
571 "[Migration] Retrieved {} tokens from source",
572 source_tokens.len()
573 );
574
575 let mut migrated_count = 0;
576 for (old_id, token) in source_tokens {
577 if let Some(&new_id) = result.provider_id_mapping.get(&old_id) {
578 tracing::debug!(
579 "[Migration] Storing token for provider {} → {} (token length: {})",
580 old_id,
581 new_id,
582 token.len()
583 );
584 target_manager
585 .token_store()
586 .await
587 .store_token(new_id, &token)
588 .await?;
589 migrated_count += 1;
590 tracing::info!(
591 "[Migration] Successfully stored token for provider {} (remapped to {})",
592 old_id,
593 new_id
594 );
595 } else {
596 let source_providers =
597 self.source_config_backend.list_providers().await?;
598 return Err(DomainError::DataConsistency(format!(
599 "Cannot migrate token for provider ID {} - provider not found in remapping.\n\n\
600 This indicates data corruption or inconsistency.\n\n\
601 Source provider IDs: {:?}\n\
602 Remapping table: {:?}\n\n\
603 This token would be lost if migration continued.\n\
604 Please investigate why this provider's configuration wasn't migrated.",
605 old_id,
606 source_providers.iter().filter_map(|p| p.id).collect::<Vec<_>>(),
607 result.provider_id_mapping
608 )));
609 }
610 }
611 tracing::info!(
612 "[Migration] Migrated {} tokens via ID remapping",
613 migrated_count
614 );
615 } else {
616 tracing::info!("[Migration] Using direct import path (no ID remapping)");
617 tracing::info!(
618 "[Migration] Importing encrypted blob ({} bytes) to target token store...",
619 encrypted_blob.len()
620 );
621 target_manager
622 .token_store()
623 .await
624 .import_encrypted(&encrypted_blob, &password)
625 .await?;
626 tracing::info!("[Migration] Direct import completed");
627 }
628
629 let count: usize = target_manager
630 .token_store()
631 .await
632 .get_all_tokens()
633 .await?
634 .len();
635 tracing::info!(
636 "[Migration] Verification: {} tokens now in target database",
637 count
638 );
639 result.stats.tokens_migrated = count;
640 }
641
642 MigrationStep::MigrateConfigs => {
643 tracing::info!(
644 "[Migration] Exporting providers from source database: {:?}",
645 self.source_config.db_path()
646 );
647 let export = self.source_config_backend.export_all().await?;
648 result.stats.providers_migrated = export.providers.len();
649 result.stats.permissions_migrated = export.permissions.len();
650
651 tracing::info!(
652 "[Migration] Importing {} providers to target database: {:?}",
653 export.providers.len(),
654 plan.to.db_path()
655 );
656 target_manager.config_backend().import_all(&export).await?;
657
658 let source_providers = self.source_config_backend.list_providers().await?;
659 let target_providers: Vec<ProviderConfig> =
660 target_manager.config_backend().list_providers().await?;
661
662 for source_provider in &source_providers {
663 if let Some(source_id) = source_provider.id {
664 if let Some(target_provider) = target_providers
665 .iter()
666 .find(|p| p.name == source_provider.name)
667 {
668 if let Some(target_id) = target_provider.id {
669 result.provider_id_mapping.insert(source_id, target_id);
670 }
671 }
672 }
673 }
674
675 if !same_database(&plan.from, &plan.to) {
676 let mut metrics_config = self.source_metrics.export_config().await?;
677 metrics_config.pipeline_configs.retain_mut(|config| {
678 match remap_pipeline_id(&config.pipeline_id, &result.provider_id_mapping) {
679 Some(pipeline_id) => {
680 config.pipeline_id = pipeline_id;
681 true
682 }
683 None => false,
684 }
685 });
686 MetricsRepository::new_from_pool(target_manager.database_pool())
687 .import_config(&metrics_config)
688 .await?;
689 }
690
691 if options.clean_target {
692 let source_names: std::collections::HashSet<String> =
693 source_providers.iter().map(|p| p.name.clone()).collect();
694
695 let orphaned: Vec<_> = target_providers
696 .iter()
697 .filter(|p| !source_names.contains(&p.name))
698 .collect();
699
700 for orphan in &orphaned {
701 if let Some(id) = orphan.id {
702 tracing::info!(
703 "[Migration] Removing orphaned provider: {} (id={})",
704 orphan.name,
705 id
706 );
707 target_manager.config_backend().delete_provider(id).await?;
708 let _ = target_manager.token_store().await.delete_token(id).await;
709 result.stats.tokens_cleaned += 1;
710 }
711 }
712
713 if !orphaned.is_empty() {
714 result.stats.providers_cleaned = orphaned.len();
715 tracing::info!(
716 "[Migration] Cleaned {} orphaned providers from target",
717 orphaned.len()
718 );
719 }
720 }
721 }
722
723 MigrationStep::MigrateCache => {
724 let objects = self.source_cache_backend.list(None).await?;
725 tracing::info!(
726 "[Migration] Migrating {} cache entries from {:?} to {:?}",
727 objects.len(),
728 self.source_config.data_dir().join("cache"),
729 plan.to.data_dir().join("cache")
730 );
731 for obj in objects {
732 let data = self.source_cache_backend.get(&obj.key).await?;
733 target_manager
734 .cache_backend()
735 .put(&obj.key, &data, obj.content_type.as_deref())
736 .await?;
737 result.stats.cache_entries_migrated += 1;
738 }
739
740 if !same_database(&plan.from, &plan.to) {
741 let target_metrics =
742 MetricsRepository::new_from_pool(target_manager.database_pool());
743 let pipeline_query = |pipeline_id: &str| MetricsQuery {
744 pipeline_id: Some(pipeline_id.to_string()),
745 limit: None,
746 ..Default::default()
747 };
748 let mut pipelines = 0;
749 for source_id in self.source_metrics.list_pipeline_ids().await? {
750 let Some(target_id) =
751 remap_pipeline_id(&source_id, &result.provider_id_mapping)
752 else {
753 continue;
754 };
755 let last_run = self
756 .source_metrics
757 .get_last_processed_run(&source_id)
758 .await?;
759 let mut runs: BTreeMap<i64, RecordMetrics> = BTreeMap::new();
760 for entry in self
761 .source_metrics
762 .query_metrics(pipeline_query(&source_id))
763 .await?
764 {
765 let slot = entry.metric_type as usize;
766 let record =
767 runs.entry(entry.run_number)
768 .or_insert_with(|| RecordMetrics {
769 run_number: entry.run_number,
770 run_hash: entry.run_hash.clone().unwrap_or_default(),
771 entries: [None, None, None],
772 });
773 record.entries[slot] = Some(entry);
774 }
775 let records: Vec<_> = runs.into_values().collect();
776 target_metrics
777 .refresh_record_metrics(&target_id, &records)
778 .await?;
779 result.stats.metrics_migrated += records
780 .iter()
781 .map(|record| record.entries.iter().flatten().count())
782 .sum::<usize>();
783 pipelines += 1;
784
785 let Some(run) = last_run else {
786 continue;
787 };
788 if target_metrics
789 .get_last_processed_run(&target_id)
790 .await?
791 .is_none_or(|target_run| target_run < run)
792 {
793 target_metrics
794 .update_last_processed_run(&target_id, run)
795 .await?;
796 }
797 }
798 tracing::info!(
799 "[Migration] Migrated {} metric entries for {} pipelines",
800 result.stats.metrics_migrated,
801 pipelines
802 );
803 }
804 }
805
806 MigrationStep::VerifyMigration => {
807 let source_providers = self.source_config_backend.list_providers().await?;
808
809 if plan.migrate_configs && !source_providers.is_empty() {
810 let target_providers: Vec<ProviderConfig> =
811 target_manager.config_backend().list_providers().await?;
812
813 for source_provider in &source_providers {
814 let found = target_providers
815 .iter()
816 .any(|t| t.name == source_provider.name);
817 if !found {
818 return Err(DomainError::DatabaseError(format!(
819 "Provider '{}' from source was not found in target after migration",
820 source_provider.name
821 )));
822 }
823 }
824
825 if options.clean_target {
826 if source_providers.len() != target_providers.len() {
827 return Err(DomainError::DatabaseError(format!(
828 "Provider count mismatch after cleanup: source={}, target={}",
829 source_providers.len(),
830 target_providers.len()
831 )));
832 }
833 tracing::info!(
834 "[Migration] Verified exact match: {} providers",
835 source_providers.len()
836 );
837 } else if target_providers.len() > source_providers.len() {
838 tracing::info!(
839 "[Migration] Merge complete: {} migrated, {} pre-existing kept (total: {})",
840 source_providers.len(),
841 target_providers.len() - source_providers.len(),
842 target_providers.len()
843 );
844 } else {
845 tracing::info!(
846 "[Migration] Verified {} providers migrated successfully",
847 source_providers.len()
848 );
849 }
850 } else if plan.migrate_configs {
851 tracing::info!("Skipping provider verification: source database is empty");
852 }
853
854 if plan.migrate_tokens {
855 let source_tokens = self.source_token_store.get_all_tokens().await?;
856
857 if !source_tokens.is_empty() {
858 let target_tokens: HashMap<i64, String> =
859 target_manager.token_store().await.get_all_tokens().await?;
860
861 if options.clean_target {
862 if source_tokens.len() != target_tokens.len() {
863 return Err(DomainError::DatabaseError(format!(
864 "Token count mismatch after cleanup: source={}, target={}",
865 source_tokens.len(),
866 target_tokens.len()
867 )));
868 }
869 tracing::info!(
870 "[Migration] Verified exact match: {} tokens",
871 source_tokens.len()
872 );
873 } else {
874 if target_tokens.len() < source_tokens.len() {
875 return Err(DomainError::DatabaseError(format!(
876 "Token count too low after migration: source={}, target={}",
877 source_tokens.len(),
878 target_tokens.len()
879 )));
880 }
881 tracing::info!(
882 "[Migration] Token migration complete: {} migrated (target has {} total)",
883 source_tokens.len(),
884 target_tokens.len()
885 );
886 }
887 } else {
888 tracing::info!("Skipping token verification: source has no tokens");
889 }
890 }
891 }
892
893 MigrationStep::UpdateConfig => {
894 use crate::infrastructure::config::ConfigLoader;
895
896 let config_path = ConfigLoader::discover_config_path();
897
898 tracing::info!(
899 "Saving migrated config to: {} (data_dir: {:?}, backend: {})",
900 config_path.display(),
901 plan.to.data_dir(),
902 plan.to.storage.backend
903 );
904
905 ConfigLoader::save(&plan.to, &config_path).map_err(|e| {
906 DomainError::DatabaseError(format!("Failed to save config: {}", e))
907 })?;
908 }
909 }
910
911 Ok(())
912 }
913
914 async fn emit_progress(&self, step: MigrationStep, step_index: usize, total_steps: usize) {
915 if let Some(ref event_bus) = self.event_bus {
916 event_bus
917 .emit(crate::event::CoreEvent::MigrationProgress {
918 step: format!("{:?}", step),
919 step_index,
920 total_steps,
921 message: step.description().to_string(),
922 })
923 .await;
924 }
925 }
926
927 async fn rollback(&self, _result: &MigrationResult) {
928 tracing::warn!("Migration failed. Manual intervention may be required.");
929 }
930}
931
932#[cfg(test)]
933mod tests {
934 use tempfile::TempDir;
935
936 use super::*;
937 use crate::{
938 infrastructure::token_store::MemoryTokenStore,
939 test_env::EnvGuard,
940 };
941
942 async fn create_test_manager(temp_dir: &TempDir) -> StorageManager {
943 let mut config = PipedashConfig::default();
944 config.storage.data_dir = temp_dir.path().to_string_lossy().to_string();
945
946 let token_store: Arc<dyn TokenStore> = Arc::new(MemoryTokenStore::new());
947 StorageManager::with_token_store(config, token_store, true)
948 .await
949 .unwrap()
950 }
951
952 #[tokio::test]
953 async fn test_plan_migration_no_changes() {
954 let temp_dir = TempDir::new().unwrap();
955 let manager = create_test_manager(&temp_dir).await;
956
957 let orchestrator = MigrationOrchestrator::from_manager(&manager, None, None).await;
958
959 let target_config = manager.config().clone();
960 let plan = orchestrator
961 .plan_migration(target_config, &MigrationOptions::default())
962 .unwrap();
963
964 assert!(!plan.steps.contains(&MigrationStep::MigrateConfigs));
965 assert!(!plan.steps.contains(&MigrationStep::MigrateTokens));
966 }
967
968 #[tokio::test]
969 async fn test_migration_dry_run() {
970 let temp_dir = TempDir::new().unwrap();
971 let manager = create_test_manager(&temp_dir).await;
972
973 let orchestrator = MigrationOrchestrator::from_manager(&manager, None, None).await;
974
975 let target_config = manager.config().clone();
976 let plan = MigrationPlan {
977 from: manager.config().clone(),
978 to: target_config,
979 steps: vec![MigrationStep::ValidateTarget],
980 migrate_tokens: false,
981 migrate_configs: false,
982 migrate_cache: false,
983 backend_changed: false,
984 data_dir_changed: false,
985 created_at: Utc::now(),
986 };
987
988 let result = orchestrator
989 .execute_migration(
990 plan,
991 MigrationOptions {
992 dry_run: true,
993 ..Default::default()
994 },
995 true, )
997 .await;
998
999 assert!(result.success);
1000 }
1001
1002 #[test]
1003 fn test_migration_step_descriptions() {
1004 assert!(!MigrationStep::ValidateTarget.description().is_empty());
1005 assert!(!MigrationStep::MigrateTokens.description().is_empty());
1006 assert!(!MigrationStep::MigrateConfigs.description().is_empty());
1007 }
1008
1009 #[test]
1010 fn test_migration_result_summary() {
1011 let result = MigrationResult {
1012 success: true,
1013 steps_completed: vec![MigrationStep::MigrateConfigs],
1014 errors: vec![],
1015 duration_ms: 100,
1016 stats: MigrationStats {
1017 providers_migrated: 5,
1018 tokens_migrated: 3,
1019 cache_entries_migrated: 10,
1020 metrics_migrated: 0,
1021 permissions_migrated: 2,
1022 providers_cleaned: 0,
1023 tokens_cleaned: 0,
1024 },
1025 provider_id_mapping: std::collections::HashMap::new(),
1026 backups: None,
1027 };
1028
1029 let summary = result.summary();
1030 assert!(summary.contains("5 providers"));
1031 assert!(summary.contains("10 cache entries"));
1032 }
1033
1034 #[tokio::test]
1035 async fn migrate_tokens_after_a_provider_was_deleted() {
1036 let source_dir = TempDir::new().unwrap();
1037 let target_dir = TempDir::new().unwrap();
1038 let config_path = source_dir.path().join("config.toml");
1039 std::fs::write(&config_path, "[storage]\nbackend = \"sqlite\"\n").unwrap();
1040 let _env = EnvGuard::set(&[("PIPEDASH_CONFIG_PATH", config_path.to_str())]);
1041
1042 let mut source = PipedashConfig::default();
1043 source.storage.data_dir = source_dir.path().to_string_lossy().to_string();
1044 source.storage.vault_password = Some("test-password".to_string());
1045 let mut target = source.clone();
1046 target.storage.data_dir = target_dir.path().to_string_lossy().to_string();
1047
1048 let manager = StorageManager::from_config(source, false).await.unwrap();
1049 let backend = manager.config_backend();
1050 let tokens = manager.token_store().await;
1051 let mut ids = Vec::new();
1052 for name in ["a", "b", "c"] {
1053 let provider = ProviderConfig {
1054 id: None,
1055 name: name.to_string(),
1056 provider_type: "github".to_string(),
1057 token: String::new(),
1058 config: HashMap::new(),
1059 refresh_interval: 30,
1060 version: None,
1061 };
1062 ids.push(backend.create_provider(&provider).await.unwrap());
1063 }
1064 backend.delete_provider(ids[1]).await.unwrap();
1065 tokens.store_token(ids[0], "tok-a").await.unwrap();
1066 tokens.store_token(ids[2], "tok-c").await.unwrap();
1067
1068 let orchestrator = MigrationOrchestrator::from_manager(&manager, None, None).await;
1069 let options = MigrationOptions {
1070 migrate_tokens: true,
1071 token_password: Some("test-password".to_string()),
1072 ..Default::default()
1073 };
1074 let plan = orchestrator
1075 .plan_migration(target.clone(), &options)
1076 .unwrap();
1077 let result = orchestrator.execute_migration(plan, options, false).await;
1078 assert!(result.success, "{:?}", result.errors);
1079
1080 let target_manager = StorageManager::from_config(target, false).await.unwrap();
1081 let mut migrated = HashMap::new();
1082 for provider in target_manager
1083 .config_backend()
1084 .list_providers()
1085 .await
1086 .unwrap()
1087 {
1088 let token = target_manager
1089 .token_store()
1090 .await
1091 .get_token(provider.id.unwrap())
1092 .await
1093 .unwrap();
1094 migrated.insert(provider.name, token);
1095 }
1096 assert_eq!(
1097 migrated,
1098 HashMap::from([
1099 ("a".to_string(), "tok-a".to_string()),
1100 ("c".to_string(), "tok-c".to_string()),
1101 ])
1102 );
1103 }
1104
1105 #[tokio::test]
1106 async fn migration_keeps_pipeline_metrics() {
1107 use crate::domain::{
1108 MetricEntry,
1109 MetricType,
1110 };
1111 use crate::infrastructure::database::RecordMetrics;
1112
1113 let source_dir = TempDir::new().unwrap();
1114 let target_dir = TempDir::new().unwrap();
1115 let config_path = source_dir.path().join("config.toml");
1116 std::fs::write(&config_path, "[storage]\nbackend = \"sqlite\"\n").unwrap();
1117 let _env = EnvGuard::set(&[("PIPEDASH_CONFIG_PATH", config_path.to_str())]);
1118
1119 let mut source = PipedashConfig::default();
1120 source.storage.data_dir = source_dir.path().to_string_lossy().to_string();
1121 source.storage.vault_password = Some("test-password".to_string());
1122 let mut target = source.clone();
1123 target.storage.data_dir = target_dir.path().to_string_lossy().to_string();
1124
1125 let manager = StorageManager::from_config(source, false).await.unwrap();
1126 let backend = manager.config_backend();
1127 let mut ids = Vec::new();
1128 for name in ["deleted", "gh"] {
1129 let provider = ProviderConfig {
1130 id: None,
1131 name: name.to_string(),
1132 provider_type: "github".to_string(),
1133 token: String::new(),
1134 config: HashMap::new(),
1135 refresh_interval: 30,
1136 version: None,
1137 };
1138 ids.push(backend.create_provider(&provider).await.unwrap());
1139 }
1140 backend.delete_provider(ids[0]).await.unwrap();
1141 let orphan_pipeline = format!("github__{}__o__r__1", ids[0]);
1142 let source_pipeline = format!("github__{}__o__r__1", ids[1]);
1143 let expired_pipeline = format!("github__{}__o__r__2", ids[1]);
1144
1145 let metrics = MetricsRepository::new_from_pool(manager.database_pool());
1146 metrics.update_global_config(true, 30).await.unwrap();
1147 metrics
1148 .upsert_pipeline_config(&source_pipeline, true, 14)
1149 .await
1150 .unwrap();
1151 metrics
1152 .upsert_pipeline_config(&orphan_pipeline, false, 3)
1153 .await
1154 .unwrap();
1155 let entry = |pipeline_id: &str, n: i64| MetricEntry {
1156 id: 0,
1157 pipeline_id: pipeline_id.to_string(),
1158 run_number: n,
1159 timestamp: Utc::now(),
1160 metric_type: MetricType::RunDuration,
1161 value: 60.0 * n as f64,
1162 metadata: None,
1163 created_at: Utc::now(),
1164 run_hash: Some(format!("hash-{n}")),
1165 };
1166 let mut entries: Vec<MetricEntry> = (1..=5).map(|n| entry(&source_pipeline, n)).collect();
1167 entries.push(entry(&orphan_pipeline, 1));
1168 metrics.insert_metrics_batch(entries).await.unwrap();
1169 metrics
1170 .update_last_processed_run(&source_pipeline, 5)
1171 .await
1172 .unwrap();
1173 metrics
1174 .update_last_processed_run(&expired_pipeline, 9)
1175 .await
1176 .unwrap();
1177
1178 let orchestrator = MigrationOrchestrator::from_manager(&manager, None, None).await;
1179 let options = MigrationOptions {
1180 migrate_tokens: true,
1181 migrate_cache: true,
1182 token_password: Some("test-password".to_string()),
1183 ..Default::default()
1184 };
1185 let plan = orchestrator
1186 .plan_migration(target.clone(), &options)
1187 .unwrap();
1188 let migrate = || orchestrator.execute_migration(plan.clone(), options.clone(), false);
1189 let result = migrate().await;
1190 assert!(result.success, "{:?}", result.errors);
1191 assert_eq!(result.stats.metrics_migrated, 5);
1192 let refreshed = MetricEntry {
1193 run_hash: Some("hash-4b".to_string()),
1194 value: 999.0,
1195 ..entry(&source_pipeline, 4)
1196 };
1197 metrics
1198 .refresh_record_metrics(
1199 &source_pipeline,
1200 &[RecordMetrics {
1201 run_number: 4,
1202 run_hash: "hash-4b".to_string(),
1203 entries: [Some(refreshed), None, None],
1204 }],
1205 )
1206 .await
1207 .unwrap();
1208 let result = migrate().await;
1209 assert!(result.success, "{:?}", result.errors);
1210
1211 let target_manager = StorageManager::from_config(target, false).await.unwrap();
1212 let target_id = target_manager
1213 .config_backend()
1214 .list_providers()
1215 .await
1216 .unwrap()[0]
1217 .id
1218 .unwrap();
1219 assert_ne!(target_id, ids[1]);
1220 let target_pipeline = format!("github__{target_id}__o__r__1");
1221 let target_metrics = MetricsRepository::new_from_pool(target_manager.database_pool());
1222
1223 let copied = target_metrics
1224 .query_metrics(MetricsQuery {
1225 pipeline_id: Some(target_pipeline.clone()),
1226 ..Default::default()
1227 })
1228 .await
1229 .unwrap();
1230 assert_eq!(copied.len(), 5);
1231 let run_4: Vec<_> = copied
1232 .into_iter()
1233 .filter(|entry| entry.run_number == 4)
1234 .map(|entry| (entry.run_hash, entry.value))
1235 .collect();
1236 assert_eq!(run_4, [(Some("hash-4b".to_string()), 999.0)]);
1237 let config = target_metrics
1238 .get_pipeline_config(&target_pipeline)
1239 .await
1240 .unwrap()
1241 .unwrap();
1242 assert!(config.enabled);
1243 assert_eq!(config.retention_days, 14);
1244 let global = target_metrics.get_global_config().await.unwrap();
1245 assert!(global.enabled);
1246 assert_eq!(global.default_retention_days, 30);
1247 assert_eq!(
1248 target_metrics
1249 .get_last_processed_run(&target_pipeline)
1250 .await
1251 .unwrap(),
1252 Some(5)
1253 );
1254 assert_eq!(
1255 target_metrics
1256 .get_last_processed_run(&format!("github__{target_id}__o__r__2"))
1257 .await
1258 .unwrap(),
1259 Some(9)
1260 );
1261 assert_eq!(
1262 target_metrics
1263 .export_config()
1264 .await
1265 .unwrap()
1266 .pipeline_configs
1267 .len(),
1268 1
1269 );
1270 assert_eq!(
1271 target_metrics
1272 .calculate_storage_stats()
1273 .await
1274 .unwrap()
1275 .total_metrics_count,
1276 5
1277 );
1278 }
1279}