212 lines
11 KiB
TypeScript
212 lines
11 KiB
TypeScript
import { randomUUID } from 'crypto'
|
||
import { Prisma, type MaterialPurgeJob } from '@prisma/client'
|
||
import { prisma } from '@/lib/prisma'
|
||
import { collectPurgeFiles, findPurgeFileReferences, purgeFileIdentity, removePurgeFile, type PurgeFile } from '@/lib/material-purge-files'
|
||
|
||
export class MaterialPurgeError extends Error {
|
||
constructor(message: string, readonly status = 409) { super(message); this.name = 'MaterialPurgeError' }
|
||
}
|
||
|
||
const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i
|
||
const LEASE_MS = 10 * 60_000
|
||
const MAX_ATTEMPTS = 3
|
||
const FILE_BATCH_SIZE = 8
|
||
|
||
export function validatePurgeIds(input: unknown): string[] {
|
||
if (!Array.isArray(input) || input.length === 0 || input.length > 100 || input.some(id => typeof id !== 'string' || !UUID.test(id))) {
|
||
throw new MaterialPurgeError('请选择 1–100 条有效素材', 400)
|
||
}
|
||
return Array.from(new Set<string>(input.map(id => id.toLowerCase()))).sort()
|
||
}
|
||
|
||
export function toPurgeJobView(job: MaterialPurgeJob) {
|
||
const files = job.files as unknown as PurgeFile[]
|
||
const snapshot = job.materialSnapshot as { productModel?: string }
|
||
return {
|
||
id: job.id, materialId: job.materialId, name: snapshot.productModel || '素材',
|
||
status: job.status, totalFiles: files.length,
|
||
processedFiles: files.filter(file => file.outcome !== 'pending').length,
|
||
sharedFiles: files.filter(file => file.outcome === 'shared').length,
|
||
error: job.errorMessage, createdAt: job.createdAt.toISOString(),
|
||
}
|
||
}
|
||
|
||
async function lock(tx: Prisma.TransactionClient, name: string) {
|
||
await tx.$queryRaw`SELECT 1 AS locked FROM (SELECT pg_advisory_xact_lock(hashtext(${name}))) AS locked`
|
||
}
|
||
|
||
/** 清单、审计和素材记录删除必须一起提交；请求路径不做物理删除。 */
|
||
export async function enqueueMaterialPurge(input: unknown, operatorUserId: string) {
|
||
const ids = validatePurgeIds(input)
|
||
return prisma.$transaction(async tx => {
|
||
await lock(tx, 'material-purge-enqueue')
|
||
await lock(tx, 'product-package-upload-create')
|
||
await lock(tx, 'product-package-upload-worker')
|
||
await tx.$queryRaw`SELECT 1 AS locked FROM pg_advisory_xact_lock(193582731)`
|
||
const jobs = await tx.materialPurgeJob.findMany({ where: { materialId: { in: ids } } })
|
||
const newIds = ids.filter(id => !jobs.some(job => job.materialId === id))
|
||
if (!newIds.length) return jobs.map(toPurgeJobView)
|
||
await tx.$queryRaw`SELECT id FROM materials WHERE id IN (${Prisma.join(newIds)}) ORDER BY id FOR UPDATE`
|
||
const materials = await tx.material.findMany({
|
||
where: { id: { in: newIds } }, include: { assets: true, skus: true },
|
||
})
|
||
if (materials.length !== newIds.length) throw new MaterialPurgeError('部分素材已不存在，请刷新后重试', 404)
|
||
if (materials.some(material => !material.deletedAt)) throw new MaterialPurgeError('只能彻底删除回收站中的素材')
|
||
const uploading = await tx.productPackageUploadJob.count({ where: { materialId: { in: newIds }, status: { in: ['queued', 'processing'] } } })
|
||
const applying = await tx.taskDetailChangeRequest.count({ where: { materialId: { in: newIds }, status: 'applying' } })
|
||
if (uploading || applying || materials.some(material => material.transcodeLeaseToken)) {
|
||
throw new MaterialPurgeError('部分素材仍在处理中，请处理完成后重试')
|
||
}
|
||
for (const material of materials) {
|
||
let files: PurgeFile[]
|
||
try { files = await collectPurgeFiles(material) } catch (error: any) {
|
||
throw new MaterialPurgeError(`${material.productModel || '素材'}：${error.message}`)
|
||
}
|
||
const snapshot = {
|
||
id: material.id, productModel: material.productModel, materialType: material.materialType,
|
||
taskNo: material.taskNo, ownerUserId: material.userId, deletedAt: material.deletedAt!.toISOString(),
|
||
}
|
||
const job = await tx.materialPurgeJob.create({ data: {
|
||
materialId: material.id, operatorUserId, materialSnapshot: snapshot,
|
||
files: files as unknown as Prisma.InputJsonValue,
|
||
} })
|
||
await tx.materialPurgeAudit.create({ data: { jobId: job.id, event: 'queued', operatorUserId } })
|
||
await tx.materialLog.create({ data: {
|
||
materialId: material.id, userId: operatorUserId, action: 'DELETE',
|
||
changes: JSON.stringify({
|
||
materialSnapshot: { oldValue: null, newValue: snapshot },
|
||
deleteMode: { oldValue: 'soft', newValue: 'permanent' },
|
||
purgeJobId: { oldValue: null, newValue: job.id },
|
||
}),
|
||
} })
|
||
// FK 清理关联资产、收藏等记录；操作/下载日志已由迁移解除素材级联关系。
|
||
await tx.material.delete({ where: { id: material.id } })
|
||
jobs.push(job)
|
||
}
|
||
return jobs.map(toPurgeJobView)
|
||
}, { timeout: 30_000 })
|
||
}
|
||
|
||
async function claimJob() {
|
||
return prisma.$transaction(async tx => {
|
||
const rows = await tx.$queryRaw<{ id: string }[]>`
|
||
SELECT id FROM material_purge_jobs
|
||
WHERE (status = 'queued' AND next_attempt_at <= clock_timestamp())
|
||
OR (status = 'processing' AND lease_expires_at <= clock_timestamp())
|
||
ORDER BY created_at LIMIT 1 FOR UPDATE SKIP LOCKED`
|
||
if (!rows.length) return null
|
||
const leaseToken = randomUUID()
|
||
const [clock] = await tx.$queryRaw<{ now: Date }[]>`SELECT clock_timestamp() AS now`
|
||
const job = await tx.materialPurgeJob.update({ where: { id: rows[0].id }, data: {
|
||
status: 'processing', leaseToken, leaseExpiresAt: new Date(clock.now.getTime() + LEASE_MS),
|
||
attempts: { increment: 1 }, errorMessage: null,
|
||
} })
|
||
await tx.materialPurgeAudit.create({ data: { jobId: job.id, event: 'processing' } })
|
||
return job
|
||
})
|
||
}
|
||
|
||
async function updateOwnedJob(job: MaterialPurgeJob, data: Prisma.MaterialPurgeJobUpdateManyMutationInput) {
|
||
const updated = await prisma.$executeRaw`
|
||
UPDATE material_purge_jobs SET lease_expires_at = clock_timestamp() + interval '10 minutes'
|
||
WHERE id = ${job.id} AND status = 'processing' AND lease_token = ${job.leaseToken}
|
||
AND lease_expires_at > clock_timestamp()`
|
||
if (!updated) throw new Error('文件清理任务已移交，等待新的处理结果')
|
||
await prisma.$transaction(async tx => {
|
||
const result = await tx.materialPurgeJob.updateMany({
|
||
where: { id: job.id, status: 'processing', leaseToken: job.leaseToken }, data: { ...data, updatedAt: new Date() },
|
||
})
|
||
if (!result.count) throw new Error('文件清理任务已移交，等待新的处理结果')
|
||
if (data.status === 'completed') await tx.materialPurgeAudit.create({ data: { jobId: job.id, event: 'completed' } })
|
||
})
|
||
}
|
||
|
||
/** 每批最多 8 个文件，逐文件记录结果；缺失文件视为成功，可从进程中断处续跑。 */
|
||
export async function processNextMaterialPurge(): Promise<boolean> {
|
||
const job = await claimJob()
|
||
if (!job) return false
|
||
const files = job.files as unknown as PurgeFile[]
|
||
let leaseLost = false
|
||
const heartbeat = setInterval(() => {
|
||
void updateOwnedJob(job, {}).catch(() => { leaseLost = true })
|
||
}, 30_000)
|
||
heartbeat.unref?.()
|
||
try {
|
||
const references = await findPurgeFileReferences(prisma, files.filter(file => file.outcome === 'pending'))
|
||
const pending = files.filter(file => file.outcome === 'pending')
|
||
for (let offset = 0; offset < pending.length; offset += FILE_BATCH_SIZE) {
|
||
if (leaseLost) throw new Error('文件清理任务已移交，等待新的处理结果')
|
||
await updateOwnedJob(job, {})
|
||
const batch = pending.slice(offset, offset + FILE_BATCH_SIZE)
|
||
const results = await Promise.allSettled(batch.map(async file => {
|
||
if (references.has(purgeFileIdentity(file))) return 'shared' as const
|
||
await removePurgeFile(file)
|
||
return 'deleted' as const
|
||
}))
|
||
results.forEach((result, index) => {
|
||
if (result.status === 'fulfilled') batch[index].outcome = result.value
|
||
})
|
||
// 即使批内部分失败，也先保存成功文件的结果，再进入原有重试流程。
|
||
await updateOwnedJob(job, { files: files as unknown as Prisma.InputJsonValue })
|
||
const failure = results.find((result): result is PromiseRejectedResult => result.status === 'rejected')
|
||
if (failure) throw failure.reason
|
||
}
|
||
await updateOwnedJob(job, { status: 'completed', completedAt: new Date(), leaseToken: null, leaseExpiresAt: null })
|
||
} catch (error: any) {
|
||
const retry = job.attempts < MAX_ATTEMPTS
|
||
const message = error instanceof Error && /^(文件|存储|对象存储)/.test(error.message)
|
||
? error.message : '暂时无法清理文件，请稍后重试'
|
||
await prisma.$transaction(async tx => {
|
||
const result = await tx.materialPurgeJob.updateMany({
|
||
where: { id: job.id, status: 'processing', leaseToken: job.leaseToken },
|
||
data: {
|
||
status: retry ? 'queued' : 'failed', errorMessage: message,
|
||
nextAttemptAt: new Date(Date.now() + job.attempts * 30_000), leaseToken: null, leaseExpiresAt: null,
|
||
},
|
||
})
|
||
if (result.count) await tx.materialPurgeAudit.create({ data: { jobId: job.id, event: retry ? 'retry_scheduled' : 'failed', message } })
|
||
})
|
||
} finally { clearInterval(heartbeat) }
|
||
return true
|
||
}
|
||
|
||
export async function retryMaterialPurge(jobId: string, operatorUserId: string) {
|
||
if (typeof jobId !== 'string' || !UUID.test(jobId)) throw new MaterialPurgeError('清理任务无效', 400)
|
||
return prisma.$transaction(async tx => {
|
||
const result = await tx.materialPurgeJob.updateMany({ where: { id: jobId, status: 'failed' }, data: {
|
||
status: 'queued', attempts: 0, errorMessage: null, nextAttemptAt: new Date(),
|
||
} })
|
||
if (!result.count) throw new MaterialPurgeError('该任务已在处理或已经完成，请刷新状态')
|
||
await tx.materialPurgeAudit.create({ data: { jobId, event: 'manual_retry', operatorUserId } })
|
||
return toPurgeJobView(await tx.materialPurgeJob.findUniqueOrThrow({ where: { id: jobId } }))
|
||
})
|
||
}
|
||
|
||
export function ensureMaterialPurgeWorkerStarted() {
|
||
if (process.env.NEXT_BUILD === '1' || process.env.NEXT_PHASE === 'phase-production-build') return
|
||
const global = globalThis as typeof globalThis & { __materialPurgeWorkerStarted?: boolean; __materialPurgePump?: () => Promise<void> }
|
||
if (global.__materialPurgeWorkerStarted) { void global.__materialPurgePump?.(); return }
|
||
global.__materialPurgeWorkerStarted = true
|
||
let running = false
|
||
let wakeRequested = false
|
||
const pump = async () => {
|
||
if (running) { wakeRequested = true; return }
|
||
running = true
|
||
wakeRequested = false
|
||
let more = false
|
||
try {
|
||
for (let count = 0; count < 20 && await processNextMaterialPurge(); count += 1) more = count === 19
|
||
}
|
||
catch (error) { console.error('[MaterialPurge] 清理队列暂不可用，将在下一轮重试:', error) }
|
||
finally {
|
||
running = false
|
||
// 入队与当前空队列检查交错时保留唤醒信号；满一轮继续处理下一批。
|
||
if (wakeRequested || more) setImmediate(() => { void pump() }).unref?.()
|
||
}
|
||
}
|
||
// 入队事务提交后直接唤醒；轮询仅用于重启恢复和延迟重试，不让新请求等下一轮。
|
||
global.__materialPurgePump = pump
|
||
setInterval(() => { void pump() }, 5_000).unref?.()
|
||
void pump()
|
||
}
|