319 lines
16 KiB
JavaScript
319 lines
16 KiB
JavaScript
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;
|
|
}
|