771 lines
23 KiB
TypeScript
771 lines
23 KiB
TypeScript
/**
|
||
* 上传会话存储管理
|
||
*
|
||
* 🔄 支持环境变量切换：
|
||
* - USE_REDIS=true → 使用Redis（生产环境推荐）
|
||
* - USE_REDIS=false 或未设置 → 使用内存Map（开发/测试环境）
|
||
*
|
||
* 📋 内存Map的局限性：
|
||
* 1. 数据不持久化 - 服务器/容器重启后数据丢失
|
||
* 2. 不支持多实例 - 多个容器实例无法共享会话数据
|
||
* 3. 内存管理 - 需要手动清理过期数据
|
||
* 4. 无法跨服务 - 不同服务无法共享数据
|
||
*
|
||
* ✅ Redis的优势：
|
||
* 1. 数据持久化 - 服务器重启后数据保留
|
||
* 2. 支持多实例 - 多个容器实例可以共享数据
|
||
* 3. 自动过期 - 支持TTL，无需手动清理
|
||
* 4. 跨服务共享 - 不同服务可以共享数据
|
||
*
|
||
* 📚 详细说明请查看：docs/upload-sessions-redis-explanation.md
|
||
*/
|
||
|
||
import path from 'path'
|
||
import { lstat, readFile } from 'fs/promises'
|
||
import {
|
||
ensureSafeDirectoryWithinRoot,
|
||
isPathFreeOfSymbolicLinks,
|
||
normalizeSafeFileName,
|
||
resolveStrictChildPath,
|
||
} from './path-security'
|
||
import { getUploadRootDir } from './upload-paths'
|
||
|
||
export interface UploadSession {
|
||
fileName: string
|
||
fileSize: number
|
||
totalChunks: number
|
||
chunks: Map<number, string> // chunkIndex -> chunkFilePath
|
||
createdAt: number
|
||
/**
|
||
* 最近一次“确实有上传流量”的时间（用于性能面板判断是否仍在上传）
|
||
* - init 时会写入
|
||
* - 每次 chunk 上传成功后会更新
|
||
*/
|
||
lastActivityAt?: number
|
||
/**
|
||
* 已接收的字节数（用于估算上传速度/进度展示）
|
||
* 注意：这是服务端接收字节，不代表最终文件写入/合并完成
|
||
*/
|
||
uploadedBytes?: number
|
||
/**
|
||
* 当前估算上传速度（字节/秒）
|
||
* 由 chunk 上传端点按增量估算，供性能面板展示
|
||
*/
|
||
currentSpeed?: number
|
||
/**
|
||
* 用于速度估算的上一次采样点
|
||
*/
|
||
_speedSampleAt?: number
|
||
_speedSampleBytes?: number
|
||
userId: string
|
||
}
|
||
|
||
function parsePositiveIntEnv(name: string, fallback: number): number {
|
||
const value = Number.parseInt(process.env[name] || '', 10)
|
||
return Number.isFinite(value) && value > 0 ? value : fallback
|
||
}
|
||
|
||
// 检查是否使用Redis
|
||
const USE_REDIS = process.env.USE_REDIS === 'true'
|
||
|
||
// 动态导入Redis实现（如果启用）
|
||
let redisModule: typeof import('./upload-sessions-redis') | null = null
|
||
|
||
if (USE_REDIS) {
|
||
try {
|
||
// 动态导入，避免在未安装ioredis时出错
|
||
redisModule = require('./upload-sessions-redis')
|
||
console.log('[UploadSessions] 使用Redis存储（生产环境）')
|
||
} catch (error) {
|
||
console.warn('[UploadSessions] Redis模块加载失败，降级到内存Map:', error)
|
||
console.warn('[UploadSessions] 请确保已安装ioredis: npm install ioredis')
|
||
}
|
||
} else {
|
||
console.log('[UploadSessions] 使用内存Map存储（开发/测试环境）')
|
||
}
|
||
|
||
// 内存存储（单实例部署可以使用）
|
||
const memorySessions = new Map<string, UploadSession>()
|
||
const MAX_UPLOAD_SESSION_AGE_MS = 8 * 60 * 60 * 1000
|
||
const SAFE_UPLOAD_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$/
|
||
const MEMORY_CACHE_LIMIT = parsePositiveIntEnv('UPLOAD_SESSION_MEMORY_CACHE_LIMIT', 500)
|
||
const MEMORY_CACHE_MAX_AGE_MS = parsePositiveIntEnv(
|
||
'UPLOAD_SESSION_MEMORY_CACHE_MAX_AGE_MS',
|
||
2 * 60 * 60 * 1000
|
||
)
|
||
const MEMORY_CACHE_PRUNE_INTERVAL_MS = parsePositiveIntEnv(
|
||
'UPLOAD_SESSION_MEMORY_CACHE_PRUNE_INTERVAL_MS',
|
||
5 * 60 * 1000
|
||
)
|
||
let lastCachePruneAt = 0
|
||
|
||
function pruneMemorySessions(now: number) {
|
||
if (!(USE_REDIS && redisModule)) {
|
||
return
|
||
}
|
||
|
||
if (
|
||
now - lastCachePruneAt < MEMORY_CACHE_PRUNE_INTERVAL_MS &&
|
||
memorySessions.size <= MEMORY_CACHE_LIMIT
|
||
) {
|
||
return
|
||
}
|
||
lastCachePruneAt = now
|
||
|
||
const expiredKeys: string[] = []
|
||
memorySessions.forEach((session, uploadId) => {
|
||
const activityAt = session.lastActivityAt || session.createdAt
|
||
if (now - activityAt > MEMORY_CACHE_MAX_AGE_MS) {
|
||
expiredKeys.push(uploadId)
|
||
}
|
||
})
|
||
expiredKeys.forEach((uploadId) => memorySessions.delete(uploadId))
|
||
|
||
while (memorySessions.size > MEMORY_CACHE_LIMIT) {
|
||
const oldestUploadId = memorySessions.keys().next().value as string | undefined
|
||
if (!oldestUploadId) {
|
||
break
|
||
}
|
||
memorySessions.delete(oldestUploadId)
|
||
}
|
||
}
|
||
|
||
function setMemorySession(uploadId: string, session: UploadSession) {
|
||
if (memorySessions.has(uploadId)) {
|
||
memorySessions.delete(uploadId)
|
||
}
|
||
memorySessions.set(uploadId, session)
|
||
pruneMemorySessions(Date.now())
|
||
}
|
||
|
||
function deleteMemorySession(uploadId: string) {
|
||
memorySessions.delete(uploadId)
|
||
}
|
||
|
||
function getMemorySession(uploadId: string): UploadSession | null {
|
||
const session = memorySessions.get(uploadId) || null
|
||
if (!session) {
|
||
return null
|
||
}
|
||
|
||
if (Date.now() - session.createdAt > MAX_UPLOAD_SESSION_AGE_MS) {
|
||
memorySessions.delete(uploadId)
|
||
return null
|
||
}
|
||
|
||
if (USE_REDIS && redisModule) {
|
||
setMemorySession(uploadId, session)
|
||
}
|
||
return session
|
||
}
|
||
|
||
export function isSafeUploadId(uploadId: string): boolean {
|
||
return SAFE_UPLOAD_ID_PATTERN.test(String(uploadId || '').trim())
|
||
}
|
||
|
||
function getUploadRoot(): string {
|
||
return getUploadRootDir()
|
||
}
|
||
|
||
function isPathInside(basePath: string, targetPath: string): boolean {
|
||
return targetPath === basePath || targetPath.startsWith(`${basePath}${path.sep}`)
|
||
}
|
||
|
||
async function getUploadSessionFromMetadata(uploadId: string): Promise<UploadSession | null> {
|
||
if (!isSafeUploadId(uploadId)) {
|
||
return null
|
||
}
|
||
|
||
const tempRoot = path.join(getUploadRoot(), 'temp')
|
||
const sessionDir = path.resolve(tempRoot, uploadId)
|
||
if (!isPathInside(tempRoot, sessionDir)) {
|
||
return null
|
||
}
|
||
|
||
try {
|
||
const rawMetadata = await readFile(path.join(sessionDir, 'meta.json'), 'utf8')
|
||
const metadata = JSON.parse(rawMetadata)
|
||
const createdAt = Number(metadata?.createdAt)
|
||
const fileSize = Number(metadata?.fileSize)
|
||
const totalChunks = Number(metadata?.totalChunks)
|
||
const fileName = typeof metadata?.fileName === 'string'
|
||
? normalizeSafeFileName(metadata.fileName)
|
||
: null
|
||
const userId = typeof metadata?.userId === 'string' ? metadata.userId : ''
|
||
const ageMs = Date.now() - createdAt
|
||
|
||
if (
|
||
!fileName ||
|
||
!userId ||
|
||
!Number.isFinite(createdAt) ||
|
||
ageMs < -5 * 60 * 1000 ||
|
||
ageMs > MAX_UPLOAD_SESSION_AGE_MS ||
|
||
!Number.isFinite(fileSize) ||
|
||
fileSize <= 0 ||
|
||
!Number.isInteger(totalChunks) ||
|
||
totalChunks <= 0
|
||
) {
|
||
return null
|
||
}
|
||
|
||
const recoveredSession: UploadSession = {
|
||
fileName,
|
||
fileSize,
|
||
totalChunks,
|
||
chunks: new Map(),
|
||
createdAt,
|
||
userId,
|
||
}
|
||
setMemorySession(uploadId, recoveredSession)
|
||
return recoveredSession
|
||
} catch {
|
||
return null
|
||
}
|
||
}
|
||
|
||
async function withRedisFallback<T>(
|
||
actionName: string,
|
||
redisAction: () => Promise<T>,
|
||
memoryFallback: () => Promise<T> | T
|
||
): Promise<T> {
|
||
if (USE_REDIS && redisModule) {
|
||
try {
|
||
return await redisAction()
|
||
} catch (error) {
|
||
console.error(`[UploadSessions] Redis ${actionName} failed, fallback to memory:`, error)
|
||
return await memoryFallback()
|
||
}
|
||
}
|
||
return await memoryFallback()
|
||
}
|
||
|
||
// 清理过期会话（8小时）- 仅内存Map需要
|
||
if (!USE_REDIS || !redisModule) {
|
||
const memoryCleanupTimer = setInterval(() => {
|
||
const now = Date.now()
|
||
const expireTime = 8 * 60 * 60 * 1000 // 8小时
|
||
const keysToDelete: string[] = []
|
||
|
||
memorySessions.forEach((session, uploadId) => {
|
||
if (now - session.createdAt > expireTime) {
|
||
keysToDelete.push(uploadId)
|
||
}
|
||
})
|
||
|
||
keysToDelete.forEach(key => {
|
||
memorySessions.delete(key)
|
||
console.log(`[UploadSessions] 清理过期会话: ${key}`)
|
||
})
|
||
|
||
if (keysToDelete.length > 0) {
|
||
console.log(`[UploadSessions] 已清理 ${keysToDelete.length} 个过期会话，当前会话数: ${memorySessions.size}`)
|
||
} else {
|
||
// 即使没有清理，也记录当前会话数（用于监控）
|
||
console.log(`[UploadSessions] 当前会话数: ${memorySessions.size}`)
|
||
}
|
||
}, 60 * 60 * 1000) // 每小时清理一次
|
||
memoryCleanupTimer.unref()
|
||
}
|
||
|
||
/**
|
||
* 保存上传会话
|
||
*/
|
||
export async function saveUploadSession(
|
||
uploadId: string,
|
||
session: UploadSession
|
||
): Promise<void> {
|
||
const normalizedUploadId = String(uploadId || '').trim()
|
||
const normalizedFileName = normalizeSafeFileName(session.fileName)
|
||
if (!isSafeUploadId(normalizedUploadId) || !normalizedFileName) {
|
||
throw new Error('上传会话包含不安全的标识或文件名')
|
||
}
|
||
|
||
const normalizedSession = {
|
||
...session,
|
||
fileName: normalizedFileName,
|
||
}
|
||
setMemorySession(normalizedUploadId, normalizedSession)
|
||
await withRedisFallback(
|
||
'saveUploadSession',
|
||
() => redisModule!.saveUploadSession(normalizedUploadId, normalizedSession),
|
||
async () => {}
|
||
)
|
||
}
|
||
|
||
/**
|
||
* 获取上传会话
|
||
*/
|
||
export async function getUploadSession(
|
||
uploadId: string
|
||
): Promise<UploadSession | null> {
|
||
return withRedisFallback(
|
||
'getUploadSession',
|
||
async () => {
|
||
const session = await redisModule!.getUploadSession(uploadId)
|
||
if (session) {
|
||
setMemorySession(uploadId, session)
|
||
return session
|
||
}
|
||
return getMemorySession(uploadId)
|
||
},
|
||
async () => getMemorySession(uploadId)
|
||
)
|
||
}
|
||
|
||
/**
|
||
* 只返回属于指定用户的上传会话。
|
||
* complete 端点会提前删除 Redis 会话，因此允许从同目录 meta.json 恢复所有权信息。
|
||
*/
|
||
export async function getOwnedUploadSession(
|
||
uploadId: string,
|
||
userId: string
|
||
): Promise<UploadSession | null> {
|
||
const normalizedUploadId = String(uploadId || '').trim()
|
||
const normalizedUserId = String(userId || '').trim()
|
||
if (!isSafeUploadId(normalizedUploadId) || !normalizedUserId) {
|
||
return null
|
||
}
|
||
|
||
const session =
|
||
(await getUploadSession(normalizedUploadId)) ||
|
||
(await getUploadSessionFromMetadata(normalizedUploadId))
|
||
|
||
return (
|
||
session?.userId === normalizedUserId &&
|
||
normalizeSafeFileName(session.fileName) !== null
|
||
)
|
||
? session
|
||
: null
|
||
}
|
||
|
||
export async function getOwnedUploadedFile(
|
||
uploadId: string,
|
||
userId: string
|
||
): Promise<string | null> {
|
||
const session = await getOwnedUploadSession(uploadId, userId)
|
||
if (!session) {
|
||
return null
|
||
}
|
||
|
||
return getUploadedFile(uploadId)
|
||
}
|
||
|
||
export interface OwnedMergedUpload {
|
||
uploadId: string
|
||
filePath: string
|
||
session: UploadSession
|
||
}
|
||
|
||
/**
|
||
* 校验分片合并文件确实位于当前用户的上传会话目录，且大小与 init 元数据一致。
|
||
*/
|
||
export async function resolveOwnedMergedUpload(input: {
|
||
existingFilePath: string
|
||
userId: string
|
||
uploadId?: string | null
|
||
}): Promise<OwnedMergedUpload | null> {
|
||
const suppliedPath = String(input.existingFilePath || '').trim()
|
||
const userId = String(input.userId || '').trim()
|
||
if (!suppliedPath || !userId) {
|
||
return null
|
||
}
|
||
|
||
const uploadRoot = getUploadRoot()
|
||
const tempRoot = path.join(uploadRoot, 'temp')
|
||
const targetPath = path.resolve(suppliedPath)
|
||
if (!isPathInside(tempRoot, targetPath)) {
|
||
return null
|
||
}
|
||
|
||
const derivedUploadId = path.basename(path.dirname(targetPath))
|
||
const suppliedUploadId = String(input.uploadId || '').trim()
|
||
if (
|
||
!isSafeUploadId(derivedUploadId) ||
|
||
(suppliedUploadId && suppliedUploadId !== derivedUploadId)
|
||
) {
|
||
return null
|
||
}
|
||
|
||
const sessionDir = path.resolve(tempRoot, derivedUploadId)
|
||
if (path.dirname(targetPath) !== sessionDir || !path.basename(targetPath).startsWith('merged-')) {
|
||
return null
|
||
}
|
||
|
||
const session = await getOwnedUploadSession(derivedUploadId, userId)
|
||
if (!session) {
|
||
return null
|
||
}
|
||
|
||
try {
|
||
const stats = await lstat(targetPath)
|
||
if (!stats.isFile() || stats.isSymbolicLink() || stats.size !== session.fileSize) {
|
||
return null
|
||
}
|
||
} catch {
|
||
return null
|
||
}
|
||
|
||
return {
|
||
uploadId: derivedUploadId,
|
||
filePath: targetPath,
|
||
session,
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 获取已上传的完整文件路径（从chunk临时目录）
|
||
* 返回组装后的完整文件路径
|
||
*/
|
||
export async function getUploadedFile(uploadId: string): Promise<string | null> {
|
||
const normalizedUploadId = String(uploadId || '').trim()
|
||
if (!isSafeUploadId(normalizedUploadId)) {
|
||
return null
|
||
}
|
||
|
||
const session = await getUploadSession(normalizedUploadId);
|
||
if (!session) {
|
||
return null;
|
||
}
|
||
|
||
const safeFileName = normalizeSafeFileName(session.fileName)
|
||
if (!safeFileName) {
|
||
console.warn(`[Upload Session] Unsafe file name rejected: ${normalizedUploadId}`)
|
||
return null
|
||
}
|
||
|
||
const { readdir, open, lstat, unlink } = await import('fs/promises');
|
||
const uploadRoot = getUploadRoot()
|
||
const tempDir = resolveStrictChildPath(uploadRoot, 'temp', normalizedUploadId)
|
||
const outputDir = await ensureSafeDirectoryWithinRoot(uploadRoot, 'assembled')
|
||
if (!tempDir || !outputDir) {
|
||
console.warn(`[Upload Session] Unsafe upload directory rejected: ${normalizedUploadId}`)
|
||
return null
|
||
}
|
||
|
||
const outputPath = resolveStrictChildPath(
|
||
outputDir,
|
||
`${normalizedUploadId}-${safeFileName}`,
|
||
)
|
||
if (
|
||
!outputPath ||
|
||
!(await isPathFreeOfSymbolicLinks(uploadRoot, tempDir)) ||
|
||
!(await isPathFreeOfSymbolicLinks(uploadRoot, outputPath, { allowMissingLeaf: true }))
|
||
) {
|
||
console.warn(`[Upload Session] Unsafe upload path rejected: ${normalizedUploadId}`)
|
||
return null
|
||
}
|
||
|
||
let outputCreated = false
|
||
|
||
try {
|
||
// 先检查文件是否已经存在（避免重复组装）
|
||
try {
|
||
const existingStats = await lstat(outputPath);
|
||
if (existingStats.isSymbolicLink() || !existingStats.isFile()) {
|
||
console.warn(`[Upload Session] Existing output is not a regular file: ${outputPath}`)
|
||
return null
|
||
}
|
||
if (existingStats.size === session.fileSize) {
|
||
console.log(`[Upload Session] File already exists, skip assembly: ${outputPath}`);
|
||
return outputPath;
|
||
}
|
||
await unlink(outputPath)
|
||
} catch (error: any) {
|
||
if (error?.code !== 'ENOENT') {
|
||
throw error
|
||
}
|
||
}
|
||
|
||
// 读取所有chunk文件
|
||
const files = await readdir(tempDir);
|
||
const chunkFiles = files
|
||
.filter(f => /^chunk-\d+$/.test(f))
|
||
.sort((a, b) => {
|
||
const indexA = parseInt(a.replace('chunk-', ''));
|
||
const indexB = parseInt(b.replace('chunk-', ''));
|
||
return indexA - indexB;
|
||
});
|
||
|
||
if (chunkFiles.length === 0) {
|
||
return null;
|
||
}
|
||
if (chunkFiles.length !== session.totalChunks) {
|
||
console.warn(
|
||
`[Upload Session] Incomplete chunks: ${uploadId}, expected=${session.totalChunks}, actual=${chunkFiles.length}`,
|
||
);
|
||
return null;
|
||
}
|
||
|
||
if (
|
||
!Number.isSafeInteger(session.fileSize) ||
|
||
session.fileSize <= 0 ||
|
||
chunkFiles.some((chunkFile, index) => chunkFile !== `chunk-${index}`)
|
||
) {
|
||
console.warn(`[Upload Session] Invalid chunk layout: ${normalizedUploadId}`)
|
||
return null
|
||
}
|
||
|
||
const validatedChunks: Array<{ path: string; size: number }> = []
|
||
let validatedTotalBytes = 0
|
||
for (const chunkFile of chunkFiles) {
|
||
const chunkPath = path.join(tempDir, chunkFile)
|
||
const chunkStats = await lstat(chunkPath)
|
||
if (
|
||
chunkStats.isSymbolicLink() ||
|
||
!chunkStats.isFile() ||
|
||
!Number.isSafeInteger(chunkStats.size) ||
|
||
chunkStats.size <= 0
|
||
) {
|
||
console.warn(`[Upload Session] Invalid chunk file: ${normalizedUploadId}/${chunkFile}`)
|
||
return null
|
||
}
|
||
|
||
validatedTotalBytes += chunkStats.size
|
||
if (!Number.isSafeInteger(validatedTotalBytes) || validatedTotalBytes > session.fileSize) {
|
||
console.warn(
|
||
`[Upload Session] Chunk bytes exceed declared size: ${normalizedUploadId}, expected=${session.fileSize}, actual>${session.fileSize}`,
|
||
)
|
||
return null
|
||
}
|
||
validatedChunks.push({ path: chunkPath, size: chunkStats.size })
|
||
}
|
||
|
||
if (validatedTotalBytes !== session.fileSize) {
|
||
console.warn(
|
||
`[Upload Session] Chunk bytes mismatch: ${normalizedUploadId}, expected=${session.fileSize}, actual=${validatedTotalBytes}`,
|
||
)
|
||
return null
|
||
}
|
||
|
||
// 使用固定大小缓冲区顺序复制。不能对每个分片直接 readFile：若物理分片
|
||
// 与会话声明大小不一致，攻击者可以让组装阶段在最终大小校验前吞入超大 Buffer。
|
||
const outputHandle = await open(outputPath, 'wx');
|
||
outputCreated = true
|
||
try {
|
||
const copyBuffer = Buffer.allocUnsafe(1024 * 1024)
|
||
let copiedTotalBytes = 0
|
||
|
||
for (const chunk of validatedChunks) {
|
||
const chunkHandle = await open(chunk.path, 'r')
|
||
let copiedChunkBytes = 0
|
||
try {
|
||
const openedChunkStats = await chunkHandle.stat()
|
||
if (!openedChunkStats.isFile() || openedChunkStats.size !== chunk.size) {
|
||
throw new Error(`上传分片在组装期间发生变化: ${path.basename(chunk.path)}`)
|
||
}
|
||
|
||
while (true) {
|
||
const { bytesRead } = await chunkHandle.read(
|
||
copyBuffer,
|
||
0,
|
||
copyBuffer.length,
|
||
null,
|
||
)
|
||
if (bytesRead === 0) {
|
||
break
|
||
}
|
||
|
||
copiedChunkBytes += bytesRead
|
||
copiedTotalBytes += bytesRead
|
||
if (copiedChunkBytes > chunk.size || copiedTotalBytes > session.fileSize) {
|
||
throw new Error(`上传分片大小超过会话声明: ${path.basename(chunk.path)}`)
|
||
}
|
||
|
||
let writtenOffset = 0
|
||
while (writtenOffset < bytesRead) {
|
||
const { bytesWritten } = await outputHandle.write(
|
||
copyBuffer,
|
||
writtenOffset,
|
||
bytesRead - writtenOffset,
|
||
null,
|
||
)
|
||
if (bytesWritten <= 0) {
|
||
throw new Error('上传文件组装写入失败')
|
||
}
|
||
writtenOffset += bytesWritten
|
||
}
|
||
}
|
||
} finally {
|
||
await chunkHandle.close()
|
||
}
|
||
|
||
if (copiedChunkBytes !== chunk.size) {
|
||
throw new Error(`上传分片大小在组装期间发生变化: ${path.basename(chunk.path)}`)
|
||
}
|
||
}
|
||
|
||
if (copiedTotalBytes !== session.fileSize) {
|
||
throw new Error('上传文件组装大小与会话声明不一致')
|
||
}
|
||
} finally {
|
||
await outputHandle.close();
|
||
}
|
||
|
||
const outputStats = await lstat(outputPath);
|
||
if (
|
||
outputStats.isSymbolicLink() ||
|
||
!outputStats.isFile() ||
|
||
outputStats.size !== session.fileSize
|
||
) {
|
||
console.warn(
|
||
`[Upload Session] Assembled size mismatch: ${normalizedUploadId}, expected=${session.fileSize}, actual=${outputStats.size}`,
|
||
);
|
||
await unlink(outputPath).catch(() => undefined)
|
||
outputCreated = false
|
||
return null;
|
||
}
|
||
|
||
console.log(`[Upload Session] Assembled file: ${outputPath} (${outputStats.size} bytes)`);
|
||
return outputPath;
|
||
} catch (error) {
|
||
if (outputCreated) {
|
||
await unlink(outputPath).catch(() => undefined)
|
||
}
|
||
console.error(`[Upload Session] Failed to get uploaded file for ${normalizedUploadId}:`, error);
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 删除上传会话
|
||
*/
|
||
export async function deleteUploadSession(uploadId: string): Promise<void> {
|
||
deleteMemorySession(uploadId)
|
||
await withRedisFallback(
|
||
'deleteUploadSession',
|
||
() => redisModule!.deleteUploadSession(uploadId),
|
||
async () => {}
|
||
)
|
||
}
|
||
|
||
export async function deleteOwnedUploadSession(uploadId: string, userId: string): Promise<boolean> {
|
||
const session = await getOwnedUploadSession(uploadId, userId)
|
||
if (!session) {
|
||
return false
|
||
}
|
||
|
||
await deleteUploadSession(uploadId)
|
||
return true
|
||
}
|
||
|
||
/**
|
||
* 检查会话是否存在
|
||
*/
|
||
export async function hasUploadSession(uploadId: string): Promise<boolean> {
|
||
return withRedisFallback(
|
||
'hasUploadSession',
|
||
() => redisModule!.hasUploadSession(uploadId),
|
||
async () => memorySessions.has(uploadId)
|
||
)
|
||
}
|
||
|
||
/**
|
||
* 更新块信息
|
||
*/
|
||
export async function updateChunk(
|
||
uploadId: string,
|
||
chunkIndex: number,
|
||
chunkPath: string
|
||
): Promise<void> {
|
||
let session = getMemorySession(uploadId)
|
||
if (!session) {
|
||
session = await getUploadSession(uploadId)
|
||
}
|
||
if (!session) {
|
||
throw new Error(`上传会话不存在: ${uploadId}`)
|
||
}
|
||
session.chunks.set(chunkIndex, chunkPath)
|
||
setMemorySession(uploadId, session)
|
||
await withRedisFallback(
|
||
'updateChunk',
|
||
() => redisModule!.updateChunk(uploadId, chunkIndex, chunkPath),
|
||
async () => {}
|
||
)
|
||
}
|
||
|
||
/**
|
||
* 更新 chunk 信息 + 写入活动/速度统计
|
||
* - bytesDelta: 本次 chunk 的字节数（通常来自 File.size）
|
||
*/
|
||
export async function updateChunkWithStats(
|
||
uploadId: string,
|
||
chunkIndex: number,
|
||
chunkPath: string,
|
||
bytesDelta: number
|
||
): Promise<void> {
|
||
let session = getMemorySession(uploadId)
|
||
if (!session) {
|
||
session = await getUploadSession(uploadId)
|
||
}
|
||
if (!session) {
|
||
throw new Error(`上传会话不存在: ${uploadId}`)
|
||
}
|
||
|
||
const now = Date.now()
|
||
const safeBytesDelta = Number.isFinite(bytesDelta) && bytesDelta > 0 ? bytesDelta : 0
|
||
|
||
session.chunks.set(chunkIndex, chunkPath)
|
||
session.lastActivityAt = now
|
||
session.uploadedBytes = (session.uploadedBytes || 0) + safeBytesDelta
|
||
|
||
// 速度估算：用上一次采样点做差
|
||
const prevAt = session._speedSampleAt
|
||
const prevBytes = session._speedSampleBytes
|
||
if (prevAt && prevBytes !== undefined) {
|
||
const dtSec = (now - prevAt) / 1000
|
||
const dBytes = session.uploadedBytes - prevBytes
|
||
if (dtSec > 0 && dBytes >= 0) {
|
||
const instant = dBytes / dtSec
|
||
// 轻量平滑，避免抖动（EMA）
|
||
const alpha = 0.3
|
||
session.currentSpeed = session.currentSpeed ? session.currentSpeed * (1 - alpha) + instant * alpha : instant
|
||
}
|
||
}
|
||
session._speedSampleAt = now
|
||
session._speedSampleBytes = session.uploadedBytes
|
||
|
||
setMemorySession(uploadId, session)
|
||
|
||
await withRedisFallback(
|
||
'updateChunkWithStats',
|
||
async () => {
|
||
// 生产环境：直接覆盖保存（redis 侧仍然调用 updateChunk 触发 saveUploadSession）
|
||
// 这里复用 updateChunk，然后再把统计字段写回（避免改动 redis 模块的签名）
|
||
await redisModule!.updateChunk(uploadId, chunkIndex, chunkPath)
|
||
await redisModule!.saveUploadSession(uploadId, session as any)
|
||
return
|
||
},
|
||
async () => {}
|
||
)
|
||
}
|
||
|
||
/**
|
||
* 刷新上传会话（保留已上传进度）
|
||
* 用于会话过期后恢复上传，不清空已上传的分片记录
|
||
*/
|
||
export async function refreshUploadSession(uploadId: string): Promise<boolean> {
|
||
return withRedisFallback(
|
||
'refreshUploadSession',
|
||
async () => {
|
||
return await redisModule!.refreshUploadSession(uploadId)
|
||
},
|
||
async () => {
|
||
// 内存Map版本：更新过期时间
|
||
const session = memorySessions.get(uploadId)
|
||
if (session) {
|
||
// 内存会话自动延长4小时
|
||
const expireTime = 4 * 60 * 60 * 1000 // 4小时
|
||
session.createdAt = Date.now() - expireTime + (4 * 60 * 60 * 1000)
|
||
console.log(`[MemoryMap] 刷新上传会话成功: ${uploadId}`)
|
||
return true
|
||
}
|
||
console.warn(`[MemoryMap] 刷新上传会话失败，会话不存在: ${uploadId}`)
|
||
return false
|
||
}
|
||
)
|
||
}
|
||
|
||
// 导出内存Map（用于向后兼容，但不推荐直接使用）
|
||
export const uploadSessions = memorySessions
|
||
|