fix(upload): 移除冗余

Author: Suyunmeng

internal/db/db.go

@@ -17,8 +17,7 @@ func Init(d *gorm.DB) {
 	if err != nil {
 		log.Fatalf("failed migrate database: %s", err.Error())
 	}
-	
-	// 清理启动前遗留的孤儿分片上传任务
+
 	if err := CleanupOrphanedSliceUploads(); err != nil {
 		log.Errorf("Failed to cleanup orphaned slice uploads: %v", err)
 	}

internal/db/slice_upload.go

@@ -70,23 +70,19 @@ func UpdateSliceUploadWithTx(su *tables.SliceUpload) error {
 // UpdateSliceStatusAtomic 原子性地更新分片状态
 func UpdateSliceStatusAtomic(taskID string, sliceNum int, status []byte) error {
 	return errors.WithStack(db.Transaction(func(tx *gorm.DB) error {
-		// 先读取当前状态
 		var su tables.SliceUpload
 		if err := tx.Where("task_id = ?", taskID).First(&su).Error; err != nil {
 			return err
 		}
 
-		// 更新分片状态
 		tables.SetSliceUploaded(su.SliceUploadStatus, sliceNum)
 
-		// 保存更新
 		return tx.Save(&su).Error
 	}))
 }
 
 // CleanupOrphanedSliceUploads 清理孤儿分片上传记录（启动时调用）
 func CleanupOrphanedSliceUploads() error {
-	// 清理超过24小时的未完成任务
 	cutoff := time.Now().Add(-24 * time.Hour)
 
 	var orphanedTasks []tables.SliceUpload
@@ -99,7 +95,6 @@ func CleanupOrphanedSliceUploads() error {
 
 	cleanedCount := 0
 	for _, task := range orphanedTasks {
-		// 清理临时文件
 		if task.TmpFile != "" {
 			if err := os.Remove(task.TmpFile); err != nil && !os.IsNotExist(err) {
 				log.Warnf("Failed to remove orphaned tmp file %s: %v", task.TmpFile, err)
@@ -108,7 +103,6 @@ func CleanupOrphanedSliceUploads() error {
 			}
 		}
 
-		// 删除数据库记录
 		if err := db.Delete(&task).Error; err != nil {
 			log.Errorf("Failed to delete orphaned slice upload task %s: %v", task.TaskID, err)
 		} else {
@@ -120,30 +114,25 @@ func CleanupOrphanedSliceUploads() error {
 		log.Infof("Cleaned up %d orphaned slice upload tasks", cleanedCount)
 	}
 
-	// 额外清理：扫描临时目录中的孤儿文件
 	return cleanupOrphanedTempFiles()
 }
 
 // cleanupOrphanedTempFiles 清理临时目录中的孤儿文件
 func cleanupOrphanedTempFiles() error {
-	// 获取临时目录路径，使用共享的tempdir包
 	tempDir := conf.GetPersistentTempDir()
 
-	// 检查临时目录是否存在
 	if _, err := os.Stat(tempDir); os.IsNotExist(err) {
 		log.Debugf("Temp directory does not exist: %s", tempDir)
 		return nil
 	}
 
-	// 获取所有活跃的分片上传任务的临时文件列表
 	var activeTasks []tables.SliceUpload
 	if err := db.Where("tmp_file IS NOT NULL AND tmp_file != '' AND status IN (?, ?)",
 		tables.SliceUploadStatusWaiting,
 		tables.SliceUploadStatusUploading).Find(&activeTasks).Error; err != nil {
 		return errors.WithStack(err)
 	}
 
-	// 构建活跃文件的映射表
 	activeFiles := make(map[string]bool)
 	for _, task := range activeTasks {
 		if task.TmpFile != "" {
@@ -152,43 +141,36 @@ func cleanupOrphanedTempFiles() error {
 	}
 
 	cleanedCount := 0
-	cutoff := time.Now().Add(-24 * time.Hour) // 只清理超过24小时的文件
+	cutoff := time.Now().Add(-24 * time.Hour)
 
-	// 遍历临时目录
 	err := filepath.WalkDir(tempDir, func(path string, d fs.DirEntry, err error) error {
 		if err != nil {
 			log.Warnf("Failed to access path %s: %v", path, err)
 			return nil // 继续处理其他文件
 		}
 
-		// 跳过目录
 		if d.IsDir() {
 			return nil
 		}
 
-		// 只处理分片上传临时文件（以slice_upload_开头）
 		if !strings.HasPrefix(d.Name(), "slice_upload_") {
 			return nil
 		}
 
-		// 检查文件是否在活跃任务列表中
 		if activeFiles[path] {
 			return nil // 文件仍在使用中，跳过
 		}
 
-		// 检查文件修改时间
 		info, err := d.Info()
 		if err != nil {
 			log.Warnf("Failed to get file info for %s: %v", path, err)
 			return nil
 		}
 
-		// 只清理超过24小时的文件
 		if info.ModTime().After(cutoff) {
 			return nil
 		}
 
-		// 删除孤儿文件
 		if err := os.Remove(path); err != nil {
 			log.Warnf("Failed to remove orphaned temp file %s: %v", path, err)
 		} else {

internal/driver/driver.go

@@ -84,23 +84,18 @@ type Remove interface {
 	Remove(ctx context.Context, obj model.Obj) error
 }
 
-// IUploadInfo 上传信息接口
 type IUploadInfo interface {
 	GetUploadInfo() *model.UploadInfo
 }
 
-// IPreup 预上传接口
 type IPreup interface {
 	Preup(ctx context.Context, srcobj model.Obj, req *reqres.PreupReq) (*model.PreupInfo, error)
 }
 
-// ISliceUpload 分片上传接口
 type ISliceUpload interface {
-	// SliceUpload 分片上传
 	SliceUpload(ctx context.Context, req *tables.SliceUpload, sliceno uint, file io.Reader) error
 }
 
-// IUploadSliceComplete 分片上传完成接口
 type IUploadSliceComplete interface {
 	UploadSliceComplete(ctx context.Context, req *tables.SliceUpload) error
 }

internal/fs/fs.go

@@ -190,19 +190,14 @@ func PutURL(ctx context.Context, path, dstName, urlStr string) error {
 	return op.PutURL(ctx, storage, dstDirActualPath, dstName, urlStr)
 }
 
-/// 分片上传功能--------------------------------------------------------------------
-
-// Preup 预上传 - 使用新的管理器重构
 func Preup(c context.Context, s driver.Driver, actualPath string, req *reqres.PreupReq) (*reqres.PreupResp, error) {
 	return getGlobalSliceManager().CreateSession(c, s, actualPath, req)
 }
 
-// UploadSlice 流式上传切片 - 使用新的管理器重构，支持流式上传
 func UploadSlice(ctx context.Context, storage driver.Driver, req *reqres.UploadSliceReq, reader io.Reader) error {
 	return getGlobalSliceManager().UploadSlice(ctx, storage, req, reader)
 }
 
-// SliceUpComplete 完成分片上传 - 使用新的管理器重构
 func SliceUpComplete(ctx context.Context, storage driver.Driver, taskID string) (*reqres.UploadSliceCompleteResp, error) {
 	return getGlobalSliceManager().CompleteUpload(ctx, storage, taskID)
 }

internal/fs/sliceup.go

@@ -54,10 +54,8 @@ func (m *SliceUploadManager) CreateSession(ctx context.Context, storage driver.D
 	}
 	user := ctx.Value(conf.UserKey).(*model.User)
 
-	// 生成唯一的TaskID
 	taskID := uuid.New().String()
 
-	//创建新的上传任务
 	createsu := &tables.SliceUpload{
 		TaskID:       taskID,
 		DstPath:      req.Path,
@@ -121,13 +119,11 @@ func (m *SliceUploadManager) CreateSession(ctx context.Context, storage driver.D
 	}, nil
 }
 
-// getOrLoadSession 获取或加载会话，提高代码复用性
 func (m *SliceUploadManager) getOrLoadSession(taskID string) (*SliceUploadSession, error) {
 	session, err, _ := m.sessionG.Do(taskID, func() (*SliceUploadSession, error) {
 		if s, ok := m.cache.Load(taskID); ok {
 			return s.(*SliceUploadSession), nil
 		}
-		// 首次加载，需要从数据库获取
 		su, err := db.GetSliceUploadByTaskID(taskID)
 		if err != nil {
 			return nil, errors.WithMessagef(err, "failed get slice upload [%s]", taskID)
@@ -141,21 +137,20 @@ func (m *SliceUploadManager) getOrLoadSession(taskID string) (*SliceUploadSessio
 	return session, err
 }
 
-// UploadSlice 流式上传分片 - 支持流式上传，避免表单上传的内存占用
+// UploadSlice 流式上传分片
 func (m *SliceUploadManager) UploadSlice(ctx context.Context, storage driver.Driver, req *reqres.UploadSliceReq, reader io.Reader) error {
 	session, err := m.getOrLoadSession(req.TaskID)
 	if err != nil {
 		log.Errorf("failed to get session: %+v", err)
 		return err
 	}
 
-	// 确保并发安全的错误处理
 	defer func() {
 		if err != nil {
 			session.mutex.Lock()
 			session.Status = tables.SliceUploadStatusFailed
 			session.Message = err.Error()
-			updateData := *session.SliceUpload // 复制数据避免锁持有时间过长
+			updateData := *session.SliceUpload
 			session.mutex.Unlock()
 
 			if updateErr := db.UpdateSliceUpload(&updateData); updateErr != nil {
@@ -166,7 +161,6 @@ func (m *SliceUploadManager) UploadSlice(ctx context.Context, storage driver.Dri
 
 	// 使用锁保护状态检查
 	session.mutex.Lock()
-	// 检查分片是否已上传过
 	if tables.IsSliceUploaded(session.SliceUploadStatus, int(req.SliceNum)) {
 		session.mutex.Unlock()
 		log.Warnf("slice already uploaded,req:%+v", req)
@@ -178,27 +172,23 @@ func (m *SliceUploadManager) UploadSlice(ctx context.Context, storage driver.Dri
 	if req.SliceHash != "" {
 		session.mutex.Lock()
 
-		// 验证分片hash值
-		if req.SliceNum == 0 { // 第一个分片，slicehash是所有的分片hash
+		if req.SliceNum == 0 {
 			hs := strings.Split(req.SliceHash, ",")
 			if len(hs) != int(session.SliceCnt) {
 				session.mutex.Unlock()
 				err := fmt.Errorf("slice hash count mismatch, expected %d, got %d", session.SliceCnt, len(hs))
 				log.Error("slice hash count mismatch", req, err)
 				return err
 			}
-			session.SliceHash = req.SliceHash // 存储完整的hash字符串
+			session.SliceHash = req.SliceHash
 		} else {
-			// 非第0个分片，不覆盖 SliceHash，保持完整的hash列表
 			log.Debugf("Slice %d hash: %s (keeping complete hash list)", req.SliceNum, req.SliceHash)
 		}
 		session.mutex.Unlock()
 	}
 
-	// 根据存储类型处理分片上传
 	switch s := storage.(type) {
 	case driver.ISliceUpload:
-		// Native slice upload: directly pass stream data, let frontend handle retry and recovery
 		if err := s.SliceUpload(ctx, session.SliceUpload, req.SliceNum, reader); err != nil {
 			log.Errorf("Native slice upload failed - TaskID: %s, SliceNum: %d, Error: %v", 
 				req.TaskID, req.SliceNum, err)
@@ -207,13 +197,12 @@ func (m *SliceUploadManager) UploadSlice(ctx context.Context, storage driver.Dri
 		log.Debugf("Native slice upload success - TaskID: %s, SliceNum: %d", 
 			req.TaskID, req.SliceNum)
 
-	default: //其他网盘先缓存到本地
+	default:
 		if err := session.ensureTmpFile(); err != nil {
 			log.Error("ensureTmpFile error", req, err)
 			return err
 		}
 
-		// 流式复制，减少内存占用
 		sw := &sliceWriter{
 			file:   session.tmpFile,
 			offset: int64(req.SliceNum) * int64(session.SliceSize),
@@ -225,10 +214,9 @@ func (m *SliceUploadManager) UploadSlice(ctx context.Context, storage driver.Dri
 		}
 	}
 
-	// 原子性更新分片状态
 	session.mutex.Lock()
 	tables.SetSliceUploaded(session.SliceUploadStatus, int(req.SliceNum))
-	updateData := *session.SliceUpload // 复制数据
+	updateData := *session.SliceUpload
 	session.mutex.Unlock()
 
 	err = db.UpdateSliceUpload(&updateData)
@@ -263,7 +251,6 @@ func (m *SliceUploadManager) CompleteUpload(ctx context.Context, storage driver.
 	}
 
 	defer func() {
-		// 确保资源清理和缓存删除
 		session.cleanup()
 		m.cache.Delete(session.TaskID)
 
@@ -278,7 +265,6 @@ func (m *SliceUploadManager) CompleteUpload(ctx context.Context, storage driver.
 				log.Errorf("Failed to update slice upload status: %v", updateErr)
 			}
 		} else {
-			// 上传成功后从数据库中删除记录，允许重复上传
 			if deleteErr := db.DeleteSliceUploadByTaskID(session.TaskID); deleteErr != nil {
 				log.Errorf("Failed to delete slice upload record: %v", deleteErr)
 			}
@@ -293,14 +279,12 @@ func (m *SliceUploadManager) CompleteUpload(ctx context.Context, storage driver.
 			return nil, err
 		}
 
-		// 原生分片上传成功，直接返回，defer中会删除数据库记录
 		return &reqres.UploadSliceCompleteResp{
 			Complete: 1,
 			TaskID:   session.TaskID,
 		}, nil
 
 	default:
-		// 其他网盘客户端上传到本地后，上传到网盘，使用任务处理
 		session.mutex.Lock()
 		tmpFile := session.tmpFile
 		session.mutex.Unlock()
@@ -330,7 +314,6 @@ func (m *SliceUploadManager) CompleteUpload(ctx context.Context, storage driver.
 
 		if session.AsTask {
 			file.SetTmpFile(tmpFile)
-			// 防止defer中清理文件
 			session.mutex.Lock()
 			session.tmpFile = nil
 			session.TmpFile = ""
@@ -360,15 +343,12 @@ func (m *SliceUploadManager) CompleteUpload(ctx context.Context, storage driver.
 	}
 }
 
-// ensureTmpFile 确保临时文件存在且正确初始化，线程安全 - 使用持久化目录
 func (s *SliceUploadSession) ensureTmpFile() error {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
 	if s.TmpFile == "" {
-		// 使用TaskID作为文件名的一部分，确保唯一性和可识别性
 		filename := fmt.Sprintf("slice_upload_%s_%s", s.TaskID, s.Name)
-		// 清理文件名中的特殊字符
 		filename = strings.ReplaceAll(filename, "/", "_")
 		filename = strings.ReplaceAll(filename, "\\", "_")
 		filename = strings.ReplaceAll(filename, ":", "_")
@@ -392,7 +372,6 @@ func (s *SliceUploadSession) ensureTmpFile() error {
 		// 更新数据库中的临时文件路径
 		if updateErr := db.UpdateSliceUpload(s.SliceUpload); updateErr != nil {
 			log.Errorf("Failed to update temp file path in database: %v", updateErr)
-			// 不返回错误，因为文件已经创建成功，只是数据库更新失败
 		}
 
 		log.Debugf("Created persistent temp file: %s", tmpPath)
@@ -410,7 +389,6 @@ func (s *SliceUploadSession) ensureTmpFile() error {
 	return nil
 }
 
-// cleanup 清理资源，线程安全 - 保持原始实现
 func (s *SliceUploadSession) cleanup() {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()

server/handles/fsup.go

@@ -238,7 +238,6 @@ func FsPreup(c *gin.Context) {
 		return
 	}
 
-	// 基本参数验证
 	if req.Name == "" {
 		common.ErrorResp(c, fmt.Errorf("file name is required"), 400)
 		return
@@ -274,7 +273,6 @@ func FsUpSlice(c *gin.Context) {
 		}
 		_ = c.Request.Body.Close()
 	}()
-	// 从HTTP头获取参数
 	taskID := c.GetHeader("X-Task-ID")
 	if taskID == "" {
 		common.ErrorResp(c, fmt.Errorf("X-Task-ID header is required"), 400)
@@ -295,14 +293,12 @@ func FsUpSlice(c *gin.Context) {
 
 	sliceHash := c.GetHeader("X-Slice-Hash")
 
-	// 构建请求对象
 	req := &reqres.UploadSliceReq{
 		TaskID:    taskID,
 		SliceHash: sliceHash,
 		SliceNum:  uint(sliceNum),
 	}
 
-	// 获取请求体作为流
 	reader := c.Request.Body
 	if reader == nil {
 		common.ErrorResp(c, fmt.Errorf("request body is required"), 400)
@@ -311,7 +307,6 @@ func FsUpSlice(c *gin.Context) {
 
 	storage := c.Request.Context().Value(conf.StorageKey).(driver.Driver)
 
-	// 调用流式上传分片函数
 	err = fs.UploadSlice(c.Request.Context(), storage, req, reader)
 	if err != nil {
 		common.ErrorResp(c, fmt.Errorf("upload slice failed: %w", err), 500)