1 Commits
Author SHA1 Message Date
4566704 455360bff0 feat: 支持 targetDir 指定上传目标目录(网盘式),固定路径模式不变
- 改动:init/check 接受可选 targetDir(相对 CloudDir,SanitizeDir 校验),merge 物理落盘 CloudDir/<targetDir>/
- 秒传跨目录 copy-on-hit:库把旧物理文件复制进目标目录,OnInstantHit 携带新路径;源缺失回退正常上传
- 断点续传按目录隔离:同 MD5 在途任务目录不同则另起新任务
- TaskInfo 新增 TargetDir(宿主须持久化);Repository 12 方法接口零变更,向后兼容 v1.0.x
- 新增 dir.go(SanitizeDir/路径拼装反解/copyFile)与 service_test.go(10 个用例:双模式秒传/续传隔离/merge 回归)
- 原因:上传位置原本固定在 CloudDir 根,宿主无法实现网盘式选目录上传
2026-09-26 08:47:55 +08:00
6 changed files with 625 additions and 31 deletions
+27 -4
View File
@@ -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/<targetDir>/<产物名>`,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 模块抽出
+98
View File
@@ -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
}
+4 -2
View File
@@ -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。
+4 -3
View File
@@ -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)
})
+68 -16
View File
@@ -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 hit, ok := s.instantHit(done, targetDir); ok {
if s.opts.OnInstantHit != nil {
if err := s.opts.OnInstantHit(*done); err != nil {
if err := s.opts.OnInstantHit(hit); err != nil {
return nil, err
}
}
return map[string]any{
"uploadId": "",
"skipUpload": true,
"fileUrl": s.fileURL(done.StoragePath),
"fileUrl": s.fileURL(hit.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
+418
View File
@@ -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`, "a<b", "a>b", "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/<nanots>_<name>,物理在 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)
}
}