780 lines
24 KiB
JavaScript
780 lines
24 KiB
JavaScript
'use strict';
|
|
|
|
const crypto = require('crypto');
|
|
|
|
const cloudDb = require('../config/cloudDatabase');
|
|
const planningDb = require('./planningDatabase');
|
|
|
|
const DEFAULT_CHUNK_SIZE = 2000;
|
|
const DEFAULT_DATETIME_LOOKBACK_MINUTES = 180;
|
|
const DEFAULT_NUMERIC_LOOKBACK_ROWS = 10000;
|
|
const LOG_TABLE = 'planning_ding_workflow_sync_log';
|
|
const DEFAULT_SOURCE_PROFILE = process.env.DING_WORKFLOW_SYNC_SOURCE_PROFILE || 'public';
|
|
const DEFAULT_SOURCE_SCHEMA = process.env.CLOUD_DB_REPORT_SCHEMA || 'rpt';
|
|
|
|
const SOURCE_TABLES = Object.freeze({
|
|
workflow: {
|
|
key: 'workflow',
|
|
sourceSchema: process.env.DING_WORKFLOW_SOURCE_SCHEMA || DEFAULT_SOURCE_SCHEMA,
|
|
sourceTable: process.env.DING_WORKFLOW_SOURCE_TABLE || 'ding_Workflow',
|
|
targetTable: process.env.DING_WORKFLOW_TARGET_TABLE || 'ding_Workflow',
|
|
primaryKeys: ['processInstanceId'],
|
|
incrementalColumn: 'updateAt',
|
|
incrementalType: 'datetime',
|
|
incrementalSourceExpression: 'COALESCE([updateAt], [createAt])',
|
|
incrementalTargetExpression: 'COALESCE(`updateAt`, `createAt`)',
|
|
},
|
|
workflow_scheme: {
|
|
key: 'workflow_scheme',
|
|
sourceSchema: process.env.DING_WORKFLOW_SCHEME_SOURCE_SCHEMA || DEFAULT_SOURCE_SCHEMA,
|
|
sourceTable: process.env.DING_WORKFLOW_SCHEME_SOURCE_TABLE || 'ding_WorkflowScheme',
|
|
targetTable: process.env.DING_WORKFLOW_SCHEME_TARGET_TABLE || 'ding_WorkflowScheme',
|
|
primaryKeys: ['id'],
|
|
incrementalColumn: 'id',
|
|
incrementalType: 'number',
|
|
lookbackRows: Number.parseInt(process.env.DING_WORKFLOW_SCHEME_LOOKBACK_ROWS || '10000', 10) || 10000,
|
|
},
|
|
workflow_task: {
|
|
key: 'workflow_task',
|
|
sourceSchema: process.env.DING_WORKFLOW_TASK_SOURCE_SCHEMA || DEFAULT_SOURCE_SCHEMA,
|
|
sourceTable: process.env.DING_WORKFLOW_TASK_SOURCE_TABLE || 'ding_WorkflowTask',
|
|
targetTable: process.env.DING_WORKFLOW_TASK_TARGET_TABLE || 'ding_WorkflowTask',
|
|
primaryKeys: ['id'],
|
|
incrementalColumn: 'id',
|
|
incrementalType: 'number',
|
|
lookbackRows: Number.parseInt(process.env.DING_WORKFLOW_TASK_LOOKBACK_ROWS || '10000', 10) || 10000,
|
|
},
|
|
});
|
|
|
|
function quoteMysqlIdentifier(identifier) {
|
|
return `\`${String(identifier).replace(/`/g, '``')}\``;
|
|
}
|
|
|
|
function quoteMssqlIdentifier(identifier) {
|
|
return `[${String(identifier).replace(/]/g, ']]')}]`;
|
|
}
|
|
|
|
function normalizePositiveInt(value, fallback) {
|
|
const parsed = Number.parseInt(value, 10);
|
|
return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
|
|
}
|
|
|
|
function normalizeTableKey(value) {
|
|
const key = String(value || 'all').trim().toLowerCase();
|
|
if (key === 'all') {
|
|
return Object.keys(SOURCE_TABLES);
|
|
}
|
|
if (SOURCE_TABLES[key]) {
|
|
return [key];
|
|
}
|
|
throw new Error(`不支持的同步表: ${value}`);
|
|
}
|
|
|
|
function mapMssqlColumnToMysql(column) {
|
|
const type = String(column.DATA_TYPE || '').toLowerCase();
|
|
const maxLength = Number(column.CHARACTER_MAXIMUM_LENGTH);
|
|
const precision = Number(column.NUMERIC_PRECISION);
|
|
const scale = Number(column.NUMERIC_SCALE);
|
|
|
|
if (['int', 'smallint', 'tinyint'].includes(type)) {
|
|
return type === 'tinyint' ? 'TINYINT' : type === 'smallint' ? 'SMALLINT' : 'INT';
|
|
}
|
|
if (type === 'bigint') {
|
|
return 'BIGINT';
|
|
}
|
|
if (type === 'bit') {
|
|
return 'TINYINT(1)';
|
|
}
|
|
if (['decimal', 'numeric'].includes(type)) {
|
|
const safePrecision = Number.isFinite(precision) && precision > 0 ? precision : 38;
|
|
const safeScale = Number.isFinite(scale) && scale >= 0 ? scale : 0;
|
|
return `DECIMAL(${Math.min(safePrecision, 65)}, ${Math.min(safeScale, 30)})`;
|
|
}
|
|
if (['money', 'smallmoney'].includes(type)) {
|
|
return 'DECIMAL(19, 4)';
|
|
}
|
|
if (['float', 'real'].includes(type)) {
|
|
return 'DOUBLE';
|
|
}
|
|
if (type === 'date') {
|
|
return 'DATE';
|
|
}
|
|
if (['datetime', 'datetime2', 'smalldatetime', 'time'].includes(type)) {
|
|
return 'DATETIME(3)';
|
|
}
|
|
if (type === 'uniqueidentifier') {
|
|
return 'VARCHAR(36)';
|
|
}
|
|
if (['nvarchar', 'varchar', 'nchar', 'char'].includes(type)) {
|
|
if (!Number.isFinite(maxLength) || maxLength < 0) {
|
|
return 'LONGTEXT';
|
|
}
|
|
const normalizedLength = ['nvarchar', 'nchar'].includes(type) ? Math.ceil(maxLength / 2) : maxLength;
|
|
if (normalizedLength <= 0) {
|
|
return 'VARCHAR(255)';
|
|
}
|
|
if (normalizedLength <= 16383) {
|
|
return `VARCHAR(${Math.max(normalizedLength, 1)})`;
|
|
}
|
|
return 'TEXT';
|
|
}
|
|
if (['text', 'ntext', 'xml'].includes(type)) {
|
|
return 'LONGTEXT';
|
|
}
|
|
if (['binary', 'varbinary', 'image'].includes(type)) {
|
|
return 'LONGBLOB';
|
|
}
|
|
|
|
return 'TEXT';
|
|
}
|
|
|
|
function toMysqlValue(value) {
|
|
if (value === undefined) {
|
|
return null;
|
|
}
|
|
if (value instanceof Date) {
|
|
const pad = (number, size = 2) => String(number).padStart(size, '0');
|
|
return [
|
|
value.getFullYear(),
|
|
'-',
|
|
pad(value.getMonth() + 1),
|
|
'-',
|
|
pad(value.getDate()),
|
|
' ',
|
|
pad(value.getHours()),
|
|
':',
|
|
pad(value.getMinutes()),
|
|
':',
|
|
pad(value.getSeconds()),
|
|
'.',
|
|
pad(value.getMilliseconds(), 3),
|
|
].join('');
|
|
}
|
|
if (typeof value === 'boolean') {
|
|
return value ? 1 : 0;
|
|
}
|
|
if (Buffer.isBuffer(value)) {
|
|
return value;
|
|
}
|
|
return value;
|
|
}
|
|
|
|
function formatDateForIncremental(date) {
|
|
const pad = (number, size = 2) => String(number).padStart(size, '0');
|
|
return [
|
|
date.getFullYear(),
|
|
'-',
|
|
pad(date.getMonth() + 1),
|
|
'-',
|
|
pad(date.getDate()),
|
|
' ',
|
|
pad(date.getHours()),
|
|
':',
|
|
pad(date.getMinutes()),
|
|
':',
|
|
pad(date.getSeconds()),
|
|
].join('');
|
|
}
|
|
|
|
function parseFlexibleDate(value) {
|
|
if (!value) {
|
|
return null;
|
|
}
|
|
if (value instanceof Date) {
|
|
return Number.isNaN(value.getTime()) ? null : value;
|
|
}
|
|
|
|
const raw = String(value).trim();
|
|
if (!raw) {
|
|
return null;
|
|
}
|
|
|
|
const candidates = [
|
|
raw,
|
|
raw.includes('T') ? raw : raw.replace(' ', 'T'),
|
|
raw.includes('Z') ? raw : `${raw.replace(' ', 'T')}Z`,
|
|
];
|
|
|
|
for (const candidate of candidates) {
|
|
const parsed = new Date(candidate);
|
|
if (!Number.isNaN(parsed.getTime())) {
|
|
return parsed;
|
|
}
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
function formatValueForLog(value) {
|
|
if (value === null || value === undefined) {
|
|
return null;
|
|
}
|
|
if (value instanceof Date) {
|
|
return value.toISOString();
|
|
}
|
|
return String(value);
|
|
}
|
|
|
|
function getMssqlIncrementalExpression(tableConfig) {
|
|
if (tableConfig.incrementalSourceExpression) {
|
|
return tableConfig.incrementalSourceExpression;
|
|
}
|
|
if (tableConfig.incrementalColumn) {
|
|
return quoteMssqlIdentifier(tableConfig.incrementalColumn);
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function getMysqlIncrementalExpression(tableConfig) {
|
|
if (tableConfig.incrementalTargetExpression) {
|
|
return tableConfig.incrementalTargetExpression;
|
|
}
|
|
if (tableConfig.incrementalColumn) {
|
|
return quoteMysqlIdentifier(tableConfig.incrementalColumn);
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function shouldUseNumericKeyset(tableConfig, primaryKeys) {
|
|
return (
|
|
tableConfig.incrementalType === 'number'
|
|
&& !tableConfig.incrementalSourceExpression
|
|
&& primaryKeys.length === 1
|
|
&& primaryKeys[0] === tableConfig.incrementalColumn
|
|
);
|
|
}
|
|
|
|
async function getSourceColumns(tableConfig, sourceDb = cloudDb) {
|
|
const rows = await sourceDb.query(
|
|
`
|
|
SELECT
|
|
COLUMN_NAME,
|
|
DATA_TYPE,
|
|
CHARACTER_MAXIMUM_LENGTH,
|
|
NUMERIC_PRECISION,
|
|
NUMERIC_SCALE,
|
|
IS_NULLABLE,
|
|
ORDINAL_POSITION
|
|
FROM INFORMATION_SCHEMA.COLUMNS
|
|
WHERE TABLE_SCHEMA = @schemaName
|
|
AND TABLE_NAME = @tableName
|
|
ORDER BY ORDINAL_POSITION
|
|
`,
|
|
{
|
|
schemaName: tableConfig.sourceSchema,
|
|
tableName: tableConfig.sourceTable,
|
|
}
|
|
);
|
|
|
|
if (!rows.length) {
|
|
throw new Error(`业务库找不到源表 ${tableConfig.sourceSchema}.${tableConfig.sourceTable}`);
|
|
}
|
|
|
|
return rows;
|
|
}
|
|
|
|
async function getSourcePrimaryKeys(tableConfig, sourceDb = cloudDb) {
|
|
const rows = await sourceDb.query(
|
|
`
|
|
SELECT KU.COLUMN_NAME
|
|
FROM INFORMATION_SCHEMA.TABLE_CONSTRAINTS TC
|
|
JOIN INFORMATION_SCHEMA.KEY_COLUMN_USAGE KU
|
|
ON TC.CONSTRAINT_NAME = KU.CONSTRAINT_NAME
|
|
AND TC.TABLE_SCHEMA = KU.TABLE_SCHEMA
|
|
AND TC.TABLE_NAME = KU.TABLE_NAME
|
|
WHERE TC.CONSTRAINT_TYPE = 'PRIMARY KEY'
|
|
AND TC.TABLE_SCHEMA = @schemaName
|
|
AND TC.TABLE_NAME = @tableName
|
|
ORDER BY KU.ORDINAL_POSITION
|
|
`,
|
|
{
|
|
schemaName: tableConfig.sourceSchema,
|
|
tableName: tableConfig.sourceTable,
|
|
}
|
|
);
|
|
|
|
return rows.map((row) => row.COLUMN_NAME).filter(Boolean);
|
|
}
|
|
|
|
async function getTargetColumns(targetTable) {
|
|
const rows = await planningDb.query(
|
|
`
|
|
SELECT COLUMN_NAME
|
|
FROM INFORMATION_SCHEMA.COLUMNS
|
|
WHERE TABLE_SCHEMA = DATABASE()
|
|
AND TABLE_NAME = ?
|
|
`,
|
|
[targetTable]
|
|
);
|
|
return new Set(rows.map((row) => row.COLUMN_NAME));
|
|
}
|
|
|
|
async function ensureLogTable() {
|
|
await planningDb.execute(`
|
|
CREATE TABLE IF NOT EXISTS ${quoteMysqlIdentifier(LOG_TABLE)} (
|
|
id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
|
|
target_table VARCHAR(128) NOT NULL,
|
|
source_schema VARCHAR(128) NOT NULL,
|
|
source_table VARCHAR(128) NOT NULL,
|
|
sync_mode VARCHAR(32) NOT NULL,
|
|
source_incremental_column VARCHAR(255) NULL,
|
|
source_after VARCHAR(128) NULL,
|
|
fetched_rows INT NOT NULL DEFAULT 0,
|
|
upserted_rows INT NOT NULL DEFAULT 0,
|
|
status VARCHAR(32) NOT NULL,
|
|
error_message TEXT NULL,
|
|
started_at DATETIME(3) NOT NULL,
|
|
finished_at DATETIME(3) NOT NULL,
|
|
elapsed_ms INT NOT NULL DEFAULT 0,
|
|
PRIMARY KEY (id),
|
|
KEY idx_target_started (target_table, started_at)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
|
|
`);
|
|
}
|
|
|
|
async function appendLog(payload) {
|
|
await ensureLogTable();
|
|
await planningDb.execute(
|
|
`
|
|
INSERT INTO ${quoteMysqlIdentifier(LOG_TABLE)}
|
|
(target_table, source_schema, source_table, sync_mode, source_incremental_column, source_after,
|
|
fetched_rows, upserted_rows, status, error_message, started_at, finished_at, elapsed_ms)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
`,
|
|
[
|
|
payload.targetTable,
|
|
payload.sourceSchema,
|
|
payload.sourceTable,
|
|
payload.syncMode,
|
|
payload.incrementalColumn || null,
|
|
payload.sourceAfter || null,
|
|
payload.fetchedRows || 0,
|
|
payload.upsertedRows || 0,
|
|
payload.status,
|
|
payload.errorMessage || null,
|
|
payload.startedAt,
|
|
payload.finishedAt,
|
|
payload.elapsedMs || 0,
|
|
]
|
|
);
|
|
}
|
|
|
|
async function ensureTargetTable(tableConfig, sourceColumns, primaryKeys) {
|
|
const pkColumns = primaryKeys.length ? primaryKeys : tableConfig.primaryKeys;
|
|
const columnDefinitions = sourceColumns.map((column) => {
|
|
const name = quoteMysqlIdentifier(column.COLUMN_NAME);
|
|
const type = mapMssqlColumnToMysql(column);
|
|
const nullable = pkColumns.includes(column.COLUMN_NAME) || String(column.IS_NULLABLE).toUpperCase() === 'NO'
|
|
? 'NOT NULL'
|
|
: 'NULL';
|
|
return `${name} ${type} ${nullable}`;
|
|
});
|
|
|
|
const metadataColumns = [
|
|
{ name: '__source_table', type: 'VARCHAR(255) NOT NULL' },
|
|
{ name: '__sync_batch_id', type: 'VARCHAR(64) NOT NULL' },
|
|
{ name: '__synced_at', type: 'DATETIME(3) NOT NULL' },
|
|
];
|
|
const pkClause = pkColumns.length
|
|
? `, PRIMARY KEY (${pkColumns.map(quoteMysqlIdentifier).join(', ')})`
|
|
: '';
|
|
const incrementalIndex = tableConfig.incrementalColumn
|
|
? `, KEY ${quoteMysqlIdentifier(`idx_${tableConfig.targetTable}_${tableConfig.incrementalColumn}`)} (${quoteMysqlIdentifier(tableConfig.incrementalColumn)})`
|
|
: '';
|
|
|
|
await planningDb.execute(`
|
|
CREATE TABLE IF NOT EXISTS ${quoteMysqlIdentifier(tableConfig.targetTable)} (
|
|
${[
|
|
...columnDefinitions,
|
|
...metadataColumns.map((column) => `${quoteMysqlIdentifier(column.name)} ${column.type}`),
|
|
].join(',\n ')}
|
|
${pkClause}
|
|
${incrementalIndex}
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
|
|
`);
|
|
|
|
const existingColumns = await getTargetColumns(tableConfig.targetTable);
|
|
for (const column of sourceColumns) {
|
|
if (existingColumns.has(column.COLUMN_NAME)) {
|
|
continue;
|
|
}
|
|
const nullable = pkColumns.includes(column.COLUMN_NAME) || String(column.IS_NULLABLE).toUpperCase() === 'NO'
|
|
? 'NOT NULL'
|
|
: 'NULL';
|
|
await planningDb.execute(
|
|
`ALTER TABLE ${quoteMysqlIdentifier(tableConfig.targetTable)} ADD COLUMN ${quoteMysqlIdentifier(column.COLUMN_NAME)} ${mapMssqlColumnToMysql(column)} ${nullable}`
|
|
);
|
|
}
|
|
|
|
for (const column of metadataColumns) {
|
|
if (existingColumns.has(column.name)) {
|
|
continue;
|
|
}
|
|
await planningDb.execute(
|
|
`ALTER TABLE ${quoteMysqlIdentifier(tableConfig.targetTable)} ADD COLUMN ${quoteMysqlIdentifier(column.name)} ${column.type}`
|
|
);
|
|
}
|
|
}
|
|
|
|
async function getLatestTargetIncremental(tableConfig) {
|
|
if (!tableConfig.incrementalColumn) {
|
|
return null;
|
|
}
|
|
const expression = getMysqlIncrementalExpression(tableConfig);
|
|
const rows = await planningDb.query(
|
|
`SELECT MAX(${expression}) AS latestValue FROM ${quoteMysqlIdentifier(tableConfig.targetTable)}`
|
|
);
|
|
return rows[0]?.latestValue || null;
|
|
}
|
|
|
|
function buildSourceWhereClause(tableConfig, sourceAfter, extraClauses = []) {
|
|
const clauses = [...extraClauses];
|
|
const incrementalExpression = getMssqlIncrementalExpression(tableConfig);
|
|
if (sourceAfter !== null && sourceAfter !== undefined && incrementalExpression) {
|
|
clauses.push(`${incrementalExpression} >= @sourceAfter`);
|
|
}
|
|
if (!clauses.length) {
|
|
return '';
|
|
}
|
|
return `WHERE ${clauses.join(' AND ')}`;
|
|
}
|
|
|
|
function buildOrderBy(tableConfig, primaryKeys) {
|
|
const parts = [];
|
|
const incrementalExpression = getMssqlIncrementalExpression(tableConfig);
|
|
if (incrementalExpression) {
|
|
parts.push(incrementalExpression);
|
|
}
|
|
primaryKeys.forEach((columnName) => {
|
|
parts.push(quoteMssqlIdentifier(columnName));
|
|
});
|
|
|
|
const uniqueParts = Array.from(new Set(parts.filter(Boolean)));
|
|
return uniqueParts.join(', ') || '(SELECT NULL)';
|
|
}
|
|
|
|
async function getSourceCount(tableConfig, sourceAfter, sourceDb = cloudDb) {
|
|
const whereClause = buildSourceWhereClause(tableConfig, sourceAfter);
|
|
const rows = await sourceDb.query(
|
|
`
|
|
SELECT COUNT(1) AS total
|
|
FROM ${quoteMssqlIdentifier(tableConfig.sourceSchema)}.${quoteMssqlIdentifier(tableConfig.sourceTable)}
|
|
${whereClause}
|
|
`,
|
|
sourceAfter !== null && sourceAfter !== undefined ? { sourceAfter } : {}
|
|
);
|
|
return Number(rows[0]?.total || 0);
|
|
}
|
|
|
|
async function fetchSourceRowsOffset(tableConfig, columns, primaryKeys, sourceAfter, offset, limit, sourceDb = cloudDb) {
|
|
const whereClause = buildSourceWhereClause(tableConfig, sourceAfter);
|
|
const orderBy = buildOrderBy(tableConfig, primaryKeys);
|
|
const selectColumns = columns.map((column) => quoteMssqlIdentifier(column.COLUMN_NAME)).join(', ');
|
|
|
|
return sourceDb.query(
|
|
`
|
|
SELECT ${selectColumns}
|
|
FROM ${quoteMssqlIdentifier(tableConfig.sourceSchema)}.${quoteMssqlIdentifier(tableConfig.sourceTable)}
|
|
${whereClause}
|
|
ORDER BY ${orderBy}
|
|
OFFSET ${offset} ROWS FETCH NEXT ${limit} ROWS ONLY
|
|
`,
|
|
sourceAfter !== null && sourceAfter !== undefined ? { sourceAfter } : {}
|
|
);
|
|
}
|
|
|
|
async function fetchSourceRowsKeyset(tableConfig, columns, sourceAfter, lastCursor, limit, sourceDb = cloudDb) {
|
|
const selectColumns = columns.map((column) => quoteMssqlIdentifier(column.COLUMN_NAME)).join(', ');
|
|
const clauses = [];
|
|
const params = {};
|
|
|
|
if (sourceAfter !== null && sourceAfter !== undefined) {
|
|
clauses.push(`${quoteMssqlIdentifier(tableConfig.incrementalColumn)} >= @sourceAfter`);
|
|
params.sourceAfter = sourceAfter;
|
|
}
|
|
if (lastCursor !== null && lastCursor !== undefined) {
|
|
clauses.push(`${quoteMssqlIdentifier(tableConfig.incrementalColumn)} > @lastCursor`);
|
|
params.lastCursor = lastCursor;
|
|
}
|
|
|
|
const whereClause = clauses.length ? `WHERE ${clauses.join(' AND ')}` : '';
|
|
|
|
return sourceDb.query(
|
|
`
|
|
SELECT TOP (${limit}) ${selectColumns}
|
|
FROM ${quoteMssqlIdentifier(tableConfig.sourceSchema)}.${quoteMssqlIdentifier(tableConfig.sourceTable)}
|
|
${whereClause}
|
|
ORDER BY ${quoteMssqlIdentifier(tableConfig.incrementalColumn)} ASC
|
|
`,
|
|
params
|
|
);
|
|
}
|
|
|
|
async function upsertRows(tableConfig, primaryKeys, sourceColumns, rows, batchId) {
|
|
if (!rows.length) {
|
|
return 0;
|
|
}
|
|
|
|
const sourceColumnNames = sourceColumns.map((column) => column.COLUMN_NAME);
|
|
const allColumns = [...sourceColumnNames, '__source_table', '__sync_batch_id', '__synced_at'];
|
|
const placeholders = rows.map(() => `(${allColumns.map(() => '?').join(', ')})`).join(', ');
|
|
const now = new Date();
|
|
const params = [];
|
|
|
|
rows.forEach((row) => {
|
|
sourceColumnNames.forEach((columnName) => {
|
|
params.push(toMysqlValue(row[columnName]));
|
|
});
|
|
params.push(`${tableConfig.sourceSchema}.${tableConfig.sourceTable}`);
|
|
params.push(batchId);
|
|
params.push(toMysqlValue(now));
|
|
});
|
|
|
|
const updateColumns = allColumns.filter((columnName) => !primaryKeys.includes(columnName));
|
|
const updateClause = updateColumns
|
|
.map((columnName) => `${quoteMysqlIdentifier(columnName)} = VALUES(${quoteMysqlIdentifier(columnName)})`)
|
|
.join(', ');
|
|
|
|
const result = await planningDb.execute(
|
|
`
|
|
INSERT INTO ${quoteMysqlIdentifier(tableConfig.targetTable)}
|
|
(${allColumns.map(quoteMysqlIdentifier).join(', ')})
|
|
VALUES ${placeholders}
|
|
ON DUPLICATE KEY UPDATE ${updateClause}
|
|
`,
|
|
params
|
|
);
|
|
|
|
return Number(result?.affectedRows || 0);
|
|
}
|
|
|
|
function applyIncrementalLookback(value, tableConfig, options) {
|
|
if (value === null || value === undefined || value === '') {
|
|
return null;
|
|
}
|
|
|
|
if (tableConfig.incrementalType === 'datetime') {
|
|
const parsed = parseFlexibleDate(value);
|
|
if (!parsed) {
|
|
return value;
|
|
}
|
|
const lookbackMinutes = normalizePositiveInt(
|
|
options.lookbackMinutes,
|
|
DEFAULT_DATETIME_LOOKBACK_MINUTES
|
|
);
|
|
const adjusted = new Date(parsed.getTime() - lookbackMinutes * 60 * 1000);
|
|
return formatDateForIncremental(adjusted);
|
|
}
|
|
|
|
if (tableConfig.incrementalType === 'number') {
|
|
const parsed = Number(value);
|
|
if (!Number.isFinite(parsed)) {
|
|
return value;
|
|
}
|
|
const lookbackRows = normalizePositiveInt(
|
|
options.lookbackRows,
|
|
Number.isFinite(Number(tableConfig.lookbackRows)) ? Number(tableConfig.lookbackRows) : DEFAULT_NUMERIC_LOOKBACK_ROWS
|
|
);
|
|
return Math.max(Math.floor(parsed) - lookbackRows, 0);
|
|
}
|
|
|
|
return value;
|
|
}
|
|
|
|
async function syncOneTable(tableConfig, options = {}) {
|
|
const startedAtDate = new Date();
|
|
const startedAt = toMysqlValue(startedAtDate);
|
|
const chunkSize = normalizePositiveInt(options.chunkSize, DEFAULT_CHUNK_SIZE);
|
|
const syncMode = options.full ? 'full' : 'incremental';
|
|
let fetchedRows = 0;
|
|
let upsertedRows = 0;
|
|
let sourceAfter = null;
|
|
|
|
try {
|
|
const sourceDb = options.sourceDb || cloudDb;
|
|
const sourceColumns = await getSourceColumns(tableConfig, sourceDb);
|
|
const sourcePrimaryKeys = await getSourcePrimaryKeys(tableConfig, sourceDb);
|
|
const primaryKeys = sourcePrimaryKeys.length ? sourcePrimaryKeys : tableConfig.primaryKeys;
|
|
|
|
await ensureTargetTable({ ...tableConfig, primaryKeys }, sourceColumns, primaryKeys);
|
|
|
|
if (!options.full) {
|
|
sourceAfter = applyIncrementalLookback(
|
|
await getLatestTargetIncremental(tableConfig),
|
|
tableConfig,
|
|
options
|
|
);
|
|
}
|
|
|
|
const totalRows = await getSourceCount(tableConfig, sourceAfter, sourceDb);
|
|
const batchId = crypto.randomUUID ? crypto.randomUUID() : crypto.randomBytes(16).toString('hex');
|
|
|
|
if (shouldUseNumericKeyset(tableConfig, primaryKeys)) {
|
|
let lastCursor = null;
|
|
|
|
while (true) {
|
|
const rows = await fetchSourceRowsKeyset(
|
|
tableConfig,
|
|
sourceColumns,
|
|
sourceAfter,
|
|
lastCursor,
|
|
chunkSize,
|
|
sourceDb
|
|
);
|
|
if (!rows.length) {
|
|
break;
|
|
}
|
|
|
|
fetchedRows += rows.length;
|
|
upsertedRows += await upsertRows(tableConfig, primaryKeys, sourceColumns, rows, batchId);
|
|
lastCursor = rows[rows.length - 1][tableConfig.incrementalColumn];
|
|
|
|
if (typeof options.onProgress === 'function') {
|
|
options.onProgress({
|
|
table: tableConfig.targetTable,
|
|
fetchedRows,
|
|
totalRows,
|
|
chunkSize,
|
|
lastCursor,
|
|
});
|
|
}
|
|
|
|
if (rows.length < chunkSize) {
|
|
break;
|
|
}
|
|
}
|
|
} else {
|
|
for (let offset = 0; offset < totalRows; offset += chunkSize) {
|
|
const rows = await fetchSourceRowsOffset(
|
|
tableConfig,
|
|
sourceColumns,
|
|
primaryKeys,
|
|
sourceAfter,
|
|
offset,
|
|
chunkSize,
|
|
sourceDb
|
|
);
|
|
fetchedRows += rows.length;
|
|
upsertedRows += await upsertRows(tableConfig, primaryKeys, sourceColumns, rows, batchId);
|
|
|
|
if (typeof options.onProgress === 'function') {
|
|
options.onProgress({
|
|
table: tableConfig.targetTable,
|
|
fetchedRows,
|
|
totalRows,
|
|
offset,
|
|
chunkSize,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
const finishedAtDate = new Date();
|
|
const result = {
|
|
key: tableConfig.key,
|
|
source: `${tableConfig.sourceSchema}.${tableConfig.sourceTable}`,
|
|
target: tableConfig.targetTable,
|
|
syncMode,
|
|
sourceAfter: formatValueForLog(sourceAfter),
|
|
fetchedRows,
|
|
upsertedRows,
|
|
elapsedMs: finishedAtDate.getTime() - startedAtDate.getTime(),
|
|
status: 'success',
|
|
};
|
|
|
|
await appendLog({
|
|
...result,
|
|
sourceSchema: tableConfig.sourceSchema,
|
|
sourceTable: tableConfig.sourceTable,
|
|
targetTable: tableConfig.targetTable,
|
|
incrementalColumn: tableConfig.incrementalColumn,
|
|
startedAt,
|
|
finishedAt: toMysqlValue(finishedAtDate),
|
|
});
|
|
|
|
return result;
|
|
} catch (error) {
|
|
const finishedAtDate = new Date();
|
|
await appendLog({
|
|
targetTable: tableConfig.targetTable,
|
|
sourceSchema: tableConfig.sourceSchema,
|
|
sourceTable: tableConfig.sourceTable,
|
|
syncMode,
|
|
incrementalColumn: tableConfig.incrementalColumn,
|
|
sourceAfter: formatValueForLog(sourceAfter),
|
|
fetchedRows,
|
|
upsertedRows,
|
|
status: 'failed',
|
|
errorMessage: error.message,
|
|
startedAt,
|
|
finishedAt: toMysqlValue(finishedAtDate),
|
|
elapsedMs: finishedAtDate.getTime() - startedAtDate.getTime(),
|
|
}).catch((logError) => {
|
|
console.warn('[ding-workflow-sync] 写入同步失败日志失败:', logError.message);
|
|
});
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async function syncPlanningDingWorkflowTables(options = {}) {
|
|
const tableKeys = normalizeTableKey(options.table);
|
|
const results = [];
|
|
const expectedTableNames = tableKeys.map((key) => SOURCE_TABLES[key].sourceTable);
|
|
const verifySchema = SOURCE_TABLES[tableKeys[0]]?.sourceSchema || 'dbo';
|
|
let sourceDbClient = options.sourceDb || null;
|
|
|
|
try {
|
|
if (!sourceDbClient) {
|
|
sourceDbClient = await cloudDb.createPinnedQueryClient({
|
|
preferredProfile: options.sourceProfile || DEFAULT_SOURCE_PROFILE,
|
|
verifySql: `
|
|
SELECT TABLE_NAME
|
|
FROM INFORMATION_SCHEMA.TABLES
|
|
WHERE TABLE_SCHEMA = @schemaName
|
|
AND TABLE_NAME IN (${expectedTableNames.map((_, index) => `@tableName${index}`).join(', ')})
|
|
`,
|
|
verifyParams: expectedTableNames.reduce(
|
|
(accumulator, tableName, index) => ({ ...accumulator, [`tableName${index}`]: tableName }),
|
|
{ schemaName: verifySchema }
|
|
),
|
|
verifyOptions: {
|
|
requestTimeout: normalizePositiveInt(options.requestTimeout, 30000),
|
|
maxAttempts: 1,
|
|
},
|
|
verifyResult: (rows) => {
|
|
const available = new Set((rows || []).map((row) => row.TABLE_NAME));
|
|
const missingTables = expectedTableNames.filter((tableName) => !available.has(tableName));
|
|
if (missingTables.length) {
|
|
return `业务数据库候选缺少源表: ${missingTables.join(', ')}`;
|
|
}
|
|
return true;
|
|
},
|
|
});
|
|
}
|
|
|
|
for (const key of tableKeys) {
|
|
results.push(await syncOneTable(SOURCE_TABLES[key], { ...options, sourceDb: sourceDbClient }));
|
|
}
|
|
} finally {
|
|
if (!options.sourceDb && sourceDbClient && typeof sourceDbClient.close === 'function') {
|
|
await sourceDbClient.close().catch((error) => {
|
|
console.warn('[ding-workflow-sync] 关闭固定业务库连接失败:', error.message);
|
|
});
|
|
}
|
|
}
|
|
|
|
return {
|
|
success: true,
|
|
mode: options.full ? 'full' : 'incremental',
|
|
sourceRuntime: sourceDbClient?.runtimeInfo || null,
|
|
tables: results,
|
|
finishedAt: new Date().toISOString(),
|
|
};
|
|
}
|
|
|
|
module.exports = {
|
|
SOURCE_TABLES,
|
|
syncPlanningDingWorkflowTables,
|
|
syncOneTable,
|
|
};
|