364 lines
27 KiB
TypeScript
364 lines
27 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'
|
||
import type { Prisma } from '@prisma/client'
|
||
import type { PurgeFile } from '../lib/material-purge-files'
|
||
|
||
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 referenceSql = await fs.readFile('prisma/migrations/20261009_material_purge_reference_indexes/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}\n${referenceSql}`,
|
||
})
|
||
assert.equal(migrated.status, 0, migrated.stderr)
|
||
const { prisma } = await import('../lib/prisma')
|
||
const { enqueueMaterialPurge, processNextMaterialPurge, retryMaterialPurge, ensureMaterialPurgeWorkerStarted } = await import('../lib/material-purge')
|
||
const { collectPurgeFiles, findPurgeFileReferences, purgeFileIdentity, identifyPurgeFile } = await import('../lib/material-purge-files')
|
||
const { findIndexedPurgeReferenceSources } = await import('../lib/material-purge-reference-query')
|
||
async function explainReferences(db: Prisma.TransactionClient, files: PurgeFile[]) {
|
||
let query: { strings: TemplateStringsArray; values: unknown[] } | undefined
|
||
const wrapped = { $queryRaw: (strings: TemplateStringsArray, ...values: unknown[]) => {
|
||
query = { strings, values }
|
||
return db.$queryRaw(strings, ...values)
|
||
} } as Prisma.TransactionClient
|
||
await findIndexedPurgeReferenceSources(wrapped, files)
|
||
assert(query)
|
||
const sql = query.strings.reduce((value, part, index) => value + (index ? '$' + index : '') + part, '')
|
||
return db.$queryRawUnsafe<any[]>('EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ' + sql, ...query.values)
|
||
}
|
||
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('部分文件失败保存进度，自动重试耗尽后人工续跑；失败/重试操作人审计完整')
|
||
|
||
const indexedSource = await file('indexed-变更[1]%.bin')
|
||
const indexedFiles = [{ ...identifyPurgeFile(indexedSource)!, outcome: 'pending' as const }]
|
||
const indexedKeeper = await make('index-maintenance')
|
||
const indexedAsset = await prisma.materialAsset.create({ data: { materialId: indexedKeeper.id, assetType: 'test', fileUrl: indexedSource } })
|
||
assert((await findPurgeFileReferences(prisma, indexedFiles)).has(purgeFileIdentity(indexedFiles[0])))
|
||
await prisma.$executeRaw`UPDATE material_assets SET file_url = 'uploads/unrelated.bin' WHERE id = ${indexedAsset.id}`
|
||
assert(!(await findPurgeFileReferences(prisma, indexedFiles)).has(purgeFileIdentity(indexedFiles[0])))
|
||
const jsonJob = await prisma.productPackageUploadJob.create({ data: { materialId: indexedKeeper.id, operatorUserId: admin.id,
|
||
assets: [{ nested: { fileUrl: indexedSource } }], status: 'queued', stage: 'queued' } })
|
||
assert((await findPurgeFileReferences(prisma, indexedFiles)).has(purgeFileIdentity(indexedFiles[0])))
|
||
await prisma.productPackageUploadJob.update({ where: { id: jsonJob.id }, data: { status: 'completed' } })
|
||
assert(!(await findPurgeFileReferences(prisma, indexedFiles)).has(purgeFileIdentity(indexedFiles[0])))
|
||
const aliases = [indexedSource, `/uploads/./indexed-变更[1]%.bin`, path.join(root, 'indexed-变更[1]%.bin'),
|
||
'\u00a0/UPLOADS/INDEXED-变更[1]%.BIN\ufeff', 'uploads\\indexed-变更[1]%.bin']
|
||
for (const previewUrl of aliases) {
|
||
await prisma.material.update({ where: { id: indexedKeeper.id }, data: { previewUrl } })
|
||
const full = await findPurgeFileReferences(prisma)
|
||
const targeted = await findPurgeFileReferences(prisma, indexedFiles)
|
||
assert.equal(targeted.has(purgeFileIdentity(indexedFiles[0])), full.has(purgeFileIdentity(indexedFiles[0])), previewUrl)
|
||
}
|
||
await prisma.material.update({ where: { id: indexedKeeper.id }, data: { previewUrl: 'data:image/png;base64,Zm9v' } })
|
||
passed.push('数据库索引随 ORM/直接 SQL 写入和队列状态变化同步；Unicode、百分号、大小写、绝对路径、点路径引用与全量审计一致')
|
||
|
||
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`
|
||
const whiteAsset = 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 displayTarget = [{ ...identifyPurgeFile(displaySource)!, outcome: 'pending' as const }]
|
||
for (const fileUrl of [localWhiteBackground + '#preview', localWhiteBackground + '?cache=1',
|
||
localWhiteBackground.replace('/preview/shared-display', '/preview/./shared-display'),
|
||
localWhiteBackground.replace('/preview/shared-display', '/preview/sub/../shared-display'),
|
||
localWhiteBackground.replace('/preview/shared-display', '/preview/%2e/shared-display')]) {
|
||
await prisma.materialAsset.update({ where: { id: whiteAsset.id }, data: { fileUrl } })
|
||
const full = await findPurgeFileReferences(prisma)
|
||
const next = await findPurgeFileReferences(prisma, displayTarget)
|
||
assert(full.has(purgeFileIdentity(displayTarget[0])), '白底派生引用确实存在')
|
||
assert(next.has(purgeFileIdentity(displayTarget[0])), `不得漏掉派生 URL 引用 ${fileUrl}`)
|
||
}
|
||
await prisma.materialAsset.update({ where: { id: whiteAsset.id }, data: { fileUrl: localWhiteBackground } })
|
||
const referenceTarget = [{ ...identifyPurgeFile(source(sharedKey))!, outcome: 'pending' as const }]
|
||
for (const previewUrl of [
|
||
source(sharedKey).replace('shared', '%73hared'), source(sharedKey).replace('shared', '%2573hared'),
|
||
`https://legacy.test/${sharedKey.replace('shared', '%73hared')}?cache=1#fragment`,
|
||
`https://legacy.test/${sharedKey.replace('shared', 'sha\tred')}`,
|
||
source(sharedKey).replace('/shared', '/unused/../shared'),
|
||
source(sharedKey).replace('/shared', '/%2e/shared'),
|
||
source(sharedKey).replace(`/${Bucket}/`, `//${Bucket}/`),
|
||
]) {
|
||
await prisma.material.update({ where: { id: objectKeeper.id }, data: { previewUrl } })
|
||
const full = await findPurgeFileReferences(prisma)
|
||
const targeted = await findPurgeFileReferences(prisma, referenceTarget)
|
||
assert.equal(targeted.has(purgeFileIdentity(referenceTarget[0])), full.has(purgeFileIdentity(referenceTarget[0])), previewUrl)
|
||
}
|
||
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()
|
||
}
|
||
|
||
if (process.argv.includes('--scale')) {
|
||
const scaleKeeper = await make('400k-reference-keeper')
|
||
for (let offset = 0; offset < 400_000; offset += 80_000) {
|
||
await prisma.$executeRaw`INSERT INTO material_assets (id, material_id, asset_type, file_url, sort_order, updated_at)
|
||
SELECT ${run + '-scale-'} || i::text, ${scaleKeeper.id}, 'qa-scale',
|
||
'https://isolated.test/project/scale/' || i::text || '/image.bin', 0, now() FROM generate_series(${offset}, ${offset + 79_999}) AS i`
|
||
console.log(JSON.stringify({ mode: 'isolated-scale-setup', assets: offset + 80_000 }))
|
||
}
|
||
await prisma.$executeRaw`ANALYZE material_assets`
|
||
const scaleSources = await Promise.all(Array.from({ length: 4 }, (_, i) => file(`scale-delete-${i}.bin`)))
|
||
const scaleMaterial = await make('scale-delete', { zipFilePath: scaleSources[0], displayImages: scaleSources.slice(1) })
|
||
const scaleFiles = await collectPurgeFiles({ ...scaleMaterial, assets: [], skus: [] })
|
||
const plan = await explainReferences(prisma, scaleFiles)
|
||
assert(JSON.stringify(plan).includes('material_assets_purge_references_idx'), '40万行必须实际使用引用索引')
|
||
const referenceStarted = Date.now()
|
||
await findPurgeFileReferences(prisma, scaleFiles)
|
||
const referenceMs = Date.now() - referenceStarted
|
||
const submitted = Date.now()
|
||
const [scaleJob] = await enqueueMaterialPurge([scaleMaterial.id], admin.id)
|
||
const acceptedMs = Date.now() - submitted
|
||
await processNextMaterialPurge()
|
||
const completedMs = Date.now() - submitted
|
||
const result = await prisma.materialPurgeJob.findUniqueOrThrow({ where: { id: scaleJob.id } })
|
||
assert.equal(result.status, 'completed')
|
||
assert((await Promise.all(scaleSources.map((_, i) => exists(`scale-delete-${i}.bin`)))).every(value => !value))
|
||
assert(referenceMs < 500 && completedMs < 1500, `索引检查 ${referenceMs}ms / 实际删除 ${completedMs}ms 超出验收线`)
|
||
const bulkFiles: PurgeFile[] = Array.from({ length: 106 }, (_, index) => ({
|
||
source: `https://isolated.test/project/bulk/${index}.bin`, storage: 's3',
|
||
bucket: process.env.S3_BUCKET || 'qa-bucket', key: `project/bulk/${index}.bin`, outcome: 'pending',
|
||
}))
|
||
bulkFiles.push(...specialFiles)
|
||
const bulkPlans = []
|
||
for (const mode of ['force_custom_plan', 'force_generic_plan']) {
|
||
const measurement = await prisma.$transaction(async tx => {
|
||
await tx.$executeRawUnsafe('SET LOCAL plan_cache_mode = ' + mode)
|
||
const started = Date.now()
|
||
const refs = await findPurgeFileReferences(tx, bulkFiles)
|
||
const ms = Date.now() - started
|
||
assert(refs.has(purgeFileIdentity(specialFiles[0])), '大批量查询仍须保留共享文件')
|
||
const executionPlan = await explainReferences(tx, bulkFiles)
|
||
assert(JSON.stringify(executionPlan).includes('material_assets_purge_references_idx'), `${mode} 大批量必须使用索引`)
|
||
assert(ms < 1500, `${mode} 大批量引用查询 ${ms}ms 超出验收线`)
|
||
return { mode, files: bulkFiles.length, ms, plan: executionPlan }
|
||
}, { timeout: 30_000 })
|
||
bulkPlans.push(measurement)
|
||
}
|
||
await fs.mkdir('tmp/material-purge-index-qa', { recursive: true })
|
||
await fs.writeFile('tmp/material-purge-index-qa/scale.json', JSON.stringify({ assets: await prisma.materialAsset.count(), referenceMs, acceptedMs, completedMs, plan, bulkPlans }, null, 2))
|
||
passed.push(`40万资产下只查目标引用 ${referenceMs}ms；提交 ${acceptedMs}ms、4个文件真实删除及审计完成 ${completedMs}ms，执行计划使用 GIN 索引`)
|
||
passed.push(`107个文件在 custom/generic 两种执行计划下均使用索引并保护共享引用：${bulkPlans.map(p => p.ms + 'ms').join(' / ')}`)
|
||
}
|
||
|
||
delete process.env.NEXT_BUILD
|
||
ensureMaterialPurgeWorkerStarted()
|
||
await new Promise(resolve => setTimeout(resolve, 100))
|
||
const wake = await make('wake-up', { zipFilePath: await file('wake-up.bin') })
|
||
const [wakeJob] = await enqueueMaterialPurge([wake.id], admin.id)
|
||
const wakeStarted = Date.now()
|
||
ensureMaterialPurgeWorkerStarted()
|
||
let wakeResult
|
||
do {
|
||
await new Promise(resolve => setTimeout(resolve, 25))
|
||
wakeResult = await prisma.materialPurgeJob.findUniqueOrThrow({ where: { id: wakeJob.id } })
|
||
} while (wakeResult.status !== 'completed' && Date.now() - wakeStarted < 1500)
|
||
assert.equal(wakeResult.status, 'completed', '新请求必须直接唤醒已启动的 worker，不能等 5 秒轮询')
|
||
process.env.NEXT_BUILD = '1'
|
||
passed.push(`已启动的 worker 收到提交立即清理，${Date.now() - wakeStarted}ms 完成，未等待5秒轮询`)
|
||
|
||
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 })
|