239 lines
13 KiB
JavaScript
239 lines
13 KiB
JavaScript
import test from 'node:test';
|
|
import assert from 'node:assert/strict';
|
|
import { ensureJournal, applyChange, rollbackChanges } from '../lib/journal.mjs';
|
|
|
|
const clone = value => Buffer.isBuffer(value) ? Buffer.from(value)
|
|
: Array.isArray(value) ? value.map(clone)
|
|
: value && typeof value === 'object' ? Object.fromEntries(Object.entries(value).map(([key, item]) => [key, clone(item)])) : value;
|
|
const same = (a, b) => Buffer.isBuffer(a) && Buffer.isBuffer(b) ? a.equals(b) : a === b || (a !== null && b !== null && String(a) === String(b));
|
|
const names = sql => [...sql.matchAll(/CAST\(`((?:``|[^`])+)` AS BINARY\)/g)].map(match => match[1].replaceAll('``', '`'));
|
|
|
|
class MemoryConnection {
|
|
constructor() {
|
|
this.calls = [];
|
|
this.rows = { '9007199254740993': { id: '9007199254740993', avatar: 'https://old/a', nullable: null, deleted: Buffer.from([0]), update_time: 'unchanged', touched: 'unchanged' } };
|
|
this.journals = [];
|
|
this.nextId = 9007199254740993n;
|
|
this.engine = 'InnoDB';
|
|
this.failJournal = false;
|
|
this.normalizeUpdate = null;
|
|
this.warningStatus = 0;
|
|
}
|
|
|
|
async beginTransaction() { this.calls.push({ transaction: 'begin' }); this.saved = { rows: clone(this.rows), journals: clone(this.journals) }; }
|
|
async commit() { this.calls.push({ transaction: 'commit' }); this.saved = undefined; }
|
|
async rollback() { this.calls.push({ transaction: 'rollback' }); if (this.saved) { this.rows = this.saved.rows; this.journals = this.saved.journals; this.saved = undefined; } }
|
|
|
|
async execute(sql, params = []) {
|
|
this.calls.push({ sql, params });
|
|
if (sql.startsWith('CREATE TABLE')) return [{ affectedRows: 0 }];
|
|
if (sql.includes('FROM information_schema.COLUMNS')) {
|
|
return [[...['id', 'avatar', 'nullable', 'deleted', 'update_time', 'touched', 'json_data', 'counter'].map(name => ({ name, engine: this.engine, extra: ['update_time', 'touched'].includes(name) ? 'on update CURRENT_TIMESTAMP' : '' }))]];
|
|
}
|
|
if (sql.startsWith('INSERT INTO')) {
|
|
if (this.failJournal) throw Object.assign(new Error('journal unavailable'), { code: 'TEST_JOURNAL_FAILURE' });
|
|
const [run_id, schema_name, table_name, pk_name, pk_value, before_json, after_json] = params;
|
|
const id = (this.nextId++).toString();
|
|
this.journals.push({ id, run_id, schema_name, table_name, pk_name, pk_value, before_json, after_json, rolled_back: 0 });
|
|
return [{ insertId: id }];
|
|
}
|
|
if (sql.startsWith('SELECT CAST(id AS CHAR)')) {
|
|
const [runId, cursor] = params;
|
|
return [this.journals.filter(row => row.run_id === runId && row.rolled_back === 0 && (cursor === undefined || BigInt(row.id) < BigInt(cursor)))
|
|
.sort((a, b) => BigInt(a.id) > BigInt(b.id) ? -1 : 1).slice(0, 100).map(row => ({ id: row.id }))];
|
|
}
|
|
if (sql.startsWith('SELECT *')) return [[this.journals.find(row => row.id === params[0] && row.run_id === params[1])].filter(Boolean)];
|
|
if (sql.startsWith('SELECT 1 AS matched')) {
|
|
const row = this.rows[String(params[0])];
|
|
return [row && names(sql).every((name, index) => same(row[name], params[index + 1])) ? [{ matched: 1 }] : []];
|
|
}
|
|
if (sql.startsWith('SELECT `')) {
|
|
const row = this.rows[String(params[0])];
|
|
const selected = [...sql.split(' FROM ')[0].matchAll(/`((?:``|[^`])+)`/g)].map(match => match[1]);
|
|
return [row ? [Object.fromEntries(selected.map(name => [name, row[name]]))] : []];
|
|
}
|
|
if (sql.startsWith('UPDATE') && sql.includes('SET rolled_back = 1')) {
|
|
const row = this.journals.find(row => row.id === params[0] && row.run_id === params[1]);
|
|
if (row) row.rolled_back = 1;
|
|
return [{ affectedRows: row ? 1 : 0 }];
|
|
}
|
|
if (sql.startsWith('UPDATE')) {
|
|
const set = sql.split(' SET ')[1].split('WHERE')[0];
|
|
const changed = [...set.matchAll(/`((?:``|[^`])+)` = \?/g)].map(match => match[1]);
|
|
const row = this.rows[String(params[changed.length])];
|
|
if (!row || !names(sql).every((name, index) => same(row[name], params[changed.length + 1 + index]))) return [{ affectedRows: 0 }];
|
|
changed.forEach((name, index) => { row[name] = params[index]; });
|
|
if (this.normalizeUpdate) this.normalizeUpdate(row, changed);
|
|
if (!set.includes('`update_time` = `update_time`')) row.update_time = 'changed';
|
|
if (!set.includes('`touched` = `touched`')) row.touched = 'changed';
|
|
return [{ affectedRows: 1, warningStatus: this.warningStatus }];
|
|
}
|
|
throw new Error(`Unexpected test query: ${sql}`);
|
|
}
|
|
}
|
|
|
|
const change = (before = 'https://old/a', after = 'https://oss/a') => ({
|
|
schema: 'app', table: 'member_user', primaryKey: 'id', pkValue: '9007199254740993',
|
|
before: { avatar: before, deleted: Buffer.from([0]), nullable: null }, after: { avatar: after }
|
|
});
|
|
|
|
async function commitChange(connection, payload = change(), runId = 'run') {
|
|
await connection.beginTransaction();
|
|
try {
|
|
const result = await applyChange(connection, 'app', runId, payload);
|
|
await connection.commit();
|
|
return result;
|
|
} catch (error) { await connection.rollback(); throw error; }
|
|
}
|
|
|
|
test('journal schema uses InnoDB and independent run/item/change records', async () => {
|
|
const connection = new MemoryConnection();
|
|
await ensureJournal(connection, 'app');
|
|
const creates = connection.calls.filter(call => call.sql?.startsWith('CREATE TABLE'));
|
|
assert.equal(creates.length, 3);
|
|
assert.ok(creates.every(call => call.sql.includes('ENGINE=InnoDB')));
|
|
assert.match(creates[1].sql, /PRIMARY KEY \(run_id, item_key\)/);
|
|
assert.match(creates[2].sql, /KEY ix_run_id_id \(run_id, id\)/);
|
|
});
|
|
|
|
test('apply logs original values with binary null-safe guards and lossless bigint IDs', async () => {
|
|
const connection = new MemoryConnection();
|
|
const result = await commitChange(connection);
|
|
assert.equal(result.journalId, '9007199254740993');
|
|
assert.equal(connection.rows['9007199254740993'].avatar, 'https://oss/a');
|
|
assert.equal(connection.rows['9007199254740993'].update_time, 'unchanged');
|
|
assert.equal(connection.rows['9007199254740993'].touched, 'unchanged');
|
|
const update = connection.calls.find(call => call.sql?.startsWith('UPDATE'));
|
|
assert.match(update.sql, /CAST\(`avatar` AS BINARY\) <=> CAST\(\? AS BINARY\)/);
|
|
assert.match(update.sql, /CAST\(`nullable` AS BINARY\) <=> CAST\(\? AS BINARY\)/);
|
|
assert.equal(update.params[1], '9007199254740993');
|
|
assert.deepEqual(update.params[3], Buffer.from([0]));
|
|
assert.equal(update.params[4], null);
|
|
assert.equal(connection.journals.length, 1);
|
|
});
|
|
|
|
test('no-op changes do not update rows or create logs', async () => {
|
|
const connection = new MemoryConnection();
|
|
assert.deepEqual(await commitChange(connection, change('https://old/a', 'https://old/a')), { changed: false });
|
|
assert.equal(connection.journals.length, 0);
|
|
assert.equal(connection.calls.filter(call => call.sql?.startsWith('UPDATE')).length, 0);
|
|
});
|
|
|
|
test('primary-key changes and missing snapshots are rejected before mutation', async () => {
|
|
const connection = new MemoryConnection();
|
|
await assert.rejects(commitChange(connection, { ...change(), after: { id: '1' } }), /不能修改主键/);
|
|
await assert.rejects(commitChange(connection, { ...change(), before: { deleted: Buffer.from([0]) } }), /缺少原值快照/);
|
|
assert.equal(connection.journals.length, 0);
|
|
});
|
|
|
|
test('concurrent edits including case-only changes fail CAS and preserve the row', async () => {
|
|
const connection = new MemoryConnection();
|
|
connection.rows['9007199254740993'].avatar = 'https://old/A';
|
|
await assert.rejects(commitChange(connection), { code: 'MIGRATION_CAS_CONFLICT' });
|
|
assert.equal(connection.rows['9007199254740993'].avatar, 'https://old/A');
|
|
assert.equal(connection.journals.length, 0);
|
|
});
|
|
|
|
test('a journal INSERT failure is rolled back with the row by the caller transaction', async () => {
|
|
const connection = new MemoryConnection();
|
|
connection.failJournal = true;
|
|
await assert.rejects(commitChange(connection), { code: 'TEST_JOURNAL_FAILURE' });
|
|
assert.equal(connection.rows['9007199254740993'].avatar, 'https://old/a');
|
|
assert.equal(connection.journals.length, 0);
|
|
assert.equal(connection.calls.at(-1).transaction, 'rollback');
|
|
});
|
|
|
|
test('journal reads back native JSON canonicalization and numeric conversions before recording rollback values', async () => {
|
|
const connection = new MemoryConnection();
|
|
connection.rows['9007199254740993'].json_data = '{"avatar": "https://old/a"}';
|
|
connection.rows['9007199254740993'].counter = 1;
|
|
connection.normalizeUpdate = (row, changed) => {
|
|
if (changed.includes('json_data')) row.json_data = JSON.stringify(JSON.parse(row.json_data)).replaceAll('\":', '\": ');
|
|
if (changed.includes('counter')) row.counter = Number(row.counter);
|
|
};
|
|
await commitChange(connection, {
|
|
...change(),
|
|
before: { json_data: '{"avatar": "https://old/a"}', counter: 1 },
|
|
after: { json_data: '{"avatar":"https://oss/a"}', counter: '2' }
|
|
});
|
|
const after = JSON.parse(connection.journals[0].after_json);
|
|
assert.equal(after.json_data.value, '{"avatar": "https://oss/a"}');
|
|
assert.equal(after.counter.value, 2);
|
|
const result = await rollbackChanges(connection, 'app', 'run');
|
|
assert.deepEqual(result, { restored: 1, alreadyRestored: 0, conflicts: [] });
|
|
assert.equal(connection.rows['9007199254740993'].json_data, '{"avatar": "https://old/a"}');
|
|
assert.equal(connection.rows['9007199254740993'].counter, 1);
|
|
});
|
|
|
|
test('silent SQL truncation warnings abort the transaction before recording success', async () => {
|
|
const connection = new MemoryConnection();
|
|
connection.warningStatus = 1;
|
|
await assert.rejects(commitChange(connection), /数据库更新产生警告/);
|
|
assert.equal(connection.rows['9007199254740993'].avatar, 'https://old/a');
|
|
assert.equal(connection.journals.length, 0);
|
|
});
|
|
|
|
test('rollback restores consecutive edits in reverse journal order without losing bigint precision', async () => {
|
|
const connection = new MemoryConnection();
|
|
await commitChange(connection);
|
|
await commitChange(connection, change('https://oss/a', 'https://oss/b'));
|
|
const result = await rollbackChanges(connection, 'app', 'run');
|
|
assert.deepEqual(result, { restored: 2, alreadyRestored: 0, conflicts: [] });
|
|
assert.equal(connection.rows['9007199254740993'].avatar, 'https://old/a');
|
|
assert.equal(connection.rows['9007199254740993'].update_time, 'unchanged');
|
|
const reads = connection.calls.filter(call => call.sql?.startsWith('SELECT *'));
|
|
assert.deepEqual(reads.map(call => call.params[0]), ['9007199254740994', '9007199254740993']);
|
|
assert.ok(connection.journals.every(row => row.rolled_back === 1));
|
|
assert.deepEqual(await rollbackChanges(connection, 'app', 'run'), { restored: 0, alreadyRestored: 0, conflicts: [] });
|
|
});
|
|
|
|
test('rollback detects an already-restored row and only marks the journal', async () => {
|
|
const connection = new MemoryConnection();
|
|
await commitChange(connection);
|
|
connection.rows['9007199254740993'].avatar = 'https://old/a';
|
|
const result = await rollbackChanges(connection, 'app', 'run');
|
|
assert.deepEqual(result, { restored: 0, alreadyRestored: 1, conflicts: [] });
|
|
assert.equal(connection.journals[0].rolled_back, 1);
|
|
});
|
|
|
|
test('rollback SQL warnings do not mark a truncated restoration as complete', async () => {
|
|
const connection = new MemoryConnection();
|
|
await commitChange(connection);
|
|
connection.warningStatus = 1;
|
|
await assert.rejects(rollbackChanges(connection, 'app', 'run'), /数据库回滚产生警告/);
|
|
assert.equal(connection.rows['9007199254740993'].avatar, 'https://oss/a');
|
|
assert.equal(connection.journals[0].rolled_back, 0);
|
|
});
|
|
|
|
test('rollback leaves later user edits and deletion guards unchanged', async () => {
|
|
for (const update of [{ avatar: 'https://user/new' }, { deleted: Buffer.from([1]) }]) {
|
|
const connection = new MemoryConnection();
|
|
await commitChange(connection);
|
|
Object.assign(connection.rows['9007199254740993'], update);
|
|
const original = clone(connection.rows['9007199254740993']);
|
|
const result = await rollbackChanges(connection, 'app', 'run');
|
|
assert.equal(result.restored, 0);
|
|
assert.equal(result.conflicts.length, 1);
|
|
assert.equal(result.conflicts[0].pkValue, '9007199254740993');
|
|
assert.deepEqual(connection.rows['9007199254740993'], original);
|
|
assert.equal(connection.journals[0].rolled_back, 0);
|
|
}
|
|
});
|
|
|
|
test('rollback does not resurrect missing rows or touch another run', async () => {
|
|
const connection = new MemoryConnection();
|
|
await commitChange(connection);
|
|
delete connection.rows['9007199254740993'];
|
|
assert.deepEqual(await rollbackChanges(connection, 'app', 'another-run'), { restored: 0, alreadyRestored: 0, conflicts: [] });
|
|
const result = await rollbackChanges(connection, 'app', 'run');
|
|
assert.equal(result.conflicts.length, 1);
|
|
assert.equal(Object.keys(connection.rows).length, 0);
|
|
});
|
|
|
|
test('non-transactional tables are rejected before any update', async () => {
|
|
const connection = new MemoryConnection();
|
|
connection.engine = 'MyISAM';
|
|
await assert.rejects(commitChange(connection), /必须为 InnoDB/);
|
|
assert.equal(connection.calls.filter(call => call.sql?.startsWith('UPDATE')).length, 0);
|
|
});
|