583 lines
36 KiB
JavaScript
583 lines
36 KiB
JavaScript
#!/usr/bin/env node
|
||
import { readFile, mkdir, writeFile } from 'node:fs/promises';
|
||
import { resolve, dirname } from 'node:path';
|
||
import { fileURLToPath } from 'node:url';
|
||
import { isDeepStrictEqual } from 'node:util';
|
||
import { aliases, bit, downloadUrl, encodePath, ensureObject, errorText, mimeType, normalizeConfig, objectKey, ossSettings, parseStorage, qualified, quote, readHttp, retry, sha256, verifyContent } from './lib/core.mjs';
|
||
import { containsLegacyReference, discoverReferenceColumns, rewriteReferences } from './lib/references.mjs';
|
||
import { ensureJournal, applyChange, rollbackChanges } from './lib/journal.mjs';
|
||
import { loadProjectConfig } from './lib/project-config.mjs';
|
||
import { discoverStorageConfig } from './lib/auto-storage.mjs';
|
||
import { loadSourceIndex } from './lib/source-index.mjs';
|
||
import { createProgress } from './lib/progress.mjs';
|
||
import { switchFiles } from './lib/transition.mjs';
|
||
import { mapBounded } from './lib/concurrency.mjs';
|
||
|
||
const SCRIPT_DIR = dirname(fileURLToPath(import.meta.url));
|
||
const PROJECT_ROOT = resolve(SCRIPT_DIR, '../..');
|
||
const TRANSITION_BATCH_SIZE = 50;
|
||
const VALIDATION_CONCURRENCY = 4;
|
||
const seconds = milliseconds => (milliseconds / 1000).toFixed(2);
|
||
|
||
const HELP = `数据库图片迁移到阿里云 OSS（Node.js 20.19+）
|
||
node migrate.mjs 自动读取项目配置，只读预检
|
||
node migrate.mjs init 自动生成本机配置，只查询存储配置及已有批次
|
||
node migrate.mjs apply 执行迁移；可加 --limit 50 先迁移少量
|
||
node migrate.mjs rollback 回滚当前批次的数据库修改
|
||
可选：--profile dev | --config 文件路径 | --run-id 新批次名称
|
||
默认复用项目当前环境的主数据库和默认 OSS，不需要填写数据库密码或存储编号。
|
||
保留原 BLOB、旧配置和 OSS 对象；rollback 只撤销本批次数据库修改。`;
|
||
|
||
export function parseArgs(argv) {
|
||
if (argv.includes('--help') || argv.includes('-h')) return { help: true };
|
||
const hasCommand = argv.length && !argv[0].startsWith('--');
|
||
const command = hasCommand ? argv[0] : 'plan';
|
||
const rest = hasCommand ? argv.slice(1) : argv;
|
||
if (!['init', 'plan', 'apply', 'rollback'].includes(command)) throw new Error('命令只能为 init、plan、apply 或 rollback');
|
||
const result = { command, limit: Infinity, configPath: resolve(SCRIPT_DIR, 'config.local.json') };
|
||
for (let i = 0; i < rest.length; i += 2) {
|
||
if (!['--config', '--limit', '--profile', '--run-id'].includes(rest[i]) || !rest[i + 1]) throw new Error('未知参数或缺少参数值');
|
||
if (rest[i] === '--config') { result.configPath = resolve(rest[i + 1]); result.explicitConfig = true; }
|
||
else if (rest[i] === '--profile') result.profile = rest[i + 1];
|
||
else if (rest[i] === '--run-id') result.runId = rest[i + 1];
|
||
else result.limit = Number(rest[i + 1]);
|
||
}
|
||
if (result.limit !== Infinity && (!Number.isSafeInteger(result.limit) || result.limit < 1 || command !== 'apply')) throw new Error('--limit 仅用于 apply，且必须为正整数');
|
||
return result;
|
||
}
|
||
|
||
export async function prepareInput(args, projectRoot = PROJECT_ROOT) {
|
||
let raw = {}, exists = false;
|
||
try { raw = JSON.parse(await readFile(args.configPath, 'utf8')); exists = true; }
|
||
catch (error) { if (error.code !== 'ENOENT' || args.explicitConfig) throw error; }
|
||
const placeholder = raw.mysql?.database === 'YOUR_DATABASE';
|
||
const projectManaged = !raw.mysql || placeholder || raw._autoDiscovery === true;
|
||
let input;
|
||
if (projectManaged) {
|
||
const project = await loadProjectConfig({ projectRoot, profile: args.profile || raw.projectProfile });
|
||
input = { ...raw, projectProfile: project.profile, mysql: project.mysql,
|
||
_projectDownloadBaseUrl: project.downloadBaseUrl, _autoDiscovery: true };
|
||
} else {
|
||
if (args.profile) throw new Error('手动 mysql 配置不能同时使用 --profile；删除 mysql 项即可复用项目配置');
|
||
input = { ...raw, mysql: { ...raw.mysql } };
|
||
}
|
||
if (args.runId) input.runId = args.runId;
|
||
if (input.targetConfigId === null) delete input.targetConfigId;
|
||
if (Array.isArray(input.sourceConfigIds) && input.sourceConfigIds.length === 0) delete input.sourceConfigIds;
|
||
const password = input.mysql.passwordEnv ? process.env[input.mysql.passwordEnv] : input.mysql.password;
|
||
if (password === undefined) throw new Error('数据库密码未提供；自动模式无需填写 mysql，手动模式请设置指定环境变量');
|
||
input.mysql.password = password;
|
||
return { input, exists, projectManaged, raw };
|
||
}
|
||
|
||
export async function persistGeneratedConfig(config, args, prepared) {
|
||
if (!prepared.projectManaged || (prepared.exists && prepared.raw._autoDiscovery !== true)) return;
|
||
const saved = {
|
||
projectProfile: config.projectProfile, runId: config.runId,
|
||
sourceConfigIds: config.sourceConfigIds, targetConfigId: config.targetConfigId,
|
||
sourceMode: config.sourceMode, oldDomains: config.oldDomains,
|
||
_autoDiscovery: true, _databaseIdentity: config._databaseIdentity
|
||
};
|
||
// Persist discovery/batch choices, but read credentials from the project's
|
||
// YAML on each invocation. No database or OSS secret is copied here.
|
||
const defaults = normalizeConfig({ runId: config.runId, mysql: config.mysql, sourceConfigIds: config.sourceConfigIds, targetConfigId: config.targetConfigId, sourceMode: 'database' });
|
||
for (const key of ['targetUrlMode', 'targetPublicDomain', 'objectPrefix', 'referenceSchemas', 'excludeReferenceTables', 'imagesOnly', 'batchSize', 'timeoutMs', 'maxFileBytes', 'retries']) {
|
||
if (config[key] !== undefined && JSON.stringify(config[key]) !== JSON.stringify(defaults[key])) saved[key] = config[key];
|
||
}
|
||
if (config.sourceMode === 'http' || config.targetUrlMode === 'backend') saved.downloadBaseUrl = config.downloadBaseUrl;
|
||
const serialized = JSON.stringify(saved, null, 2) + '\n';
|
||
if (prepared.exists && JSON.stringify(prepared.raw) === JSON.stringify(saved)) return;
|
||
// Never overwrite a pre-existing manual file; generated files may be
|
||
// refreshed to pin an explicitly requested new batch.
|
||
await writeFile(args.configPath, serialized, { mode: 0o600, flag: prepared.exists ? 'w' : 'wx' });
|
||
}
|
||
|
||
const table = (config, name) => qualified(config.mysql.database, name);
|
||
const runs = config => table(config, 'infra_file_oss_migration_runs');
|
||
const items = config => table(config, 'infra_file_oss_migration_items');
|
||
const parseJson = value => typeof value === 'string' ? JSON.parse(value) : value;
|
||
const itemKey = item => sha256(`${item.configId}\0${item.path}`);
|
||
const samePlannedItem = (left, right) => {
|
||
const canonical = item => ({ ...item, aliases: [...item.aliases].sort(), files: [...item.files].sort((a, b) => BigInt(a.id) < BigInt(b.id) ? -1 : BigInt(a.id) > BigInt(b.id) ? 1 : 0) });
|
||
return isDeepStrictEqual(canonical(left), canonical(right));
|
||
};
|
||
|
||
export async function loadSettings(db, config) {
|
||
const ids = [...config.sourceConfigIds, config.targetConfigId];
|
||
const [rows] = await db.execute(`SELECT id, storage, master, config FROM ${table(config, 'infra_file_config')} WHERE deleted=0 AND id IN (${ids.map(() => '?').join(',')})`, ids);
|
||
const byId = new Map(rows.map(row => [String(row.id), row]));
|
||
for (const sourceId of config.sourceConfigIds) {
|
||
const row = byId.get(sourceId);
|
||
if (!row || Number(row.storage) !== 1) throw new Error(`源配置 ${sourceId} 不存在或不是数据库存储`);
|
||
if (bit(row.master)) throw new Error(`源配置 ${sourceId} 仍是默认上传配置，请先切换默认配置到 OSS`);
|
||
}
|
||
const targetRow = byId.get(config.targetConfigId);
|
||
if (!targetRow) throw new Error('目标存储配置不存在');
|
||
const oss = ossSettings(targetRow, config);
|
||
const [identity] = await db.query('SELECT @@server_uuid AS server_uuid');
|
||
const safe = {
|
||
version: 1, server: identity[0].server_uuid, schema: config.mysql.database,
|
||
sourceConfigIds: config.sourceConfigIds, targetConfigId: config.targetConfigId,
|
||
sourceDomains: config.sourceConfigIds.map(sourceId => parseStorage(byId.get(sourceId)).domain || ''),
|
||
sourceMode: config.sourceMode, downloadBaseUrl: config.downloadBaseUrl || '',
|
||
oldDomains: config.oldDomains, targetUrlMode: config.targetUrlMode,
|
||
endpoint: oss.endpoint, bucket: oss.bucket, publicDomain: oss.publicDomain,
|
||
objectPrefix: config.objectPrefix, imagesOnly: config.imagesOnly,
|
||
referenceSchemas: config.referenceSchemas, excludeReferenceTables: config.excludeReferenceTables
|
||
};
|
||
return { byId, oss, safe };
|
||
}
|
||
|
||
const progressOf = config => config._progress || { stage: message => console.log(message), update() {} };
|
||
|
||
function sourceItem(group, config, settings) {
|
||
const type = mimeType(group.path, group.files);
|
||
if (config.imagesOnly && !type.toLowerCase().startsWith('image/')) return null;
|
||
encodePath(group.path);
|
||
const item = { contentId: String(group.contentId), configId: String(group.configId), path: group.path, type, files: group.files };
|
||
const aliasRows = [...group.files, ...group.spellings.map(path => ({ path }))];
|
||
item.aliases = aliases(item, aliasRows, parseStorage(settings.byId.get(item.configId)).domain, config);
|
||
return item;
|
||
}
|
||
|
||
const missingCount = (index, config) => index.missingFiles.filter(file => !config.imagesOnly || mimeType(file.path, [file]).toLowerCase().startsWith('image/')).length;
|
||
|
||
export async function sourceCurrent(db, config, item, lock = false) {
|
||
// Keep large content out of the unindexed path/order lookup. On the actual
|
||
// database the combined lookup + BLOB projection took seconds per image;
|
||
// a metadata lookup followed by a primary-key hash read takes milliseconds.
|
||
// With lock=true the first SELECT retains the original range-lock semantics.
|
||
const [[latest]] = await db.execute(`SELECT id FROM ${table(config, 'infra_file_content')} WHERE deleted=0 AND config_id=? AND path=? ORDER BY id DESC LIMIT 1${lock ? ' FOR UPDATE' : ''}`, [item.configId, item.path]);
|
||
if (!latest || String(latest.id) !== String(item.contentId)) throw new Error('源图片在任务建立后发生变化或已删除，请检查该文件');
|
||
const [[current]] = await db.execute(`SELECT id,OCTET_LENGTH(content) AS size,LOWER(SHA2(content,256)) AS sha256 FROM ${table(config, 'infra_file_content')} WHERE id=? AND config_id=? AND path=? AND deleted=0${lock ? ' FOR UPDATE' : ''}`, [item.contentId, item.configId, item.path]);
|
||
if (!current || String(current.id) !== String(item.contentId) || current.sha256 !== item.sha256 || Number(current.size) !== item.size) throw new Error('源图片在任务建立后发生变化或已删除，请检查该文件');
|
||
}
|
||
|
||
export async function* inventory(db, config, settings, maxId, summary, sourceIndex) {
|
||
const progress = progressOf(config);
|
||
const index = sourceIndex || await loadSourceIndex(db, config, { maxId, progress });
|
||
const candidates = [];
|
||
for (const group of index.groups) {
|
||
summary.sourceGroups++;
|
||
const item = sourceItem(group, config, settings);
|
||
if (item) candidates.push(item); else summary.skippedNonImages++;
|
||
}
|
||
progress.stage(`建立迁移清单：按主键读取并校验 ${candidates.length} 个文件的内容`);
|
||
for (let start = 0; start < candidates.length; start += config.batchSize) {
|
||
const batch = candidates.slice(start, start + config.batchSize);
|
||
progress.update(`建立迁移清单：已校验 ${start}/${candidates.length} 个文件，正在读取下一批`);
|
||
const [rows] = await db.execute(`SELECT id,config_id,path,OCTET_LENGTH(content) AS size,LOWER(SHA2(content,256)) AS sha256 FROM ${table(config, 'infra_file_content')} WHERE deleted=0 AND id IN (${batch.map(() => '?').join(',')})`, batch.map(item => item.contentId));
|
||
const byId = new Map(rows.map(row => [String(row.id), row]));
|
||
for (const item of batch) {
|
||
const row = byId.get(item.contentId);
|
||
if (!row || String(row.config_id) !== item.configId || row.path !== item.path) throw new Error(`内容记录 ${item.contentId} 在建立清单期间发生变化，请检查后重试`);
|
||
item.size = Number(row.size);
|
||
item.sha256 = row.sha256;
|
||
if (!item.size || item.size > config.maxFileBytes) throw new Error(`内容记录 ${item.contentId} 为空或超过 maxFileBytes`);
|
||
item.key = objectKey(config, item);
|
||
item.url = config.targetUrlMode === 'backend' ? downloadUrl(config.downloadBaseUrl, config.targetConfigId, item.key) : `${settings.oss.publicDomain}/${encodePath(item.key)}`;
|
||
if (item.key.length > 512 || item.url.length > 1024) throw new Error('目标路径或访问地址超出 infra_file 字段长度');
|
||
summary.candidates++;
|
||
summary.bytes += item.size;
|
||
if (!item.files.length) summary.withoutFileRecord++;
|
||
yield item;
|
||
}
|
||
}
|
||
progress.stage(`迁移清单已建立：${summary.candidates} 个文件，内容校验完成`);
|
||
}
|
||
|
||
function addMappings(map, item) {
|
||
for (const oldUrl of item.aliases) {
|
||
if (map.has(oldUrl) && map.get(oldUrl) !== item.url) throw new Error('同一旧地址对应多个目标，必须先解决数据冲突');
|
||
map.set(oldUrl, item.url);
|
||
}
|
||
}
|
||
|
||
const emptyInventory = () => ({ sourceGroups: 0, candidates: 0, bytes: 0, skippedNonImages: 0, withoutFileRecord: 0 });
|
||
|
||
async function scanReferences(db, config, map, { apply = false } = {}) {
|
||
const progress = progressOf(config);
|
||
progress.stage('检查需要扫描的业务表和图片地址字段');
|
||
const discovery = await discoverReferenceColumns(db, config.referenceSchemas, config.excludeReferenceTables);
|
||
const groups = new Map();
|
||
for (const column of discovery.columns) {
|
||
const key = JSON.stringify([column.schema, column.table]);
|
||
if (!groups.has(key)) groups.set(key, { ...column, columns: [] });
|
||
groups.get(key).columns.push(column.column);
|
||
}
|
||
const result = { scannedRows: 0, changedRows: 0, replacements: 0, unresolvedRows: 0, conflicts: 0, skippedTables: discovery.skipped.filter(entry => entry.reason !== 'excluded table'), excludedTables: discovery.skipped.filter(entry => entry.reason === 'excluded table'), tables: [] };
|
||
let tableNumber = 0;
|
||
const readBatchSize = Math.max(config.batchSize, 1000);
|
||
for (const group of groups.values()) {
|
||
tableNumber++;
|
||
let scanned = 0;
|
||
progress.stage(`引用扫描 ${tableNumber}/${groups.size}：${group.schema}.${group.table}`);
|
||
const stat = { schema: group.schema, table: group.table, changedRows: 0, unresolvedRows: 0 };
|
||
let cursor;
|
||
for (;;) {
|
||
const columns = [group.primaryKey, ...group.columns];
|
||
const [rows] = await db.query(`SELECT ${columns.map(quote).join(',')} FROM ${qualified(group.schema, group.table)} ${cursor === undefined ? '' : `WHERE ${quote(group.primaryKey)}>?`} ORDER BY ${quote(group.primaryKey)} LIMIT ${readBatchSize}`, cursor === undefined ? [] : [cursor]);
|
||
if (!rows.length) break;
|
||
for (const row of rows) {
|
||
cursor = row[group.primaryKey];
|
||
result.scannedRows++;
|
||
scanned++;
|
||
progress.update(`引用扫描 ${tableNumber}/${groups.size}：${group.schema}.${group.table}，已检查 ${scanned} 行，匹配 ${stat.changedRows} 行`);
|
||
const before = {}, after = {};
|
||
let unresolved = false, replacements = 0;
|
||
for (const column of group.columns) {
|
||
const rewritten = rewriteReferences(row[column], map);
|
||
if (rewritten.changed) { before[column] = row[column]; after[column] = rewritten.value; replacements += rewritten.replacements; }
|
||
if (containsLegacyReference(rewritten.value, config.sourceConfigIds)) unresolved = true;
|
||
}
|
||
if (unresolved) { result.unresolvedRows++; stat.unresolvedRows++; }
|
||
if (!Object.keys(after).length) continue;
|
||
if (apply) {
|
||
await db.beginTransaction();
|
||
try {
|
||
await applyChange(db, config.mysql.database, config.runId, { schema: group.schema, table: group.table, primaryKey: group.primaryKey, pkValue: cursor, before, after });
|
||
await db.commit();
|
||
} catch (error) {
|
||
await db.rollback();
|
||
result.conflicts++;
|
||
console.error(`引用更新未完成 ${group.schema}.${group.table}：${errorText(error)}`);
|
||
continue;
|
||
}
|
||
}
|
||
result.changedRows++; stat.changedRows++; result.replacements += replacements;
|
||
}
|
||
}
|
||
if (stat.changedRows || stat.unresolvedRows) result.tables.push(stat);
|
||
console.log(`引用扫描 ${group.schema}.${group.table}：匹配 ${stat.changedRows} 行，仍含旧地址 ${stat.unresolvedRows} 行`);
|
||
}
|
||
return result;
|
||
}
|
||
|
||
async function setItem(db, config, item, status, error = null) {
|
||
await db.execute(`UPDATE ${items(config)} SET status=?,error=?,updated_at=CURRENT_TIMESTAMP WHERE run_id=? AND item_key=?`, [status, error, config.runId, itemKey(item)]);
|
||
}
|
||
|
||
async function* savedItems(db, config, statuses = null) {
|
||
let cursor = '';
|
||
for (;;) {
|
||
const [rows] = await db.execute(`SELECT item_key,data,status FROM ${items(config)} WHERE run_id=? AND item_key>? ${statuses ? `AND status IN (${statuses.map(() => '?').join(',')})` : ''} ORDER BY item_key LIMIT ${config.batchSize}`, [config.runId, cursor, ...(statuses || [])]);
|
||
if (!rows.length) return;
|
||
for (const row of rows) { cursor = row.item_key; yield { ...parseJson(row.data), status: row.status }; }
|
||
}
|
||
}
|
||
|
||
async function migrateItem(db, config, oss, item) {
|
||
const timings = { sourceMs: 0, ossMs: 0, publicMs: 0, recordMs: 0 };
|
||
const measure = async (key, label, action) => {
|
||
progressOf(config).update(`文件 ${item.contentId}：${label}`);
|
||
const started = performance.now();
|
||
try { return await action(); } finally { timings[key] += performance.now() - started; }
|
||
};
|
||
await measure('sourceMs', '复核源文件版本', () => sourceCurrent(db, config, item));
|
||
let bytes;
|
||
if (config.sourceMode === 'http') {
|
||
bytes = await measure('sourceMs', '读取原图片', () => readHttp(downloadUrl(config.downloadBaseUrl, item.configId, item.path), config, item.size));
|
||
} else {
|
||
const [[row]] = await measure('sourceMs', '读取数据库原图片', () => db.execute(`SELECT content FROM ${table(config, 'infra_file_content')} WHERE id=? AND deleted=0`, [item.contentId]));
|
||
if (!row) throw new Error('源内容已删除');
|
||
bytes = row.content;
|
||
}
|
||
verifyContent(bytes, item);
|
||
await measure('ossMs', '上传 OSS 并读取校验', () => ensureObject(oss, item.key, bytes, item));
|
||
// SDK 可读不代表页面可读：按最终将写入业务表的地址匿名访问并再次校验。
|
||
verifyContent(await measure('publicMs', '校验最终图片访问地址', () => readHttp(item.url, config, item.size)), item);
|
||
await measure('sourceMs', '复核源图片未发生变化', () => sourceCurrent(db, config, item));
|
||
await measure('recordMs', '保存迁移进度', () => setItem(db, config, item, 'verified'));
|
||
return timings;
|
||
}
|
||
|
||
export async function verifyCheckpoint(oss, config, item) {
|
||
// A verified checkpoint needs no source download or PUT. Still check the
|
||
// authenticated object and its final public URL before trusting it again.
|
||
const head = await oss.head(item.key);
|
||
if (Number(head.res.headers['content-length']) !== item.size || item.size > config.maxFileBytes) {
|
||
throw new Error('OSS 文件长度不一致或超过限制');
|
||
}
|
||
verifyContent((await oss.get(item.key)).content, item);
|
||
verifyContent(await readHttp(item.url, config, item.size), item);
|
||
}
|
||
|
||
export async function verifySwitchedBatch(db, config, batch) {
|
||
// A resumed run must not reuse a mapping whose object/metadata was deleted
|
||
// or changed after the earlier successful batch. Read metadata once; only
|
||
// HTTP validation is concurrent, never transactions on the shared connection.
|
||
const ids = [...new Set(batch.flatMap(item => item.files.map(file => String(file.id))))];
|
||
const byId = new Map();
|
||
for (let start = 0; start < ids.length; start += 1000) {
|
||
const chunk = ids.slice(start, start + 1000);
|
||
const [rows] = await db.execute(`SELECT id,config_id,path,url,deleted FROM ${table(config, 'infra_file')} WHERE id IN (${chunk.map(() => '?').join(',')})`, chunk);
|
||
for (const row of rows) byId.set(String(row.id), row);
|
||
}
|
||
return mapBounded(batch, VALIDATION_CONCURRENCY, async item => {
|
||
try {
|
||
for (const original of item.files) {
|
||
const current = byId.get(String(original.id));
|
||
if (!current || bit(current.deleted) || String(current.config_id) !== config.targetConfigId || current.path !== item.key || current.url !== item.url) {
|
||
throw new Error('已迁移文件记录后来发生变化或已删除，停止复用该地址映射');
|
||
}
|
||
}
|
||
await retry(async () => verifyContent(await readHttp(item.url, config, item.size), item), config.retries);
|
||
return { item };
|
||
} catch (error) {
|
||
return { item, error };
|
||
}
|
||
});
|
||
}
|
||
|
||
async function* itemBatches(iterable, size) {
|
||
let batch = [];
|
||
for await (const item of iterable) {
|
||
batch.push(item);
|
||
if (batch.length === size) { yield batch; batch = []; }
|
||
}
|
||
if (batch.length) yield batch;
|
||
}
|
||
|
||
export async function runPlan(db, config, settings) {
|
||
const progress = progressOf(config);
|
||
progress.stage('开始只读预检：只读取文件信息，不读取图片内容或计算 SHA-256');
|
||
const index = await loadSourceIndex(db, config, { progress });
|
||
const summary = { command: 'plan', ...emptyInventory(), bytes: null, estimatedBytes: 0, sizePending: 0, contentVerified: false, missingContent: missingCount(index, config) };
|
||
const mappings = new Map();
|
||
for (const group of index.groups) {
|
||
summary.sourceGroups++;
|
||
const item = sourceItem(group, config, settings);
|
||
if (!item) { summary.skippedNonImages++; continue; }
|
||
summary.candidates++;
|
||
const sizes = item.files.map(file => Number(file.size)).filter(size => Number.isSafeInteger(size) && size >= 0);
|
||
if (sizes.length) summary.estimatedBytes += Math.max(...sizes); else summary.sizePending++;
|
||
if (!item.files.length) summary.withoutFileRecord++;
|
||
// Preview markers exist only in memory. Actual object keys require the
|
||
// verified content hash and are created exclusively during apply.
|
||
item.url = `migration-preview://${item.configId}/${item.contentId}`;
|
||
addMappings(mappings, item);
|
||
}
|
||
progress.stage(`已识别 ${summary.candidates} 个图片文件；开始检查业务引用（实际内容在迁移时校验）`);
|
||
summary.references = await scanReferences(db, config, mappings);
|
||
summary.status = summary.missingContent || summary.references.unresolvedRows || summary.references.skippedTables.length ? 'needs-review' : 'planned';
|
||
progress.stage('只读预检完成');
|
||
console.log('预检未读取图片二进制内容、未计算 SHA-256；容量仅为文件表估算。未验证 HTTP/OSS 连通性，也未创建迁移表、上传或修改数据。');
|
||
return summary;
|
||
}
|
||
|
||
export async function runApply(db, config, settings, limit = Infinity, createOss = null) {
|
||
const progress = progressOf(config);
|
||
progress.stage('读取迁移批次和断点记录');
|
||
// Batch source checks depend on InnoDB next-key locks and a snapshot first
|
||
// established AFTER their range lock, even if the server defaults to RC.
|
||
await db.query('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ');
|
||
const [engines] = await db.execute(`SELECT TABLE_NAME AS name,ENGINE AS engine FROM information_schema.TABLES WHERE TABLE_SCHEMA=? AND TABLE_NAME IN ('infra_file_content','infra_file')`, [config.mysql.database]);
|
||
if (engines.length !== 2 || engines.some(row => String(row.engine).toLowerCase() !== 'innodb')) {
|
||
throw new Error('infra_file_content 和 infra_file 必须为 InnoDB，才能保证源校验与记录更新的事务保护');
|
||
}
|
||
await ensureJournal(db, config.mysql.database);
|
||
const [[saved]] = await db.execute(`SELECT settings,status FROM ${runs(config)} WHERE run_id=?`, [config.runId]);
|
||
let runSettings, sourceIndex;
|
||
if (saved) {
|
||
runSettings = parseJson(saved.settings);
|
||
if (JSON.stringify(runSettings.binding) !== JSON.stringify(settings.safe)) throw new Error('同一 runId 的数据库、源/目标或引用范围发生变化，请恢复原配置或使用新 runId');
|
||
if (saved.status.startsWith('rollback') || saved.status === 'rolled_back') throw new Error('本批次已开始回滚，不能继续迁移，请继续回滚或使用新 runId');
|
||
} else {
|
||
sourceIndex = await loadSourceIndex(db, config, { progress });
|
||
const missing = missingCount(sourceIndex, config);
|
||
if (missing) throw new Error(`发现 ${missing} 条图片记录没有数据库内容，请先执行 plan 并处理`);
|
||
runSettings = { binding: settings.safe, maxId: sourceIndex.maxId };
|
||
await db.execute(`INSERT INTO ${runs(config)} (run_id,settings,status) VALUES (?,?,'inventory')`, [config.runId, JSON.stringify(runSettings)]);
|
||
}
|
||
const summary = { command: 'apply', ...emptyInventory(), ...runSettings.inventory, uploaded: 0, revalidated: 0, switched: 0, failed: 0, limited: false };
|
||
if (!saved || saved.status === 'inventory') {
|
||
for await (const item of inventory(db, config, settings, runSettings.maxId, summary, sourceIndex)) {
|
||
const [[existing]] = await db.execute(`SELECT data FROM ${items(config)} WHERE run_id=? AND item_key=?`, [config.runId, itemKey(item)]);
|
||
if (existing) {
|
||
if (!samePlannedItem(parseJson(existing.data), item)) throw new Error('建立清单期间源数据发生变化，请检查后建立新批次');
|
||
} else {
|
||
await db.execute(`INSERT INTO ${items(config)} (run_id,item_key,data,status) VALUES (?,?,?,'planned')`, [config.runId, itemKey(item), JSON.stringify(item)]);
|
||
}
|
||
}
|
||
runSettings.inventory = Object.fromEntries(Object.keys(emptyInventory()).map(key => [key, summary[key]]));
|
||
await db.execute(`UPDATE ${runs(config)} SET settings=?,status='active',updated_at=CURRENT_TIMESTAMP WHERE run_id=?`, [JSON.stringify(runSettings), config.runId]);
|
||
}
|
||
// 先验证所有映射没有歧义，再发生 OSS 写入。
|
||
const allMappings = new Map();
|
||
let initialSwitched = 0, totalItems = 0;
|
||
progress.stage('检查迁移地址映射是否存在冲突');
|
||
for await (const item of savedItems(db, config)) {
|
||
addMappings(allMappings, item);
|
||
totalItems++;
|
||
if (item.status === 'switched') initialSwitched++;
|
||
}
|
||
const { publicDomain, ...ossOptions } = settings.oss;
|
||
const oss = createOss ? createOss() : new (await import('ali-oss')).default({ ...ossOptions, timeout: config.timeoutMs });
|
||
let processed = 0;
|
||
const selected = [], checkpoints = [];
|
||
progress.stage('开始上传并校验 OSS 文件');
|
||
for await (const item of savedItems(db, config, ['planned', 'failed', 'verified'])) {
|
||
if (processed >= limit) { summary.limited = true; break; }
|
||
processed++;
|
||
if (item.status === 'verified') { checkpoints.push(item); continue; }
|
||
try {
|
||
const started = performance.now();
|
||
const timings = await retry(() => migrateItem(db, config, oss, item), config.retries);
|
||
summary.uploaded++;
|
||
selected.push(item);
|
||
console.log(`上传校验通过：内容记录 ${item.contentId}（本次 ${processed}，总计 ${seconds(performance.now() - started)} 秒；源读取/复核 ${seconds(timings.sourceMs)} 秒，OSS 上传/校验 ${seconds(timings.ossMs)} 秒，访问校验 ${seconds(timings.publicMs)} 秒）`);
|
||
} catch (error) {
|
||
await setItem(db, config, item, 'failed', errorText(error));
|
||
summary.failed++;
|
||
console.error(`上传未完成：内容记录 ${item.contentId}，${errorText(error)}`);
|
||
}
|
||
}
|
||
if (checkpoints.length) {
|
||
let checked = 0;
|
||
progress.stage(`复核已上传断点：0/${checkpoints.length}，最多 ${VALIDATION_CONCURRENCY} 张同时校验`);
|
||
const results = await mapBounded(checkpoints, VALIDATION_CONCURRENCY, async item => {
|
||
try {
|
||
await retry(() => verifyCheckpoint(oss, config, item), config.retries);
|
||
return { item };
|
||
} catch (error) {
|
||
return { item, error };
|
||
} finally {
|
||
progress.update(`复核已上传断点：${++checked}/${checkpoints.length}`);
|
||
}
|
||
});
|
||
// All network workers have settled before using the database again.
|
||
for (const { item, error } of results) {
|
||
if (error) {
|
||
await setItem(db, config, item, 'failed', errorText(error));
|
||
summary.failed++;
|
||
console.error(`已上传断点校验失败：内容记录 ${item.contentId}，${errorText(error)}`);
|
||
} else { selected.push(item); summary.revalidated++; }
|
||
}
|
||
}
|
||
if (summary.failed) {
|
||
summary.status = 'partial';
|
||
const [counts] = await db.execute(`SELECT status,COUNT(*) AS count FROM ${items(config)} WHERE run_id=? GROUP BY status`, [config.runId]);
|
||
summary.items = Object.fromEntries(counts.map(row => [row.status, Number(row.count)]));
|
||
await db.execute(`UPDATE ${runs(config)} SET status='partial',updated_at=CURRENT_TIMESTAMP WHERE run_id=?`, [config.runId]);
|
||
return summary;
|
||
}
|
||
const switchProgress = () => `更新已校验文件记录：本次完成 ${summary.switched}/${selected.length}，累计 ${initialSwitched + summary.switched}/${totalItems}`;
|
||
progress.stage(switchProgress());
|
||
for (let start = 0; start < selected.length; start += TRANSITION_BATCH_SIZE) {
|
||
const batch = selected.slice(start, start + TRANSITION_BATCH_SIZE);
|
||
const started = performance.now();
|
||
try {
|
||
const timing = await switchFiles(db, config, batch);
|
||
summary.switched += batch.length;
|
||
console.log(`文件记录批次完成：${batch.length} 张，${seconds(performance.now() - started)} 秒（源校验 ${seconds(timing.sourceMs)} 秒，文件及回滚日志 ${seconds(timing.writeMs)} 秒，进度 ${seconds(timing.statusMs)} 秒，提交 ${seconds(timing.commitMs)} 秒）；累计 ${initialSwitched + summary.switched}/${totalItems}`);
|
||
} catch (error) {
|
||
// Isolate known data conflicts after the whole batch rolled back. Never
|
||
// replay a transaction with an ambiguous commit or broken connection.
|
||
if (error.fatal || !['MIGRATION_CAS_CONFLICT', 'MIGRATION_SOURCE_CHANGED'].includes(error.code)) throw error;
|
||
console.error(`该批文件存在修改冲突，事务已回滚；逐项定位：${errorText(error)}`);
|
||
for (const item of batch) {
|
||
try { await switchFiles(db, config, [item]); summary.switched++; }
|
||
catch (itemError) {
|
||
if (itemError.fatal || !['MIGRATION_CAS_CONFLICT', 'MIGRATION_SOURCE_CHANGED'].includes(itemError.code)) throw itemError;
|
||
summary.failed++;
|
||
console.error(`文件记录未切换：内容记录 ${item.contentId}，${errorText(itemError)}`);
|
||
}
|
||
}
|
||
}
|
||
progress.update(switchProgress());
|
||
}
|
||
const mappings = new Map();
|
||
let checkedSwitched = 0;
|
||
progress.stage(`复核已迁移文件的访问地址：0/${initialSwitched + summary.switched}`);
|
||
for await (const batch of itemBatches(savedItems(db, config, ['switched']), TRANSITION_BATCH_SIZE)) {
|
||
for (const { item, error } of await verifySwitchedBatch(db, config, batch)) {
|
||
if (error) {
|
||
summary.failed++;
|
||
console.error(`目标地址不可复用：内容记录 ${item.contentId}，${errorText(error)}`);
|
||
} else addMappings(mappings, item);
|
||
}
|
||
checkedSwitched += batch.length;
|
||
progress.update(`复核已迁移文件的访问地址：${checkedSwitched}/${initialSwitched + summary.switched}`);
|
||
}
|
||
summary.references = await scanReferences(db, config, mappings, { apply: true });
|
||
const [counts] = await db.execute(`SELECT status,COUNT(*) AS count FROM ${items(config)} WHERE run_id=? GROUP BY status`, [config.runId]);
|
||
summary.items = Object.fromEntries(counts.map(row => [row.status, Number(row.count)]));
|
||
const unfinished = counts.some(row => row.status !== 'switched');
|
||
summary.status = unfinished || summary.failed || summary.references.conflicts ? 'partial'
|
||
: summary.references.unresolvedRows || summary.references.skippedTables.length ? 'needs-review' : 'complete';
|
||
await db.execute(`UPDATE ${runs(config)} SET status=?,updated_at=CURRENT_TIMESTAMP WHERE run_id=?`, [summary.status, config.runId]);
|
||
return summary;
|
||
}
|
||
|
||
export async function runRollback(db, config) {
|
||
const [[run]] = await db.execute(`SELECT settings,status FROM ${runs(config)} WHERE run_id=?`, [config.runId]);
|
||
if (!run) throw new Error('找不到此 runId 的迁移日志');
|
||
const [[identity]] = await db.query('SELECT @@server_uuid AS server_uuid');
|
||
const binding = parseJson(run.settings).binding;
|
||
if (binding.schema !== config.mysql.database || binding.server !== identity.server_uuid) throw new Error('回滚数据库与迁移时不一致');
|
||
await db.execute(`UPDATE ${runs(config)} SET status='rollback',updated_at=CURRENT_TIMESTAMP WHERE run_id=?`, [config.runId]);
|
||
const result = await rollbackChanges(db, config.mysql.database, config.runId);
|
||
const status = result.conflicts.length ? 'rollback_partial' : 'rolled_back';
|
||
await db.execute(`UPDATE ${runs(config)} SET status=?,updated_at=CURRENT_TIMESTAMP WHERE run_id=?`, [status, config.runId]);
|
||
return { command: 'rollback', status, ...result };
|
||
}
|
||
|
||
export async function main(argv = process.argv.slice(2)) {
|
||
const args = parseArgs(argv);
|
||
if (args.help) { console.log(HELP); return; }
|
||
const prepared = await prepareInput(args);
|
||
const { default: mysql } = await import('mysql2/promise');
|
||
const { passwordEnv, ...mysqlOptions } = prepared.input.mysql;
|
||
console.log(`读取${prepared.input.projectProfile ? `项目 ${prepared.input.projectProfile} 环境` : '指定配置'}：MySQL ${mysqlOptions.host || 'localhost'}:${mysqlOptions.port || 3306}/${mysqlOptions.database}`);
|
||
const db = await mysql.createConnection({ connectTimeout: 10000, ...mysqlOptions, multipleStatements: false, supportBigNumbers: true, bigNumberStrings: true, jsonStrings: true, dateStrings: true });
|
||
const progress = createProgress();
|
||
let locked = false;
|
||
const lockName = `oss-file-migration:${sha256(mysqlOptions.database).slice(0, 40)}`;
|
||
try {
|
||
progress.stage('识别旧存储、默认 OSS 和迁移批次');
|
||
await db.query("SET SESSION sql_mode = CONCAT_WS(',', NULLIF(@@SESSION.sql_mode,''), 'STRICT_ALL_TABLES')");
|
||
if (['apply', 'rollback'].includes(args.command)) {
|
||
const [[lock]] = await db.execute('SELECT GET_LOCK(?,0) AS acquired', [lockName]);
|
||
if (Number(lock.acquired) !== 1) throw new Error('已有文件迁移或回滚任务运行，请等待其结束');
|
||
locked = true;
|
||
}
|
||
const auto = prepared.projectManaged || !prepared.input.runId || !prepared.input.sourceConfigIds?.length || !prepared.input.targetConfigId;
|
||
const config = normalizeConfig(auto ? await discoverStorageConfig(db, prepared.input, { command: args.command }) : prepared.input);
|
||
config._progress = progress;
|
||
await db.query(`SET SESSION max_execution_time=${config.timeoutMs}`);
|
||
await persistGeneratedConfig(config, args, prepared);
|
||
if (args.command === 'init') {
|
||
await loadSettings(db, config);
|
||
progress.close();
|
||
console.log(`配置已就绪：旧数据库存储 ${config.sourceConfigIds.join(',')} → 默认 OSS ${config.targetConfigId}；批次 ${config.runId}`);
|
||
console.log(`配置文件：${args.configPath}；数据库密码和 OSS 密钥未复制到此文件。`);
|
||
console.log('只读取了项目和数据库配置，未上传图片或修改数据库。');
|
||
return { status: 'configured', configPath: args.configPath };
|
||
}
|
||
let summary;
|
||
if (args.command === 'rollback') summary = await runRollback(db, config);
|
||
else {
|
||
const settings = await loadSettings(db, config);
|
||
console.log(`源配置 ${config.sourceConfigIds.join(',')} → OSS 配置 ${config.targetConfigId}；批次 ${config.runId}`);
|
||
summary = args.command === 'plan' ? await runPlan(db, config, settings) : await runApply(db, config, settings, args.limit);
|
||
}
|
||
const reportDir = resolve(dirname(args.configPath), 'reports');
|
||
await mkdir(reportDir, { recursive: true, mode: 0o700 });
|
||
const reportPath = resolve(reportDir, `${config.runId}-${args.command}-${Date.now()}.json`);
|
||
await writeFile(reportPath, JSON.stringify(summary, null, 2), { mode: 0o600 });
|
||
progress.close();
|
||
console.log(JSON.stringify(summary, null, 2));
|
||
console.log(`报告：${reportPath}`);
|
||
if (['partial', 'needs-review', 'rollback_partial'].includes(summary.status)) process.exitCode = 2;
|
||
return summary;
|
||
} finally {
|
||
progress.close();
|
||
try { if (locked) await db.execute('SELECT RELEASE_LOCK(?)', [lockName]); }
|
||
finally { await db.end(); }
|
||
}
|
||
}
|
||
|
||
if (process.argv[1] && resolve(process.argv[1]) === fileURLToPath(import.meta.url)) {
|
||
main().catch(error => { console.error(`执行停止：${errorText(error)}`); process.exitCode = 1; });
|
||
}
|