gongxiangp2/lib/chunk-upload.ts
2026-09-16 15:28:42 +08:00

1051 lines
37 KiB
TypeScript
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

// 分块上传工具函数
// 支持断点续传,自动重试,静默处理错误
import { getOptimalChunkSize, getChunkTimeout, getOptimalConcurrency } from './chunk-upload-config'
import { resolveUploadChunkSize } from './material-upload-chunk-policy'
interface ChunkUploadOptions {
chunkSize?: number // 每块大小(字节),默认根据网络环境自动选择(内网20MB,公网10MB)
onProgress?: (progress: number, speed?: number) => void // progress: 0-100, speed: 字节/秒
onError?: (error: Error) => void
maxRetries?: number // 最大重试次数,默认5次
retryDelay?: number // 重试延迟(毫秒),默认1000ms
onStageChange?: (
stage: 'uploading' | 'processing',
payload?: {
message?: string
}
) => void
onRequest?: (xhr: XMLHttpRequest | null) => void // 暴露当前的XHR请求以便取消
skipComplete?: boolean // 如果为true,分片上传完成后立即返回,不调用completeUpload(用于后台异步处理)
retainStateOnSkipComplete?: boolean // 仅特殊批次使用:建档确认前保留 IndexedDB 断点
isCancelled?: () => boolean // 取消检查回调,返回 true 时停止上传
onActiveXhrChange?: (xhrs: Set<XMLHttpRequest>) => void // 暴露所有活跃的XHR请求集合(用于批量取消)
requestHeaders?: Record<string, string> // 供钉钉等独立登录态透传认证头
chunkConcurrency?: number // 单次文件上传的并发块数;不传时继续按网络环境自动选择
}
interface ChunkUploadState {
fileId: string // 文件唯一ID(基于文件内容hash)
fileName: string
fileSize: number
chunkSize: number
totalChunks: number
uploadedChunks: Set<number> // 已上传的块索引
uploadId?: string // 服务器返回的上传ID
metadata?: Record<string, any> // 表单元数据(用于恢复时预填)
}
interface UploadHttpError extends Error {
status?: number
retryable?: boolean
outcomeUncertain?: boolean
}
function createUploadHttpError(message: string, status?: number, retryable: boolean = true): UploadHttpError {
const error = new Error(message) as UploadHttpError
error.status = status
error.retryable = retryable
return error
}
function createUncertainUploadError(message: string): UploadHttpError {
const error = createUploadHttpError(message, undefined, true)
error.outcomeUncertain = true
return error
}
function isUploadOutcomeUncertain(error: unknown): boolean {
return Boolean(
error
&& typeof error === 'object'
&& 'outcomeUncertain' in error
&& (error as UploadHttpError).outcomeUncertain,
)
}
function isRetryableUploadError(error: unknown): boolean {
if (error && typeof error === 'object' && 'retryable' in error) {
return Boolean((error as UploadHttpError).retryable)
}
return true
}
function isNonRetryableUploadError(error: unknown): boolean {
return !isRetryableUploadError(error)
}
function isRetryableHttpStatus(status: number): boolean {
return status >= 500 || status === 408 || status === 429
}
function getUploadHttpErrorMessage(action: string, status: number, fallback: string): string {
if (status === 413) {
return `${action}失败:上传分片超过入口服务器限制(413),请联系管理员检查公网入口上传大小限制`
}
if (status === 401 || status === 403) {
return `${action}失败:登录状态已失效或无权限,请重新登录后再试`
}
return fallback
}
// 存储上传状态到IndexedDB
const DB_NAME = 'chunk-upload-db'
const STORE_NAME = 'upload-states'
async function getDB(): Promise<IDBDatabase> {
return new Promise((resolve, reject) => {
const request = indexedDB.open(DB_NAME, 1)
request.onerror = () => reject(request.error)
request.onsuccess = () => resolve(request.result)
request.onupgradeneeded = (event) => {
const db = (event.target as IDBOpenDBRequest).result
if (!db.objectStoreNames.contains(STORE_NAME)) {
db.createObjectStore(STORE_NAME, { keyPath: 'fileId' })
}
}
})
}
async function saveUploadState(state: ChunkUploadState): Promise<void> {
try {
const db = await getDB()
const transaction = db.transaction([STORE_NAME], 'readwrite')
const store = transaction.objectStore(STORE_NAME)
await new Promise<void>((resolve, reject) => {
const request = store.put(state)
request.onsuccess = () => resolve()
request.onerror = () => reject(request.error)
})
} catch (error) {
console.error('[ChunkUpload] Failed to save state:', error)
}
}
async function loadUploadState(fileId: string): Promise<ChunkUploadState | null> {
try {
const db = await getDB()
const transaction = db.transaction([STORE_NAME], 'readonly')
const store = transaction.objectStore(STORE_NAME)
return new Promise<ChunkUploadState | null>((resolve, reject) => {
const request = store.get(fileId)
request.onsuccess = () => resolve(request.result || null)
request.onerror = () => reject(request.error)
})
} catch (error) {
console.error('[ChunkUpload] Failed to load state:', error)
return null
}
}
export async function deleteUploadState(fileId: string): Promise<void> {
try {
const db = await getDB()
const transaction = db.transaction([STORE_NAME], 'readwrite')
const store = transaction.objectStore(STORE_NAME)
await new Promise<void>((resolve, reject) => {
const request = store.delete(fileId)
request.onsuccess = () => resolve()
request.onerror = () => reject(request.error)
})
} catch (error) {
console.error('[ChunkUpload] Failed to delete state:', error)
}
}
// 导出:按fileId获取上传状态(用于恢复预填)
export async function getUploadStateById(fileId: string): Promise<ChunkUploadState | null> {
return loadUploadState(fileId)
}
// 生成文件唯一ID(基于文件名、大小、修改时间)
export function generateFileId(file: File): string {
const data = `${file.name}-${file.size}-${file.lastModified}`
// 简单的稳定hash函数(不包含时间),确保同一文件生成相同ID以便恢复
let hash = 0
for (let i = 0; i < data.length; i++) {
const char = data.charCodeAt(i)
hash = ((hash << 5) - hash) + char
hash = hash & hash // Convert to 32bit integer
}
return `file-${Math.abs(hash)}`
}
// 初始化上传(获取uploadId)
async function initUpload(
fileName: string,
fileSize: number,
totalChunks: number,
chunkSize: number,
uploadPurpose: string | undefined,
requestHeaders: Record<string, string> = {},
): Promise<string> {
const response = await fetch('/api/materials/upload/init', {
method: 'POST',
headers: { 'Content-Type': 'application/json', ...requestHeaders },
body: JSON.stringify({ fileName, fileSize, totalChunks, chunkSize, uploadPurpose }),
})
if (!response.ok) {
let errorMessage = `初始化上传失败: ${response.status}`
try {
const errorData = await response.json()
errorMessage = errorData?.error || errorData?.message || errorMessage
} catch {
// 忽略错误响应解析失败,保留默认文案
}
const retryable = isRetryableHttpStatus(response.status)
throw createUploadHttpError(errorMessage, response.status, retryable)
}
const data = await response.json()
return data.uploadId
}
// 上传单个块
async function uploadChunk(
uploadId: string,
chunkIndex: number,
chunk: Blob,
fileName: string,
options: {
onChunkProgress?: (loaded: number, total: number) => void
onRequest?: (xhr: XMLHttpRequest | null) => void
requestHeaders?: Record<string, string>
} = {}
): Promise<void> {
const startTime = Date.now()
let uploadStartTime = 0
let firstByteTime = 0
let lastProgressTime = 0
let lastProgressLoaded = 0
return new Promise((resolve, reject) => {
const xhr = new XMLHttpRequest()
// 设置超时时间:根据分片大小动态调整
// 10MB在500KB/s下需要20秒,20MB需要40秒,加上50%缓冲
xhr.timeout = getChunkTimeout(chunk.size)
options.onRequest?.(xhr)
xhr.open('POST', '/api/materials/upload/chunk')
xhr.setRequestHeader('Content-Type', 'application/octet-stream')
xhr.setRequestHeader('X-Upload-Id', uploadId)
xhr.setRequestHeader('X-Chunk-Index', chunkIndex.toString())
xhr.setRequestHeader('X-File-Name', encodeURIComponent(fileName))
xhr.setRequestHeader('X-Chunk-Size', chunk.size.toString())
Object.entries(options.requestHeaders || {}).forEach(([key, value]) => xhr.setRequestHeader(key, value))
// 记录开始上传时间
xhr.upload.onloadstart = () => {
uploadStartTime = Date.now()
const prepareTime = uploadStartTime - startTime
console.log(`[ChunkUpload] 🚀 Chunk ${chunkIndex} 开始上传 (准备耗时: ${prepareTime}ms)`)
}
xhr.upload.onprogress = (event) => {
if (event.lengthComputable) {
// 记录第一个字节发送时间
if (firstByteTime === 0 && event.loaded > 0) {
firstByteTime = Date.now()
const firstByteDelay = firstByteTime - uploadStartTime
console.log(`[ChunkUpload] 📤 Chunk ${chunkIndex} 首字节发送: ${firstByteDelay}ms`)
}
options.onChunkProgress?.(event.loaded, event.total)
// 每秒输出一次进度和速度
const now = Date.now()
if (now - lastProgressTime >= 1000) {
const elapsed = (now - uploadStartTime) / 1000
const speed = event.loaded / elapsed / 1024 / 1024
const instantSpeed = (event.loaded - lastProgressLoaded) / ((now - lastProgressTime) / 1000) / 1024 / 1024
console.log(`[ChunkUpload] 📊 Chunk ${chunkIndex} 进度: ${(event.loaded/event.total*100).toFixed(1)}%, 平均速度: ${speed.toFixed(2)}MB/s, 瞬时速度: ${instantSpeed.toFixed(2)}MB/s`)
lastProgressTime = now
lastProgressLoaded = event.loaded
}
}
}
xhr.onload = () => {
options.onRequest?.(null)
const totalTime = Date.now() - startTime
const uploadTime = Date.now() - uploadStartTime
const serverTime = totalTime - uploadTime
console.log(`[ChunkUpload] ✅ Chunk ${chunkIndex} 完成:`, {
总耗时: `${totalTime}ms`,
上传耗时: `${uploadTime}ms (${(uploadTime/totalTime*100).toFixed(1)}%)`,
服务器处理: `${serverTime}ms (${(serverTime/totalTime*100).toFixed(1)}%)`,
大小: `${(chunk.size / 1024 / 1024).toFixed(2)}MB`,
平均速度: `${(chunk.size / 1024 / 1024 / (uploadTime / 1000)).toFixed(2)}MB/s`
})
if (xhr.status >= 200 && xhr.status < 300) {
try {
const response = JSON.parse(xhr.responseText)
if (response.serverProcessTime) {
console.log(`[ChunkUpload] 🖥️ Chunk ${chunkIndex} 服务器详细耗时: ${response.serverProcessTime}ms`)
}
} catch (e) {
// 忽略JSON解析错误
}
resolve()
} else if (xhr.status === 404) {
// 会话不存在或已过期,抛出特殊错误以便重新初始化
console.warn(`[ChunkUpload] ⚠️ Chunk ${chunkIndex} 会话不存在(可能已过期),需要重新初始化`)
reject(new Error('UPLOAD_SESSION_EXPIRED'))
} else {
const retryable = isRetryableHttpStatus(xhr.status)
const message = getUploadHttpErrorMessage(
`上传块 ${chunkIndex}`,
xhr.status,
`上传块 ${chunkIndex} 失败: ${xhr.status}`,
)
reject(createUploadHttpError(message, xhr.status, retryable))
}
}
xhr.onerror = () => {
options.onRequest?.(null)
const totalTime = Date.now() - startTime
console.error(`[ChunkUpload] ❌ Chunk ${chunkIndex} 失败 (耗时: ${totalTime}ms)`)
reject(new Error(`上传块 ${chunkIndex} 失败: 网络错误`))
}
xhr.onabort = () => {
options.onRequest?.(null)
reject(new Error('上传已取消'))
}
xhr.ontimeout = () => {
options.onRequest?.(null)
const totalTime = Date.now() - startTime
console.error(`[ChunkUpload] ❌ Chunk ${chunkIndex} 超时 (耗时: ${totalTime}ms, 超过60秒)`)
reject(new Error(`上传块 ${chunkIndex} 超时`))
}
xhr.send(chunk)
})
}
// 完成上传(合并所有块)
export async function completeUpload(
uploadId: string | null,
metadata: Record<string, any>
): Promise<any> {
// 添加超时控制(5分钟)
const timeout = 5 * 60 * 1000 // 5分钟
const controller = new AbortController()
const timeoutId = setTimeout(() => controller.abort(), timeout)
try {
const response = await fetch('/api/materials/upload/complete', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ uploadId: uploadId || null, metadata }),
signal: controller.signal,
})
clearTimeout(timeoutId)
if (!response.ok) {
const errorData = await response.json().catch(() => ({}))
const errorMessage = errorData.error || `完成上传失败: ${response.status}`
// 如果是会话过期错误,抛出特殊错误以便前端处理
if (response.status === 404 && errorMessage.includes('会话')) {
throw new Error('UPLOAD_SESSION_EXPIRED')
}
const retryable = isRetryableHttpStatus(response.status)
throw createUploadHttpError(errorMessage, response.status, retryable)
}
try {
return await response.json()
} catch {
// 2xx 已由服务器返回,但响应体在链路中截断时,服务端可能已经完成合并。
throw createUncertainUploadError('完成响应不完整,服务器处理结果尚未确认')
}
} catch (error) {
clearTimeout(timeoutId)
if (error instanceof Error) {
if (error.name === 'AbortError') {
throw createUncertainUploadError('完成请求超时,服务器处理结果尚未确认')
}
if (error.message === 'UPLOAD_SESSION_EXPIRED') {
throw error
}
if ('retryable' in error || isUploadOutcomeUncertain(error)) {
throw error
}
}
// fetch 的网络错误没有可靠的“服务端未执行”语义,必须按结果不确定处理。
throw createUncertainUploadError(
error instanceof Error
? `完成请求中断:${error.message}`
: '完成请求中断,服务器处理结果尚未确认',
)
}
}
// 主上传函数
const activeFileUploadIds = new Set<string>()
export async function chunkUpload(
file: File,
metadata: Record<string, any>,
options: ChunkUploadOptions = {},
): Promise<any> {
const fileId = file?.name?.trim() && Number.isFinite(file.size)
? generateFileId(file)
: ''
if (fileId && activeFileUploadIds.has(fileId)) {
throw new Error(`文件 ${file.name} 正在上传,请勿在同一批次重复选择`)
}
if (fileId) {
activeFileUploadIds.add(fileId)
}
try {
return await runChunkUpload(file, metadata, options)
} finally {
if (fileId) {
activeFileUploadIds.delete(fileId)
}
}
}
async function runChunkUpload(
file: File,
metadata: Record<string, any>,
options: ChunkUploadOptions = {}
): Promise<any> {
if (!file || !file.name?.trim()) {
throw new Error('文件名不能为空,无法上传')
}
if (!Number.isFinite(file.size) || file.size <= 0) {
throw new Error(`文件 ${file.name} 为空,无法上传`)
}
// 如果没有指定chunkSize,根据网络环境自动选择最优值
let defaultChunkSize = 10 * 1024 * 1024 // 默认10MB(公网配置)
if (!options.chunkSize) {
try {
defaultChunkSize = await getOptimalChunkSize()
} catch (error) {
console.warn('[ChunkUpload] 获取最优分片大小失败,使用默认10MB:', error)
}
}
const {
chunkSize: requestedChunkSize = defaultChunkSize, // 根据网络环境自动选择(内网20MB,公网10MB)
onProgress,
maxRetries = 5,
retryDelay = 1000,
onStageChange,
onRequest,
skipComplete = false, // 默认不跳过complete
retainStateOnSkipComplete = false,
isCancelled, // 取消检查回调
onActiveXhrChange, // 活跃XHR集合回调
requestHeaders = {},
} = options
// 跟踪所有活跃的 XHR 请求(用于批量取消)
const activeXhrs = new Set<XMLHttpRequest>()
onStageChange?.('uploading', { message: '正在上传素材文件…' })
const fileId = generateFileId(file)
let chunkSize = Number(requestedChunkSize)
let totalChunks = Math.ceil(file.size / chunkSize)
if (resolveUploadChunkSize(file.size, totalChunks, chunkSize) !== chunkSize) {
throw new Error('上传文件大小或分片配置无效(单文件最多10GB、单分片最多20MB)')
}
// 加载已有状态
let state = await loadUploadState(fileId)
if (state) {
const requestedUploadPurpose = typeof metadata?.uploadPurpose === 'string'
? metadata.uploadPurpose.trim()
: ''
const storedUploadPurpose = typeof state.metadata?.uploadPurpose === 'string'
? state.metadata.uploadPurpose.trim()
: ''
const storedUploadedChunks = state.uploadedChunks instanceof Set
? state.uploadedChunks
: null
const storedChunkSize = resolveUploadChunkSize(
file.size,
Number(state.totalChunks),
state.chunkSize,
)
const storedStateMatchesFile = state.fileName === file.name
&& state.fileSize === file.size
&& storedUploadedChunks !== null
&& Array.from(storedUploadedChunks).every((index) => (
Number.isInteger(index)
&& index >= 0
&& index < state!.totalChunks
))
const storedStateMatchesPurpose = storedUploadPurpose === requestedUploadPurpose
if (state.uploadId && storedStateMatchesFile && storedChunkSize && storedStateMatchesPurpose) {
// 服务端会话的分片布局在 init 时已经固定。网络从公网切到内网(或反向)
// 时仍沿用已保存的 10/20MiB 布局,不能拿新探测值切旧会话。
chunkSize = storedChunkSize
totalChunks = state.totalChunks
} else if (state.uploadId || !storedStateMatchesFile || !storedChunkSize || !storedStateMatchesPurpose) {
// 无法安全证明旧状态与当前文件/布局一致时,清理旧会话并重新初始化,
// 避免把错误切片继续写进服务端既有会话。
if (state.uploadId) {
fetch('/api/materials/upload/cleanup', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ uploadId: state.uploadId }),
}).catch(() => undefined)
}
await deleteUploadState(fileId).catch(() => undefined)
state = null
} else {
// 尚未拿到服务端 uploadId,不存在可恢复的远端进度,按本次网络配置重建。
state.chunkSize = chunkSize
state.totalChunks = totalChunks
state.uploadedChunks.clear()
}
}
if (!state) {
// 初始化新上传
state = {
fileId,
fileName: file.name,
fileSize: file.size,
chunkSize,
totalChunks,
uploadedChunks: new Set(),
metadata, // 首次保存metadata
}
await saveUploadState(state)
} else {
// 恢复上传
console.log(`[ChunkUpload] 恢复上传: ${state.uploadedChunks.size}/${totalChunks} 块已上传`)
// 用最新metadata覆盖旧值(以便完成时使用最新表单信息)
state.metadata = metadata
await saveUploadState(state)
}
// 初始化上传(如果需要)
if (!state.uploadId) {
try {
state.uploadId = await initUpload(
file.name,
file.size,
totalChunks,
chunkSize,
typeof metadata?.uploadPurpose === 'string' ? metadata.uploadPurpose : undefined,
requestHeaders,
)
await saveUploadState(state)
} catch (error) {
if (!isRetryableUploadError(error)) {
await deleteUploadState(fileId).catch(() => {})
throw error
}
// 自动重试初始化
let retries = 0
while (retries < maxRetries) {
await new Promise(resolve => setTimeout(resolve, retryDelay))
try {
state.uploadId = await initUpload(
file.name,
file.size,
totalChunks,
chunkSize,
typeof metadata?.uploadPurpose === 'string' ? metadata.uploadPurpose : undefined,
requestHeaders,
)
await saveUploadState(state)
break
} catch (e) {
if (!isRetryableUploadError(e)) {
await deleteUploadState(fileId).catch(() => {})
throw e
}
retries++
if (retries >= maxRetries) {
throw new Error('初始化上传失败,已重试多次')
}
}
}
}
}
// 网速计算相关变量
let lastUpdateTime = Date.now()
let lastUploadedBytes = 0
// 计算已上传的字节数(包括恢复上传时已上传的块)
let completedBytes = 0
for (let i = 0; i < totalChunks; i++) {
if (state.uploadedChunks.has(i)) {
const start = i * chunkSize
const end = Math.min(start + chunkSize, file.size)
completedBytes += (end - start)
}
}
lastUploadedBytes = completedBytes // 初始化已上传字节数
const speedHistory: number[] = [] // 用于平滑网速计算
const SPEED_HISTORY_SIZE = 5 // 保留最近5次的速度记录
const inFlightChunkBytes = new Map<number, number>()
const getInFlightBytes = () => {
let total = 0
inFlightChunkBytes.forEach((bytes) => {
total += bytes
})
return total
}
const updateProgress = (uploadedBytes: number) => {
const clampedBytes = Math.min(file.size, Math.max(0, uploadedBytes))
const progress = Math.max(
uploadedBytes > 0 ? 1 : 0,
Math.min(100, Math.round((clampedBytes / file.size) * 100))
)
const currentTime = Date.now()
const timeDelta = (currentTime - lastUpdateTime) / 1000 // 秒
const bytesDelta = clampedBytes - lastUploadedBytes
// 优化:降低速度显示更新频率,从0.1秒改为1秒
// 避免速度数字跳动过快,用户无法看清
const UPDATE_INTERVAL = 1 // 1秒更新一次
if (timeDelta >= UPDATE_INTERVAL) {
const currentSpeed = bytesDelta / timeDelta
if (currentSpeed > 0 && currentSpeed < file.size * 100) {
speedHistory.push(currentSpeed)
if (speedHistory.length > SPEED_HISTORY_SIZE) {
speedHistory.shift()
}
}
let avgSpeed = 0
if (speedHistory.length > 0) {
avgSpeed = speedHistory.reduce((sum, s) => sum + s, 0) / speedHistory.length
}
onProgress?.(progress, avgSpeed > 0 ? avgSpeed : undefined)
lastUpdateTime = currentTime
lastUploadedBytes = clampedBytes
} else {
// 在更新间隔内,使用上一次的速度显示(避免数字跳动)
if (speedHistory.length > 0) {
const lastSpeed = speedHistory[speedHistory.length - 1]
onProgress?.(progress, lastSpeed)
} else {
onProgress?.(progress)
}
}
}
if (completedBytes > 0) {
updateProgress(completedBytes)
}
// 动态并发上传配置(根据网络环境自动调整)
let defaultConcurrency = options.chunkConcurrency ?? 2 // 默认2并发
if (options.chunkConcurrency === undefined) {
try {
defaultConcurrency = await getOptimalConcurrency()
} catch (error) {
console.warn('[ChunkUpload] 获取最优并发数失败,使用默认值(2):', error)
}
}
const CONCURRENT_CHUNKS = Math.max(1, Math.min(8, Math.floor(Number(defaultConcurrency) || 1)))
// 创建待上传chunk队列
const pendingChunks: number[] = []
for (let chunkIndex = 0; chunkIndex < totalChunks; chunkIndex++) {
if (!state.uploadedChunks.has(chunkIndex)) {
pendingChunks.push(chunkIndex)
}
}
console.log(`[ChunkUpload] 开始并发上传: ${pendingChunks.length}个chunk待上传,${CONCURRENT_CHUNKS}个并发`)
// 会话过期标志(用于协调所有worker)
let sessionExpired = false
let reinitPromise: Promise<void> | null = null
// 并发上传worker
const uploadWorker = async () => {
while (pendingChunks.length > 0) {
// 检查取消状态
if (isCancelled?.()) {
console.log('[ChunkUpload] Worker 检测到取消信号,停止上传')
throw new Error('上传已取消')
}
const chunkIndex = pendingChunks.shift()
if (chunkIndex === undefined) break
// 读取块
const start = chunkIndex * chunkSize
const end = Math.min(start + chunkSize, file.size)
const chunk = file.slice(start, end)
const chunkSizeBytes = end - start
// 上传块(带自动重试)
let retries = 0
let success = false
while (!success && retries < maxRetries) {
// 每次重试前检查取消状态
if (isCancelled?.()) {
console.log('[ChunkUpload] Worker 检测到取消信号,停止上传')
throw new Error('上传已取消')
}
// 当前分片的 XHR 引用
let currentXhr: XMLHttpRequest | null = null
try {
await uploadChunk(state.uploadId!, chunkIndex, chunk, file.name, {
onChunkProgress: (loaded) => {
inFlightChunkBytes.set(chunkIndex, Math.min(loaded, chunkSizeBytes))
updateProgress(completedBytes + getInFlightBytes())
},
onRequest: (xhr) => {
// 跟踪所有活跃的 XHR
if (xhr) {
currentXhr = xhr
activeXhrs.add(xhr)
onActiveXhrChange?.(activeXhrs)
} else if (currentXhr) {
// XHR 完成,从集合中移除
activeXhrs.delete(currentXhr)
onActiveXhrChange?.(activeXhrs)
currentXhr = null
}
// 同时调用原来的回调
onRequest?.(xhr)
},
requestHeaders,
})
state.uploadedChunks.add(chunkIndex)
await saveUploadState(state)
success = true
// 更新已上传字节数(完成整个块)
inFlightChunkBytes.delete(chunkIndex)
completedBytes += chunkSizeBytes
updateProgress(completedBytes + getInFlightBytes())
console.log(`[ChunkUpload] Chunk ${chunkIndex}/${totalChunks} 完成,剩余 ${pendingChunks.length} 个`)
} catch (error) {
inFlightChunkBytes.delete(chunkIndex)
updateProgress(completedBytes + getInFlightBytes())
// 清理当前 XHR
if (currentXhr) {
activeXhrs.delete(currentXhr)
onActiveXhrChange?.(activeXhrs)
currentXhr = null
}
if ((error as Error)?.message === '上传已取消') {
onRequest?.(null)
// 取消时清理 IndexedDB 状态
await deleteUploadState(fileId).catch(() => {})
// 调用后端清理 API 删除服务器临时文件
if (state.uploadId) {
console.log('[ChunkUpload] 取消上传,调用后端清理临时文件...')
fetch('/api/materials/upload/cleanup', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ uploadId: state.uploadId })
}).catch(err => {
console.warn('[ChunkUpload] 清理临时文件失败(忽略):', err)
})
}
throw error
}
if (isNonRetryableUploadError(error)) {
await deleteUploadState(fileId).catch(() => {})
if (state.uploadId) {
console.log('[ChunkUpload] 上传分片失败且不可重试,调用后端清理临时文件...')
fetch('/api/materials/upload/cleanup', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ uploadId: state.uploadId })
}).catch(err => {
console.warn('[ChunkUpload] 清理临时文件失败(忽略):', err)
})
}
throw error
}
// 如果是会话过期,尝试刷新会话(保留进度)
if ((error as Error)?.message === 'UPLOAD_SESSION_EXPIRED') {
// ⚠️ 重要修复:如果所有分片都已上传成功,不要重新初始化
// 这种情况可能是 complete 阶段的竞争条件导致的,应该直接继续到 complete
if (state.uploadedChunks.size >= totalChunks) {
console.warn(`[ChunkUpload] ⚠️ 会话过期但所有分片已上传完成 (${state.uploadedChunks.size}/${totalChunks}),跳过重新初始化`)
sessionExpired = false
continue
}
// 如果已经有其他worker在处理,等待它完成
if (reinitPromise) {
console.log(`[ChunkUpload] 等待其他worker完成会话刷新...`)
await reinitPromise
continue
}
// 标记会话过期,防止其他worker重复处理
if (!sessionExpired) {
sessionExpired = true
console.warn(`[ChunkUpload] 🔄 检测到会话过期,尝试刷新会话(保留已上传进度)...`)
reinitPromise = (async () => {
try {
// 尝试刷新会话(保留uploadId和已上传进度)
const refreshResponse = await fetch('/api/materials/upload/refresh', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ uploadId: state.uploadId })
})
const refreshResult = await refreshResponse.json()
if (refreshResult.success) {
console.log(`[ChunkUpload] ✅ 会话刷新成功,保留 ${state.uploadedChunks.size} 个已上传分片,继续上传`)
sessionExpired = false
// 不清空 uploadedChunks,保留进度
return
} else {
// ⚠️ 修复:如果刷新失败但所有分片已上传,不要重新初始化
if (state.uploadedChunks.size >= totalChunks) {
console.warn(`[ChunkUpload] ⚠️ 会话刷新失败但所有分片已上传完成,跳过重新初始化`)
sessionExpired = false
return
}
console.warn(`[ChunkUpload] ⚠️ 会话刷新失败(${refreshResult.error}),重新初始化上传...`)
// 刷新失败,需要重新初始化
state.uploadId = await initUpload(
file.name,
file.size,
totalChunks,
chunkSize,
typeof metadata?.uploadPurpose === 'string' ? metadata.uploadPurpose : undefined,
requestHeaders,
)
state.uploadedChunks.clear()
completedBytes = 0
await saveUploadState(state)
// 将所有chunk重新加入队列
pendingChunks.length = 0
for (let i = 0; i < totalChunks; i++) {
pendingChunks.push(i)
}
console.log(`[ChunkUpload] ✅ 会话重新初始化成功,重新开始上传`)
sessionExpired = false
return
}
} catch (refreshError) {
console.error(`[ChunkUpload] ❌ 会话刷新/重新初始化失败:`, refreshError)
throw new Error('上传会话已过期且无法恢复,请刷新页面后重试')
}
})()
await reinitPromise
reinitPromise = null
// 不增加重试次数,直接继续
continue
} else {
// 等待刷新/重新初始化完成
if (reinitPromise) {
await reinitPromise
continue
}
}
}
retries++
if (retries < maxRetries) {
// 等待后重试
await new Promise(resolve => setTimeout(resolve, retryDelay * retries))
} else {
throw new Error(`上传块 ${chunkIndex} 失败,已重试多次`)
}
}
}
}
}
// 启动多个并发worker
const workers = []
for (let i = 0; i < CONCURRENT_CHUNKS; i++) {
workers.push(uploadWorker())
}
// 等待所有worker完成
await Promise.all(workers)
// 如果设置了skipComplete,分片上传完成后立即返回,不调用completeUpload
if (skipComplete) {
console.log('[ChunkUpload] 分片上传完成,跳过complete处理(将在后台异步处理)')
if (!retainStateOnSkipComplete) {
try {
await deleteUploadState(fileId)
} catch (cleanupError) {
console.warn('[ChunkUpload] 清理上传状态失败(将忽略并继续):', cleanupError)
}
}
return {
uploadId: state.uploadId,
metadata,
chunksUploaded: true,
...(retainStateOnSkipComplete
? {
confirmMaterialCreated: async () => {
await deleteUploadState(fileId)
},
}
: {}),
}
}
// 完成上传
let retries = 0
let result: any = null
while (!result && retries < maxRetries) {
try {
onStageChange?.('processing', {
message:
metadata?.materialType === '模特' || metadata?.materialType === '实拍图片' || metadata?.materialType === '买家秀图片' || metadata?.materialType === '小红书种草' || metadata?.materialType === '主图模版' || metadata?.materialType === 'SKU模版'
? `展示图处理中(共 ${metadata?.modelImages?.length || 0} 张)…`
: '正在处理素材,请稍候…',
})
result = await completeUpload(state.uploadId!, metadata)
// 删除保存的状态
await deleteUploadState(fileId)
return result
} catch (error) {
// 如果是会话过期错误,不再重试,直接抛出
if (error instanceof Error && error.message === 'UPLOAD_SESSION_EXPIRED') {
await deleteUploadState(fileId)
// 清理后端临时文件
if (state.uploadId) {
fetch('/api/materials/upload/cleanup', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ uploadId: state.uploadId })
}).catch(() => {})
}
throw new Error('上传会话已过期,请重新上传文件')
}
if (isNonRetryableUploadError(error)) {
await deleteUploadState(fileId).catch(() => {})
if (state.uploadId) {
fetch('/api/materials/upload/cleanup', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ uploadId: state.uploadId })
}).catch(() => {})
}
throw error
}
retries++
if (retries < maxRetries) {
await new Promise(resolve => setTimeout(resolve, retryDelay * retries))
} else {
// 5xx、超时和网络错误都不能证明服务端没有完成合并。保留本地断点,
// 且绝不 destructive cleanup,避免删除一个已经成功生成的 merged 文件。
const detail = error instanceof Error ? error.message : '服务器处理结果尚未确认'
const suffix = isUploadOutcomeUncertain(error)
? '服务器可能已经完成处理'
: '服务暂时不可用'
throw createUncertainUploadError(
`${detail};${suffix},已保留上传断点,请稍后重试确认`,
)
}
}
}
return result
}
// 检查是否有未完成的上传
export async function hasPendingUploads(): Promise<boolean> {
try {
const db = await getDB()
const transaction = db.transaction([STORE_NAME], 'readonly')
const store = transaction.objectStore(STORE_NAME)
return new Promise<boolean>((resolve) => {
const request = store.count()
request.onsuccess = () => resolve(request.result > 0)
request.onerror = () => resolve(false)
})
} catch {
return false
}
}
// 获取所有未完成的上传状态
export async function getAllPendingUploads(): Promise<ChunkUploadState[]> {
try {
const db = await getDB()
const transaction = db.transaction([STORE_NAME], 'readonly')
const store = transaction.objectStore(STORE_NAME)
return new Promise<ChunkUploadState[]>((resolve) => {
const request = store.getAll()
request.onsuccess = () => resolve(request.result || [])
request.onerror = () => resolve([])
})
} catch (error) {
console.error('[ChunkUpload] Failed to get pending uploads:', error)
return []
}
}
// 清理所有上传状态
export async function clearAllUploadStates(): Promise<void> {
try {
const db = await getDB()
const transaction = db.transaction([STORE_NAME], 'readwrite')
const store = transaction.objectStore(STORE_NAME)
await new Promise<void>((resolve, reject) => {
const request = store.clear()
request.onsuccess = () => resolve()
request.onerror = () => reject(request.error)
})
} catch (error) {
console.error('[ChunkUpload] Failed to clear states:', error)
}
}