diff --git a/README.md b/README.md index 5cf2ef8..846f5cc 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # fileupload 通用大文件分片上传 Go/Gin 服务端核心(独立库,原 zogo/fileupload 子包)。 -六接口(init/chunk/merge/check/task/abort)+ 分片幂等 + 断点续传 + `.part` 原子改名 + 合并整文件 MD5 校验 + 秒传。 +六接口(init/chunk/merge/check/task/abort)+ 分片幂等 + 断点续传 + `.part` 原子改名 + 合并整文件 MD5 校验 + 秒传 + 目标目录(targetDir)。 不依赖宿主的响应封装、ORM 或业务表——存储走 `Repository` 接口注入,业务衔接走 `Options` 回调。 ## 用法 @@ -11,7 +11,7 @@ import "git.zeroonesoft.cn/golib/fileupload" opts := fileupload.Options{ ChunkRoot: "./data/upload-chunks", // 分片临时目录(建议在静态根之外) - CloudDir: "./data/uploads/cloud", // 合并产物目录 + CloudDir: "./data/uploads/cloud", // 合并产物目录(targetDir 的根) WebPathPrefix: "cloud", // storagePath 的 Web 前缀(须与 CloudDir 相对静态根的子目录一致) URLPrefix: "/uploads", // fileUrl = URLPrefix + "/" + storagePath MaxTotalSizeMB: 2048, // 0=不限 @@ -20,7 +20,7 @@ opts := fileupload.Options{ return nil }, OnInstantHit: func(t fileupload.TaskInfo) error { - // 秒传命中:宿主确保业务侧可见行(可空回调) + // 秒传命中:宿主确保业务侧可见行(可空回调);跨目录命中时携带新 StoragePath/TargetDir return nil }, Envelope: func(c *gin.Context, resp any, err error) { @@ -36,6 +36,29 @@ _ = fileupload.Mount(api.Group("/upload"), opts, myRepo, `Repository`(12 个方法)由宿主按自己的表结构实现;软删/硬删自定,但删除后查询必须不可见。 鉴权由宿主路由组的 JWT 中间件统一负责,本库不感知。 +## 目标目录(targetDir) + +上传位置不再固定:init 请求可带 `targetDir`(相对 `CloudDir`,`/` 分隔,**空=根目录**), +合并产物物理落盘 `CloudDir//<产物名>`,storagePath = `WebPathPrefix + "/" + targetDir + "/" + 产物名`。 +`/check` 亦接受可选 query `targetDir`。宿主可用 `fileupload.SanitizeDir` 复用同款目录校验 +(拒绝 `..`/`.`/空段/绝对路径/Windows 非法字符,限长 500、限深 16)。 + +三档宿主形态(服务端无模式概念,字段缺省即固定路径模式,v1.0.x 行为不变): + +| 形态 | 客户端行为 | 服务端表现 | +|---|---|---| +| 固定路径(默认) | 不传 targetDir | 与 v1.0.x 完全一致,落 CloudDir 根 | +| 固定子目录 | 每次上传带同一 targetDir | 固定落该子目录 | +| 网盘式自选 | 用户选目录后随 init 传入 | 落所选子目录,`MkdirAll` 自动创建 | + +语义要点: + +- **秒传 copy-on-hit**:秒传按全局 MD5 命中。同目录(含根目录)命中直接复用旧文件;跨目录命中由库把旧物理文件复制进目标目录,`OnInstantHit` 携带新 `StoragePath`/`TargetDir`;旧物理文件缺失或复制失败自动回退为正常分片上传。 +- **断点续传按目录隔离**:同 MD5 在途任务目标目录不同时不复用,另起新任务。 +- **宿主须持久化 `TaskInfo.TargetDir`**(建议 `upload_task` 加 `target_dir varchar(500)`):不持久化时目录上传的断点续传会退化为每次新建任务,其余功能不受影响。 + ## 版本 -- v0.1.0 首版:自 FileUpload 项目 restful/upload 模块抽出 +- v1.1.0 新增 targetDir:init/check 支持目标目录,跨目录秒传 copy-on-hit,续传按目录隔离;Repository 接口零变更 +- v1.0.0 首个独立 tag:自 zogo/fileupload v0.3.2 抽出 +- v0.1.0 初版:自 FileUpload 项目 restful/upload 模块抽出 diff --git a/dir.go b/dir.go new file mode 100644 index 0000000..744628b --- /dev/null +++ b/dir.go @@ -0,0 +1,98 @@ +package fileupload + +import ( + "fmt" + "io" + "os" + "path/filepath" + "strings" +) + +// 目标目录(targetDir)约束:相对存储根(Options.CloudDir)。 +const ( + MaxDirLength = 500 // 总长度上限 + MaxDirDepth = 16 // 层级深度上限 +) + +// SanitizeDir 规范客户端传入的目标目录,返回以 / 分隔的相对路径; +// 空串表示根目录(固定路径模式)。防路径穿越规则: +// - 反斜杠归一为 /,剔除控制字符,去首尾 /,按段校验; +// - 拒绝 . / .. 段、空段、超长、超深、段内 Windows 非法字符(: * ? " < > |); +// - 一律按相对路径处理(前导 / 仅剥离,不表示绝对路径)。 +func SanitizeDir(dir string) (string, error) { + dir = strings.Map(func(r rune) rune { + if r < 0x20 { + return -1 + } + return r + }, strings.ReplaceAll(dir, "\\", "/")) + dir = strings.TrimSpace(dir) + dir = strings.Trim(dir, "/") + if dir == "" { + return "", nil + } + if len(dir) > MaxDirLength { + return "", fmt.Errorf("目标目录超过长度上限 %d", MaxDirLength) + } + segs := strings.Split(dir, "/") + if len(segs) > MaxDirDepth { + return "", fmt.Errorf("目标目录超过层级上限 %d", MaxDirDepth) + } + for _, seg := range segs { + switch seg { + case "", ".", "..": + return "", fmt.Errorf("目标目录含非法段 %q", seg) + } + if strings.ContainsAny(seg, `:*?"<>|`) { + return "", fmt.Errorf("目标目录段 %q 含非法字符", seg) + } + } + return strings.Join(segs, "/"), nil +} + +// buildStoragePath 拼 storagePath(Web 相对路径): +// WebPathPrefix + "/" + targetDir + "/" + 产物名(前缀与目录为空时跳过对应段)。 +func buildStoragePath(webPathPrefix, targetDir, storageName string) string { + parts := make([]string, 0, 3) + if webPathPrefix != "" { + parts = append(parts, webPathPrefix) + } + if targetDir != "" { + parts = append(parts, targetDir) + } + parts = append(parts, storageName) + return strings.Join(parts, "/") +} + +// physicalPath 由 storagePath(Web 相对路径)反解物理落盘路径: +// 剥离 WebPathPrefix 后相对 CloudDir。秒传复制旧文件时用。 +func (s *service) physicalPath(storagePath string) string { + rel := strings.ReplaceAll(storagePath, "\\", "/") + if s.opts.WebPathPrefix != "" && strings.HasPrefix(rel, s.opts.WebPathPrefix+"/") { + rel = rel[len(s.opts.WebPathPrefix)+1:] + } + return filepath.Join(s.opts.CloudDir, filepath.FromSlash(rel)) +} + +// copyFile 文件级复制,失败时清理半成品。 +func copyFile(src, dst string) error { + in, err := os.Open(src) + if err != nil { + return err + } + defer in.Close() + out, err := os.Create(dst) + if err != nil { + return err + } + if _, err = io.Copy(out, in); err != nil { + _ = out.Close() + _ = os.Remove(dst) + return err + } + if err = out.Close(); err != nil { + _ = os.Remove(dst) + return err + } + return nil +} diff --git a/fileupload.go b/fileupload.go index 33aef29..180cdae 100644 --- a/fileupload.go +++ b/fileupload.go @@ -35,6 +35,7 @@ type TaskInfo struct { ChunkSize int64 TotalChunk int32 Status string // uploading / merging / done / error + TargetDir string // 目标目录(相对 CloudDir,/ 分隔,空=根目录);宿主须持久化该字段 StoragePath string // 相对路径(如 cloud/xxx),由宿主的目录约定决定 Operator string TenantId int64 @@ -70,8 +71,9 @@ type Options struct { // CloudDir 合并产物目录(宿主的云盘/文件目录约定)。 CloudDir string // WebPathPrefix storagePath 的 Web 相对前缀(如 "cloud"),空则无前缀。 - // 最终 storagePath = WebPathPrefix + "/" + 产物名(存库、回调、BS 拼接均用该值), - // 需与 CloudDir 相对宿主静态根的子目录一致。 + // 最终 storagePath = WebPathPrefix + "/" + [targetDir + "/"] + 产物名 + // (存库、回调、BS 拼接均用该值);targetDir 物理上是 CloudDir 下的真实子目录, + // 前缀则只是 URL 拼接约定,需与 CloudDir 相对宿主静态根的子目录一致。 WebPathPrefix string // URLPrefix 拼接 fileUrl 的静态前缀,默认 "/uploads"; // 最终 fileUrl = URLPrefix + "/" + storagePath。 diff --git a/mount.go b/mount.go index 7c9c1f1..e0acdeb 100644 --- a/mount.go +++ b/mount.go @@ -9,10 +9,11 @@ import ( // Mount 挂载分片上传路由到 r(宿主自行决定路由组,如 api.Group("/upload")): // -// POST /init 初始化(JSON:fileName/fileSize/fileMD5/mimeType/chunkSize/totalChunk) +// POST /init 初始化(JSON:fileName/fileSize/fileMD5/mimeType/chunkSize/totalChunk +// + 可选 targetDir:目标目录,相对 CloudDir,空=根目录/固定路径模式) // POST /chunk 上传分片(multipart:uploadId/chunkIndex/chunk),幂等 // POST /merge 合并分片(JSON:uploadId),幂等 -// GET /check?fileMD5= 秒传/续传检查 +// GET /check?fileMD5=&targetDir= 秒传/续传检查(targetDir 可选) // GET /task/:uploadId 任务详情与已确认分片 // DELETE /task/:uploadId 取消任务并清理 // @@ -88,7 +89,7 @@ func Mount(r *gin.RouterGroup, opts Options, repo Repository, mos ...MountOption }) r.GET("/check", func(c *gin.Context) { - resp, err := svc.check(c.Query("fileMD5")) + resp, err := svc.check(c.Query("fileMD5"), c.Query("targetDir")) opts.Envelope(c, resp, err) }) diff --git a/service.go b/service.go index ef16207..1fd13d7 100644 --- a/service.go +++ b/service.go @@ -7,8 +7,10 @@ import ( "io" "mime/multipart" "os" + "path" "path/filepath" "strconv" + "strings" "time" ) @@ -39,6 +41,8 @@ type InitReq struct { 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) { @@ -58,31 +62,38 @@ func (s *service) init(req *InitReq, operator string, tenantID int64) (any, erro 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 已完成任务,复用其文件 + // 1. 秒传:同 MD5 已完成任务 done, err := s.repo.FindDoneByMD5(req.FileMD5) if err != nil { return nil, err } if done != nil && done.StoragePath != "" { - if s.opts.OnInstantHit != nil { - if err := s.opts.OnInstantHit(*done); err != nil { - return nil, err + 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 } - return map[string]any{ - "uploadId": "", - "skipUpload": true, - "fileUrl": s.fileURL(done.StoragePath), - }, nil + // 秒传不可用(旧物理文件缺失/复制失败)→ 落入正常上传流程 } - // 2. 断点续传:同 MD5 进行中任务 + // 2. 断点续传:同 MD5 进行中任务(目标目录不同则不复用,另起新任务) resuming, err := s.repo.FindUploadingByMD5(req.FileMD5) if err != nil { return nil, err } - if resuming != nil { + if resuming != nil && resuming.TargetDir == targetDir { indexes, err := s.repo.ListChunkIndexes(resuming.UploadId) if err != nil { return nil, err @@ -108,6 +119,7 @@ func (s *service) init(req *InitReq, operator string, tenantID int64) (any, erro ChunkSize: req.ChunkSize, TotalChunk: int32(req.TotalChunk), Status: StatusUploading, + TargetDir: targetDir, Operator: operator, TenantId: tenantID, } @@ -121,6 +133,33 @@ func (s *service) init(req *InitReq, operator string, tenantID int64) (any, erro }, 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) { @@ -208,6 +247,11 @@ func (s *service) merge(uploadID string) (any, error) { 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 != "" { @@ -265,20 +309,21 @@ func (s *service) merge(uploadID string) (any, error) { return nil, fmt.Errorf("合并结果校验失败:实际 MD5 %s 与声称的 %s 不一致", actual, task.FileMd5) } - // 改名进宿主产物目录 - if err := os.MkdirAll(s.opts.CloudDir, 0o766); err != nil { + // 改名进宿主产物目录(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 前缀,如 cloud/xxx), - // 物理落盘固定在 CloudDir 根下(前缀只是 URL/拼接约定,不是子目录) - finalPath := filepath.Join(s.opts.CloudDir, storageName) - storagePath := storageName - if s.opts.WebPathPrefix != "" { - storagePath = s.opts.WebPathPrefix + "/" + storageName - } + // 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) @@ -308,10 +353,17 @@ func (s *service) merge(uploadID string) (any, error) { // ── check / task / abort ──────────────────────────────────────────── -func (s *service) check(fileMD5 string) (any, error) { +// 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 @@ -326,7 +378,7 @@ func (s *service) check(fileMD5 string) (any, error) { if err != nil { return nil, err } - if resuming != nil { + if resuming != nil && resuming.TargetDir == dir { indexes, err := s.repo.ListChunkIndexes(resuming.UploadId) if err != nil { return nil, err diff --git a/service_test.go b/service_test.go new file mode 100644 index 0000000..4b081b4 --- /dev/null +++ b/service_test.go @@ -0,0 +1,418 @@ +package fileupload + +import ( + "crypto/md5" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "testing" + + "github.com/gin-gonic/gin" +) + +// ── 内存版 Repository(测试桩) ────────────────────────────────────── + +type memRepo struct { + nextID int64 + tasks map[int64]*TaskInfo + byUID map[string]int64 + chunks map[string]map[int32]struct{} +} + +func newMemRepo() *memRepo { + return &memRepo{ + tasks: map[int64]*TaskInfo{}, + byUID: map[string]int64{}, + chunks: map[string]map[int32]struct{}{}, + } +} + +func (r *memRepo) CreateTask(t *TaskInfo) error { + r.nextID++ + cp := *t + cp.Id = r.nextID + r.tasks[cp.Id] = &cp + r.byUID[cp.UploadId] = cp.Id + return nil +} + +func (r *memRepo) GetTaskByUploadID(uploadID string) (*TaskInfo, error) { + id, ok := r.byUID[uploadID] + if !ok { + return nil, nil + } + cp := *r.tasks[id] + return &cp, nil +} + +func (r *memRepo) FindDoneByMD5(md5 string) (*TaskInfo, error) { + var best *TaskInfo + for _, t := range r.tasks { + if t.FileMd5 == md5 && t.Status == StatusDone { + if best == nil || t.Id > best.Id { + best = t + } + } + } + if best == nil { + return nil, nil + } + cp := *best + return &cp, nil +} + +func (r *memRepo) FindUploadingByMD5(md5 string) (*TaskInfo, error) { + var best *TaskInfo + for _, t := range r.tasks { + if t.FileMd5 == md5 && t.Status == StatusUploading { + if best == nil || t.Id > best.Id { + best = t + } + } + } + if best == nil { + return nil, nil + } + cp := *best + return &cp, nil +} + +func (r *memRepo) UpdateTaskMerged(id int64, storagePath string) error { + r.tasks[id].Status = StatusDone + r.tasks[id].StoragePath = storagePath + return nil +} + +func (r *memRepo) SetTaskStatus(id int64, status string) error { + r.tasks[id].Status = status + return nil +} + +func (r *memRepo) CountChunks(uploadID string) (int, error) { + return len(r.chunks[uploadID]), nil +} + +func (r *memRepo) ChunkExists(uploadID string, index int32) (bool, error) { + _, ok := r.chunks[uploadID][index] + return ok, nil +} + +func (r *memRepo) CreateChunk(c *ChunkInfo) error { + if r.chunks[c.UploadId] == nil { + r.chunks[c.UploadId] = map[int32]struct{}{} + } + r.chunks[c.UploadId][c.ChunkIndex] = struct{}{} + return nil +} + +func (r *memRepo) ListChunkIndexes(uploadID string) ([]int32, error) { + var out []int32 + for i := range r.chunks[uploadID] { + out = append(out, i) + } + sort.Slice(out, func(a, b int) bool { return out[a] < out[b] }) + return out, nil +} + +func (r *memRepo) DeleteChunks(uploadID string) error { + delete(r.chunks, uploadID) + return nil +} + +func (r *memRepo) DeleteTaskByUploadID(uploadID string) error { + id, ok := r.byUID[uploadID] + if !ok { + return nil + } + delete(r.tasks, id) + delete(r.byUID, uploadID) + return nil +} + +// ── 测试工具 ───────────────────────────────────────────────────────── + +func newTestSvc(t *testing.T) (*service, *memRepo, string) { + t.Helper() + repo := newMemRepo() + cloud := t.TempDir() + svc := newService(Options{ + ChunkRoot: t.TempDir(), + CloudDir: cloud, + WebPathPrefix: "cloud", + URLPrefix: "/uploads", + Envelope: func(c *gin.Context, resp any, err error) {}, + }, repo) + return svc, repo, cloud +} + +func mustMD5(b []byte) string { return fmt.Sprintf("%x", md5.Sum(b)) } + +func seedDoneTask(t *testing.T, svc *service, repo *memRepo, md5Str, targetDir string) *TaskInfo { + t.Helper() + task := TaskInfo{ + UploadId: newUploadID(), + FileName: "old.bin", + FileMd5: md5Str, + Status: StatusDone, + TargetDir: targetDir, + StoragePath: buildStoragePath("cloud", targetDir, "old.bin"), + } + if err := repo.CreateTask(&task); err != nil { + t.Fatal(err) + } + // 物理文件落盘 + phys := svc.physicalPath(task.StoragePath) + if err := os.MkdirAll(filepath.Dir(phys), 0o766); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(phys, []byte("OLD-CONTENT"), 0o644); err != nil { + t.Fatal(err) + } + return &task +} + +// ── SanitizeDir ───────────────────────────────────────────────────── + +func TestSanitizeDir(t *testing.T) { + ok := map[string]string{ + "": "", // 根目录(固定模式) + "a": "a", + "a/": "a", // 尾随 / 归一 + "a/b": "a/b", + `a\b`: "a/b", + "/abs/": "abs", // 前导 / 剥离,按相对处理 + " docs / reports ": "docs / reports", + "云盘/资料": "云盘/资料", + "a\x01b": "ab", // 控制字符剔除 + } + for in, want := range ok { + got, err := SanitizeDir(in) + if err != nil || got != want { + t.Errorf("SanitizeDir(%q) = %q, %v; want %q", in, got, err, want) + } + } + bad := []string{ + "..", "a/../b", "a/./b", "a//b", + strings.Repeat("a/", 20) + "b", // 超深 + strings.Repeat("a", MaxDirLength+1), // 超长 + "a:b", "a*b", `a"b`, "ab", "a|b", // Windows 非法字符 + } + for _, in := range bad { + if got, err := SanitizeDir(in); err == nil { + t.Errorf("SanitizeDir(%q) = %q, want error", in, got) + } + } +} + +// ── init:秒传双模式 ────────────────────────────────────────────────── + +func initReq(name, md5Str string, targetDir string) *InitReq { + return &InitReq{ + FileName: name, + FileSize: 10, + FileMD5: md5Str, + ChunkSize: 10, + TotalChunk: 1, + TargetDir: targetDir, + } +} + +func TestInitInstantHitFixedMode(t *testing.T) { + svc, repo, cloud := newTestSvc(t) + sum := mustMD5([]byte("OLD-CONTENT")) + seedDoneTask(t, svc, repo, sum, "") + + resp, err := svc.init(initReq("old.bin", sum, ""), "op", 1) + if err != nil { + t.Fatal(err) + } + if resp.(map[string]any)["skipUpload"] != true { + t.Fatalf("固定模式应秒传: %v", resp) + } + if got := resp.(map[string]any)["fileUrl"]; got != "/uploads/cloud/old.bin" { + t.Fatalf("fileUrl = %v, 与 v1.0.x 行为不一致", got) + } + // 不应产生复制:cloud 根下仍只有 old.bin + entries, _ := os.ReadDir(cloud) + if len(entries) != 1 { + t.Fatalf("固定模式秒传不应复制文件, cloud 根条目数 = %d", len(entries)) + } +} + +func TestInitInstantHitCrossDirCopy(t *testing.T) { + svc, repo, cloud := newTestSvc(t) + sum := mustMD5([]byte("OLD-CONTENT")) + seedDoneTask(t, svc, repo, sum, "") + + var hit TaskInfo + svc.opts.OnInstantHit = func(ti TaskInfo) error { hit = ti; return nil } + + resp, err := svc.init(initReq("old.bin", sum, "docs/reports"), "op", 1) + if err != nil { + t.Fatal(err) + } + if got := resp.(map[string]any)["fileUrl"]; got != "/uploads/cloud/docs/reports/old.bin" { + t.Fatalf("fileUrl = %v", got) + } + if hit.StoragePath != "cloud/docs/reports/old.bin" || hit.TargetDir != "docs/reports" { + t.Fatalf("OnInstantHit 携带路径错误: %+v", hit) + } + // 物理复制存在且内容一致 + data, err := os.ReadFile(filepath.Join(cloud, "docs", "reports", "old.bin")) + if err != nil || string(data) != "OLD-CONTENT" { + t.Fatalf("目标目录物理副本缺失或不一致: %v, %q", err, data) + } +} + +func TestInitInstantHitFallbackWhenSourceMissing(t *testing.T) { + svc, repo, _ := newTestSvc(t) + sum := mustMD5([]byte("OLD-CONTENT")) + done := seedDoneTask(t, svc, repo, sum, "") + if err := os.Remove(svc.physicalPath(done.StoragePath)); err != nil { + t.Fatal(err) + } + + resp, err := svc.init(initReq("old.bin", sum, "docs"), "op", 1) + if err != nil { + t.Fatal(err) + } + m := resp.(map[string]any) + if m["skipUpload"] != false || m["uploadId"] == "" { + t.Fatalf("旧物理文件缺失应回退正常上传: %v", m) + } +} + +// ── init:断点续传目录隔离 ──────────────────────────────────────────── + +func TestInitResumeDirIsolation(t *testing.T) { + svc, repo, _ := newTestSvc(t) + sum := mustMD5([]byte("OLD-CONTENT")) + + first, err := svc.init(initReq("old.bin", sum, "a"), "op", 1) + if err != nil { + t.Fatal(err) + } + oldUID := first.(map[string]any)["uploadId"].(string) + + same, err := svc.init(initReq("old.bin", sum, "a"), "op", 1) + if err != nil { + t.Fatal(err) + } + if got := same.(map[string]any)["uploadId"]; got != oldUID { + t.Fatalf("同目录续传应复用任务: %v != %v", got, oldUID) + } + + diff, err := svc.init(initReq("old.bin", sum, "b"), "op", 1) + if err != nil { + t.Fatal(err) + } + newUID := diff.(map[string]any)["uploadId"].(string) + if newUID == oldUID { + t.Fatal("目录不同应另起新任务") + } + if task, _ := repo.GetTaskByUploadID(newUID); task.TargetDir != "b" { + t.Fatalf("新任务 TargetDir 应为 b: %+v", task) + } + + fixed, err := svc.init(initReq("old.bin", sum, ""), "op", 1) + if err != nil { + t.Fatal(err) + } + if got := fixed.(map[string]any)["uploadId"]; got == oldUID { + t.Fatal("固定模式与目录 a 的在途任务也应隔离") + } +} + +// ── merge:落盘目标目录 ────────────────────────────────────────────── + +func mergeTask(t *testing.T, svc *service, repo *memRepo, targetDir, content string) string { + t.Helper() + sum := mustMD5([]byte(content)) + resp, err := svc.init(initReq("new.bin", sum, targetDir), "op", 1) + if err != nil { + t.Fatal(err) + } + uid := resp.(map[string]any)["uploadId"].(string) + + dir := svc.chunkDir(uid) + if err := os.MkdirAll(dir, 0o766); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "0"), []byte(content), 0o644); err != nil { + t.Fatal(err) + } + if err := repo.CreateChunk(&ChunkInfo{UploadId: uid, ChunkIndex: 0}); err != nil { + t.Fatal(err) + } + out, err := svc.merge(uid) + if err != nil { + t.Fatal(err) + } + if got := out.(map[string]any)["fileUrl"]; got != "/uploads/"+repo.tasks[repo.byUID[uid]].StoragePath { + t.Fatalf("fileUrl 与 storagePath 不对应: %v vs %s", got, repo.tasks[repo.byUID[uid]].StoragePath) + } + return repo.tasks[repo.byUID[uid]].StoragePath +} + +func TestMergeIntoTargetDir(t *testing.T) { + svc, repo, cloud := newTestSvc(t) + sp := mergeTask(t, svc, repo, "docs/reports", "HELLO") + + if sp != "cloud/docs/reports/"+filepath.Base(sp) { + t.Fatalf("storagePath 形态错误: %s", sp) + } + data, err := os.ReadFile(filepath.Join(cloud, "docs", "reports", filepath.Base(sp))) + if err != nil || string(data) != "HELLO" { + t.Fatalf("目标目录物理文件缺失或不一致: %v", err) + } +} + +func TestMergeFixedModeRegression(t *testing.T) { + svc, repo, cloud := newTestSvc(t) + sp := mergeTask(t, svc, repo, "", "HELLO") + + // 与 v1.0.x 逐字节一致:cloud/_,物理在 cloud 根 + if !strings.HasPrefix(sp, "cloud/") || strings.Contains(strings.TrimPrefix(sp, "cloud/"), "/") { + t.Fatalf("固定模式 storagePath 应平铺在 cloud 根: %s", sp) + } + if _, err := os.Stat(filepath.Join(cloud, filepath.Base(sp))); err != nil { + t.Fatalf("固定模式物理文件应在 cloud 根: %v", err) + } +} + +// ── check:目录过滤 ────────────────────────────────────────────────── + +func TestCheckUploadingDirFilter(t *testing.T) { + svc, repo, _ := newTestSvc(t) + sum := mustMD5([]byte("OLD-CONTENT")) + if _, err := svc.init(initReq("old.bin", sum, "a"), "op", 1); err != nil { + t.Fatal(err) + } + uploading, err := svc.check(sum, "a") + if err != nil { + t.Fatal(err) + } + if uploading.(map[string]any)["status"] != StatusUploading { + t.Fatalf("同目录 check 应报 uploading: %v", uploading) + } + + noneResp, err := svc.check(sum, "b") + if err != nil { + t.Fatal(err) + } + if noneResp.(map[string]any)["status"] != "none" { + t.Fatalf("异目录 check 应报 none: %v", noneResp) + } + + // done 命中不受目录影响(init 会 copy-on-hit) + seedDoneTask(t, svc, repo, mustMD5([]byte("OTHER")), "") + doneResp, err := svc.check(mustMD5([]byte("OTHER")), "zzz") + if err != nil { + t.Fatal(err) + } + if doneResp.(map[string]any)["status"] != StatusDone { + t.Fatalf("done 命中应报 done: %v", doneResp) + } +}