/** 仅使用独立的本机 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 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 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 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 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 })