a-share-analysis/tools/database/mysql2-schema-exporter.mjs

import { createRequire } from 'node:module';

const require = createRequire(import.meta.url);
const mysql2Version = require('mysql2/package.json').version;
const DEVELOPMENT_DATABASE = 'a_share_analysis_dev';

const CATALOG_QUERIES = Object.freeze({
  tables: `SELECT TABLE_NAME, TABLE_TYPE
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = ?
ORDER BY TABLE_NAME`,
  viewDependencies: `SELECT VIEW_NAME, TABLE_SCHEMA AS REFERENCED_SCHEMA_NAME,
       TABLE_NAME AS REFERENCED_TABLE_NAME
FROM information_schema.VIEW_TABLE_USAGE
WHERE VIEW_SCHEMA = ?
ORDER BY VIEW_NAME, REFERENCED_SCHEMA_NAME, REFERENCED_TABLE_NAME`,
  routines: `SELECT ROUTINE_NAME, ROUTINE_TYPE
FROM information_schema.ROUTINES
WHERE ROUTINE_SCHEMA = ?
ORDER BY ROUTINE_TYPE, ROUTINE_NAME`,
  triggers: `SELECT TRIGGER_NAME, EVENT_OBJECT_TABLE, ACTION_TIMING,
       EVENT_MANIPULATION, ACTION_ORDER
FROM information_schema.TRIGGERS
WHERE TRIGGER_SCHEMA = ?
ORDER BY EVENT_OBJECT_TABLE, ACTION_TIMING, EVENT_MANIPULATION,
         ACTION_ORDER, TRIGGER_NAME`,
  events: `SELECT EVENT_NAME
FROM information_schema.EVENTS
WHERE EVENT_SCHEMA = ?
ORDER BY EVENT_NAME`,
});

export const MYSQL2_EXPORTER_METADATA = Object.freeze({
  name: 'mysql2-show-create',
  version: '1.0.0',
  driverName: 'mysql2',
  driverVersion: mysql2Version,
});

export const DEFAULT_EXPORT_LIMITS = Object.freeze({
  connectTimeoutMs: 5_000,
  queryTimeoutMs: 5_000,
  totalTimeoutMs: 30_000,
  maxObjects: 5_000,
  maxDefinitionBytes: 1_048_576,
});

export class SchemaExportError extends Error {
  constructor(code, message) {
    super(message);
    this.name = 'SchemaExportError';
    this.code = code;
  }
}

/**
 * 配置在 Schema 治理入口已完成 development、主机白名单和固定库名校验；这里仅映射
 * 连接所需字段，固定禁用多语句以防止未来元数据查询被拼接为任意 SQL。
 */
export function buildConnectionOptions(config) {
  return {
    host: config.MYSQL_HOST,
    port: Number(config.MYSQL_PORT),
    user: config.MYSQL_USERNAME,
    password: config.MYSQL_PASSWORD,
    database: config.MYSQL_DATABASE,
    charset: 'utf8mb4',
    timezone: '+08:00',
    connectTimeout: DEFAULT_EXPORT_LIMITS.connectTimeoutMs,
    multipleStatements: false,
    enableKeepAlive: true,
  };
}

/**
 * 元数据对象名仍需防御性转义，避免控制字符破坏 `SHOW CREATE` 查询边界；名称不合法时
 * 不回显输入，确保异常路径不会把潜在敏感值写入日志。
 */
export function quoteIdentifier(identifier) {
  if (typeof identifier !== 'string' || identifier.length === 0 || /[\u0000-\u001F\u007F]/.test(identifier)) {
    throw new SchemaExportError('INVALID_IDENTIFIER', 'Schema 标识符不能为空且不得包含控制字符');
  }

  return `\`${identifier.replaceAll('`', '``')}\``;
}

/**
 * 导出器只接受项目数据库兼容基线，拒绝 MariaDB 和 MySQL 5.7 之外的服务器，避免相同
 * `SHOW CREATE` 文本在不兼容语义下被误认为可恢复的 Schema 快照。
 */
export function assertMysql57ServerMetadata(metadata) {
  const serverVersion = metadata?.serverVersion ?? '';
  if (!/^5\.7\.\d+(?:[-+].*)?$/.test(serverVersion) || /mariadb/i.test(serverVersion)) {
    throw new SchemaExportError('UNSUPPORTED_SERVER_VERSION', 'MySQL 服务端必须为 5.7.x');
  }
  if (metadata.databaseCharset !== 'utf8mb4') {
    throw new SchemaExportError('INVALID_DATABASE_CHARSET', '数据库字符集必须为 utf8mb4');
  }
  if (metadata.databaseCollation !== 'utf8mb4_general_ci') {
    throw new SchemaExportError('INVALID_DATABASE_COLLATION', '数据库排序规则必须为 utf8mb4_general_ci');
  }
}

function requireCatalogText(value) {
  if (typeof value !== 'string' || value.length === 0) {
    throw new SchemaExportError('CATALOG_SHAPE_INVALID', 'Schema 对象目录字段无效');
  }
  return value;
}

async function queryCatalogRows(connection, sql, database, limits) {
  const result = await connection.query({
    sql,
    values: [database],
    timeout: limits.queryTimeoutMs,
  });
  if (!Array.isArray(result) || !Array.isArray(result[0])) {
    throw new SchemaExportError('CATALOG_SHAPE_INVALID', 'Schema 对象目录查询结果无效');
  }
  return result[0];
}

function assertObjectLimit(count, limits) {
  if (!Number.isInteger(limits.maxObjects) || limits.maxObjects < 0 || count > limits.maxObjects) {
    throw new SchemaExportError('OBJECT_LIMIT_EXCEEDED', 'Schema 对象数量超过安全上限');
  }
}

/**
 * 目录查询只从 information_schema 读取固定列，库名始终作为参数传入。每类对象
 * 在进入后续 SHOW CREATE 前先转换成稳定的小对象，并在累计数量越界时立即停止。
 */
export async function readSchemaCatalog(connection, database, limits = DEFAULT_EXPORT_LIMITS) {
  const tableRows = await queryCatalogRows(connection, CATALOG_QUERIES.tables, database, limits);
  const tables = [];
  const views = [];
  for (const row of tableRows) {
    const name = requireCatalogText(row.TABLE_NAME);
    if (row.TABLE_TYPE === 'BASE TABLE') tables.push(name);
    else if (row.TABLE_TYPE === 'VIEW') views.push(name);
    else throw new SchemaExportError('UNKNOWN_TABLE_TYPE', 'Schema 目录包含未知表类型');
  }
  tables.sort();
  views.sort();
  let objectCount = tables.length + views.length;
  assertObjectLimit(objectCount, limits);

  const dependencyRows = await queryCatalogRows(
    connection,
    CATALOG_QUERIES.viewDependencies,
    database,
    limits,
  );
  const viewDependencies = dependencyRows.map((row) => ({
    view: requireCatalogText(row.VIEW_NAME),
    referencedSchema: requireCatalogText(row.REFERENCED_SCHEMA_NAME),
    referencedObject: requireCatalogText(row.REFERENCED_TABLE_NAME),
  }));

  const routineRows = await queryCatalogRows(connection, CATALOG_QUERIES.routines, database, limits);
  const routines = routineRows.map((row) => {
    const type = requireCatalogText(row.ROUTINE_TYPE);
    if (!['FUNCTION', 'PROCEDURE'].includes(type)) {
      throw new SchemaExportError('CATALOG_SHAPE_INVALID', 'Schema 目录包含未知例程类型');
    }
    return { name: requireCatalogText(row.ROUTINE_NAME), type };
  });
  routines.sort((left, right) => left.type.localeCompare(right.type) || left.name.localeCompare(right.name));
  objectCount += routines.length;
  assertObjectLimit(objectCount, limits);

  const triggerRows = await queryCatalogRows(connection, CATALOG_QUERIES.triggers, database, limits);
  const triggers = triggerRows.map((row) => {
    const order = Number(row.ACTION_ORDER);
    if (!Number.isInteger(order) || order < 0) {
      throw new SchemaExportError('CATALOG_SHAPE_INVALID', 'Schema 触发器执行顺序无效');
    }
    return {
      name: requireCatalogText(row.TRIGGER_NAME),
      table: requireCatalogText(row.EVENT_OBJECT_TABLE),
      timing: requireCatalogText(row.ACTION_TIMING),
      event: requireCatalogText(row.EVENT_MANIPULATION),
      order,
    };
  });
  triggers.sort(
    (left, right) =>
      left.table.localeCompare(right.table) ||
      left.timing.localeCompare(right.timing) ||
      left.event.localeCompare(right.event) ||
      left.order - right.order ||
      left.name.localeCompare(right.name),
  );
  objectCount += triggers.length;
  assertObjectLimit(objectCount, limits);

  const eventRows = await queryCatalogRows(connection, CATALOG_QUERIES.events, database, limits);
  const events = eventRows.map((row) => requireCatalogText(row.EVENT_NAME)).sort();
  objectCount += events.length;
  assertObjectLimit(objectCount, limits);

  return Object.freeze({
    tables: Object.freeze(tables),
    views: Object.freeze(views),
    viewDependencies: Object.freeze(viewDependencies.map((dependency) => Object.freeze(dependency))),
    routines: Object.freeze(routines.map((routine) => Object.freeze(routine))),
    triggers: Object.freeze(triggers.map((trigger) => Object.freeze(trigger))),
    events: Object.freeze(events),
  });
}

/**
 * 视图恢复顺序使用稳定 Kahn 拓扑排序。跨库或未知依赖不能被默默忽略，否则恢复
 * 出来的 Schema 可能引用未捕获对象；同层节点始终按名称排序以保证逐字节确定性。
 */
export function orderViewNames(catalog) {
  const tables = new Set(catalog.tables ?? []);
  const views = new Set(catalog.views ?? []);
  const incoming = new Map([...views].map((view) => [view, 0]));
  const dependents = new Map([...views].map((view) => [view, []]));

  for (const dependency of catalog.viewDependencies ?? []) {
    if (dependency.referencedSchema !== DEVELOPMENT_DATABASE) {
      throw new SchemaExportError(
        'CROSS_SCHEMA_VIEW_DEPENDENCY',
        'Schema 视图包含跨库依赖',
      );
    }
    if (!views.has(dependency.view)) {
      throw new SchemaExportError('UNKNOWN_VIEW_DEPENDENCY', 'Schema 视图依赖来源未知');
    }
    if (tables.has(dependency.referencedObject)) continue;
    if (!views.has(dependency.referencedObject)) {
      throw new SchemaExportError('UNKNOWN_VIEW_DEPENDENCY', 'Schema 视图依赖对象未知');
    }
    incoming.set(dependency.view, incoming.get(dependency.view) + 1);
    dependents.get(dependency.referencedObject).push(dependency.view);
  }

  for (const names of dependents.values()) names.sort();
  const ready = [...views].filter((view) => incoming.get(view) === 0).sort();
  const ordered = [];
  while (ready.length > 0) {
    const view = ready.shift();
    ordered.push(view);
    for (const dependent of dependents.get(view)) {
      incoming.set(dependent, incoming.get(dependent) - 1);
      if (incoming.get(dependent) === 0) {
        ready.push(dependent);
        ready.sort();
      }
    }
  }
  if (ordered.length !== views.size) {
    throw new SchemaExportError('VIEW_DEPENDENCY_CYCLE', 'Schema 视图依赖存在循环');
  }
  return ordered;
}