feat(compare): keep schema and data compares running in background
This commit is contained in:
jearhe authored and GitHub committed
1 parent
29c6d1fbed
commit
dc7b905220
15 files changed
+1643
-535
No files matched your search
@@ -0,0 +1,96 @@
|
||||
import { strict as assert } from "node:assert";
|
||||
import { beforeEach, test, vi } from "vitest";
|
||||
|
||||
const apiMock = vi.hoisted(() => ({
|
||||
buildDataCompareSyncPlan: vi.fn(),
|
||||
getColumns: vi.fn(),
|
||||
prepareDataCompareFromTables: vi.fn(),
|
||||
prepareDataCompareMissingTarget: vi.fn(),
|
||||
}));
|
||||
|
||||
const openMock = vi.hoisted(() => vi.fn());
|
||||
const trackerMock = vi.hoisted(() => ({
|
||||
addDataCompareTask: vi.fn(),
|
||||
updateCompareTask: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/backend/api", () => apiMock);
|
||||
vi.mock("@/composables/useDialogSources", () => ({ openDataCompareSession: openMock }));
|
||||
vi.mock("@/composables/useExportTracker", () => ({ useExportTracker: () => trackerMock }));
|
||||
|
||||
const { startDataCompareSession } = await import("../useDataCompareSession.ts");
|
||||
|
||||
test("runs a data compare session independently of the dialog and retains its result", async () => {
|
||||
apiMock.getColumns.mockResolvedValue([
|
||||
{ name: "id", is_primary_key: true },
|
||||
{ name: "name", is_primary_key: false },
|
||||
]);
|
||||
apiMock.prepareDataCompareFromTables.mockResolvedValue({
|
||||
result: {
|
||||
added: [{ key: "1", keyValues: { id: 1 }, values: { id: 1, name: "new" } }],
|
||||
removed: [],
|
||||
modified: [],
|
||||
},
|
||||
syncStatements: ["INSERT INTO users (id, name) VALUES (1, 'new')"],
|
||||
syncSql: "INSERT INTO users (id, name) VALUES (1, 'new')",
|
||||
preSyncStatements: [],
|
||||
sourceRowCount: 1,
|
||||
targetRowCount: 0,
|
||||
sourceTruncated: false,
|
||||
targetTruncated: false,
|
||||
});
|
||||
apiMock.buildDataCompareSyncPlan.mockResolvedValue({
|
||||
insertCount: 1,
|
||||
updateCount: 0,
|
||||
deleteCount: 0,
|
||||
statementCount: 1,
|
||||
syncStatements: ["INSERT"],
|
||||
syncSql: "INSERT",
|
||||
});
|
||||
|
||||
const session = startDataCompareSession(
|
||||
{
|
||||
sourceConnectionId: "source",
|
||||
sourceDatabase: "app",
|
||||
sourceSchema: "public",
|
||||
sourceDatabases: ["app"],
|
||||
sourceSchemas: ["public"],
|
||||
sourceTables: ["users"],
|
||||
selectedSourceTables: ["users"],
|
||||
targetConnectionId: "target",
|
||||
targetDatabase: "warehouse",
|
||||
targetSchema: "public",
|
||||
targetDatabases: ["warehouse"],
|
||||
targetSchemas: ["public"],
|
||||
targetTables: ["users"],
|
||||
targetTable: "users",
|
||||
keyColumns: ["id"],
|
||||
label: "app → warehouse",
|
||||
},
|
||||
[{ sourceTable: "users", targetTable: "users" }],
|
||||
{
|
||||
ensureConnected: vi.fn().mockResolvedValue(undefined),
|
||||
getConfig: () => ({ db_type: "mysql" }),
|
||||
},
|
||||
);
|
||||
|
||||
for (let attempt = 0; attempt < 20 && session.status === "running"; attempt += 1) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
}
|
||||
|
||||
assert.equal(session.status, "completed");
|
||||
assert.equal(session.batchResults.length, 1);
|
||||
assert.equal(session.batchResults[0]?.status, "different");
|
||||
assert.equal(session.batchResults[0]?.diff.added[0]?.selected, true);
|
||||
assert.equal(session.syncPlan.statementCount, 1);
|
||||
assert.equal(trackerMock.addDataCompareTask.mock.calls.length, 1);
|
||||
assert.equal(trackerMock.updateCompareTask.mock.calls.at(-1)?.[1].status, "Done");
|
||||
|
||||
const onOpen = trackerMock.addDataCompareTask.mock.calls[0]?.[2] as (() => void) | undefined;
|
||||
onOpen?.();
|
||||
assert.equal(openMock.mock.calls.at(-1)?.[0], session.id);
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
@@ -0,0 +1,66 @@
|
||||
import { strict as assert } from "node:assert";
|
||||
import { test, vi } from "vitest";
|
||||
|
||||
const apiMock = vi.hoisted(() => ({
|
||||
prepareSchemaDiff: vi.fn(),
|
||||
}));
|
||||
const openMock = vi.hoisted(() => vi.fn());
|
||||
const trackerMock = vi.hoisted(() => ({
|
||||
addSchemaDiffTask: vi.fn(),
|
||||
updateCompareTask: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/backend/api", () => apiMock);
|
||||
vi.mock("@/composables/useDialogSources", () => ({ openSchemaDiffSession: openMock }));
|
||||
vi.mock("@/composables/useExportTracker", () => ({ useExportTracker: () => trackerMock }));
|
||||
vi.mock("@/lib/schema/schemaDiffMetadataLoad", () => ({ loadSchemaDetails: vi.fn().mockResolvedValue([]) }));
|
||||
|
||||
const { startSchemaDiffSession } = await import("../useSchemaDiffSession.ts");
|
||||
|
||||
test("runs a schema diff session after the dialog is closed and retains the prepared result", async () => {
|
||||
apiMock.prepareSchemaDiff.mockResolvedValue({
|
||||
diffs: [],
|
||||
functionDiffs: [],
|
||||
sequenceDiffs: [],
|
||||
ruleDiffs: [],
|
||||
ownerDiffs: [],
|
||||
renameCandidates: [],
|
||||
syncSql: "",
|
||||
rollbackSyncSql: "",
|
||||
});
|
||||
|
||||
const tableListLoader = {
|
||||
load: vi.fn().mockResolvedValue([]),
|
||||
};
|
||||
const session = startSchemaDiffSession(
|
||||
{
|
||||
sourceConnectionId: "source",
|
||||
sourceDatabase: "app",
|
||||
sourceSchema: "public",
|
||||
targetConnectionId: "target",
|
||||
targetDatabase: "warehouse",
|
||||
targetSchema: "public",
|
||||
sourceDbType: "mysql",
|
||||
targetDbType: "mysql",
|
||||
options: {},
|
||||
ignoreComments: false,
|
||||
label: "app → warehouse",
|
||||
},
|
||||
{ tableListLoader },
|
||||
);
|
||||
|
||||
for (let attempt = 0; attempt < 20 && session.status === "running"; attempt += 1) {
|
||||
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||
}
|
||||
|
||||
assert.equal(session.status, "completed");
|
||||
assert.deepEqual(session.result?.diffs, []);
|
||||
assert.equal(tableListLoader.load.mock.calls.length, 2);
|
||||
assert.equal(tableListLoader.load.mock.calls[0]?.[1]?.refresh, true);
|
||||
assert.equal(trackerMock.addSchemaDiffTask.mock.calls.length, 1);
|
||||
assert.equal(trackerMock.updateCompareTask.mock.calls.at(-1)?.[1].status, "Done");
|
||||
|
||||
const onOpen = trackerMock.addSchemaDiffTask.mock.calls[0]?.[2] as (() => void) | undefined;
|
||||
onOpen?.();
|
||||
assert.equal(openMock.mock.calls.at(-1)?.[0], session.id);
|
||||
});
|
||||
@@ -0,0 +1,436 @@
|
||||
import { reactive, shallowReactive } from "vue";
|
||||
import { uuid } from "@/lib/common/utils";
|
||||
import * as api from "@/lib/backend/api";
|
||||
import { openDataCompareSession } from "@/composables/useDialogSources";
|
||||
import { useExportTracker } from "@/composables/useExportTracker";
|
||||
import { inferCompareKeyColumns, intersectCompareColumns, matchColumnNameIgnoreCase, type DataCompareCellValue, type DataCompareFromTablesPreparation, type DataCompareResult, type DataCompareSyncPlan, type DataCompareSyncPlanTableOptions } from "@/lib/dataGrid/dataCompare";
|
||||
import type { ColumnInfo, DatabaseType } from "@/types/database";
|
||||
|
||||
export type CompareColumn = ColumnInfo;
|
||||
|
||||
export interface DataCompareTableTask {
|
||||
sourceTable: string;
|
||||
targetTable: string;
|
||||
}
|
||||
|
||||
export type DataCompareTableStatus = "different" | "same" | "error";
|
||||
export type DiffKind = "added" | "removed" | "modified";
|
||||
|
||||
export interface SelectableDataCompareRow {
|
||||
key: string;
|
||||
keyValues: Record<string, DataCompareCellValue>;
|
||||
values: Record<string, DataCompareCellValue>;
|
||||
selected: boolean;
|
||||
}
|
||||
|
||||
export interface SelectableDataCompareModifiedRow {
|
||||
key: string;
|
||||
keyValues: Record<string, DataCompareCellValue>;
|
||||
sourceValues: Record<string, DataCompareCellValue>;
|
||||
targetValues: Record<string, DataCompareCellValue>;
|
||||
changes: { column: string; source: DataCompareCellValue; target: DataCompareCellValue }[];
|
||||
selected: boolean;
|
||||
}
|
||||
|
||||
export interface SelectableDataCompareResult {
|
||||
added: SelectableDataCompareRow[];
|
||||
removed: SelectableDataCompareRow[];
|
||||
modified: SelectableDataCompareModifiedRow[];
|
||||
}
|
||||
|
||||
export interface DataCompareTableResult {
|
||||
sourceTable: string;
|
||||
targetTable: string;
|
||||
keyColumns: string[];
|
||||
columns: string[];
|
||||
columnInfo: CompareColumn[];
|
||||
status: DataCompareTableStatus;
|
||||
added: number;
|
||||
removed: number;
|
||||
modified: number;
|
||||
sourceRowCount: number;
|
||||
targetRowCount: number;
|
||||
sourceTruncated: boolean;
|
||||
targetTruncated: boolean;
|
||||
databaseType?: DatabaseType;
|
||||
preSyncStatements?: string[];
|
||||
diff: SelectableDataCompareResult;
|
||||
expanded: boolean;
|
||||
showAll: Record<DiffKind, boolean>;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
export interface DataCompareSessionConfig {
|
||||
sourceConnectionId: string;
|
||||
sourceDatabase: string;
|
||||
sourceSchema: string;
|
||||
sourceDatabases: string[];
|
||||
sourceSchemas: string[];
|
||||
sourceTables: string[];
|
||||
selectedSourceTables: string[];
|
||||
targetConnectionId: string;
|
||||
targetDatabase: string;
|
||||
targetSchema: string;
|
||||
targetDatabases: string[];
|
||||
targetSchemas: string[];
|
||||
targetTables: string[];
|
||||
targetTable: string;
|
||||
keyColumns: string[];
|
||||
label: string;
|
||||
}
|
||||
|
||||
export interface DataCompareSessionProgress {
|
||||
current: number;
|
||||
total: number;
|
||||
table: string;
|
||||
}
|
||||
|
||||
export type DataCompareSessionStatus = "running" | "completed" | "failed";
|
||||
export type DataCompareSessionErrorKind = "noKeyColumns" | "missingKeyColumns" | "noCommonColumns";
|
||||
|
||||
export interface DataCompareSession {
|
||||
id: string;
|
||||
version: number;
|
||||
status: DataCompareSessionStatus;
|
||||
config: DataCompareSessionConfig;
|
||||
progress: DataCompareSessionProgress | null;
|
||||
batchResults: DataCompareTableResult[];
|
||||
syncPlan: DataCompareSyncPlan;
|
||||
error: string | null;
|
||||
startedAt: number;
|
||||
finishedAt?: number;
|
||||
}
|
||||
|
||||
export interface DataCompareSessionDependencies {
|
||||
ensureConnected(connectionId: string): Promise<unknown>;
|
||||
getConfig(connectionId: string): { db_type?: DatabaseType } | undefined;
|
||||
formatError?: (kind: DataCompareSessionErrorKind, columns?: string) => string;
|
||||
}
|
||||
|
||||
const sessions = reactive(new Map<string, DataCompareSession>());
|
||||
|
||||
export function emptyDataCompareSyncPlan(): DataCompareSyncPlan {
|
||||
return {
|
||||
insertCount: 0,
|
||||
updateCount: 0,
|
||||
deleteCount: 0,
|
||||
statementCount: 0,
|
||||
syncStatements: [],
|
||||
syncSql: "",
|
||||
};
|
||||
}
|
||||
|
||||
function cloneConfig(config: DataCompareSessionConfig): DataCompareSessionConfig {
|
||||
return {
|
||||
...config,
|
||||
sourceDatabases: [...config.sourceDatabases],
|
||||
sourceSchemas: [...config.sourceSchemas],
|
||||
sourceTables: [...config.sourceTables],
|
||||
selectedSourceTables: [...config.selectedSourceTables],
|
||||
targetDatabases: [...config.targetDatabases],
|
||||
targetSchemas: [...config.targetSchemas],
|
||||
targetTables: [...config.targetTables],
|
||||
keyColumns: [...config.keyColumns],
|
||||
};
|
||||
}
|
||||
|
||||
function sessionError(error: unknown): string {
|
||||
return error instanceof Error ? error.message : (error as { message?: string } | null)?.message || String(error);
|
||||
}
|
||||
|
||||
function compareErrorMessage(dependencies: DataCompareSessionDependencies, kind: DataCompareSessionErrorKind, columns?: string): string {
|
||||
return dependencies.formatError?.(kind, columns) ?? (kind === "missingKeyColumns" ? `Missing key columns: ${columns ?? ""}` : kind === "noCommonColumns" ? "No common columns" : "No key columns available");
|
||||
}
|
||||
|
||||
function toSelectableDiff(diff: DataCompareResult): SelectableDataCompareResult {
|
||||
return {
|
||||
added: diff.added.map((row) => ({ ...row, selected: true })),
|
||||
removed: diff.removed.map((row) => ({ ...row, selected: true })),
|
||||
modified: diff.modified.map((row) => ({ ...row, selected: true })),
|
||||
};
|
||||
}
|
||||
|
||||
export function buildDataCompareSyncPlanTables(results: DataCompareTableResult[], targetSchema: string): DataCompareSyncPlanTableOptions[] {
|
||||
return results
|
||||
.filter((table) => table.status === "different")
|
||||
.map((table) => ({
|
||||
tableName: table.targetTable,
|
||||
schema: targetSchema,
|
||||
columns: table.columns,
|
||||
keyColumns: table.keyColumns,
|
||||
columnInfo: table.columnInfo,
|
||||
diff: {
|
||||
added: table.diff.added.filter((row) => row.selected).map(({ selected: _selected, ...row }) => row),
|
||||
removed: table.diff.removed.filter((row) => row.selected).map(({ selected: _selected, ...row }) => row),
|
||||
modified: table.diff.modified.filter((row) => row.selected).map(({ selected: _selected, ...row }) => row),
|
||||
},
|
||||
databaseType: table.databaseType,
|
||||
preSyncStatements: table.preSyncStatements ?? [],
|
||||
}))
|
||||
.filter((table) => table.preSyncStatements.length > 0 || table.diff.added.length > 0 || table.diff.removed.length > 0 || table.diff.modified.length > 0);
|
||||
}
|
||||
|
||||
function publishProgress(session: DataCompareSession, progress: DataCompareSessionProgress): void {
|
||||
session.version += 1;
|
||||
session.progress = progress;
|
||||
const results = session.batchResults;
|
||||
const failedTableCount = results.filter((item) => item.status === "error").length;
|
||||
const tracker = useExportTracker();
|
||||
tracker.updateCompareTask(session.id, {
|
||||
status: "Running",
|
||||
compareCurrent: progress.current,
|
||||
compareTotal: progress.total,
|
||||
compareCurrentObject: progress.table,
|
||||
compareResultCount: results.length,
|
||||
compareSameCount: results.filter((item) => item.status === "same").length,
|
||||
compareDifferentCount: results.filter((item) => item.status === "different").length,
|
||||
compareFailedCount: failedTableCount,
|
||||
compareAddedCount: results.reduce((sum, item) => sum + item.added, 0),
|
||||
compareRemovedCount: results.reduce((sum, item) => sum + item.removed, 0),
|
||||
compareModifiedCount: results.reduce((sum, item) => sum + item.modified, 0),
|
||||
});
|
||||
}
|
||||
|
||||
function updateResults(session: DataCompareSession, results: DataCompareTableResult[]): void {
|
||||
session.batchResults = [...results];
|
||||
}
|
||||
|
||||
async function loadColumnsWithCache(cache: Map<string, CompareColumn[]>, connectionId: string, database: string, schema: string, table: string): Promise<CompareColumn[]> {
|
||||
const key = `${connectionId}:${database}:${schema}:${table}`;
|
||||
const cached = cache.get(key);
|
||||
if (cached) return cached;
|
||||
const columns = (await api.getColumns(connectionId, database, schema, table)) as CompareColumn[];
|
||||
cache.set(key, columns);
|
||||
return columns;
|
||||
}
|
||||
|
||||
async function inferKeyColumnsForTable(input: DataCompareSessionConfig, table: string, sourceColumnCache: Map<string, CompareColumn[]>): Promise<string[]> {
|
||||
if (!input.sourceConnectionId || !input.sourceDatabase || !input.sourceSchema || !table) return [];
|
||||
const columns = await loadColumnsWithCache(sourceColumnCache, input.sourceConnectionId, input.sourceDatabase, input.sourceSchema, table);
|
||||
return inferCompareKeyColumns(columns);
|
||||
}
|
||||
|
||||
function buildTableResult(task: DataCompareTableTask, preparation: DataCompareFromTablesPreparation, keyColumns: string[], columns: string[], columnInfo: CompareColumn[], databaseType: DatabaseType | undefined): DataCompareTableResult {
|
||||
const added = preparation.result.added.length;
|
||||
const removed = preparation.result.removed.length;
|
||||
const modified = preparation.result.modified.length;
|
||||
const status: DataCompareTableStatus = added || removed || modified ? "different" : "same";
|
||||
return {
|
||||
sourceTable: task.sourceTable,
|
||||
targetTable: task.targetTable,
|
||||
keyColumns,
|
||||
columns,
|
||||
columnInfo,
|
||||
status,
|
||||
added,
|
||||
removed,
|
||||
modified,
|
||||
sourceRowCount: preparation.sourceRowCount,
|
||||
targetRowCount: preparation.targetRowCount,
|
||||
sourceTruncated: preparation.sourceTruncated,
|
||||
targetTruncated: preparation.targetTruncated,
|
||||
databaseType,
|
||||
preSyncStatements: preparation.preSyncStatements,
|
||||
diff: toSelectableDiff(preparation.result),
|
||||
expanded: status === "different",
|
||||
showAll: { added: false, removed: false, modified: false },
|
||||
};
|
||||
}
|
||||
|
||||
function buildMissingTargetResult(task: DataCompareTableTask, preparation: DataCompareFromTablesPreparation, sourceColumns: CompareColumn[], keyColumns: string[], databaseType: DatabaseType | undefined): DataCompareTableResult {
|
||||
return {
|
||||
sourceTable: task.sourceTable,
|
||||
targetTable: task.targetTable,
|
||||
keyColumns,
|
||||
columns: sourceColumns.map((column) => column.name),
|
||||
columnInfo: sourceColumns,
|
||||
status: "different",
|
||||
added: preparation.result.added.length,
|
||||
removed: 0,
|
||||
modified: 0,
|
||||
sourceRowCount: preparation.sourceRowCount,
|
||||
targetRowCount: 0,
|
||||
sourceTruncated: preparation.sourceTruncated,
|
||||
targetTruncated: false,
|
||||
databaseType,
|
||||
preSyncStatements: preparation.preSyncStatements,
|
||||
diff: toSelectableDiff(preparation.result),
|
||||
expanded: preparation.result.added.length > 0,
|
||||
showAll: { added: false, removed: false, modified: false },
|
||||
};
|
||||
}
|
||||
|
||||
function buildErrorResult(task: DataCompareTableTask, keyColumns: string[], databaseType: DatabaseType | undefined, error: unknown): DataCompareTableResult {
|
||||
return {
|
||||
sourceTable: task.sourceTable,
|
||||
targetTable: task.targetTable,
|
||||
keyColumns,
|
||||
columns: [],
|
||||
columnInfo: [],
|
||||
status: "error",
|
||||
added: 0,
|
||||
removed: 0,
|
||||
modified: 0,
|
||||
sourceRowCount: 0,
|
||||
targetRowCount: 0,
|
||||
sourceTruncated: false,
|
||||
targetTruncated: false,
|
||||
databaseType,
|
||||
preSyncStatements: [],
|
||||
diff: { added: [], removed: [], modified: [] },
|
||||
expanded: false,
|
||||
showAll: { added: false, removed: false, modified: false },
|
||||
error: sessionError(error),
|
||||
};
|
||||
}
|
||||
|
||||
async function runDataCompareSession(session: DataCompareSession, tasks: DataCompareTableTask[], dependencies: DataCompareSessionDependencies): Promise<void> {
|
||||
const input = session.config;
|
||||
const sourceColumnCache = new Map<string, CompareColumn[]>();
|
||||
const targetColumnCache = new Map<string, CompareColumn[]>();
|
||||
const results: DataCompareTableResult[] = [];
|
||||
const databaseType = dependencies.getConfig(input.targetConnectionId)?.db_type;
|
||||
|
||||
try {
|
||||
publishProgress(session, { current: 0, total: tasks.length, table: "" });
|
||||
await Promise.all([dependencies.ensureConnected(input.sourceConnectionId), dependencies.ensureConnected(input.targetConnectionId)]);
|
||||
|
||||
for (const [index, task] of tasks.entries()) {
|
||||
publishProgress(session, { current: index + 1, total: tasks.length, table: task.sourceTable });
|
||||
let keyColumns = input.keyColumns;
|
||||
try {
|
||||
if (input.targetTables.includes(task.targetTable)) {
|
||||
const resolvedKeys = input.keyColumns.length > 0 ? input.keyColumns : await inferKeyColumnsForTable(input, task.sourceTable, sourceColumnCache);
|
||||
if (resolvedKeys.length === 0) throw new Error(compareErrorMessage(dependencies, "noKeyColumns"));
|
||||
|
||||
const sourceColumns = await loadColumnsWithCache(sourceColumnCache, input.sourceConnectionId, input.sourceDatabase, input.sourceSchema, task.sourceTable);
|
||||
const targetColumns = await loadColumnsWithCache(targetColumnCache, input.targetConnectionId, input.targetDatabase, input.targetSchema, task.targetTable);
|
||||
const matched = intersectCompareColumns(sourceColumns, targetColumns);
|
||||
const columns = matched.columns;
|
||||
const columnInfo = columns.map((column) => targetColumns.find((target) => target.name === column)).filter((column): column is CompareColumn => !!column);
|
||||
const canonicalKeyColumns: string[] = [];
|
||||
const missingKeys: string[] = [];
|
||||
for (const key of resolvedKeys) {
|
||||
const canonical = matchColumnNameIgnoreCase(key, columns);
|
||||
if (!canonical) missingKeys.push(key);
|
||||
else if (!canonicalKeyColumns.includes(canonical)) canonicalKeyColumns.push(canonical);
|
||||
}
|
||||
if (missingKeys.length > 0) throw new Error(compareErrorMessage(dependencies, "missingKeyColumns", missingKeys.join(", ")));
|
||||
if (columns.length === 0) throw new Error(compareErrorMessage(dependencies, "noCommonColumns"));
|
||||
|
||||
const preparation = await api.prepareDataCompareFromTables({
|
||||
sourceConnectionId: input.sourceConnectionId,
|
||||
sourceDatabase: input.sourceDatabase,
|
||||
sourceSchema: input.sourceSchema,
|
||||
sourceTable: task.sourceTable,
|
||||
targetConnectionId: input.targetConnectionId,
|
||||
targetDatabase: input.targetDatabase,
|
||||
targetSchema: input.targetSchema,
|
||||
targetTable: task.targetTable,
|
||||
columns,
|
||||
keyColumns: canonicalKeyColumns,
|
||||
sourceColumns: matched.sourceColumns,
|
||||
});
|
||||
keyColumns = canonicalKeyColumns;
|
||||
results.push(buildTableResult(task, preparation, canonicalKeyColumns, columns, columnInfo, databaseType));
|
||||
} else {
|
||||
const sourceColumns = await loadColumnsWithCache(sourceColumnCache, input.sourceConnectionId, input.sourceDatabase, input.sourceSchema, task.sourceTable);
|
||||
const resolvedKeys = input.keyColumns.length > 0 ? input.keyColumns : [];
|
||||
const sourceColumnNames = sourceColumns.map((column) => column.name);
|
||||
keyColumns = resolvedKeys.map((key) => matchColumnNameIgnoreCase(key, sourceColumnNames) ?? key);
|
||||
const preparation = await api.prepareDataCompareMissingTarget({
|
||||
sourceConnectionId: input.sourceConnectionId,
|
||||
sourceDatabase: input.sourceDatabase,
|
||||
sourceSchema: input.sourceSchema,
|
||||
sourceTable: task.sourceTable,
|
||||
targetConnectionId: input.targetConnectionId,
|
||||
targetDatabase: input.targetDatabase,
|
||||
targetSchema: input.targetSchema,
|
||||
targetTable: task.targetTable,
|
||||
keyColumns,
|
||||
});
|
||||
results.push(buildMissingTargetResult(task, preparation, sourceColumns, keyColumns, databaseType));
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
results.push(buildErrorResult(task, keyColumns, databaseType, error));
|
||||
}
|
||||
updateResults(session, results);
|
||||
publishProgress(session, { current: index + 1, total: tasks.length, table: task.sourceTable });
|
||||
}
|
||||
|
||||
const plan = await api.buildDataCompareSyncPlan({ tables: buildDataCompareSyncPlanTables(results, input.targetSchema) });
|
||||
session.syncPlan = plan;
|
||||
session.progress = null;
|
||||
session.status = "completed";
|
||||
session.version += 1;
|
||||
session.finishedAt = Date.now();
|
||||
const sameCount = results.filter((item) => item.status === "same").length;
|
||||
const differentCount = results.filter((item) => item.status === "different").length;
|
||||
const failedCount = results.filter((item) => item.status === "error").length;
|
||||
useExportTracker().updateCompareTask(session.id, {
|
||||
status: failedCount > 0 ? "Error" : "Done",
|
||||
compareCurrent: tasks.length,
|
||||
compareTotal: tasks.length,
|
||||
compareCurrentObject: "",
|
||||
compareResultCount: results.length,
|
||||
compareSameCount: sameCount,
|
||||
compareDifferentCount: differentCount,
|
||||
compareFailedCount: failedCount,
|
||||
compareAddedCount: results.reduce((sum, item) => sum + item.added, 0),
|
||||
compareRemovedCount: results.reduce((sum, item) => sum + item.removed, 0),
|
||||
compareModifiedCount: results.reduce((sum, item) => sum + item.modified, 0),
|
||||
errorMessage: failedCount > 0 ? `${failedCount} table(s) failed` : null,
|
||||
});
|
||||
} catch (error: unknown) {
|
||||
const message = sessionError(error);
|
||||
const lastProgress = session.progress;
|
||||
session.error = message;
|
||||
session.progress = null;
|
||||
session.status = "failed";
|
||||
session.version += 1;
|
||||
session.finishedAt = Date.now();
|
||||
useExportTracker().updateCompareTask(session.id, {
|
||||
status: "Error",
|
||||
errorMessage: message,
|
||||
compareCurrent: lastProgress?.current,
|
||||
compareTotal: lastProgress?.total,
|
||||
compareCurrentObject: lastProgress?.table,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export function startDataCompareSession(input: DataCompareSessionConfig, tasks: DataCompareTableTask[], dependencies: DataCompareSessionDependencies): DataCompareSession {
|
||||
const id = uuid();
|
||||
const session = shallowReactive<DataCompareSession>({
|
||||
id,
|
||||
version: 0,
|
||||
status: "running",
|
||||
config: cloneConfig(input),
|
||||
progress: null,
|
||||
batchResults: [],
|
||||
syncPlan: emptyDataCompareSyncPlan(),
|
||||
error: null,
|
||||
startedAt: Date.now(),
|
||||
});
|
||||
sessions.set(id, session);
|
||||
|
||||
const tracker = useExportTracker();
|
||||
tracker.addDataCompareTask(
|
||||
id,
|
||||
input.label,
|
||||
() => openDataCompareSession(id),
|
||||
() => removeDataCompareSession(id),
|
||||
);
|
||||
void runDataCompareSession(session, tasks, dependencies);
|
||||
return session;
|
||||
}
|
||||
|
||||
export function getDataCompareSession(id: string | null | undefined): DataCompareSession | undefined {
|
||||
return id ? sessions.get(id) : undefined;
|
||||
}
|
||||
|
||||
export function removeDataCompareSession(id: string): boolean {
|
||||
const session = sessions.get(id);
|
||||
if (session?.status === "running") return false;
|
||||
return sessions.delete(id);
|
||||
}
|
||||
@@ -45,10 +45,12 @@ const transferPrefillTargetSchema = ref("");
|
||||
const schemaDiffPrefillConnectionId = ref("");
|
||||
const schemaDiffPrefillDatabase = ref("");
|
||||
const schemaDiffPrefillSchema = ref("");
|
||||
const schemaDiffSessionId = ref<string | null>(null);
|
||||
const dataComparePrefillConnectionId = ref("");
|
||||
const dataComparePrefillDatabase = ref("");
|
||||
const dataComparePrefillSchema = ref("");
|
||||
const dataComparePrefillTable = ref("");
|
||||
const dataCompareSessionId = ref<string | null>(null);
|
||||
const sqlFilePrefillConnectionId = ref("");
|
||||
const sqlFilePrefillDatabase = ref("");
|
||||
const sqlFilePrefillFilePath = ref("");
|
||||
@@ -96,6 +98,16 @@ function clearTransferPrefill() {
|
||||
transferPrefillTargetSchema.value = "";
|
||||
}
|
||||
|
||||
export function openSchemaDiffSession(sessionId: string): void {
|
||||
schemaDiffSessionId.value = sessionId;
|
||||
showSchemaDiffDialog.value = true;
|
||||
}
|
||||
|
||||
export function openDataCompareSession(sessionId: string): void {
|
||||
dataCompareSessionId.value = sessionId;
|
||||
showDataCompareDialog.value = true;
|
||||
}
|
||||
|
||||
export function useDialogSources() {
|
||||
const { t } = useI18n();
|
||||
const connectionStore = useConnectionStore();
|
||||
@@ -134,6 +146,7 @@ export function useDialogSources() {
|
||||
schemaDiffPrefillConnectionId.value = v.connectionId;
|
||||
schemaDiffPrefillDatabase.value = v.database;
|
||||
schemaDiffPrefillSchema.value = v.schema ?? "";
|
||||
schemaDiffSessionId.value = null;
|
||||
showSchemaDiffDialog.value = true;
|
||||
connectionStore.schemaDiffSource = null;
|
||||
}
|
||||
@@ -148,12 +161,21 @@ export function useDialogSources() {
|
||||
dataComparePrefillDatabase.value = v.database;
|
||||
dataComparePrefillSchema.value = v.schema ?? "";
|
||||
dataComparePrefillTable.value = v.tableName ?? "";
|
||||
dataCompareSessionId.value = null;
|
||||
showDataCompareDialog.value = true;
|
||||
connectionStore.dataCompareSource = null;
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
watch(showSchemaDiffDialog, (open) => {
|
||||
if (!open) schemaDiffSessionId.value = null;
|
||||
});
|
||||
|
||||
watch(showDataCompareDialog, (open) => {
|
||||
if (!open) dataCompareSessionId.value = null;
|
||||
});
|
||||
|
||||
watch(
|
||||
() => connectionStore.sqlFileSource,
|
||||
(v) => {
|
||||
@@ -502,10 +524,12 @@ export function useDialogSources() {
|
||||
schemaDiffPrefillConnectionId,
|
||||
schemaDiffPrefillDatabase,
|
||||
schemaDiffPrefillSchema,
|
||||
schemaDiffSessionId,
|
||||
dataComparePrefillConnectionId,
|
||||
dataComparePrefillDatabase,
|
||||
dataComparePrefillSchema,
|
||||
dataComparePrefillTable,
|
||||
dataCompareSessionId,
|
||||
sqlFilePrefillConnectionId,
|
||||
sqlFilePrefillDatabase,
|
||||
sqlFilePrefillFilePath,
|
||||
|
||||
@@ -2,7 +2,7 @@ import { reactive, computed } from "vue";
|
||||
import * as api from "@/lib/backend/api";
|
||||
import { isTerminalTransferProgress } from "@/lib/backend/transferProgress";
|
||||
|
||||
export type BackgroundTaskKind = "table-export" | "database-export" | "sql-file" | "data-transfer" | "multi-db-execution";
|
||||
export type BackgroundTaskKind = "table-export" | "database-export" | "sql-file" | "data-transfer" | "multi-db-execution" | "schema-diff" | "data-compare";
|
||||
export type BackgroundTaskStatus = "Running" | "Writing" | "Cancelling" | "Done" | "Error" | "Cancelled";
|
||||
export type DatabaseExportSource = "manual" | "scheduled";
|
||||
|
||||
@@ -70,7 +70,36 @@ export interface ExportTask {
|
||||
multiDbSkippedCount?: number;
|
||||
multiDbNotExecutedCount?: number;
|
||||
currentTarget?: { connectionId: string; catalog?: string; database: string; schema?: string };
|
||||
comparePhase?: string;
|
||||
compareCurrent?: number;
|
||||
compareTotal?: number;
|
||||
compareCurrentObject?: string;
|
||||
compareResultCount?: number;
|
||||
compareSameCount?: number;
|
||||
compareDifferentCount?: number;
|
||||
compareFailedCount?: number;
|
||||
compareAddedCount?: number;
|
||||
compareRemovedCount?: number;
|
||||
compareModifiedCount?: number;
|
||||
canCancel?: boolean;
|
||||
onOpen?: () => void;
|
||||
onRemove?: () => void;
|
||||
}
|
||||
|
||||
export interface CompareTaskProgress {
|
||||
status: Extract<BackgroundTaskStatus, "Running" | "Done" | "Error" | "Cancelled">;
|
||||
comparePhase?: string;
|
||||
compareCurrent?: number;
|
||||
compareTotal?: number;
|
||||
compareCurrentObject?: string;
|
||||
compareResultCount?: number;
|
||||
compareSameCount?: number;
|
||||
compareDifferentCount?: number;
|
||||
compareFailedCount?: number;
|
||||
compareAddedCount?: number;
|
||||
compareRemovedCount?: number;
|
||||
compareModifiedCount?: number;
|
||||
errorMessage?: string | null;
|
||||
}
|
||||
|
||||
export interface MultiDbExecutionTaskProgress {
|
||||
@@ -458,6 +487,37 @@ export function useExportTracker() {
|
||||
return task;
|
||||
}
|
||||
|
||||
function addCompareTask(kind: "schema-diff" | "data-compare", sessionId: string, label: string, onOpen?: () => void, onRemove?: () => void): ExportTask {
|
||||
const task = reactive<ExportTask>({
|
||||
exportId: sessionId,
|
||||
kind,
|
||||
tableName: label,
|
||||
format: "compare",
|
||||
filePath: "",
|
||||
rowsExported: 0,
|
||||
totalRows: null,
|
||||
status: "Running",
|
||||
errorMessage: null,
|
||||
compareCurrent: 0,
|
||||
compareTotal: 0,
|
||||
compareCurrentObject: "",
|
||||
canCancel: false,
|
||||
onOpen,
|
||||
onRemove,
|
||||
startedAt: Date.now(),
|
||||
});
|
||||
taskMap.set(sessionId, task);
|
||||
return task;
|
||||
}
|
||||
|
||||
function addSchemaDiffTask(sessionId: string, label: string, onOpen?: () => void, onRemove?: () => void): ExportTask {
|
||||
return addCompareTask("schema-diff", sessionId, label, onOpen, onRemove);
|
||||
}
|
||||
|
||||
function addDataCompareTask(sessionId: string, label: string, onOpen?: () => void, onRemove?: () => void): ExportTask {
|
||||
return addCompareTask("data-compare", sessionId, label, onOpen, onRemove);
|
||||
}
|
||||
|
||||
function addMultiDbExecutionTask(batchId: string, label: string, sourceTabId: string, onOpen?: () => void): ExportTask {
|
||||
const task = reactive<ExportTask>({
|
||||
exportId: batchId,
|
||||
@@ -642,7 +702,36 @@ export function useExportTracker() {
|
||||
task.totalRows = progress.totalRows ?? task.totalRows;
|
||||
}
|
||||
|
||||
function updateCompareTask(sessionId: string, progress: CompareTaskProgress): void {
|
||||
const task = taskMap.get(sessionId);
|
||||
if (!task || (task.kind !== "schema-diff" && task.kind !== "data-compare")) return;
|
||||
task.status = progress.status;
|
||||
task.comparePhase = progress.comparePhase;
|
||||
task.compareCurrent = progress.compareCurrent;
|
||||
task.compareTotal = progress.compareTotal;
|
||||
task.compareCurrentObject = progress.compareCurrentObject;
|
||||
task.compareResultCount = progress.compareResultCount;
|
||||
task.compareSameCount = progress.compareSameCount;
|
||||
task.compareDifferentCount = progress.compareDifferentCount;
|
||||
task.compareFailedCount = progress.compareFailedCount;
|
||||
task.compareAddedCount = progress.compareAddedCount;
|
||||
task.compareRemovedCount = progress.compareRemovedCount;
|
||||
task.compareModifiedCount = progress.compareModifiedCount;
|
||||
task.errorMessage = progress.errorMessage ?? null;
|
||||
if (task.status === "Done" || task.status === "Error" || task.status === "Cancelled") finishExportTask(task);
|
||||
}
|
||||
|
||||
function cleanupFinishedTask(task: ExportTask): void {
|
||||
try {
|
||||
task.onRemove?.();
|
||||
} catch {
|
||||
// cleanup hooks must not prevent the task from being removed
|
||||
}
|
||||
}
|
||||
|
||||
function removeTask(exportId: string) {
|
||||
const task = taskMap.get(exportId);
|
||||
if (task && (task.status === "Done" || task.status === "Error" || task.status === "Cancelled")) cleanupFinishedTask(task);
|
||||
taskMap.delete(exportId);
|
||||
taskCancelHandlers.delete(exportId);
|
||||
transferFailureStates.delete(exportId);
|
||||
@@ -652,6 +741,7 @@ export function useExportTracker() {
|
||||
function clearFinished() {
|
||||
for (const [id, task] of taskMap) {
|
||||
if (task.status === "Done" || task.status === "Error" || task.status === "Cancelled") {
|
||||
cleanupFinishedTask(task);
|
||||
taskMap.delete(id);
|
||||
taskCancelHandlers.delete(id);
|
||||
transferFailureStates.delete(id);
|
||||
@@ -670,6 +760,7 @@ export function useExportTracker() {
|
||||
|
||||
async function cancelTask(exportId: string) {
|
||||
const task = taskMap.get(exportId);
|
||||
if (task?.canCancel === false) return;
|
||||
try {
|
||||
const customHandler = taskCancelHandlers.get(exportId);
|
||||
if (customHandler) {
|
||||
@@ -696,6 +787,8 @@ export function useExportTracker() {
|
||||
addDatabaseExportTask,
|
||||
addSqlFileTask,
|
||||
addDataTransferTask,
|
||||
addSchemaDiffTask,
|
||||
addDataCompareTask,
|
||||
addMultiDbExecutionTask,
|
||||
updateMultiDbExecutionTask,
|
||||
startDataTransferTask,
|
||||
@@ -705,6 +798,7 @@ export function useExportTracker() {
|
||||
restoreDatabaseExportTaskRunning,
|
||||
updateSqlFileTask,
|
||||
updateDataTransferTask,
|
||||
updateCompareTask,
|
||||
registerTaskCancelHandler,
|
||||
unregisterTaskCancelHandler,
|
||||
removeTask,
|
||||
|
||||
@@ -0,0 +1,284 @@
|
||||
import { reactive, shallowReactive } from "vue";
|
||||
import { uuid } from "@/lib/common/utils";
|
||||
import * as api from "@/lib/backend/api";
|
||||
import { openSchemaDiffSession } from "@/composables/useDialogSources";
|
||||
import { useExportTracker } from "@/composables/useExportTracker";
|
||||
import { filterSchemaDiffTables } from "@/lib/schema/schemaDiffTableFilter";
|
||||
import { compileSchemaDiffTableFilter } from "@/lib/schema/schemaDiffTableFilter";
|
||||
import { loadSchemaDetails } from "@/lib/schema/schemaDiffMetadataLoad";
|
||||
import type { SchemaDiffTableIdentity, SchemaDiffTableListLoader } from "@/lib/schema/schemaDiffTableList";
|
||||
import { getSchemaDiffNextProgressStep, isSchemaDiffPostgresLike, shouldLoadSchemaDiffExtraObjects, type SchemaDiffProgressPhase } from "@/lib/schema/schemaDiffProgress";
|
||||
import { convertToSchemaDiffObjects, databaseTypeToDialectKind, normalizeDialectKind, schemaDiffDeployTargetSchema, type SchemaDiffObject } from "@/lib/schema/schemaDiff";
|
||||
import { normalizeSchemaDiffCompareOptions, type SchemaDiffCompareOptions } from "@/types/schemaDiff";
|
||||
import type { DatabaseType } from "@/types/database";
|
||||
import type { SchemaDiffPreparation } from "@/lib/schema/schemaDiff";
|
||||
|
||||
export interface SchemaDiffSessionProgress {
|
||||
phase: SchemaDiffProgressPhase;
|
||||
current?: number;
|
||||
total?: number;
|
||||
objectName?: string;
|
||||
}
|
||||
|
||||
export interface SchemaDiffSessionConfig {
|
||||
sourceConnectionId: string;
|
||||
sourceDatabase: string;
|
||||
sourceSchema: string;
|
||||
targetConnectionId: string;
|
||||
targetDatabase: string;
|
||||
targetSchema: string;
|
||||
sourceDbType: string;
|
||||
targetDbType: DatabaseType;
|
||||
options: SchemaDiffCompareOptions;
|
||||
ignoreComments: boolean;
|
||||
label: string;
|
||||
}
|
||||
|
||||
export type SchemaDiffSessionStatus = "running" | "completed" | "failed";
|
||||
|
||||
export interface SchemaDiffSession {
|
||||
id: string;
|
||||
version: number;
|
||||
status: SchemaDiffSessionStatus;
|
||||
config: SchemaDiffSessionConfig;
|
||||
progress: SchemaDiffSessionProgress | null;
|
||||
result: SchemaDiffPreparation | null;
|
||||
error: string | null;
|
||||
startedAt: number;
|
||||
finishedAt?: number;
|
||||
}
|
||||
|
||||
export interface SchemaDiffSessionDependencies {
|
||||
tableListLoader: SchemaDiffTableListLoader;
|
||||
}
|
||||
|
||||
const sessions = reactive(new Map<string, SchemaDiffSession>());
|
||||
|
||||
function cloneOptions(options: SchemaDiffCompareOptions): SchemaDiffCompareOptions {
|
||||
return {
|
||||
...options,
|
||||
selectedTables: options.selectedTables ? [...options.selectedTables] : options.selectedTables,
|
||||
tableMappings: options.tableMappings?.map((mapping) => ({ ...mapping })),
|
||||
fieldMappings: options.fieldMappings?.map((mapping) => ({ ...mapping })),
|
||||
};
|
||||
}
|
||||
|
||||
function resultObjectCount(result: SchemaDiffPreparation): number {
|
||||
return result.diffs.length + (result.functionDiffs?.length ?? 0) + (result.sequenceDiffs?.length ?? 0) + (result.ruleDiffs?.length ?? 0) + (result.ownerDiffs?.length ?? 0);
|
||||
}
|
||||
|
||||
function sessionError(error: unknown): string {
|
||||
return error instanceof Error ? error.message : (error as { message?: string } | null)?.message || String(error);
|
||||
}
|
||||
|
||||
function publishProgress(session: SchemaDiffSession, progress: SchemaDiffSessionProgress): void {
|
||||
session.version += 1;
|
||||
session.progress = progress;
|
||||
useExportTracker().updateCompareTask(session.id, {
|
||||
status: "Running",
|
||||
comparePhase: progress.phase,
|
||||
compareCurrent: progress.current,
|
||||
compareTotal: progress.total,
|
||||
compareCurrentObject: progress.objectName,
|
||||
});
|
||||
}
|
||||
|
||||
async function runSchemaDiffSession(session: SchemaDiffSession, dependencies: SchemaDiffSessionDependencies): Promise<void> {
|
||||
const input = session.config;
|
||||
const tracker = useExportTracker();
|
||||
const options = normalizeSchemaDiffCompareOptions(input.options, input.targetDbType);
|
||||
const isPostgresLike = isSchemaDiffPostgresLike(input.targetDbType);
|
||||
const hasExtraObjectPhase = shouldLoadSchemaDiffExtraObjects(input.targetDbType, options);
|
||||
const tableFilter = compileSchemaDiffTableFilter(options);
|
||||
|
||||
try {
|
||||
publishProgress(session, { phase: "loading-table-lists" });
|
||||
|
||||
const sourceTableIdentity: SchemaDiffTableIdentity = {
|
||||
connectionId: input.sourceConnectionId,
|
||||
database: input.sourceDatabase,
|
||||
schema: input.sourceSchema,
|
||||
};
|
||||
const targetTableIdentity: SchemaDiffTableIdentity = {
|
||||
connectionId: input.targetConnectionId,
|
||||
database: input.targetDatabase,
|
||||
schema: input.targetSchema,
|
||||
};
|
||||
const [sourceTableList, targetTableList] = await Promise.all([dependencies.tableListLoader.load(sourceTableIdentity, { refresh: true }), dependencies.tableListLoader.load(targetTableIdentity, { refresh: true })]);
|
||||
const { sourceTables, targetTables } = filterSchemaDiffTables(sourceTableList, targetTableList, tableFilter, options, options.selectedTables);
|
||||
|
||||
publishProgress(session, { phase: "loading-source-details", current: 0, total: sourceTables.length });
|
||||
const sourceDetails = await loadSchemaDetails(
|
||||
sourceTables,
|
||||
{
|
||||
connectionId: input.sourceConnectionId,
|
||||
database: input.sourceDatabase,
|
||||
schema: input.sourceSchema,
|
||||
dbType: input.sourceDbType,
|
||||
options,
|
||||
onProgress: (progress) => publishProgress(session, { phase: "loading-source-details", ...progress }),
|
||||
},
|
||||
api,
|
||||
);
|
||||
|
||||
publishProgress(session, { phase: "loading-target-details", current: 0, total: targetTables.length });
|
||||
const targetDetails = await loadSchemaDetails(
|
||||
targetTables,
|
||||
{
|
||||
connectionId: input.targetConnectionId,
|
||||
database: input.targetDatabase,
|
||||
schema: input.targetSchema,
|
||||
dbType: input.targetDbType,
|
||||
options,
|
||||
onProgress: (progress) => publishProgress(session, { phase: "loading-target-details", ...progress }),
|
||||
},
|
||||
api,
|
||||
);
|
||||
|
||||
const promises: Promise<unknown>[] = [];
|
||||
if (isPostgresLike && options.functions) {
|
||||
promises.push(api.listFunctions(input.sourceConnectionId, input.sourceDatabase, input.sourceSchema));
|
||||
promises.push(api.listFunctions(input.targetConnectionId, input.targetDatabase, input.targetSchema));
|
||||
}
|
||||
if (isPostgresLike && options.sequences) {
|
||||
promises.push(api.listSequences(input.sourceConnectionId, input.sourceDatabase, input.sourceSchema, !!options.sequenceLastValues));
|
||||
promises.push(api.listSequences(input.targetConnectionId, input.targetDatabase, input.targetSchema, !!options.sequenceLastValues));
|
||||
}
|
||||
if (isPostgresLike && options.rules) {
|
||||
promises.push(api.listRules(input.sourceConnectionId, input.sourceDatabase, input.sourceSchema));
|
||||
promises.push(api.listRules(input.targetConnectionId, input.targetDatabase, input.targetSchema));
|
||||
}
|
||||
if (isPostgresLike && options.owners) {
|
||||
promises.push(api.listOwners(input.sourceConnectionId, input.sourceDatabase, input.sourceSchema));
|
||||
promises.push(api.listOwners(input.targetConnectionId, input.targetDatabase, input.targetSchema));
|
||||
}
|
||||
|
||||
if (hasExtraObjectPhase) publishProgress(session, { phase: "loading-extra-objects" });
|
||||
const extraObjects = await Promise.all(promises);
|
||||
let index = 0;
|
||||
const sourceFunctions = options.functions && isPostgresLike ? (extraObjects[index++] as Awaited<ReturnType<typeof api.listFunctions>>) : [];
|
||||
const targetFunctions = options.functions && isPostgresLike ? (extraObjects[index++] as Awaited<ReturnType<typeof api.listFunctions>>) : [];
|
||||
const sourceSequences = options.sequences && isPostgresLike ? (extraObjects[index++] as Awaited<ReturnType<typeof api.listSequences>>) : [];
|
||||
const targetSequences = options.sequences && isPostgresLike ? (extraObjects[index++] as Awaited<ReturnType<typeof api.listSequences>>) : [];
|
||||
const sourceRules = options.rules && isPostgresLike ? (extraObjects[index++] as Awaited<ReturnType<typeof api.listRules>>) : [];
|
||||
const targetRules = options.rules && isPostgresLike ? (extraObjects[index++] as Awaited<ReturnType<typeof api.listRules>>) : [];
|
||||
const sourceOwners = options.owners && isPostgresLike ? (extraObjects[index++] as Awaited<ReturnType<typeof api.listOwners>>) : [];
|
||||
const targetOwners = options.owners && isPostgresLike ? (extraObjects[index++] as Awaited<ReturnType<typeof api.listOwners>>) : [];
|
||||
|
||||
publishProgress(session, { phase: "comparing" });
|
||||
const result = await api.prepareSchemaDiff({
|
||||
sourceTables,
|
||||
targetTables,
|
||||
sourceDetails,
|
||||
targetDetails,
|
||||
sourceFunctions,
|
||||
targetFunctions,
|
||||
sourceSequences,
|
||||
targetSequences,
|
||||
sourceRules,
|
||||
targetRules,
|
||||
sourceOwners,
|
||||
targetOwners,
|
||||
tableMappings: options.selectedTables === undefined ? undefined : options.tableMappings,
|
||||
databaseType: input.targetDbType,
|
||||
targetSchema: schemaDiffDeployTargetSchema(input.targetDbType, input.targetDatabase, input.targetSchema),
|
||||
ignoreComments: input.ignoreComments,
|
||||
cascadeDelete: options.cascadeDelete ?? false,
|
||||
compareColumnOrder: options.compareColumnOrder,
|
||||
ignoreTableNameCase: options.ignoreTableNameCase,
|
||||
ignoreColumnNameCase: options.ignoreColumnNameCase,
|
||||
detectRenames: options.detectRenames ?? false,
|
||||
detectTableRenames: options.detectTableRenames ?? false,
|
||||
renameThreshold: options.renameThreshold ?? 0.5,
|
||||
enableRollback: options.enableRollback ?? false,
|
||||
batchPatterns: options.batchPatterns
|
||||
? options.batchPatterns
|
||||
.split(",")
|
||||
.map((value) => value.trim())
|
||||
.filter(Boolean)
|
||||
: undefined,
|
||||
sourceDialect: options.sourceDialect ? normalizeDialectKind(options.sourceDialect) : input.sourceDbType ? databaseTypeToDialectKind(input.sourceDbType as DatabaseType) : undefined,
|
||||
targetDialect: options.targetDialect ? normalizeDialectKind(options.targetDialect) : input.targetDbType ? databaseTypeToDialectKind(input.targetDbType) : undefined,
|
||||
compatibilityThreshold: options.compatibilityThreshold ?? 0.5,
|
||||
fieldMappings:
|
||||
options.fieldMappings?.map((mapping) => ({
|
||||
sourceType: mapping.sourceType,
|
||||
targetType: mapping.targetType,
|
||||
paramStrategy: mapping.paramStrategy ?? "preserve",
|
||||
customParams: mapping.customParams,
|
||||
})) || [],
|
||||
});
|
||||
|
||||
publishProgress(session, { phase: "generating" });
|
||||
session.result = result;
|
||||
session.progress = { phase: "complete" };
|
||||
session.version += 1;
|
||||
session.status = "completed";
|
||||
session.finishedAt = Date.now();
|
||||
tracker.updateCompareTask(session.id, {
|
||||
status: "Done",
|
||||
comparePhase: "complete",
|
||||
compareCurrent: 1,
|
||||
compareTotal: 1,
|
||||
compareResultCount: resultObjectCount(result),
|
||||
});
|
||||
} catch (error: unknown) {
|
||||
const message = sessionError(error);
|
||||
session.error = message;
|
||||
session.progress = null;
|
||||
session.version += 1;
|
||||
session.status = "failed";
|
||||
session.finishedAt = Date.now();
|
||||
tracker.updateCompareTask(session.id, {
|
||||
status: "Error",
|
||||
errorMessage: message,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
export function startSchemaDiffSession(input: Omit<SchemaDiffSessionConfig, "options"> & { options: SchemaDiffCompareOptions }, dependencies: SchemaDiffSessionDependencies): SchemaDiffSession {
|
||||
const id = uuid();
|
||||
const session = shallowReactive<SchemaDiffSession>({
|
||||
id,
|
||||
version: 0,
|
||||
status: "running",
|
||||
config: {
|
||||
...input,
|
||||
options: cloneOptions(input.options),
|
||||
},
|
||||
progress: null,
|
||||
result: null,
|
||||
error: null,
|
||||
startedAt: Date.now(),
|
||||
});
|
||||
sessions.set(id, session);
|
||||
|
||||
const tracker = useExportTracker();
|
||||
tracker.addSchemaDiffTask(
|
||||
id,
|
||||
input.label,
|
||||
() => openSchemaDiffSession(id),
|
||||
() => removeSchemaDiffSession(id),
|
||||
);
|
||||
void runSchemaDiffSession(session, dependencies);
|
||||
return session;
|
||||
}
|
||||
|
||||
export function getSchemaDiffSession(id: string | null | undefined): SchemaDiffSession | undefined {
|
||||
return id ? sessions.get(id) : undefined;
|
||||
}
|
||||
|
||||
export function removeSchemaDiffSession(id: string): boolean {
|
||||
const session = sessions.get(id);
|
||||
if (session?.status === "running") return false;
|
||||
return sessions.delete(id);
|
||||
}
|
||||
|
||||
export function schemaDiffSessionNextProgressStep(session: SchemaDiffSession): string | null {
|
||||
return getSchemaDiffNextProgressStep(session.progress?.phase, shouldLoadSchemaDiffExtraObjects(session.config.targetDbType, session.config.options));
|
||||
}
|
||||
|
||||
export function schemaDiffSessionObjects(result: SchemaDiffPreparation | null): SchemaDiffObject[] {
|
||||
if (!result) return [];
|
||||
return convertToSchemaDiffObjects(result.diffs, result.functionDiffs, result.sequenceDiffs, result.ruleDiffs, result.ownerDiffs, result.renameCandidates);
|
||||
}
|
||||
Reference in new issue
Block a user