准备迁移1

2026-08-23 12:54:38 +08:00
parent 360c5ea20e
commit 6d7f9b5f3f
16 changed files with 999 additions and 339 deletions
@@ -0,0 +1,176 @@
+import crypto from 'node:crypto'
+import process from 'node:process'
+
+import {
+  createConfiguredObjectStorage,
+  type ObjectStorage,
+} from './services/file-storage/object-storage.js'
+
+const command = process.argv[2] || ''
+const concurrency = readConcurrency()
+if (!['copy', 'verify'].includes(command)) {
+  console.error('用法：node dist/storage-migrate.js copy|verify')
+  process.exit(2)
+}
+
+const source = createConfiguredObjectStorage(readStorageConfig('STORAGE'))
+const destination = createConfiguredObjectStorage(readStorageConfig('STORAGE_OSS'))
+
+if (command === 'copy') {
+  await copyObjects(source, destination)
+} else {
+  await verifyObjects(source, destination)
+}
+
+function readStorageConfig(prefix: 'STORAGE' | 'STORAGE_OSS') {
+  return {
+    endpoint: required(`${prefix}_ENDPOINT`),
+    bucket: required(`${prefix}_BUCKET`),
+    accessKeyId: required(`${prefix}_ACCESS_KEY_ID`),
+    secretAccessKey: required(`${prefix}_SECRET_ACCESS_KEY`),
+    region: process.env[`${prefix}_REGION`] || 'us-east-1',
+  }
+}
+
+async function copyObjects(sourceStorage: ObjectStorage, destinationStorage: ObjectStorage) {
+  const objects = await sourceStorage.listObjects()
+  console.info(`源存储对象数：${objects.length}`)
+  let copied = 0
+  let skipped = 0
+  let completed = 0
+  await runConcurrent(objects, concurrency, async (object) => {
+    const existing = await readStat(destinationStorage, object.key)
+    const sourceEtag = normalizeEtag(object.etag)
+    const destinationEtag = normalizeEtag(existing?.etag)
+    if (
+      existing &&
+      Number(existing.size || 0) === object.size &&
+      sourceEtag &&
+      destinationEtag &&
+      destinationEtag === sourceEtag
+    ) {
+      skipped += 1
+    } else {
+      const stored = await sourceStorage.getObject(object.key)
+      const content = await readBuffer(stored.reader)
+      await destinationStorage.putObject({
+        key: object.key,
+        content,
+        contentType: stored.contentType,
+      })
+      copied += 1
+    }
+    completed += 1
+    logProgress('复制', completed, objects.length)
+  })
+  console.info(`复制完成：新增/覆盖 ${copied}，已存在且大小、ETag 一致 ${skipped}`)
+}
+
+async function verifyObjects(sourceStorage: ObjectStorage, destinationStorage: ObjectStorage) {
+  const sourceObjects = await sourceStorage.listObjects()
+  const destinationObjects = await destinationStorage.listObjects()
+  const destinationByKey = new Map(destinationObjects.map((item) => [item.key, item]))
+  const missing = sourceObjects.filter((item) => !destinationByKey.has(item.key))
+  const mismatches: string[] = []
+  let completed = 0
+  await runConcurrent(sourceObjects, concurrency, async (object) => {
+    const destination = destinationByKey.get(object.key)
+    if (!destination) {
+      completed += 1
+      logProgress('校验', completed, sourceObjects.length)
+      return
+    }
+    if (Number(destination.size || 0) !== object.size) {
+      mismatches.push(object.key)
+      completed += 1
+      logProgress('校验', completed, sourceObjects.length)
+      return
+    }
+    const sourceStored = await sourceStorage.getObject(object.key)
+    const destinationStored = await destinationStorage.getObject(object.key)
+    const [sourceHash, destinationHash] = await Promise.all([
+      hashStream(sourceStored.reader),
+      hashStream(destinationStored.reader),
+    ])
+    if (
+      sourceHash !== destinationHash ||
+      sourceStored.contentType !== destinationStored.contentType ||
+      sourceStored.size !== destinationStored.size
+    ) {
+      mismatches.push(object.key)
+    }
+    completed += 1
+    logProgress('校验', completed, sourceObjects.length)
+  })
+  if (missing.length || mismatches.length) {
+    console.error(`校验失败：缺失 ${missing.length}，内容不一致 ${mismatches.length}`)
+    if (missing.length) console.error(`缺失对象：${missing.slice(0, 20).join(', ')}`)
+    if (mismatches.length) console.error(`不一致对象：${mismatches.slice(0, 20).join(', ')}`)
+    process.exit(1)
+  }
+  console.info(`校验通过：${sourceObjects.length} 个源对象全部存在且内容一致`)
+}
+
+async function runConcurrent<T>(
+  items: T[],
+  limit: number,
+  worker: (item: T) => Promise<void>,
+): Promise<void> {
+  let nextIndex = 0
+  const workers = Array.from({ length: Math.min(limit, items.length) }, async () => {
+    while (true) {
+      const index = nextIndex++
+      const item = items[index]
+      if (item === undefined) return
+      await worker(item)
+    }
+  })
+  await Promise.all(workers)
+}
+
+function normalizeEtag(value: unknown): string {
+  return String(value || '')
+    .trim()
+    .replace(/^"|"$/g, '')
+    .toLowerCase()
+}
+
+function logProgress(label: string, completed: number, total: number): void {
+  if (completed === total || completed % 100 === 0) {
+    console.info(`${label}进度：${completed}/${total}（并发 ${concurrency}）`)
+  }
+}
+
+async function readStat(storage: ObjectStorage, key: string) {
+  try {
+    return await storage.statObject(key)
+  } catch {
+    return null
+  }
+}
+
+async function readBuffer(reader: NodeJS.ReadableStream): Promise<Buffer> {
+  const chunks: Buffer[] = []
+  for await (const chunk of reader) chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
+  return Buffer.concat(chunks)
+}
+
+async function hashStream(reader: NodeJS.ReadableStream): Promise<string> {
+  const hash = crypto.createHash('sha256')
+  for await (const chunk of reader) hash.update(chunk)
+  return hash.digest('hex')
+}
+
+function required(name: string): string {
+  const value = String(process.env[name] || '').trim()
+  if (!value) {
+    console.error(`缺少环境变量：${name}`)
+    process.exit(2)
+  }
+  return value
+}
+
+function readConcurrency(): number {
+  const value = Number(process.env.STORAGE_MIGRATION_CONCURRENCY || 8)
+  return Number.isInteger(value) && value > 0 ? Math.min(value, 32) : 8
+}