7326e273e1
Protect upload cleanup and atomic completion paths, add durable media worker leases and brand asset validation, and introduce verified migration/bootstrap/deployment gates.
340 lines
9.9 KiB
JavaScript
340 lines
9.9 KiB
JavaScript
#!/usr/bin/env node
|
|
|
|
// LEGACY_READ_ONLY: 唯一用途是将显式旧来源迁移到当前存储桶，默认 dry-run。
|
|
|
|
const { PrismaClient } = require('@prisma/client')
|
|
const { S3Client, HeadObjectCommand, PutObjectCommand } = require('@aws-sdk/client-s3')
|
|
|
|
const prisma = new PrismaClient()
|
|
|
|
const DEFAULT_OLD_PREFIX = 'https://s3.1080p.hk/jsxiangche/'
|
|
|
|
function getArg(name, fallback = '') {
|
|
const prefix = `--${name}=`
|
|
const match = process.argv.slice(2).find((arg) => arg.startsWith(prefix))
|
|
return match ? match.slice(prefix.length) : fallback
|
|
}
|
|
|
|
function hasFlag(name) {
|
|
return process.argv.includes(`--${name}`)
|
|
}
|
|
|
|
function normalizePrefix(prefix) {
|
|
return String(prefix || '').trim().replace(/\/+$/, '') + '/'
|
|
}
|
|
|
|
function normalizePath(input) {
|
|
return String(input || '')
|
|
.trim()
|
|
.replace(/\\/g, '/')
|
|
.replace(/^\/+/, '')
|
|
.replace(/\/{2,}/g, '/')
|
|
}
|
|
|
|
function parseOldObjectKey(oldUrl, oldBucket) {
|
|
const parsed = new URL(oldUrl)
|
|
let key = decodeURIComponent(parsed.pathname).replace(/^\/+/, '')
|
|
if (oldBucket && key.startsWith(`${oldBucket}/`)) {
|
|
key = key.slice(oldBucket.length + 1)
|
|
}
|
|
return normalizePath(key)
|
|
}
|
|
|
|
function ensureDestinationKey(oldKey) {
|
|
const s3FilePath = normalizePath(process.env.S3_FILE_PATH || '')
|
|
if (!s3FilePath) {
|
|
return oldKey
|
|
}
|
|
|
|
if (oldKey.startsWith(`${s3FilePath}/`)) {
|
|
return oldKey
|
|
}
|
|
|
|
if (oldKey.startsWith('public-display/')) {
|
|
return `${s3FilePath}/${oldKey}`
|
|
}
|
|
|
|
return oldKey
|
|
}
|
|
|
|
function buildPublicUrl(key) {
|
|
const publicDomain = String(process.env.S3_PUBLIC_DOMAIN || '').trim().replace(/\/+$/, '')
|
|
if (!publicDomain) {
|
|
throw new Error('Missing S3_PUBLIC_DOMAIN')
|
|
}
|
|
return `${publicDomain}/${key}`
|
|
}
|
|
|
|
function guessContentType(url) {
|
|
const lower = url.toLowerCase()
|
|
if (lower.endsWith('.webp')) return 'image/webp'
|
|
if (lower.endsWith('.png')) return 'image/png'
|
|
if (lower.endsWith('.jpg') || lower.endsWith('.jpeg')) return 'image/jpeg'
|
|
return 'application/octet-stream'
|
|
}
|
|
|
|
function createS3Client() {
|
|
const endpoint = String(process.env.S3_ENDPOINT || '').split('#')[0].trim()
|
|
if (!endpoint) throw new Error('Missing S3_ENDPOINT')
|
|
if (!process.env.S3_BUCKET) throw new Error('Missing S3_BUCKET')
|
|
if (!process.env.S3_ACCESS_KEY_ID) throw new Error('Missing S3_ACCESS_KEY_ID')
|
|
if (!process.env.S3_SECRET_ACCESS_KEY) throw new Error('Missing S3_SECRET_ACCESS_KEY')
|
|
|
|
return new S3Client({
|
|
endpoint,
|
|
region: process.env.S3_REGION || 'us-east-1',
|
|
credentials: {
|
|
accessKeyId: process.env.S3_ACCESS_KEY_ID,
|
|
secretAccessKey: process.env.S3_SECRET_ACCESS_KEY,
|
|
},
|
|
forcePathStyle: process.env.S3_ENABLE_PATH_STYLE !== '0',
|
|
maxAttempts: 5,
|
|
})
|
|
}
|
|
|
|
async function destinationExists(client, bucket, key) {
|
|
try {
|
|
await client.send(new HeadObjectCommand({ Bucket: bucket, Key: key }))
|
|
return true
|
|
} catch (error) {
|
|
const statusCode = error && error.$metadata && error.$metadata.httpStatusCode
|
|
if (statusCode === 404 || error.name === 'NotFound') {
|
|
return false
|
|
}
|
|
throw error
|
|
}
|
|
}
|
|
|
|
async function fetchOldObject(oldUrl) {
|
|
const response = await fetch(oldUrl)
|
|
if (!response.ok) {
|
|
throw new Error(`Fetch failed ${response.status}`)
|
|
}
|
|
|
|
return {
|
|
body: Buffer.from(await response.arrayBuffer()),
|
|
contentType: response.headers.get('content-type') || guessContentType(oldUrl),
|
|
}
|
|
}
|
|
|
|
async function getOldPreviewUrls(oldPrefix, limit) {
|
|
const limitedSql = limit > 0 ? `LIMIT ${Number(limit)}` : ''
|
|
return prisma.$queryRawUnsafe(`
|
|
WITH history_materials AS (
|
|
SELECT DISTINCT "materialId" AS id
|
|
FROM material_logs
|
|
WHERE action = 'HISTORY_IMPORT'
|
|
), active_history AS (
|
|
SELECT m.id, m."previewUrl"
|
|
FROM materials m
|
|
JOIN history_materials h ON h.id = m.id
|
|
WHERE m."deletedAt" IS NULL
|
|
)
|
|
SELECT DISTINCT old_url
|
|
FROM (
|
|
SELECT "previewUrl" AS old_url
|
|
FROM active_history
|
|
WHERE "previewUrl" LIKE $1
|
|
UNION
|
|
SELECT ma.file_url AS old_url
|
|
FROM material_assets ma
|
|
JOIN active_history h ON h.id = ma.material_id
|
|
WHERE ma.asset_type IN ('main-preview', 'sku-preview', 'detail-preview')
|
|
AND ma.file_url LIKE $1
|
|
) urls
|
|
WHERE old_url IS NOT NULL AND old_url <> ''
|
|
ORDER BY old_url
|
|
${limitedSql}
|
|
`, `${oldPrefix}%`)
|
|
}
|
|
|
|
async function updateDatabaseUrl(oldUrl, newUrl) {
|
|
const materialUpdates = await prisma.$executeRawUnsafe(`
|
|
WITH history_materials AS (
|
|
SELECT DISTINCT "materialId" AS id
|
|
FROM material_logs
|
|
WHERE action = 'HISTORY_IMPORT'
|
|
)
|
|
UPDATE materials m
|
|
SET "previewUrl" = $1,
|
|
"updatedAt" = NOW()
|
|
FROM history_materials h
|
|
WHERE h.id = m.id
|
|
AND m."deletedAt" IS NULL
|
|
AND m."previewUrl" = $2
|
|
`, newUrl, oldUrl)
|
|
|
|
const assetUpdates = await prisma.$executeRawUnsafe(`
|
|
WITH history_materials AS (
|
|
SELECT DISTINCT "materialId" AS id
|
|
FROM material_logs
|
|
WHERE action = 'HISTORY_IMPORT'
|
|
)
|
|
UPDATE material_assets ma
|
|
SET file_url = $1,
|
|
updated_at = NOW()
|
|
FROM history_materials h
|
|
JOIN materials m ON m.id = h.id
|
|
WHERE ma.material_id = m.id
|
|
AND m."deletedAt" IS NULL
|
|
AND ma.asset_type IN ('main-preview', 'sku-preview', 'detail-preview')
|
|
AND ma.file_url = $2
|
|
`, newUrl, oldUrl)
|
|
|
|
return { materialUpdates, assetUpdates }
|
|
}
|
|
|
|
async function bulkUpdateDatabaseUrls(oldPrefix, newPrefix) {
|
|
const materialUpdates = await prisma.$executeRawUnsafe(`
|
|
WITH history_materials AS (
|
|
SELECT DISTINCT "materialId" AS id
|
|
FROM material_logs
|
|
WHERE action = 'HISTORY_IMPORT'
|
|
)
|
|
UPDATE materials m
|
|
SET "previewUrl" = REPLACE(m."previewUrl", $1, $2),
|
|
"updatedAt" = NOW()
|
|
FROM history_materials h
|
|
WHERE h.id = m.id
|
|
AND m."deletedAt" IS NULL
|
|
AND m."previewUrl" LIKE $3
|
|
`, oldPrefix, newPrefix, `${oldPrefix}%`)
|
|
|
|
const assetUpdates = await prisma.$executeRawUnsafe(`
|
|
WITH history_materials AS (
|
|
SELECT DISTINCT "materialId" AS id
|
|
FROM material_logs
|
|
WHERE action = 'HISTORY_IMPORT'
|
|
)
|
|
UPDATE material_assets ma
|
|
SET file_url = REPLACE(ma.file_url, $1, $2),
|
|
updated_at = NOW()
|
|
FROM history_materials h
|
|
JOIN materials m ON m.id = h.id
|
|
WHERE ma.material_id = m.id
|
|
AND m."deletedAt" IS NULL
|
|
AND ma.asset_type IN ('main-preview', 'sku-preview', 'detail-preview')
|
|
AND ma.file_url LIKE $3
|
|
`, oldPrefix, newPrefix, `${oldPrefix}%`)
|
|
|
|
return { materialUpdates, assetUpdates }
|
|
}
|
|
|
|
async function runPool(items, concurrency, worker) {
|
|
let index = 0
|
|
const workers = Array.from({ length: concurrency }, async () => {
|
|
while (index < items.length) {
|
|
const currentIndex = index
|
|
index += 1
|
|
await worker(items[currentIndex], currentIndex)
|
|
}
|
|
})
|
|
await Promise.all(workers)
|
|
}
|
|
|
|
async function main() {
|
|
const apply = hasFlag('apply')
|
|
const bulkDbUpdate = hasFlag('bulk-db-update')
|
|
const skipDbUpdate = hasFlag('skip-db-update')
|
|
const oldPrefix = normalizePrefix(getArg('old-prefix', DEFAULT_OLD_PREFIX))
|
|
const oldBucket = getArg('old-bucket', 'jsxiangche')
|
|
const limit = Number(getArg('limit', '0')) || 0
|
|
const concurrency = Math.max(1, Number(getArg('concurrency', '8')) || 8)
|
|
const oldRows = await getOldPreviewUrls(oldPrefix, limit)
|
|
const mappings = oldRows.map((row) => {
|
|
const oldUrl = row.old_url
|
|
const key = ensureDestinationKey(parseOldObjectKey(oldUrl, oldBucket))
|
|
return {
|
|
oldUrl,
|
|
key,
|
|
newUrl: buildPublicUrl(key),
|
|
}
|
|
})
|
|
|
|
console.log(JSON.stringify({
|
|
mode: apply ? 'apply' : 'dry-run',
|
|
oldPrefix,
|
|
totalUrls: mappings.length,
|
|
concurrency,
|
|
bulkDbUpdate,
|
|
skipDbUpdate,
|
|
sample: mappings.slice(0, 3),
|
|
}, null, 2))
|
|
|
|
if (!apply) {
|
|
return
|
|
}
|
|
|
|
const s3Client = createS3Client()
|
|
const bucket = process.env.S3_BUCKET
|
|
const stats = {
|
|
uploaded: 0,
|
|
skippedExisting: 0,
|
|
dbMaterialUpdates: 0,
|
|
dbAssetUpdates: 0,
|
|
failed: 0,
|
|
}
|
|
const failures = []
|
|
|
|
await runPool(mappings, concurrency, async (mapping, currentIndex) => {
|
|
try {
|
|
const exists = await destinationExists(s3Client, bucket, mapping.key)
|
|
if (exists) {
|
|
stats.skippedExisting += 1
|
|
} else {
|
|
const oldObject = await fetchOldObject(mapping.oldUrl)
|
|
await s3Client.send(new PutObjectCommand({
|
|
Bucket: bucket,
|
|
Key: mapping.key,
|
|
Body: oldObject.body,
|
|
ContentType: oldObject.contentType,
|
|
ACL: process.env.S3_SET_ACL === '0' ? undefined : 'public-read',
|
|
}))
|
|
stats.uploaded += 1
|
|
}
|
|
|
|
if (!bulkDbUpdate && !skipDbUpdate) {
|
|
const dbResult = await updateDatabaseUrl(mapping.oldUrl, mapping.newUrl)
|
|
stats.dbMaterialUpdates += Number(dbResult.materialUpdates || 0)
|
|
stats.dbAssetUpdates += Number(dbResult.assetUpdates || 0)
|
|
}
|
|
} catch (error) {
|
|
stats.failed += 1
|
|
failures.push({
|
|
oldUrl: mapping.oldUrl,
|
|
message: error instanceof Error ? error.message : String(error),
|
|
})
|
|
}
|
|
|
|
if ((currentIndex + 1) % 1000 === 0 || currentIndex + 1 === mappings.length) {
|
|
console.log(JSON.stringify({ processed: currentIndex + 1, total: mappings.length, stats }))
|
|
}
|
|
})
|
|
|
|
if (bulkDbUpdate && stats.failed === 0) {
|
|
const newPrefix = mappings[0]
|
|
? mappings[0].newUrl.slice(0, mappings[0].newUrl.length - mappings[0].key.length)
|
|
: ''
|
|
if (!newPrefix) {
|
|
throw new Error('Cannot infer new prefix for bulk database update')
|
|
}
|
|
const dbResult = await bulkUpdateDatabaseUrls(oldPrefix, newPrefix)
|
|
stats.dbMaterialUpdates += Number(dbResult.materialUpdates || 0)
|
|
stats.dbAssetUpdates += Number(dbResult.assetUpdates || 0)
|
|
}
|
|
|
|
console.log(JSON.stringify({ done: true, stats, failures: failures.slice(0, 20) }, null, 2))
|
|
if (stats.failed > 0) {
|
|
process.exitCode = 1
|
|
}
|
|
}
|
|
|
|
main()
|
|
.catch((error) => {
|
|
console.error(error)
|
|
process.exitCode = 1
|
|
})
|
|
.finally(async () => {
|
|
await prisma.$disconnect()
|
|
})
|