feat(export-tracker): 增强导出任务跟踪功能 (#1958)
- 添加了对数据库导出、SQL 文件导出和数据传输任务的支持 - 引入了新的状态和错误处理机制 - 实现了任务的重叠检查,防止同时进行相同目标表的数据传输 - 更新了 API 调用以支持新的任务类型 - 增加了相应的国际化文本 - 添加了单元测试以确保新功能的正确性
This commit is contained in:
15 files changed
+838
-208
No files matched your search
@@ -1,18 +1,79 @@
|
||||
import { reactive, computed } from "vue";
|
||||
import * as api from "@/lib/api";
|
||||
|
||||
export type BackgroundTaskKind = "table-export" | "database-export" | "sql-file" | "data-transfer";
|
||||
export type BackgroundTaskStatus = "Running" | "Writing" | "Done" | "Error" | "Cancelled";
|
||||
|
||||
export interface ExportTask {
|
||||
exportId: string;
|
||||
kind: BackgroundTaskKind;
|
||||
tableName: string;
|
||||
format: "csv" | "xlsx";
|
||||
format: string;
|
||||
filePath: string;
|
||||
rowsExported: number;
|
||||
totalRows: number | null;
|
||||
status: string;
|
||||
status: BackgroundTaskStatus;
|
||||
errorMessage: string | null;
|
||||
objectIndex?: number;
|
||||
totalObjects?: number;
|
||||
statementIndex?: number;
|
||||
successCount?: number;
|
||||
failureCount?: number;
|
||||
affectedRows?: number;
|
||||
elapsedMs?: number;
|
||||
statementSummary?: string;
|
||||
tableIndex?: number;
|
||||
totalTables?: number;
|
||||
currentTable?: string;
|
||||
targetConnectionId?: string;
|
||||
targetDatabase?: string;
|
||||
targetSchema?: string;
|
||||
targetTables?: string[];
|
||||
}
|
||||
|
||||
const taskMap = reactive<Map<string, ExportTask>>(new Map());
|
||||
const activeTransferRuns = new Set<string>();
|
||||
|
||||
function normalizeExportStatus(status: string): BackgroundTaskStatus {
|
||||
if (status === "Writing" || status === "Done" || status === "Error" || status === "Cancelled") return status;
|
||||
return "Running";
|
||||
}
|
||||
|
||||
function normalizeSqlFileStatus(status: api.SqlFileStatus): BackgroundTaskStatus {
|
||||
if (status === "done") return "Done";
|
||||
if (status === "error") return "Error";
|
||||
if (status === "cancelled") return "Cancelled";
|
||||
return "Running";
|
||||
}
|
||||
|
||||
function normalizeTransferStatus(status: api.TransferProgress["status"]): BackgroundTaskStatus {
|
||||
if (status === "done") return "Done";
|
||||
if (status === "error") return "Error";
|
||||
if (status === "cancelled") return "Cancelled";
|
||||
return "Running";
|
||||
}
|
||||
|
||||
function targetTableName(table: string, nameCase: api.TransferTableNameCase): string {
|
||||
if (nameCase === "lower") return table.toLowerCase();
|
||||
if (nameCase === "upper") return table.toUpperCase();
|
||||
return table;
|
||||
}
|
||||
|
||||
function findActiveOverlappingTransfer(request: api.TransferRequest): string[] {
|
||||
const requestedTables = new Set(request.tables.map((table) => targetTableName(table, request.targetTableNameCase)));
|
||||
for (const task of taskMap.values()) {
|
||||
if (task.exportId === request.transferId) continue;
|
||||
if (task.kind !== "data-transfer") continue;
|
||||
if (task.status !== "Running" && task.status !== "Writing") continue;
|
||||
if (task.targetConnectionId !== request.targetConnectionId) continue;
|
||||
if (task.targetDatabase !== request.targetDatabase) continue;
|
||||
if (task.targetSchema !== request.targetSchema) continue;
|
||||
|
||||
const overlappingTables = (task.targetTables ?? []).filter((table) => requestedTables.has(table));
|
||||
if (overlappingTables.length > 0) return overlappingTables;
|
||||
}
|
||||
return [];
|
||||
}
|
||||
|
||||
export function useExportTracker() {
|
||||
const tasks = computed(() => Array.from(taskMap.values()));
|
||||
@@ -33,10 +94,11 @@ export function useExportTracker() {
|
||||
.replace(/^(.{8})(.{4})(.{4})(.{4})(.{12})$/, "$1-$2-$3-$4-$5");
|
||||
}
|
||||
|
||||
function addTask(tableName: string, format: "csv" | "xlsx", filePath: string): ExportTask {
|
||||
function addTask(tableName: string, format: string, filePath: string): ExportTask {
|
||||
const exportId = generateUUID();
|
||||
const task = reactive<ExportTask>({
|
||||
exportId,
|
||||
kind: "table-export",
|
||||
tableName,
|
||||
format,
|
||||
filePath,
|
||||
@@ -49,6 +111,174 @@ export function useExportTracker() {
|
||||
return task;
|
||||
}
|
||||
|
||||
function addDatabaseExportTask(exportId: string, label: string, filePath: string): ExportTask {
|
||||
const task = reactive<ExportTask>({
|
||||
exportId,
|
||||
kind: "database-export",
|
||||
tableName: label,
|
||||
format: "sql",
|
||||
filePath,
|
||||
rowsExported: 0,
|
||||
totalRows: null,
|
||||
status: "Running",
|
||||
errorMessage: null,
|
||||
objectIndex: 0,
|
||||
totalObjects: 0,
|
||||
});
|
||||
taskMap.set(exportId, task);
|
||||
return task;
|
||||
}
|
||||
|
||||
function addSqlFileTask(executionId: string, fileName: string, filePath: string): ExportTask {
|
||||
const task = reactive<ExportTask>({
|
||||
exportId: executionId,
|
||||
kind: "sql-file",
|
||||
tableName: fileName,
|
||||
format: "sql",
|
||||
filePath,
|
||||
rowsExported: 0,
|
||||
totalRows: null,
|
||||
status: "Running",
|
||||
errorMessage: null,
|
||||
statementIndex: 0,
|
||||
successCount: 0,
|
||||
failureCount: 0,
|
||||
affectedRows: 0,
|
||||
elapsedMs: 0,
|
||||
statementSummary: "",
|
||||
});
|
||||
taskMap.set(executionId, task);
|
||||
return task;
|
||||
}
|
||||
|
||||
function addDataTransferTask(transferId: string, label: string, totalTables: number): ExportTask {
|
||||
const task = reactive<ExportTask>({
|
||||
exportId: transferId,
|
||||
kind: "data-transfer",
|
||||
tableName: label,
|
||||
format: "transfer",
|
||||
filePath: "",
|
||||
rowsExported: 0,
|
||||
totalRows: null,
|
||||
status: "Running",
|
||||
errorMessage: null,
|
||||
tableIndex: 0,
|
||||
totalTables,
|
||||
currentTable: "",
|
||||
});
|
||||
taskMap.set(transferId, task);
|
||||
return task;
|
||||
}
|
||||
|
||||
function startDataTransferTask(
|
||||
request: api.TransferRequest,
|
||||
label: string,
|
||||
options: {
|
||||
onDone?: () => void | Promise<void>;
|
||||
formatOverlapError?: (tables: string[]) => string;
|
||||
} = {},
|
||||
): ExportTask {
|
||||
const existingTask = taskMap.get(request.transferId);
|
||||
const task = existingTask ?? addDataTransferTask(request.transferId, label, request.tables.length);
|
||||
task.targetConnectionId = request.targetConnectionId;
|
||||
task.targetDatabase = request.targetDatabase;
|
||||
task.targetSchema = request.targetSchema;
|
||||
task.targetTables = request.tables.map((table) => targetTableName(table, request.targetTableNameCase));
|
||||
if (activeTransferRuns.has(request.transferId)) return task;
|
||||
|
||||
const overlappingTables = findActiveOverlappingTransfer(request);
|
||||
if (overlappingTables.length > 0) {
|
||||
task.status = "Error";
|
||||
const visibleTables = overlappingTables.slice(0, 5);
|
||||
task.errorMessage = options.formatOverlapError?.(visibleTables) ?? `Another data transfer is already running for target table(s): ${visibleTables.join(", ")}`;
|
||||
return task;
|
||||
}
|
||||
|
||||
activeTransferRuns.add(request.transferId);
|
||||
let terminalStatus: api.TransferProgress["status"] | null = null;
|
||||
|
||||
void (async () => {
|
||||
try {
|
||||
await api.startTransfer(request, (progress) => {
|
||||
terminalStatus = progress.status === "done" || progress.status === "error" || progress.status === "cancelled" ? progress.status : terminalStatus;
|
||||
updateDataTransferTask(progress.transferId, progress);
|
||||
});
|
||||
|
||||
if (terminalStatus === "done" && task.status === "Done") {
|
||||
await options.onDone?.();
|
||||
}
|
||||
} catch (e: any) {
|
||||
updateDataTransferTask(request.transferId, {
|
||||
transferId: request.transferId,
|
||||
table: task.currentTable || "",
|
||||
tableIndex: task.tableIndex ?? 0,
|
||||
totalTables: task.totalTables ?? request.tables.length,
|
||||
rowsTransferred: task.rowsExported,
|
||||
totalRows: task.totalRows,
|
||||
status: "error",
|
||||
error: e?.message || String(e),
|
||||
});
|
||||
} finally {
|
||||
activeTransferRuns.delete(request.transferId);
|
||||
}
|
||||
})();
|
||||
|
||||
return task;
|
||||
}
|
||||
|
||||
function updateTableExportTask(exportId: string, progress: api.TableExportProgress) {
|
||||
const task = taskMap.get(exportId);
|
||||
if (!task) return;
|
||||
task.tableName = progress.tableName || task.tableName;
|
||||
task.rowsExported = progress.rowsExported;
|
||||
task.totalRows = progress.totalRows;
|
||||
task.status = normalizeExportStatus(progress.status);
|
||||
task.errorMessage = progress.errorMessage || null;
|
||||
}
|
||||
|
||||
function updateDatabaseExportTask(exportId: string, progress: api.ExportProgress) {
|
||||
const task = taskMap.get(exportId);
|
||||
if (!task) return;
|
||||
task.tableName = progress.currentObject || task.tableName;
|
||||
task.rowsExported = progress.rowsExported;
|
||||
task.totalRows = progress.totalRows;
|
||||
task.status = normalizeExportStatus(progress.status);
|
||||
task.errorMessage = progress.error || null;
|
||||
task.objectIndex = progress.objectIndex;
|
||||
task.totalObjects = progress.totalObjects;
|
||||
}
|
||||
|
||||
function updateSqlFileTask(executionId: string, progress: api.SqlFileProgress) {
|
||||
const task = taskMap.get(executionId);
|
||||
if (!task) return;
|
||||
task.status = normalizeSqlFileStatus(progress.status);
|
||||
task.errorMessage = progress.error || null;
|
||||
task.statementIndex = progress.statementIndex;
|
||||
task.successCount = progress.successCount;
|
||||
task.failureCount = progress.failureCount;
|
||||
task.affectedRows = progress.affectedRows;
|
||||
task.elapsedMs = progress.elapsedMs;
|
||||
task.statementSummary = progress.statementSummary;
|
||||
task.rowsExported = progress.successCount + progress.failureCount;
|
||||
task.totalRows = Math.max(progress.statementIndex, progress.successCount + progress.failureCount) || null;
|
||||
}
|
||||
|
||||
function updateDataTransferTask(transferId: string, progress: api.TransferProgress) {
|
||||
const task = taskMap.get(transferId);
|
||||
if (!task) return;
|
||||
const nextStatus = normalizeTransferStatus(progress.status);
|
||||
const hadError = task.status === "Error";
|
||||
task.status = hadError && nextStatus === "Done" ? "Error" : nextStatus;
|
||||
task.errorMessage = progress.error || task.errorMessage || null;
|
||||
task.tableIndex = progress.tableIndex;
|
||||
task.totalTables = progress.totalTables;
|
||||
task.currentTable = progress.table || task.currentTable;
|
||||
if (progress.table || progress.rowsTransferred > 0 || task.rowsExported === 0) {
|
||||
task.rowsExported = progress.rowsTransferred;
|
||||
}
|
||||
task.totalRows = progress.totalRows ?? task.totalRows;
|
||||
}
|
||||
|
||||
function removeTask(exportId: string) {
|
||||
taskMap.delete(exportId);
|
||||
}
|
||||
@@ -62,8 +292,17 @@ export function useExportTracker() {
|
||||
}
|
||||
|
||||
async function cancelTask(exportId: string) {
|
||||
const task = taskMap.get(exportId);
|
||||
try {
|
||||
await api.cancelTableExport(exportId);
|
||||
if (task?.kind === "database-export") {
|
||||
await api.cancelDatabaseExport(exportId);
|
||||
} else if (task?.kind === "sql-file") {
|
||||
await api.cancelSqlFileExecution(exportId);
|
||||
} else if (task?.kind === "data-transfer") {
|
||||
await api.cancelTransfer(exportId);
|
||||
} else {
|
||||
await api.cancelTableExport(exportId);
|
||||
}
|
||||
} catch {
|
||||
// ignore
|
||||
}
|
||||
@@ -74,6 +313,14 @@ export function useExportTracker() {
|
||||
activeCount,
|
||||
hasActive,
|
||||
addTask,
|
||||
addDatabaseExportTask,
|
||||
addSqlFileTask,
|
||||
addDataTransferTask,
|
||||
startDataTransferTask,
|
||||
updateTableExportTask,
|
||||
updateDatabaseExportTask,
|
||||
updateSqlFileTask,
|
||||
updateDataTransferTask,
|
||||
removeTask,
|
||||
clearFinished,
|
||||
cancelTask,
|
||||
|
||||
Reference in new issue
Block a user