88 lines
4.5 KiB
JavaScript
88 lines
4.5 KiB
JavaScript
import { id, qualified } from './core.mjs';
|
|
|
|
function sourceChanged() {
|
|
return Object.assign(new Error('源图片在任务建立后发生变化或已删除，请检查该批文件'), {
|
|
code: 'MIGRATION_SOURCE_CHANGED'
|
|
});
|
|
}
|
|
|
|
/**
|
|
* Validate a bounded batch immediately after the caller begins an InnoDB
|
|
* REPEATABLE READ transaction, BEFORE any consistent read or explicitly created
|
|
* consistent snapshot. The plain metadata read must establish its snapshot
|
|
* only after the primary-key range has been locked.
|
|
*
|
|
* The first locking read scans from the minimum expected ID through the upper
|
|
* key range, including deleted rows and potential newer versions. This retains
|
|
* range/gap protection until commit/rollback, including against undeleting or
|
|
* changing another row into a newer matching version. Do not restrict this
|
|
* range lock to only expected IDs, active rows, or matching paths.
|
|
*
|
|
* The caller owns transaction boundaries, rollback on failure and batch size.
|
|
*/
|
|
export async function verifySourceBatch(db, config, items) {
|
|
if (!Array.isArray(items) || items.length === 0) throw new TypeError('源文件校验批次不能为空');
|
|
if (!config?.mysql?.database) throw new TypeError('源文件校验需要数据库名称');
|
|
const expected = new Map();
|
|
for (const item of items) {
|
|
const contentId = id(item?.contentId);
|
|
const configId = id(item?.configId);
|
|
if (expected.has(contentId)) throw new TypeError('源文件校验批次不能包含重复的内容编号');
|
|
if (typeof item.path !== 'string' || !item.path.length) throw new TypeError('源文件路径必须为非空字符串');
|
|
if (!Number.isSafeInteger(item.size) || item.size < 0 || typeof item.sha256 !== 'string' || !item.sha256.length) {
|
|
throw new TypeError('源文件校验需要有效的预期大小和 SHA-256');
|
|
}
|
|
expected.set(contentId, { ...item, contentId, configId });
|
|
}
|
|
const entries = [...expected.values()];
|
|
const minimumId = entries.reduce((minimum, item) => BigInt(item.contentId) < BigInt(minimum) ? item.contentId : minimum, entries[0].contentId);
|
|
const sourceTable = qualified(config.mysql.database, 'infra_file_content');
|
|
const cases = entries.map(() => 'WHEN config_id=? AND path=? THEN ?').join(' ');
|
|
const predicates = entries.map(() => '(config_id=? AND path=? AND id>=?)').join(' OR ');
|
|
const caseParams = entries.flatMap(item => [item.configId, item.path, item.contentId]);
|
|
const predicateParams = entries.flatMap(item => [item.configId, item.path, item.contentId]);
|
|
|
|
// A primary-key-only locking scan avoids the large-row access path triggered
|
|
// by locking reads that evaluate config/path predicates. Do not filter out
|
|
// deleted rows: their undelete could otherwise create a new active version.
|
|
await db.execute(
|
|
`SELECT id FROM ${sourceTable} WHERE id>=? ORDER BY id FOR UPDATE`,
|
|
[minimumId]
|
|
);
|
|
|
|
// CASE is evaluated by MySQL using the path column's collation. The returned
|
|
// expected ID is the only grouping key used in JavaScript. This consistent
|
|
// read establishes a fresh snapshot while the whole relevant range is held.
|
|
const [latestRows] = await db.execute(
|
|
`SELECT id,CASE ${cases} END AS expected_content_id FROM ${sourceTable}
|
|
WHERE deleted=0 AND id>=? AND (${predicates}) ORDER BY id DESC`,
|
|
[...caseParams, minimumId, ...predicateParams]
|
|
);
|
|
const latest = new Map();
|
|
for (const row of latestRows) {
|
|
const expectedId = String(row.expected_content_id);
|
|
if (!expected.has(expectedId)) throw sourceChanged();
|
|
const actualId = id(row.id);
|
|
if (!latest.has(expectedId) || BigInt(actualId) > BigInt(latest.get(expectedId))) latest.set(expectedId, actualId);
|
|
}
|
|
for (const contentId of expected.keys()) {
|
|
if (latest.get(contentId) !== contentId) throw sourceChanged();
|
|
}
|
|
|
|
// The range is already locked and each config/path version is validated.
|
|
// Digest work now touches only the selected primary keys, not a sorted scan.
|
|
const [contentRows] = await db.execute(
|
|
`SELECT id,config_id,path,OCTET_LENGTH(content) AS size,LOWER(SHA2(content,256)) AS sha256
|
|
FROM ${sourceTable} WHERE deleted=0 AND id IN (${entries.map(() => '?').join(',')}) FOR UPDATE`,
|
|
entries.map(item => item.contentId)
|
|
);
|
|
const verified = new Set();
|
|
for (const row of contentRows) {
|
|
const contentId = id(row.id);
|
|
const item = expected.get(contentId);
|
|
if (!item || verified.has(contentId) || Number(row.size) !== item.size || row.sha256 !== item.sha256) throw sourceChanged();
|
|
verified.add(contentId);
|
|
}
|
|
if (verified.size !== expected.size) throw sourceChanged();
|
|
}
|