'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,
};