From 8b30a42b65d8cbe2e28bc0383852afa0ba55641e Mon Sep 17 00:00:00 2001 From: jearhe <0verme7753@gmail.com> Date: Fri, 9 Oct 2026 09:44:41 +0800 Subject: [PATCH] feat(transfer): persist task run history --- apps/desktop/src/lib/backend/api.ts | 3 + apps/desktop/src/lib/backend/http.ts | 30 + apps/desktop/src/lib/backend/tauri.ts | 109 ++ crates/dbx-core/src/data/transfer.rs | 75 +- .../dbx-core/src/data/transfer/iris_tests.rs | 7 +- crates/dbx-core/src/persistence/mod.rs | 1 + crates/dbx-core/src/persistence/storage.rs | 663 +++++++++++ .../dbx-core/src/persistence/task_history.rs | 1018 +++++++++++++++++ crates/dbx-web/src/error.rs | 4 + crates/dbx-web/src/main.rs | 3 + crates/dbx-web/src/routes/mod.rs | 1 + crates/dbx-web/src/routes/task_history.rs | 75 ++ crates/dbx-web/src/routes/transfer.rs | 137 ++- crates/dbx-web/src/sse.rs | 22 + src-tauri/src/commands/mod.rs | 1 + src-tauri/src/commands/task_history.rs | 27 + src-tauri/src/commands/transfer.rs | 134 ++- src-tauri/src/lib.rs | 3 + 18 files changed, 2274 insertions(+), 39 deletions(-) create mode 100644 crates/dbx-core/src/persistence/task_history.rs create mode 100644 crates/dbx-web/src/routes/task_history.rs create mode 100644 src-tauri/src/commands/task_history.rs diff --git a/apps/desktop/src/lib/backend/api.ts b/apps/desktop/src/lib/backend/api.ts index f5e5c79be7..eb6428f451 100644 --- a/apps/desktop/src/lib/backend/api.ts +++ b/apps/desktop/src/lib/backend/api.ts @@ -931,6 +931,9 @@ export const saveHistory = forward("saveHistory"); export const loadHistory = forward("loadHistory"); export const searchHistory = forward("searchHistory"); export const loadHistoryConnectionOptions = forward("loadHistoryConnectionOptions"); +export const loadTaskRuns = forward("loadTaskRuns"); +export const loadTaskRun = forward("loadTaskRun"); +export const loadTaskRunItems = forward("loadTaskRunItems"); export const loadRedisHistory = forward("loadRedisHistory"); export const clearHistory = forward("clearHistory"); export const clearHistoryBySource = forward("clearHistoryBySource"); diff --git a/apps/desktop/src/lib/backend/http.ts b/apps/desktop/src/lib/backend/http.ts index be40d9a9ab..c9fd4bc12d 100644 --- a/apps/desktop/src/lib/backend/http.ts +++ b/apps/desktop/src/lib/backend/http.ts @@ -151,6 +151,11 @@ import type { HistorySearchRequest, HistorySearchResult, HistoryConnectionOption, + TaskRunDetail, + TaskRunItemsPage, + TaskRunItemsQuery, + TaskRunListQuery, + TaskRunPage, SqlFileRequest, SqlFilePreview, SqlFileTable, @@ -5300,6 +5305,31 @@ export async function loadHistoryConnectionOptions(): Promise { + return get( + `/api/task-runs?${qs({ + limit: query.limit, + cursorCreatedAt: query.cursor?.createdAt, + cursorRunId: query.cursor?.runId, + taskType: query.taskType, + status: query.status, + })}`, + ); +} + +export async function loadTaskRun(runId: string): Promise { + return get(`/api/task-runs/${encodeURIComponent(runId)}`); +} + +export async function loadTaskRunItems(runId: string, query: TaskRunItemsQuery = {}): Promise { + return get( + `/api/task-runs/${encodeURIComponent(runId)}/items?${qs({ + limit: query.limit, + afterItemIndex: query.afterItemIndex, + })}`, + ); +} + export async function loadRedisHistory(limit = 100, offset = 0): Promise { return loadHistory(limit, offset, "redis_command"); } diff --git a/apps/desktop/src/lib/backend/tauri.ts b/apps/desktop/src/lib/backend/tauri.ts index ecb3f5f4be..bf5861cbb8 100644 --- a/apps/desktop/src/lib/backend/tauri.ts +++ b/apps/desktop/src/lib/backend/tauri.ts @@ -5364,6 +5364,115 @@ export interface HistoryConnectionOption extends HistoryConnectionFilter { databases: string[]; } +export type TaskType = "transfer"; +export type TaskLifecycleOwner = "tauri" | "web"; +export type TaskRunStatus = "running" | "succeeded" | "partial_failed" | "failed" | "cancelled"; +export type TaskItemKind = "table" | "view" | "materialized_view" | "procedure" | "function" | "trigger" | "sequence" | "event" | "object"; +export type TaskItemStatus = "pending" | "running" | "succeeded" | "skipped" | "failed" | "cancelled" | "not_started" | "incomplete"; +export type TaskRowCountState = "not_applicable" | "known" | "unknown" | "incomplete"; +export type TransferRunContent = "structure_and_data" | "structure_only" | "data_only"; +export type TransferRunMode = "append" | "overwrite" | "upsert"; +export type TransferRunObjectSelectionMode = "unspecified" | "explicit"; + +export interface TaskEndpointSnapshot { + connectionId: string; + databaseType: string; + database: string; + schema: string; + catalog?: string | null; +} + +export interface TaskRun { + runId: string; + taskType: TaskType; + lifecycleOwner: TaskLifecycleOwner; + status: TaskRunStatus; + createdAt: string; + startedAt: string; + finishedAt?: string | null; + ownerInstanceId: string; + errorCode?: string | null; + safeErrorSummary?: string | null; + historyComplete: boolean; + source: TaskEndpointSnapshot; + target: TaskEndpointSnapshot; +} + +export interface TransferRunDetails { + runId: string; + content: TransferRunContent; + mode: TransferRunMode; + batchSize: number; + createTable: boolean; + dropTargetBeforeCreate: boolean; + targetTableNameCase: "preserve" | "lower" | "upper"; + quoteTargetColumnNames: boolean; + ownershipPolicy: "preserve" | "skip" | "reassign_missing"; + filteredTableCount: number; + tableTotal: number; + objectSelectionMode: TransferRunObjectSelectionMode; + selectedObjectCount?: number | null; +} + +export interface TaskRunDetail { + run: TaskRun; + transfer?: TransferRunDetails | null; +} + +export interface TaskRunCursor { + createdAt: string; + runId: string; +} + +export interface TaskRunListQuery { + limit?: number; + cursor?: TaskRunCursor | null; + taskType?: TaskType; + status?: TaskRunStatus; +} + +export interface TaskRunPage { + items: TaskRun[]; + nextCursor?: TaskRunCursor | null; +} + +export interface TaskRunItem { + runId: string; + itemIndex: number; + itemKind: TaskItemKind; + sourceObject: string; + targetObject: string; + status: TaskItemStatus; + sourceRowCount?: number | null; + movedRowCount?: number | null; + targetRowCount?: number | null; + rowCountState: TaskRowCountState; + hasTableFilter: boolean; + safeErrorSummary?: string | null; +} + +export interface TaskRunItemsQuery { + limit?: number; + afterItemIndex?: number; +} + +export interface TaskRunItemsPage { + items: TaskRunItem[]; + nextAfterItemIndex?: number | null; +} + +export async function loadTaskRuns(query: TaskRunListQuery = {}): Promise { + return invoke("load_task_runs", { query }); +} + +export async function loadTaskRun(runId: string): Promise { + return invoke("load_task_run", { runId }); +} + +export async function loadTaskRunItems(runId: string, query: TaskRunItemsQuery = {}): Promise { + return invoke("load_task_run_items", { runId, query }); +} + export async function saveHistory(entry: HistoryEntry): Promise { return invoke("save_history", { entry }); } diff --git a/crates/dbx-core/src/data/transfer.rs b/crates/dbx-core/src/data/transfer.rs index b45c6205a3..7c68d060d6 100644 --- a/crates/dbx-core/src/data/transfer.rs +++ b/crates/dbx-core/src/data/transfer.rs @@ -485,6 +485,13 @@ impl TransferRequest { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct TransferTableResult { + pub moved_rows: u64, + /// Source COUNT result when the executor obtained it. It is not a target row count. + pub source_row_count: Option, +} + #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] pub struct TransferProgress { @@ -10516,15 +10523,16 @@ async fn create_transfer_target_table( Ok(()) } -/// Transfer a single table. Returns rows transferred. -/// `progress_callback` is invoked for progress updates. +/// Transfer a single table, returning moved rows and the source COUNT when available. +/// `progress_callback` is invoked for progress updates; source-count observation is separate +/// so the existing progress event payload remains unchanged. /// /// `preexisting_backup_names` carries the output of [`rename_tables_to_backup`] and is /// required whenever `drop_target_before_create` is set — this pass only checks whether the /// table was renamed aside, and never renames or drops anything itself. Removing the backups /// is [`drop_backup_tables`], after every table has succeeded. #[allow(clippy::too_many_arguments)] -async fn transfer_table_inner( +async fn transfer_table_inner( state: &Arc, request: &TransferRequest, table: &str, @@ -10537,14 +10545,17 @@ async fn transfer_table_inner( pending_fk_alters: &mut Vec<(String, String)>, preexisting_backup_names: Option<&HashMap>, mut progress_callback: F, -) -> Result + mut source_count_callback: C, +) -> Result where F: FnMut(TransferProgress), + C: FnMut(Option), { if *target_db_type == DatabaseType::Db2 { db2::validate_request(request)?; } if is_mongodb_transfer_type(source_db_type) || is_mongodb_transfer_type(target_db_type) { + source_count_callback(None); return transfer_mongodb_table( state, request, @@ -10556,7 +10567,8 @@ where target_pool_key, progress_callback, ) - .await; + .await + .map(|moved_rows| TransferTableResult { moved_rows, source_row_count: None }); } let table_filter = transfer_table_filter_for(request, table)?; @@ -10762,6 +10774,7 @@ where } else { None }; + source_count_callback(total_rows); log::info!("[transfer] {} total_rows={:?}", table, total_rows); let server_side_complex_copy = @@ -10847,7 +10860,7 @@ where ) .await?; } - return Ok(0); + return Ok(TransferTableResult { moved_rows: 0, source_row_count: None }); } // A preexisting target also needs its columns read, even for a data-only @@ -11032,7 +11045,7 @@ where error: None, terminal: false, }); - return Ok(copied); + return Ok(TransferTableResult { moved_rows: copied, source_row_count: total_rows }); } // COPY fast path: PG-family append/overwrite transfers stream the whole @@ -11394,7 +11407,7 @@ where .await?; } - Ok(total_transferred) + Ok(TransferTableResult { moved_rows: total_transferred, source_row_count: total_rows }) } /// Free the constraint names the backups are still holding (MySQL family). @@ -11705,10 +11718,49 @@ pub async fn transfer_table( known_foreign_keys: &HashMap>, pending_fk_alters: &mut Vec<(String, String)>, preexisting_backup_names: Option<&HashMap>, - mut progress_callback: F, + progress_callback: F, ) -> Result where F: FnMut(TransferProgress), +{ + transfer_table_with_result( + state, + request, + table, + table_index, + source_db_type, + target_db_type, + source_pool_key, + target_pool_key, + known_foreign_keys, + pending_fk_alters, + preexisting_backup_names, + progress_callback, + |_| {}, + ) + .await + .map(|result| result.moved_rows) +} + +#[allow(clippy::too_many_arguments)] +pub async fn transfer_table_with_result( + state: &Arc, + request: &TransferRequest, + table: &str, + table_index: usize, + source_db_type: &DatabaseType, + target_db_type: &DatabaseType, + source_pool_key: &str, + target_pool_key: &str, + known_foreign_keys: &HashMap>, + pending_fk_alters: &mut Vec<(String, String)>, + preexisting_backup_names: Option<&HashMap>, + mut progress_callback: F, + mut source_count_callback: C, +) -> Result +where + F: FnMut(TransferProgress), + C: FnMut(Option), { let state = state.clone(); let request = request.clone(); @@ -11731,6 +11783,7 @@ where let request_target_schema = request.target_schema.clone(); let request_target_catalog = request.target_catalog.clone(); let (progress_tx, mut progress_rx) = tokio::sync::mpsc::channel(TRANSFER_PROGRESS_CHANNEL_CAPACITY); + let (source_count_tx, mut source_count_rx) = tokio::sync::mpsc::channel(1); let mut task = tokio::spawn(async move { let mut task_pending_fk_alters = Vec::new(); @@ -11749,6 +11802,9 @@ where move |progress| { try_send_transfer_progress(&progress_tx, progress); }, + move |source_count| { + let _ = source_count_tx.try_send(source_count); + }, ) .await; (result, task_pending_fk_alters) @@ -11759,6 +11815,7 @@ where tokio::select! { biased; Some(progress) = progress_rx.recv() => progress_callback(progress), + Some(source_count) = source_count_rx.recv() => source_count_callback(source_count), result = &mut task => { let (result, task_pending_fk_alters) = result.map_err(|error| format!("Transfer table task failed: {error}"))?; diff --git a/crates/dbx-core/src/data/transfer/iris_tests.rs b/crates/dbx-core/src/data/transfer/iris_tests.rs index 08117748de..5cb8217ae2 100644 --- a/crates/dbx-core/src/data/transfer/iris_tests.rs +++ b/crates/dbx-core/src/data/transfer/iris_tests.rs @@ -310,6 +310,7 @@ async fn iris_transfer_agent_preserves_errors_count_fallback_and_cancellation() for failure in ["source", "target", "count", "cancel"] { let (directory, state, request, source, target) = iris_transfer_fixture(None, 5, failure, true).await; let mut totals = Vec::new(); + let mut observed_source_count = None; let result = Box::pin(transfer_table_inner( &state, &request, @@ -328,12 +329,16 @@ async fn iris_transfer_agent_preserves_errors_count_fallback_and_cancellation() CANCELLED.try_write().unwrap().insert(request.transfer_id.clone()); } }, + |source_count| observed_source_count = source_count, )) .await; clear_cancelled(&request.transfer_id).await; state.shutdown(std::time::Duration::from_secs(1)).await; if failure == "count" { - assert_eq!(result.unwrap(), 5); + let result = result.unwrap(); + assert_eq!(result.moved_rows, 5); + assert_eq!(result.source_row_count, None); + assert_eq!(observed_source_count, None); assert!(totals.iter().all(Option::is_none)); } else { let error = result.unwrap_err(); diff --git a/crates/dbx-core/src/persistence/mod.rs b/crates/dbx-core/src/persistence/mod.rs index 42ff8df5a8..d8cc71d266 100644 --- a/crates/dbx-core/src/persistence/mod.rs +++ b/crates/dbx-core/src/persistence/mod.rs @@ -5,5 +5,6 @@ pub mod saved_sql; pub mod secret_codec; pub mod state_persistence; pub mod storage; +pub mod task_history; #[cfg(any(test, feature = "test-support"))] pub mod test_storage; diff --git a/crates/dbx-core/src/persistence/storage.rs b/crates/dbx-core/src/persistence/storage.rs index ceb08a4e33..e838508907 100644 --- a/crates/dbx-core/src/persistence/storage.rs +++ b/crates/dbx-core/src/persistence/storage.rs @@ -37,6 +37,12 @@ use crate::models::connection::{ConnectionConfig, DatabaseConnectionInfo, Databa use crate::persistence::secret_codec::{ key_file_candidates, SecretCodec, SecretKeyPolicy, SecretKeyResolution, SecretKeySource, }; +use crate::persistence::task_history::{ + RowCountState, TaskEndpointSnapshot, TaskHistoryStorageError, TaskItemKind, TaskItemStatus, TaskLifecycleOwner, + TaskRun, TaskRunCursor, TaskRunDetail, TaskRunItem, TaskRunItemsPage, TaskRunItemsQuery, TaskRunListQuery, + TaskRunPage, TaskRunStatus, TaskType, TransferObjectSelectionMode, TransferRunContent, TransferRunDetails, + TransferRunMode, TransferRunOwnershipPolicy, TransferRunTargetTableNameCase, +}; use crate::prompt_template::PromptTemplate; use crate::saved_sql::{SavedSqlFile, SavedSqlFolder, SavedSqlLibrary}; @@ -83,6 +89,9 @@ const USER_DATA_TABLES: &[&str] = &[ "connections", "connection_secrets", "history", + "task_runs", + "transfer_run_details", + "task_run_items", "ai_config", "ai_provider_configs", "ai_conversations", @@ -1044,6 +1053,62 @@ const SCHEMA_STATEMENTS: &[&str] = &[ ,mcp_response_json TEXT ,mcp_session_id TEXT )", + "CREATE TABLE IF NOT EXISTS task_runs ( + run_id TEXT PRIMARY KEY NOT NULL, + task_type TEXT NOT NULL, + lifecycle_owner TEXT NOT NULL, + status TEXT NOT NULL, + created_at TEXT NOT NULL, + started_at TEXT NOT NULL, + finished_at TEXT, + owner_instance_id TEXT NOT NULL, + error_code TEXT, + safe_error_summary TEXT, + history_complete INTEGER NOT NULL DEFAULT 0, + source_connection_id TEXT NOT NULL, + source_database_type TEXT NOT NULL, + source_database TEXT NOT NULL, + source_schema TEXT NOT NULL, + source_catalog TEXT, + target_connection_id TEXT NOT NULL, + target_database_type TEXT NOT NULL, + target_database TEXT NOT NULL, + target_schema TEXT NOT NULL, + target_catalog TEXT + )", + "CREATE INDEX IF NOT EXISTS idx_task_runs_created ON task_runs(created_at DESC, run_id DESC)", + "CREATE INDEX IF NOT EXISTS idx_task_runs_type_status_created ON task_runs(task_type, status, created_at DESC, run_id DESC)", + "CREATE TABLE IF NOT EXISTS transfer_run_details ( + run_id TEXT PRIMARY KEY NOT NULL, + content TEXT NOT NULL, + mode TEXT NOT NULL, + batch_size INTEGER NOT NULL, + create_table INTEGER NOT NULL, + drop_target_before_create INTEGER NOT NULL, + target_table_name_case TEXT NOT NULL, + quote_target_column_names INTEGER NOT NULL, + ownership_policy TEXT NOT NULL, + filtered_table_count INTEGER NOT NULL, + table_total INTEGER NOT NULL, + object_selection_mode TEXT NOT NULL, + selected_object_count INTEGER + )", + "CREATE TABLE IF NOT EXISTS task_run_items ( + run_id TEXT NOT NULL, + item_index INTEGER NOT NULL, + item_kind TEXT NOT NULL, + source_object TEXT NOT NULL, + target_object TEXT NOT NULL, + status TEXT NOT NULL, + source_row_count INTEGER, + moved_row_count INTEGER, + target_row_count INTEGER, + row_count_state TEXT NOT NULL, + has_table_filter INTEGER NOT NULL DEFAULT 0, + safe_error_summary TEXT, + PRIMARY KEY (run_id, item_index) + )", + "CREATE INDEX IF NOT EXISTS idx_task_run_items_run_status ON task_run_items(run_id, status, item_index)", "CREATE TABLE IF NOT EXISTS ai_config ( id INTEGER PRIMARY KEY CHECK (id = 1), config_json TEXT NOT NULL @@ -3836,6 +3901,475 @@ impl Storage { } } +impl Storage { + pub async fn create_task_run( + &self, + run: &TaskRun, + details: &TransferRunDetails, + items: &[TaskRunItem], + ) -> Result<(), TaskHistoryStorageError> { + let run = run.clone(); + let details = details.clone(); + let items = items.to_vec(); + let inserted = self + .with_conn(move |conn| { + let tx = conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(|error| error.to_string())?; + let exists = tx + .query_row("SELECT 1 FROM task_runs WHERE run_id = ?1", [&run.run_id], |row| row.get::<_, i64>(0)) + .optional() + .map_err(|error| error.to_string())? + .is_some(); + if exists { + return Ok(false); + } + tx.execute( + "INSERT INTO task_runs ( + run_id, task_type, lifecycle_owner, status, created_at, started_at, finished_at, + owner_instance_id, error_code, safe_error_summary, history_complete, + source_connection_id, source_database_type, source_database, source_schema, source_catalog, + target_connection_id, target_database_type, target_database, target_schema, target_catalog + ) VALUES ( + ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, + ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21 + )", + params![ + run.run_id, + run.task_type.as_storage_str(), + run.lifecycle_owner.as_storage_str(), + run.status.as_storage_str(), + run.created_at, + run.started_at, + run.finished_at, + run.owner_instance_id, + run.error_code, + run.safe_error_summary, + run.history_complete, + run.source.connection_id, + run.source.database_type, + run.source.database, + run.source.schema, + run.source.catalog, + run.target.connection_id, + run.target.database_type, + run.target.database, + run.target.schema, + run.target.catalog, + ], + ) + .map_err(|error| error.to_string())?; + tx.execute( + "INSERT INTO transfer_run_details ( + run_id, content, mode, batch_size, create_table, drop_target_before_create, + target_table_name_case, quote_target_column_names, ownership_policy, + filtered_table_count, table_total, object_selection_mode, selected_object_count + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)", + params![ + details.run_id, + details.content.as_storage_str(), + details.mode.as_storage_str(), + details.batch_size, + details.create_table, + details.drop_target_before_create, + details.target_table_name_case.as_storage_str(), + details.quote_target_column_names, + details.ownership_policy.as_storage_str(), + details.filtered_table_count, + details.table_total, + details.object_selection_mode.as_storage_str(), + details.selected_object_count, + ], + ) + .map_err(|error| error.to_string())?; + for item in items { + insert_task_run_item(&tx, &item)?; + } + tx.commit().map_err(|error| error.to_string())?; + Ok(true) + }) + .await + .map_err(|_| TaskHistoryStorageError::StorageUnavailable)?; + if inserted { + Ok(()) + } else { + Err(TaskHistoryStorageError::RunIdConflict) + } + } + + pub async fn save_task_run_item(&self, item: &TaskRunItem) -> Result { + let item = item.clone(); + self.with_conn(move |conn| { + let tx = conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(|error| error.to_string())?; + let run_status = tx + .query_row("SELECT status FROM task_runs WHERE run_id = ?1", [&item.run_id], |row| { + row.get::<_, String>(0) + }) + .optional() + .map_err(|error| error.to_string())?; + if run_status.as_deref() != Some("running") { + return Ok(false); + } + let changed = tx + .execute( + "INSERT INTO task_run_items ( + run_id, item_index, item_kind, source_object, target_object, status, + source_row_count, moved_row_count, target_row_count, row_count_state, + has_table_filter, safe_error_summary + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12) + ON CONFLICT(run_id, item_index) DO UPDATE SET + item_kind = excluded.item_kind, + source_object = excluded.source_object, + target_object = excluded.target_object, + status = excluded.status, + source_row_count = excluded.source_row_count, + moved_row_count = excluded.moved_row_count, + target_row_count = excluded.target_row_count, + row_count_state = excluded.row_count_state, + has_table_filter = excluded.has_table_filter, + safe_error_summary = excluded.safe_error_summary + WHERE task_run_items.status NOT IN ('succeeded', 'skipped', 'failed', 'cancelled', 'not_started', 'incomplete')", + params![ + item.run_id, + item.item_index, + item.item_kind.as_storage_str(), + item.source_object, + item.target_object, + item.status.as_storage_str(), + item.source_row_count, + item.moved_row_count, + item.target_row_count, + item.row_count_state.as_storage_str(), + item.has_table_filter, + item.safe_error_summary, + ], + ) + .map_err(|error| error.to_string())?; + tx.commit().map_err(|error| error.to_string())?; + Ok(changed > 0) + }) + .await + .map_err(|_| TaskHistoryStorageError::StorageUnavailable) + } + + pub async fn finish_task_run( + &self, + run_id: &str, + status: TaskRunStatus, + error_code: Option<&str>, + safe_error_summary: Option<&str>, + history_complete: bool, + ) -> Result { + let run_id = run_id.to_string(); + let error_code = error_code.map(str::to_string); + let safe_error_summary = safe_error_summary.map(str::to_string); + self.with_conn(move |conn| { + let tx = conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(|error| error.to_string())?; + let current = tx + .query_row("SELECT status FROM task_runs WHERE run_id = ?1", [&run_id], |row| { + row.get::<_, String>(0) + }) + .optional() + .map_err(|error| error.to_string())?; + let Some(current) = current else { + return Err("Task run does not exist".to_string()); + }; + if current != "running" { + return Ok(false); + } + if !status.is_terminal() { + return Err("A task run cannot be finished with a running status".to_string()); + } + let finished_at = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true); + let changed = tx + .execute( + "UPDATE task_runs SET status = ?2, finished_at = ?3, error_code = ?4, + safe_error_summary = ?5, history_complete = ?6 WHERE run_id = ?1 AND status = 'running'", + params![run_id, status.as_storage_str(), finished_at, error_code, safe_error_summary, history_complete], + ) + .map_err(|error| error.to_string())?; + if changed == 0 { + return Ok(false); + } + tx.execute( + "UPDATE task_run_items SET status = 'not_started' + WHERE run_id = ?1 AND status = 'pending'", + [&run_id], + ) + .map_err(|error| error.to_string())?; + tx.execute( + "UPDATE task_run_items SET status = 'incomplete', + row_count_state = CASE WHEN row_count_state = 'not_applicable' THEN row_count_state ELSE 'incomplete' END, + safe_error_summary = COALESCE(safe_error_summary, ?2) + WHERE run_id = ?1 AND status = 'running'", + params![run_id, "Transfer item did not complete."], + ) + .map_err(|error| error.to_string())?; + tx.commit().map_err(|error| error.to_string())?; + Ok(true) + }) + .await + .map_err(|_| TaskHistoryStorageError::StorageUnavailable) + } + + pub async fn get_task_run_detail(&self, run_id: &str) -> Result, TaskHistoryStorageError> { + let run_id = run_id.to_string(); + self.with_conn(move |conn| { + let run = conn + .query_row( + "SELECT run_id, task_type, lifecycle_owner, status, created_at, started_at, finished_at, + owner_instance_id, error_code, safe_error_summary, history_complete, + source_connection_id, source_database_type, source_database, source_schema, source_catalog, + target_connection_id, target_database_type, target_database, target_schema, target_catalog + FROM task_runs WHERE run_id = ?1", + [&run_id], + map_task_run_row, + ) + .optional() + .map_err(|error| error.to_string())?; + let Some(run) = run else { + return Ok(None); + }; + let transfer = conn + .query_row( + "SELECT run_id, content, mode, batch_size, create_table, drop_target_before_create, + target_table_name_case, quote_target_column_names, ownership_policy, + filtered_table_count, table_total, object_selection_mode, selected_object_count + FROM transfer_run_details WHERE run_id = ?1", + [&run_id], + map_transfer_run_details_row, + ) + .optional() + .map_err(|error| error.to_string())?; + Ok(Some(TaskRunDetail { run, transfer })) + }) + .await + .map_err(|_| TaskHistoryStorageError::StorageUnavailable) + } + + pub async fn list_task_runs(&self, query: TaskRunListQuery) -> Result { + self.with_conn(move |conn| { + let limit = query.limit.filter(|limit| *limit > 0).unwrap_or(50).clamp(1, 100); + let mut predicates = Vec::new(); + let mut values = Vec::::new(); + if let Some(task_type) = query.task_type { + predicates.push("task_type = ?".to_string()); + values.push(Value::Text(task_type.as_storage_str().to_string())); + } + if let Some(status) = query.status { + predicates.push("status = ?".to_string()); + values.push(Value::Text(status.as_storage_str().to_string())); + } + if let Some(cursor) = query.cursor { + predicates.push("(created_at < ? OR (created_at = ? AND run_id < ?))".to_string()); + values.push(Value::Text(cursor.created_at.clone())); + values.push(Value::Text(cursor.created_at)); + values.push(Value::Text(cursor.run_id)); + } + let where_clause = + if predicates.is_empty() { String::new() } else { format!(" WHERE {}", predicates.join(" AND ")) }; + let sql = format!( + "SELECT run_id, task_type, lifecycle_owner, status, created_at, started_at, finished_at, + owner_instance_id, error_code, safe_error_summary, history_complete, + source_connection_id, source_database_type, source_database, source_schema, source_catalog, + target_connection_id, target_database_type, target_database, target_schema, target_catalog + FROM task_runs{where_clause} ORDER BY created_at DESC, run_id DESC LIMIT ?" + ); + values.push(Value::Integer((limit + 1) as i64)); + let mut stmt = conn.prepare(&sql).map_err(|error| error.to_string())?; + let rows = + stmt.query_map(params_from_iter(values.iter()), map_task_run_row).map_err(|error| error.to_string())?; + let mut items = rows.collect::, _>>().map_err(|error| error.to_string())?; + let has_more = items.len() > limit; + items.truncate(limit); + let next_cursor = if has_more { + items.last().map(|run| TaskRunCursor { created_at: run.created_at.clone(), run_id: run.run_id.clone() }) + } else { + None + }; + Ok(TaskRunPage { items, next_cursor }) + }) + .await + .map_err(|_| TaskHistoryStorageError::StorageUnavailable) + } + + pub async fn list_task_run_items( + &self, + run_id: &str, + query: TaskRunItemsQuery, + ) -> Result { + let run_id = run_id.to_string(); + self.with_conn(move |conn| { + let limit = query.limit.filter(|limit| *limit > 0).unwrap_or(100).clamp(1, 200); + let after = query.after_item_index.unwrap_or(-1); + let mut stmt = conn + .prepare( + "SELECT run_id, item_index, item_kind, source_object, target_object, status, + source_row_count, moved_row_count, target_row_count, row_count_state, + has_table_filter, safe_error_summary + FROM task_run_items WHERE run_id = ?1 AND item_index > ?2 + ORDER BY item_index ASC LIMIT ?3", + ) + .map_err(|error| error.to_string())?; + let rows = stmt + .query_map(params![run_id, after, (limit + 1) as i64], map_task_run_item_row) + .map_err(|error| error.to_string())?; + let mut items = rows.collect::, _>>().map_err(|error| error.to_string())?; + let has_more = items.len() > limit; + items.truncate(limit); + let next_after_item_index = if has_more { items.last().map(|item| item.item_index) } else { None }; + Ok(TaskRunItemsPage { items, next_after_item_index }) + }) + .await + .map_err(|_| TaskHistoryStorageError::StorageUnavailable) + } + + pub async fn delete_task_run(&self, run_id: &str) -> Result { + const RUNNING_SENTINEL: &str = "TASK_RUN_STILL_RUNNING"; + let run_id = run_id.to_string(); + let result = self + .with_conn(move |conn| { + let tx = conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(|error| error.to_string())?; + let current = tx + .query_row("SELECT status FROM task_runs WHERE run_id = ?1", [&run_id], |row| { + row.get::<_, String>(0) + }) + .optional() + .map_err(|error| error.to_string())?; + let Some(current) = current else { + return Ok(false); + }; + if TaskRunStatus::from_storage(¤t).is_none_or(|status| !status.is_terminal()) { + return Err(RUNNING_SENTINEL.to_string()); + } + tx.execute("DELETE FROM task_run_items WHERE run_id = ?1", [&run_id]) + .map_err(|error| error.to_string())?; + tx.execute("DELETE FROM transfer_run_details WHERE run_id = ?1", [&run_id]) + .map_err(|error| error.to_string())?; + tx.execute("DELETE FROM task_runs WHERE run_id = ?1", [&run_id]).map_err(|error| error.to_string())?; + tx.commit().map_err(|error| error.to_string())?; + Ok(true) + }) + .await; + match result { + Ok(deleted) => Ok(deleted), + Err(error) if error == RUNNING_SENTINEL => Err(TaskHistoryStorageError::RunStillRunning), + Err(_) => Err(TaskHistoryStorageError::StorageUnavailable), + } + } +} + +fn insert_task_run_item(tx: &rusqlite::Transaction<'_>, item: &TaskRunItem) -> Result<(), String> { + tx.execute( + "INSERT INTO task_run_items ( + run_id, item_index, item_kind, source_object, target_object, status, + source_row_count, moved_row_count, target_row_count, row_count_state, + has_table_filter, safe_error_summary + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)", + params![ + item.run_id, + item.item_index, + item.item_kind.as_storage_str(), + item.source_object, + item.target_object, + item.status.as_storage_str(), + item.source_row_count, + item.moved_row_count, + item.target_row_count, + item.row_count_state.as_storage_str(), + item.has_table_filter, + item.safe_error_summary, + ], + ) + .map(|_| ()) + .map_err(|error| error.to_string()) +} + +fn map_task_run_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let task_type: String = row.get(1)?; + let lifecycle_owner: String = row.get(2)?; + let status: String = row.get(3)?; + Ok(TaskRun { + run_id: row.get(0)?, + task_type: TaskType::from_storage(&task_type).ok_or(rusqlite::Error::InvalidQuery)?, + lifecycle_owner: TaskLifecycleOwner::from_storage(&lifecycle_owner).ok_or(rusqlite::Error::InvalidQuery)?, + status: TaskRunStatus::from_storage(&status).ok_or(rusqlite::Error::InvalidQuery)?, + created_at: row.get(4)?, + started_at: row.get(5)?, + finished_at: row.get(6)?, + owner_instance_id: row.get(7)?, + error_code: row.get(8)?, + safe_error_summary: row.get(9)?, + history_complete: row.get(10)?, + source: TaskEndpointSnapshot { + connection_id: row.get(11)?, + database_type: row.get(12)?, + database: row.get(13)?, + schema: row.get(14)?, + catalog: row.get(15)?, + }, + target: TaskEndpointSnapshot { + connection_id: row.get(16)?, + database_type: row.get(17)?, + database: row.get(18)?, + schema: row.get(19)?, + catalog: row.get(20)?, + }, + }) +} + +fn map_transfer_run_details_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let content: String = row.get(1)?; + let mode: String = row.get(2)?; + let table_name_case: String = row.get(6)?; + let ownership_policy: String = row.get(8)?; + let selection_mode: String = row.get(11)?; + Ok(TransferRunDetails { + run_id: row.get(0)?, + content: TransferRunContent::from_storage(&content).ok_or(rusqlite::Error::InvalidQuery)?, + mode: TransferRunMode::from_storage(&mode).ok_or(rusqlite::Error::InvalidQuery)?, + batch_size: row.get(3)?, + create_table: row.get(4)?, + drop_target_before_create: row.get(5)?, + target_table_name_case: TransferRunTargetTableNameCase::from_storage(&table_name_case) + .ok_or(rusqlite::Error::InvalidQuery)?, + quote_target_column_names: row.get(7)?, + ownership_policy: TransferRunOwnershipPolicy::from_storage(&ownership_policy) + .ok_or(rusqlite::Error::InvalidQuery)?, + filtered_table_count: row.get(9)?, + table_total: row.get(10)?, + object_selection_mode: TransferObjectSelectionMode::from_storage(&selection_mode) + .ok_or(rusqlite::Error::InvalidQuery)?, + selected_object_count: row.get(12)?, + }) +} + +fn map_task_run_item_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let item_kind: String = row.get(2)?; + let status: String = row.get(5)?; + let row_count_state: String = row.get(9)?; + Ok(TaskRunItem { + run_id: row.get(0)?, + item_index: row.get(1)?, + item_kind: TaskItemKind::from_storage(&item_kind).ok_or(rusqlite::Error::InvalidQuery)?, + source_object: row.get(3)?, + target_object: row.get(4)?, + status: TaskItemStatus::from_storage(&status).ok_or(rusqlite::Error::InvalidQuery)?, + source_row_count: row.get(6)?, + moved_row_count: row.get(7)?, + target_row_count: row.get(8)?, + row_count_state: RowCountState::from_storage(&row_count_state).ok_or(rusqlite::Error::InvalidQuery)?, + has_table_filter: row.get(10)?, + safe_error_summary: row.get(11)?, + }) +} + // AI Config fn ai_provider_key(provider: &AiProvider) -> String { @@ -14243,4 +14777,133 @@ mod tests { .unwrap(); assert_eq!(storage.stored_connection_count().await.unwrap(), None); } + + #[tokio::test] + async fn task_run_storage_is_idempotent_paginated_and_deletes_only_terminal_runs() { + use crate::persistence::task_history::{ + RowCountState, TaskEndpointSnapshot, TaskHistoryStorageError, TaskItemKind, TaskItemStatus, + TaskLifecycleOwner, TaskRun, TaskRunItem, TaskRunItemsQuery, TaskRunListQuery, TaskRunStatus, TaskType, + TransferObjectSelectionMode, TransferRunContent, TransferRunDetails, TransferRunMode, + TransferRunOwnershipPolicy, TransferRunTargetTableNameCase, + }; + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("dbx.db"); + let storage = crate::persistence::test_storage::open_unmigrated(&path).await.unwrap(); + let run = TaskRun { + run_id: "history-storage-run".to_string(), + task_type: TaskType::Transfer, + lifecycle_owner: TaskLifecycleOwner::Web, + status: TaskRunStatus::Running, + created_at: "2025-01-01T00:00:00.000Z".to_string(), + started_at: "2025-01-01T00:00:00.000Z".to_string(), + finished_at: None, + owner_instance_id: "test-instance".to_string(), + error_code: None, + safe_error_summary: None, + history_complete: false, + source: TaskEndpointSnapshot { + connection_id: "source-id".to_string(), + database_type: "sqlite".to_string(), + database: "source-db".to_string(), + schema: "main".to_string(), + catalog: None, + }, + target: TaskEndpointSnapshot { + connection_id: "target-id".to_string(), + database_type: "sqlite".to_string(), + database: "target-db".to_string(), + schema: "main".to_string(), + catalog: None, + }, + }; + let details = TransferRunDetails { + run_id: run.run_id.clone(), + content: TransferRunContent::DataOnly, + mode: TransferRunMode::Append, + batch_size: 500, + create_table: false, + drop_target_before_create: false, + target_table_name_case: TransferRunTargetTableNameCase::Preserve, + quote_target_column_names: true, + ownership_policy: TransferRunOwnershipPolicy::Preserve, + filtered_table_count: 0, + table_total: 1, + object_selection_mode: TransferObjectSelectionMode::Explicit, + selected_object_count: Some(0), + }; + let item = TaskRunItem { + run_id: run.run_id.clone(), + item_index: 0, + item_kind: TaskItemKind::Table, + source_object: "orders".to_string(), + target_object: "orders".to_string(), + status: TaskItemStatus::Pending, + source_row_count: None, + moved_row_count: None, + target_row_count: None, + row_count_state: RowCountState::Unknown, + has_table_filter: false, + safe_error_summary: None, + }; + + storage.create_task_run(&run, &details, std::slice::from_ref(&item)).await.unwrap(); + assert_eq!( + storage.create_task_run(&run, &details, std::slice::from_ref(&item)).await, + Err(TaskHistoryStorageError::RunIdConflict) + ); + assert_eq!(storage.delete_task_run(&run.run_id).await, Err(TaskHistoryStorageError::RunStillRunning)); + + let completed_item = TaskRunItem { + status: TaskItemStatus::Succeeded, + source_row_count: Some(0), + moved_row_count: Some(0), + row_count_state: RowCountState::Known, + ..item + }; + assert!(storage.save_task_run_item(&completed_item).await.unwrap()); + assert!(storage.finish_task_run(&run.run_id, TaskRunStatus::Succeeded, None, None, true).await.unwrap()); + assert!(!storage + .finish_task_run(&run.run_id, TaskRunStatus::Failed, Some("late"), Some("late"), true) + .await + .unwrap()); + + let older_run = TaskRun { + run_id: "history-storage-run-older".to_string(), + created_at: "2024-12-31T00:00:00.000Z".to_string(), + started_at: "2024-12-31T00:00:00.000Z".to_string(), + ..run.clone() + }; + let older_details = TransferRunDetails { run_id: older_run.run_id.clone(), ..details.clone() }; + storage.create_task_run(&older_run, &older_details, &[]).await.unwrap(); + assert!(storage.finish_task_run(&older_run.run_id, TaskRunStatus::Succeeded, None, None, true).await.unwrap()); + + let page = storage.list_task_runs(TaskRunListQuery { limit: Some(1), ..Default::default() }).await.unwrap(); + assert_eq!(page.items.len(), 1); + assert_eq!(page.items[0].run_id, run.run_id); + let cursor = page.next_cursor.clone().expect("older run must be available on the next page"); + let next_page = storage + .list_task_runs(TaskRunListQuery { limit: Some(1), cursor: Some(cursor), ..Default::default() }) + .await + .unwrap(); + assert_eq!(next_page.items.len(), 1); + assert_eq!(next_page.items[0].run_id, older_run.run_id); + assert!(next_page.next_cursor.is_none()); + let item_page = storage + .list_task_run_items(&run.run_id, TaskRunItemsQuery { limit: Some(1), ..Default::default() }) + .await + .unwrap(); + assert_eq!(item_page.items[0].source_row_count, Some(0)); + assert_eq!(item_page.items[0].moved_row_count, Some(0)); + assert_eq!(item_page.items[0].target_row_count, None); + + drop(storage); + let reopened = crate::persistence::test_storage::open_unmigrated(&path).await.unwrap(); + let detail = reopened.get_task_run_detail(&run.run_id).await.unwrap().unwrap(); + assert_eq!(detail.run.status, TaskRunStatus::Succeeded); + assert!(detail.run.history_complete); + assert_eq!(detail.transfer.unwrap().selected_object_count, Some(0)); + assert!(reopened.delete_task_run(&run.run_id).await.unwrap()); + assert!(reopened.get_task_run_detail(&run.run_id).await.unwrap().is_none()); + } } diff --git a/crates/dbx-core/src/persistence/task_history.rs b/crates/dbx-core/src/persistence/task_history.rs new file mode 100644 index 0000000000..e76a56abbb --- /dev/null +++ b/crates/dbx-core/src/persistence/task_history.rs @@ -0,0 +1,1018 @@ +use std::collections::{HashMap, HashSet}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::{LazyLock, Mutex}; + +use chrono::{SecondsFormat, Utc}; +use serde::{Deserialize, Serialize}; + +use crate::connection::AppState; +use crate::data::transfer::{ + TransferContent, TransferMode, TransferObjectOutcome, TransferOwnershipPolicy, TransferProgress, TransferRequest, + TransferStatus, TransferTableNameCase, +}; +use crate::persistence::storage::Storage; + +const HISTORY_STORAGE_UNAVAILABLE: &str = "Task history storage is unavailable."; +const ITEM_FAILED_SUMMARY: &str = "Transfer item failed."; +const ITEM_CANCELLED_SUMMARY: &str = "Transfer item was cancelled."; +const ITEM_INCOMPLETE_SUMMARY: &str = "Transfer item did not complete."; +const RUN_FAILED_SUMMARY: &str = "Transfer failed before completion."; +const RUN_PARTIAL_SUMMARY: &str = "Transfer completed with failed items."; +const RUN_CANCELLED_SUMMARY: &str = "Transfer was cancelled."; + +/// A storage failure is deliberately distinct from a duplicate run identifier. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TaskHistoryStorageError { + RunIdConflict, + RunStillRunning, + StorageUnavailable, +} + +impl TaskHistoryStorageError { + pub const fn code(self) -> &'static str { + match self { + Self::RunIdConflict => "TRANSFER_RUN_ID_CONFLICT", + Self::RunStillRunning => "TASK_RUN_STILL_RUNNING", + Self::StorageUnavailable => "TASK_HISTORY_STORAGE_UNAVAILABLE", + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TaskType { + Transfer, +} + +impl TaskType { + pub(crate) const fn as_storage_str(self) -> &'static str { + match self { + Self::Transfer => "transfer", + } + } + + pub(crate) fn from_storage(value: &str) -> Option { + match value { + "transfer" => Some(Self::Transfer), + _ => None, + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TaskLifecycleOwner { + Tauri, + Web, +} + +impl TaskLifecycleOwner { + pub(crate) const fn as_storage_str(self) -> &'static str { + match self { + Self::Tauri => "tauri", + Self::Web => "web", + } + } + + pub(crate) fn from_storage(value: &str) -> Option { + match value { + "tauri" => Some(Self::Tauri), + "web" => Some(Self::Web), + _ => None, + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TaskRunStatus { + Running, + Succeeded, + PartialFailed, + Failed, + Cancelled, +} + +impl TaskRunStatus { + pub(crate) const fn as_storage_str(self) -> &'static str { + match self { + Self::Running => "running", + Self::Succeeded => "succeeded", + Self::PartialFailed => "partial_failed", + Self::Failed => "failed", + Self::Cancelled => "cancelled", + } + } + + pub(crate) fn from_storage(value: &str) -> Option { + match value { + "running" => Some(Self::Running), + "succeeded" => Some(Self::Succeeded), + "partial_failed" => Some(Self::PartialFailed), + "failed" => Some(Self::Failed), + "cancelled" => Some(Self::Cancelled), + _ => None, + } + } + + pub const fn is_terminal(self) -> bool { + !matches!(self, Self::Running) + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TransferRunContent { + StructureAndData, + StructureOnly, + DataOnly, +} + +impl TransferRunContent { + fn from_transfer(value: TransferContent) -> Self { + match value { + TransferContent::StructureAndData => Self::StructureAndData, + TransferContent::StructureOnly => Self::StructureOnly, + TransferContent::DataOnly => Self::DataOnly, + } + } + + pub(crate) const fn as_storage_str(self) -> &'static str { + match self { + Self::StructureAndData => "structure_and_data", + Self::StructureOnly => "structure_only", + Self::DataOnly => "data_only", + } + } + + pub(crate) fn from_storage(value: &str) -> Option { + match value { + "structure_and_data" => Some(Self::StructureAndData), + "structure_only" => Some(Self::StructureOnly), + "data_only" => Some(Self::DataOnly), + _ => None, + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TransferRunMode { + Append, + Overwrite, + Upsert, +} + +impl TransferRunMode { + fn from_transfer(value: &TransferMode) -> Self { + match value { + TransferMode::Append => Self::Append, + TransferMode::Overwrite => Self::Overwrite, + TransferMode::Upsert => Self::Upsert, + } + } + + pub(crate) const fn as_storage_str(self) -> &'static str { + match self { + Self::Append => "append", + Self::Overwrite => "overwrite", + Self::Upsert => "upsert", + } + } + + pub(crate) fn from_storage(value: &str) -> Option { + match value { + "append" => Some(Self::Append), + "overwrite" => Some(Self::Overwrite), + "upsert" => Some(Self::Upsert), + _ => None, + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TransferRunTargetTableNameCase { + Preserve, + Lower, + Upper, +} + +impl TransferRunTargetTableNameCase { + fn from_transfer(value: &TransferTableNameCase) -> Self { + match value { + TransferTableNameCase::Preserve => Self::Preserve, + TransferTableNameCase::Lower => Self::Lower, + TransferTableNameCase::Upper => Self::Upper, + } + } + + pub(crate) const fn as_storage_str(self) -> &'static str { + match self { + Self::Preserve => "preserve", + Self::Lower => "lower", + Self::Upper => "upper", + } + } + + pub(crate) fn from_storage(value: &str) -> Option { + match value { + "preserve" => Some(Self::Preserve), + "lower" => Some(Self::Lower), + "upper" => Some(Self::Upper), + _ => None, + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TransferRunOwnershipPolicy { + Preserve, + Skip, + ReassignMissing, +} + +impl TransferRunOwnershipPolicy { + fn from_transfer(value: &TransferOwnershipPolicy) -> Self { + match value { + TransferOwnershipPolicy::Preserve => Self::Preserve, + TransferOwnershipPolicy::Skip => Self::Skip, + TransferOwnershipPolicy::ReassignMissing => Self::ReassignMissing, + } + } + + pub(crate) const fn as_storage_str(self) -> &'static str { + match self { + Self::Preserve => "preserve", + Self::Skip => "skip", + Self::ReassignMissing => "reassign_missing", + } + } + + pub(crate) fn from_storage(value: &str) -> Option { + match value { + "preserve" => Some(Self::Preserve), + "skip" => Some(Self::Skip), + "reassign_missing" => Some(Self::ReassignMissing), + _ => None, + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TransferObjectSelectionMode { + Unspecified, + Explicit, +} + +impl TransferObjectSelectionMode { + pub(crate) const fn as_storage_str(self) -> &'static str { + match self { + Self::Unspecified => "unspecified", + Self::Explicit => "explicit", + } + } + + pub(crate) fn from_storage(value: &str) -> Option { + match value { + "unspecified" => Some(Self::Unspecified), + "explicit" => Some(Self::Explicit), + _ => None, + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TaskItemKind { + Table, + View, + MaterializedView, + Procedure, + Function, + Trigger, + Sequence, + Event, + Object, +} + +impl TaskItemKind { + pub(crate) const fn as_storage_str(self) -> &'static str { + match self { + Self::Table => "table", + Self::View => "view", + Self::MaterializedView => "materialized_view", + Self::Procedure => "procedure", + Self::Function => "function", + Self::Trigger => "trigger", + Self::Sequence => "sequence", + Self::Event => "event", + Self::Object => "object", + } + } + + pub(crate) fn from_storage(value: &str) -> Option { + match value { + "table" => Some(Self::Table), + "view" => Some(Self::View), + "materialized_view" => Some(Self::MaterializedView), + "procedure" => Some(Self::Procedure), + "function" => Some(Self::Function), + "trigger" => Some(Self::Trigger), + "sequence" => Some(Self::Sequence), + "event" => Some(Self::Event), + "object" => Some(Self::Object), + _ => None, + } + } + + fn from_transfer(value: &str) -> (Self, String) { + let Some((kind, name)) = value.split_once(':') else { + return (Self::Object, value.to_string()); + }; + let kind = match kind { + "Table" => Self::Table, + "View" => Self::View, + "MaterializedView" => Self::MaterializedView, + "Procedure" => Self::Procedure, + "Function" => Self::Function, + "Trigger" => Self::Trigger, + "Sequence" => Self::Sequence, + "Event" => Self::Event, + _ => return (Self::Object, value.to_string()), + }; + (kind, name.to_string()) + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TaskItemStatus { + Pending, + Running, + Succeeded, + Skipped, + Failed, + Cancelled, + NotStarted, + Incomplete, +} + +impl TaskItemStatus { + pub(crate) const fn as_storage_str(self) -> &'static str { + match self { + Self::Pending => "pending", + Self::Running => "running", + Self::Succeeded => "succeeded", + Self::Skipped => "skipped", + Self::Failed => "failed", + Self::Cancelled => "cancelled", + Self::NotStarted => "not_started", + Self::Incomplete => "incomplete", + } + } + + pub(crate) fn from_storage(value: &str) -> Option { + match value { + "pending" => Some(Self::Pending), + "running" => Some(Self::Running), + "succeeded" => Some(Self::Succeeded), + "skipped" => Some(Self::Skipped), + "failed" => Some(Self::Failed), + "cancelled" => Some(Self::Cancelled), + "not_started" => Some(Self::NotStarted), + "incomplete" => Some(Self::Incomplete), + _ => None, + } + } +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum RowCountState { + NotApplicable, + Known, + Unknown, + Incomplete, +} + +impl RowCountState { + pub(crate) const fn as_storage_str(self) -> &'static str { + match self { + Self::NotApplicable => "not_applicable", + Self::Known => "known", + Self::Unknown => "unknown", + Self::Incomplete => "incomplete", + } + } + + pub(crate) fn from_storage(value: &str) -> Option { + match value { + "not_applicable" => Some(Self::NotApplicable), + "known" => Some(Self::Known), + "unknown" => Some(Self::Unknown), + "incomplete" => Some(Self::Incomplete), + _ => None, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct TaskEndpointSnapshot { + pub connection_id: String, + pub database_type: String, + pub database: String, + pub schema: String, + pub catalog: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct TaskRun { + pub run_id: String, + pub task_type: TaskType, + pub lifecycle_owner: TaskLifecycleOwner, + pub status: TaskRunStatus, + pub created_at: String, + pub started_at: String, + pub finished_at: Option, + pub owner_instance_id: String, + pub error_code: Option, + pub safe_error_summary: Option, + /// False until a terminal write succeeds; a crash therefore cannot look complete. + pub history_complete: bool, + pub source: TaskEndpointSnapshot, + pub target: TaskEndpointSnapshot, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct TransferRunDetails { + pub run_id: String, + pub content: TransferRunContent, + pub mode: TransferRunMode, + pub batch_size: i64, + pub create_table: bool, + pub drop_target_before_create: bool, + pub target_table_name_case: TransferRunTargetTableNameCase, + pub quote_target_column_names: bool, + pub ownership_policy: TransferRunOwnershipPolicy, + pub filtered_table_count: i64, + pub table_total: i64, + pub object_selection_mode: TransferObjectSelectionMode, + pub selected_object_count: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct TaskRunItem { + pub run_id: String, + pub item_index: i64, + pub item_kind: TaskItemKind, + pub source_object: String, + pub target_object: String, + pub status: TaskItemStatus, + pub source_row_count: Option, + /// Rows moved by this Transfer item, never the target's final row count. + pub moved_row_count: Option, + pub target_row_count: Option, + /// Null target_row_count means the target row count was not validated. + pub row_count_state: RowCountState, + pub has_table_filter: bool, + pub safe_error_summary: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct TaskRunDetail { + pub run: TaskRun, + pub transfer: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct TaskRunCursor { + pub created_at: String, + pub run_id: String, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct TaskRunListQuery { + pub limit: Option, + pub cursor: Option, + pub task_type: Option, + pub status: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct TaskRunPage { + pub items: Vec, + pub next_cursor: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct TaskRunItemsQuery { + pub limit: Option, + pub after_item_index: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct TaskRunItemsPage { + pub items: Vec, + pub next_after_item_index: Option, +} + +#[derive(Debug, Clone)] +struct TableSnapshot { + source_object: String, + target_object: String, + has_table_filter: bool, +} + +#[derive(Debug, Clone, Copy, Default)] +struct ObservedRows { + source_row_count: Option, + moved_row_count: Option, +} + +/// Per-transfer observer. It persists only request facts and result-boundary updates; +/// raw TransferRequest, table filters, SQL, and driver error strings are never retained. +#[derive(Clone)] +pub struct TransferTaskJournal { + storage: Storage, + run_id: String, + content: TransferRunContent, + tables: std::sync::Arc>, + history_complete: std::sync::Arc, + successful_items: std::sync::Arc, + observed_rows: std::sync::Arc>>, +} + +static UNPERSISTED_RUN_ID_CLAIMS: LazyLock>> = LazyLock::new(|| Mutex::new(HashSet::new())); +static OWNER_INSTANCE_ID: LazyLock = + LazyLock::new(|| format!("{}-{}", std::process::id(), uuid::Uuid::new_v4())); + +impl TransferTaskJournal { + /// Create the durable acceptance record. Storage errors degrade to an unavailable + /// observer; ID collisions remain an explicit rejection before any Transfer starts. + pub async fn accept( + storage: &Storage, + app: &AppState, + request: &TransferRequest, + lifecycle_owner: TaskLifecycleOwner, + ) -> Result, TaskHistoryStorageError> { + { + let mut claims = UNPERSISTED_RUN_ID_CLAIMS.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + if !claims.insert(request.transfer_id.clone()) { + return Err(TaskHistoryStorageError::RunIdConflict); + } + } + + let (source_type, target_type) = { + let configs = app.configs.read().await; + ( + configs.get(&request.source_connection_id).map(|config| config.db_type.as_str().to_string()), + configs.get(&request.target_connection_id).map(|config| config.db_type.as_str().to_string()), + ) + }; + let now = Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true); + let source = endpoint_snapshot( + &request.source_connection_id, + source_type.unwrap_or_else(|| "unknown".to_string()), + &request.source_database, + &request.source_schema, + request.source_catalog.clone(), + ); + let target = endpoint_snapshot( + &request.target_connection_id, + target_type.unwrap_or_else(|| "unknown".to_string()), + &request.target_database, + &request.target_schema, + request.target_catalog.clone(), + ); + let content = TransferRunContent::from_transfer(request.content); + let tables = std::sync::Arc::new( + request + .tables + .iter() + .map(|table| TableSnapshot { + source_object: table.clone(), + target_object: request.target_table_name(table), + has_table_filter: request.table_filters.get(table).is_some_and(|filter| !filter.trim().is_empty()), + }) + .collect::>(), + ); + let run = TaskRun { + run_id: request.transfer_id.clone(), + task_type: TaskType::Transfer, + lifecycle_owner, + status: TaskRunStatus::Running, + created_at: now.clone(), + started_at: now, + finished_at: None, + owner_instance_id: OWNER_INSTANCE_ID.clone(), + error_code: None, + safe_error_summary: None, + history_complete: false, + source, + target, + }; + let details = transfer_details(request); + let items = tables + .iter() + .enumerate() + .map(|(index, table)| TaskRunItem { + run_id: request.transfer_id.clone(), + item_index: index as i64, + item_kind: TaskItemKind::Table, + source_object: table.source_object.clone(), + target_object: table.target_object.clone(), + status: TaskItemStatus::Pending, + source_row_count: None, + moved_row_count: None, + target_row_count: None, + row_count_state: if content == TransferRunContent::StructureOnly { + RowCountState::NotApplicable + } else { + RowCountState::Unknown + }, + has_table_filter: table.has_table_filter, + safe_error_summary: None, + }) + .collect::>(); + + match storage.create_task_run(&run, &details, &items).await { + Ok(()) => { + let mut claims = UNPERSISTED_RUN_ID_CLAIMS.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + claims.remove(&request.transfer_id); + Ok(Some(Self { + storage: storage.clone(), + run_id: request.transfer_id.clone(), + content, + tables, + history_complete: std::sync::Arc::new(AtomicBool::new(true)), + successful_items: std::sync::Arc::new(AtomicUsize::new(0)), + observed_rows: std::sync::Arc::new(Mutex::new(HashMap::new())), + })) + } + Err(TaskHistoryStorageError::RunIdConflict) => { + let mut claims = UNPERSISTED_RUN_ID_CLAIMS.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + claims.remove(&request.transfer_id); + Err(TaskHistoryStorageError::RunIdConflict) + } + Err(TaskHistoryStorageError::RunStillRunning) => unreachable!("run creation cannot return this error"), + Err(TaskHistoryStorageError::StorageUnavailable) => { + log::warn!("[task-history] {HISTORY_STORAGE_UNAVAILABLE} Transfer will continue without a durable history record."); + // Keep this process-local claim until exit: a second submission with the same + // ID must not start another Transfer after the initial journal write failed. + Ok(None) + } + } + } + + pub fn run_id(&self) -> &str { + &self.run_id + } + + /// Return the stable request-order index for a table after the executor sorts its work. + pub fn item_index_for_table(&self, source_object: &str) -> Option { + self.tables.iter().position(|table| table.source_object == source_object) + } + + pub async fn start_table(&self, item_index: usize) { + let Some(table) = self.tables.get(item_index) else { + self.mark_write_failed("table start"); + return; + }; + let item = TaskRunItem { + run_id: self.run_id.clone(), + item_index: item_index as i64, + item_kind: TaskItemKind::Table, + source_object: table.source_object.clone(), + target_object: table.target_object.clone(), + status: TaskItemStatus::Running, + source_row_count: None, + moved_row_count: None, + target_row_count: None, + row_count_state: if self.content == TransferRunContent::StructureOnly { + RowCountState::NotApplicable + } else { + RowCountState::Unknown + }, + has_table_filter: table.has_table_filter, + safe_error_summary: None, + }; + self.persist_item(item).await; + } + + /// Capture row progress in memory only. It is flushed at the table's result boundary. + pub fn observe_table_progress(&self, item_index: usize, moved_row_count: u64) { + if let Ok(mut rows) = self.observed_rows.lock() { + rows.entry(item_index).or_default().moved_row_count = Some(moved_row_count); + } + } + + /// Record the source COUNT result from the Transfer executor without changing its public + /// progress event contract. `None` means the source count was not obtained. + pub fn observe_source_count(&self, item_index: usize, source_row_count: Option) { + if let Ok(mut rows) = self.observed_rows.lock() { + rows.entry(item_index).or_default().source_row_count = source_row_count; + } + } + + pub async fn finish_table( + &self, + item_index: usize, + status: TaskItemStatus, + source_row_count: Option, + moved_row_count: Option, + ) { + let Some(table) = self.tables.get(item_index) else { + self.mark_write_failed("table result"); + return; + }; + let observed = + self.observed_rows.lock().ok().and_then(|rows| rows.get(&item_index).copied()).unwrap_or_default(); + let source_count = source_row_count.or(observed.source_row_count); + let moved_count = moved_row_count.or(observed.moved_row_count); + let (source_row_count, moved_row_count, row_count_state) = if self.content == TransferRunContent::StructureOnly + { + (None, None, RowCountState::NotApplicable) + } else if matches!(status, TaskItemStatus::Succeeded) { + let source = source_count.and_then(to_sqlite_count); + let moved = moved_count.and_then(to_sqlite_count); + let state = if source.is_some() && moved.is_some() { RowCountState::Known } else { RowCountState::Unknown }; + (source, moved, state) + } else { + (source_count.and_then(to_sqlite_count), moved_count.and_then(to_sqlite_count), RowCountState::Incomplete) + }; + self.persist_item(TaskRunItem { + run_id: self.run_id.clone(), + item_index: item_index as i64, + item_kind: TaskItemKind::Table, + source_object: table.source_object.clone(), + target_object: table.target_object.clone(), + status, + source_row_count, + moved_row_count, + target_row_count: None, + row_count_state, + has_table_filter: table.has_table_filter, + safe_error_summary: safe_item_summary(status), + }) + .await; + } + + pub async fn record_object_outcome(&self, outcome: &TransferObjectOutcome) { + let mut next_index = self.tables.len(); + for (items, status) in [ + (&outcome.transferred, TaskItemStatus::Succeeded), + (&outcome.skipped, TaskItemStatus::Skipped), + (&outcome.failed, TaskItemStatus::Failed), + ] { + for raw in items { + let (item_kind, source_object) = TaskItemKind::from_transfer(raw); + self.persist_item(TaskRunItem { + run_id: self.run_id.clone(), + item_index: next_index as i64, + item_kind, + source_object: source_object.clone(), + target_object: source_object, + status, + source_row_count: None, + moved_row_count: None, + target_row_count: None, + row_count_state: RowCountState::NotApplicable, + has_table_filter: false, + safe_error_summary: safe_item_summary(status), + }) + .await; + next_index += 1; + } + } + } + + pub async fn record_schema_objects_error(&self, cancelled: bool) { + let status = if cancelled { TaskItemStatus::Cancelled } else { TaskItemStatus::Failed }; + self.persist_item(TaskRunItem { + run_id: self.run_id.clone(), + item_index: self.tables.len() as i64, + item_kind: TaskItemKind::Object, + source_object: "schema objects".to_string(), + target_object: "schema objects".to_string(), + status, + source_row_count: None, + moved_row_count: None, + target_row_count: None, + row_count_state: RowCountState::NotApplicable, + has_table_filter: false, + safe_error_summary: safe_item_summary(status), + }) + .await; + } + + pub async fn finish(&self, progress: &TransferProgress) { + if !progress.terminal { + return; + } + let (status, error_code, safe_summary) = match &progress.status { + TransferStatus::Done => (TaskRunStatus::Succeeded, None, None), + TransferStatus::Cancelled => ( + TaskRunStatus::Cancelled, + Some("TRANSFER_CANCELLED".to_string()), + Some(RUN_CANCELLED_SUMMARY.to_string()), + ), + TransferStatus::Error => { + if self.successful_items.load(Ordering::Relaxed) > 0 { + ( + TaskRunStatus::PartialFailed, + Some("TRANSFER_PARTIAL_FAILURE".to_string()), + Some(RUN_PARTIAL_SUMMARY.to_string()), + ) + } else { + (TaskRunStatus::Failed, Some("TRANSFER_FAILED".to_string()), Some(RUN_FAILED_SUMMARY.to_string())) + } + } + TransferStatus::Running | TransferStatus::TableDone => return, + }; + let history_complete = self.history_complete.load(Ordering::Relaxed); + if self + .storage + .finish_task_run(&self.run_id, status, error_code.as_deref(), safe_summary.as_deref(), history_complete) + .await + .is_err() + { + log::warn!("[task-history] terminal history write failed; transfer result is unchanged and the run remains unconfirmed."); + } + } + + async fn persist_item(&self, item: TaskRunItem) { + if item.status == TaskItemStatus::Succeeded { + self.successful_items.fetch_add(1, Ordering::Relaxed); + } + if self.storage.save_task_run_item(&item).await.is_err() { + self.mark_write_failed("item result"); + } + } + + fn mark_write_failed(&self, stage: &'static str) { + self.history_complete.store(false, Ordering::Relaxed); + log::warn!("[task-history] {stage} could not be persisted; Transfer continues and history is incomplete."); + } +} + +fn endpoint_snapshot( + connection_id: &str, + database_type: String, + database: &str, + schema: &str, + catalog: Option, +) -> TaskEndpointSnapshot { + TaskEndpointSnapshot { + connection_id: connection_id.to_string(), + database_type, + database: database.to_string(), + schema: schema.to_string(), + catalog, + } +} + +fn transfer_details(request: &TransferRequest) -> TransferRunDetails { + let object_selection_mode = if request.objects.is_some() { + TransferObjectSelectionMode::Explicit + } else { + TransferObjectSelectionMode::Unspecified + }; + let selected_object_count = request + .objects + .as_ref() + .map(|objects| objects.iter().map(|selection| selection.names.len()).sum::() as i64); + TransferRunDetails { + run_id: request.transfer_id.clone(), + content: TransferRunContent::from_transfer(request.content), + mode: TransferRunMode::from_transfer(&request.mode), + batch_size: i64::try_from(request.batch_size).unwrap_or(i64::MAX), + create_table: request.create_table, + drop_target_before_create: request.drop_target_before_create, + target_table_name_case: TransferRunTargetTableNameCase::from_transfer(&request.target_table_name_case), + quote_target_column_names: request.quote_target_column_names, + ownership_policy: TransferRunOwnershipPolicy::from_transfer(&request.ownership_policy), + filtered_table_count: request + .tables + .iter() + .filter(|table| request.table_filters.get(*table).is_some_and(|filter| !filter.trim().is_empty())) + .count() as i64, + table_total: request.tables.len() as i64, + object_selection_mode, + selected_object_count, + } +} + +fn safe_item_summary(status: TaskItemStatus) -> Option { + match status { + TaskItemStatus::Failed => Some(ITEM_FAILED_SUMMARY.to_string()), + TaskItemStatus::Cancelled => Some(ITEM_CANCELLED_SUMMARY.to_string()), + TaskItemStatus::Incomplete => Some(ITEM_INCOMPLETE_SUMMARY.to_string()), + _ => None, + } +} + +fn to_sqlite_count(value: u64) -> Option { + i64::try_from(value).ok() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::data::transfer::{ + TransferMode, TransferObjectSelection, TransferOwnershipPolicy, TransferTableNameCase, + }; + + #[tokio::test] + async fn journal_drops_raw_filters_and_driver_errors_from_persistent_history() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("dbx.db"); + let storage = crate::persistence::test_storage::open_unmigrated(&db_path).await.unwrap(); + let app = AppState::new_with_plugin_dir(storage.clone(), dir.path().join("plugins")); + let filter_secret = "task_history_filter_secret_marker"; + let driver_secret = "task_history_driver_error_secret_marker"; + let request = TransferRequest { + transfer_id: format!("task-history-test-{}", uuid::Uuid::new_v4()), + source_connection_id: "source-id".to_string(), + source_database: "source-db".to_string(), + source_schema: "main".to_string(), + source_catalog: None, + target_connection_id: "target-id".to_string(), + target_database: "target-db".to_string(), + target_schema: "main".to_string(), + target_catalog: None, + tables: vec!["orders".to_string()], + create_table: false, + content: TransferContent::DataOnly, + objects: Some(vec![TransferObjectSelection::default()]), + mode: TransferMode::Append, + target_table_name_case: TransferTableNameCase::Preserve, + quote_target_column_names: true, + ownership_policy: TransferOwnershipPolicy::Preserve, + batch_size: 500, + table_filters: HashMap::from([ + ("orders".to_string(), format!("WHERE note = '{filter_secret}'")), + ("unselected_table".to_string(), format!("WHERE note = '{filter_secret}'")), + ]), + drop_target_before_create: false, + drop_target_confirmed: false, + }; + let journal = + TransferTaskJournal::accept(&storage, &app, &request, TaskLifecycleOwner::Web).await.unwrap().unwrap(); + assert!(matches!( + TransferTaskJournal::accept(&storage, &app, &request, TaskLifecycleOwner::Web).await, + Err(TaskHistoryStorageError::RunIdConflict) + )); + + journal.start_table(0).await; + journal.observe_source_count(0, None); + journal.observe_table_progress(0, 2); + journal.finish_table(0, TaskItemStatus::Failed, None, None).await; + journal + .finish(&TransferProgress { + transfer_id: request.transfer_id.clone(), + table: "orders".to_string(), + table_index: 0, + total_tables: 1, + rows_transferred: 2, + total_rows: None, + status: TransferStatus::Error, + error: Some(driver_secret.to_string()), + terminal: true, + }) + .await; + + let detail = storage.get_task_run_detail(&request.transfer_id).await.unwrap().unwrap(); + assert_eq!(detail.run.status, TaskRunStatus::Failed); + assert!(detail.run.history_complete); + assert_eq!(detail.run.error_code.as_deref(), Some("TRANSFER_FAILED")); + assert_eq!(detail.transfer.as_ref().unwrap().filtered_table_count, 1); + let items = storage.list_task_run_items(&request.transfer_id, TaskRunItemsQuery::default()).await.unwrap(); + assert_eq!(items.items[0].source_row_count, None); + assert_eq!(items.items[0].moved_row_count, Some(2)); + assert_eq!(items.items[0].target_row_count, None); + assert_eq!(items.items[0].row_count_state, RowCountState::Incomplete); + assert_eq!(items.items[0].safe_error_summary.as_deref(), Some(ITEM_FAILED_SUMMARY)); + + let serialized = serde_json::to_string(&(detail, items)).unwrap(); + assert!(!serialized.contains(filter_secret)); + assert!(!serialized.contains(driver_secret)); + for path in [db_path.clone(), db_path.with_extension("db-wal"), db_path.with_extension("db-shm")] { + let bytes = std::fs::read(path).unwrap_or_default(); + assert!(!bytes.windows(filter_secret.len()).any(|window| window == filter_secret.as_bytes())); + assert!(!bytes.windows(driver_secret.len()).any(|window| window == driver_secret.as_bytes())); + } + } +} diff --git a/crates/dbx-web/src/error.rs b/crates/dbx-web/src/error.rs index 2e2f56bf68..2f93aa2634 100644 --- a/crates/dbx-web/src/error.rs +++ b/crates/dbx-web/src/error.rs @@ -26,6 +26,10 @@ impl AppError { Self::with_status(msg.into(), StatusCode::BAD_REQUEST) } + pub fn conflict(msg: impl Into) -> Self { + Self::with_status(msg.into(), StatusCode::CONFLICT) + } + pub fn forbidden(msg: impl Into) -> Self { Self::with_status(msg.into(), StatusCode::FORBIDDEN) } diff --git a/crates/dbx-web/src/main.rs b/crates/dbx-web/src/main.rs index 1f34486779..8e3d15c410 100644 --- a/crates/dbx-web/src/main.rs +++ b/crates/dbx-web/src/main.rs @@ -1322,6 +1322,9 @@ async fn serve() -> Result<(), String> { // Transfer .route("/transfer/start", post(routes::transfer::start_transfer)) .route("/transfer/ownership-preview", post(routes::transfer::preview_transfer_ownership)) + .route("/task-runs", get(routes::task_history::list_task_runs)) + .route("/task-runs/{run_id}", get(routes::task_history::get_task_run)) + .route("/task-runs/{run_id}/items", get(routes::task_history::list_task_run_items)) .route("/transfer/progress/{transferId}", get(routes::transfer::transfer_progress)) .route("/transfer/cancel", post(routes::transfer::cancel_transfer)) .route("/transfer/sort-tables-by-fk", post(routes::transfer::sort_tables_by_fk_dependency)) diff --git a/crates/dbx-web/src/routes/mod.rs b/crates/dbx-web/src/routes/mod.rs index cad1245de3..9cefd08656 100644 --- a/crates/dbx-web/src/routes/mod.rs +++ b/crates/dbx-web/src/routes/mod.rs @@ -40,6 +40,7 @@ pub mod ssh_prompt; pub mod tab_runtime_cache; pub mod table_export; pub mod table_import; +pub mod task_history; pub mod text_export; pub mod transfer; pub mod tunnel_profiles; diff --git a/crates/dbx-web/src/routes/task_history.rs b/crates/dbx-web/src/routes/task_history.rs new file mode 100644 index 0000000000..e43711399f --- /dev/null +++ b/crates/dbx-web/src/routes/task_history.rs @@ -0,0 +1,75 @@ +use std::sync::Arc; + +use axum::extract::{Path, Query, State}; +use axum::Json; +use dbx_core::persistence::task_history::{ + TaskRunCursor, TaskRunDetail, TaskRunItemsPage, TaskRunItemsQuery, TaskRunListQuery, TaskRunPage, TaskRunStatus, + TaskType, +}; +use serde::Deserialize; + +use crate::error::AppError; +use crate::state::WebState; + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TaskRunListParams { + pub limit: Option, + pub cursor_created_at: Option, + pub cursor_run_id: Option, + pub task_type: Option, + pub status: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TaskRunItemsParams { + pub limit: Option, + pub after_item_index: Option, +} + +pub async fn list_task_runs( + State(state): State>, + Query(params): Query, +) -> Result, AppError> { + let cursor = match (params.cursor_created_at, params.cursor_run_id) { + (Some(created_at), Some(run_id)) => Some(TaskRunCursor { created_at, run_id }), + (None, None) => None, + _ => return Err(AppError::bad_request("Both cursorCreatedAt and cursorRunId are required.")), + }; + let query = TaskRunListQuery { limit: params.limit, cursor, task_type: params.task_type, status: params.status }; + state.app.storage.list_task_runs(query).await.map(Json).map_err(|error| AppError::internal(error.code())) +} + +pub async fn get_task_run( + State(state): State>, + Path(run_id): Path, +) -> Result, AppError> { + state + .app + .storage + .get_task_run_detail(&run_id) + .await + .map_err(|error| AppError::internal(error.code()))? + .map(Json) + .ok_or_else(|| AppError::not_found("TASK_RUN_NOT_FOUND")) +} + +pub async fn list_task_run_items( + State(state): State>, + Path(run_id): Path, + Query(params): Query, +) -> Result, AppError> { + if state.app.storage.get_task_run_detail(&run_id).await.map_err(|error| AppError::internal(error.code()))?.is_none() + { + return Err(AppError::not_found("TASK_RUN_NOT_FOUND")); + } + let query = TaskRunItemsQuery { limit: params.limit, after_item_index: params.after_item_index }; + state + .app + .storage + .list_task_run_items(&run_id, query) + .await + .map(Json) + .map_err(|error| AppError::internal(error.code())) +} diff --git a/crates/dbx-web/src/routes/transfer.rs b/crates/dbx-web/src/routes/transfer.rs index f3e77c15c5..3cb21f1d0a 100644 --- a/crates/dbx-web/src/routes/transfer.rs +++ b/crates/dbx-web/src/routes/transfer.rs @@ -3,6 +3,9 @@ use std::sync::Arc; use axum::extract::{Path, State}; use axum::response::sse::{Event, Sse}; use axum::Json; +use dbx_core::persistence::task_history::{ + TaskHistoryStorageError, TaskItemStatus, TaskLifecycleOwner, TransferTaskJournal, +}; use dbx_core::transfer::{self, TransferRequest, TransferStatus}; use futures::stream::Stream; use serde::Deserialize; @@ -15,6 +18,9 @@ use crate::state::WebState; const COMPLETED_TRANSFER_CHANNEL_TTL: Duration = Duration::from_secs(60); fn send_transfer_progress(channel: &TransferProgressChannel, progress: &transfer::TransferProgress) { + if progress.terminal { + channel.remember_terminal_progress(progress.clone()); + } if let Ok(json) = serde_json::to_string(progress) { let kind = if progress.terminal { TransferReplayEventKind::Terminal @@ -42,6 +48,7 @@ fn terminal_transfer_error(req: &TransferRequest, error: impl ToString) -> trans } async fn finish_transfer_channel(state: &Arc, transfer_id: &str, channel: &Arc) { + channel.finish_task_history().await; transfer::clear_cancelled(transfer_id).await; let state = state.clone(); let transfer_id = transfer_id.to_string(); @@ -106,10 +113,22 @@ pub async fn start_transfer( } let transfer_id = req.transfer_id.clone(); + let history = if state.demo_mode { + None + } else { + match TransferTaskJournal::accept(&state.app.storage, &state.app, &req, TaskLifecycleOwner::Web).await { + Ok(history) => history, + Err(TaskHistoryStorageError::RunIdConflict) => { + return Err(AppError::conflict("TRANSFER_RUN_ID_CONFLICT")); + } + Err(error) => return Err(AppError::internal(error.code())), + } + }; // Keep bounded replay state so a web EventSource opened after this POST // still receives early table failures and the terminal result. let progress_channel = Arc::new(TransferProgressChannel::new()); + progress_channel.set_task_history_journal(history.clone()); state.transfer_progress_channels.write().await.insert(transfer_id.clone(), progress_channel.clone()); let app = state.app.clone(); @@ -370,10 +389,19 @@ pub async fn start_transfer( return; } + let history_item_index = + history.as_ref().and_then(|journal| journal.item_index_for_table(table)).unwrap_or(i); + if let Some(journal) = history.as_ref() { + journal.start_table(history_item_index).await; + } let progress_channel_clone = progress_channel.clone(); let mut last_rows_transferred = 0_u64; let mut last_total_rows = None; - let result = transfer::transfer_table( + let mut source_row_count = None; + let mut moved_row_count = None; + let history_for_progress = history.as_ref(); + let history_for_count = history.as_ref(); + let result = transfer::transfer_table_with_result( &app, &req, table, @@ -388,20 +416,40 @@ pub async fn start_transfer( |progress| { last_rows_transferred = progress.rows_transferred; last_total_rows = progress.total_rows; + moved_row_count = Some(progress.rows_transferred); + if let Some(journal) = history_for_progress { + journal.observe_table_progress(history_item_index, progress.rows_transferred); + } send_transfer_progress(&progress_channel_clone, &progress); }, + |source_count| { + source_row_count = source_count; + if let Some(journal) = history_for_count { + journal.observe_source_count(history_item_index, source_count); + } + }, ) .await; match result { - Ok(rows) => { + Ok(result) => { + if let Some(journal) = history.as_ref() { + journal + .finish_table( + history_item_index, + TaskItemStatus::Succeeded, + result.source_row_count, + Some(result.moved_rows), + ) + .await; + } let progress = transfer::TransferProgress { transfer_id: req.transfer_id.clone(), table: table.clone(), table_index: i, total_tables: tables.len(), - rows_transferred: rows, - total_rows: last_total_rows.or(Some(rows)), + rows_transferred: result.moved_rows, + total_rows: last_total_rows.or(Some(result.moved_rows)), status: TransferStatus::TableDone, error: None, terminal: false, @@ -410,6 +458,16 @@ pub async fn start_transfer( } Err(e) => { if e == "Cancelled" { + if let Some(journal) = history.as_ref() { + journal + .finish_table( + history_item_index, + TaskItemStatus::Cancelled, + source_row_count, + moved_row_count, + ) + .await; + } let progress = transfer::TransferProgress { transfer_id: req.transfer_id.clone(), table: table.clone(), @@ -425,6 +483,11 @@ pub async fn start_transfer( finish_transfer_channel(&state_clone, &req.transfer_id, &progress_channel).await; return; } + if let Some(journal) = history.as_ref() { + journal + .finish_table(history_item_index, TaskItemStatus::Failed, source_row_count, moved_row_count) + .await; + } failed_tables.push(table.clone()); let progress = transfer::TransferProgress { transfer_id: req.transfer_id.clone(), @@ -476,9 +539,15 @@ pub async fn start_transfer( .await { Ok(outcome) => { + if let Some(journal) = history.as_ref() { + journal.record_object_outcome(&outcome).await; + } object_outcome = outcome; } Err(e) if e == "Cancelled" => { + if let Some(journal) = history.as_ref() { + journal.record_schema_objects_error(true).await; + } let progress = transfer::TransferProgress { transfer_id: req.transfer_id.clone(), table: "schema objects".to_string(), @@ -495,6 +564,9 @@ pub async fn start_transfer( return; } Err(e) => { + if let Some(journal) = history.as_ref() { + journal.record_schema_objects_error(false).await; + } failed_tables.push("schema objects".to_string()); let progress = transfer::TransferProgress { transfer_id: req.transfer_id.clone(), @@ -673,6 +745,7 @@ mod tests { default_connect_timeout_secs, default_idle_timeout_secs, default_keepalive_interval_secs, default_query_timeout_secs, ConnectionConfig, DatabaseType, }; + use dbx_core::persistence::task_history::{TaskRunItemsQuery, TaskRunStatus, TransferObjectSelectionMode}; use dbx_core::transfer::{ TransferContent, TransferMode, TransferOwnershipPolicy, TransferRequest, TransferTableNameCase, }; @@ -801,6 +874,14 @@ mod tests { let response = start_transfer(State(state.clone()), Json(StartTransferRequest { request: req })).await.unwrap(); let _ = response.into_response(); + let duplicate = start_transfer( + State(state.clone()), + Json(StartTransferRequest { request: transfer_request("src", "dst", &dir) }), + ) + .await + .unwrap_err(); + assert_eq!(duplicate.status, axum::http::StatusCode::CONFLICT); + // Both HTTP and Tauri call the same Core schema-object stage // unconditionally. DataOnly must resolve there to a no-op without an // object progress event or database-family fallback. @@ -837,6 +918,54 @@ mod tests { terminal_error.is_none(), "DataOnly must complete without a schema-object error, got: {terminal_error:?}" ); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let task_run = loop { + let run = state.app.storage.get_task_run_detail(&transfer_id).await.unwrap(); + if run.as_ref().is_some_and(|detail| detail.run.status.is_terminal()) { + break run.unwrap(); + } + if std::time::Instant::now() > deadline { + panic!("task history did not reach a terminal state"); + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + }; + assert_eq!(task_run.run.lifecycle_owner, TaskLifecycleOwner::Web); + assert_eq!(task_run.run.status, TaskRunStatus::Succeeded); + assert!(task_run.run.history_complete); + let transfer_details = task_run.transfer.expect("accepted transfer details persisted"); + assert_eq!(transfer_details.object_selection_mode, TransferObjectSelectionMode::Explicit); + assert_eq!(transfer_details.selected_object_count, Some(0)); + let items = state.app.storage.list_task_run_items(&transfer_id, TaskRunItemsQuery::default()).await.unwrap(); + assert!(items.items.is_empty()); + let _ = std::fs::remove_dir_all(&dir); + } + + #[tokio::test] + async fn demo_mode_transfer_does_not_write_task_history() { + let (mut state, dir) = test_web_state().await; + Arc::get_mut(&mut state).unwrap().demo_mode = true; + let req = transfer_request("src", "dst", &dir); + let transfer_id = req.transfer_id.clone(); + let _ = start_transfer(State(state.clone()), Json(StartTransferRequest { request: req })).await.unwrap(); + let channel = { + let channels = state.transfer_progress_channels.read().await; + channels.get(&transfer_id).cloned().expect("transfer channel registered") + }; + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + if channel + .latest() + .and_then(|data| serde_json::from_str::(&data).ok()) + .is_some_and(|progress| progress["terminal"].as_bool() == Some(true)) + { + break; + } + if std::time::Instant::now() > deadline { + panic!("demo transfer did not reach a terminal event"); + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + assert!(state.app.storage.get_task_run_detail(&transfer_id).await.unwrap().is_none()); let _ = std::fs::remove_dir_all(&dir); } } diff --git a/crates/dbx-web/src/sse.rs b/crates/dbx-web/src/sse.rs index 43373e7a94..8158a02855 100644 --- a/crates/dbx-web/src/sse.rs +++ b/crates/dbx-web/src/sse.rs @@ -1,4 +1,6 @@ use axum::response::sse::{Event, KeepAlive, Sse}; +use dbx_core::persistence::task_history::TransferTaskJournal; +use dbx_core::transfer::TransferProgress; use futures::stream::Stream; use serde_json::Value; use std::collections::VecDeque; @@ -22,6 +24,8 @@ struct TransferReplayHistory { failure_bytes: usize, latest: Option, omitted_failures: usize, + task_history_journal: Option, + terminal_progress: Option, } pub struct TransferProgressChannel { @@ -35,6 +39,24 @@ impl TransferProgressChannel { Self { tx, history: Mutex::new(TransferReplayHistory::default()) } } + pub fn set_task_history_journal(&self, journal: Option) { + self.history.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).task_history_journal = journal; + } + + pub fn remember_terminal_progress(&self, progress: TransferProgress) { + self.history.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).terminal_progress = Some(progress); + } + + pub async fn finish_task_history(&self) { + let (journal, progress) = { + let history = self.history.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + (history.task_history_journal.clone(), history.terminal_progress.clone()) + }; + if let (Some(journal), Some(progress)) = (journal, progress) { + journal.finish(&progress).await; + } + } + pub fn send(&self, data: String, kind: TransferReplayEventKind) { let mut history = self.history.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); match kind { diff --git a/src-tauri/src/commands/mod.rs b/src-tauri/src/commands/mod.rs index 5f97efa5c6..018764dc3a 100644 --- a/src-tauri/src/commands/mod.rs +++ b/src-tauri/src/commands/mod.rs @@ -64,6 +64,7 @@ pub mod system_fonts; pub mod tab_runtime_cache; pub mod table_export; pub mod table_import; +pub mod task_history; pub mod text_export; pub mod transfer; pub mod tunnel_profiles; diff --git a/src-tauri/src/commands/task_history.rs b/src-tauri/src/commands/task_history.rs new file mode 100644 index 0000000000..08c77052eb --- /dev/null +++ b/src-tauri/src/commands/task_history.rs @@ -0,0 +1,27 @@ +use std::sync::Arc; + +use tauri::State; + +use super::connection::AppState; +use dbx_core::persistence::task_history::{ + TaskRunDetail, TaskRunItemsPage, TaskRunItemsQuery, TaskRunListQuery, TaskRunPage, +}; + +#[tauri::command] +pub async fn load_task_runs(state: State<'_, Arc>, query: TaskRunListQuery) -> Result { + state.storage.list_task_runs(query).await.map_err(|error| error.code().to_string()) +} + +#[tauri::command] +pub async fn load_task_run(state: State<'_, Arc>, run_id: String) -> Result, String> { + state.storage.get_task_run_detail(&run_id).await.map_err(|error| error.code().to_string()) +} + +#[tauri::command] +pub async fn load_task_run_items( + state: State<'_, Arc>, + run_id: String, + query: TaskRunItemsQuery, +) -> Result { + state.storage.list_task_run_items(&run_id, query).await.map_err(|error| error.code().to_string()) +} diff --git a/src-tauri/src/commands/transfer.rs b/src-tauri/src/commands/transfer.rs index 4bb090f05b..3fa47bab80 100644 --- a/src-tauri/src/commands/transfer.rs +++ b/src-tauri/src/commands/transfer.rs @@ -5,6 +5,9 @@ use crate::commands::connection::{ensure_connection_writable, AppState}; // Re-export types and functions used by other modules use dbx_core::models::connection::DatabaseType; +use dbx_core::persistence::task_history::{ + TaskHistoryStorageError, TaskItemStatus, TaskLifecycleOwner, TransferTaskJournal, +}; pub use dbx_core::transfer::{ get_db_type, TransferOwnershipPreview, TransferProgress, TransferRequest, TransferStatus, }; @@ -13,6 +16,13 @@ fn emit_progress(app: &AppHandle, progress: TransferProgress) { let _ = app.emit("transfer-progress", progress); } +async fn emit_terminal_progress(app: &AppHandle, history: Option<&TransferTaskJournal>, progress: TransferProgress) { + if let Some(history) = history { + history.finish(&progress).await; + } + emit_progress(app, progress); +} + #[tauri::command] pub async fn start_transfer( app: AppHandle, @@ -70,6 +80,12 @@ pub async fn start_transfer( dbx_core::transfer::ensure_transfer_source_types_supported(&state, &request, &source_pool_key).await?; + let history = match TransferTaskJournal::accept(&state.storage, &state, &request, TaskLifecycleOwner::Tauri).await { + Ok(history) => history, + Err(TaskHistoryStorageError::RunIdConflict) => return Err("TRANSFER_RUN_ID_CONFLICT".to_string()), + Err(error) => return Err(error.code().to_string()), + }; + 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 @@ -157,8 +173,9 @@ pub async fn start_transfer( { Ok(names) => Some(names), Err(e) if e == "Cancelled" => { - emit_progress( + emit_terminal_progress( &app, + history.as_ref(), TransferProgress { transfer_id: transfer_id.clone(), table: "rename pre-pass".to_string(), @@ -170,13 +187,15 @@ pub async fn start_transfer( error: None, terminal: true, }, - ); + ) + .await; dbx_core::transfer::clear_cancelled(&transfer_id).await; return; } Err(e) => { - emit_progress( + emit_terminal_progress( &app, + history.as_ref(), TransferProgress { transfer_id: transfer_id.clone(), table: "rename pre-pass".to_string(), @@ -188,7 +207,8 @@ pub async fn start_transfer( error: Some(e), terminal: true, }, - ); + ) + .await; dbx_core::transfer::clear_cancelled(&transfer_id).await; return; } @@ -215,8 +235,9 @@ pub async fn start_transfer( { Ok(()) => {} Err(e) if e == "Cancelled" => { - emit_progress( + emit_terminal_progress( &app, + history.as_ref(), TransferProgress { transfer_id: transfer_id.clone(), table: "schema dependencies".to_string(), @@ -228,13 +249,15 @@ pub async fn start_transfer( error: None, terminal: true, }, - ); + ) + .await; dbx_core::transfer::clear_cancelled(&transfer_id).await; return; } Err(e) => { - emit_progress( + emit_terminal_progress( &app, + history.as_ref(), TransferProgress { transfer_id: transfer_id.clone(), table: "schema dependencies".to_string(), @@ -246,7 +269,8 @@ pub async fn start_transfer( error: Some(e), terminal: true, }, - ); + ) + .await; dbx_core::transfer::clear_cancelled(&transfer_id).await; return; } @@ -254,8 +278,9 @@ pub async fn start_transfer( } for (i, table) in sorted_tables.iter().enumerate() { if dbx_core::transfer::is_cancelled(&transfer_id).await { - emit_progress( + emit_terminal_progress( &app, + history.as_ref(), TransferProgress { transfer_id: transfer_id.clone(), table: table.clone(), @@ -267,14 +292,23 @@ pub async fn start_transfer( error: None, terminal: true, }, - ); + ) + .await; dbx_core::transfer::clear_cancelled(&transfer_id).await; return; } log::info!("[transfer] table {}/{}: {}", i + 1, total_tables, table); - - match dbx_core::transfer::transfer_table( + let history_item_index = + history.as_ref().and_then(|journal| journal.item_index_for_table(table)).unwrap_or(i); + if let Some(journal) = history.as_ref() { + journal.start_table(history_item_index).await; + } + let mut source_row_count = None; + let mut moved_row_count = None; + let history_for_progress = history.as_ref(); + let history_for_count = history.as_ref(); + let result = dbx_core::transfer::transfer_table_with_result( &state, &request, table, @@ -289,12 +323,32 @@ pub async fn start_transfer( |progress| { last_rows_transferred = progress.rows_transferred; last_total_rows = progress.total_rows; + moved_row_count = Some(progress.rows_transferred); + if let Some(journal) = history_for_progress { + journal.observe_table_progress(history_item_index, progress.rows_transferred); + } emit_progress(&app, progress); }, + |source_count| { + source_row_count = source_count; + if let Some(journal) = history_for_count { + journal.observe_source_count(history_item_index, source_count); + } + }, ) - .await - { - Ok(rows) => { + .await; + match result { + Ok(result) => { + if let Some(journal) = history.as_ref() { + journal + .finish_table( + history_item_index, + TaskItemStatus::Succeeded, + result.source_row_count, + Some(result.moved_rows), + ) + .await; + } emit_progress( &app, TransferProgress { @@ -302,8 +356,8 @@ pub async fn start_transfer( table: table.clone(), table_index: i, total_tables, - rows_transferred: rows, - total_rows: last_total_rows.or(Some(rows)), + rows_transferred: result.moved_rows, + total_rows: last_total_rows.or(Some(result.moved_rows)), status: TransferStatus::TableDone, error: None, terminal: false, @@ -312,23 +366,40 @@ pub async fn start_transfer( } Err(e) => { if e == "Cancelled" { - emit_progress( + if let Some(journal) = history.as_ref() { + journal + .finish_table( + history_item_index, + TaskItemStatus::Cancelled, + source_row_count, + moved_row_count, + ) + .await; + } + emit_terminal_progress( &app, + history.as_ref(), TransferProgress { transfer_id: transfer_id.clone(), table: table.clone(), table_index: i, total_tables, - rows_transferred: last_rows_transferred, - total_rows: last_total_rows, + rows_transferred: 0, + total_rows: None, status: TransferStatus::Cancelled, error: None, terminal: true, }, - ); + ) + .await; dbx_core::transfer::clear_cancelled(&transfer_id).await; return; } + if let Some(journal) = history.as_ref() { + journal + .finish_table(history_item_index, TaskItemStatus::Failed, source_row_count, moved_row_count) + .await; + } failed_tables.push(table.clone()); emit_progress( &app, @@ -385,11 +456,18 @@ pub async fn start_transfer( .await { Ok(outcome) => { + if let Some(journal) = history.as_ref() { + journal.record_object_outcome(&outcome).await; + } object_outcome = outcome; } Err(e) if e == "Cancelled" => { - emit_progress( + if let Some(journal) = history.as_ref() { + journal.record_schema_objects_error(true).await; + } + emit_terminal_progress( &app, + history.as_ref(), TransferProgress { transfer_id: transfer_id.clone(), table: "schema objects".to_string(), @@ -401,11 +479,15 @@ pub async fn start_transfer( error: None, terminal: true, }, - ); + ) + .await; dbx_core::transfer::clear_cancelled(&transfer_id).await; return; } Err(e) => { + if let Some(journal) = history.as_ref() { + journal.record_schema_objects_error(false).await; + } failed_tables.push("schema objects".to_string()); emit_progress( &app, @@ -461,8 +543,9 @@ pub async fn start_transfer( String::new() }; - emit_progress( + emit_terminal_progress( &app, + history.as_ref(), TransferProgress { transfer_id: transfer_id.clone(), table: String::new(), @@ -487,7 +570,8 @@ pub async fn start_transfer( }, terminal: true, }, - ); + ) + .await; dbx_core::transfer::clear_cancelled(&transfer_id).await; }); diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 54429caf3e..24e84eef3d 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -2583,6 +2583,9 @@ pub fn run() { commands::history::clear_history_by_source, commands::history::cleanup_mcp_history_retention, commands::history::delete_history_entry, + commands::task_history::load_task_runs, + commands::task_history::load_task_run, + commands::task_history::load_task_run_items, commands::mcp::check_mcp_server_status, commands::mcp::install_mcp_server, commands::mcp::install_native_mcp_server,