GithubBackup/MyGoNavi
1
0
Fork 0
mirror of https://github.com/Syngnat/GoNavi.git synced 2026-08-22 08:53:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

✨ feat(data-sync): 建立同步任务模型与运行网关

Browse Source 
- 定义任务、映射、调度、预检和运行记录前端模型
- 接入 Wails 元数据、审批、生命周期、检查点与错误行 API
- 严格解码后端协议并补齐模型和网关回归测试
This commit is contained in:
Syngnat
2026-08-08 19:26:57 +08:00
parent 11dbb1f9dd
commit 53dde950fd
11 changed files with 5229 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

167
frontend/src/components/data-sync/gateway.test.ts Normal file
View File

@@ -0,0 +1,167 @@
import { describe, expect, it } from 'vitest';
import { createStaticDataSyncWorkbenchGateway } from './gateway';
import {
createDataSyncTableMapping,
createDataSyncTaskDraft,
reviseDataSyncTask,
} from './model';
const configuredTask = () => {
const draft = createDataSyncTaskDraft({
id: 'static-task',
kind: 'migration',
now: '2026-08-08T00:00:00.000Z',
});
return reviseDataSyncTask(draft, {
name: 'Static migration',
lifecycle: 'ready',
source: { ...draft.source, connectionId: 'source' },
target: { ...draft.target, connectionId: 'target' },
mappings: [createDataSyncTableMapping('map-1', 'source.orders', 'target.orders')],
});
};
describe('static data sync workbench gateway', () => {
it('persists task references in memory and binds preflight to the task revision', async () => {
const task = configuredTask();
const gateway = createStaticDataSyncWorkbenchGateway({
tasks: [task],
capabilities: {
[task.id]: {
level: 'full',
canExecute: true,
supportsAutoCreate: true,
supportsCdc: false,
},
},
now: () => '2026-08-08T01:00:00.000Z',
});
const preflight = await gateway.preflightTask(task);
expect(preflight).toMatchObject({
taskId: task.id,
taskRevision: task.revision,
status: 'passed',
issues: [],
definitionHash: `static:${task.id}:${task.revision}`,
approvalRequired: false,
});
const run = await gateway.startTask(task, preflight);
expect(run).toMatchObject({
taskId: task.id,
status: 'queued',
trigger: 'manual',
});
expect(await gateway.listRuns(task.id)).toHaveLength(1);
const renamed = reviseDataSyncTask(task, { name: 'Renamed migration' });
await gateway.saveTask(renamed);
expect(await gateway.listTasks()).toContainEqual(renamed);
});
it('fails closed with an explicit warning when no backend capability is injected', async () => {
const task = configuredTask();
const gateway = createStaticDataSyncWorkbenchGateway({ tasks: [task] });
expect(await gateway.resolveCapability(task)).toMatchObject({
level: 'unknown',
canExecute: false,
});
expect((await gateway.preflightTask(task)).issues.map((issue) => issue.code)).toContain(
'capability_unverified',
);
});
it('exposes credential-free metadata fixtures across connection, object, and field levels', async () => {
const gateway = createStaticDataSyncWorkbenchGateway();
const connections = await gateway.listSavedConnections();
const source = connections.find((item) => item.id === 'fixture-mysql-sales')!;
expect(Object.keys(source).sort()).toEqual([
'id',
'name',
'readable',
'type',
'writable',
]);
expect(await gateway.listDatabases(source.id)).toEqual([{ name: 'sales' }]);
const endpoint = {
connectionId: source.id,
connectionName: source.name,
type: source.type,
database: 'sales',
schema: '',
};
expect(await gateway.listObjects(endpoint)).toContainEqual({
name: 'orders',
kind: 'table',
});
expect(await gateway.listFields(endpoint, 'orders')).toContainEqual(
expect.objectContaining({ name: 'id', key: true }),
);
});
it('does not mint approval tokens and blocks an approval-required run', async () => {
const task = configuredTask();
const gateway = createStaticDataSyncWorkbenchGateway({
tasks: [task],
approvalRequiredByTask: { [task.id]: true },
definitionHashByTask: { [task.id]: 'definition-hash' },
});
const preflight = await gateway.preflightTask(task);
expect(preflight).toMatchObject({
approvalRequired: true,
definitionHash: 'definition-hash',
});
await expect(gateway.startTask(task, preflight)).rejects.toThrow(
'preflight is not current',
);
await expect(
gateway.beginApproval(task, preflight),
).rejects.toThrow('approval gateway is not configured');
await expect(gateway.approveTask(task, preflight)).rejects.toThrow(
'approval gateway is not configured',
);
});
it('resets checkpoints only for the current paused task revision', async () => {
const ready = configuredTask();
const paused = { ...ready, lifecycle: 'paused' as const };
const checkpoint = {
taskId: paused.id,
runId: 'run-1',
kind: 'watermark',
phase: 'batch_committed',
cursorPreview: '{"id":42}',
updatedAt: '2026-08-08T00:30:00.000Z',
};
const gateway = createStaticDataSyncWorkbenchGateway({
tasks: [paused],
checkpointsByTask: { [paused.id]: checkpoint },
now: () => '2026-08-08T01:00:00.000Z',
});
await expect(
gateway.resetCheckpoint(paused.id, paused.revision - 1),
).rejects.toThrow('revision changed');
const saved = await gateway.resetCheckpoint(paused.id, paused.revision);
expect(saved).toMatchObject({
id: paused.id,
lifecycle: 'paused',
revision: paused.revision + 1,
});
expect(await gateway.getCheckpoint(paused.id)).toBeNull();
const readyGateway = createStaticDataSyncWorkbenchGateway({
tasks: [ready],
checkpointsByTask: { [ready.id]: checkpoint },
});
await expect(
readyGateway.resetCheckpoint(ready.id, ready.revision),
).rejects.toThrow('requires a paused task');
});
});

401
frontend/src/components/data-sync/gateway.ts Normal file
View File

@@ -0,0 +1,401 @@
import {
resolveDataSyncPreflightStatus,
validateDataSyncTask,
type DataSyncApprovalChallenge,
type DataSyncApprovalGrant,
type DataSyncCdcSourceStatus,
type DataSyncCheckpointSummary,
type DataSyncDatabaseMetadata,
type DataSyncErrorRow,
type DataSyncFieldMetadata,
type DataSyncObjectMetadata,
type DataSyncPreflightSnapshot,
type DataSyncRouteCapability,
type DataSyncRunRecord,
type DataSyncSavedConnectionView,
type DataSyncScheduleSummary,
type DataSyncTaskDefinition,
type DataSyncEndpointRef,
type DataSyncValidationIssue,
} from './model';
export interface DataSyncWorkbenchGateway {
readonly capabilities: {
/** Row retry is exposed only when the backend can replay captured payloads. */
errorRowRetry: boolean;
};
/** Returns credential-free connection summaries only. */
listSavedConnections(): Promise<DataSyncSavedConnectionView[]>;
listDatabases(connectionId: string): Promise<DataSyncDatabaseMetadata[]>;
listObjects(endpoint: DataSyncEndpointRef): Promise<DataSyncObjectMetadata[]>;
listFields(
endpoint: DataSyncEndpointRef,
objectName: string,
): Promise<DataSyncFieldMetadata[]>;
listTasks(): Promise<DataSyncTaskDefinition[]>;
saveTask(task: DataSyncTaskDefinition): Promise<DataSyncTaskDefinition>;
resolveCapability(task: DataSyncTaskDefinition): Promise<DataSyncRouteCapability>;
preflightTask(task: DataSyncTaskDefinition): Promise<DataSyncPreflightSnapshot>;
beginApproval(
task: DataSyncTaskDefinition,
preflight: DataSyncPreflightSnapshot,
): Promise<DataSyncApprovalChallenge>;
approveTask(
task: DataSyncTaskDefinition,
preflight: DataSyncPreflightSnapshot,
): Promise<DataSyncApprovalGrant>;
startTask(
task: DataSyncTaskDefinition,
preflight: DataSyncPreflightSnapshot,
): Promise<DataSyncRunRecord>;
listRuns(taskId?: string): Promise<DataSyncRunRecord[]>;
listErrorRows(runId: string): Promise<DataSyncErrorRow[]>;
listSchedules(): Promise<DataSyncScheduleSummary[]>;
listCdcAdapters(): Promise<string[]>;
listCdcSources(): Promise<DataSyncCdcSourceStatus[]>;
getCheckpoint(taskId: string): Promise<DataSyncCheckpointSummary | null>;
resetCheckpoint(
taskId: string,
expectedJobRevision: number,
): Promise<DataSyncTaskDefinition>;
cancelRun(runId: string): Promise<void>;
resumeRun(runId: string): Promise<DataSyncRunRecord>;
retryRun(runId: string): Promise<DataSyncRunRecord>;
discardErrorRow(errorRowId: string): Promise<void>;
retryErrorRow(errorRowId: string): Promise<DataSyncErrorRow>;
}
export type StaticDataSyncGatewayFixtures = {
savedConnections?: DataSyncSavedConnectionView[];
databasesByConnection?: Record<string, DataSyncDatabaseMetadata[]>;
objectsByEndpoint?: Record<string, DataSyncObjectMetadata[]>;
fieldsByObject?: Record<string, DataSyncFieldMetadata[]>;
tasks?: DataSyncTaskDefinition[];
capabilities?: Record<string, DataSyncRouteCapability>;
runs?: DataSyncRunRecord[];
errorRowsByRun?: Record<string, DataSyncErrorRow[]>;
schedules?: DataSyncScheduleSummary[];
cdcSources?: DataSyncCdcSourceStatus[];
cdcAdapters?: string[];
checkpointsByTask?: Record<string, DataSyncCheckpointSummary>;
extraPreflightIssues?: DataSyncValidationIssue[];
approvalRequiredByTask?: Record<string, boolean>;
definitionHashByTask?: Record<string, string>;
now?: () => string;
};
const DEFAULT_SAVED_CONNECTIONS: DataSyncSavedConnectionView[] = [
{
id: 'fixture-mysql-sales',
name: 'MySQL Sales',
type: 'mysql',
readable: true,
writable: true,
},
{
id: 'fixture-postgres-analytics',
name: 'PostgreSQL Analytics',
type: 'postgresql',
readable: true,
writable: true,
},
];
const DEFAULT_DATABASES: Record<string, DataSyncDatabaseMetadata[]> = {
'fixture-mysql-sales': [{ name: 'sales' }],
'fixture-postgres-analytics': [{ name: 'analytics' }],
};
export const dataSyncEndpointMetadataKey = (
endpoint: Pick<DataSyncEndpointRef, 'connectionId' | 'database' | 'schema'>,
): string =>
[endpoint.connectionId, endpoint.database, endpoint.schema]
.map((part) => encodeURIComponent(part.trim().toLowerCase()))
.join('|');
export const dataSyncObjectMetadataKey = (
endpoint: Pick<DataSyncEndpointRef, 'connectionId' | 'database' | 'schema'>,
objectName: string,
): string =>
`${dataSyncEndpointMetadataKey(endpoint)}|${encodeURIComponent(
objectName.trim().toLowerCase(),
)}`;
const DEFAULT_OBJECTS: Record<string, DataSyncObjectMetadata[]> = {
[dataSyncEndpointMetadataKey({
connectionId: 'fixture-mysql-sales',
database: 'sales',
schema: '',
})]: [
{ name: 'orders', kind: 'table' },
{ name: 'customers', kind: 'table' },
],
[dataSyncEndpointMetadataKey({
connectionId: 'fixture-postgres-analytics',
database: 'analytics',
schema: '',
})]: [
{ name: 'fact_orders', kind: 'table' },
{ name: 'dim_customers', kind: 'table' },
],
};
const DEFAULT_FIELDS: Record<string, DataSyncFieldMetadata[]> = {
[dataSyncObjectMetadataKey(
{
connectionId: 'fixture-mysql-sales',
database: 'sales',
schema: '',
},
'orders',
)]: [
{ name: 'id', type: 'bigint', nullable: false, ordinal: 1, key: true },
{ name: 'customer_id', type: 'bigint', nullable: false, ordinal: 2, key: false },
{ name: 'amount', type: 'decimal(18,2)', nullable: false, ordinal: 3, key: false },
{ name: 'created_at', type: 'datetime', nullable: false, ordinal: 4, key: false },
],
[dataSyncObjectMetadataKey(
{
connectionId: 'fixture-postgres-analytics',
database: 'analytics',
schema: '',
},
'fact_orders',
)]: [
{ name: 'id', type: 'int8', nullable: false, ordinal: 1, key: true },
{ name: 'customer_id', type: 'int8', nullable: false, ordinal: 2, key: false },
{ name: 'amount', type: 'numeric(18,2)', nullable: false, ordinal: 3, key: false },
{ name: 'created_at', type: 'timestamptz', nullable: false, ordinal: 4, key: false },
],
};
const copy = <T,>(value: T): T => {
if (typeof structuredClone === 'function') return structuredClone(value);
return JSON.parse(JSON.stringify(value)) as T;
};
const unresolvedCapability: DataSyncRouteCapability = {
level: 'unknown',
canExecute: false,
supportsAutoCreate: false,
supportsCdc: false,
};
/**
* Static adapter used until persisted task/run APIs are available.
* It deliberately does not call Wails and stores only non-secret task references.
*/
export const createStaticDataSyncWorkbenchGateway = (
fixtures: StaticDataSyncGatewayFixtures = {},
): DataSyncWorkbenchGateway => {
const taskMap = new Map(
(fixtures.tasks || []).map((task) => [task.id, copy(task)] as const),
);
const runs = copy(fixtures.runs || []);
const errors = copy(fixtures.errorRowsByRun || {});
const schedules = copy(fixtures.schedules || []);
const cdcSources = copy(fixtures.cdcSources || []);
const cdcAdapters = copy(fixtures.cdcAdapters || ['mongodb-change-stream']);
const checkpoints = copy(fixtures.checkpointsByTask || {});
const savedConnections = copy(
fixtures.savedConnections || DEFAULT_SAVED_CONNECTIONS,
);
const databases = copy(fixtures.databasesByConnection || DEFAULT_DATABASES);
const objects = copy(fixtures.objectsByEndpoint || DEFAULT_OBJECTS);
const fields = copy(fixtures.fieldsByObject || DEFAULT_FIELDS);
const now = fixtures.now || (() => new Date().toISOString());
return {
capabilities: { errorRowRetry: false },
async listSavedConnections() {
return savedConnections.map(copy);
},
async listDatabases(connectionId) {
return (databases[connectionId] || []).map(copy);
},
async listObjects(endpoint) {
const exactKey = dataSyncEndpointMetadataKey(endpoint);
const withoutSchemaKey = dataSyncEndpointMetadataKey({
...endpoint,
schema: '',
});
return (objects[exactKey] || objects[withoutSchemaKey] || []).map(copy);
},
async listFields(endpoint, objectName) {
const exactKey = dataSyncObjectMetadataKey(endpoint, objectName);
const withoutSchemaKey = dataSyncObjectMetadataKey(
{ ...endpoint, schema: '' },
objectName,
);
return (fields[exactKey] || fields[withoutSchemaKey] || []).map(copy);
},
async listTasks() {
return Array.from(taskMap.values()).map(copy);
},
async saveTask(task) {
const saved = copy(task);
taskMap.set(saved.id, saved);
return copy(saved);
},
async resolveCapability(task) {
const capability = fixtures.capabilities?.[task.id];
return copy(capability || unresolvedCapability);
},
async preflightTask(task) {
const capability = fixtures.capabilities?.[task.id];
const issues = [
...validateDataSyncTask(task),
...(fixtures.extraPreflightIssues || []),
];
if (!capability || capability.level === 'unknown') {
issues.push({
id: 'capability_unverified',
severity: 'warning',
code: 'capability_unverified',
stage: 'endpoints',
});
}
return {
taskId: task.id,
taskRevision: task.revision,
status: resolveDataSyncPreflightStatus(issues),
issues: copy(issues),
definitionHash:
fixtures.definitionHashByTask?.[task.id] ||
`static:${task.id}:${task.revision}`,
approvalRequired: Boolean(fixtures.approvalRequiredByTask?.[task.id]),
approvalSatisfied: false,
checkedAt: now(),
};
},
async beginApproval() {
throw new Error('data sync approval gateway is not configured');
},
async approveTask() {
// The static adapter cannot mint production authorization tokens.
throw new Error('data sync approval gateway is not configured');
},
async listRuns(taskId) {
return runs
.filter((run) => !taskId || run.taskId === taskId)
.map(copy);
},
async startTask(task, preflight) {
if (
(task.lifecycle !== 'ready' && task.lifecycle !== 'enabled') ||
preflight.taskId !== task.id ||
preflight.taskRevision !== task.revision ||
preflight.status === 'blocked' ||
preflight.approvalRequired !== false
) {
throw new Error('data sync preflight is not current');
}
const startedAt = now();
const run: DataSyncRunRecord = {
id: `${task.id}:run:${startedAt}`,
taskId: task.id,
taskName: task.name,
status: task.kind === 'cdc' ? 'streaming' : 'queued',
trigger:
task.trigger.mode === 'manual'
? 'manual'
: task.trigger.mode === 'continuous'
? 'continuous'
: 'schedule',
attempt: 1,
resumable: false,
message: '',
startedAt,
finishedAt: '',
rowsRead: 0,
rowsWritten: 0,
rowsFailed: 0,
throughput: 0,
checkpoint: '',
};
runs.unshift(run);
return copy(run);
},
async listErrorRows(runId) {
return (errors[runId] || []).map(copy);
},
async listSchedules() {
return schedules.map(copy);
},
async listCdcAdapters() {
return cdcAdapters.slice();
},
async listCdcSources() {
return cdcSources.map(copy);
},
async getCheckpoint(taskId) {
return checkpoints[taskId] ? copy(checkpoints[taskId]) : null;
},
async resetCheckpoint(taskId, expectedJobRevision) {
const task = taskMap.get(taskId);
if (!task) throw new Error('data sync task not found');
if (task.revision !== expectedJobRevision) {
throw new Error('data sync task revision changed');
}
if (task.lifecycle !== 'paused') {
throw new Error('data sync checkpoint reset requires a paused task');
}
delete checkpoints[taskId];
const saved = {
...task,
revision: task.revision + 1,
updatedAt: now(),
};
taskMap.set(taskId, saved);
return copy(saved);
},
async cancelRun(runId) {
const run = runs.find((item) => item.id === runId);
if (!run) throw new Error('data sync run not found');
run.status = 'cancelling';
},
async resumeRun(runId) {
const previous = runs.find((item) => item.id === runId);
if (!previous) throw new Error('data sync run not found');
const resumed = {
...previous,
id: `${previous.id}:resume:${now()}`,
status: 'queued' as const,
trigger: 'resume' as const,
attempt: previous.attempt + 1,
startedAt: '',
finishedAt: '',
};
runs.unshift(resumed);
return copy(resumed);
},
async retryRun(runId) {
const previous = runs.find((item) => item.id === runId);
if (!previous) throw new Error('data sync run not found');
const retried = {
...previous,
id: `${previous.id}:retry:${now()}`,
status: 'queued' as const,
trigger: 'retry' as const,
attempt: previous.attempt + 1,
startedAt: '',
finishedAt: '',
};
runs.unshift(retried);
return copy(retried);
},
async discardErrorRow(errorRowId) {
for (const rows of Object.values(errors)) {
const row = rows.find((item) => item.id === errorRowId);
if (row) {
row.status = 'discarded';
return;
}
}
throw new Error('data sync error row not found');
},
async retryErrorRow() {
throw new Error('data sync error row retry is not supported');
},
};
};

324
frontend/src/components/data-sync/model.test.ts Normal file
View File

@@ -0,0 +1,324 @@
import { describe, expect, it } from 'vitest';
import {
autoMatchDataSyncFields,
canUseDataSyncRowErrorIsolation,
canStartDataSyncTask,
buildDataSyncMappingsFromSelection,
createDataSyncTableMapping,
createDataSyncTaskDraft,
isDataSyncPreflightCurrent,
resolveDataSyncPreflightStatus,
reviseDataSyncTask,
validateDataSyncTask,
type DataSyncPreflightSnapshot,
} from './model';
describe('data sync task model', () => {
it('creates versioned defaults for compare and CDC tasks', () => {
const compare = createDataSyncTaskDraft({
id: 'compare-1',
kind: 'compare',
compareMode: 'schema',
now: '2026-08-08T00:00:00.000Z',
});
const cdc = createDataSyncTaskDraft({
id: 'cdc-1',
kind: 'cdc',
now: '2026-08-08T00:00:00.000Z',
});
expect(compare).toMatchObject({
schemaVersion: 1,
revision: 1,
compareMode: 'schema',
delivery: { writeMode: 'none' },
trigger: { mode: 'manual' },
incremental: { mode: 'snapshot' },
});
expect(cdc).toMatchObject({
schemaVersion: 1,
revision: 1,
delivery: { writeMode: 'upsert' },
trigger: { mode: 'continuous' },
incremental: {
mode: 'cdc',
initialSnapshot: false,
startPosition: 'latest',
adapter: '',
},
});
});
it('increments revisions and makes an older preflight stale', () => {
const task = createDataSyncTaskDraft({
id: 'task-1',
kind: 'migration',
now: '2026-08-08T00:00:00.000Z',
});
const snapshot: DataSyncPreflightSnapshot = {
taskId: task.id,
taskRevision: task.revision,
status: 'passed',
issues: [],
definitionHash: 'hash-1',
approvalRequired: false,
approvalSatisfied: false,
checkedAt: '2026-08-08T00:01:00.000Z',
};
const revised = reviseDataSyncTask(
task,
{ name: 'Customer migration' },
'2026-08-08T00:02:00.000Z',
);
expect(revised.revision).toBe(2);
expect(revised.createdAt).toBe(task.createdAt);
expect(revised.updatedAt).toBe('2026-08-08T00:02:00.000Z');
expect(isDataSyncPreflightCurrent(task, snapshot)).toBe(true);
expect(isDataSyncPreflightCurrent(revised, snapshot)).toBe(false);
expect(canStartDataSyncTask(revised, snapshot)).toBe(false);
expect(canStartDataSyncTask(task, { ...snapshot, approvalRequired: true })).toBe(
false,
);
});
it('reports endpoint, mapping, key, and delivery blockers by stage', () => {
const task = createDataSyncTaskDraft({ id: 'task-2', kind: 'reconcile' });
const issues = validateDataSyncTask(task);
expect(issues.map((item) => item.code)).toEqual(
expect.arrayContaining([
'task_name_required',
'source_connection_required',
'target_connection_required',
'mapping_required',
]),
);
expect(issues).not.toEqual(
expect.arrayContaining([
expect.objectContaining({ code: 'source_object_required' }),
expect.objectContaining({ code: 'target_object_required' }),
]),
);
expect(resolveDataSyncPreflightStatus(issues)).toBe('blocked');
});
it('builds multiple mappings with safe same-name target suggestions', () => {
const existing = [createDataSyncTableMapping('empty-row')];
const migration = buildDataSyncMappingsFromSelection({
taskId: 'migration-1',
taskKind: 'migration',
sourceNames: ['sales.Orders', 'customers', 'CUSTOMERS'],
targetObjects: [
{ name: 'orders', kind: 'table' },
{ name: 'archive', kind: 'table' },
],
existingMappings: existing,
keyColumnsBySource: { 'sales.orders': ['id'] },
});
expect(migration).toHaveLength(2);
expect(migration[0]).toMatchObject({
sourceObject: 'sales.Orders',
targetObject: 'orders',
targetMode: 'existing_only',
keyColumns: ['id'],
});
expect(migration[1]).toMatchObject({
sourceObject: 'customers',
targetObject: 'customers',
targetMode: 'create_or_reuse',
});
const reconcile = buildDataSyncMappingsFromSelection({
taskId: 'reconcile-1',
taskKind: 'reconcile',
sourceNames: ['missing_target'],
targetObjects: [],
existingMappings: [],
});
expect(reconcile[0]).toMatchObject({
sourceObject: 'missing_target',
targetObject: '',
targetMode: 'existing_only',
});
});
it('accepts a complete reconcile task and detects duplicate targets', () => {
const base = createDataSyncTaskDraft({ id: 'task-3', kind: 'reconcile' });
const first = {
...createDataSyncTableMapping('map-1', 'sales.orders', 'ods.orders'),
keyColumns: ['id'],
};
const configured = reviseDataSyncTask(base, {
name: 'Orders sync',
source: { ...base.source, connectionId: 'mysql-prod' },
target: { ...base.target, connectionId: 'pg-warehouse' },
mappings: [first],
});
expect(validateDataSyncTask(configured)).toEqual([]);
const duplicate = reviseDataSyncTask(configured, {
mappings: [
first,
{
...createDataSyncTableMapping('map-2', 'sales.order_lines', 'ODS.ORDERS'),
keyColumns: ['id'],
},
],
});
expect(validateDataSyncTask(duplicate).map((item) => item.code)).toContain(
'duplicate_target_object',
);
const duplicateSource = reviseDataSyncTask(configured, {
mappings: [
first,
{
...createDataSyncTableMapping('map-3', 'SALES.ORDERS', 'ods.orders_copy'),
keyColumns: ['id'],
},
],
});
expect(validateDataSyncTask(duplicateSource).map((item) => item.code)).toContain(
'duplicate_source_object',
);
});
it('requires Cron and watermark configuration without weakening CDC invariants', () => {
const base = createDataSyncTaskDraft({ id: 'task-4', kind: 'cdc' });
const invalid = reviseDataSyncTask(base, {
trigger: { mode: 'cron', expression: '', timezone: '', overlap: 'skip' },
incremental: {
mode: 'watermark',
column: '',
tieBreaker: '',
overlapWindowMs: 0,
},
});
const codes = validateDataSyncTask(invalid).map((item) => item.code);
expect(codes).toEqual(
expect.arrayContaining([
'cron_expression_required',
'timezone_required',
'watermark_column_required',
'cdc_incremental_required',
'cdc_trigger_required',
]),
);
});
it('auto-matches field names case-insensitively and keeps existing transforms', () => {
const matched = autoMatchDataSyncFields(
'orders-map',
[
{ name: 'ID', type: 'bigint', nullable: false, ordinal: 1, key: true },
{ name: 'amount', type: 'decimal', nullable: false, ordinal: 2, key: false },
{ name: 'source_only', type: 'text', nullable: true, ordinal: 3, key: false },
],
[
{ name: 'id', type: 'int8', nullable: false, ordinal: 1, key: true },
{ name: 'AMOUNT', type: 'numeric', nullable: true, ordinal: 2, key: false },
],
[
{
id: 'keep-transform',
sourceField: 'amount',
targetField: 'amount',
sourceType: 'old',
targetType: 'old',
transform: 'upper',
nullable: false,
},
],
);
expect(matched).toEqual([
expect.objectContaining({
sourceField: 'ID',
targetField: 'id',
sourceType: 'bigint',
targetType: 'int8',
}),
expect.objectContaining({
id: 'keep-transform',
sourceField: 'amount',
targetField: 'AMOUNT',
transform: 'upper',
nullable: true,
}),
]);
});
it('enables row isolation only for CDC or safe data-only atomic SQL routes', () => {
const base = createDataSyncTaskDraft({ id: 'row-errors', kind: 'reconcile' });
const safe = reviseDataSyncTask(base, {
name: 'Safe row isolation',
source: { ...base.source, connectionId: 'source', type: 'mysql' },
target: {
...base.target,
connectionId: 'target',
type: 'postgresql',
},
mappings: [
{
...createDataSyncTableMapping('map', 'orders', 'orders'),
targetMode: 'existing_only',
keyColumns: ['id'],
},
],
});
expect(canUseDataSyncRowErrorIsolation(safe)).toBe(true);
expect(
canUseDataSyncRowErrorIsolation({
...safe,
delivery: { ...safe.delivery, autoAddColumns: true },
}),
).toBe(false);
expect(
validateDataSyncTask({
...safe,
target: { ...safe.target, type: 'clickhouse' },
delivery: { ...safe.delivery, errorPolicy: 'quarantine' },
}).map((item) => item.code),
).toContain('row_error_isolation_unsupported');
});
it('keeps query sinks to one target mapping and blocks watermark append', () => {
const query = createDataSyncTaskDraft({ id: 'query', kind: 'querySink' });
expect(
validateDataSyncTask({
...query,
delivery: { ...query.delivery, writeMode: 'none' },
}).map((item) => item.code),
).toContain('write_mode_required');
expect(
validateDataSyncTask({
...query,
mappings: [
...query.mappings,
createDataSyncTableMapping('query-map-2', '', 'archive'),
],
}).map((item) => item.code),
).toContain('query_sink_single_mapping_required');
const watermark = {
...createDataSyncTaskDraft({ id: 'watermark', kind: 'reconcile' }),
incremental: {
mode: 'watermark' as const,
column: 'updated_at',
tieBreaker: 'id',
overlapWindowMs: 0,
},
};
expect(
validateDataSyncTask({
...watermark,
delivery: { ...watermark.delivery, writeMode: 'append', retryLimit: 0 },
}).map((item) => item.code),
).toContain('watermark_append_unsupported');
});
});

863
frontend/src/components/data-sync/model.ts Normal file
View File

@@ -0,0 +1,863 @@
export const DATA_SYNC_TASK_SCHEMA_VERSION = 1 as const;
export type DataSyncTaskKind =
| 'migration'
| 'reconcile'
| 'querySink'
| 'compare'
| 'cdc';
export type DataSyncTaskLifecycle =
| 'draft'
| 'ready'
| 'enabled'
| 'paused'
| 'archived';
export type DataSyncTaskStage =
| 'endpoints'
| 'mappings'
| 'delivery'
| 'trigger'
| 'preflight';
export type DataSyncCompareMode = 'schema' | 'data' | 'both';
export type DataSyncEndpointRef = {
connectionId: string;
connectionName: string;
type: string;
database: string;
schema: string;
};
/**
* Credential-free connection data that is safe to expose to the workbench.
* Concrete gateways must never add usernames, passwords, DSNs, or SSH secrets.
*/
export type DataSyncSavedConnectionView = {
id: string;
name: string;
type: string;
readable: boolean;
writable: boolean;
};
export type DataSyncDatabaseMetadata = {
name: string;
};
export type DataSyncObjectMetadata = {
name: string;
kind: 'table' | 'view' | 'collection';
rowCount?: number;
dataBytes?: number;
indexBytes?: number;
};
export type DataSyncFieldMetadata = {
name: string;
type: string;
nullable: boolean;
ordinal: number;
key: boolean;
};
export type DataSyncFieldMapping = {
id: string;
sourceField: string;
targetField: string;
sourceType: string;
targetType: string;
transform: string;
transformArgument?: string;
nullable: boolean;
};
export type DataSyncTableMapping = {
id: string;
enabled: boolean;
sourceObject: string;
targetObject: string;
targetMode: 'create_or_reuse' | 'existing_only';
keyColumns: string[];
watermark?: {
column: string;
tieBreaker: string;
};
fields: DataSyncFieldMapping[];
};
export type DataSyncDeliveryPolicy = {
writeMode: 'none' | 'append' | 'upsert' | 'overwrite';
errorPolicy: 'stop' | 'skip' | 'quarantine';
batchSize: number;
commitEvery: number;
retryLimit: number;
retryBackoffMs: number;
propagateDeletes: boolean;
autoAddColumns: boolean;
createIndexes: boolean;
captureErrorPayload: boolean;
};
export type DataSyncTriggerPolicy =
| { mode: 'manual' }
| { mode: 'once'; runAt: string; timezone: string }
| { mode: 'interval'; intervalSeconds: number; timezone: string }
| {
mode: 'cron';
expression: string;
timezone: string;
overlap: 'skip' | 'queue';
}
| { mode: 'continuous' };
export type DataSyncIncrementalPolicy =
| { mode: 'snapshot' }
| {
mode: 'watermark';
column: string;
tieBreaker: string;
overlapWindowMs: number;
}
| {
mode: 'cdc';
initialSnapshot: boolean;
startPosition: 'latest' | 'earliest' | 'checkpoint';
adapter: string;
slotName: string;
publicationName: string;
};
export type DataSyncTaskDefinition = {
schemaVersion: typeof DATA_SYNC_TASK_SCHEMA_VERSION;
id: string;
revision: number;
name: string;
kind: DataSyncTaskKind;
lifecycle: DataSyncTaskLifecycle;
compareMode?: DataSyncCompareMode;
sourceMode: 'tables' | 'query';
sourceQuery: string;
source: DataSyncEndpointRef;
target: DataSyncEndpointRef;
mappings: DataSyncTableMapping[];
delivery: DataSyncDeliveryPolicy;
trigger: DataSyncTriggerPolicy;
incremental: DataSyncIncrementalPolicy;
concurrencyPolicy: 'forbid' | 'queue';
resumePolicy: 'never' | 'manual' | 'auto';
createdAt: string;
updatedAt: string;
};
export type DataSyncValidationSeverity = 'blocker' | 'warning' | 'info';
export type DataSyncValidationCode =
| 'definition_invalid'
| 'definition_hash_failed'
| 'task_name_required'
| 'source_connection_required'
| 'target_connection_required'
| 'source_connection_failed'
| 'target_connection_failed'
| 'source_connect_failed'
| 'source_ping_failed'
| 'target_connect_failed'
| 'target_ping_failed'
| 'same_endpoint'
| 'same_object'
| 'route_unsupported'
| 'source_query_required'
| 'source_query_not_read_only'
| 'mapping_required'
| 'source_object_required'
| 'target_object_required'
| 'duplicate_source_object'
| 'duplicate_target_object'
| 'mapping_compile_failed'
| 'source_columns_failed'
| 'target_columns_failed'
| 'source_column_missing'
| 'target_column_missing'
| 'key_column_missing'
| 'watermark_column_missing'
| 'target_table_check_failed'
| 'target_table_missing'
| 'target_table_will_be_created'
| 'key_columns_required'
| 'batch_size_invalid'
| 'commit_every_invalid'
| 'write_mode_required'
| 'target_protection_blocked'
| 'row_error_isolation_unsupported'
| 'append_retry_unsupported'
| 'append_retry_unsafe'
| 'append_resume_unsafe'
| 'full_overwrite_non_atomic'
| 'watermark_append_unsupported'
| 'watermark_tie_breaker_required'
| 'watermark_initial_value_unsupported'
| 'watermark_runtime_unsupported'
| 'watermark_overwrite_unsupported'
| 'watermark_delete_unsupported'
| 'query_sink_single_mapping_required'
| 'query_key_required'
| 'query_target_pk_mismatch'
| 'query_schema_runtime_validation'
| 'watermark_column_required'
| 'cron_expression_required'
| 'timezone_required'
| 'interval_invalid'
| 'cdc_incremental_required'
| 'cdc_trigger_required'
| 'cdc_adapter_required'
| 'cdc_initial_snapshot_unsupported'
| 'cdc_initial_snapshot_handoff_unsupported'
| 'cdc_earliest_unsupported'
| 'cdc_probe_failed'
| 'cdc_adapter_not_ready'
| 'cdc_existing_target_required'
| 'cdc_upsert_required'
| 'cdc_target_non_atomic'
| 'cdc_authoritative_columns_required'
| 'cdc_checkpoint_unavailable'
| 'cdc_checkpoint_required'
| 'cdc_checkpoint_incompatible'
| 'compare_route_unsupported'
| 'capability_unverified';
export type DataSyncValidationIssue = {
id: string;
severity: DataSyncValidationSeverity;
code: DataSyncValidationCode;
stage: DataSyncTaskStage;
mappingId?: string;
message?: string;
};
export type DataSyncPreflightStatus =
| 'stale'
| 'running'
| 'blocked'
| 'warning'
| 'passed';
export type DataSyncPreflightSnapshot = {
taskId: string;
taskRevision: number;
status: Exclude<DataSyncPreflightStatus, 'stale' | 'running'>;
issues: DataSyncValidationIssue[];
/** Backend-owned hash of the exact definition that was checked. */
definitionHash: string;
/** Production writes remain disabled until an explicit approval grants a token. */
approvalRequired: boolean;
/** True only when the exact enriched definition already carries a valid approval. */
approvalSatisfied: boolean;
checkedAt: string;
};
export type DataSyncApprovalChallenge = {
definitionHash: string;
notBefore: string;
expiresAt: string;
};
export type DataSyncApprovalGrant = {
definitionHash: string;
expiresAt: string;
};
export type DataSyncRouteCapability = {
level: 'full' | 'partial' | 'unsupported' | 'unknown';
canExecute: boolean;
supportsAutoCreate: boolean;
supportsAutoAddColumns?: boolean;
requiresExistingTarget?: boolean;
supportsCdc: boolean;
};
export type DataSyncRunStatus =
| 'queued'
| 'running'
| 'cancelling'
| 'preflighting'
| 'snapshotting'
| 'catching_up'
| 'streaming'
| 'paused'
| 'succeeded'
| 'partial'
| 'failed'
| 'canceled'
| 'cancelled'
| 'interrupted';
export type DataSyncRunRecord = {
id: string;
taskId: string;
taskName: string;
status: DataSyncRunStatus;
trigger: 'manual' | 'schedule' | 'resume' | 'retry' | 'once' | 'cron' | 'continuous';
attempt: number;
resumable: boolean;
message: string;
startedAt: string;
finishedAt: string;
rowsRead: number;
rowsWritten: number;
rowsFailed: number;
throughput: number;
checkpoint: string;
};
export type DataSyncErrorRow = {
id: string;
runId: string;
taskId: string;
mappingId: string;
sourceObject: string;
reason: string;
payloadPreview: string;
retryable: boolean;
status: 'pending' | 'resolved' | 'discarded' | 'unknown';
operation: string;
};
export type DataSyncScheduleSummary = {
id: string;
taskId: string;
taskName: string;
enabled: boolean;
expression: string;
timezone: string;
nextRunAt: string;
};
export type DataSyncCdcSourceStatus = {
taskId: string;
connectionId: string;
connectionName: string;
type: string;
adapter: string;
status: 'ready' | 'probing' | 'lagging' | 'offline' | 'unsupported' | 'unknown';
lagMs: number | null;
checkpoint: string;
reason: string;
};
export type DataSyncCheckpointSummary = {
taskId: string;
runId: string;
kind: string;
phase: string;
cursorPreview: string;
updatedAt: string;
};
type CreateDataSyncTaskInput = {
id: string;
kind: DataSyncTaskKind;
name?: string;
now?: string;
compareMode?: DataSyncCompareMode;
sourceConnectionId?: string;
};
const emptyEndpoint = (connectionId = ''): DataSyncEndpointRef => ({
connectionId,
connectionName: '',
type: '',
database: '',
schema: '',
});
export const createDataSyncTableMapping = (
id: string,
sourceObject = '',
targetObject = '',
): DataSyncTableMapping => ({
id,
enabled: true,
sourceObject,
targetObject,
targetMode: 'existing_only',
keyColumns: [],
fields: [],
});
const normalizeMetadataName = (value: string): string =>
value.trim().toLowerCase();
const metadataObjectBaseName = (value: string): string => {
const parts = value.trim().split('.');
return (parts[parts.length - 1] || '')
.trim()
.replace(/^[`"\[]/, '')
.replace(/[`"\]]$/, '');
};
const pristineDataSyncMapping = (mapping: DataSyncTableMapping): boolean =>
!mapping.sourceObject.trim() &&
!mapping.targetObject.trim() &&
mapping.keyColumns.length === 0 &&
mapping.fields.length === 0;
type BuildDataSyncMappingsInput = {
taskId: string;
taskKind: DataSyncTaskKind;
sourceNames: string[];
targetObjects: DataSyncObjectMetadata[];
existingMappings: DataSyncTableMapping[];
keyColumnsBySource?: Record<string, string[]>;
allowTargetCreate?: boolean;
};
/**
* Turns a user's object selection into safe, editable mappings. Existing work is
* retained, duplicate sources are ignored, and only one unconfigured starter row
* is replaced. Missing targets are auto-created only for one-time migrations.
*/
export const buildDataSyncMappingsFromSelection = ({
taskId,
taskKind,
sourceNames,
targetObjects,
existingMappings,
keyColumnsBySource = {},
allowTargetCreate,
}: BuildDataSyncMappingsInput): DataSyncTableMapping[] => {
const retained = existingMappings.filter((mapping) => !pristineDataSyncMapping(mapping));
const seenSources = new Set(
retained
.map((mapping) => normalizeMetadataName(mapping.sourceObject))
.filter(Boolean),
);
const usedMappingIds = new Set(retained.map((mapping) => mapping.id));
const targetCandidates = targetObjects.filter((object) => object.kind !== 'view');
const matchTarget = (sourceName: string): string => {
const sourceFull = normalizeMetadataName(sourceName);
const sourceBase = normalizeMetadataName(metadataObjectBaseName(sourceName));
const exact = targetCandidates.find(
(candidate) => normalizeMetadataName(candidate.name) === sourceFull,
);
if (exact) return exact.name;
const baseMatches = targetCandidates.filter(
(candidate) =>
normalizeMetadataName(metadataObjectBaseName(candidate.name)) === sourceBase,
);
return baseMatches.length === 1 ? baseMatches[0].name : '';
};
const additions = sourceNames.flatMap((rawName, selectionIndex) => {
const sourceObject = rawName.trim();
const sourceKey = normalizeMetadataName(sourceObject);
if (!sourceKey || seenSources.has(sourceKey)) return [];
seenSources.add(sourceKey);
const matchedTarget = matchTarget(sourceObject);
const canCreateTarget =
(allowTargetCreate ?? taskKind === 'migration') && !matchedTarget;
const idStem = sourceKey.replace(/[^a-z0-9_-]+/g, '-').replace(/^-|-$/g, '');
let mappingId = `${taskId}:mapping:${idStem || selectionIndex + 1}`;
let suffix = 2;
while (usedMappingIds.has(mappingId)) {
mappingId = `${taskId}:mapping:${idStem || selectionIndex + 1}:${suffix}`;
suffix += 1;
}
usedMappingIds.add(mappingId);
const mapping = createDataSyncTableMapping(
mappingId,
sourceObject,
matchedTarget || (canCreateTarget ? metadataObjectBaseName(sourceObject) : ''),
);
mapping.targetMode = canCreateTarget ? 'create_or_reuse' : 'existing_only';
mapping.keyColumns = [
...(keyColumnsBySource[sourceKey] ||
keyColumnsBySource[normalizeMetadataName(metadataObjectBaseName(sourceObject))] ||
[]),
];
return [mapping];
});
return [...retained, ...additions];
};
/**
* Produces a deterministic, editable mapping for fields that share a name.
* Existing transforms are retained so refreshing metadata never discards work.
*/
export const autoMatchDataSyncFields = (
mappingId: string,
sourceFields: DataSyncFieldMetadata[],
targetFields: DataSyncFieldMetadata[],
existing: DataSyncFieldMapping[] = [],
): DataSyncFieldMapping[] => {
const targetByName = new Map(
targetFields.map((field) => [normalizeMetadataName(field.name), field] as const),
);
const existingByPair = new Map<string, DataSyncFieldMapping>(
existing.map((field) => [
`${normalizeMetadataName(field.sourceField)}\u0000${normalizeMetadataName(
field.targetField,
)}`,
field,
] as const),
);
return sourceFields.flatMap((sourceField, index) => {
const targetField = targetByName.get(normalizeMetadataName(sourceField.name));
if (!targetField) return [];
const pairKey = `${normalizeMetadataName(sourceField.name)}\u0000${normalizeMetadataName(
targetField.name,
)}`;
const previous = existingByPair.get(pairKey);
return [
{
id: previous?.id || `${mappingId}:field:${index + 1}`,
sourceField: sourceField.name,
targetField: targetField.name,
sourceType: sourceField.type,
targetType: targetField.type,
transform: previous?.transform || '',
transformArgument: previous?.transformArgument || '',
nullable: targetField.nullable,
},
];
});
};
const defaultWriteMode = (
kind: DataSyncTaskKind,
): DataSyncDeliveryPolicy['writeMode'] => {
if (kind === 'compare') return 'none';
if (kind === 'querySink') return 'append';
return 'upsert';
};
export const createDataSyncTaskDraft = ({
id,
kind,
name = '',
now = new Date().toISOString(),
compareMode,
sourceConnectionId = '',
}: CreateDataSyncTaskInput): DataSyncTaskDefinition => ({
schemaVersion: DATA_SYNC_TASK_SCHEMA_VERSION,
id,
revision: 1,
name,
kind,
lifecycle: 'draft',
compareMode: kind === 'compare' ? compareMode || 'data' : undefined,
sourceMode: kind === 'querySink' ? 'query' : 'tables',
sourceQuery: '',
source: emptyEndpoint(sourceConnectionId),
target: emptyEndpoint(),
mappings:
kind === 'querySink'
? [createDataSyncTableMapping(`${id}:mapping:1`)]
: [],
delivery: {
writeMode: defaultWriteMode(kind),
errorPolicy: 'stop',
batchSize: 1_000,
commitEvery: 1_000,
retryLimit: kind === 'querySink' ? 0 : 3,
retryBackoffMs: 500,
propagateDeletes: false,
autoAddColumns: false,
createIndexes: false,
captureErrorPayload: false,
},
trigger: kind === 'cdc' ? { mode: 'continuous' } : { mode: 'manual' },
incremental:
kind === 'cdc'
? {
mode: 'cdc',
initialSnapshot: false,
startPosition: 'latest',
adapter: '',
slotName: '',
publicationName: '',
}
: { mode: 'snapshot' },
concurrencyPolicy: 'forbid',
resumePolicy: 'manual',
createdAt: now,
updatedAt: now,
});
export const reviseDataSyncTask = (
task: DataSyncTaskDefinition,
patch: Partial<Omit<DataSyncTaskDefinition, 'id' | 'schemaVersion' | 'revision' | 'createdAt'>>,
now = new Date().toISOString(),
): DataSyncTaskDefinition => ({
...task,
...patch,
revision: task.revision + 1,
updatedAt: now,
});
const normalize = (value: unknown): string => String(value ?? '').trim();
const normalizeAtomicTargetType = (value: string): string => {
const normalized = value.trim().toLowerCase();
if (normalized === 'postgresql') return 'postgres';
if (['mssql', 'sql_server', 'sql-server'].includes(normalized)) return 'sqlserver';
if (['kingbase8', 'kingbasees', 'kingbasev8'].includes(normalized)) return 'kingbase';
if (['open_gauss', 'open-gauss'].includes(normalized)) return 'opengauss';
if (['gauss_db', 'gauss-db'].includes(normalized)) return 'gaussdb';
if (['intersystems', 'intersystemsiris', 'inter-systems', 'inter-systems-iris'].includes(normalized)) return 'iris';
if (['dm', 'dm8'].includes(normalized)) return 'dameng';
if (normalized === 'sqlite3') return 'sqlite';
if (['goldendb', 'greatdb', 'gdb'].includes(normalized)) return 'mysql';
return normalized;
};
const ATOMIC_ROW_ISOLATION_TARGETS = new Set([
'mysql',
'mariadb',
'oceanbase',
'postgres',
'kingbase',
'highgo',
'vastbase',
'opengauss',
'gaussdb',
'oracle',
'sqlserver',
'dameng',
'sqlite',
'duckdb',
'iris',
]);
/** Mirrors the backend's fail-closed snapshot/query row-isolation contract. */
export const canUseDataSyncRowErrorIsolation = (
task: DataSyncTaskDefinition,
): boolean => {
if (task.kind === 'cdc') return true;
if (
(task.kind !== 'reconcile' && task.kind !== 'querySink') ||
task.incremental.mode !== 'snapshot' ||
task.delivery.writeMode === 'overwrite' ||
task.delivery.autoAddColumns ||
task.delivery.createIndexes ||
task.delivery.propagateDeletes ||
!ATOMIC_ROW_ISOLATION_TARGETS.has(normalizeAtomicTargetType(task.target.type))
) {
return false;
}
const enabledMappings = task.mappings.filter((mapping) => mapping.enabled);
return (
enabledMappings.length > 0 &&
enabledMappings.every((mapping) => mapping.targetMode === 'existing_only')
);
};
const issue = (
code: DataSyncValidationCode,
severity: DataSyncValidationSeverity,
stage: DataSyncTaskStage,
mappingId?: string,
): DataSyncValidationIssue => ({
id: mappingId ? `${code}:${mappingId}` : code,
code,
severity,
stage,
...(mappingId ? { mappingId } : {}),
});
export const validateDataSyncTask = (
task: DataSyncTaskDefinition,
): DataSyncValidationIssue[] => {
const issues: DataSyncValidationIssue[] = [];
if (!normalize(task.name)) {
issues.push(issue('task_name_required', 'blocker', 'endpoints'));
}
if (!normalize(task.source.connectionId)) {
issues.push(issue('source_connection_required', 'blocker', 'endpoints'));
}
if (!normalize(task.target.connectionId)) {
issues.push(issue('target_connection_required', 'blocker', 'endpoints'));
}
if (
normalize(task.source.connectionId) &&
normalize(task.source.connectionId) === normalize(task.target.connectionId) &&
normalize(task.source.database).toLowerCase() ===
normalize(task.target.database).toLowerCase()
) {
issues.push(issue('same_endpoint', 'warning', 'endpoints'));
}
if (task.sourceMode === 'query' && !normalize(task.sourceQuery)) {
issues.push(issue('source_query_required', 'blocker', 'endpoints'));
}
const enabledMappings = task.mappings.filter((mapping) => mapping.enabled);
if (enabledMappings.length === 0) {
issues.push(issue('mapping_required', 'blocker', 'mappings'));
}
if (task.kind === 'querySink' && task.mappings.length !== 1) {
issues.push(
issue('query_sink_single_mapping_required', 'blocker', 'mappings'),
);
}
const sourceKeys = new Set<string>();
const targetKeys = new Set<string>();
enabledMappings.forEach((mapping) => {
if (task.kind !== 'querySink') {
const sourceObject = normalize(mapping.sourceObject);
if (!sourceObject) {
issues.push(
issue('source_object_required', 'blocker', 'mappings', mapping.id),
);
} else {
const sourceKey = sourceObject.toLowerCase();
if (sourceKeys.has(sourceKey)) {
issues.push(
issue('duplicate_source_object', 'blocker', 'mappings', mapping.id),
);
}
sourceKeys.add(sourceKey);
}
}
const targetObject = normalize(mapping.targetObject);
if (!targetObject) {
issues.push(
issue('target_object_required', 'blocker', 'mappings', mapping.id),
);
} else {
const targetKey = targetObject.toLowerCase();
if (targetKeys.has(targetKey)) {
issues.push(
issue('duplicate_target_object', 'blocker', 'mappings', mapping.id),
);
}
targetKeys.add(targetKey);
}
if (
(task.kind === 'reconcile' || task.kind === 'cdc') &&
mapping.keyColumns.map(normalize).filter(Boolean).length === 0
) {
issues.push(
issue('key_columns_required', 'blocker', 'mappings', mapping.id),
);
}
});
if (
!Number.isInteger(task.delivery.batchSize) ||
task.delivery.batchSize < 1 ||
task.delivery.batchSize > 10_000
) {
issues.push(issue('batch_size_invalid', 'blocker', 'delivery'));
}
if (
!Number.isInteger(task.delivery.commitEvery) ||
task.delivery.commitEvery < task.delivery.batchSize
) {
issues.push(issue('commit_every_invalid', 'blocker', 'delivery'));
}
if (task.kind !== 'compare' && task.delivery.writeMode === 'none') {
issues.push(issue('write_mode_required', 'blocker', 'delivery'));
}
if (
task.delivery.errorPolicy !== 'stop' &&
!canUseDataSyncRowErrorIsolation(task)
) {
issues.push(
issue('row_error_isolation_unsupported', 'blocker', 'delivery'),
);
}
if (task.delivery.writeMode === 'append' && task.delivery.retryLimit !== 0) {
issues.push(issue('append_retry_unsupported', 'blocker', 'delivery'));
}
if (
task.incremental.mode === 'watermark' &&
task.delivery.writeMode === 'append'
) {
issues.push(issue('watermark_append_unsupported', 'blocker', 'delivery'));
}
if (
task.incremental.mode === 'watermark' &&
!normalize(task.incremental.column)
) {
issues.push(issue('watermark_column_required', 'blocker', 'trigger'));
}
if (task.trigger.mode === 'cron') {
if (!normalize(task.trigger.expression)) {
issues.push(issue('cron_expression_required', 'blocker', 'trigger'));
}
if (!normalize(task.trigger.timezone)) {
issues.push(issue('timezone_required', 'blocker', 'trigger'));
}
}
if (
task.trigger.mode === 'interval' &&
(!Number.isInteger(task.trigger.intervalSeconds) || task.trigger.intervalSeconds < 60)
) {
issues.push(issue('interval_invalid', 'blocker', 'trigger'));
}
if (task.kind === 'cdc') {
if (task.incremental.mode !== 'cdc') {
issues.push(issue('cdc_incremental_required', 'blocker', 'trigger'));
}
if (task.trigger.mode !== 'continuous') {
issues.push(issue('cdc_trigger_required', 'blocker', 'trigger'));
}
if (task.incremental.mode === 'cdc') {
if (!normalize(task.incremental.adapter)) {
issues.push(issue('cdc_adapter_required', 'blocker', 'trigger'));
}
if (task.incremental.initialSnapshot) {
issues.push(
issue('cdc_initial_snapshot_unsupported', 'blocker', 'trigger'),
);
}
if (task.incremental.startPosition === 'earliest') {
issues.push(issue('cdc_earliest_unsupported', 'blocker', 'trigger'));
}
}
}
return issues;
};
export const resolveDataSyncPreflightStatus = (
issues: DataSyncValidationIssue[],
): DataSyncPreflightSnapshot['status'] => {
if (issues.some((item) => item.severity === 'blocker')) return 'blocked';
if (issues.some((item) => item.severity === 'warning')) return 'warning';
return 'passed';
};
export const isDataSyncPreflightCurrent = (
task: DataSyncTaskDefinition,
preflight: DataSyncPreflightSnapshot | null,
): boolean =>
Boolean(
preflight &&
preflight.taskId === task.id &&
preflight.taskRevision === task.revision,
);
export const canStartDataSyncTask = (
task: DataSyncTaskDefinition,
preflight: DataSyncPreflightSnapshot | null,
approval: DataSyncApprovalGrant | null = null,
now = Date.now(),
): boolean =>
(task.lifecycle === 'ready' || task.lifecycle === 'enabled') &&
isDataSyncPreflightCurrent(task, preflight) &&
(preflight?.approvalRequired === false ||
preflight?.approvalSatisfied === true ||
Boolean(
approval &&
approval.definitionHash === preflight?.definitionHash &&
Date.parse(approval.expiresAt) > now,
)) &&
(preflight?.status === 'passed' || preflight?.status === 'warning');

368
frontend/src/components/data-sync/wailsDto.test.ts Normal file
View File

@@ -0,0 +1,368 @@
import { describe, expect, it } from 'vitest';
import {
createDataSyncTableMapping,
createDataSyncTaskDraft,
reviseDataSyncTask,
} from './model';
import {
DataSyncGatewayProtocolError,
decodeDataSyncJobDefinition,
decodeObjectMetadata,
decodeDataSyncPreflightQuery,
decodeRouteCapability,
encodeDataSyncJobDefinition,
requireWailsQueryData,
} from './wailsDto';
const configuredTask = (
kind: 'migration' | 'reconcile' | 'querySink' | 'compare' | 'cdc' = 'reconcile',
) => {
const draft = createDataSyncTaskDraft({
id: `persisted-${kind}`,
kind,
name: `${kind} task`,
now: '2026-08-08T00:00:00.000Z',
});
return reviseDataSyncTask(
draft,
{
lifecycle: 'ready',
source: {
connectionId: 'source-id',
connectionName: 'Source',
type: 'mysql',
database: 'sales',
schema: 'public',
},
target: {
connectionId: 'target-id',
connectionName: 'Target',
type: 'postgresql',
database: 'warehouse',
schema: 'ods',
},
mappings: [
{
...createDataSyncTableMapping('map-1', 'public.orders', 'ods.orders'),
targetMode: 'create_or_reuse',
keyColumns: ['id'],
fields: [
{
id: 'field-1',
sourceField: 'name',
targetField: 'name',
sourceType: 'varchar',
targetType: 'text',
transform: 'trim',
transformArgument: '{"unicode":true}',
nullable: false,
},
],
},
],
},
'2026-08-08T00:01:00.000Z',
);
};
describe('data sync Wails DTO boundary', () => {
it('decodes optional object size metadata without inventing missing values', () => {
expect(
decodeObjectMetadata(
[
{
Table: 'orders',
Rows: '1250',
Data_length: '65536',
Index_length: '16384',
},
{ Table: 'empty_table', Rows: '0' },
{ Table: 'unknown_size' },
],
'mysql',
),
).toEqual([
{
name: 'orders',
kind: 'table',
rowCount: 1250,
dataBytes: 65536,
indexBytes: 16384,
},
{ name: 'empty_table', kind: 'table', rowCount: 0 },
{ name: 'unknown_size', kind: 'table' },
]);
});
it('decodes route target capabilities and defaults absent flags to false', () => {
expect(
decodeRouteCapability({
supportLevel: 'full',
canExecute: true,
supportsAutoCreate: true,
supportsAutoAddColumns: true,
requiresExistingTarget: true,
}),
).toMatchObject({
level: 'full',
canExecute: true,
supportsAutoCreate: true,
supportsAutoAddColumns: true,
requiresExistingTarget: true,
});
expect(
decodeRouteCapability({
supportLevel: 'partial',
canExecute: true,
supportsAutoCreate: false,
}),
).toMatchObject({
supportsAutoAddColumns: false,
requiresExistingTarget: false,
});
});
it('round-trips versioned delivery, mapping, schedule, and target strategy fields', () => {
const task = reviseDataSyncTask(configuredTask(), {
delivery: {
...configuredTask().delivery,
writeMode: 'upsert',
errorPolicy: 'stop',
batchSize: 250,
commitEvery: 250,
retryLimit: 4,
retryBackoffMs: 750,
autoAddColumns: true,
createIndexes: true,
propagateDeletes: true,
captureErrorPayload: false,
},
trigger: {
mode: 'cron',
expression: '0 */5 * * * *',
timezone: 'Asia/Shanghai',
overlap: 'queue',
},
concurrencyPolicy: 'queue',
resumePolicy: 'auto',
});
const wire = encodeDataSyncJobDefinition(task, {
approval: { definitionHash: 'untrusted' },
});
expect(wire).not.toHaveProperty('approval');
expect(wire).toMatchObject({
kind: 'reconcile',
incrementalMode: 'snapshot',
concurrencyPolicy: 'queue',
resumePolicy: 'auto',
options: {
syncMode: 'insert_update',
targetTableStrategy: 'smart',
batchSize: 250,
maxRetries: 4,
retryBackoffMillis: 750,
},
schedule: { kind: 'cron', cronExpression: '0 */5 * * * *' },
});
expect(decodeDataSyncJobDefinition(wire)).toMatchObject({
kind: 'reconcile',
delivery: {
writeMode: 'upsert',
batchSize: 250,
retryLimit: 4,
retryBackoffMs: 750,
},
trigger: { mode: 'cron', overlap: 'queue' },
concurrencyPolicy: 'queue',
resumePolicy: 'auto',
});
});
it('maps query sink, compare content, and CDC into backend wire values', () => {
const querySink = reviseDataSyncTask(configuredTask('querySink'), {
sourceQuery: 'SELECT id FROM orders',
});
const compare = configuredTask('compare');
const cdcBase = configuredTask('cdc');
const cdc = reviseDataSyncTask(cdcBase, {
incremental: {
mode: 'cdc',
initialSnapshot: false,
startPosition: 'latest',
adapter: 'mongodb-change-stream',
slotName: '',
publicationName: '',
},
trigger: { mode: 'continuous' },
});
expect(encodeDataSyncJobDefinition(querySink)).toMatchObject({
kind: 'query_sink',
sourceQuery: 'SELECT id FROM orders',
options: { syncMode: 'insert_only', maxRetries: 0 },
});
expect(encodeDataSyncJobDefinition(compare)).toMatchObject({
kind: 'compare',
options: { content: 'data' },
});
expect(encodeDataSyncJobDefinition(cdc)).toMatchObject({
kind: 'reconcile',
incrementalMode: 'cdc',
cdc: {
adapter: 'mongodb-change-stream',
initialSnapshot: false,
startPosition: 'latest',
},
});
});
it('fails closed for malformed QueryResult and unsafe policies', () => {
expect(() => requireWailsQueryData({ success: true }, 'Example')).toThrow(
DataSyncGatewayProtocolError,
);
expect(() => requireWailsQueryData({ success: false, data: [] }, 'Example')).toThrow(
DataSyncGatewayProtocolError,
);
const snapshot = configuredTask('migration');
expect(() =>
encodeDataSyncJobDefinition(
reviseDataSyncTask(snapshot, {
delivery: { ...snapshot.delivery, errorPolicy: 'skip' },
}),
),
).toThrow('row isolation requires');
const querySink = configuredTask('querySink');
expect(() =>
encodeDataSyncJobDefinition(
reviseDataSyncTask(querySink, {
delivery: { ...querySink.delivery, writeMode: 'none' },
}),
),
).toThrow('requires an explicit delivery mode');
expect(() =>
encodeDataSyncJobDefinition(
reviseDataSyncTask(querySink, {
delivery: { ...querySink.delivery, retryLimit: 1 },
}),
),
).toThrow('requires retryLimit 0');
});
it('encodes safe snapshot row isolation and preserves per-mapping watermarks', () => {
const base = configuredTask('reconcile');
const safeIsolation = reviseDataSyncTask(base, {
mappings: base.mappings.map((mapping) => ({
...mapping,
targetMode: 'existing_only',
})),
delivery: {
...base.delivery,
errorPolicy: 'quarantine',
captureErrorPayload: true,
},
});
expect(encodeDataSyncJobDefinition(safeIsolation)).toMatchObject({
options: { errorPolicy: 'skip_row', captureErrorPayload: true },
});
const second = {
...createDataSyncTableMapping('map-2', 'public.customers', 'ods.customers'),
keyColumns: ['customer_id'],
watermark: { column: 'modified_at', tieBreaker: 'customer_id' },
};
const watermark = reviseDataSyncTask(base, {
incremental: {
mode: 'watermark',
column: 'updated_at',
tieBreaker: 'id',
overlapWindowMs: 0,
},
mappings: [
{
...base.mappings[0],
watermark: { column: 'updated_at', tieBreaker: 'id' },
},
second,
],
});
const decoded = decodeDataSyncJobDefinition(
encodeDataSyncJobDefinition(watermark),
);
expect(decoded.mappings.map((mapping) => mapping.watermark)).toEqual([
{ column: 'updated_at', tieBreaker: 'id' },
{ column: 'modified_at', tieBreaker: 'customer_id' },
]);
});
it('accepts only a structured blocked payload when preflight success is false', () => {
const task = configuredTask();
const definition = encodeDataSyncJobDefinition(task);
const blocked = decodeDataSyncPreflightQuery(
{
success: false,
message: 'blocked',
data: {
status: 'blocked',
definition,
definitionHash: 'hash',
approvalRequired: false,
capability: {
supportLevel: 'full',
canExecute: true,
supportsAutoCreate: true,
},
issues: [
{
code: 'route_unsupported',
severity: 'blocker',
stage: 'endpoints',
message: 'unsupported route',
},
],
checkedAt: Date.parse('2026-08-08T00:02:00.000Z'),
},
},
task,
);
expect(blocked.snapshot).toMatchObject({
status: 'blocked',
approvalSatisfied: false,
issues: [{ message: 'unsupported route' }],
});
const earlyBlocked = decodeDataSyncPreflightQuery(
{
success: false,
data: {
status: 'blocked',
definition,
approvalRequired: false,
issues: [
{
code: 'definition_invalid',
severity: 'blocker',
stage: 'endpoints',
message: 'invalid definition',
},
],
checkedAt: Date.parse('2026-08-08T00:02:00.000Z'),
},
},
task,
);
expect(earlyBlocked.capability).toMatchObject({
level: 'unknown',
canExecute: false,
});
expect(() =>
decodeDataSyncPreflightQuery(
{ success: false, data: { ...blocked, status: 'passed' } },
task,
),
).toThrow(DataSyncGatewayProtocolError);
});
});

1192
frontend/src/components/data-sync/wailsDto.ts Normal file
View File

File diff suppressed because it is too large Load Diff

465
frontend/src/components/data-sync/wailsGateway.test.ts Normal file
View File

@@ -0,0 +1,465 @@
import { describe, expect, it, vi } from 'vitest';
import {
createDataSyncTableMapping,
createDataSyncTaskDraft,
reviseDataSyncTask,
} from './model';
import {
createWailsDataSyncWorkbenchGateway,
type WailsDataSyncApi,
} from './wailsGateway';
import { encodeDataSyncJobDefinition } from './wailsDto';
const NOW = Date.parse('2030-08-08T00:00:20.000Z');
const taskFixture = () => {
const draft = createDataSyncTaskDraft({
id: 'persisted-task',
kind: 'reconcile',
name: 'Orders sync',
now: '2030-08-08T00:00:00.000Z',
});
return reviseDataSyncTask(draft, {
lifecycle: 'ready',
source: {
connectionId: 'source-id',
connectionName: 'Source',
type: 'mysql',
database: 'sales',
schema: '',
},
target: {
connectionId: 'target-id',
connectionName: 'Target',
type: 'postgresql',
database: 'warehouse',
schema: 'ods',
},
mappings: [
{
...createDataSyncTableMapping('map-1', 'orders', 'ods.orders'),
keyColumns: ['id'],
},
],
});
};
const success = (data?: unknown) => ({ success: true, message: '', data });
const apiFixture = (
overrides: Partial<WailsDataSyncApi> = {},
): WailsDataSyncApi => ({
GetSavedConnections: vi.fn(async () => [
{
id: 'source-id',
name: 'Source',
config: {
type: 'mysql',
host: 'sanitized-host',
user: '',
password: '',
readOnly: false,
protection: {},
},
},
]),
DataSyncDatabaseList: vi.fn(async () =>
success([{ Database: 'sales' }]),
),
DataSyncObjectList: vi.fn(async () => success([{ Table: 'orders' }])),
DataSyncFieldList: vi.fn(async () =>
success([{ name: 'id', type: 'bigint', nullable: 'NO', key: 'PRI' }]),
),
DataSyncCapabilityResolve: vi.fn(async () =>
success({
supportLevel: 'full',
canExecute: true,
supportsAutoCreate: true,
}),
),
DataSyncCDCAdapterList: vi.fn(async () => success(['mongodb-change-stream'])),
DataSyncCDCProbe: vi.fn(async () =>
success({
adapter: 'mongodb-change-stream',
supported: true,
ready: true,
reason: '',
}),
),
DataSyncCheckpointGet: vi.fn(async () => ({
success: false,
message: 'data sync job record not found',
})),
DataSyncCheckpointReset: vi.fn(async () => success()),
DataSyncErrorRowDiscard: vi.fn(async () => success()),
DataSyncErrorRowList: vi.fn(async () => success([])),
DataSyncErrorRowRetry: vi.fn(async () => success()),
DataSyncJobApprovalBegin: vi.fn(async () =>
success({
challenge: 'server-challenge',
notBefore: NOW + 10_000,
expiresAt: NOW + 120_000,
}),
),
DataSyncJobApprove: vi.fn(async () =>
success({
token: 'one-time-token',
expiresAt: NOW + 600_000,
}),
),
DataSyncJobList: vi.fn(async () => success([])),
DataSyncJobPreflight: vi.fn(async (definition) =>
success({
status: 'passed',
definition,
definitionHash: 'definition-hash',
approvalRequired: true,
capability: {
supportLevel: 'full',
canExecute: true,
supportsAutoCreate: true,
},
issues: [],
checkedAt: NOW,
}),
),
DataSyncJobSave: vi.fn(async (definition) => success(definition)),
DataSyncRunCancel: vi.fn(async () => success()),
DataSyncRunList: vi.fn(async () => success([])),
DataSyncRunResume: vi.fn(async () => success(runFixture('resume-run', 'resume'))),
DataSyncRunRetry: vi.fn(async () => success(runFixture('retry-run', 'retry'))),
DataSyncRunStart: vi.fn(async () => success(runFixture('run-1', 'manual'))),
...overrides,
});
const runFixture = (id: string, trigger: string) => ({
id,
jobId: 'persisted-task',
trigger,
status: 'queued',
attempt: 1,
resumable: false,
message: '',
queuedAt: NOW,
startedAt: 0,
finishedAt: 0,
rowsInserted: 0,
rowsUpdated: 0,
rowsDeleted: 0,
rowsFailed: 0,
});
const errorRowFixture = (status = 'pending') => ({
id: 'error-row-1',
runId: 'run-1',
jobId: 'persisted-task',
sourceTable: 'orders',
targetTable: 'ods.orders',
operation: 'insert',
payloadPolicy: 'full',
error: 'invalid timestamp',
status,
});
const preflightData = (task = taskFixture(), approvalRequired = true) => ({
status: 'passed',
definition: encodeDataSyncJobDefinition(task),
definitionHash: 'definition-hash',
approvalRequired,
capability: {
supportLevel: 'full',
canExecute: true,
supportsAutoCreate: true,
},
issues: [],
checkedAt: NOW,
});
describe('real Wails data sync gateway', () => {
it('returns localized-form validation codes before calling backend preflight', async () => {
const base = taskFixture();
const task = reviseDataSyncTask(base, {
mappings: [createDataSyncTableMapping('map-empty', '', '')],
});
const api = apiFixture();
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => NOW });
const preflight = await gateway.preflightTask(task);
expect(preflight).toMatchObject({
status: 'blocked',
definitionHash: '',
approvalRequired: false,
});
expect(preflight.issues.map((issue) => issue.code)).toEqual(
expect.arrayContaining([
'source_object_required',
'target_object_required',
'key_columns_required',
]),
);
expect(api.DataSyncJobPreflight).not.toHaveBeenCalled();
});
it('uses only saved connection IDs for metadata and never forwards sanitized configs', async () => {
const api = apiFixture();
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => NOW });
const connections = await gateway.listSavedConnections();
await gateway.listDatabases('source-id');
await gateway.listObjects({
connectionId: 'source-id',
connectionName: 'Source',
type: 'mysql',
database: 'sales',
schema: 'public',
});
await gateway.listFields(
{
connectionId: 'source-id',
connectionName: 'Source',
type: 'mysql',
database: 'sales',
schema: 'public',
},
'public.orders',
);
expect(connections[0]).toEqual({
id: 'source-id',
name: 'Source',
type: 'mysql',
readable: true,
writable: true,
});
expect(api.DataSyncDatabaseList).toHaveBeenCalledWith('source-id');
expect(api.DataSyncObjectList).toHaveBeenCalledWith(
'source-id',
'sales',
'public',
);
expect(api.DataSyncFieldList).toHaveBeenCalledWith(
'source-id',
'sales',
'public',
'orders',
);
});
it('accepts a structured blocked preflight while rejecting malformed success payloads', async () => {
const task = taskFixture();
const blockedApi = apiFixture({
DataSyncJobPreflight: vi.fn(async (definition) => ({
success: false,
message: 'blocked',
data: {
...preflightData(task, false),
status: 'blocked',
definition,
issues: [
{
code: 'route_unsupported',
severity: 'blocker',
stage: 'endpoints',
message: 'route unavailable',
},
],
},
})),
});
const blockedGateway = createWailsDataSyncWorkbenchGateway({
api: blockedApi,
now: () => NOW,
});
expect(await blockedGateway.preflightTask(task)).toMatchObject({
status: 'blocked',
issues: [{ message: 'route unavailable' }],
});
const malformedGateway = createWailsDataSyncWorkbenchGateway({
api: apiFixture({
DataSyncJobList: vi.fn(async () => ({ success: true, message: '' })),
}),
now: () => NOW,
});
await expect(malformedGateway.listTasks()).rejects.toThrow('omitted data');
});
it('mints a memory-only token only after countdown and consumes it once on start', async () => {
const task = taskFixture();
const api = apiFixture();
let clock = NOW;
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => clock });
const preflight = await gateway.preflightTask(task);
const challenge = await gateway.beginApproval(task, preflight);
expect(challenge).toEqual({
definitionHash: 'definition-hash',
notBefore: new Date(NOW + 10_000).toISOString(),
expiresAt: new Date(NOW + 120_000).toISOString(),
});
expect(challenge).not.toHaveProperty('challenge');
await expect(
gateway.approveTask(task, preflight),
).rejects.toThrow('countdown is incomplete');
await gateway.beginApproval(task, preflight);
clock = NOW + 10_000;
const grant = await gateway.approveTask(task, preflight);
expect(grant).toEqual({
definitionHash: 'definition-hash',
expiresAt: new Date(NOW + 600_000).toISOString(),
});
expect(grant).not.toHaveProperty('token');
expect(api.DataSyncJobApprove).toHaveBeenCalledTimes(1);
const approvedDefinition = vi.mocked(api.DataSyncJobApprove).mock.calls[0][0];
expect(approvedDefinition.approval).toBeUndefined();
expect(api.DataSyncJobApprove).toHaveBeenCalledWith(
approvedDefinition,
'server-challenge',
);
await expect(gateway.startTask(task, preflight)).resolves.toMatchObject({
id: 'run-1',
status: 'queued',
});
expect(api.DataSyncRunStart).toHaveBeenCalledWith(
task.id,
task.revision,
'one-time-token',
);
await expect(gateway.startTask(task, preflight)).rejects.toThrow(
'explicit production approval is required',
);
expect(api.DataSyncRunStart).toHaveBeenCalledTimes(1);
});
it('binds authorization to the exact local signature and runnable lifecycle', async () => {
const task = taskFixture();
const api = apiFixture();
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => NOW });
const preflight = await gateway.preflightTask(task);
await gateway.beginApproval(task, preflight);
const paused = reviseDataSyncTask(task, { lifecycle: 'paused' });
await expect(gateway.startTask(paused, preflight)).rejects.toThrow(
'only ready or enabled tasks can run',
);
expect(api.DataSyncRunStart).not.toHaveBeenCalled();
});
it('allows pausing without production approval while ready/enabled saves stay guarded', async () => {
const task = taskFixture();
const api = apiFixture();
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => NOW });
const paused = reviseDataSyncTask(task, { lifecycle: 'paused' });
await expect(gateway.saveTask(paused)).resolves.toMatchObject({
lifecycle: 'paused',
});
expect(api.DataSyncJobSave).toHaveBeenCalledWith(
expect.objectContaining({ lifecycle: 'paused' }),
'',
);
await expect(gateway.saveTask(task)).rejects.toThrow(
'run preflight again',
);
});
it('retries only a listed full-payload row at the current task revision', async () => {
const task = taskFixture();
const api = apiFixture({
DataSyncJobList: vi.fn(async () =>
success([encodeDataSyncJobDefinition(task)]),
),
DataSyncErrorRowList: vi.fn(async () => success([errorRowFixture()])),
DataSyncErrorRowRetry: vi.fn(async () =>
success(errorRowFixture('resolved')),
),
});
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => NOW });
await gateway.listTasks();
const [row] = await gateway.listErrorRows('run-1');
expect(gateway.capabilities.errorRowRetry).toBe(true);
expect(row).toMatchObject({
taskId: task.id,
retryable: true,
status: 'pending',
});
await expect(gateway.retryErrorRow(row.id)).resolves.toMatchObject({
id: row.id,
status: 'resolved',
});
expect(api.DataSyncErrorRowRetry).toHaveBeenCalledWith(
row.id,
task.revision,
'',
);
await expect(gateway.retryErrorRow(row.id)).rejects.toThrow(
'capture its full payload before retrying',
);
expect(api.DataSyncErrorRowRetry).toHaveBeenCalledTimes(1);
});
it('consumes an explicit production approval token once for error-row retry', async () => {
const task = taskFixture();
let clock = NOW;
const api = apiFixture({
DataSyncJobList: vi.fn(async () =>
success([encodeDataSyncJobDefinition(task)]),
),
DataSyncErrorRowList: vi.fn(async () => success([errorRowFixture()])),
DataSyncErrorRowRetry: vi.fn(async () =>
success(errorRowFixture('resolved')),
),
});
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => clock });
await gateway.listTasks();
const preflight = await gateway.preflightTask(task);
await gateway.beginApproval(task, preflight);
clock = NOW + 10_000;
await gateway.approveTask(task, preflight);
const [row] = await gateway.listErrorRows('run-1');
await gateway.retryErrorRow(row.id);
expect(api.DataSyncErrorRowRetry).toHaveBeenCalledWith(
row.id,
task.revision,
'one-time-token',
);
});
it('passes the paused task revision when resetting a checkpoint and refreshes the cached task', async () => {
const task = { ...taskFixture(), lifecycle: 'paused' as const };
const saved = { ...task, revision: task.revision + 1 };
const api = apiFixture({
DataSyncJobList: vi.fn(async () =>
success([encodeDataSyncJobDefinition(task)]),
),
DataSyncCheckpointReset: vi.fn(async () =>
success(encodeDataSyncJobDefinition(saved)),
),
});
const gateway = createWailsDataSyncWorkbenchGateway({ api, now: () => NOW });
await gateway.listTasks();
await expect(
gateway.resetCheckpoint(task.id, task.revision),
).resolves.toMatchObject({
id: task.id,
lifecycle: 'paused',
revision: saved.revision,
});
expect(api.DataSyncCheckpointReset).toHaveBeenCalledWith(
task.id,
task.revision,
);
await expect(
gateway.resetCheckpoint(task.id, task.revision),
).rejects.toThrow('current paused task revision');
expect(api.DataSyncCheckpointReset).toHaveBeenCalledTimes(1);
});
});

791
frontend/src/components/data-sync/wailsGateway.ts Normal file
View File

@@ -0,0 +1,791 @@
import * as WailsApp from '../../../wailsjs/go/app/App';
import { syncjob } from '../../../wailsjs/go/models';
import type { DataSyncWorkbenchGateway } from './gateway';
import type {
DataSyncApprovalChallenge,
DataSyncApprovalGrant,
DataSyncCheckpointSummary,
DataSyncRouteCapability,
DataSyncRunRecord,
DataSyncTaskDefinition,
} from './model';
import { validateDataSyncTask } from './model';
import {
cdcSourceFromProbe,
decodeCDCAdapters,
decodeCDCProbe,
decodeCheckpoint,
decodeDataSyncApproval,
decodeDataSyncApprovalChallenge,
decodeDataSyncJobDefinition,
decodeDataSyncPreflightQuery,
decodeDatabaseMetadata,
decodeErrorRow,
decodeFieldMetadata,
decodeObjectMetadata,
decodeRouteCapability,
decodeRunRecord,
decodeSavedConnectionViews,
decodeScheduleSummary,
encodeDataSyncJobDefinition,
isLocalDataSyncTaskId,
requireWailsCommandSuccess,
requireWailsQueryData,
DataSyncGatewayProtocolError,
type WailsDataSyncJobDefinition,
type WailsQueryResultLike,
} from './wailsDto';
type QueryResultPromise = Promise<WailsQueryResultLike>;
/** Narrow seam around Wails so protocol handling can be tested without a runtime. */
export interface WailsDataSyncApi {
GetSavedConnections(): Promise<unknown>;
DataSyncDatabaseList(connectionId: string): QueryResultPromise;
DataSyncObjectList(
connectionId: string,
database: string,
schema: string,
): QueryResultPromise;
DataSyncFieldList(
connectionId: string,
database: string,
schema: string,
objectName: string,
): QueryResultPromise;
DataSyncCapabilityResolve(
sourceConnectionId: string,
sourceDatabase: string,
sourceSchema: string,
targetConnectionId: string,
targetDatabase: string,
targetSchema: string,
): QueryResultPromise;
DataSyncCDCAdapterList(): QueryResultPromise;
DataSyncCDCProbe(
connectionId: string,
database: string,
schema: string,
adapter: string,
): QueryResultPromise;
DataSyncCheckpointGet(taskId: string): QueryResultPromise;
DataSyncCheckpointReset(
taskId: string,
expectedJobRevision: number,
): QueryResultPromise;
DataSyncErrorRowDiscard(errorRowId: string): QueryResultPromise;
DataSyncErrorRowRetry(
errorRowId: string,
expectedJobRevision: number,
approvalToken: string,
): QueryResultPromise;
DataSyncErrorRowList(
runId: string,
status: string,
limit: number,
): QueryResultPromise;
DataSyncJobApprovalBegin(definition: syncjob.JobDefinition): QueryResultPromise;
DataSyncJobApprove(
definition: syncjob.JobDefinition,
challenge: string,
): QueryResultPromise;
DataSyncJobList(): QueryResultPromise;
DataSyncJobPreflight(definition: syncjob.JobDefinition): QueryResultPromise;
DataSyncJobSave(
definition: syncjob.JobDefinition,
approvalToken: string,
): QueryResultPromise;
DataSyncRunCancel(runId: string): QueryResultPromise;
DataSyncRunList(taskId: string, limit: number): QueryResultPromise;
DataSyncRunResume(runId: string): QueryResultPromise;
DataSyncRunRetry(runId: string): QueryResultPromise;
DataSyncRunStart(
taskId: string,
expectedRevision: number,
approvalToken: string,
): QueryResultPromise;
}
type GatewayOptions = {
api?: WailsDataSyncApi;
now?: () => number;
};
type CachedPreflight = {
taskRevision: number;
taskSignature: string;
definitionHash: string;
approvalRequired: boolean;
canExecute: boolean;
definition: WailsDataSyncJobDefinition;
};
type ApprovalToken = {
token: string;
expiresAt: string;
definitionHash: string;
taskSignature: string;
};
type ApprovalChallenge = {
challenge: string;
notBefore: string;
expiresAt: string;
definitionHash: string;
taskSignature: string;
};
const UNKNOWN_CAPABILITY: DataSyncRouteCapability = {
level: 'unknown',
canExecute: false,
supportsAutoCreate: false,
supportsCdc: false,
};
const asApi = (): WailsDataSyncApi =>
WailsApp as unknown as WailsDataSyncApi;
const asJobDefinition = (
value: WailsDataSyncJobDefinition,
): syncjob.JobDefinition => new syncjob.JobDefinition(value);
const taskSignature = (
task: DataSyncTaskDefinition,
previous?: WailsDataSyncJobDefinition,
): string => JSON.stringify(encodeDataSyncJobDefinition(task, previous));
const sanitizedWireDefinition = (
value: unknown,
): WailsDataSyncJobDefinition => {
if (!value || typeof value !== 'object' || Array.isArray(value)) {
throw new DataSyncGatewayProtocolError('job', 'expected object');
}
const sanitized = { ...(value as WailsDataSyncJobDefinition) };
// Backend approval evidence is never retained or reflected into the UI.
delete sanitized.approval;
return sanitized;
};
const queryFailureMessage = (result: WailsQueryResultLike): string =>
typeof result?.message === 'string' ? result.message.trim() : '';
const isCheckpointMissing = (result: WailsQueryResultLike): boolean =>
result?.success === false &&
queryFailureMessage(result) === 'data sync job record not found';
const stripEndpointSchema = (schema: string, objectName: string): string => {
const prefix = `${schema.trim()}.`;
return prefix !== '.' && objectName.trim().startsWith(prefix)
? objectName.trim().slice(prefix.length)
: objectName.trim();
};
const isCurrentPreflight = (
cached: CachedPreflight | undefined,
task: DataSyncTaskDefinition,
previous?: WailsDataSyncJobDefinition,
): cached is CachedPreflight =>
Boolean(
cached &&
cached.taskRevision === task.revision &&
cached.taskSignature === taskSignature(task, previous),
);
export const createWailsDataSyncWorkbenchGateway = (
options: GatewayOptions = {},
): DataSyncWorkbenchGateway => {
const api = options.api || asApi();
const now = options.now || Date.now;
const wireJobs = new Map<string, WailsDataSyncJobDefinition>();
const taskNames = new Map<string, string>();
const preflights = new Map<string, CachedPreflight>();
const approvalTokens = new Map<string, ApprovalToken>();
const approvalChallenges = new Map<string, ApprovalChallenge>();
const errorRows = new Map<string, ReturnType<typeof decodeErrorRow>>();
const connectionTypes = new Map<string, string>();
let tasksLoaded = false;
const decodeRuns = (value: unknown): DataSyncRunRecord[] => {
if (!Array.isArray(value)) {
throw new DataSyncGatewayProtocolError('DataSyncRunList.data', 'expected array');
}
return value.map((run) => decodeRunRecord(run, taskNames));
};
const takeApprovalToken = (
task: DataSyncTaskDefinition,
preflight: CachedPreflight,
): string => {
const approval = approvalTokens.get(task.id);
if (
!approval ||
approval.definitionHash !== preflight.definitionHash ||
approval.taskSignature !== preflight.taskSignature ||
Date.parse(approval.expiresAt) <= now()
) {
approvalTokens.delete(task.id);
return '';
}
// The backend token is one-time. Remove it before crossing the boundary so
// retries can never accidentally reuse a token with uncertain outcome.
approvalTokens.delete(task.id);
return approval.token;
};
const requireCurrentPreflight = (
task: DataSyncTaskDefinition,
): CachedPreflight => {
const previous = wireJobs.get(task.id);
const cached = preflights.get(task.id);
if (!isCurrentPreflight(cached, task, previous)) {
throw new DataSyncGatewayProtocolError(
'data sync preflight',
'task definition changed; run preflight again',
);
}
return cached;
};
const getCheckpoint = async (
taskId: string,
): Promise<DataSyncCheckpointSummary | null> => {
if (!taskId.trim() || isLocalDataSyncTaskId(taskId)) return null;
const result = await api.DataSyncCheckpointGet(taskId);
if (isCheckpointMissing(result)) return null;
return decodeCheckpoint(requireWailsQueryData(result, 'DataSyncCheckpointGet'));
};
const gateway: DataSyncWorkbenchGateway = {
capabilities: { errorRowRetry: true },
async listSavedConnections() {
const connections = decodeSavedConnectionViews(
await api.GetSavedConnections(),
);
connectionTypes.clear();
connections.forEach((connection) => {
connectionTypes.set(connection.id, connection.type);
});
return connections;
},
async listDatabases(connectionId) {
return decodeDatabaseMetadata(
requireWailsQueryData(
await api.DataSyncDatabaseList(connectionId),
'DataSyncDatabaseList',
),
);
},
async listObjects(endpoint) {
return decodeObjectMetadata(
requireWailsQueryData(
await api.DataSyncObjectList(
endpoint.connectionId,
endpoint.database,
endpoint.schema,
),
'DataSyncObjectList',
),
endpoint.type || connectionTypes.get(endpoint.connectionId) || '',
);
},
async listFields(endpoint, objectName) {
return decodeFieldMetadata(
requireWailsQueryData(
await api.DataSyncFieldList(
endpoint.connectionId,
endpoint.database,
endpoint.schema,
stripEndpointSchema(endpoint.schema, objectName),
),
'DataSyncFieldList',
),
);
},
async listTasks() {
const value = requireWailsQueryData(
await api.DataSyncJobList(),
'DataSyncJobList',
);
if (!Array.isArray(value)) {
throw new DataSyncGatewayProtocolError('DataSyncJobList.data', 'expected array');
}
wireJobs.clear();
taskNames.clear();
const tasks = value.map((item) => {
const wire = sanitizedWireDefinition(item);
const task = decodeDataSyncJobDefinition(wire);
wireJobs.set(task.id, wire);
taskNames.set(task.id, task.name);
return task;
});
tasksLoaded = true;
return tasks;
},
async saveTask(task) {
const previous = wireJobs.get(task.id);
let definition = encodeDataSyncJobDefinition(task, previous);
let token = '';
if (task.lifecycle === 'ready' || task.lifecycle === 'enabled') {
const preflight = requireCurrentPreflight(task);
definition = preflight.definition;
if (preflight.approvalRequired) {
token = takeApprovalToken(task, preflight);
if (!token) {
throw new DataSyncGatewayProtocolError(
'DataSyncJobSave',
'explicit production approval is required',
);
}
}
}
const savedValue = requireWailsQueryData(
await api.DataSyncJobSave(asJobDefinition(definition), token),
'DataSyncJobSave',
);
const savedWire = sanitizedWireDefinition(savedValue);
const saved = decodeDataSyncJobDefinition(savedWire);
wireJobs.delete(task.id);
taskNames.delete(task.id);
preflights.delete(task.id);
wireJobs.set(saved.id, savedWire);
taskNames.set(saved.id, saved.name);
return saved;
},
async resolveCapability(task) {
if (!task.source.connectionId || !task.target.connectionId) {
return { ...UNKNOWN_CAPABILITY };
}
const base = decodeRouteCapability(
requireWailsQueryData(
await api.DataSyncCapabilityResolve(
task.source.connectionId,
task.source.database,
task.source.schema,
task.target.connectionId,
task.target.database,
task.target.schema,
),
'DataSyncCapabilityResolve',
),
);
if (task.kind !== 'cdc') return base;
if (task.incremental.mode !== 'cdc' || !task.incremental.adapter) {
return { ...base, canExecute: false, supportsAutoCreate: false, supportsCdc: false };
}
const probe = decodeCDCProbe(
requireWailsQueryData(
await api.DataSyncCDCProbe(
task.source.connectionId,
task.source.database,
task.source.schema,
task.incremental.adapter,
),
'DataSyncCDCProbe',
),
);
return {
...base,
canExecute: base.canExecute && probe.supported && probe.ready,
supportsAutoCreate: false,
supportsCdc: probe.supported,
};
},
async preflightTask(task) {
const localIssues = validateDataSyncTask(task);
if (localIssues.some((issue) => issue.severity === 'blocker')) {
preflights.delete(task.id);
approvalChallenges.delete(task.id);
approvalTokens.delete(task.id);
return {
taskId: task.id,
taskRevision: task.revision,
status: 'blocked',
issues: localIssues,
definitionHash: '',
approvalRequired: false,
approvalSatisfied: false,
checkedAt: new Date(now()).toISOString(),
};
}
const previous = wireJobs.get(task.id);
const input = encodeDataSyncJobDefinition(task, previous);
const decoded = decodeDataSyncPreflightQuery(
await api.DataSyncJobPreflight(asJobDefinition(input)),
task,
);
if (task.kind === 'compare' && !decoded.capability.canExecute) {
decoded.snapshot.status = 'blocked';
decoded.snapshot.issues.push({
id: 'compare-capability-unsupported',
code: 'compare_route_unsupported',
severity: 'blocker',
stage: 'endpoints',
message: 'the selected source-target route cannot execute compare tasks',
});
}
const definition = sanitizedWireDefinition(decoded.definition);
approvalChallenges.delete(task.id);
approvalTokens.delete(task.id);
preflights.set(task.id, {
taskRevision: task.revision,
taskSignature: JSON.stringify(input),
definitionHash: decoded.snapshot.definitionHash,
approvalRequired: decoded.snapshot.approvalRequired,
canExecute: decoded.capability.canExecute,
definition,
});
return decoded.snapshot;
},
async beginApproval(task, preflight): Promise<DataSyncApprovalChallenge> {
const cached = requireCurrentPreflight(task);
if (
!preflight.approvalRequired ||
preflight.status === 'blocked' ||
cached.definitionHash !== preflight.definitionHash
) {
throw new DataSyncGatewayProtocolError(
'DataSyncJobApprovalBegin',
'approval does not match the current passed preflight',
);
}
const challenge = decodeDataSyncApprovalChallenge(
requireWailsQueryData(
await api.DataSyncJobApprovalBegin(asJobDefinition(cached.definition)),
'DataSyncJobApprovalBegin',
),
);
if (Date.parse(challenge.expiresAt) <= now()) {
throw new DataSyncGatewayProtocolError(
'DataSyncJobApprovalBegin',
'approval challenge expired before it could be stored',
);
}
approvalChallenges.set(task.id, {
...challenge,
definitionHash: preflight.definitionHash,
taskSignature: cached.taskSignature,
});
return {
definitionHash: preflight.definitionHash,
notBefore: challenge.notBefore,
expiresAt: challenge.expiresAt,
};
},
async approveTask(task, preflight): Promise<DataSyncApprovalGrant> {
const cached = requireCurrentPreflight(task);
const challenge = approvalChallenges.get(task.id);
if (
!challenge ||
challenge.definitionHash !== preflight.definitionHash ||
challenge.definitionHash !== cached.definitionHash ||
challenge.taskSignature !== cached.taskSignature ||
Date.parse(challenge.notBefore) > now() ||
Date.parse(challenge.expiresAt) <= now()
) {
approvalChallenges.delete(task.id);
throw new DataSyncGatewayProtocolError(
'DataSyncJobApprove',
'backend approval countdown is incomplete, expired, or stale',
);
}
// The backend challenge is also one-time. Remove it before crossing the
// boundary so an uncertain response can never be replayed.
approvalChallenges.delete(task.id);
const approved = decodeDataSyncApproval(
requireWailsQueryData(
await api.DataSyncJobApprove(
asJobDefinition(cached.definition),
challenge.challenge,
),
'DataSyncJobApprove',
),
);
if (Date.parse(approved.expiresAt) <= now()) {
throw new DataSyncGatewayProtocolError(
'DataSyncJobApprove',
'approval token expired before it could be stored',
);
}
approvalTokens.set(task.id, {
...approved,
definitionHash: preflight.definitionHash,
taskSignature: cached.taskSignature,
});
return {
definitionHash: preflight.definitionHash,
expiresAt: approved.expiresAt,
};
},
async startTask(task, preflight) {
if (isLocalDataSyncTaskId(task.id)) {
throw new DataSyncGatewayProtocolError(
'DataSyncRunStart',
'save the task before running it',
);
}
if (task.lifecycle !== 'ready' && task.lifecycle !== 'enabled') {
throw new DataSyncGatewayProtocolError(
'DataSyncRunStart',
'only ready or enabled tasks can run',
);
}
const cached = requireCurrentPreflight(task);
if (
cached.definitionHash !== preflight.definitionHash ||
preflight.status === 'blocked' ||
!cached.canExecute
) {
throw new DataSyncGatewayProtocolError(
'DataSyncRunStart',
'preflight is blocked or stale',
);
}
let token = '';
if (cached.approvalRequired) {
token = takeApprovalToken(task, cached);
if (!token) {
throw new DataSyncGatewayProtocolError(
'DataSyncRunStart',
'explicit production approval is required',
);
}
}
return decodeRunRecord(
requireWailsQueryData(
await api.DataSyncRunStart(task.id, task.revision, token),
'DataSyncRunStart',
),
taskNames,
);
},
async listRuns(taskId) {
return decodeRuns(
requireWailsQueryData(
await api.DataSyncRunList(taskId || '', 200),
'DataSyncRunList',
),
);
},
async listErrorRows(runId) {
const value = requireWailsQueryData(
await api.DataSyncErrorRowList(runId, '', 500),
'DataSyncErrorRowList',
);
if (!Array.isArray(value)) {
throw new DataSyncGatewayProtocolError(
'DataSyncErrorRowList.data',
'expected array',
);
}
return value.map((item) => {
const row = decodeErrorRow(item);
errorRows.set(row.id, row);
return row;
});
},
async listSchedules() {
if (!tasksLoaded) await gateway.listTasks();
return Array.from(wireJobs.values()).flatMap((job) => {
const schedule = decodeScheduleSummary(job);
return schedule ? [schedule] : [];
});
},
async listCdcAdapters() {
return decodeCDCAdapters(
requireWailsQueryData(
await api.DataSyncCDCAdapterList(),
'DataSyncCDCAdapterList',
),
);
},
async listCdcSources() {
const tasks =
!tasksLoaded
? await gateway.listTasks()
: Array.from(wireJobs.values()).map(decodeDataSyncJobDefinition);
const cdcTasks = tasks.filter(
(
task,
): task is DataSyncTaskDefinition & {
incremental: Extract<DataSyncTaskDefinition['incremental'], { mode: 'cdc' }>;
} => task.kind === 'cdc' && task.incremental.mode === 'cdc',
);
if (cdcTasks.length === 0) return [];
let adapters: string[] = [];
let adapterListError = '';
try {
adapters = await gateway.listCdcAdapters();
} catch (error) {
adapterListError = error instanceof Error ? error.message : String(error);
}
return Promise.all(
cdcTasks.map(async (task) => {
let checkpoint: DataSyncCheckpointSummary | null = null;
let checkpointError = '';
try {
checkpoint = await getCheckpoint(task.id);
} catch (error) {
checkpointError = error instanceof Error ? error.message : String(error);
}
if (adapterListError) {
return cdcSourceFromProbe(task, null, checkpoint, adapterListError);
}
if (!adapters.includes(task.incremental.adapter)) {
return {
...cdcSourceFromProbe(
task,
null,
checkpoint,
checkpointError || 'selected CDC adapter is not registered',
),
status: 'unsupported' as const,
};
}
try {
const probe = decodeCDCProbe(
requireWailsQueryData(
await api.DataSyncCDCProbe(
task.source.connectionId,
task.source.database,
task.source.schema,
task.incremental.adapter,
),
'DataSyncCDCProbe',
),
);
return cdcSourceFromProbe(task, probe, checkpoint, checkpointError);
} catch (error) {
const reason = error instanceof Error ? error.message : String(error);
return cdcSourceFromProbe(task, null, checkpoint, reason);
}
}),
);
},
getCheckpoint,
async resetCheckpoint(taskId, expectedJobRevision) {
const previous = wireJobs.get(taskId);
if (!previous) {
throw new DataSyncGatewayProtocolError(
'DataSyncCheckpointReset',
'the current task revision is unavailable; refresh tasks before resetting the checkpoint',
);
}
const task = decodeDataSyncJobDefinition(previous);
if (task.lifecycle !== 'paused' || task.revision !== expectedJobRevision) {
throw new DataSyncGatewayProtocolError(
'DataSyncCheckpointReset',
'checkpoint reset requires the current paused task revision',
);
}
const savedWire = sanitizedWireDefinition(
requireWailsQueryData(
await api.DataSyncCheckpointReset(taskId, expectedJobRevision),
'DataSyncCheckpointReset',
),
);
const saved = decodeDataSyncJobDefinition(savedWire);
wireJobs.set(saved.id, savedWire);
taskNames.set(saved.id, saved.name);
preflights.delete(taskId);
approvalTokens.delete(taskId);
approvalChallenges.delete(taskId);
return saved;
},
async cancelRun(runId) {
requireWailsCommandSuccess(
await api.DataSyncRunCancel(runId),
'DataSyncRunCancel',
);
},
async resumeRun(runId) {
return decodeRunRecord(
requireWailsQueryData(
await api.DataSyncRunResume(runId),
'DataSyncRunResume',
),
taskNames,
);
},
async retryRun(runId) {
return decodeRunRecord(
requireWailsQueryData(
await api.DataSyncRunRetry(runId),
'DataSyncRunRetry',
),
taskNames,
);
},
async discardErrorRow(errorRowId) {
requireWailsCommandSuccess(
await api.DataSyncErrorRowDiscard(errorRowId),
'DataSyncErrorRowDiscard',
);
},
async retryErrorRow(errorRowId) {
const row = errorRows.get(errorRowId);
if (!row || !row.retryable || row.status !== 'pending') {
throw new DataSyncGatewayProtocolError(
'DataSyncErrorRowRetry',
'refresh the error row and capture its full payload before retrying',
);
}
const previous = wireJobs.get(row.taskId);
if (!previous) {
throw new DataSyncGatewayProtocolError(
'DataSyncErrorRowRetry',
'the current task revision is unavailable; refresh tasks before retrying',
);
}
const task = decodeDataSyncJobDefinition(previous);
const cached = preflights.get(task.id);
let token = '';
if (
cached &&
isCurrentPreflight(cached, task, previous) &&
cached.approvalRequired &&
approvalTokens.has(task.id)
) {
token = takeApprovalToken(task, cached);
}
const retried = decodeErrorRow(
requireWailsQueryData(
await api.DataSyncErrorRowRetry(errorRowId, task.revision, token),
'DataSyncErrorRowRetry',
),
);
if (retried.id !== row.id || retried.taskId !== row.taskId) {
throw new DataSyncGatewayProtocolError(
'DataSyncErrorRowRetry.data',
'backend returned a different error row or task',
);
}
errorRows.set(retried.id, retried);
return retried;
},
};
return gateway;
};

57
frontend/wailsjs/go/app/App.d.ts vendored
View File

@@ -4,6 +4,7 @@ import {connection} from '../models';
import {sqlaudit} from '../models'; import {sqlaudit} from '../models';
import {app} from '../models'; import {app} from '../models';
import {sync} from '../models'; import {sync} from '../models';
import {syncjob} from '../models';
import {jvm} from '../models'; import {jvm} from '../models';
import {redis} from '../models'; import {redis} from '../models';
import {resultdiff} from '../models'; import {resultdiff} from '../models';
@@ -118,8 +119,64 @@ export function DataSync(arg1:sync.SyncConfig):Promise<sync.SyncResult>;
export function DataSyncAnalyze(arg1:sync.SyncConfig):Promise<connection.QueryResult>; export function DataSyncAnalyze(arg1:sync.SyncConfig):Promise<connection.QueryResult>;
export function DataSyncCDCAdapterList():Promise<connection.QueryResult>;
export function DataSyncCDCProbe(arg1:string,arg2:string,arg3:string,arg4:string):Promise<connection.QueryResult>;
export function DataSyncCapability(arg1:connection.ConnectionConfig,arg2:connection.ConnectionConfig):Promise<sync.MigrationCapability>;
export function DataSyncCapabilityResolve(arg1:string,arg2:string,arg3:string,arg4:string,arg5:string,arg6:string):Promise<connection.QueryResult>;
export function DataSyncCheckpointGet(arg1:string):Promise<connection.QueryResult>;
export function DataSyncCheckpointReset(arg1:string,arg2:number):Promise<connection.QueryResult>;
export function DataSyncDatabaseList(arg1:string):Promise<connection.QueryResult>;
export function DataSyncErrorRowDiscard(arg1:string):Promise<connection.QueryResult>;
export function DataSyncErrorRowGet(arg1:string):Promise<connection.QueryResult>;
export function DataSyncErrorRowList(arg1:string,arg2:string,arg3:number):Promise<connection.QueryResult>;
export function DataSyncErrorRowRetry(arg1:string,arg2:number,arg3:string):Promise<connection.QueryResult>;
export function DataSyncFieldList(arg1:string,arg2:string,arg3:string,arg4:string):Promise<connection.QueryResult>;
export function DataSyncJobApprovalBegin(arg1:syncjob.JobDefinition):Promise<connection.QueryResult>;
export function DataSyncJobApprove(arg1:syncjob.JobDefinition,arg2:string):Promise<connection.QueryResult>;
export function DataSyncJobDelete(arg1:string):Promise<connection.QueryResult>;
export function DataSyncJobGet(arg1:string):Promise<connection.QueryResult>;
export function DataSyncJobList():Promise<connection.QueryResult>;
export function DataSyncJobPreflight(arg1:syncjob.JobDefinition):Promise<connection.QueryResult>;
export function DataSyncJobSave(arg1:syncjob.JobDefinition,arg2:string):Promise<connection.QueryResult>;
export function DataSyncObjectList(arg1:string,arg2:string,arg3:string):Promise<connection.QueryResult>;
export function DataSyncPreview(arg1:sync.SyncConfig,arg2:string,arg3:number):Promise<connection.QueryResult>; export function DataSyncPreview(arg1:sync.SyncConfig,arg2:string,arg3:number):Promise<connection.QueryResult>;
export function DataSyncRunCancel(arg1:string):Promise<connection.QueryResult>;
export function DataSyncRunEventList(arg1:string,arg2:number,arg3:number):Promise<connection.QueryResult>;
export function DataSyncRunGet(arg1:string):Promise<connection.QueryResult>;
export function DataSyncRunList(arg1:string,arg2:number):Promise<connection.QueryResult>;
export function DataSyncRunResume(arg1:string):Promise<connection.QueryResult>;
export function DataSyncRunRetry(arg1:string):Promise<connection.QueryResult>;
export function DataSyncRunStart(arg1:string,arg2:number,arg3:string):Promise<connection.QueryResult>;
export function DataSyncSchedulePreview(arg1:syncjob.JobDefinition,arg2:number):Promise<connection.QueryResult>;
export function DeleteConnection(arg1:string):Promise<void>; export function DeleteConnection(arg1:string):Promise<void>;
export function DeleteQuery(arg1:string):Promise<void>; export function DeleteQuery(arg1:string):Promise<void>;

112
frontend/wailsjs/go/app/App.js
View File

@@ -222,10 +222,122 @@ export function DataSyncAnalyze(arg1) {
return window['go']['app']['App']['DataSyncAnalyze'](arg1); return window['go']['app']['App']['DataSyncAnalyze'](arg1);
} }
export function DataSyncCDCAdapterList() {
return window['go']['app']['App']['DataSyncCDCAdapterList']();
}
export function DataSyncCDCProbe(arg1, arg2, arg3, arg4) {
return window['go']['app']['App']['DataSyncCDCProbe'](arg1, arg2, arg3, arg4);
}
export function DataSyncCapability(arg1, arg2) {
return window['go']['app']['App']['DataSyncCapability'](arg1, arg2);
}
export function DataSyncCapabilityResolve(arg1, arg2, arg3, arg4, arg5, arg6) {
return window['go']['app']['App']['DataSyncCapabilityResolve'](arg1, arg2, arg3, arg4, arg5, arg6);
}
export function DataSyncCheckpointGet(arg1) {
return window['go']['app']['App']['DataSyncCheckpointGet'](arg1);
}
export function DataSyncCheckpointReset(arg1, arg2) {
return window['go']['app']['App']['DataSyncCheckpointReset'](arg1, arg2);
}
export function DataSyncDatabaseList(arg1) {
return window['go']['app']['App']['DataSyncDatabaseList'](arg1);
}
export function DataSyncErrorRowDiscard(arg1) {
return window['go']['app']['App']['DataSyncErrorRowDiscard'](arg1);
}
export function DataSyncErrorRowGet(arg1) {
return window['go']['app']['App']['DataSyncErrorRowGet'](arg1);
}
export function DataSyncErrorRowList(arg1, arg2, arg3) {
return window['go']['app']['App']['DataSyncErrorRowList'](arg1, arg2, arg3);
}
export function DataSyncErrorRowRetry(arg1, arg2, arg3) {
return window['go']['app']['App']['DataSyncErrorRowRetry'](arg1, arg2, arg3);
}
export function DataSyncFieldList(arg1, arg2, arg3, arg4) {
return window['go']['app']['App']['DataSyncFieldList'](arg1, arg2, arg3, arg4);
}
export function DataSyncJobApprovalBegin(arg1) {
return window['go']['app']['App']['DataSyncJobApprovalBegin'](arg1);
}
export function DataSyncJobApprove(arg1, arg2) {
return window['go']['app']['App']['DataSyncJobApprove'](arg1, arg2);
}
export function DataSyncJobDelete(arg1) {
return window['go']['app']['App']['DataSyncJobDelete'](arg1);
}
export function DataSyncJobGet(arg1) {
return window['go']['app']['App']['DataSyncJobGet'](arg1);
}
export function DataSyncJobList() {
return window['go']['app']['App']['DataSyncJobList']();
}
export function DataSyncJobPreflight(arg1) {
return window['go']['app']['App']['DataSyncJobPreflight'](arg1);
}
export function DataSyncJobSave(arg1, arg2) {
return window['go']['app']['App']['DataSyncJobSave'](arg1, arg2);
}
export function DataSyncObjectList(arg1, arg2, arg3) {
return window['go']['app']['App']['DataSyncObjectList'](arg1, arg2, arg3);
}
export function DataSyncPreview(arg1, arg2, arg3) { export function DataSyncPreview(arg1, arg2, arg3) {
return window['go']['app']['App']['DataSyncPreview'](arg1, arg2, arg3); return window['go']['app']['App']['DataSyncPreview'](arg1, arg2, arg3);
} }
export function DataSyncRunCancel(arg1) {
return window['go']['app']['App']['DataSyncRunCancel'](arg1);
}
export function DataSyncRunEventList(arg1, arg2, arg3) {
return window['go']['app']['App']['DataSyncRunEventList'](arg1, arg2, arg3);
}
export function DataSyncRunGet(arg1) {
return window['go']['app']['App']['DataSyncRunGet'](arg1);
}
export function DataSyncRunList(arg1, arg2) {
return window['go']['app']['App']['DataSyncRunList'](arg1, arg2);
}
export function DataSyncRunResume(arg1) {
return window['go']['app']['App']['DataSyncRunResume'](arg1);
}
export function DataSyncRunRetry(arg1) {
return window['go']['app']['App']['DataSyncRunRetry'](arg1);
}
export function DataSyncRunStart(arg1, arg2, arg3) {
return window['go']['app']['App']['DataSyncRunStart'](arg1, arg2, arg3);
}
export function DataSyncSchedulePreview(arg1, arg2) {
return window['go']['app']['App']['DataSyncSchedulePreview'](arg1, arg2);
}
export function DeleteConnection(arg1) { export function DeleteConnection(arg1) {
return window['go']['app']['App']['DeleteConnection'](arg1); return window['go']['app']['App']['DeleteConnection'](arg1);
} }

495
frontend/wailsjs/go/models.ts
View File

@@ -2704,7 +2704,163 @@ export namespace sqlaudit {
} }
export namespace sync { export namespace sync {
export class MigrationCapability {
sourceType: string;
targetType: string;
sourceModel: string;
targetModel: string;
planner: string;
supportLevel: string;
canExecute: boolean;
supportsAutoCreate: boolean;
supportsAutoAddColumns: boolean;
requiresExistingTarget: boolean;
static createFrom(source: any = {}) {
return new MigrationCapability(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.sourceType = source["sourceType"];
this.targetType = source["targetType"];
this.sourceModel = source["sourceModel"];
this.targetModel = source["targetModel"];
this.planner = source["planner"];
this.supportLevel = source["supportLevel"];
this.canExecute = source["canExecute"];
this.supportsAutoCreate = source["supportsAutoCreate"];
this.supportsAutoAddColumns = source["supportsAutoAddColumns"];
this.requiresExistingTarget = source["requiresExistingTarget"];
}
}
export class SyncValueTransform {
type: string;
args?: Record<string, string>;
static createFrom(source: any = {}) {
return new SyncValueTransform(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.type = source["type"];
this.args = source["args"];
}
}
export class SyncDefaultValue {
when?: string[];
valueType?: string;
value?: string;
static createFrom(source: any = {}) {
return new SyncDefaultValue(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.when = source["when"];
this.valueType = source["valueType"];
this.value = source["value"];
}
}
export class SyncColumnMapping {
source?: string;
target?: string;
drop?: boolean;
default?: SyncDefaultValue;
transforms?: SyncValueTransform[];
static createFrom(source: any = {}) {
return new SyncColumnMapping(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.source = source["source"];
this.target = source["target"];
this.drop = source["drop"];
this.default = this.convertValues(source["default"], SyncDefaultValue);
this.transforms = this.convertValues(source["transforms"], SyncValueTransform);
}
convertValues(a: any, classs: any, asMap: boolean = false): any {
if (!a) {
return a;
}
if (a.slice && a.map) {
return (a as any[]).map(elem => this.convertValues(elem, classs));
} else if ("object" === typeof a) {
if (asMap) {
for (const key of Object.keys(a)) {
a[key] = new classs(a[key]);
}
return a;
}
return new classs(a);
}
return a;
}
}
export class SyncObjectRef {
catalog?: string;
database?: string;
schema?: string;
name: string;
static createFrom(source: any = {}) {
return new SyncObjectRef(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.catalog = source["catalog"];
this.database = source["database"];
this.schema = source["schema"];
this.name = source["name"];
}
}
export class SyncObjectMapping {
id?: string;
source: SyncObjectRef;
target: SyncObjectRef;
keyColumns?: string[];
filter?: string;
columns?: SyncColumnMapping[];
static createFrom(source: any = {}) {
return new SyncObjectMapping(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.id = source["id"];
this.source = this.convertValues(source["source"], SyncObjectRef);
this.target = this.convertValues(source["target"], SyncObjectRef);
this.keyColumns = source["keyColumns"];
this.filter = source["filter"];
this.columns = this.convertValues(source["columns"], SyncColumnMapping);
}
convertValues(a: any, classs: any, asMap: boolean = false): any {
if (!a) {
return a;
}
if (a.slice && a.map) {
return (a as any[]).map(elem => this.convertValues(elem, classs));
} else if ("object" === typeof a) {
if (asMap) {
for (const key of Object.keys(a)) {
a[key] = new classs(a[key]);
}
return a;
}
return new classs(a);
}
return a;
}
}
export class TableOptions { export class TableOptions {
insert?: boolean; insert?: boolean;
update?: boolean; update?: boolean;
@@ -2743,11 +2899,14 @@ export namespace sync {
createIndexes?: boolean; createIndexes?: boolean;
mongoCollectionName?: string; mongoCollectionName?: string;
tableOptions?: Record<string, TableOptions>; tableOptions?: Record<string, TableOptions>;
mappings?: SyncObjectMapping[];
batchSize?: number;
rowErrorPolicy?: string;
static createFrom(source: any = {}) { static createFrom(source: any = {}) {
return new SyncConfig(source); return new SyncConfig(source);
} }
constructor(source: any = {}) { constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source); if ('string' === typeof source) source = JSON.parse(source);
this.sourceConfig = this.convertValues(source["sourceConfig"], connection.ConnectionConfig); this.sourceConfig = this.convertValues(source["sourceConfig"], connection.ConnectionConfig);
@@ -2765,8 +2924,11 @@ export namespace sync {
this.createIndexes = source["createIndexes"]; this.createIndexes = source["createIndexes"];
this.mongoCollectionName = source["mongoCollectionName"]; this.mongoCollectionName = source["mongoCollectionName"];
this.tableOptions = this.convertValues(source["tableOptions"], TableOptions, true); this.tableOptions = this.convertValues(source["tableOptions"], TableOptions, true);
this.mappings = this.convertValues(source["mappings"], SyncObjectMapping);
this.batchSize = source["batchSize"];
this.rowErrorPolicy = source["rowErrorPolicy"];
} }
convertValues(a: any, classs: any, asMap: boolean = false): any { convertValues(a: any, classs: any, asMap: boolean = false): any {
if (!a) { if (!a) {
return a; return a;
@@ -2793,11 +2955,13 @@ export namespace sync {
rowsInserted: number; rowsInserted: number;
rowsUpdated: number; rowsUpdated: number;
rowsDeleted: number; rowsDeleted: number;
rowsSkipped?: number;
cancelled?: boolean;
static createFrom(source: any = {}) { static createFrom(source: any = {}) {
return new SyncResult(source); return new SyncResult(source);
} }
constructor(source: any = {}) { constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source); if ('string' === typeof source) source = JSON.parse(source);
this.success = source["success"]; this.success = source["success"];
@@ -2807,7 +2971,326 @@ export namespace sync {
this.rowsInserted = source["rowsInserted"]; this.rowsInserted = source["rowsInserted"];
this.rowsUpdated = source["rowsUpdated"]; this.rowsUpdated = source["rowsUpdated"];
this.rowsDeleted = source["rowsDeleted"]; this.rowsDeleted = source["rowsDeleted"];
this.rowsSkipped = source["rowsSkipped"];
this.cancelled = source["cancelled"];
} }
} }
}
export namespace syncjob {
export class CDCSpec {
adapter?: string;
startPosition?: string;
initialSnapshot?: boolean;
slotName?: string;
publicationName?: string;
static createFrom(source: any = {}) {
return new CDCSpec(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.adapter = source["adapter"];
this.startPosition = source["startPosition"];
this.initialSnapshot = source["initialSnapshot"];
this.slotName = source["slotName"];
this.publicationName = source["publicationName"];
}
}
export class TransformSpec {
kind?: string;
argument?: number[];
static createFrom(source: any = {}) {
return new TransformSpec(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.kind = source["kind"];
this.argument = source["argument"];
}
}
export class ColumnMapping {
source?: string;
target: string;
transform?: TransformSpec;
defaultValue?: number[];
required?: boolean;
static createFrom(source: any = {}) {
return new ColumnMapping(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.source = source["source"];
this.target = source["target"];
this.transform = this.convertValues(source["transform"], TransformSpec);
this.defaultValue = source["defaultValue"];
this.required = source["required"];
}
convertValues(a: any, classs: any, asMap: boolean = false): any {
if (!a) {
return a;
}
if (a.slice && a.map) {
return (a as any[]).map(elem => this.convertValues(elem, classs));
} else if ("object" === typeof a) {
if (asMap) {
for (const key of Object.keys(a)) {
a[key] = new classs(a[key]);
}
return a;
}
return new classs(a);
}
return a;
}
}
export class EndpointRef {
connectionId: string;
connectionType?: string;
connectionName?: string;
database?: string;
schema?: string;
fingerprint?: string;
static createFrom(source: any = {}) {
return new EndpointRef(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.connectionId = source["connectionId"];
this.connectionType = source["connectionType"];
this.connectionName = source["connectionName"];
this.database = source["database"];
this.schema = source["schema"];
this.fingerprint = source["fingerprint"];
}
}
export class ExecutionApproval {
definitionHash: string;
targetFingerprint: string;
approvedAt: number;
approvedByRuntime: string;
static createFrom(source: any = {}) {
return new ExecutionApproval(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.definitionHash = source["definitionHash"];
this.targetFingerprint = source["targetFingerprint"];
this.approvedAt = source["approvedAt"];
this.approvedByRuntime = source["approvedByRuntime"];
}
}
export class ExecutionOptions {
content?: string;
syncMode?: string;
targetTableStrategy?: string;
autoAddColumns?: boolean;
createIndexes?: boolean;
propagateDeletes?: boolean;
batchSize?: number;
errorPolicy?: string;
maxRetries?: number;
retryBackoffMillis?: number;
captureErrorPayload?: boolean;
static createFrom(source: any = {}) {
return new ExecutionOptions(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.content = source["content"];
this.syncMode = source["syncMode"];
this.targetTableStrategy = source["targetTableStrategy"];
this.autoAddColumns = source["autoAddColumns"];
this.createIndexes = source["createIndexes"];
this.propagateDeletes = source["propagateDeletes"];
this.batchSize = source["batchSize"];
this.errorPolicy = source["errorPolicy"];
this.maxRetries = source["maxRetries"];
this.retryBackoffMillis = source["retryBackoffMillis"];
this.captureErrorPayload = source["captureErrorPayload"];
}
}
export class ScheduleSpec {
kind: string;
runAt?: number;
intervalSeconds?: number;
cronExpression?: string;
timezone?: string;
anchorAt?: number;
misfirePolicy?: string;
static createFrom(source: any = {}) {
return new ScheduleSpec(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.kind = source["kind"];
this.runAt = source["runAt"];
this.intervalSeconds = source["intervalSeconds"];
this.cronExpression = source["cronExpression"];
this.timezone = source["timezone"];
this.anchorAt = source["anchorAt"];
this.misfirePolicy = source["misfirePolicy"];
}
}
export class WatermarkSpec {
column: string;
initialValue?: number[];
tieBreakerColumns?: string[];
static createFrom(source: any = {}) {
return new WatermarkSpec(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.column = source["column"];
this.initialValue = source["initialValue"];
this.tieBreakerColumns = source["tieBreakerColumns"];
}
}
export class TableMapping {
sourceSchema?: string;
sourceTable: string;
targetSchema?: string;
targetTable: string;
targetTableStrategy?: string;
filter?: string;
keyColumns?: string[];
columns?: ColumnMapping[];
watermark?: WatermarkSpec;
enabled: boolean;
static createFrom(source: any = {}) {
return new TableMapping(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.sourceSchema = source["sourceSchema"];
this.sourceTable = source["sourceTable"];
this.targetSchema = source["targetSchema"];
this.targetTable = source["targetTable"];
this.targetTableStrategy = source["targetTableStrategy"];
this.filter = source["filter"];
this.keyColumns = source["keyColumns"];
this.columns = this.convertValues(source["columns"], ColumnMapping);
this.watermark = this.convertValues(source["watermark"], WatermarkSpec);
this.enabled = source["enabled"];
}
convertValues(a: any, classs: any, asMap: boolean = false): any {
if (!a) {
return a;
}
if (a.slice && a.map) {
return (a as any[]).map(elem => this.convertValues(elem, classs));
} else if ("object" === typeof a) {
if (asMap) {
for (const key of Object.keys(a)) {
a[key] = new classs(a[key]);
}
return a;
}
return new classs(a);
}
return a;
}
}
export class JobDefinition {
version: number;
id: string;
name: string;
description?: string;
lifecycle: string;
enabled: boolean;
kind: string;
incrementalMode: string;
source: EndpointRef;
target: EndpointRef;
sourceQuery?: string;
mappings: TableMapping[];
options: ExecutionOptions;
schedule: ScheduleSpec;
cdc?: CDCSpec;
approval?: ExecutionApproval;
concurrencyPolicy?: string;
resumePolicy?: string;
revision: number;
createdAt: number;
updatedAt: number;
nextRunAt?: number;
lastScheduledAt?: number;
archivedAt?: number;
static createFrom(source: any = {}) {
return new JobDefinition(source);
}
constructor(source: any = {}) {
if ('string' === typeof source) source = JSON.parse(source);
this.version = source["version"];
this.id = source["id"];
this.name = source["name"];
this.description = source["description"];
this.lifecycle = source["lifecycle"];
this.enabled = source["enabled"];
this.kind = source["kind"];
this.incrementalMode = source["incrementalMode"];
this.source = this.convertValues(source["source"], EndpointRef);
this.target = this.convertValues(source["target"], EndpointRef);
this.sourceQuery = source["sourceQuery"];
this.mappings = this.convertValues(source["mappings"], TableMapping);
this.options = this.convertValues(source["options"], ExecutionOptions);
this.schedule = this.convertValues(source["schedule"], ScheduleSpec);
this.cdc = this.convertValues(source["cdc"], CDCSpec);
this.approval = this.convertValues(source["approval"], ExecutionApproval);
this.concurrencyPolicy = source["concurrencyPolicy"];
this.resumePolicy = source["resumePolicy"];
this.revision = source["revision"];
this.createdAt = source["createdAt"];
this.updatedAt = source["updatedAt"];
this.nextRunAt = source["nextRunAt"];
this.lastScheduledAt = source["lastScheduledAt"];
this.archivedAt = source["archivedAt"];
}
convertValues(a: any, classs: any, asMap: boolean = false): any {
if (!a) {
return a;
}
if (a.slice && a.map) {
return (a as any[]).map(elem => this.convertValues(elem, classs));
} else if ("object" === typeof a) {
if (asMap) {
for (const key of Object.keys(a)) {
a[key] = new classs(a[key]);
}
return a;
}
return new classs(a);
}
return a;
}
}
} }