use std::sync::Arc; use tauri::{AppHandle, Emitter, State}; use crate::commands::connection::{ensure_connection_writable, AppState}; // Re-export types and functions used by other modules use dbx_core::models::connection::DatabaseType; pub use dbx_core::transfer::{ get_db_type, TransferOwnershipPreview, TransferProgress, TransferRequest, TransferStatus, }; fn emit_progress(app: &AppHandle, progress: TransferProgress) { let _ = app.emit("transfer-progress", progress); } #[tauri::command] pub async fn start_transfer( app: AppHandle, state: State<'_, Arc>, request: TransferRequest, ) -> Result<(), String> { let state = state.inner().clone(); let transfer_id = request.transfer_id.clone(); // Reject transfer early if the target connection is read-only — writing to it is inherently required ensure_connection_writable(&state, &request.target_connection_id, "Transfer").await?; // Validate connections exist let source_db_type = get_db_type(&state, &request.source_connection_id).await?; let target_db_type = get_db_type(&state, &request.target_connection_id).await?; dbx_core::transfer::validate_transfer_request(&request)?; // Cross-family object transfers are validated inside transfer_schema_objects: // only mechanically rewriteable kinds (views, sequences) are allowed. // Structure-only data transfer is unsupported for MongoDB. if matches!(request.content, dbx_core::transfer::TransferContent::StructureOnly) && (matches!(source_db_type, DatabaseType::MongoDb) || matches!(target_db_type, DatabaseType::MongoDb)) { return Err("MongoDB 暂不支持仅结构传输".to_string()); } // External Doris/StarRocks catalogs: pool is created with `catalog=` URL // setup (SET catalog) and without USE . See ensure_transfer_pool. let source_pool_key = dbx_core::transfer::ensure_transfer_pool( &state, &request.source_connection_id, &request.source_database, request.source_catalog.as_deref(), ) .await?; let target_pool_key = dbx_core::transfer::ensure_transfer_pool( &state, &request.target_connection_id, &request.target_database, request.target_catalog.as_deref(), ) .await?; tokio::spawn(async move { // Sort tables by FK dependency so referenced tables are transferred first, // and keep the foreign key metadata fetched along the way — MySQL-family // targets reuse it per table below instead of re-querying it. // Skip for external Doris/StarRocks catalogs — the database name does not // exist in the default catalog and sorting is unnecessary (no FK constraints). let (sorted_tables, known_foreign_keys) = { let skip_fk_sort = { let configs = state.configs.read().await; configs .get(&request.source_connection_id) .and_then(|config| { dbx_core::transfer::resolve_external_transfer_catalog_for_config( request.source_catalog.as_deref(), config, ) }) .is_some() }; if skip_fk_sort { (request.tables.clone(), std::collections::HashMap::new()) } else { dbx_core::transfer::sort_tables_by_fk_dependency_with_foreign_keys( &state, &request.source_connection_id, &request.source_database, &request.source_schema, &request.tables, true, ) .await .unwrap_or_else(|e| { log::warn!("[transfer] failed to sort tables by FK dependency, using original order: {e}"); (request.tables.clone(), std::collections::HashMap::new()) }) } }; let total_tables = sorted_tables.len(); log::info!("[transfer] starting transfer_id={} tables={}", transfer_id, total_tables); let mut failed_tables: Vec = Vec::new(); let mut pending_fk_alters: Vec<(String, String)> = Vec::new(); let mut last_rows_transferred = 0_u64; let mut last_total_rows = None; if matches!(source_db_type, dbx_core::models::connection::DatabaseType::Postgres) && matches!(target_db_type, dbx_core::models::connection::DatabaseType::Postgres) { match dbx_core::transfer::transfer_postgres_schema_dependencies( &state, &request, &source_pool_key, &target_pool_key, |progress| { last_rows_transferred = progress.rows_transferred; last_total_rows = progress.total_rows; emit_progress(&app, progress); }, ) .await { Ok(()) => {} Err(e) if e == "Cancelled" => { emit_progress( &app, TransferProgress { transfer_id: transfer_id.clone(), table: "schema dependencies".to_string(), table_index: 0, total_tables, rows_transferred: last_rows_transferred, total_rows: last_total_rows, status: TransferStatus::Cancelled, error: None, terminal: true, }, ); dbx_core::transfer::clear_cancelled(&transfer_id).await; return; } Err(e) => { emit_progress( &app, TransferProgress { transfer_id: transfer_id.clone(), table: "schema dependencies".to_string(), table_index: 0, total_tables, rows_transferred: last_rows_transferred, total_rows: last_total_rows, status: TransferStatus::Error, error: Some(e), terminal: true, }, ); dbx_core::transfer::clear_cancelled(&transfer_id).await; return; } } } for (i, table) in sorted_tables.iter().enumerate() { if dbx_core::transfer::is_cancelled(&transfer_id).await { emit_progress( &app, TransferProgress { transfer_id: transfer_id.clone(), table: table.clone(), table_index: i, total_tables, rows_transferred: last_rows_transferred, total_rows: last_total_rows, status: TransferStatus::Cancelled, error: None, terminal: true, }, ); dbx_core::transfer::clear_cancelled(&transfer_id).await; return; } log::info!("[transfer] table {}/{}: {}", i + 1, total_tables, table); match dbx_core::transfer::transfer_table( &state, &request, table, i, &source_db_type, &target_db_type, &source_pool_key, &target_pool_key, &known_foreign_keys, &mut pending_fk_alters, |progress| { last_rows_transferred = progress.rows_transferred; last_total_rows = progress.total_rows; emit_progress(&app, progress); }, ) .await { Ok(rows) => { emit_progress( &app, TransferProgress { transfer_id: transfer_id.clone(), table: table.clone(), table_index: i, total_tables, rows_transferred: rows, total_rows: last_total_rows.or(Some(rows)), status: TransferStatus::TableDone, error: None, terminal: false, }, ); } Err(e) => { if e == "Cancelled" { emit_progress( &app, TransferProgress { transfer_id: transfer_id.clone(), table: table.clone(), table_index: i, total_tables, rows_transferred: last_rows_transferred, total_rows: last_total_rows, status: TransferStatus::Cancelled, error: None, terminal: true, }, ); dbx_core::transfer::clear_cancelled(&transfer_id).await; return; } failed_tables.push(table.clone()); emit_progress( &app, TransferProgress { transfer_id: transfer_id.clone(), table: table.clone(), table_index: i, total_tables, rows_transferred: last_rows_transferred, total_rows: last_total_rows, status: TransferStatus::Error, error: Some(e), terminal: false, }, ); } } } // Add any foreign keys deferred during MySQL-family table creation now that // every selected table exists — see transfer_table's use of // strip_inline_foreign_key_constraint_lines for why these can't be created // inline (a foreign key cycle has no valid CREATE TABLE order at all). let mut failed_fk_tables: Vec = Vec::new(); let mut failed_fk_count = 0usize; for (table, alter_sql) in &pending_fk_alters { if let Err(e) = dbx_core::transfer::execute_on_pool(&state, &target_pool_key, alter_sql).await { log::warn!("[transfer] failed to add deferred foreign key constraint for {table}: {e}"); failed_fk_count += 1; failed_fk_tables.push(table.clone()); } } if failed_fk_count > 0 { failed_fk_tables.sort(); failed_fk_tables.dedup(); failed_tables.push(format!("{} foreign key(s) on: {}", failed_fk_count, failed_fk_tables.join(", "))); } // Transfer selected non-table objects (views, procedures, functions, // triggers, sequences, events) after the per-table loop. The shared // Core decision handles all content modes: DataOnly never // transfers schema objects; PG→PG keeps the legacy empty-selection // default only when structure participates in the transfer. let mut object_outcome = dbx_core::transfer::TransferObjectOutcome::default(); match dbx_core::transfer::transfer_schema_objects( &state, &request, &source_pool_key, &target_pool_key, |progress| emit_progress(&app, progress), ) .await { Ok(outcome) => { object_outcome = outcome; } Err(e) if e == "Cancelled" => { emit_progress( &app, TransferProgress { transfer_id: transfer_id.clone(), table: "schema objects".to_string(), table_index: total_tables, total_tables, rows_transferred: 0, total_rows: None, status: TransferStatus::Cancelled, error: None, terminal: true, }, ); dbx_core::transfer::clear_cancelled(&transfer_id).await; return; } Err(e) => { failed_tables.push("schema objects".to_string()); emit_progress( &app, TransferProgress { transfer_id: transfer_id.clone(), table: "schema objects".to_string(), table_index: total_tables, total_tables, rows_transferred: 0, total_rows: None, status: TransferStatus::Error, error: Some(e), terminal: false, }, ); } } if !object_outcome.failed.is_empty() { failed_tables.push(format!("schema objects ({})", object_outcome.failed.len())); } let skip_suffix = if !object_outcome.skipped.is_empty() && failed_tables.is_empty() { format!(",跳过 {} 个已存在对象", object_outcome.skipped.len()) } else if !object_outcome.skipped.is_empty() { format!(";跳过 {} 个已存在对象", object_outcome.skipped.len()) } else { String::new() }; emit_progress( &app, TransferProgress { transfer_id: transfer_id.clone(), table: String::new(), table_index: total_tables, total_tables, rows_transferred: last_rows_transferred, total_rows: last_total_rows, status: if failed_tables.is_empty() { TransferStatus::Done } else { TransferStatus::Error }, error: if failed_tables.is_empty() { if skip_suffix.is_empty() { None } else { Some(skip_suffix.clone()) } } else { Some(format!( "{} table(s) failed: {}{}", failed_tables.len(), failed_tables.iter().take(5).cloned().collect::>().join(", "), skip_suffix )) }, terminal: true, }, ); dbx_core::transfer::clear_cancelled(&transfer_id).await; }); Ok(()) } #[tauri::command] pub async fn preview_transfer_ownership( state: State<'_, Arc>, request: TransferRequest, ) -> Result { let state = state.inner().clone(); let source_db_type = get_db_type(&state, &request.source_connection_id).await?; let target_db_type = get_db_type(&state, &request.target_connection_id).await?; dbx_core::transfer::validate_transfer_request(&request)?; let source_pool_key = dbx_core::transfer::ensure_transfer_pool( &state, &request.source_connection_id, &request.source_database, request.source_catalog.as_deref(), ) .await?; let target_pool_key = dbx_core::transfer::ensure_transfer_pool( &state, &request.target_connection_id, &request.target_database, request.target_catalog.as_deref(), ) .await?; dbx_core::transfer::preview_transfer_ownership( &state, &request, &source_db_type, &target_db_type, &source_pool_key, &target_pool_key, ) .await } #[tauri::command] pub async fn cancel_transfer(transfer_id: String) -> Result<(), String> { dbx_core::transfer::set_cancelled(&transfer_id).await; Ok(()) } /// Sort table names by foreign key dependency. /// `parents_first: true` → parent tables first (insert/export order). /// `parents_first: false` → child tables first (drop order). #[allow(dead_code)] #[tauri::command] pub async fn sort_tables_by_fk_dependency( state: tauri::State<'_, std::sync::Arc>, connection_id: String, database: String, schema: String, tables: Vec, parents_first: bool, ) -> Result, String> { dbx_core::transfer::sort_tables_by_fk_dependency(&state, &connection_id, &database, &schema, &tables, parents_first) .await }