232 lines
17 KiB
TypeScript
232 lines
17 KiB
TypeScript
/** 仅使用独立的本机 PostgreSQL 和临时上传目录，不读取生产配置。 */
|
|
import assert from 'node:assert/strict'
|
|
import { randomUUID } from 'node:crypto'
|
|
import { promises as fs } from 'node:fs'
|
|
import os from 'node:os'
|
|
import path from 'node:path'
|
|
|
|
async function main() {
|
|
const url = new URL(process.env.DATABASE_URL || '')
|
|
assert(['127.0.0.1', 'localhost'].includes(url.hostname) && url.port === '55439' && /^\/material_purge(?:_migration)?_qa$/.test(url.pathname), '拒绝连接非独立验收数据库')
|
|
process.env.NEXT_BUILD = '1'
|
|
process.env.NODE_ENV = 'production'
|
|
const s3Enabled = process.argv.includes('--s3')
|
|
if (s3Enabled) {
|
|
Object.assign(process.env, { S3_ENDPOINT: 'http://127.0.0.1:59000', S3_PUBLIC_DOMAIN: 'http://127.0.0.1:59000/purge-qa',
|
|
S3_BUCKET: 'purge-qa', S3_FILE_PATH: 'project-purge-qa', S3_ACCESS_KEY_ID: 'purgeqa', S3_SECRET_ACCESS_KEY: 'purge_local_storage_qa' })
|
|
}
|
|
const root = await fs.mkdtemp(path.join(os.tmpdir(), 'gongxiangp2-purge-integration-'))
|
|
process.env.UPLOAD_DIR = root
|
|
// 每次使用独立数据库，原生 SQL 与 Prisma 使用同一个 public schema。
|
|
const { PrismaClient } = await import('@prisma/client')
|
|
const database = `material_purge_qa_${randomUUID().replace(/-/g, '')}`
|
|
const setup = new PrismaClient({ datasourceUrl: url.toString() })
|
|
await setup.$executeRawUnsafe(`CREATE DATABASE "${database}"`)
|
|
await setup.$disconnect()
|
|
url.pathname = `/${database}`
|
|
url.searchParams.set('schema', 'public')
|
|
process.env.DATABASE_URL = url.toString()
|
|
const { spawnSync } = await import('node:child_process')
|
|
const pushed = spawnSync(process.execPath, ['node_modules/prisma/build/index.js', 'db', 'push', '--skip-generate'], { encoding: 'utf8', env: process.env })
|
|
assert.equal(pushed.status, 0, pushed.stderr)
|
|
// 重建迁移前的日志级联关系，在空的专用数据库中执行真实 reviewed SQL。
|
|
const migrationSql = await fs.readFile('prisma/migrations/20261009_material_purge_outbox/migration.sql', 'utf8')
|
|
const baselineSql = `DROP TABLE material_purge_audits; DROP TABLE material_purge_jobs;
|
|
ALTER TABLE material_logs ADD CONSTRAINT "material_logs_materialId_fkey" FOREIGN KEY ("materialId") REFERENCES materials(id) ON DELETE CASCADE;
|
|
ALTER TABLE material_download_logs ADD CONSTRAINT "material_download_logs_materialId_fkey" FOREIGN KEY ("materialId") REFERENCES materials(id) ON DELETE CASCADE;`
|
|
const migrated = spawnSync('docker', ['exec', '-i', 'gongxiangp2-purge-qa-20261009', 'psql', '-U', 'postgres', '-d', database, '-v', 'ON_ERROR_STOP=1'], {
|
|
encoding: 'utf8', input: `${baselineSql}\n${migrationSql}`,
|
|
})
|
|
assert.equal(migrated.status, 0, migrated.stderr)
|
|
const { prisma } = await import('../lib/prisma')
|
|
const { enqueueMaterialPurge, processNextMaterialPurge, retryMaterialPurge } = await import('../lib/material-purge')
|
|
const { collectPurgeFiles, findPurgeFileReferences, purgeFileIdentity } = await import('../lib/material-purge-files')
|
|
const run = randomUUID()
|
|
const passed: string[] = []
|
|
passed.push('真实升级 SQL 创建清理表、重试索引与审计表，并解除历史日志的素材级联删除')
|
|
const admin = await prisma.user.create({ data: { email: `${run}@purge.test`, username: run, password: 'local-qa-only', role: 'ADMIN' } })
|
|
const make = async (name: string, extra: any = {}) => prisma.material.create({ data: {
|
|
productModel: `${run}-${name}`, previewUrl: 'data:image/svg+xml;base64,Zm9v', categoryName: '验收',
|
|
sceneName: '验收', materialType: '标准模型', authorName: '隔离验收', userId: admin.id, deletedAt: new Date(), ...extra,
|
|
} })
|
|
const file = async (name: string) => { await fs.writeFile(path.join(root, name), 'isolated content'); return `uploads/${name}` }
|
|
const exists = async (name: string) => fs.stat(path.join(root, name)).then(() => true, () => false)
|
|
try {
|
|
const active = await make('active', { deletedAt: null })
|
|
const bin = await make('atomic', { zipFilePath: await file('atomic.bin') })
|
|
await assert.rejects(enqueueMaterialPurge([bin.id, active.id], admin.id), /只能彻底删除/)
|
|
assert(await prisma.material.findUnique({ where: { id: bin.id } }))
|
|
assert.equal(await prisma.materialPurgeJob.count({ where: { materialId: bin.id } }), 0)
|
|
assert(await exists('atomic.bin'))
|
|
passed.push('混合批量整批拒绝，素材/文件/任务均无部分提交')
|
|
|
|
await assert.rejects(enqueueMaterialPurge([bin.id], 'missing-operator'), /Foreign key/)
|
|
assert(await prisma.material.findUnique({ where: { id: bin.id } }))
|
|
assert.equal(await prisma.materialPurgeJob.count({ where: { materialId: bin.id } }), 0)
|
|
passed.push('审计写入失败时整个删除事务回滚')
|
|
|
|
const uploading = await make('uploading')
|
|
await prisma.productPackageUploadJob.create({ data: { materialId: uploading.id, operatorUserId: admin.id, assets: [], status: 'queued', stage: 'queued' } })
|
|
await assert.rejects(enqueueMaterialPurge([uploading.id], admin.id), /仍在处理中/)
|
|
const transcoding = await make('transcoding', { transcodeLeaseToken: 'active-lease' })
|
|
await assert.rejects(enqueueMaterialPurge([transcoding.id], admin.id), /仍在处理中/)
|
|
passed.push('上传/转码未结束的素材拒绝进入清理队列')
|
|
|
|
const own = await make('single', { zipFilePath: await file('single.bin') })
|
|
await prisma.materialAsset.create({ data: { materialId: own.id, assetType: 'arbitrary-type', fileUrl: await file('asset.bin') } })
|
|
await prisma.materialSku.create({ data: { materialId: own.id, skuName: 'test', filePath: await file('sku.bin') } })
|
|
await prisma.materialLog.create({ data: { materialId: own.id, userId: admin.id, action: 'UPLOAD' } })
|
|
await prisma.materialDownloadLog.create({ data: { materialId: own.id, userId: admin.id } })
|
|
const requests = await Promise.all([enqueueMaterialPurge([own.id, own.id], admin.id), enqueueMaterialPurge([own.id], admin.id)])
|
|
assert.equal(requests[0][0].id, requests[1][0].id)
|
|
assert.equal(await prisma.materialPurgeJob.count({ where: { materialId: own.id } }), 1)
|
|
assert.equal(await prisma.materialLog.count({ where: { materialId: own.id } }), 2)
|
|
assert.equal(await prisma.materialDownloadLog.count({ where: { materialId: own.id } }), 1)
|
|
assert.equal(await prisma.materialAsset.count({ where: { materialId: own.id } }), 0)
|
|
assert.equal(await prisma.materialSku.count({ where: { materialId: own.id } }), 0)
|
|
assert(await exists('single.bin'))
|
|
await processNextMaterialPurge()
|
|
assert(!await exists('single.bin') && !await exists('asset.bin') && !await exists('sku.bin'))
|
|
assert.equal((await prisma.materialPurgeJob.findUniqueOrThrow({ where: { materialId: own.id } })).status, 'completed')
|
|
passed.push('并发/重复提交幂等；关联记录清理，旧日志和下载日志保留；所有类型资产纳入清单')
|
|
|
|
const sharedSource = await file('shared.bin')
|
|
const keeper = await make('shared-keeper', { previewUrl: '/uploads/shared.bin' })
|
|
const sharer = await make('shared-delete', { zipFilePath: sharedSource })
|
|
await enqueueMaterialPurge([sharer.id], admin.id)
|
|
await processNextMaterialPurge()
|
|
assert(await exists('shared.bin'))
|
|
const sharedJob = await prisma.materialPurgeJob.findUniqueOrThrow({ where: { materialId: sharer.id } })
|
|
assert.equal((sharedJob.files as any)[0].outcome, 'shared')
|
|
assert(await prisma.material.findUnique({ where: { id: keeper.id } }))
|
|
passed.push('URL 等价引用与回收站引用保护共用文件')
|
|
|
|
const specialSource = await file('shared[1]+.bin')
|
|
const specialKeeper = await make('special-keeper')
|
|
await prisma.materialAsset.create({ data: { materialId: specialKeeper.id, assetType: 'test', fileUrl: '/uploads/shared[1]+.bin' } })
|
|
const special = await make('special-delete', { zipFilePath: specialSource })
|
|
const specialFiles = await collectPurgeFiles({ ...special, assets: [], skus: [] })
|
|
const fullReferences = await findPurgeFileReferences(prisma)
|
|
const targetedReferences = await findPurgeFileReferences(prisma, specialFiles)
|
|
for (const target of specialFiles) assert.equal(targetedReferences.has(purgeFileIdentity(target)), fullReferences.has(purgeFileIdentity(target)))
|
|
await enqueueMaterialPurge([special.id], admin.id)
|
|
await processNextMaterialPurge()
|
|
assert(await exists('shared[1]+.bin'))
|
|
const otherDirectory = await make('same-filename', { zipFilePath: await file('same-filename.bin') })
|
|
await prisma.materialAsset.create({ data: { materialId: specialKeeper.id, assetType: 'test', fileUrl: 'uploads/other/same-filename.bin' } })
|
|
await enqueueMaterialPurge([otherDirectory.id], admin.id)
|
|
await processNextMaterialPurge()
|
|
assert(!await exists('same-filename.bin'))
|
|
await findPurgeFileReferences(prisma, Array.from({ length: 2000 }, () => ({
|
|
source: '', storage: 'local' as const, key: `${randomUUID()}.bin`, outcome: 'pending' as const,
|
|
})))
|
|
passed.push('真实 PostgreSQL 候选筛选与全量引用结果一致：特殊文件名共用资产保留，同名不同路径文件正常删除')
|
|
|
|
const batch = await Promise.all([make('batch-a', { zipFilePath: await file('batch-a.bin') }), make('batch-b', { zipFilePath: await file('batch-b.bin') })])
|
|
const batchJobs = await enqueueMaterialPurge(batch.map(material => material.id), admin.id)
|
|
assert.equal(batchJobs.length, 2)
|
|
// 模拟作业进程中断：持久化处理中状态和已过期租约，下一轮自动接管。
|
|
await prisma.materialPurgeJob.update({ where: { id: batchJobs[0].id }, data: { status: 'processing', leaseToken: 'interrupted-process', leaseExpiresAt: new Date(0) } })
|
|
await Promise.all([processNextMaterialPurge(), processNextMaterialPurge()])
|
|
assert(!await exists('batch-a.bin') && !await exists('batch-b.bin'))
|
|
assert.equal(await prisma.materialPurgeJob.count({ where: { id: { in: batchJobs.map(job => job.id) }, status: 'completed' } }), 2)
|
|
passed.push('批量作业并发领取不重复，重启后过期租约可接管并完成')
|
|
|
|
const wide = await make('wide-batch')
|
|
const wideFiles = await Promise.all(Array.from({ length: 19 }, (_, index) => file(`wide-${index}.bin`)))
|
|
await prisma.materialAsset.createMany({ data: wideFiles.map(fileUrl => ({ materialId: wide.id, assetType: 'test', fileUrl })) })
|
|
const [wideJob] = await enqueueMaterialPurge([wide.id], admin.id)
|
|
const wideStarted = Date.now()
|
|
await processNextMaterialPurge()
|
|
const wideResult = await prisma.materialPurgeJob.findUniqueOrThrow({ where: { id: wideJob.id } })
|
|
assert.equal(wideResult.status, 'completed')
|
|
assert.equal((wideResult.files as any[]).filter(f => f.outcome === 'deleted').length, 19)
|
|
assert((await Promise.all(wideFiles.map((_, index) => exists(`wide-${index}.bin`)))).every(value => !value))
|
|
passed.push(`真实文件系统跨三个清理批次完成 19 个文件，结果全部持久化，耗时 ${Date.now() - wideStarted}ms`)
|
|
|
|
const broken = await make('retry', { zipFilePath: await file('retry-first.bin'), displayImages: [await file('retry-second.bin')] })
|
|
const [retryJob] = await enqueueMaterialPurge([broken.id], admin.id)
|
|
await fs.rename(path.join(root, 'retry-second.bin'), path.join(root, 'saved-second.bin'))
|
|
await fs.mkdir(path.join(root, 'retry-second.bin'))
|
|
for (let attempt = 0; attempt < 3; attempt += 1) {
|
|
await prisma.materialPurgeJob.update({ where: { id: retryJob.id }, data: { nextAttemptAt: new Date(0) } })
|
|
await processNextMaterialPurge()
|
|
}
|
|
let failed = await prisma.materialPurgeJob.findUniqueOrThrow({ where: { id: retryJob.id } })
|
|
assert.equal(failed.status, 'failed')
|
|
assert(!await exists('retry-first.bin'))
|
|
assert.equal((failed.files as any)[0].outcome, 'deleted')
|
|
await fs.rmdir(path.join(root, 'retry-second.bin'))
|
|
await fs.rename(path.join(root, 'saved-second.bin'), path.join(root, 'retry-second.bin'))
|
|
await retryMaterialPurge(retryJob.id, admin.id)
|
|
await processNextMaterialPurge()
|
|
failed = await prisma.materialPurgeJob.findUniqueOrThrow({ where: { id: retryJob.id } })
|
|
assert.equal(failed.status, 'completed')
|
|
assert(!await exists('retry-second.bin'))
|
|
const audits = await prisma.materialPurgeAudit.findMany({ where: { jobId: retryJob.id }, orderBy: { createdAt: 'asc' } })
|
|
assert(audits.some(audit => audit.event === 'failed'))
|
|
assert(audits.some(audit => audit.event === 'manual_retry' && audit.operatorUserId === admin.id))
|
|
passed.push('部分文件失败保存进度，自动重试耗尽后人工续跑；失败/重试操作人审计完整')
|
|
|
|
if (s3Enabled) {
|
|
const { S3Client, CreateBucketCommand, PutObjectCommand, HeadObjectCommand } = await import('@aws-sdk/client-s3')
|
|
const client = new S3Client({ endpoint: process.env.S3_ENDPOINT, region: 'us-east-1', forcePathStyle: true,
|
|
credentials: { accessKeyId: process.env.S3_ACCESS_KEY_ID!, secretAccessKey: process.env.S3_SECRET_ACCESS_KEY! }, maxAttempts: 1 })
|
|
const Bucket = process.env.S3_BUCKET!
|
|
try { await client.send(new CreateBucketCommand({ Bucket })) } catch (error: any) { if (error.name !== 'BucketAlreadyOwnedByYou') throw error }
|
|
const ownKey = `project-purge-qa/${run}/owned.bin`
|
|
const sharedKey = `project-purge-qa/${run}/shared.bin`
|
|
const foreignKey = `other-project/${run}/sentinel.bin`
|
|
for (const Key of [ownKey, sharedKey, foreignKey]) await client.send(new PutObjectCommand({ Bucket, Key, Body: 'isolated-object' }))
|
|
const source = (key: string) => `http://127.0.0.1:59000/${Bucket}/${key}`
|
|
const objectMaterial = await make('object-own', { zipFilePath: source(ownKey) })
|
|
await enqueueMaterialPurge([objectMaterial.id], admin.id)
|
|
await processNextMaterialPurge()
|
|
await assert.rejects(client.send(new HeadObjectCommand({ Bucket, Key: ownKey })), error => (error as any).$metadata.httpStatusCode === 404)
|
|
const objectKeeper = await make('object-keeper', { previewUrl: `${source(sharedKey)}?cache=keeper` })
|
|
const objectShared = await make('object-shared', { zipFilePath: source(sharedKey) })
|
|
await enqueueMaterialPurge([objectShared.id], admin.id)
|
|
await processNextMaterialPurge()
|
|
await client.send(new HeadObjectCommand({ Bucket, Key: sharedKey }))
|
|
assert(await prisma.material.findUnique({ where: { id: objectKeeper.id } }))
|
|
const encodedKey = `project-purge-qa/${run}/encoded[1]+.bin`
|
|
await client.send(new PutObjectCommand({ Bucket, Key: encodedKey, Body: 'isolated-object' }))
|
|
await prisma.materialAsset.create({ data: { materialId: objectKeeper.id, assetType: 'test',
|
|
fileUrl: `https://legacy-storage.test/${Bucket}/${encodedKey.replace('encoded', '%65ncoded')}?cache=1` } })
|
|
const encoded = await make('encoded-shared', { zipFilePath: source(encodedKey) })
|
|
await enqueueMaterialPurge([encoded.id], admin.id)
|
|
await processNextMaterialPurge()
|
|
await client.send(new HeadObjectCommand({ Bucket, Key: encodedKey }))
|
|
const localWhiteBackground = `uploads/skus/${objectKeeper.id}/white-background/${run}/preview/shared-display.png`
|
|
await prisma.materialAsset.create({ data: { materialId: objectKeeper.id, assetType: 'test', fileUrl: localWhiteBackground } })
|
|
const { buildPublicDisplayUrl, buildWhiteBackgroundPublicDisplayPathFromAssetPath } = await import('../lib/public-display')
|
|
const displaySource = buildPublicDisplayUrl(buildWhiteBackgroundPublicDisplayPathFromAssetPath(localWhiteBackground)!)!
|
|
const { getTrustedS3MutationKey } = await import('../lib/s3')
|
|
const displayKey = getTrustedS3MutationKey(displaySource)
|
|
await client.send(new PutObjectCommand({ Bucket, Key: displayKey, Body: 'isolated-display' }))
|
|
const display = await make('derived-shared', { zipFilePath: displaySource })
|
|
await enqueueMaterialPurge([display.id], admin.id)
|
|
await processNextMaterialPurge()
|
|
await client.send(new HeadObjectCommand({ Bucket, Key: displayKey }))
|
|
const objectForeign = await make('object-foreign', { zipFilePath: source(foreignKey) })
|
|
await assert.rejects(enqueueMaterialPurge([objectForeign.id], admin.id), /业务前缀之外/)
|
|
await client.send(new HeadObjectCommand({ Bucket, Key: foreignKey }))
|
|
assert(await prisma.material.findUnique({ where: { id: objectForeign.id } }))
|
|
passed.push('真实 MinIO 验证：本项目对象删除成功，共用对象保留，其他项目前缀拒绝删除')
|
|
client.destroy()
|
|
}
|
|
|
|
const output = { database: url.pathname.slice(1), passed, checks: passed.length, at: new Date().toISOString() }
|
|
await fs.mkdir('tmp/material-purge-qa', { recursive: true })
|
|
await fs.writeFile('tmp/material-purge-qa/integration.json', JSON.stringify(output, null, 2))
|
|
console.log(JSON.stringify(output, null, 2))
|
|
} finally {
|
|
await prisma.$disconnect()
|
|
// root 是本次 mkdtemp 创建的目录，仅移除这次隔离文件。
|
|
assert(path.dirname(root) === os.tmpdir() && path.basename(root).startsWith('gongxiangp2-purge-integration-'))
|
|
await fs.rm(root, { recursive: true, force: true })
|
|
}
|
|
}
|
|
|
|
main().catch(error => { console.error(error); process.exitCode = 1 })
|