From 6487f03c4a5b23136ff36e3ccf0705a6541d9ce7 Mon Sep 17 00:00:00 2001 From: yiyiis Date: Mon, 11 May 2026 20:49:55 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=E8=A7=86=E9=A2=91?= =?UTF-8?q?=E5=88=86=E7=89=87=E4=B8=8A=E4=BC=A0=E4=B8=8E=E6=96=AD=E7=82=B9?= =?UTF-8?q?=E7=BB=AD=E4=BC=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 后端：init/upload/status/complete 四个分片上传接口 - 后端：基于 Redis 的上传会话管理，支持断点续传 - 后端：分片 MD5 校验、幂等上传、顺序合并 - 测试：覆盖完整上传流程、断点续传、幂等性、哈希校验、未完成合并、状态查询 - 前端：Vue 3 分片上传组件，支持并发上传与进度展示 --- backend/internal/http/router.go | 5 + backend/internal/middleware/redis/redis.go | 4 + backend/internal/video/chunk_entity.go | 55 ++ backend/internal/video/chunk_handler.go | 344 ++++++++ backend/internal/video/chunk_handler_test.go | 427 +++++++++ frontend/package-lock.json | 7 + frontend/package.json | 1 + frontend/src/api/video.ts | 98 ++- frontend/src/types/spark-md5.d.ts | 33 + frontend/src/views/VideoView.vue | 863 +++++++++++-------- 10 files changed, 1461 insertions(+), 376 deletions(-) create mode 100644 backend/internal/video/chunk_entity.go create mode 100644 backend/internal/video/chunk_handler.go create mode 100644 backend/internal/video/chunk_handler_test.go create mode 100644 frontend/src/types/spark-md5.d.ts diff --git a/backend/internal/http/router.go b/backend/internal/http/router.go index 50c86e8..6431930 100644 --- a/backend/internal/http/router.go +++ b/backend/internal/http/router.go @@ -62,6 +62,7 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g } videoService := video.NewVideoService(videoRepository, cache, popularityMQ) videoHandler := video.NewVideoHandler(videoService, accountService) + chunkHandler := video.NewChunkUploadHandler(cache) videoGroup := r.Group("/video") { videoGroup.POST("/listByAuthorID", videoHandler.ListByAuthorID) @@ -73,6 +74,10 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g protectedVideoGroup.POST("/uploadVideo", videoHandler.UploadVideo) protectedVideoGroup.POST("/uploadCover", videoHandler.UploadCover) protectedVideoGroup.POST("/publish", videoHandler.PublishVideo) + protectedVideoGroup.POST("/chunk/init", chunkHandler.InitChunkUpload) + protectedVideoGroup.POST("/chunk/upload", chunkHandler.UploadChunk) + protectedVideoGroup.POST("/chunk/status", chunkHandler.ChunkStatus) + protectedVideoGroup.POST("/chunk/complete", chunkHandler.CompleteChunkUpload) } // like likeMQ, err := rabbitmq.NewLikeMQ(rmq) diff --git a/backend/internal/middleware/redis/redis.go b/backend/internal/middleware/redis/redis.go index 1f7f011..25d55de 100644 --- a/backend/internal/middleware/redis/redis.go +++ b/backend/internal/middleware/redis/redis.go @@ -19,6 +19,10 @@ type Client struct { const defaultKeyPrefix = "v1:" +func NewClient(rdb *redis.Client, keyPrefix string) *Client { + return &Client{rdb: rdb, keyPrefix: keyPrefix} +} + func NewFromEnv(cfg *config.RedisConfig) (*Client, error) { rdb := redis.NewClient(&redis.Options{ Addr: cfg.Host + ":" + strconv.Itoa(cfg.Port), diff --git a/backend/internal/video/chunk_entity.go b/backend/internal/video/chunk_entity.go new file mode 100644 index 0000000..8e29faf --- /dev/null +++ b/backend/internal/video/chunk_entity.go @@ -0,0 +1,55 @@ +package video + +const ChunkSize = 5 << 20 // 5 MB + +type ChunkUploadSession struct { + UploadID string `json:"upload_id"` + AccountID uint `json:"account_id"` + Filename string `json:"filename"` + FileSize int64 `json:"file_size"` + ChunkSize int64 `json:"chunk_size"` + TotalChunks int `json:"total_chunks"` + FileHash string `json:"file_hash"` + UploadedBits []bool `json:"uploaded_bits"` +} + +func (s *ChunkUploadSession) UploadedChunks() []int { + var indices []int + for i, uploaded := range s.UploadedBits { + if uploaded { + indices = append(indices, i) + } + } + return indices +} + +func (s *ChunkUploadSession) IsComplete() bool { + for _, b := range s.UploadedBits { + if !b { + return false + } + } + return true +} + +type InitChunkUploadRequest struct { + Filename string `json:"filename" binding:"required"` + FileSize int64 `json:"file_size" binding:"required,min=1"` + ChunkSize int64 `json:"chunk_size" binding:"required,min=1"` + TotalChunks int `json:"total_chunks" binding:"required,min=1"` + FileHash string `json:"file_hash" binding:"required"` +} + +type UploadChunkRequest struct { + UploadID string `form:"upload_id" binding:"required"` + ChunkIndex int `form:"chunk_index" binding:"min=0"` + ChunkHash string `form:"chunk_hash" binding:"required"` +} + +type ChunkStatusRequest struct { + UploadID string `json:"upload_id" binding:"required"` +} + +type CompleteChunkUploadRequest struct { + UploadID string `json:"upload_id" binding:"required"` +} diff --git a/backend/internal/video/chunk_handler.go b/backend/internal/video/chunk_handler.go new file mode 100644 index 0000000..b7cf753 --- /dev/null +++ b/backend/internal/video/chunk_handler.go @@ -0,0 +1,344 @@ +package video + +import ( + "crypto/md5" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "time" + + "feedsystem_video_go/internal/middleware/jwt" + rediscache "feedsystem_video_go/internal/middleware/redis" + + "github.com/gin-gonic/gin" +) + +const sessionTTL = 24 * time.Hour + +type ChunkUploadHandler struct { + cache *rediscache.Client +} + +func NewChunkUploadHandler(cache *rediscache.Client) *ChunkUploadHandler { + return &ChunkUploadHandler{cache: cache} +} + +func (h *ChunkUploadHandler) sessionKey(uploadID string) string { + return h.cache.Key("chunk_upload:%s", uploadID) +} + +func (h *ChunkUploadHandler) hashKey(accountID uint, fileHash string) string { + return h.cache.Key("chunk_upload_hash:%d:%s", accountID, fileHash) +} + +func (h *ChunkUploadHandler) getSession(ctx *gin.Context, uploadID string) (*ChunkUploadSession, error) { + b, err := h.cache.GetBytes(ctx.Request.Context(), h.sessionKey(uploadID)) + if err != nil { + return nil, fmt.Errorf("upload session not found") + } + var s ChunkUploadSession + if err := json.Unmarshal(b, &s); err != nil { + return nil, fmt.Errorf("invalid session data") + } + return &s, nil +} + +func (h *ChunkUploadHandler) saveSession(ctx *gin.Context, s *ChunkUploadSession) error { + b, err := json.Marshal(s) + if err != nil { + return err + } + return h.cache.SetBytes(ctx.Request.Context(), h.sessionKey(s.UploadID), b, sessionTTL) +} + +func (h *ChunkUploadHandler) InitChunkUpload(c *gin.Context) { + var req InitChunkUploadRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + accountID, err := jwt.GetAccountID(c) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + const maxSize = 200 << 20 + if req.FileSize > maxSize { + c.JSON(http.StatusBadRequest, gin.H{"error": "file size exceeds 200MB limit"}) + return + } + + // Check for existing session (resume) + hashKey := h.hashKey(accountID, req.FileHash) + existingID, err := h.cache.GetBytes(c.Request.Context(), hashKey) + if err == nil && len(existingID) > 0 { + session, sessErr := h.getSession(c, string(existingID)) + if sessErr == nil { + // Refresh TTL on resume + _ = h.cache.SetBytes(c.Request.Context(), hashKey, existingID, sessionTTL) + _ = h.saveSession(c, session) + c.JSON(http.StatusOK, gin.H{ + "upload_id": session.UploadID, + "uploaded_chunks": session.UploadedChunks(), + }) + return + } + } + + id, _ := randHex(16) + uploadID := id + fmt.Sprintf("%d", time.Now().UnixNano()) + session := &ChunkUploadSession{ + UploadID: uploadID, + AccountID: accountID, + Filename: req.Filename, + FileSize: req.FileSize, + ChunkSize: req.ChunkSize, + TotalChunks: req.TotalChunks, + FileHash: req.FileHash, + UploadedBits: make([]bool, req.TotalChunks), + } + + if err := h.saveSession(c, session); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create session"}) + return + } + + if err := h.cache.SetBytes(c.Request.Context(), hashKey, []byte(uploadID), sessionTTL); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create session"}) + return + } + + c.JSON(http.StatusOK, gin.H{ + "upload_id": uploadID, + "uploaded_chunks": []int{}, + }) +} + +func (h *ChunkUploadHandler) UploadChunk(c *gin.Context) { + var req UploadChunkRequest + if err := c.ShouldBind(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + session, err := h.getSession(c, req.UploadID) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + return + } + + accountID, err := jwt.GetAccountID(c) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + if session.AccountID != accountID { + c.JSON(http.StatusForbidden, gin.H{"error": "forbidden"}) + return + } + + if req.ChunkIndex < 0 || req.ChunkIndex >= session.TotalChunks { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid chunk_index"}) + return + } + + if session.UploadedBits[req.ChunkIndex] { + c.JSON(http.StatusOK, gin.H{"chunk_index": req.ChunkIndex}) + return + } + + f, err := c.FormFile("file") + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "missing file"}) + return + } + + chunkFile, err := f.Open() + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to read chunk"}) + return + } + defer chunkFile.Close() + + hash := md5.New() + if _, err := io.Copy(hash, chunkFile); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to hash chunk"}) + return + } + actualHash := fmt.Sprintf("%x", hash.Sum(nil)) + + if actualHash != req.ChunkHash { + c.JSON(http.StatusBadRequest, gin.H{"error": "chunk hash mismatch", "expected": req.ChunkHash, "actual": actualHash}) + return + } + + tmpDir := filepath.Join(".run", "uploads", "tmp", req.UploadID) + if err := os.MkdirAll(tmpDir, 0o755); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create temp dir"}) + return + } + + chunkPath := filepath.Join(tmpDir, fmt.Sprintf("%d", req.ChunkIndex)) + if _, seekErr := chunkFile.Seek(0, io.SeekStart); seekErr != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to read chunk"}) + return + } + + dst, err := os.Create(chunkPath) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to save chunk"}) + return + } + defer dst.Close() + + if _, err := io.Copy(dst, chunkFile); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to save chunk"}) + return + } + + session.UploadedBits[req.ChunkIndex] = true + if err := h.saveSession(c, session); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update session"}) + return + } + + c.JSON(http.StatusOK, gin.H{"chunk_index": req.ChunkIndex}) +} + +func (h *ChunkUploadHandler) ChunkStatus(c *gin.Context) { + var req ChunkStatusRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + session, err := h.getSession(c, req.UploadID) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + return + } + + accountID, err := jwt.GetAccountID(c) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + if session.AccountID != accountID { + c.JSON(http.StatusForbidden, gin.H{"error": "forbidden"}) + return + } + + c.JSON(http.StatusOK, gin.H{ + "upload_id": session.UploadID, + "uploaded_chunks": session.UploadedChunks(), + "total_chunks": session.TotalChunks, + }) +} + +func (h *ChunkUploadHandler) CompleteChunkUpload(c *gin.Context) { + var req CompleteChunkUploadRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + session, err := h.getSession(c, req.UploadID) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + return + } + + accountID, err := jwt.GetAccountID(c) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + if session.AccountID != accountID { + c.JSON(http.StatusForbidden, gin.H{"error": "forbidden"}) + return + } + + if !session.IsComplete() { + missing := 0 + for _, uploaded := range session.UploadedBits { + if !uploaded { + missing++ + if missing > 5 { + missing = 5 + break + } + } + } + c.JSON(http.StatusBadRequest, gin.H{ + "error": "not all chunks uploaded", + "missing": missing, + "completed": len(session.UploadedChunks()), + "total": session.TotalChunks, + }) + return + } + + date := time.Now().Format("20060102") + relDir := filepath.Join("videos", fmt.Sprintf("%d", accountID), date) + root := filepath.Join(".run", "uploads") + absDir := filepath.Join(root, relDir) + if err := os.MkdirAll(absDir, 0o755); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create output dir"}) + return + } + + filename, err := randHex(16) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to generate filename"}) + return + } + finalPath := filepath.Join(absDir, filename+".mp4") + + finalFile, err := os.Create(finalPath) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create final file"}) + return + } + defer finalFile.Close() + + tmpDir := filepath.Join(".run", "uploads", "tmp", req.UploadID) + for i := 0; i < session.TotalChunks; i++ { + chunkPath := filepath.Join(tmpDir, fmt.Sprintf("%d", i)) + cf, err := os.Open(chunkPath) + if err != nil { + finalFile.Close() + os.Remove(finalPath) + c.JSON(http.StatusInternalServerError, gin.H{"error": fmt.Sprintf("chunk %d missing", i)}) + return + } + _, err = io.Copy(finalFile, cf) + cf.Close() + if err != nil { + finalFile.Close() + os.Remove(finalPath) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to merge chunks"}) + return + } + } + finalFile.Close() + + // Clean up temp chunks + os.RemoveAll(tmpDir) + + // Clean up Redis session + h.cache.Del(c.Request.Context(), h.sessionKey(req.UploadID)) + h.cache.Del(c.Request.Context(), h.hashKey(accountID, session.FileHash)) + + urlPath := fmt.Sprintf("/static/videos/%d/%s/%s.mp4", accountID, date, filename) + playURL := buildAbsoluteURL(c, urlPath) + + c.JSON(http.StatusOK, gin.H{ + "url": playURL, + "play_url": playURL, + }) +} diff --git a/backend/internal/video/chunk_handler_test.go b/backend/internal/video/chunk_handler_test.go new file mode 100644 index 0000000..8a09dd2 --- /dev/null +++ b/backend/internal/video/chunk_handler_test.go @@ -0,0 +1,427 @@ +package video + +import ( + "bytes" + "crypto/md5" + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "mime/multipart" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + + rediscache "feedsystem_video_go/internal/middleware/redis" + + "github.com/alicebob/miniredis/v2" + "github.com/gin-gonic/gin" + goredis "github.com/redis/go-redis/v9" +) + +// ── helpers ── + +const testAccountID uint = 1 + +func setupTestEnv(t *testing.T) (*ChunkUploadHandler, func()) { + t.Helper() + gin.SetMode(gin.TestMode) + + mr, err := miniredis.Run() + if err != nil { + t.Fatalf("start miniredis: %v", err) + } + + client := rediscache.NewClient( + goredis.NewClient(&goredis.Options{Addr: mr.Addr()}), + "", + ) + handler := NewChunkUploadHandler(client) + + origDir, _ := os.Getwd() + tmpDir, err := os.MkdirTemp("", "chunk-upload-test-*") + if err != nil { + t.Fatalf("create temp dir: %v", err) + } + if err := os.Chdir(tmpDir); err != nil { + t.Fatalf("chdir: %v", err) + } + + cleanup := func() { + os.Chdir(origDir) + os.RemoveAll(tmpDir) + client.Close() + mr.Close() + } + return handler, cleanup +} + +func newJSONContext(t *testing.T, path string, body interface{}) (*gin.Context, *httptest.ResponseRecorder) { + t.Helper() + b, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshal: %v", err) + } + req := httptest.NewRequest(http.MethodPost, path, bytes.NewReader(b)) + req.Header.Set("Content-Type", "application/json") + req.Host = "localhost" + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = req + c.Set("accountID", testAccountID) + return c, rec +} + +func newMultipartContext(t *testing.T, path string, fields map[string]string, fileContent []byte) (*gin.Context, *httptest.ResponseRecorder) { + t.Helper() + var buf bytes.Buffer + writer := multipart.NewWriter(&buf) + + for k, v := range fields { + if err := writer.WriteField(k, v); err != nil { + t.Fatalf("write field %s: %v", k, err) + } + } + part, err := writer.CreateFormFile("file", "chunk.bin") + if err != nil { + t.Fatalf("create form file: %v", err) + } + if _, err := part.Write(fileContent); err != nil { + t.Fatalf("write file: %v", err) + } + writer.Close() + + req := httptest.NewRequest(http.MethodPost, path, &buf) + req.Header.Set("Content-Type", writer.FormDataContentType()) + req.Host = "localhost" + + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Request = req + c.Set("accountID", testAccountID) + return c, rec +} + +func computeMD5(data []byte) string { + h := md5.Sum(data) + return hex.EncodeToString(h[:]) +} + +func parseJSON(t *testing.T, rec *httptest.ResponseRecorder) map[string]interface{} { + t.Helper() + var m map[string]interface{} + if err := json.Unmarshal(rec.Body.Bytes(), &m); err != nil { + t.Fatalf("parse json: %v, body: %s", err, rec.Body.String()) + } + return m +} + +func makeTestChunks(t *testing.T, totalChunks, chunkSize int) ([][]byte, []string, string) { + t.Helper() + chunks := make([][]byte, totalChunks) + chunkHashes := make([]string, totalChunks) + var full bytes.Buffer + for i := 0; i < totalChunks; i++ { + data := make([]byte, chunkSize) + if _, err := rand.Read(data); err != nil { + t.Fatalf("rand read: %v", err) + } + chunks[i] = data + chunkHashes[i] = computeMD5(data) + full.Write(data) + } + return chunks, chunkHashes, computeMD5(full.Bytes()) +} + +func initUpload(t *testing.T, h *ChunkUploadHandler, filename string, fileSize int64, chunkSize int64, totalChunks int, fileHash string) string { + t.Helper() + c, rec := newJSONContext(t, "/video/chunk/init", InitChunkUploadRequest{ + Filename: filename, + FileSize: fileSize, + ChunkSize: chunkSize, + TotalChunks: totalChunks, + FileHash: fileHash, + }) + h.InitChunkUpload(c) + if rec.Code != http.StatusOK { + t.Fatalf("init: expected 200, got %d, body: %s", rec.Code, rec.Body.String()) + } + resp := parseJSON(t, rec) + return resp["upload_id"].(string) +} + +func uploadChunk(t *testing.T, h *ChunkUploadHandler, uploadID string, chunkIndex int, chunkHash string, chunkData []byte) { + t.Helper() + c, rec := newMultipartContext(t, "/video/chunk/upload", map[string]string{ + "upload_id": uploadID, + "chunk_index": fmt.Sprintf("%d", chunkIndex), + "chunk_hash": chunkHash, + }, chunkData) + h.UploadChunk(c) + if rec.Code != http.StatusOK { + t.Fatalf("upload chunk %d: expected 200, got %d, body: %s", chunkIndex, rec.Code, rec.Body.String()) + } +} + +func completeUpload(t *testing.T, h *ChunkUploadHandler, uploadID string) map[string]interface{} { + t.Helper() + c, rec := newJSONContext(t, "/video/chunk/complete", CompleteChunkUploadRequest{UploadID: uploadID}) + h.CompleteChunkUpload(c) + return parseJSON(t, rec) +} + +// readMergedFile extracts the file path from the response URL and reads it. +// URL format: http://localhost/static/videos///.mp4 +func readMergedFile(t *testing.T, resp map[string]interface{}) []byte { + t.Helper() + urlStr := resp["url"].(string) + idx := len("http://localhost/static/") + fsPath := filepath.Join(".run", "uploads", urlStr[idx:]) + data, err := os.ReadFile(fsPath) + if err != nil { + t.Fatalf("read merged file %s: %v", fsPath, err) + } + return data +} + +// ── tests ── + +func TestFullChunkUploadFlow(t *testing.T) { + h, cleanup := setupTestEnv(t) + defer cleanup() + + chunkSize := 1024 + totalChunks := 3 + chunks, chunkHashes, fileHash := makeTestChunks(t, totalChunks, chunkSize) + + uploadID := initUpload(t, h, "test.mp4", int64(totalChunks*chunkSize), int64(chunkSize), totalChunks, fileHash) + + for i := 0; i < totalChunks; i++ { + uploadChunk(t, h, uploadID, i, chunkHashes[i], chunks[i]) + } + + c, rec := newJSONContext(t, "/video/chunk/complete", CompleteChunkUploadRequest{UploadID: uploadID}) + h.CompleteChunkUpload(c) + if rec.Code != http.StatusOK { + t.Fatalf("complete: expected 200, got %d, body: %s", rec.Code, rec.Body.String()) + } + resp := parseJSON(t, rec) + if resp["url"] == nil || resp["play_url"] == nil { + t.Fatal("complete response missing url or play_url") + } + + // verify merged file content matches original chunks + merged := readMergedFile(t, resp) + var expected bytes.Buffer + for _, ch := range chunks { + expected.Write(ch) + } + if !bytes.Equal(merged, expected.Bytes()) { + t.Fatalf("merged file content mismatch: got %d bytes, want %d bytes", len(merged), expected.Len()) + } + + // verify temp dir cleaned up + tmpDir := filepath.Join(".run", "uploads", "tmp", uploadID) + if _, err := os.Stat(tmpDir); !os.IsNotExist(err) { + t.Fatalf("temp dir should be removed: %s", tmpDir) + } +} + +func TestBreakpointResume(t *testing.T) { + h, cleanup := setupTestEnv(t) + defer cleanup() + + chunkSize := 512 + totalChunks := 4 + chunks, chunkHashes, fileHash := makeTestChunks(t, totalChunks, chunkSize) + + uploadID := initUpload(t, h, "resume.mp4", int64(totalChunks*chunkSize), int64(chunkSize), totalChunks, fileHash) + + // upload first 2 chunks + for i := 0; i < 2; i++ { + uploadChunk(t, h, uploadID, i, chunkHashes[i], chunks[i]) + } + + // re-init with same file_hash → should resume existing session + c, rec := newJSONContext(t, "/video/chunk/init", InitChunkUploadRequest{ + Filename: "resume.mp4", + FileSize: int64(totalChunks * chunkSize), + ChunkSize: int64(chunkSize), + TotalChunks: totalChunks, + FileHash: fileHash, + }) + h.InitChunkUpload(c) + if rec.Code != http.StatusOK { + t.Fatalf("re-init: expected 200, got %d, body: %s", rec.Code, rec.Body.String()) + } + resp := parseJSON(t, rec) + + resumedID := resp["upload_id"].(string) + if resumedID != uploadID { + t.Fatalf("resume should return same upload_id: got %s, want %s", resumedID, uploadID) + } + + chunkList := resp["uploaded_chunks"].([]interface{}) + if len(chunkList) != 2 { + t.Fatalf("expected 2 uploaded chunks on resume, got %d", len(chunkList)) + } + + // upload remaining chunks and complete + for i := 2; i < totalChunks; i++ { + uploadChunk(t, h, uploadID, i, chunkHashes[i], chunks[i]) + } + + resp = completeUpload(t, h, uploadID) + merged := readMergedFile(t, resp) + var expected bytes.Buffer + for _, ch := range chunks { + expected.Write(ch) + } + if !bytes.Equal(merged, expected.Bytes()) { + t.Fatalf("merged file mismatch after resume") + } +} + +func TestIdempotentChunkUpload(t *testing.T) { + h, cleanup := setupTestEnv(t) + defer cleanup() + + chunkSize := 256 + totalChunks := 2 + chunks, chunkHashes, fileHash := makeTestChunks(t, totalChunks, chunkSize) + + uploadID := initUpload(t, h, "idempotent.mp4", int64(totalChunks*chunkSize), int64(chunkSize), totalChunks, fileHash) + + // upload chunk 0 twice — both should succeed + uploadChunk(t, h, uploadID, 0, chunkHashes[0], chunks[0]) + uploadChunk(t, h, uploadID, 0, chunkHashes[0], chunks[0]) + + uploadChunk(t, h, uploadID, 1, chunkHashes[1], chunks[1]) + + c, rec := newJSONContext(t, "/video/chunk/complete", CompleteChunkUploadRequest{UploadID: uploadID}) + h.CompleteChunkUpload(c) + if rec.Code != http.StatusOK { + t.Fatalf("complete: expected 200, got %d, body: %s", rec.Code, rec.Body.String()) + } +} + +func TestHashMismatch(t *testing.T) { + h, cleanup := setupTestEnv(t) + defer cleanup() + + chunkSize := 256 + totalChunks := 1 + chunks, _, fileHash := makeTestChunks(t, totalChunks, chunkSize) + + uploadID := initUpload(t, h, "hashfail.mp4", int64(chunkSize), int64(chunkSize), totalChunks, fileHash) + + // upload with wrong hash + c, rec := newMultipartContext(t, "/video/chunk/upload", map[string]string{ + "upload_id": uploadID, + "chunk_index": "0", + "chunk_hash": "deadbeef000000000000000000000000", + }, chunks[0]) + h.UploadChunk(c) + if rec.Code != http.StatusBadRequest { + t.Fatalf("expected 400 for hash mismatch, got %d", rec.Code) + } + resp := parseJSON(t, rec) + if resp["error"] != "chunk hash mismatch" { + t.Fatalf("unexpected error: %v", resp["error"]) + } + if resp["expected"] == nil || resp["actual"] == nil { + t.Fatal("response should contain expected and actual hash") + } + actualHash := computeMD5(chunks[0]) + if resp["actual"] != actualHash { + t.Fatalf("actual hash: got %v, want %s", resp["actual"], actualHash) + } + + // retry with correct hash should succeed + uploadChunk(t, h, uploadID, 0, actualHash, chunks[0]) +} + +func TestIncompleteMerge(t *testing.T) { + h, cleanup := setupTestEnv(t) + defer cleanup() + + chunkSize := 128 + totalChunks := 5 + chunks, chunkHashes, fileHash := makeTestChunks(t, totalChunks, chunkSize) + + uploadID := initUpload(t, h, "incomplete.mp4", int64(totalChunks*chunkSize), int64(chunkSize), totalChunks, fileHash) + + // upload only 3 out of 5 + for i := 0; i < 3; i++ { + uploadChunk(t, h, uploadID, i, chunkHashes[i], chunks[i]) + } + + c, rec := newJSONContext(t, "/video/chunk/complete", CompleteChunkUploadRequest{UploadID: uploadID}) + h.CompleteChunkUpload(c) + if rec.Code != http.StatusBadRequest { + t.Fatalf("expected 400 for incomplete, got %d", rec.Code) + } + resp := parseJSON(t, rec) + if resp["error"] != "not all chunks uploaded" { + t.Fatalf("unexpected error: %v", resp["error"]) + } + if int(resp["missing"].(float64)) != 2 { + t.Fatalf("expected missing=2, got %v", resp["missing"]) + } + if int(resp["completed"].(float64)) != 3 { + t.Fatalf("expected completed=3, got %v", resp["completed"]) + } + if int(resp["total"].(float64)) != 5 { + t.Fatalf("expected total=5, got %v", resp["total"]) + } + + // upload remaining and retry + for i := 3; i < totalChunks; i++ { + uploadChunk(t, h, uploadID, i, chunkHashes[i], chunks[i]) + } + c, rec = newJSONContext(t, "/video/chunk/complete", CompleteChunkUploadRequest{UploadID: uploadID}) + h.CompleteChunkUpload(c) + if rec.Code != http.StatusOK { + t.Fatalf("complete after fix: expected 200, got %d, body: %s", rec.Code, rec.Body.String()) + } +} + +func TestChunkStatus(t *testing.T) { + h, cleanup := setupTestEnv(t) + defer cleanup() + + chunkSize := 256 + totalChunks := 3 + chunks, chunkHashes, fileHash := makeTestChunks(t, totalChunks, chunkSize) + + uploadID := initUpload(t, h, "status.mp4", int64(totalChunks*chunkSize), int64(chunkSize), totalChunks, fileHash) + + assertStatus := func(wantCount int) { + t.Helper() + c, rec := newJSONContext(t, "/video/chunk/status", ChunkStatusRequest{UploadID: uploadID}) + h.ChunkStatus(c) + if rec.Code != http.StatusOK { + t.Fatalf("status: expected 200, got %d", rec.Code) + } + resp := parseJSON(t, rec) + list, _ := resp["uploaded_chunks"].([]interface{}) + if len(list) != wantCount { + t.Fatalf("expected %d uploaded chunks, got %d", wantCount, len(list)) + } + if int(resp["total_chunks"].(float64)) != totalChunks { + t.Fatalf("expected total_chunks=%d", totalChunks) + } + } + + assertStatus(0) + + uploadChunk(t, h, uploadID, 0, chunkHashes[0], chunks[0]) + assertStatus(1) + + uploadChunk(t, h, uploadID, 1, chunkHashes[1], chunks[1]) + uploadChunk(t, h, uploadID, 2, chunkHashes[2], chunks[2]) + assertStatus(3) +} diff --git a/frontend/package-lock.json b/frontend/package-lock.json index 70bfc23..13e0f85 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -9,6 +9,7 @@ "version": "0.0.0", "dependencies": { "pinia": "^3.0.4", + "spark-md5": "^3.0.2", "vue": "^3.5.24", "vue-router": "^4.6.4" }, @@ -1173,6 +1174,12 @@ "node": ">=0.10.0" } }, + "node_modules/spark-md5": { + "version": "3.0.2", + "resolved": "https://registry.npmmirror.com/spark-md5/-/spark-md5-3.0.2.tgz", + "integrity": "sha512-wcFzz9cDfbuqe0FZzfi2or1sgyIrsDwmPwfZC4hiNidPdPINjeUwNfv5kldczoEAcjl9Y1L3SM7Uz2PUEQzxQw==", + "license": "(WTFPL OR MIT)" + }, "node_modules/speakingurl": { "version": "14.0.1", "resolved": "https://registry.npmmirror.com/speakingurl/-/speakingurl-14.0.1.tgz", diff --git a/frontend/package.json b/frontend/package.json index 939ce30..82e5319 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -10,6 +10,7 @@ }, "dependencies": { "pinia": "^3.0.4", + "spark-md5": "^3.0.2", "vue": "^3.5.24", "vue-router": "^4.6.4" }, diff --git a/frontend/src/api/video.ts b/frontend/src/api/video.ts index 6eee891..e762870 100644 --- a/frontend/src/api/video.ts +++ b/frontend/src/api/video.ts @@ -1,30 +1,68 @@ -import { postForm, postJson } from './client' -import { normalizeVideoList } from './normalize' -import type { Video } from './types' - -export function publishVideo(input: { title: string; description: string; play_url: string; cover_url: string }) { - return postJson