feat(upload): extract upload subsystem + move sweep to scanner ticker (B16)

- internal/upload owns chunked-upload domain logic (fingerprint resume, part tmp+rename, assemble, session sweep) and the shared UniquePath helper used by both single-file and chunked completion paths - handlers/uploads.go is now HTTP-only: bind params, call upload.U, map sentinel errors to the unchanged status/code/message contract - B16: session sweep moved out of the UploadInit request path onto the scanner ticker via a Sweeper hook (upload.U satisfies it) - unit tests for the package without PG/Redis; contract pinned by the existing handler integration tests (all green)
2026-09-14 22:00:25 +08:00
parent 443f4acfa9
commit fbd5cd243d
10 changed files with 708 additions and 263 deletions
@@ -10,11 +10,12 @@ import (
 	"booklib/internal/redispkg"
 	"booklib/internal/scanner"
 	"booklib/internal/store"
+	"booklib/internal/upload"
 )

-func NewRouter(cfg *config.Config, st *store.Store, rdb *redispkg.R, sc *scanner.Scanner) *gin.Engine {
+func NewRouter(cfg *config.Config, st *store.Store, rdb *redispkg.R, sc *scanner.Scanner, up *upload.U) *gin.Engine {
 	gin.SetMode(gin.ReleaseMode)
-	h := handlers.New(cfg, st, rdb, sc)
+	h := handlers.New(cfg, st, rdb, sc, up)
 	r := gin.New()
 	if e := r.SetTrustedProxies(cfg.TrustedProxies); e != nil {
 		panic(e)
@@ -16,7 +16,7 @@ func testCfg() *config.Config {
 }

 func TestHealthz(t *testing.T) {
-	r := NewRouter(testCfg(), nil, redispkg.New(""), nil)
+	r := NewRouter(testCfg(), nil, redispkg.New(""), nil, nil)
 	req := httptest.NewRequest(http.MethodGet, "/api/healthz", nil)
 	w := httptest.NewRecorder()
 	r.ServeHTTP(w, req)
@@ -21,6 +21,7 @@ import (
 	"booklib/internal/redispkg"
 	"booklib/internal/scanner"
 	"booklib/internal/store"
+	"booklib/internal/upload"
 )

 func testCfg() *config.Config {
@@ -57,8 +58,9 @@ func setupAPI(t *testing.T) (*store.Store, *scanner.Scanner, http.Handler, strin
 	cfg.BooksDir = booksDir
 	cfg.CacheDir = t.TempDir()
 	rdb := redispkg.New(os.Getenv("REDIS_URL"))
-	sc := scanner.New(st, cfg, rdb)
-	r := api.NewRouter(cfg, st, rdb, sc)
+	up := upload.New(cfg.BooksDir, cfg.UploadMaxMB)
+	sc := scanner.New(st, cfg, rdb, up)
+	r := api.NewRouter(cfg, st, rdb, sc, up)
 	if u := os.Getenv("REDIS_URL"); u != "" { // 测试卫生: 共享 redis 上重置登录限流桶, 防跨测试累计 429
 		if opt, e := redis.ParseURL(u); e == nil {
 			rc := redis.NewClient(opt)
@@ -18,6 +18,7 @@ import (
 	"booklib/internal/redispkg"
 	"booklib/internal/scanner"
 	"booklib/internal/store"
+	"booklib/internal/upload"
 )

 type H struct {
@@ -25,10 +26,11 @@ type H struct {
 	st  *store.Store
 	rdb *redispkg.R
 	sc  *scanner.Scanner
+	up  *upload.U
 }

-func New(cfg *config.Config, st *store.Store, rdb *redispkg.R, sc *scanner.Scanner) *H {
-	return &H{cfg: cfg, st: st, rdb: rdb, sc: sc}
+func New(cfg *config.Config, st *store.Store, rdb *redispkg.R, sc *scanner.Scanner, up *upload.U) *H {
+	return &H{cfg: cfg, st: st, rdb: rdb, sc: sc, up: up}
 }

 func err(c *gin.Context, status int, code, msg string) {
@@ -136,12 +136,12 @@ func (h *H) Upload(c *gin.Context) {
 		return
 	}
 	// B12: retry on O_EXCL collision — concurrent uploads with the same name
-	// can both get the same candidate from uniquePath (stat-then-create race).
+	// can both get the same candidate from UniquePath (stat-then-create race).
 	var dst, tmp string
 	var out *os.File
 	for attempt := 0; attempt < 5; attempt++ {
 		var e error
-		dst, e = h.uniquePath(root, name)
+		dst, e = h.up.UniquePath(root, name)
 		if e != nil {
 			err(c, http.StatusForbidden, "forbidden", e.Error())
 			return
@@ -190,28 +190,3 @@ func (h *H) Upload(c *gin.Context) {
 	}
 	c.JSON(http.StatusAccepted, gin.H{"accepted": true, "path": strings.TrimPrefix(dst, root+string(os.PathSeparator))})
 }
-
-// uniquePath 清洗后的 name 必须仍在 root 内;重名加 " (n)" 后缀
-func (h *H) uniquePath(root, name string) (string, error) {
-	ext := filepath.Ext(name)
-	base := strings.TrimSuffix(name, ext)
-	for i := 0; ; i++ {
-		cand := base + ext
-		if i > 0 {
-			cand = base + " (" + strconv.Itoa(i) + ")" + ext
-		}
-		p := filepath.Join(root, cand)
-		if filepath.Clean(p) != filepath.Join(root, filepath.Clean(cand)) ||
-			!strings.HasPrefix(filepath.Clean(p), root+string(os.PathSeparator)) {
-			return "", os.ErrInvalid
-		}
-		if _, e := os.Stat(p); os.IsNotExist(e) {
-			return p, nil
-		} else if e != nil {
-			return "", e
-		}
-		if i > 999 {
-			return "", os.ErrExist
-		}
-	}
-}
@@ -1,86 +1,22 @@
 package handlers

 import (
-	"crypto/sha256"
-	"encoding/hex"
-	"encoding/json"
 	"errors"
-	"fmt"
-	"io"
 	"net/http"
 	"os"
-	"path/filepath"
-	"sort"
 	"strconv"
-	"strings"
-	"time"

 	"github.com/gin-gonic/gin"

-	"booklib/internal/bookfile"
+	"booklib/internal/ports"
+	"booklib/internal/upload"
 )

-// 分片上传:init(指纹→确定性 uploadId,天然支持续传)→ PUT parts → complete 拼接原子落盘。
-// 会话即 <BooksDir>/.uploads/<uid>/(meta.json + parts/N),无独立状态存储;24h 未 complete  opportunistic 清扫。
+// 分片上传:HTTP 层只做参数绑定与错误映射,域逻辑(指纹续传/分片落盘/拼接/清扫)
+// 全部在 internal/upload;会话清扫由 scanner ticker 接管(B16),不在请求路径。

-const (
-	maxChunkBytes   = 32 << 20
-	defaultChunk    = 8 << 20
-	uploadSessTTL   = 24 * time.Hour
-	uploadSessionIn = ".uploads"
-)
-
-type uploadMeta struct {
-	Name      string `json:"name"`
-	Size      int64  `json:"size"`
-	ChunkSize int64  `json:"chunkSize"`
-	LibraryID int64  `json:"libraryId"`
-}
-
-func validUploadID(s string) bool {
-	if len(s) != 32 {
-		return false
-	}
-	for _, r := range s {
-		if !((r >= '0' && r <= '9') || (r >= 'a' && r <= 'f')) {
-			return false
-		}
-	}
-	return true
-}
-
-func uploadIDFor(libID int64, name string, size, chunk int64) string {
-	h := sha256.Sum256([]byte(fmt.Sprintf("%d|%s|%d|%d", libID, name, size, chunk)))
-	return hex.EncodeToString(h[:16])
-}
-
-func (h *H) uploadDir(uid string) string {
-	return filepath.Join(filepath.Clean(h.cfg.BooksDir), uploadSessionIn, uid)
-}
-
-func chunkRange(m uploadMeta, i int64) (int64, int64) {
-	lo := i * m.ChunkSize
-	hi := min(lo+m.ChunkSize, m.Size)
-	return lo, hi
-}
-
-func (h *H) numParts(m uploadMeta) int64 {
-	return (m.Size + m.ChunkSize - 1) / m.ChunkSize
-}
-
-// sweepUploads 删除过期会话目录;尽力而为,失败不影响主流程
-func (h *H) sweepUploads() {
-	base := filepath.Join(filepath.Clean(h.cfg.BooksDir), uploadSessionIn)
-	es, e := os.ReadDir(base)
-	if e != nil {
-		return
-	}
-	for _, en := range es {
-		if fi, e := en.Info(); e == nil && time.Since(fi.ModTime()) > uploadSessTTL {
-			os.RemoveAll(filepath.Join(base, en.Name()))
-		}
-	}
-}
+// maxChunkBytes 与 upload 包内常量同值,作为 PutPart 的防御性读取上限。
+const maxChunkBytes = 32 << 20

 func (h *H) UploadInit(c *gin.Context) {
 	lib, ok := h.getLibrary(c)
@@ -99,154 +35,47 @@ func (h *H) UploadInit(c *gin.Context) {
 		err(c, http.StatusBadRequest, "bad_request", "name, size required")
 		return
 	}
-	name := bookfile.SafeName(req.Name)
-	if name == "" {
-		err(c, http.StatusBadRequest, "bad_request", "bad name")
-		return
-	}
-	if bookfile.FormatFromExt(name) == "" {
-		err(c, http.StatusBadRequest, "bad_format", "extension must be cbz/pdf/epub/txt/md")
-		return
-	}
-	if req.Size <= 0 {
-		err(c, http.StatusBadRequest, "bad_request", "bad size")
-		return
-	}
-	if req.Size > h.cfg.UploadMaxMB<<20 {
-		err(c, http.StatusRequestEntityTooLarge, "too_large", "file exceeds upload limit of "+strconv.FormatInt(h.cfg.UploadMaxMB, 10)+"MB")
-		return
-	}
-	if req.ChunkSize == 0 {
-		req.ChunkSize = defaultChunk
-	}
-	if req.ChunkSize > maxChunkBytes {
-		err(c, http.StatusBadRequest, "bad_request", "chunkSize must be <= 33554432")
-		return
-	}
-	uid := uploadIDFor(lib.ID, name, req.Size, req.ChunkSize)
-	dir := h.uploadDir(uid)
-	meta := uploadMeta{Name: name, Size: req.Size, ChunkSize: req.ChunkSize, LibraryID: lib.ID}
-	if b, e := os.ReadFile(filepath.Join(dir, "meta.json")); e == nil {
-		var old uploadMeta
-		if json.Unmarshal(b, &old) == nil && old == meta { // 同指纹 → 复用会话(续传)
-			c.JSON(http.StatusOK, gin.H{"uploadId": uid})
-			return
-		}
-		os.RemoveAll(dir) // 指纹撞上但内容不同 → 从头再来
-	}
-	h.sweepUploads()
-	if e := os.MkdirAll(filepath.Join(dir, "parts"), 0o755); e != nil {
-		err(c, http.StatusInternalServerError, "internal", "create session")
-		return
-	}
-	b, _ := json.Marshal(meta)
-	if e := os.WriteFile(filepath.Join(dir, "meta.json"), b, 0o644); e != nil {
-		err(c, http.StatusInternalServerError, "internal", "write meta")
+	uid, e := h.up.Init(c, lib.ID, req.Name, req.Size, req.ChunkSize)
+	if e != nil {
+		h.mapUploadErr(c, e)
 		return
 	}
 	c.JSON(http.StatusOK, gin.H{"uploadId": uid})
 }

-// loadMeta 校验 uid 与路径,404/400 已回复
-func (h *H) loadMeta(c *gin.Context) (uploadMeta, string, bool) {
-	uid := c.Param("uid")
-	if !validUploadID(uid) {
-		err(c, http.StatusBadRequest, "bad_request", "bad upload id")
-		return uploadMeta{}, "", false
-	}
-	dir := h.uploadDir(uid)
-	var m uploadMeta
-	b, e := os.ReadFile(filepath.Join(dir, "meta.json"))
-	if e != nil {
-		err(c, http.StatusNotFound, "not_found", "no such upload")
-		return uploadMeta{}, "", false
-	}
-	if json.Unmarshal(b, &m) != nil {
-		err(c, http.StatusInternalServerError, "internal", "corrupt session")
-		return uploadMeta{}, "", false
-	}
-	return m, dir, true
-}
-
 func (h *H) UploadStatus(c *gin.Context) {
-	_, dir, ok := h.loadMeta(c)
-	if !ok {
+	recv, e := h.up.Status(c, c.Param("uid"))
+	if e != nil {
+		h.mapUploadErr(c, e)
 		return
 	}
-	recv := []int64{}
-	es, e := os.ReadDir(filepath.Join(dir, "parts"))
-	if e == nil {
-		for _, en := range es {
-			if i, e := strconv.ParseInt(en.Name(), 10, 64); e == nil {
-				recv = append(recv, i)
-			}
-		}
-	}
-	sort.Slice(recv, func(i, j int) bool { return recv[i] < recv[j] })
 	c.JSON(http.StatusOK, gin.H{"received": recv})
 }

 func (h *H) UploadPart(c *gin.Context) {
-	m, dir, ok := h.loadMeta(c)
-	if !ok {
-		return
-	}
+	uid := c.Param("uid")
 	idx, e := strconv.ParseInt(c.Param("index"), 10, 64)
-	if e != nil || idx < 0 || idx >= h.numParts(m) {
+	if e != nil {
 		err(c, http.StatusBadRequest, "bad_request", "bad part index")
 		return
 	}
-	lo, hi := chunkRange(m, idx)
-	c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, hi-lo)
-	// B4: write to .tmp then rename — prevents truncated parts from being
-	// reported as "received" by UploadStatus if the process crashes mid-write.
-	p := filepath.Join(dir, "parts", strconv.FormatInt(idx, 10))
-	tmp := p + ".tmp"
-	f, e := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o644)
+	// maxSize 防御性上限:分片声明大小由会话 meta 决定,这里再垫一层 32MB 全局上限
+	e = h.up.PutPart(c, uid, idx, c.Request.Body, maxChunkBytes)
 	if e != nil {
-		err(c, http.StatusInternalServerError, "internal", "create part")
-		return
-	}
-	n, e := io.Copy(f, c.Request.Body)
-	f.Close()
-	if e != nil || n != hi-lo {
-		os.Remove(tmp)
-		var mbe *http.MaxBytesError
-		code, msg := "too_large", "part size mismatch"
-		if errors.As(e, &mbe) {
-			msg = "part exceeds declared size"
-		}
-		err(c, http.StatusRequestEntityTooLarge, code, msg)
-		return
-	}
-	if e := os.Rename(tmp, p); e != nil {
-		os.Remove(tmp)
-		err(c, http.StatusInternalServerError, "internal", "rename part")
+		h.mapUploadErr(c, e)
 		return
 	}
 	c.JSON(http.StatusAccepted, gin.H{"accepted": true})
 }

 func (h *H) UploadComplete(c *gin.Context) {
-	m, dir, ok := h.loadMeta(c)
-	if !ok {
+	uid := c.Param("uid")
+	libID, e := h.up.LibraryID(c, uid)
+	if e != nil {
+		h.mapUploadErr(c, e)
 		return
 	}
-	var total int64
-	for i := int64(0); i < h.numParts(m); i++ {
-		lo, hi := chunkRange(m, i)
-		fi, e := os.Stat(filepath.Join(dir, "parts", strconv.FormatInt(i, 10)))
-		if e != nil || fi.Size() != hi-lo {
-			err(c, http.StatusBadRequest, "bad_request", "upload incomplete; missing or corrupt parts, re-upload them")
-			return
-		}
-		total += fi.Size()
-	}
-	if total != m.Size {
-		err(c, http.StatusBadRequest, "bad_request", "total size mismatch")
-		return
-	}
-	lib, e := h.st.GetLibrary(c, m.LibraryID)
+	lib, e := h.st.GetLibrary(c, libID)
 	if e != nil {
 		dbErr(c, e)
 		return
@@ -255,39 +84,53 @@ func (h *H) UploadComplete(c *gin.Context) {
 	if !ok {
 		return
 	}
-	dst, e := h.uniquePath(root, m.Name)
+	rel, e := h.up.Complete(c, uid, root)
 	if e != nil {
-		err(c, http.StatusForbidden, "forbidden", e.Error())
+		h.mapUploadErr(c, e)
 		return
 	}
-	tmp := filepath.Join(dir, "assembled")
-	out, e := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o644)
-	if e != nil {
-		err(c, http.StatusInternalServerError, "internal", "create tmp")
-		return
-	}
-	for i := int64(0); i < h.numParts(m); i++ {
-		pf, e := os.Open(filepath.Join(dir, "parts", strconv.FormatInt(i, 10)))
-		if e != nil {
-			out.Close()
-			err(c, http.StatusInternalServerError, "internal", "open part")
-			return
-		}
-		if _, e := io.Copy(out, pf); e != nil {
-			pf.Close()
-			out.Close()
-			os.Remove(tmp)
-			err(c, http.StatusInternalServerError, "internal", "assemble")
+	c.JSON(http.StatusAccepted, gin.H{"accepted": true, "path": rel})
+}
+
+// mapUploadErr 把 upload 包的 sentinel 错误映射为原契约的 status/code/message。
+func (h *H) mapUploadErr(c *gin.Context, e error) {
+	switch {
+	case errors.Is(e, ports.ErrTooLarge):
+		err(c, http.StatusRequestEntityTooLarge, "too_large", "file exceeds upload limit of "+strconv.FormatInt(h.cfg.UploadMaxMB, 10)+"MB")
+	case errors.Is(e, upload.ErrBadName):
+		err(c, http.StatusBadRequest, "bad_request", "bad name")
+	case errors.Is(e, upload.ErrBadFormat):
+		err(c, http.StatusBadRequest, "bad_format", "extension must be cbz/pdf/epub/txt/md")
+	case errors.Is(e, upload.ErrBadSize):
+		err(c, http.StatusBadRequest, "bad_request", "bad size")
+	case errors.Is(e, upload.ErrBadChunk):
+		err(c, http.StatusBadRequest, "bad_request", "chunkSize must be <= 33554432")
+	case errors.Is(e, upload.ErrBadUploadID):
+		err(c, http.StatusBadRequest, "bad_request", "bad upload id")
+	case errors.Is(e, upload.ErrBadIndex):
+		err(c, http.StatusBadRequest, "bad_request", "bad part index")
+	case errors.Is(e, upload.ErrPartTooBig):
+		err(c, http.StatusRequestEntityTooLarge, "too_large", "part exceeds declared size")
+	case errors.Is(e, upload.ErrPartSizeMismatch):
+		err(c, http.StatusRequestEntityTooLarge, "too_large", "part size mismatch")
+	case errors.Is(e, ports.ErrNotFound):
+		err(c, http.StatusNotFound, "not_found", "no such upload")
+	case errors.Is(e, upload.ErrCorrupt):
+		err(c, http.StatusInternalServerError, "internal", "corrupt session")
+	case errors.Is(e, ports.ErrIncomplete):
+		err(c, http.StatusBadRequest, "bad_request", "upload incomplete; missing or corrupt parts, re-upload them")
+	case errors.Is(e, ports.ErrSizeMismatch):
+		err(c, http.StatusBadRequest, "bad_request", "total size mismatch")
+	case errors.Is(e, os.ErrInvalid), errors.Is(e, os.ErrExist):
+		err(c, http.StatusForbidden, "forbidden", e.Error())
+	case errors.Is(e, os.ErrPermission), errors.Is(e, os.ErrClosed):
+		err(c, http.StatusInternalServerError, "internal", "io error")
+	default:
+		var oe *upload.OpError
+		if errors.As(e, &oe) {
+			err(c, http.StatusInternalServerError, "internal", oe.Op)
 			return
 		}
-		pf.Close()
-	}
-	out.Close()
-	if e := os.Rename(tmp, dst); e != nil { // 原子落盘,scanner 自动收编
-		os.Remove(tmp)
-		err(c, http.StatusInternalServerError, "internal", "rename")
-		return
+		err(c, http.StatusInternalServerError, "internal", "upload failed")
 	}
-	os.RemoveAll(dir)
-	c.JSON(http.StatusAccepted, gin.H{"accepted": true, "path": strings.TrimPrefix(dst, root+string(os.PathSeparator))})
 }
@@ -16,6 +16,7 @@ import (
 	"booklib/internal/scanner"
 	"booklib/internal/seed"
 	"booklib/internal/store"
+	"booklib/internal/upload"
 )

 func main() {
@@ -39,11 +40,12 @@ func main() {
 		log.Fatalf("seed: %v", err)
 	}
 	rdb := redispkg.New(cfg.RedisURL)
-	sc := scanner.New(st, cfg, rdb)
+	up := upload.New(cfg.BooksDir, cfg.UploadMaxMB)
+	sc := scanner.New(st, cfg, rdb, up) // B16: sweep rides the scan ticker
 	go sc.Run(ctx)

 	serveErr := make(chan error, 1)
-	srv := &http.Server{Addr: cfg.Addr, Handler: api.NewRouter(cfg, st, rdb, sc),
+	srv := &http.Server{Addr: cfg.Addr, Handler: api.NewRouter(cfg, st, rdb, sc, up),
 		ReadHeaderTimeout: 10 * time.Second}
 	go func() {
 		log.Printf("listening on %s", cfg.Addr)