977 lines
29 KiB
TypeScript
977 lines
29 KiB
TypeScript
/**
|
||
* Material upload 临时目录的统一安全清理协议。
|
||
*
|
||
* 候选扫描可以在锁外进行，但任何删除决定都必须在跨容器独占锁内重新检查：
|
||
* 会话、owner、目录年龄、seal 和 merged 文件。取消时先持久化 cancelled
|
||
* tombstone，再清理 Redis + 当前进程内存会话；释放锁后才删除目录。
|
||
*/
|
||
|
||
import { lstat, readFile, readdir, rm } from 'fs/promises'
|
||
import path from 'path'
|
||
|
||
import {
|
||
acquireMaterialUploadStorageLock,
|
||
acquireTaskMaterialUploadSessionLocks,
|
||
type MaterialUploadStorageLockRelease,
|
||
} from './material-upload-cross-process-lock'
|
||
import {
|
||
cancelUploadSessionSeal,
|
||
cleanupExpiredUploadCompletionSeals,
|
||
readUploadSessionSeal,
|
||
type UploadSessionSealState,
|
||
} from './material-upload-session-seal'
|
||
import { MaterialUploadRequestError } from './material-upload-request-guard'
|
||
import { normalizeSafeFileName, resolveStrictChildPath } from './path-security'
|
||
import { getUploadRootDir } from './upload-paths'
|
||
import { uploadSessions, type UploadSession } from './upload-sessions'
|
||
|
||
const MAX_AGE_MS = 8 * 60 * 60 * 1000
|
||
const ABANDONED_BRAND_KNOWLEDGE_RETENTION_MS = 24 * 60 * 60 * 1000
|
||
const CLEANUP_INTERVAL_MS = 60 * 60 * 1000
|
||
const SAFE_UPLOAD_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$/
|
||
const REDIS_SESSION_KEY_PREFIX = 'upload:session:'
|
||
const INTERNAL_CONSUMED_CLEANUP = Symbol('internal-consumed-material-upload-cleanup')
|
||
const INTERNAL_ABANDONED_BRAND_KNOWLEDGE_CLEANUP = Symbol('internal-abandoned-brand-knowledge-cleanup')
|
||
|
||
let cleanupTimer: NodeJS.Timeout | null = null
|
||
let isCleanupRunning = false
|
||
|
||
export type SafeMaterialUploadCleanupStatus =
|
||
| 'deleted'
|
||
| 'absent'
|
||
| 'skipped_active'
|
||
| 'skipped_young'
|
||
| 'protected_completed'
|
||
| 'protected_sealing'
|
||
| 'not_consumed'
|
||
| 'not_abandoned_brand_knowledge'
|
||
| 'owner_mismatch'
|
||
| 'unavailable'
|
||
|
||
export interface SafeMaterialUploadCleanupResult {
|
||
status: SafeMaterialUploadCleanupStatus
|
||
success: boolean
|
||
deletedFiles: number
|
||
freedSpace: number
|
||
error?: string
|
||
code?: string
|
||
httpStatus?: number
|
||
}
|
||
|
||
export interface SafeMaterialUploadCleanupInput {
|
||
uploadId: string
|
||
intent: 'cancel' | 'sweep'
|
||
expectedOwnerId?: string
|
||
minAgeMs?: number
|
||
allowYoungOrphaned?: boolean
|
||
}
|
||
|
||
interface InternalCleanupInput extends SafeMaterialUploadCleanupInput {
|
||
intent: 'cancel' | 'sweep'
|
||
capability?:
|
||
| typeof INTERNAL_CONSUMED_CLEANUP
|
||
| typeof INTERNAL_ABANDONED_BRAND_KNOWLEDGE_CLEANUP
|
||
}
|
||
|
||
interface StoredSessionEvidence {
|
||
source: 'memory' | 'redis'
|
||
userId: string
|
||
fileName: string
|
||
fileSize: number | null
|
||
createdAt: number | null
|
||
uploadPurpose: string | null
|
||
}
|
||
|
||
interface UploadDirectorySnapshot {
|
||
path: string
|
||
exists: boolean
|
||
mtimeMs: number | null
|
||
metadata: {
|
||
userId: string
|
||
fileName: string
|
||
fileSize: number
|
||
createdAt: number
|
||
uploadPurpose: string | null
|
||
} | null
|
||
files: Array<{ name: string; size: number }>
|
||
deletedFiles: number
|
||
freedSpace: number
|
||
}
|
||
|
||
interface CleanupEvidence {
|
||
directory: UploadDirectorySnapshot
|
||
sessions: StoredSessionEvidence[]
|
||
ownerId: string | null
|
||
expectedFileSize: number | null
|
||
activeSession: boolean
|
||
ageMs: number | null
|
||
hasValidMerged: boolean
|
||
sealState: UploadSessionSealState | null
|
||
assembledFilePath: string | null
|
||
uploadPurpose: string | null
|
||
}
|
||
|
||
/**
|
||
* 公开取消接口和自动清理只能调用这个入口；它们没有能力进入“已消费”模式。
|
||
*/
|
||
export async function safelyCleanupMaterialUploadSession(
|
||
input: SafeMaterialUploadCleanupInput,
|
||
): Promise<SafeMaterialUploadCleanupResult> {
|
||
return executeSafeCleanup(input)
|
||
}
|
||
|
||
/**
|
||
* 仅供素材已经成功落库/转存后的服务端业务清理使用。这个入口不会写 cancelled，
|
||
* 且只有 completed + 有效 merged 的会话才允许删除临时目录。公开 cleanup 路由
|
||
* 不持有 capability，不能借此绕过 completed 保护。
|
||
*/
|
||
export async function cleanupConsumedUploadSession(
|
||
uploadId: string,
|
||
): Promise<SafeMaterialUploadCleanupResult> {
|
||
return executeSafeCleanup({
|
||
uploadId,
|
||
intent: 'cancel',
|
||
capability: INTERNAL_CONSUMED_CLEANUP,
|
||
})
|
||
}
|
||
|
||
/**
|
||
* 兼容现有服务端素材建档调用。语义已收紧为“已消费清理”，不能用于用户取消。
|
||
* 新代码应使用 cleanupConsumedUploadSession 这个明确命名的入口。
|
||
*/
|
||
export async function cleanupUploadSession(uploadId: string): Promise<boolean> {
|
||
const result = await cleanupConsumedUploadSession(uploadId)
|
||
return result.success
|
||
}
|
||
|
||
/**
|
||
* 启动时和每小时扫描一次。Redis/状态 I/O 任何异常都只计为跳过，绝不删目录。
|
||
*/
|
||
export async function runCleanup(): Promise<{
|
||
cleaned: number
|
||
failed: number
|
||
freedBytes: number
|
||
skippedActive: number
|
||
skippedProtected: number
|
||
skippedUnavailable: number
|
||
}> {
|
||
if (isCleanupRunning) {
|
||
console.log('[TempCleanup] 清理任务正在运行，跳过')
|
||
return emptyRunResult()
|
||
}
|
||
|
||
isCleanupRunning = true
|
||
const startTime = Date.now()
|
||
const result = emptyRunResult()
|
||
|
||
try {
|
||
const tempDir = getTempDir()
|
||
try {
|
||
const stats = await lstat(tempDir)
|
||
if (stats.isSymbolicLink() || !stats.isDirectory()) {
|
||
throw new Error('temp 路径不是安全目录')
|
||
}
|
||
} catch (error: any) {
|
||
if (error?.code === 'ENOENT') {
|
||
console.log('[TempCleanup] temp 目录不存在，跳过清理')
|
||
return result
|
||
}
|
||
throw error
|
||
}
|
||
|
||
const dirs = await readdir(tempDir)
|
||
const now = Date.now()
|
||
console.log(`[TempCleanup] 检查 ${dirs.length} 个目录...`)
|
||
|
||
for (const uploadId of dirs) {
|
||
if (!SAFE_UPLOAD_ID_PATTERN.test(uploadId)) {
|
||
continue
|
||
}
|
||
|
||
const dirPath = resolveStrictChildPath(tempDir, uploadId)
|
||
if (!dirPath) {
|
||
result.skippedUnavailable += 1
|
||
continue
|
||
}
|
||
|
||
try {
|
||
// 锁外只筛候选，helper 会在独占锁内重新读取并判断年龄。
|
||
const stats = await lstat(dirPath)
|
||
if (stats.isSymbolicLink() || !stats.isDirectory()) {
|
||
result.skippedUnavailable += 1
|
||
continue
|
||
}
|
||
if (now - stats.mtimeMs < MAX_AGE_MS) {
|
||
continue
|
||
}
|
||
|
||
const uploadPurpose = await readUploadPurposeCandidate(dirPath)
|
||
const cleanup = uploadPurpose === 'brand-knowledge'
|
||
? await cleanupAbandonedBrandKnowledgeUploadSession(uploadId)
|
||
: await safelyCleanupMaterialUploadSession({
|
||
uploadId,
|
||
intent: 'sweep',
|
||
minAgeMs: MAX_AGE_MS,
|
||
})
|
||
applyRunCleanupResult(result, cleanup)
|
||
} catch (error) {
|
||
result.failed += 1
|
||
console.warn(`[TempCleanup] 处理目录失败: ${uploadId}`, error)
|
||
}
|
||
}
|
||
|
||
// terminal seal 至少保留 24 小时；sealing 永不由这里自动删除。
|
||
try {
|
||
const cleanedSeals = await cleanupExpiredUploadCompletionSeals(now)
|
||
if (cleanedSeals > 0) {
|
||
console.log(`[TempCleanup] 已清理 ${cleanedSeals} 个过期上传终态标记`)
|
||
}
|
||
} catch (error) {
|
||
console.warn('[TempCleanup] 清理上传终态标记失败:', error)
|
||
}
|
||
|
||
const duration = Date.now() - startTime
|
||
const freedMB = (result.freedBytes / 1024 / 1024).toFixed(2)
|
||
console.log(
|
||
`[TempCleanup] 清理完成: 删除 ${result.cleaned} 个目录，释放 ${freedMB} MB，`
|
||
+ `跳过活跃 ${result.skippedActive} 个，保护完成态 ${result.skippedProtected} 个，`
|
||
+ `状态不确定 ${result.skippedUnavailable} 个，耗时 ${duration}ms`,
|
||
)
|
||
return result
|
||
} finally {
|
||
isCleanupRunning = false
|
||
}
|
||
}
|
||
|
||
async function cleanupAbandonedBrandKnowledgeUploadSession(
|
||
uploadId: string,
|
||
): Promise<SafeMaterialUploadCleanupResult> {
|
||
let releaseUploadSessionLock: MaterialUploadStorageLockRelease | null = null
|
||
try {
|
||
// Keep the same lock order as brand-knowledge finalization: scoped uploadId,
|
||
// then the global storage lock acquired inside executeSafeCleanup.
|
||
releaseUploadSessionLock = await acquireTaskMaterialUploadSessionLocks([uploadId])
|
||
const seal = await readUploadSessionSeal(uploadId)
|
||
if (seal?.state !== 'completed') {
|
||
return await executeSafeCleanup({
|
||
uploadId,
|
||
intent: 'sweep',
|
||
minAgeMs: MAX_AGE_MS,
|
||
})
|
||
}
|
||
const completedAt = Number(seal.completedAt)
|
||
const completedAgeMs = Date.now() - completedAt
|
||
if (!Number.isFinite(completedAt) || completedAt <= 0 || completedAgeMs < -5 * 60 * 1000) {
|
||
return unavailableResult(
|
||
'上传完成时间状态异常，已保留上传文件',
|
||
'UPLOAD_CLEANUP_COMPLETED_AT_INVALID',
|
||
)
|
||
}
|
||
if (completedAgeMs < ABANDONED_BRAND_KNOWLEDGE_RETENTION_MS) {
|
||
return skippedResult('skipped_young')
|
||
}
|
||
// Database availability is a fail-closed gate, but it does not need the
|
||
// global upload-storage lock. The scoped upload lock already prevents a
|
||
// concurrent brand-knowledge finalization for this uploadId.
|
||
await assertBrandKnowledgeDatabaseAvailable(uploadId)
|
||
return await executeSafeCleanup({
|
||
uploadId,
|
||
intent: 'sweep',
|
||
minAgeMs: ABANDONED_BRAND_KNOWLEDGE_RETENTION_MS,
|
||
capability: INTERNAL_ABANDONED_BRAND_KNOWLEDGE_CLEANUP,
|
||
})
|
||
} catch (error) {
|
||
return resultFromError(error)
|
||
} finally {
|
||
await releaseUploadSessionLock?.()
|
||
}
|
||
}
|
||
|
||
export function startCleanupScheduler(): void {
|
||
if (cleanupTimer) {
|
||
console.log('[TempCleanup] 定时任务已在运行')
|
||
return
|
||
}
|
||
|
||
console.log('[TempCleanup] 启动定时清理任务（每小时执行）')
|
||
runCleanup().catch((error) => {
|
||
console.error('[TempCleanup] 启动时清理失败:', error)
|
||
})
|
||
cleanupTimer = setInterval(() => {
|
||
runCleanup().catch((error) => {
|
||
console.error('[TempCleanup] 定时清理失败:', error)
|
||
})
|
||
}, CLEANUP_INTERVAL_MS)
|
||
cleanupTimer.unref()
|
||
}
|
||
|
||
export function stopCleanupScheduler(): void {
|
||
if (cleanupTimer) {
|
||
clearInterval(cleanupTimer)
|
||
cleanupTimer = null
|
||
console.log('[TempCleanup] 定时清理任务已停止')
|
||
}
|
||
}
|
||
|
||
async function executeSafeCleanup(
|
||
input: InternalCleanupInput,
|
||
): Promise<SafeMaterialUploadCleanupResult> {
|
||
const uploadId = String(input.uploadId || '').trim()
|
||
if (!SAFE_UPLOAD_ID_PATTERN.test(uploadId)) {
|
||
return unavailableResult('uploadId 无效', 'UPLOAD_ID_INVALID', 400)
|
||
}
|
||
|
||
let releaseLock: MaterialUploadStorageLockRelease | null = null
|
||
let directoryToDelete: UploadDirectorySnapshot | null = null
|
||
let assembledFileToDelete: string | null = null
|
||
let decision: SafeMaterialUploadCleanupResult | null = null
|
||
|
||
try {
|
||
releaseLock = await acquireMaterialUploadStorageLock('exclusive')
|
||
releaseLock.assertHeld()
|
||
|
||
const evidence = await readCleanupEvidence(uploadId)
|
||
releaseLock.assertHeld()
|
||
|
||
decision = decideCleanup(input, evidence)
|
||
if (!decision) {
|
||
const isConsumedCleanup = input.capability === INTERNAL_CONSUMED_CLEANUP
|
||
|| input.capability === INTERNAL_ABANDONED_BRAND_KNOWLEDGE_CLEANUP
|
||
if (!isConsumedCleanup) {
|
||
const tombstone = await cancelUploadSessionSeal(uploadId)
|
||
if (tombstone.state === 'completed') {
|
||
decision = protectedResult('protected_completed')
|
||
}
|
||
releaseLock.assertHeld()
|
||
}
|
||
|
||
if (!decision) {
|
||
// Redis 必须先成功删除，再同步移除当前进程内存副本。Redis 异常时内存和
|
||
// 目录都保持不动；cancelled tombstone 留下用于阻止迟到 chunk/complete。
|
||
await deleteSessionStateFailClosed(uploadId)
|
||
releaseLock.assertHeld()
|
||
directoryToDelete = evidence.directory
|
||
assembledFileToDelete = isConsumedCleanup ? evidence.assembledFilePath : null
|
||
}
|
||
}
|
||
} catch (error) {
|
||
console.warn('[TempCleanup] 安全清理状态确认失败，已保留目录:', {
|
||
uploadId,
|
||
error,
|
||
})
|
||
decision = resultFromError(error)
|
||
} finally {
|
||
if (releaseLock) {
|
||
try {
|
||
await releaseLock()
|
||
} catch (error) {
|
||
console.warn('[TempCleanup] 释放上传存储锁失败，已保留目录:', {
|
||
uploadId,
|
||
error,
|
||
})
|
||
directoryToDelete = null
|
||
assembledFileToDelete = null
|
||
decision ||= resultFromError(error)
|
||
}
|
||
}
|
||
}
|
||
|
||
if (decision) {
|
||
return decision
|
||
}
|
||
if (!directoryToDelete?.exists && !assembledFileToDelete) {
|
||
return successResult('absent', 0, 0)
|
||
}
|
||
|
||
try {
|
||
if (assembledFileToDelete) {
|
||
await rm(assembledFileToDelete, { force: true })
|
||
}
|
||
if (directoryToDelete?.exists) {
|
||
await rm(directoryToDelete.path, { recursive: true, force: true })
|
||
}
|
||
console.log(`[TempCleanup] 已安全清理上传会话: ${uploadId}`)
|
||
return successResult(
|
||
'deleted',
|
||
(directoryToDelete?.deletedFiles || 0) + (assembledFileToDelete ? 1 : 0),
|
||
directoryToDelete?.freedSpace || 0,
|
||
)
|
||
} catch (error) {
|
||
console.warn('[TempCleanup] tombstone 已写入，但目录删除失败并已保留:', {
|
||
uploadId,
|
||
error,
|
||
})
|
||
return resultFromError(error)
|
||
}
|
||
}
|
||
|
||
function decideCleanup(
|
||
input: InternalCleanupInput,
|
||
evidence: CleanupEvidence,
|
||
): SafeMaterialUploadCleanupResult | null {
|
||
const expectedOwnerId = String(input.expectedOwnerId || '').trim()
|
||
if (expectedOwnerId && evidence.ownerId !== expectedOwnerId) {
|
||
return skippedResult(
|
||
'owner_mismatch',
|
||
'上传会话不存在或不属于当前用户',
|
||
'UPLOAD_SESSION_NOT_OWNED',
|
||
404,
|
||
)
|
||
}
|
||
|
||
const isConsumedCleanup = input.capability === INTERNAL_CONSUMED_CLEANUP
|
||
const isAbandonedBrandKnowledgeCleanup = input.capability === INTERNAL_ABANDONED_BRAND_KNOWLEDGE_CLEANUP
|
||
if (isAbandonedBrandKnowledgeCleanup) {
|
||
if (evidence.uploadPurpose !== 'brand-knowledge') {
|
||
return skippedResult('not_abandoned_brand_knowledge')
|
||
}
|
||
if (
|
||
evidence.sealState !== 'completed'
|
||
|| evidence.ageMs === null
|
||
|| evidence.ageMs < ABANDONED_BRAND_KNOWLEDGE_RETENTION_MS
|
||
) {
|
||
if (evidence.sealState === 'completed') {
|
||
return skippedResult('skipped_young')
|
||
}
|
||
if (evidence.sealState === 'sealing') {
|
||
return protectedResult('protected_sealing')
|
||
}
|
||
return skippedResult(
|
||
'not_consumed',
|
||
'未确认 completed 终态，已保留上传临时目录',
|
||
'UPLOAD_NOT_CONFIRMED_COMPLETED',
|
||
409,
|
||
)
|
||
}
|
||
return null
|
||
}
|
||
if (isConsumedCleanup) {
|
||
if (
|
||
!evidence.directory.exists
|
||
&& evidence.sessions.length === 0
|
||
) {
|
||
return successResult('absent', 0, 0)
|
||
}
|
||
if (evidence.sealState !== 'completed') {
|
||
return skippedResult(
|
||
'not_consumed',
|
||
'未确认 completed 终态，已保留上传临时目录',
|
||
'UPLOAD_NOT_CONFIRMED_CONSUMED',
|
||
409,
|
||
)
|
||
}
|
||
return null
|
||
}
|
||
|
||
return decideCancellation(input, evidence)
|
||
}
|
||
|
||
function decideCancellation(
|
||
input: SafeMaterialUploadCleanupInput,
|
||
evidence: CleanupEvidence,
|
||
): SafeMaterialUploadCleanupResult | null {
|
||
const sealState = evidence.sealState
|
||
if (sealState === 'completed') {
|
||
return protectedResult('protected_completed')
|
||
}
|
||
if (
|
||
sealState === 'sealing'
|
||
&& (input.intent === 'sweep' || evidence.hasValidMerged)
|
||
) {
|
||
return protectedResult('protected_sealing')
|
||
}
|
||
if (!sealState && evidence.hasValidMerged) {
|
||
return protectedResult('protected_sealing')
|
||
}
|
||
|
||
if (input.intent === 'sweep') {
|
||
if (evidence.activeSession) {
|
||
return skippedResult('skipped_active')
|
||
}
|
||
const minAgeMs = input.minAgeMs ?? MAX_AGE_MS
|
||
const isOldEnough = evidence.ageMs !== null && evidence.ageMs >= minAgeMs
|
||
if (!isOldEnough && !(input.allowYoungOrphaned && !evidence.activeSession)) {
|
||
return skippedResult('skipped_young')
|
||
}
|
||
}
|
||
|
||
if (
|
||
!evidence.directory.exists
|
||
&& evidence.sessions.length === 0
|
||
&& !evidence.ownerId
|
||
) {
|
||
return successResult('absent', 0, 0)
|
||
}
|
||
return null
|
||
}
|
||
|
||
async function readCleanupEvidence(uploadId: string): Promise<CleanupEvidence> {
|
||
const directory = await readUploadDirectory(uploadId)
|
||
const sessions = await readSessionEvidenceFailClosed(uploadId)
|
||
const ownerIds = new Set<string>()
|
||
if (directory.metadata?.userId) {
|
||
ownerIds.add(directory.metadata.userId)
|
||
}
|
||
sessions.forEach((session) => {
|
||
if (session.userId) {
|
||
ownerIds.add(session.userId)
|
||
}
|
||
})
|
||
if (ownerIds.size > 1) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传会话 owner 状态冲突，已保留数据',
|
||
503,
|
||
'UPLOAD_CLEANUP_OWNER_CONFLICT',
|
||
)
|
||
}
|
||
|
||
const fileSizes = new Set<number>()
|
||
if (directory.metadata?.fileSize) {
|
||
fileSizes.add(directory.metadata.fileSize)
|
||
}
|
||
sessions.forEach((session) => {
|
||
if (session.fileSize) {
|
||
fileSizes.add(session.fileSize)
|
||
}
|
||
})
|
||
if (fileSizes.size > 1) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传会话文件大小状态冲突，已保留数据',
|
||
503,
|
||
'UPLOAD_CLEANUP_SESSION_CONFLICT',
|
||
)
|
||
}
|
||
|
||
const fileNames = new Set<string>()
|
||
if (directory.metadata?.fileName) {
|
||
fileNames.add(directory.metadata.fileName)
|
||
}
|
||
sessions.forEach((session) => fileNames.add(session.fileName))
|
||
if (fileNames.size > 1) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传会话文件名状态冲突，已保留数据',
|
||
503,
|
||
'UPLOAD_CLEANUP_SESSION_CONFLICT',
|
||
)
|
||
}
|
||
|
||
const uploadPurposes = new Set<string>()
|
||
if (directory.metadata?.uploadPurpose) {
|
||
uploadPurposes.add(directory.metadata.uploadPurpose)
|
||
}
|
||
sessions.forEach((session) => {
|
||
if (session.uploadPurpose) uploadPurposes.add(session.uploadPurpose)
|
||
})
|
||
if (uploadPurposes.size > 1) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传会话用途状态冲突，已保留数据',
|
||
503,
|
||
'UPLOAD_CLEANUP_SESSION_CONFLICT',
|
||
)
|
||
}
|
||
|
||
const expectedFileSize = fileSizes.values().next().value as number | undefined
|
||
const hasValidTempMerged = directory.files.some((file) => (
|
||
file.name.startsWith('merged-')
|
||
&& file.size > 0
|
||
&& (expectedFileSize === undefined || file.size === expectedFileSize)
|
||
))
|
||
const safeFileName = fileNames.values().next().value as string | undefined
|
||
const assembledFilePath = safeFileName
|
||
? await inspectAssembledFile(uploadId, safeFileName, expectedFileSize)
|
||
: null
|
||
const createdAt = directory.metadata?.createdAt ?? directory.mtimeMs
|
||
const ageMs = createdAt === null ? null : Date.now() - createdAt
|
||
if (ageMs !== null && (!Number.isFinite(ageMs) || ageMs < -5 * 60 * 1000)) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传目录时间状态异常，已保留数据',
|
||
503,
|
||
'UPLOAD_CLEANUP_AGE_INVALID',
|
||
)
|
||
}
|
||
|
||
const seal = await readUploadSessionSeal(uploadId)
|
||
const uploadPurpose = uploadPurposes.values().next().value as string | undefined
|
||
return {
|
||
directory,
|
||
sessions,
|
||
ownerId: ownerIds.values().next().value as string | undefined || null,
|
||
expectedFileSize: expectedFileSize ?? null,
|
||
activeSession: sessions.length > 0,
|
||
ageMs,
|
||
hasValidMerged: hasValidTempMerged || assembledFilePath !== null,
|
||
sealState: seal?.state ?? null,
|
||
assembledFilePath,
|
||
uploadPurpose: uploadPurpose || null,
|
||
}
|
||
}
|
||
|
||
async function readUploadDirectory(uploadId: string): Promise<UploadDirectorySnapshot> {
|
||
const tempDir = getTempDir()
|
||
const dirPath = resolveStrictChildPath(tempDir, uploadId)
|
||
if (!dirPath) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传临时目录路径无效',
|
||
503,
|
||
'UPLOAD_CLEANUP_PATH_INVALID',
|
||
)
|
||
}
|
||
|
||
let directoryStats: Awaited<ReturnType<typeof lstat>>
|
||
try {
|
||
directoryStats = await lstat(dirPath)
|
||
} catch (error: any) {
|
||
if (error?.code === 'ENOENT') {
|
||
return {
|
||
path: dirPath,
|
||
exists: false,
|
||
mtimeMs: null,
|
||
metadata: null,
|
||
files: [],
|
||
deletedFiles: 0,
|
||
freedSpace: 0,
|
||
}
|
||
}
|
||
throw error
|
||
}
|
||
if (directoryStats.isSymbolicLink() || !directoryStats.isDirectory()) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传临时目录类型异常，已保留数据',
|
||
503,
|
||
'UPLOAD_CLEANUP_PATH_INVALID',
|
||
)
|
||
}
|
||
|
||
const names = await readdir(dirPath)
|
||
const files: Array<{ name: string; size: number }> = []
|
||
let freedSpace = 0
|
||
for (const name of names) {
|
||
const filePath = resolveStrictChildPath(dirPath, name)
|
||
if (!filePath) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传临时文件路径异常，已保留数据',
|
||
503,
|
||
'UPLOAD_CLEANUP_PATH_INVALID',
|
||
)
|
||
}
|
||
const stats = await lstat(filePath)
|
||
if (stats.isSymbolicLink() || !stats.isFile()) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传临时目录包含未知文件类型，已保留数据',
|
||
503,
|
||
'UPLOAD_CLEANUP_FILE_INVALID',
|
||
)
|
||
}
|
||
files.push({ name, size: stats.size })
|
||
freedSpace += stats.size
|
||
}
|
||
|
||
const metadata = names.includes('meta.json')
|
||
? await readMetadataFailClosed(path.join(dirPath, 'meta.json'))
|
||
: null
|
||
return {
|
||
path: dirPath,
|
||
exists: true,
|
||
mtimeMs: directoryStats.mtimeMs,
|
||
metadata,
|
||
files,
|
||
deletedFiles: files.length,
|
||
freedSpace,
|
||
}
|
||
}
|
||
|
||
async function readMetadataFailClosed(metadataPath: string): Promise<{
|
||
userId: string
|
||
fileName: string
|
||
fileSize: number
|
||
createdAt: number
|
||
uploadPurpose: string | null
|
||
}> {
|
||
try {
|
||
const parsed = JSON.parse(await readFile(metadataPath, 'utf8')) as Record<string, unknown>
|
||
const userId = typeof parsed.userId === 'string' ? parsed.userId.trim() : ''
|
||
const fileName = normalizeSafeFileName(
|
||
typeof parsed.fileName === 'string' ? parsed.fileName : '',
|
||
)
|
||
const fileSize = Number(parsed.fileSize)
|
||
const createdAt = Number(parsed.createdAt)
|
||
const uploadPurpose = typeof parsed.uploadPurpose === 'string'
|
||
? parsed.uploadPurpose.trim()
|
||
: ''
|
||
if (
|
||
!userId
|
||
|| !fileName
|
||
|| !Number.isFinite(fileSize)
|
||
|| fileSize <= 0
|
||
|| !Number.isFinite(createdAt)
|
||
|| createdAt <= 0
|
||
|| (uploadPurpose && uploadPurpose !== 'brand-knowledge')
|
||
) {
|
||
throw new Error('meta.json 字段无效')
|
||
}
|
||
return { userId, fileName, fileSize, createdAt, uploadPurpose: uploadPurpose || null }
|
||
} catch (error) {
|
||
throw new MaterialUploadRequestError(
|
||
`无法确认上传目录元数据，已保留数据: ${error instanceof Error ? error.message : '未知错误'}`,
|
||
503,
|
||
'UPLOAD_CLEANUP_METADATA_UNAVAILABLE',
|
||
)
|
||
}
|
||
}
|
||
|
||
async function readSessionEvidenceFailClosed(
|
||
uploadId: string,
|
||
): Promise<StoredSessionEvidence[]> {
|
||
const sessions: StoredSessionEvidence[] = []
|
||
const memorySession = uploadSessions.get(uploadId)
|
||
if (memorySession) {
|
||
sessions.push(normalizeSessionEvidence(memorySession, 'memory'))
|
||
}
|
||
|
||
if (process.env.USE_REDIS === 'true') {
|
||
try {
|
||
const { redis } = await import('./upload-sessions-redis')
|
||
const raw = await redis.get(`${REDIS_SESSION_KEY_PREFIX}${uploadId}`)
|
||
if (raw) {
|
||
const parsed = JSON.parse(raw) as UploadSession
|
||
sessions.push(normalizeSessionEvidence(parsed, 'redis'))
|
||
}
|
||
} catch (error) {
|
||
throw new MaterialUploadRequestError(
|
||
`无法确认 Redis 上传会话，已保留数据: ${error instanceof Error ? error.message : '未知错误'}`,
|
||
503,
|
||
'UPLOAD_CLEANUP_REDIS_UNAVAILABLE',
|
||
)
|
||
}
|
||
}
|
||
return sessions
|
||
}
|
||
|
||
function normalizeSessionEvidence(
|
||
session: Partial<UploadSession>,
|
||
source: StoredSessionEvidence['source'],
|
||
): StoredSessionEvidence {
|
||
const userId = typeof session.userId === 'string' ? session.userId.trim() : ''
|
||
const fileName = normalizeSafeFileName(
|
||
typeof session.fileName === 'string' ? session.fileName : '',
|
||
)
|
||
const fileSize = Number(session.fileSize)
|
||
const createdAt = Number(session.createdAt)
|
||
const uploadPurpose = typeof session.uploadPurpose === 'string'
|
||
? session.uploadPurpose.trim()
|
||
: ''
|
||
if (
|
||
!userId
|
||
|| !fileName
|
||
|| !Number.isFinite(fileSize)
|
||
|| fileSize <= 0
|
||
|| !Number.isFinite(createdAt)
|
||
|| createdAt <= 0
|
||
|| (uploadPurpose && uploadPurpose !== 'brand-knowledge')
|
||
) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传会话内容异常，已保留数据',
|
||
503,
|
||
'UPLOAD_CLEANUP_SESSION_INVALID',
|
||
)
|
||
}
|
||
return { source, userId, fileName, fileSize, createdAt, uploadPurpose: uploadPurpose || null }
|
||
}
|
||
|
||
async function readUploadPurposeCandidate(dirPath: string): Promise<string | null> {
|
||
try {
|
||
const metadataPath = resolveStrictChildPath(dirPath, 'meta.json')
|
||
if (!metadataPath) return null
|
||
const stats = await lstat(metadataPath)
|
||
if (stats.isSymbolicLink() || !stats.isFile() || stats.size > 16 * 1024) return null
|
||
const parsed = JSON.parse(await readFile(metadataPath, 'utf8')) as Record<string, unknown>
|
||
return typeof parsed.uploadPurpose === 'string' ? parsed.uploadPurpose.trim() || null : null
|
||
} catch {
|
||
// The locked cleanup path will parse the same metadata fail-closed.
|
||
return null
|
||
}
|
||
}
|
||
|
||
async function assertBrandKnowledgeDatabaseAvailable(uploadId: string): Promise<void> {
|
||
try {
|
||
const { prisma } = await import('./prisma')
|
||
await prisma.brandKnowledge.findFirst({
|
||
where: {
|
||
filePath: { startsWith: `uploads/brand-knowledge/source-${uploadId}.` },
|
||
},
|
||
select: { id: true },
|
||
})
|
||
} catch (error) {
|
||
throw new MaterialUploadRequestError(
|
||
`无法确认品牌知识建档状态，已保留上传文件: ${error instanceof Error ? error.message : '未知错误'}`,
|
||
503,
|
||
'UPLOAD_CLEANUP_DATABASE_UNAVAILABLE',
|
||
)
|
||
}
|
||
}
|
||
|
||
async function inspectAssembledFile(
|
||
uploadId: string,
|
||
fileName: string,
|
||
expectedFileSize?: number,
|
||
): Promise<string | null> {
|
||
const assembledDir = resolveStrictChildPath(getUploadRootDir(), 'assembled')
|
||
const assembledPath = assembledDir
|
||
? resolveStrictChildPath(assembledDir, `${uploadId}-${fileName}`)
|
||
: null
|
||
if (!assembledPath) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传合并文件路径异常，已保留数据',
|
||
503,
|
||
'UPLOAD_CLEANUP_PATH_INVALID',
|
||
)
|
||
}
|
||
try {
|
||
const stats = await lstat(assembledPath)
|
||
if (
|
||
stats.isSymbolicLink()
|
||
|| !stats.isFile()
|
||
|| stats.size <= 0
|
||
|| (expectedFileSize !== undefined && stats.size !== expectedFileSize)
|
||
) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传合并文件状态异常，已保留数据',
|
||
503,
|
||
'UPLOAD_CLEANUP_MERGED_INVALID',
|
||
)
|
||
}
|
||
return assembledPath
|
||
} catch (error: any) {
|
||
if (error?.code === 'ENOENT') {
|
||
return null
|
||
}
|
||
throw error
|
||
}
|
||
}
|
||
|
||
async function deleteSessionStateFailClosed(uploadId: string): Promise<void> {
|
||
if (process.env.USE_REDIS === 'true') {
|
||
try {
|
||
const { redis } = await import('./upload-sessions-redis')
|
||
await redis.del(`${REDIS_SESSION_KEY_PREFIX}${uploadId}`)
|
||
} catch (error) {
|
||
throw new MaterialUploadRequestError(
|
||
`无法删除 Redis 上传会话，已保留目录: ${error instanceof Error ? error.message : '未知错误'}`,
|
||
503,
|
||
'UPLOAD_CLEANUP_REDIS_UNAVAILABLE',
|
||
)
|
||
}
|
||
}
|
||
uploadSessions.delete(uploadId)
|
||
}
|
||
|
||
function applyRunCleanupResult(
|
||
run: ReturnType<typeof emptyRunResult>,
|
||
cleanup: SafeMaterialUploadCleanupResult,
|
||
): void {
|
||
switch (cleanup.status) {
|
||
case 'deleted':
|
||
run.cleaned += 1
|
||
run.freedBytes += cleanup.freedSpace
|
||
break
|
||
case 'skipped_active':
|
||
run.skippedActive += 1
|
||
break
|
||
case 'protected_completed':
|
||
case 'protected_sealing':
|
||
run.skippedProtected += 1
|
||
break
|
||
case 'unavailable':
|
||
run.skippedUnavailable += 1
|
||
break
|
||
case 'absent':
|
||
case 'skipped_young':
|
||
case 'not_consumed':
|
||
case 'not_abandoned_brand_knowledge':
|
||
case 'owner_mismatch':
|
||
break
|
||
default:
|
||
run.failed += 1
|
||
}
|
||
}
|
||
|
||
function emptyRunResult() {
|
||
return {
|
||
cleaned: 0,
|
||
failed: 0,
|
||
freedBytes: 0,
|
||
skippedActive: 0,
|
||
skippedProtected: 0,
|
||
skippedUnavailable: 0,
|
||
}
|
||
}
|
||
|
||
function getTempDir(): string {
|
||
return path.join(getUploadRootDir(), 'temp')
|
||
}
|
||
|
||
function successResult(
|
||
status: 'deleted' | 'absent',
|
||
deletedFiles: number,
|
||
freedSpace: number,
|
||
): SafeMaterialUploadCleanupResult {
|
||
return { status, success: true, deletedFiles, freedSpace }
|
||
}
|
||
|
||
function protectedResult(
|
||
status: 'protected_completed' | 'protected_sealing',
|
||
): SafeMaterialUploadCleanupResult {
|
||
return skippedResult(
|
||
status,
|
||
status === 'protected_completed'
|
||
? '上传会话已经完成，公开清理不得删除合并文件'
|
||
: '上传会话正在完成且已有有效合并文件，已保留数据',
|
||
status === 'protected_completed'
|
||
? 'UPLOAD_SESSION_COMPLETED'
|
||
: 'UPLOAD_SESSION_SEALING',
|
||
409,
|
||
)
|
||
}
|
||
|
||
function skippedResult(
|
||
status: Exclude<SafeMaterialUploadCleanupStatus, 'deleted' | 'absent' | 'unavailable'>,
|
||
error?: string,
|
||
code?: string,
|
||
httpStatus?: number,
|
||
): SafeMaterialUploadCleanupResult {
|
||
return {
|
||
status,
|
||
success: false,
|
||
deletedFiles: 0,
|
||
freedSpace: 0,
|
||
error,
|
||
code,
|
||
httpStatus,
|
||
}
|
||
}
|
||
|
||
function unavailableResult(
|
||
error: string,
|
||
code = 'UPLOAD_CLEANUP_STATE_UNAVAILABLE',
|
||
httpStatus = 503,
|
||
): SafeMaterialUploadCleanupResult {
|
||
return {
|
||
status: 'unavailable',
|
||
success: false,
|
||
deletedFiles: 0,
|
||
freedSpace: 0,
|
||
error,
|
||
code,
|
||
httpStatus,
|
||
}
|
||
}
|
||
|
||
function resultFromError(error: unknown): SafeMaterialUploadCleanupResult {
|
||
if (error instanceof MaterialUploadRequestError) {
|
||
return unavailableResult(error.message, error.code, error.status)
|
||
}
|
||
return unavailableResult(
|
||
error instanceof Error ? error.message : '无法确认清理状态，已保留数据',
|
||
)
|
||
}
|