order_site/apps/backend/src/storage-migrate.ts

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
}