package fileupload import ( "crypto/md5" "errors" "fmt" "io" "mime/multipart" "os" "path" "path/filepath" "strconv" "strings" "time" ) // service 核心业务,持 Options + Repository。 type service struct { opts Options repo Repository } func newService(opts Options, repo Repository) *service { return &service{opts: opts, repo: repo} } func (s *service) chunkDir(uploadID string) string { return filepath.Join(s.opts.ChunkRoot, uploadID) } func (s *service) fileURL(storagePath string) string { return s.opts.URLPrefix + "/" + storagePath } // ── init ──────────────────────────────────────────────────────────── type InitReq struct { FileName string `json:"fileName" binding:"required,max=255"` FileSize int64 `json:"fileSize" binding:"min=0"` FileMD5 string `json:"fileMD5" binding:"required,len=32"` MimeType string `json:"mimeType" binding:"max=128"` ChunkSize int64 `json:"chunkSize" binding:"required,min=1"` TotalChunk int `json:"totalChunk" binding:"required,min=1"` // TargetDir 目标目录(相对 CloudDir,/ 分隔),空=根目录(固定路径模式)。 TargetDir string `json:"targetDir" binding:"max=500"` } func (s *service) init(req *InitReq, operator string, tenantID int64) (any, error) { if req.ChunkSize <= 0 || req.ChunkSize > s.opts.maxChunkBytes() { return nil, fmt.Errorf("chunkSize 超出范围 (0, %d]", s.opts.maxChunkBytes()) } if req.FileSize < 0 { return nil, errors.New("fileSize 非法") } if len(req.FileMD5) != 32 { return nil, errors.New("fileMD5 必须为 32 位十六进制") } if expect := expectChunks(req.FileSize, req.ChunkSize); req.TotalChunk != expect { return nil, fmt.Errorf("totalChunk(%d) 与 fileSize(%d)/chunkSize(%d) 不匹配,应为 %d", req.TotalChunk, req.FileSize, req.ChunkSize, expect) } if s.opts.MaxTotalSizeMB > 0 && req.FileSize > s.opts.MaxTotalSizeMB*1024*1024 { return nil, fmt.Errorf("文件超过总大小上限 %dMB", s.opts.MaxTotalSizeMB) } targetDir, err := SanitizeDir(req.TargetDir) if err != nil { return nil, err } // 1. 秒传:同 MD5 已完成任务 done, err := s.repo.FindDoneByMD5(req.FileMD5) if err != nil { return nil, err } if done != nil && done.StoragePath != "" { if hit, ok := s.instantHit(done, targetDir); ok { if s.opts.OnInstantHit != nil { if err := s.opts.OnInstantHit(hit); err != nil { return nil, err } } return map[string]any{ "uploadId": "", "skipUpload": true, "fileUrl": s.fileURL(hit.StoragePath), }, nil } // 秒传不可用(旧物理文件缺失/复制失败)→ 落入正常上传流程 } // 2. 断点续传:同 MD5 进行中任务(目标目录不同则不复用,另起新任务) resuming, err := s.repo.FindUploadingByMD5(req.FileMD5) if err != nil { return nil, err } if resuming != nil && resuming.TargetDir == targetDir { indexes, err := s.repo.ListChunkIndexes(resuming.UploadId) if err != nil { return nil, err } return map[string]any{ "uploadId": resuming.UploadId, "skipUpload": false, "uploadedChunks": indexes, }, nil } // 3. 新建任务 name := sanitizeFileName(req.FileName) if name == "" { return nil, errors.New("文件名非法") } task := TaskInfo{ UploadId: newUploadID(), FileName: name, FileSize: req.FileSize, FileMd5: req.FileMD5, MimeType: req.MimeType, ChunkSize: req.ChunkSize, TotalChunk: int32(req.TotalChunk), Status: StatusUploading, TargetDir: targetDir, Operator: operator, TenantId: tenantID, } if err := s.repo.CreateTask(&task); err != nil { return nil, err } return map[string]any{ "uploadId": task.UploadId, "skipUpload": false, "uploadedChunks": []int32{}, }, nil } // instantHit 处理秒传命中: // - 目标目录为空(固定模式)或与旧任务一致 → 直接复用旧 storagePath,行为与 v1.0.x 一致; // - 跨目录 → 把旧物理文件复制进目标目录,返回携带新 storagePath/TargetDir 的命中视图; // // ok=false 表示无法秒传(旧物理文件缺失/复制失败),调用方回退正常分片上传。 func (s *service) instantHit(done *TaskInfo, targetDir string) (TaskInfo, bool) { hit := *done if targetDir == "" || targetDir == done.TargetDir { return hit, true } src := s.physicalPath(done.StoragePath) if _, err := os.Stat(src); err != nil { return TaskInfo{}, false } name := path.Base(strings.ReplaceAll(done.StoragePath, "\\", "/")) dstDir := filepath.Join(s.opts.CloudDir, filepath.FromSlash(targetDir)) if err := os.MkdirAll(dstDir, 0o766); err != nil { return TaskInfo{}, false } if err := copyFile(src, filepath.Join(dstDir, name)); err != nil { return TaskInfo{}, false } hit.StoragePath = buildStoragePath(s.opts.WebPathPrefix, targetDir, name) hit.TargetDir = targetDir return hit, true } // ── chunk ─────────────────────────────────────────────────────────── func (s *service) chunk(uploadID string, chunkIndex int, fh *multipart.FileHeader) (any, error) { if uploadID == "" { return nil, errors.New("缺少 uploadId") } task, err := s.repo.GetTaskByUploadID(uploadID) if err != nil { return nil, err } if task == nil { return nil, errors.New("上传任务不存在") } if task.Status != StatusUploading { return nil, fmt.Errorf("任务状态异常: %s", task.Status) } if chunkIndex < 0 || int32(chunkIndex) >= task.TotalChunk { return nil, fmt.Errorf("chunkIndex %d 超出范围 [0, %d)", chunkIndex, task.TotalChunk) } if fh.Size > s.opts.maxChunkBytes() { return nil, fmt.Errorf("分片超过单片上限 %d MB", s.opts.maxChunkBytes()/1024/1024) } // 幂等:已确认的分片直接成功 exists, err := s.repo.ChunkExists(uploadID, int32(chunkIndex)) if err != nil { return nil, err } if exists { return map[string]any{"chunkIndex": chunkIndex}, nil } dir := s.chunkDir(uploadID) if err := os.MkdirAll(dir, 0o766); err != nil { return nil, err } chunkPath := filepath.Join(dir, strconv.Itoa(chunkIndex)) dst, err := os.Create(chunkPath) if err != nil { return nil, err } src, err := fh.Open() if err != nil { _ = dst.Close() return nil, err } hash := md5.New() written, copyErr := io.Copy(io.MultiWriter(dst, hash), src) closeErr := dst.Close() _ = src.Close() if copyErr != nil { return nil, copyErr } if closeErr != nil { return nil, closeErr } if err := s.repo.CreateChunk(&ChunkInfo{ UploadId: uploadID, ChunkIndex: int32(chunkIndex), ChunkSize: written, ChunkMd5: fmt.Sprintf("%x", hash.Sum(nil)), }); err != nil { // 并发下另一请求刚写入同分片:唯一冲突视为幂等成功 again, e2 := s.repo.ChunkExists(uploadID, int32(chunkIndex)) if e2 == nil && again { return map[string]any{"chunkIndex": chunkIndex}, nil } return nil, err } return map[string]any{"chunkIndex": chunkIndex}, nil } // ── merge ─────────────────────────────────────────────────────────── type MergeReq struct { UploadId string `json:"uploadId" binding:"required"` } func (s *service) merge(uploadID string) (any, error) { task, err := s.repo.GetTaskByUploadID(uploadID) if err != nil { return nil, err } if task == nil { return nil, errors.New("上传任务不存在") } // 目标目录在 init 时已 sanitize,此处防御性重校验(防宿主存脏数据) targetDir, err := SanitizeDir(task.TargetDir) if err != nil { return nil, err } // 幂等:已完成直接返回原地址;合并中拒绝重复触发 if task.Status == StatusDone && task.StoragePath != "" { return map[string]any{"fileUrl": s.fileURL(task.StoragePath)}, nil } if task.Status == StatusMerging { return nil, errors.New("任务正在合并中,请勿重复提交") } count, err := s.repo.CountChunks(uploadID) if err != nil { return nil, err } if count < int(task.TotalChunk) { return nil, fmt.Errorf("分片不完整,已上传 %d/%d", count, task.TotalChunk) } _ = s.repo.SetTaskStatus(task.Id, StatusMerging) // 先写 .part,全部成功后改名,避免半成品文件残留 partPath := filepath.Join(s.chunkDir(uploadID), "merged.part") part, err := os.Create(partPath) if err != nil { _ = s.repo.SetTaskStatus(task.Id, StatusError) return nil, err } hash := md5.New() for i := int32(0); i < task.TotalChunk; i++ { src, err := os.Open(filepath.Join(s.chunkDir(uploadID), strconv.Itoa(int(i)))) if err != nil { _ = part.Close() _ = os.Remove(partPath) _ = s.repo.SetTaskStatus(task.Id, StatusError) return nil, fmt.Errorf("读取分片 %d 失败: %w", i, err) } _, copyErr := io.Copy(part, io.TeeReader(src, hash)) _ = src.Close() if copyErr != nil { _ = part.Close() _ = os.Remove(partPath) _ = s.repo.SetTaskStatus(task.Id, StatusError) return nil, fmt.Errorf("合并分片 %d 失败: %w", i, copyErr) } } if err := part.Close(); err != nil { _ = os.Remove(partPath) _ = s.repo.SetTaskStatus(task.Id, StatusError) return nil, err } // 整文件 MD5 校验:保证秒传可信 if actual := fmt.Sprintf("%x", hash.Sum(nil)); actual != task.FileMd5 { _ = os.Remove(partPath) _ = s.repo.SetTaskStatus(task.Id, StatusError) return nil, fmt.Errorf("合并结果校验失败:实际 MD5 %s 与声称的 %s 不一致", actual, task.FileMd5) } // 改名进宿主产物目录(targetDir 是 CloudDir 下的真实子目录,空=根目录) finalDir := s.opts.CloudDir if targetDir != "" { finalDir = filepath.Join(s.opts.CloudDir, filepath.FromSlash(targetDir)) } if err := os.MkdirAll(finalDir, 0o766); err != nil { _ = os.Remove(partPath) _ = s.repo.SetTaskStatus(task.Id, StatusError) return nil, err } storageName := strconv.FormatInt(time.Now().UnixNano(), 10) + "_" + task.FileName // storagePath 是 Web 相对路径(WebPathPrefix + targetDir + 产物名), // 前缀只是 URL/拼接约定,targetDir 才是物理子目录 finalPath := filepath.Join(finalDir, storageName) storagePath := buildStoragePath(s.opts.WebPathPrefix, targetDir, storageName) if err := os.Rename(partPath, finalPath); err != nil { _ = os.Remove(partPath) _ = s.repo.SetTaskStatus(task.Id, StatusError) return nil, err } task.Status = StatusDone task.StoragePath = storagePath if err := s.repo.UpdateTaskMerged(task.Id, storagePath); err != nil { return nil, err } // 宿主业务衔接(如写云盘记录行);失败则 merge 整体失败 if s.opts.OnMerged != nil { if err := s.opts.OnMerged(*task); err != nil { return nil, err } } // 分片目录异步清理 go func(dir string) { _ = os.RemoveAll(dir) }(s.chunkDir(uploadID)) return map[string]any{"fileUrl": s.fileURL(storagePath)}, nil } // ── check / task / abort ──────────────────────────────────────────── // check 秒传/续传预检,targetDir 与 init 同源: // - done 命中照常返回(init 会按目录做 copy-on-hit); // - uploading 命中仅在目标目录一致时返回(目录不同时 init 会另起新任务,此处报 none)。 func (s *service) check(fileMD5, targetDir string) (any, error) { if len(fileMD5) != 32 { return map[string]any{"status": "none"}, nil } dir, err := SanitizeDir(targetDir) if err != nil { return map[string]any{"status": "none"}, nil } done, err := s.repo.FindDoneByMD5(fileMD5) if err != nil { return nil, err } if done != nil && done.StoragePath != "" { return map[string]any{ "status": StatusDone, "fileUrl": s.fileURL(done.StoragePath), }, nil } resuming, err := s.repo.FindUploadingByMD5(fileMD5) if err != nil { return nil, err } if resuming != nil && resuming.TargetDir == dir { indexes, err := s.repo.ListChunkIndexes(resuming.UploadId) if err != nil { return nil, err } return map[string]any{ "status": StatusUploading, "uploadId": resuming.UploadId, "uploadedChunks": indexes, }, nil } return map[string]any{"status": "none"}, nil } func (s *service) taskInfo(uploadID string) (any, error) { task, err := s.repo.GetTaskByUploadID(uploadID) if err != nil { return nil, err } if task == nil { return nil, errors.New("任务不存在") } indexes, err := s.repo.ListChunkIndexes(uploadID) if err != nil { return nil, err } return map[string]any{"task": task, "uploadedChunks": indexes}, nil } func (s *service) abort(uploadID string) (any, error) { task, err := s.repo.GetTaskByUploadID(uploadID) if err != nil { return nil, err } if task == nil { return nil, errors.New("上传任务不存在") } if task.Status == StatusDone { return nil, errors.New("任务已完成,无需取消") } if task.Status == StatusMerging { return nil, errors.New("任务正在合并中,无法取消") } if err := s.repo.DeleteChunks(uploadID); err != nil { return nil, err } if err := s.repo.DeleteTaskByUploadID(uploadID); err != nil { return nil, err } go func(dir string) { _ = os.RemoveAll(dir) }(s.chunkDir(uploadID)) return nil, nil }