import { randomUUID } from 'crypto' import { mkdir, open, rename, rm } from 'fs/promises' import path from 'path' import { NextRequest, NextResponse } from 'next/server' import { getAuthenticatedUploadUserId } from '@/lib/upload-auth' import { getUploadSession, updateChunkWithStats } from '@/lib/upload-sessions' import { getExpectedUploadChunkBytes, MAX_UPLOAD_CHUNK_BYTES, resolveUploadChunkSize, } from '@/lib/material-upload-chunk-policy' import { MaterialUploadRequestError, materialUploadChunkRequestLimiter, } from '@/lib/material-upload-request-guard' import { acquireMaterialUploadStorageLock, type MaterialUploadStorageLockRelease, } from '@/lib/material-upload-cross-process-lock' import { materialUploadStorageReservations } from '@/lib/material-upload-storage-reservation' import { assertUploadSessionAcceptingChunks } from '@/lib/material-upload-session-seal' export const runtime = 'nodejs' export const dynamic = 'force-dynamic' export const maxDuration = 10800 const MEBIBYTE = 1024 * 1024 const MAX_CHUNK_BYTES = MAX_UPLOAD_CHUNK_BYTES const MAX_MULTIPART_CHUNK_REQUEST_BYTES = 21 * MEBIBYTE class ChunkUploadRequestError extends Error { constructor( message: string, public readonly status: number, public readonly code: string, ) { super(message) this.name = 'ChunkUploadRequestError' } } interface ChunkRequestPayload { uploadId: string chunkIndex: number fileName: string chunkSizeHint: number mode: 'raw' | 'multipart' chunkFile?: File | null } export async function POST(req: NextRequest) { const requestStartTime = Date.now() let releaseChunkSlot: (() => void) | null = null let releaseCrossProcessLock: MaterialUploadStorageLockRelease | null = null let releaseStorageReservation: (() => void) | null = null let pendingChunkPath: string | null = null console.log(`[UploadChunk] 📥 收到请求 at ${new Date().toISOString()}`) try { const authStartedAt = Date.now() const userId = await getAuthenticatedUploadUserId(req) const authTime = Date.now() - authStartedAt console.log(`[UploadChunk] 🔐 认证完成: ${authTime}ms`) if (!userId) { return NextResponse.json({ error: '请先登录' }, { status: 401 }) } validateChunkRequestHeaders(req) // multipart 的 formData() 会完整解析请求体,必须先进入并发槽,不能让 // 任意数量的 21MiB 请求在限流之外同时占用 Node 内存。 releaseChunkSlot = await materialUploadChunkRequestLimiter.acquire() const payloadStartedAt = Date.now() const payload = await resolveChunkPayload(req) const payloadResolveTime = Date.now() - payloadStartedAt console.log( `[UploadChunk] 📦 请求参数解析完成: ${payloadResolveTime}ms, mode=${payload.mode}, chunkIndex=${payload.chunkIndex}`, ) if (!payload.uploadId || Number.isNaN(payload.chunkIndex)) { return NextResponse.json({ error: '缺少必要参数' }, { status: 400 }) } if (!Number.isSafeInteger(payload.chunkSizeHint) || payload.chunkSizeHint <= 0) { return NextResponse.json({ error: '上传分片大小无效' }, { status: 400 }) } if (payload.chunkSizeHint > MAX_CHUNK_BYTES) { return NextResponse.json( { error: '单个上传分片不能超过 20MB', code: 'CHUNK_TOO_LARGE' }, { status: 413 }, ) } const sessionStartedAt = Date.now() const sessionData = await getUploadSession(payload.uploadId) const getSessionTime = Date.now() - sessionStartedAt console.log(`[UploadChunk] 🔍 获取会话完成: ${getSessionTime}ms, 会话存在: ${!!sessionData}`) if (!sessionData) { console.log(`[UploadChunk] ❌ 会话不存在: ${payload.uploadId}`) return NextResponse.json({ error: '上传会话不存在或已过期' }, { status: 404 }) } if (sessionData.userId !== userId) { return NextResponse.json({ error: '无权访问此上传会话' }, { status: 403 }) } // 已经完成/正在完成的会话不能重新接收分片。这里先快速拒绝;慢网络写完后 // 还会在跨容器共享锁内复查一次,封住 complete 与迟到请求的竞态窗口。 await assertUploadSessionAcceptingChunks(payload.uploadId) if (payload.chunkIndex < 0 || payload.chunkIndex >= sessionData.totalChunks) { console.error( `[UploadChunk] ❌ chunkIndex超出范围: ${payload.chunkIndex}, 总数: ${sessionData.totalChunks}`, ) return NextResponse.json( { error: `chunkIndex超出范围: ${payload.chunkIndex}/${sessionData.totalChunks}` }, { status: 400 }, ) } const sessionChunkSize = resolveUploadChunkSize( sessionData.fileSize, sessionData.totalChunks, sessionData.chunkSize, ) const expectedChunkBytes = sessionChunkSize ? getExpectedUploadChunkBytes({ fileSize: sessionData.fileSize, totalChunks: sessionData.totalChunks, chunkSize: sessionChunkSize, chunkIndex: payload.chunkIndex, }) : null if (!expectedChunkBytes) { return NextResponse.json( { error: '上传会话的分片布局无效,请重新开始上传', code: 'CHUNK_LAYOUT_INVALID' }, { status: 409 }, ) } if (payload.chunkSizeHint !== expectedChunkBytes) { return NextResponse.json( { error: '上传分片大小与会话声明不一致', code: 'CHUNK_SIZE_MISMATCH' }, { status: 400 }, ) } const wasChunkAlreadyRecorded = sessionData.chunks.has(payload.chunkIndex) const nextUploadedChunkCount = wasChunkAlreadyRecorded ? sessionData.chunks.size : sessionData.chunks.size + 1 const mkdirStartedAt = Date.now() const tempDir = path.join(process.env.UPLOAD_DIR || './uploads', 'temp', payload.uploadId) await mkdir(tempDir, { recursive: true }) const mkdirTime = Date.now() - mkdirStartedAt releaseStorageReservation = await materialUploadStorageReservations.acquire( tempDir, expectedChunkBytes, ) const chunkFilePath = path.join(tempDir, `chunk-${payload.chunkIndex}`) pendingChunkPath = `${chunkFilePath}.partial-${randomUUID()}` const writeStartedAt = Date.now() const writtenBytes = payload.mode === 'raw' ? await writeRawRequestBodyToFile(req, pendingChunkPath, MAX_CHUNK_BYTES) : await writeMultipartChunkToFile(payload.chunkFile, pendingChunkPath, MAX_CHUNK_BYTES) if (writtenBytes !== payload.chunkSizeHint) { throw new ChunkUploadRequestError( '上传分片实际大小与声明不一致', 400, 'CHUNK_SIZE_MISMATCH', ) } // 慢公网接收期间只写唯一 partial,不持有全局共享锁,避免一个 10/20MiB // 弱网分片让所有用户的 complete 长时间饥饿。写盘完成后才用短临界区 // 发布 chunk-N;跨容器在途分片的最大体量由 20MiB*并发上限约束,且 // statfs 始终额外保留 2GiB,不依赖这把锁承担网络阶段的空间预留。 releaseCrossProcessLock = await acquireMaterialUploadStorageLock('shared') releaseCrossProcessLock.assertHeld() await assertUploadSessionAcceptingChunks(payload.uploadId) await rename(pendingChunkPath, chunkFilePath) pendingChunkPath = null const writeTime = Date.now() - writeStartedAt const statsStartedAt = Date.now() await updateChunkWithStats(payload.uploadId, payload.chunkIndex, chunkFilePath, writtenBytes) const updateChunkTime = Date.now() - statsStartedAt const totalTime = Date.now() - requestStartTime console.log(`[UploadChunk] ⏱️ Chunk ${payload.chunkIndex}/${sessionData.totalChunks} 总耗时: ${totalTime}ms`, { 认证: `${authTime}ms (${percent(authTime, totalTime)}%)`, 请求参数解析: `${payloadResolveTime}ms (${percent(payloadResolveTime, totalTime)}%)`, 获取会话: `${getSessionTime}ms (${percent(getSessionTime, totalTime)}%)`, 创建目录: `${mkdirTime}ms (${percent(mkdirTime, totalTime)}%)`, 流式写盘: `${writeTime}ms (${percent(writeTime, totalTime)}%)`, 更新会话: `${updateChunkTime}ms (${percent(updateChunkTime, totalTime)}%)`, 大小: `${(writtenBytes / 1024 / 1024).toFixed(2)}MB`, 模式: payload.mode, }) if (writeTime / totalTime > 0.8) { console.log( `[UploadChunk] ⚠️ 流式写盘阶段占比 ${percent(writeTime, totalTime)}%,当前主要瓶颈仍在网络接收`, ) } return NextResponse.json({ success: true, uploadedChunks: nextUploadedChunkCount, totalChunks: sessionData.totalChunks, serverProcessTime: totalTime, receivedBytes: writtenBytes, mode: payload.mode, }) } catch (error) { if (pendingChunkPath) { await rm(pendingChunkPath, { force: true }).catch(() => undefined) pendingChunkPath = null } if (error instanceof MaterialUploadRequestError) { return NextResponse.json( { error: error.message, code: error.code }, { status: error.status, headers: error.retryAfterSeconds ? { 'Retry-After': String(error.retryAfterSeconds) } : undefined, }, ) } if (error instanceof ChunkUploadRequestError) { return NextResponse.json( { error: error.message, code: error.code }, { status: error.status }, ) } console.error('[UploadChunk] Error:', error) return NextResponse.json({ error: '上传块失败' }, { status: 500 }) } finally { releaseStorageReservation?.() await releaseCrossProcessLock?.() releaseChunkSlot?.() } } function validateChunkRequestHeaders(req: NextRequest): void { const contentType = (req.headers.get('content-type') || '').toLowerCase() const isMultipart = contentType.includes('multipart/form-data') const rawContentLength = req.headers.get('content-length') if (!rawContentLength || !/^\d+$/.test(rawContentLength.trim())) { throw new ChunkUploadRequestError( '上传分片缺少有效的 Content-Length', 411, 'CONTENT_LENGTH_REQUIRED', ) } const contentLength = Number(rawContentLength) const requestLimit = isMultipart ? MAX_MULTIPART_CHUNK_REQUEST_BYTES : MAX_CHUNK_BYTES if (!Number.isSafeInteger(contentLength) || contentLength <= 0) { throw new ChunkUploadRequestError( '上传分片的 Content-Length 无效', 411, 'CONTENT_LENGTH_REQUIRED', ) } if (contentLength > requestLimit) { throw new ChunkUploadRequestError( '单个上传分片不能超过 20MB', 413, 'CHUNK_TOO_LARGE', ) } const rawHint = req.headers.get('x-chunk-size') if (rawHint) { if (!/^\d+$/.test(rawHint.trim())) { throw new ChunkUploadRequestError('X-Chunk-Size 无效', 400, 'CHUNK_SIZE_INVALID') } const hint = Number(rawHint) if (!Number.isSafeInteger(hint) || hint <= 0 || hint > MAX_CHUNK_BYTES) { throw new ChunkUploadRequestError( '单个上传分片不能超过 20MB', 413, 'CHUNK_TOO_LARGE', ) } if (!isMultipart && hint !== contentLength) { throw new ChunkUploadRequestError( 'Content-Length 与 X-Chunk-Size 不一致', 400, 'CHUNK_SIZE_MISMATCH', ) } } } async function resolveChunkPayload(req: NextRequest): Promise { const contentType = (req.headers.get('content-type') || '').toLowerCase() if (!contentType.includes('multipart/form-data')) { const uploadId = (req.headers.get('x-upload-id') || req.nextUrl.searchParams.get('uploadId') || '').trim() const chunkIndexRaw = req.headers.get('x-chunk-index') || req.nextUrl.searchParams.get('chunkIndex') || '' const fileNameRaw = req.headers.get('x-file-name') || req.nextUrl.searchParams.get('fileName') || '' const chunkSizeHint = Number( req.headers.get('x-chunk-size') || req.headers.get('content-length') || 0, ) return { uploadId, chunkIndex: Number.parseInt(chunkIndexRaw, 10), fileName: safeDecodeURIComponent(fileNameRaw), chunkSizeHint: Number.isFinite(chunkSizeHint) ? chunkSizeHint : 0, mode: 'raw', chunkFile: null, } } const formData = await req.formData() const uploadId = (formData.get('uploadId') as string) || '' const chunkIndex = Number.parseInt((formData.get('chunkIndex') as string) || '', 10) const fileName = (formData.get('fileName') as string) || '' const chunk = formData.get('chunk') as File | null return { uploadId: uploadId.trim(), chunkIndex, fileName, chunkSizeHint: Number(chunk?.size || 0), mode: 'multipart', chunkFile: chunk, } } async function writeRawRequestBodyToFile( req: NextRequest, chunkFilePath: string, maxBytes: number, ): Promise { if (!req.body) { throw new Error('请求体为空') } return writeReadableStreamToFile(req.body, chunkFilePath, maxBytes) } async function writeMultipartChunkToFile( chunk: File | null | undefined, chunkFilePath: string, maxBytes: number, ): Promise { if (!chunk) { throw new Error('缺少 chunk 文件') } if (!Number.isSafeInteger(chunk.size) || chunk.size <= 0 || chunk.size > maxBytes) { throw new ChunkUploadRequestError( '单个上传分片不能超过 20MB', 413, 'CHUNK_TOO_LARGE', ) } return writeReadableStreamToFile(chunk.stream(), chunkFilePath, maxBytes) } async function writeReadableStreamToFile( stream: ReadableStream, targetPath: string, maxBytes: number, ): Promise { const reader = stream.getReader() let fileHandle: Awaited> | null = null let writtenBytes = 0 try { // FileHandle 的 open/write/sync 都是可 await 的 Promise;磁盘满、权限和 // EIO 会回到当前 try/catch,不会像无 error listener 的 WriteStream 那样 // 触发未捕获事件并杀掉整个 Node 进程。 fileHandle = await open(targetPath, 'wx') while (true) { const { done, value } = await reader.read() if (done) { break } if (!value || value.length === 0) { continue } writtenBytes += value.length if (writtenBytes > maxBytes) { await reader.cancel('chunk too large').catch(() => undefined) throw new ChunkUploadRequestError( '单个上传分片不能超过 20MB', 413, 'CHUNK_TOO_LARGE', ) } let offset = 0 while (offset < value.length) { const { bytesWritten } = await fileHandle.write( value, offset, value.length - offset, null, ) if (bytesWritten <= 0) { throw new Error('上传分片写入失败') } offset += bytesWritten } } await fileHandle.sync() await fileHandle.close() fileHandle = null return writtenBytes } catch (error) { if (fileHandle) { await fileHandle.close().catch(() => undefined) } await rm(targetPath, { force: true }).catch(() => undefined) throw error } finally { reader.releaseLock() } } function safeDecodeURIComponent(value: string): string { if (!value) { return '' } try { return decodeURIComponent(value) } catch { return value } } function percent(part: number, total: number): string { if (total <= 0) { return '0.0' } return ((part / total) * 100).toFixed(1) }