import { qualified, quote } from './core.mjs';

const RUNS = 'infra_file_oss_migration_runs';
const ITEMS = 'infra_file_oss_migration_items';
const CHANGES = 'infra_file_oss_migration_changes';
const metadataCache = new WeakMap();
const own = (object, key) => Object.prototype.hasOwnProperty.call(object, key);

// Snapshots must preserve BIT/Buffer and 64-bit keys without Number conversion.
function pack(value) {
  if (Buffer.isBuffer(value)) return { type: 'buffer', value: value.toString('base64') };
  if (typeof value === 'bigint') return { type: 'bigint', value: value.toString() };
  if (value instanceof Date) return { type: 'date', value: value.toISOString() };
  if (value === undefined) throw new Error('变更快照不能包含 undefined');
  return { type: 'json', value };
}

function unpack(value) {
  if (value.type === 'buffer') return Buffer.from(value.value, 'base64');
  if (value.type === 'bigint') return BigInt(value.value);
  if (value.type === 'date') return new Date(value.value);
  if (value.type === 'json') return value.value;
  throw new Error('变更日志值类型不受支持');
}

function snapshot(record) {
  return JSON.stringify(Object.fromEntries(Object.entries(record).map(([key, value]) => [key, pack(value)])));
}

function restoreSnapshot(text) {
  return Object.fromEntries(Object.entries(JSON.parse(text)).map(([key, value]) => [key, unpack(value)]));
}

function parameter(value) {
  if (typeof value === 'bigint') return value.toString();
  if (value !== null && typeof value === 'object' && !Buffer.isBuffer(value) && !(value instanceof Date)) return JSON.stringify(value);
  return value;
}

const equal = (left, right) => JSON.stringify(pack(left)) === JSON.stringify(pack(right));

function condition(values) {
  const entries = Object.entries(values);
  return {
    sql: entries.map(([key]) => `CAST(${quote(key)} AS BINARY) <=> CAST(? AS BINARY)`).join(' AND '),
    params: entries.map(([, value]) => parameter(value))
  };
}

async function tableMetadata(connection, schema, table) {
  let cache = metadataCache.get(connection);
  if (!cache) { cache = new Map(); metadataCache.set(connection, cache); }
  const key = JSON.stringify([schema, table]);
  if (!cache.has(key)) {
    const [rows] = await connection.execute(
      `SELECT c.COLUMN_NAME AS name, c.EXTRA AS extra, t.ENGINE AS engine
       FROM information_schema.COLUMNS c JOIN information_schema.TABLES t
         ON t.TABLE_SCHEMA = c.TABLE_SCHEMA AND t.TABLE_NAME = c.TABLE_NAME
       WHERE c.TABLE_SCHEMA = ? AND c.TABLE_NAME = ?`, [schema, table]);
    if (!rows.length || rows.some(row => String(row.engine).toLowerCase() !== 'innodb')) {
      throw new Error(`表 ${schema}.${table} 必须为 InnoDB，才能保证更新和日志原子提交`);
    }
    cache.set(key, {
      names: new Set(rows.map(row => row.name)),
      automatic: rows.filter(row => /on update/i.test(row.extra || '')).map(row => row.name)
    });
  }
  return cache.get(key);
}

function assertFields(metadata, primaryKey, before, after) {
  if (own(after, primaryKey)) throw new Error('迁移不能修改主键');
  for (const name of [primaryKey, ...Object.keys(before), ...Object.keys(after)]) {
    if (!metadata.names.has(name)) throw new Error(`变更列 ${name} 不存在`);
  }
  for (const name of Object.keys(after)) {
    if (!own(before, name)) throw new Error(`变更列 ${name} 缺少原值快照`);
  }
}

function assignments(values, metadata) {
  const entries = Object.entries(values);
  return {
    sql: [
      ...entries.map(([name]) => `${quote(name)} = ?`),
      ...metadata.automatic.filter(name => !own(values, name)).map(name => `${quote(name)} = ${quote(name)}`)
    ].join(', '),
    params: entries.map(([, value]) => parameter(value))
  };
}

export async function ensureJournal(connection, schema) {
  await connection.execute(`CREATE TABLE IF NOT EXISTS ${qualified(schema, RUNS)} (
    run_id VARCHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL PRIMARY KEY,
    settings LONGTEXT NOT NULL,
    status VARCHAR(32) NOT NULL DEFAULT 'active',
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin`);
  await connection.execute(`CREATE TABLE IF NOT EXISTS ${qualified(schema, ITEMS)} (
    run_id VARCHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
    item_key CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
    data LONGTEXT NOT NULL,
    status VARCHAR(32) NOT NULL,
    error TEXT NULL,
    updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    PRIMARY KEY (run_id, item_key)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin`);
  await connection.execute(`CREATE TABLE IF NOT EXISTS ${qualified(schema, CHANGES)} (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
    run_id VARCHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
    schema_name VARCHAR(64) NOT NULL,
    table_name VARCHAR(64) NOT NULL,
    pk_name VARCHAR(64) NOT NULL,
    pk_value TEXT NOT NULL,
    before_json LONGTEXT NOT NULL,
    after_json LONGTEXT NOT NULL,
    rolled_back TINYINT NOT NULL DEFAULT 0,
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    KEY ix_run_id_id (run_id, id)
  ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin`);
  for (const table of [RUNS, ITEMS, CHANGES]) await tableMetadata(connection, schema, table);
}

/** The caller owns BEGIN/COMMIT/ROLLBACK, including the change-journal INSERT. */
export async function applyChange(connection, journalSchema, runId, change) {
  const { schema, table, primaryKey, pkValue, before, after } = change;
  const metadata = await tableMetadata(connection, schema, table);
  assertFields(metadata, primaryKey, before, after);
  const changed = Object.fromEntries(Object.entries(after).filter(([name, value]) => !equal(before[name], value)));
  if (!Object.keys(changed).length) return { changed: false };
  const update = assignments(changed, metadata);
  const guard = condition(before);
  const [result] = await connection.execute(
    `UPDATE ${qualified(schema, table)} SET ${update.sql}
     WHERE ${quote(primaryKey)} = ?${guard.sql ? ` AND ${guard.sql}` : ''}`,
    [...update.params, parameter(pkValue), ...guard.params]);
  if (result.affectedRows !== 1) {
    const error = new Error(`数据已变化或不存在：${schema}.${table}，本次更新已停止`);
    error.code = 'MIGRATION_CAS_CONFLICT';
    throw error;
  }
  if (result.warningStatus > 0) {
    throw new Error('数据库更新产生警告，可能发生字段截断或类型转换；事务必须回滚');
  }
  // Native JSON columns normalize whitespace/key order, and MySQL may convert
  // numeric request values. Rollback must compare against the bytes/type that
  // the database actually persisted, not the JavaScript UPDATE parameters.
  const [storedRows] = await connection.execute(
    `SELECT ${Object.keys(changed).map(quote).join(', ')} FROM ${qualified(schema, table)}
     WHERE ${quote(primaryKey)} = ? FOR UPDATE`, [parameter(pkValue)]);
  if (storedRows.length !== 1 || Object.keys(changed).some(name => !own(storedRows[0], name))) {
    throw new Error('无法读取更新后的记录，事务必须回滚');
  }
  const persisted = Object.fromEntries(Object.keys(changed).map(name => [name, storedRows[0][name]]));
  const [journal] = await connection.execute(
    `INSERT INTO ${qualified(journalSchema, CHANGES)}
     (run_id, schema_name, table_name, pk_name, pk_value, before_json, after_json)
     VALUES (?, ?, ?, ?, ?, ?, ?)`,
    [runId, schema, table, primaryKey, JSON.stringify(pack(pkValue)), snapshot(before), snapshot(persisted)]);
  return { changed: true, journalId: String(journal.insertId) };
}

function primaryKeyIdentity(value) {
  if (Buffer.isBuffer(value)) return `buffer:${value.toString('base64')}`;
  if (!['string', 'number', 'bigint'].includes(typeof value)
      || (typeof value === 'number' && !Number.isSafeInteger(value))) {
    throw new Error('批量变更主键必须是字符串、安全整数、BigInt 或 Buffer');
  }
  // mysql2 may return integer primary keys as either strings or numbers.
  return `scalar:${String(value)}`;
}

/**
 * Apply one homogeneous batch with one CASE UPDATE, one persisted-value read,
 * and one journal INSERT. The caller must roll back the entire transaction on
 * any error: the UPDATE can have matched only part of a concurrently edited
 * batch before affectedRows reports the conflict.
 */
export async function applyChanges(connection, journalSchema, runId, changes) {
  if (!Array.isArray(changes)) throw new Error('批量变更必须为数组');
  if (!changes.length) return { changed: 0, skipped: 0 };
  const { schema, table, primaryKey } = changes[0];
  const afterFields = Object.keys(changes[0].after).sort();
  const identities = new Set();
  for (const change of changes) {
    if (change.schema !== schema || change.table !== table || change.primaryKey !== primaryKey) {
      throw new Error('批量变更必须使用相同的数据库、表和主键列');
    }
    if (JSON.stringify(Object.keys(change.after).sort()) !== JSON.stringify(afterFields)) {
      throw new Error('批量变更的目标字段集合必须一致');
    }
    const identity = primaryKeyIdentity(change.pkValue);
    if (identities.has(identity)) throw new Error('批量变更包含重复主键');
    identities.add(identity);
  }
  const metadata = await tableMetadata(connection, schema, table);
  const active = [];
  for (const change of changes) {
    assertFields(metadata, primaryKey, change.before, change.after);
    const changed = Object.fromEntries(Object.entries(change.after).filter(([name, value]) => !equal(change.before[name], value)));
    if (Object.keys(changed).length) active.push({ ...change, changed });
  }
  const skipped = changes.length - active.length;
  if (!active.length) return { changed: 0, skipped };

  const fields = afterFields.filter(name => active.some(change => own(change.changed, name)));
  const updateParams = [];
  const set = fields.map(name => {
    const branches = active.filter(change => own(change.changed, name)).map(change => {
      updateParams.push(parameter(change.pkValue), parameter(change.changed[name]));
      return 'WHEN ? THEN ?';
    });
    return `${quote(name)} = CASE ${quote(primaryKey)} ${branches.join(' ')} ELSE ${quote(name)} END`;
  });
  set.push(...metadata.automatic.filter(name => !fields.includes(name)).map(name => `${quote(name)} = ${quote(name)}`));
  const where = active.map(change => {
    const guard = condition(change.before);
    updateParams.push(parameter(change.pkValue), ...guard.params);
    return `(${quote(primaryKey)} = ?${guard.sql ? ` AND ${guard.sql}` : ''})`;
  });
  const [updated] = await connection.execute(
    `UPDATE ${qualified(schema, table)} SET ${set.join(', ')} WHERE ${where.join(' OR ')}`, updateParams);
  if (updated.affectedRows !== active.length) {
    const error = new Error(`批量更新发现数据已变化或不存在：${schema}.${table}，必须回滚整批事务`);
    error.code = 'MIGRATION_CAS_CONFLICT';
    throw error;
  }
  if (updated.warningStatus > 0) throw new Error('数据库批量更新产生警告，可能发生字段截断或类型转换；事务必须回滚');

  const [storedRows] = await connection.execute(
    `SELECT ${[primaryKey, ...fields].map(quote).join(', ')} FROM ${qualified(schema, table)}
     WHERE ${quote(primaryKey)} IN (${active.map(() => '?').join(', ')}) FOR UPDATE`,
    active.map(change => parameter(change.pkValue)));
  if (storedRows.length !== active.length) throw new Error('无法读取批量更新后的全部记录，事务必须回滚');
  const storedByKey = new Map();
  for (const row of storedRows) {
    if (![primaryKey, ...fields].every(name => own(row, name))) throw new Error('批量更新读回记录缺少字段，事务必须回滚');
    const identity = primaryKeyIdentity(row[primaryKey]);
    if (storedByKey.has(identity)) throw new Error('批量更新读回记录包含重复主键，事务必须回滚');
    storedByKey.set(identity, row);
  }
  const journalParams = [];
  for (const change of active) {
    const row = storedByKey.get(primaryKeyIdentity(change.pkValue));
    if (!row) throw new Error('批量更新读回主键与请求不一致，事务必须回滚');
    const persisted = Object.fromEntries(Object.keys(change.changed).map(name => [name, row[name]]));
    journalParams.push(runId, schema, table, primaryKey, JSON.stringify(pack(change.pkValue)), snapshot(change.before), snapshot(persisted));
  }
  const [journal] = await connection.execute(
    `INSERT INTO ${qualified(journalSchema, CHANGES)}
     (run_id, schema_name, table_name, pk_name, pk_value, before_json, after_json)
     VALUES ${active.map(() => '(?, ?, ?, ?, ?, ?, ?)').join(', ')}`, journalParams);
  if (journal.affectedRows !== active.length || journal.warningStatus > 0) {
    throw new Error('批量回滚日志未完整保存，事务必须回滚');
  }
  return { changed: active.length, skipped };
}

export async function rollbackChanges(connection, journalSchema, runId) {
  const result = { restored: 0, alreadyRestored: 0, conflicts: [] };
  let cursor = null;
  const journal = qualified(journalSchema, CHANGES);
  while (true) {
    const [batch] = await connection.execute(
      `SELECT CAST(id AS CHAR) AS id FROM ${journal}
       WHERE run_id = ? AND rolled_back = 0${cursor === null ? '' : ' AND id < ?'} ORDER BY id DESC LIMIT 100`,
      cursor === null ? [runId] : [runId, cursor]);
    if (!batch.length) break;
    for (const entry of batch) {
      cursor = String(entry.id);
      await connection.beginTransaction();
      try {
        const [[row]] = await connection.execute(`SELECT * FROM ${journal} WHERE id = ? AND run_id = ? FOR UPDATE`, [cursor, runId]);
        if (!row || Number(row.rolled_back) === 1) { await connection.commit(); continue; }
        const before = restoreSnapshot(row.before_json);
        const after = restoreSnapshot(row.after_json);
        const pkValue = unpack(JSON.parse(row.pk_value));
        const metadata = await tableMetadata(connection, row.schema_name, row.table_name);
        assertFields(metadata, row.pk_name, before, after);
        const target = qualified(row.schema_name, row.table_name);
        const expectedAfter = condition({ ...before, ...after });
        const [matchesAfter] = await connection.execute(
          `SELECT 1 AS matched FROM ${target} WHERE ${quote(row.pk_name)} = ? AND ${expectedAfter.sql} FOR UPDATE`,
          [parameter(pkValue), ...expectedAfter.params]);
        let alreadyRestored = false;
        if (!matchesAfter.length) {
          const expectedBefore = condition(before);
          const [matchesBefore] = await connection.execute(
            `SELECT 1 AS matched FROM ${target} WHERE ${quote(row.pk_name)} = ? AND ${expectedBefore.sql} FOR UPDATE`,
            [parameter(pkValue), ...expectedBefore.params]);
          if (!matchesBefore.length) {
            await connection.rollback();
            result.conflicts.push({ id: cursor, schema: row.schema_name, table: row.table_name, pkValue: typeof pkValue === 'bigint' ? pkValue.toString() : pkValue });
            continue;
          }
          alreadyRestored = true;
        } else {
          const values = Object.fromEntries(Object.keys(after).map(name => [name, before[name]]));
          const update = assignments(values, metadata);
          const [updated] = await connection.execute(
            `UPDATE ${target} SET ${update.sql} WHERE ${quote(row.pk_name)} = ? AND ${expectedAfter.sql}`,
            [...update.params, parameter(pkValue), ...expectedAfter.params]);
          if (updated.affectedRows !== 1) throw new Error('已锁定的回滚记录发生变化，事务已终止');
          if (updated.warningStatus > 0) throw new Error('数据库回滚产生警告，事务已终止');
        }
        await connection.execute(`UPDATE ${journal} SET rolled_back = 1 WHERE id = ? AND run_id = ?`, [cursor, runId]);
        await connection.commit();
        if (alreadyRestored) result.alreadyRestored++;
        else result.restored++;
      } catch (error) {
        await connection.rollback();
        throw error;
      }
    }
  }
  return result;
}