package system import ( "context" "crypto/md5" "encoding/hex" "errors" "fmt" "io" "os" "path" "path/filepath" "sort" "strings" "time" "github.com/google/uuid" ) var ErrUploadSessionNotFound = errors.New("upload session not found") func validMD5(value string) bool { if len(value) != md5.Size*2 { return false } _, err := hex.DecodeString(value) return err == nil } func (uc *MediaUsecase) validateUploadLayout(size, chunkSize int64, total int) error { if size <= 0 || chunkSize <= 0 || total <= 0 { return errors.New("文件大小和分片参数必须大于 0") } if size > uc.maxMediaFileSize() { return ErrMediaTooLarge } expectedTotal := (size-1)/chunkSize + 1 if int64(total) != expectedTotal { return fmt.Errorf("分片数量不匹配: %d/%d", total, expectedTotal) } return nil } func expectedChunkSize(session *UploadSession, index int) (int64, error) { if session == nil || session.FileSize <= 0 || session.ChunkSize <= 0 || session.ChunkTotal <= 0 { return 0, errors.New("上传会话分片参数非法") } if index < 0 || index >= session.ChunkTotal { return 0, errors.New("分片序号超出范围") } if index < session.ChunkTotal-1 { return session.ChunkSize, nil } last := session.FileSize - session.ChunkSize*int64(session.ChunkTotal-1) if last <= 0 || last > session.ChunkSize { return 0, errors.New("上传会话分片参数非法") } return last, nil } func (uc *MediaUsecase) uploadSession(ctx context.Context, id uint) (*UploadSession, error) { session, err := uc.FindUploadSession(ctx, id) if errors.Is(err, ErrUploadSessionNotFound) { return nil, fmt.Errorf("上传会话不存在: %w", ErrUploadSessionNotFound) } if err != nil { return nil, err } return session, nil } func (uc *MediaUsecase) chunkPrefix(uploadID uint) string { directory := "uploads/chunks" if uc.settings != nil { if configured := strings.Trim(uc.settings.MediaSettings().ChunkDir, "/\\ "); configured != "" { directory = configured } } return path.Join(directory, fmt.Sprintf("%d", uploadID)) } func (uc *MediaUsecase) chunkKey(uploadID uint, index int) string { return path.Join(uc.chunkPrefix(uploadID), fmt.Sprintf("%08d", index)) } func (uc *MediaUsecase) InitUpload(ctx context.Context, userID uint, name, hash string, size, chunkSize int64, total int) (*UploadSession, *MediaFile, []int, error) { if err := validateMediaName(name); err != nil { return nil, nil, nil, err } if !validMD5(hash) { return nil, nil, nil, errors.New("文件 MD5 非法") } if err := uc.validateUploadLayout(size, chunkSize, total); err != nil { return nil, nil, nil, err } completed, err := uc.FindCompletedSession(ctx, userID, strings.ToLower(hash)) if err == nil && completed.MediaID != 0 { if media, findErr := uc.FindMedia(ctx, completed.MediaID); findErr == nil { copy := &MediaFile{Name: name, URL: media.URL, Tag: media.Tag, Key: media.Key, Size: media.Size, MD5: media.MD5, Mime: media.Mime, UserID: userID} if createErr := uc.CreateMedia(ctx, copy); createErr == nil { return nil, copy, nil, nil } } } else if err != nil && !errors.Is(err, ErrUploadSessionNotFound) { return nil, nil, nil, err } hash = strings.ToLower(hash) session, err := uc.FindUploadingSession(ctx, userID, hash) if errors.Is(err, ErrUploadSessionNotFound) { session = &UploadSession{UserID: userID, FileName: name, FileHash: hash, FileSize: size, ChunkSize: chunkSize, ChunkTotal: total, Status: "uploading"} if err = uc.CreateUploadSession(ctx, session); err != nil { return nil, nil, nil, err } } else if err != nil { return nil, nil, nil, err } else if session.FileSize != size || session.ChunkSize != chunkSize || session.ChunkTotal != total { return nil, nil, nil, errors.New("上传参数与已有会话不一致") } chunks, err := uc.ListChunks(ctx, session.ID) if err != nil { return nil, nil, nil, err } indices := make([]int, 0, len(chunks)) for _, v := range chunks { indices = append(indices, v.Index) } sort.Ints(indices) return session, nil, indices, nil } func (uc *MediaUsecase) SaveChunk(ctx context.Context, userID, uploadID uint, index int, expected string, reader io.Reader) error { session, err := uc.uploadSession(ctx, uploadID) if err != nil { return err } if session.UserID != userID { return errors.New("无权操作该上传") } if session.Status != "uploading" { return errors.New("上传会话状态不允许收片") } expected = strings.ToLower(strings.TrimSpace(expected)) if !validMD5(expected) { return errors.New("分片 MD5 非法") } wantSize, err := expectedChunkSize(session, index) if err != nil { return err } hash := md5.New() key := uc.chunkKey(uploadID, index) temporary, err := os.CreateTemp("", "kra-upload-chunk-*") if err != nil { return err } temporaryName := temporary.Name() defer os.Remove(temporaryName) defer temporary.Close() written, err := io.Copy(io.MultiWriter(temporary, hash), io.LimitReader(reader, wantSize+1)) if err != nil { return err } if written != wantSize { return fmt.Errorf("分片 %d 大小不匹配: %d/%d", index, written, wantSize) } actual := hex.EncodeToString(hash.Sum(nil)) if !strings.EqualFold(actual, expected) { return fmt.Errorf("分片 %d 校验失败", index) } if _, err = temporary.Seek(0, io.SeekStart); err != nil { return err } stored, err := uc.files.Put(ctx, key, temporary) if err != nil { return err } if stored == nil || stored.Size != written { _ = uc.files.Delete(ctx, key) return errors.New("分片存储大小不一致") } if err = uc.UpsertChunk(ctx, uploadID, &UploadChunk{Index: index, Hash: actual, Size: written}); err != nil { _ = uc.files.Delete(ctx, key) return err } return nil } func (uc *MediaUsecase) CompleteUpload(ctx context.Context, userID, uploadID uint, mime string) (*MediaFile, error) { session, err := uc.uploadSession(ctx, uploadID) if err != nil { return nil, err } if session.UserID != userID { return nil, errors.New("无权操作该上传") } if err = validateMediaName(session.FileName); err != nil { return nil, err } claimed, err := uc.ClaimUploadSession(ctx, uploadID) if err != nil { return nil, err } if !claimed { return nil, errors.New("上传不在可合并状态(可能已在合并或已完成)") } fail := func(value error) (*MediaFile, error) { _ = uc.FailUploadSession(ctx, uploadID) return nil, value } chunks, err := uc.ListChunks(ctx, uploadID) if err != nil { return fail(err) } if len(chunks) != session.ChunkTotal { return fail(fmt.Errorf("分片不全: %d/%d", len(chunks), session.ChunkTotal)) } sort.Slice(chunks, func(i, j int) bool { return chunks[i].Index < chunks[j].Index }) names := make([]string, 0, len(chunks)) var totalSize int64 for index, chunk := range chunks { if chunk.Index != index { return fail(errors.New("分片序号不连续")) } expectedSize, sizeErr := expectedChunkSize(session, index) if sizeErr != nil || chunk.Size != expectedSize || chunk.Size > session.FileSize-totalSize { return fail(errors.New("分片大小不一致")) } totalSize += chunk.Size names = append(names, uc.chunkKey(uploadID, index)) } if totalSize != session.FileSize { return fail(fmt.Errorf("文件大小不匹配: %d/%d", totalSize, session.FileSize)) } ext := strings.ToLower(filepath.Ext(session.FileName)) key := time.Now().Format("20060102") + "/" + uuid.NewString() + ext stored, hash, err := uc.files.Compose(ctx, names, key) if err != nil { return fail(err) } if stored == nil || stored.Size != session.FileSize { if stored != nil { _ = uc.files.Delete(ctx, key) } return fail(errors.New("合并文件大小不匹配")) } if !strings.EqualFold(hash, session.FileHash) { _ = uc.files.Delete(ctx, key) return fail(errors.New("整文件校验失败")) } media := &MediaFile{Name: session.FileName, URL: stored.URL, Tag: strings.TrimPrefix(filepath.Ext(session.FileName), "."), Key: key, Size: stored.Size, Mime: mime, MD5: hash, UserID: userID} if err = uc.CreateMedia(ctx, media); err != nil { return fail(err) } if err = uc.CompleteUploadSession(ctx, uploadID, key, media.ID); err != nil { _ = uc.FailUploadSession(ctx, uploadID) return nil, fmt.Errorf("完成上传会话失败: %w", err) } _ = uc.DeleteChunks(ctx, uploadID) _ = uc.files.DeletePrefix(ctx, uc.chunkPrefix(uploadID)) return media, nil } func (uc *MediaUsecase) CancelUpload(ctx context.Context, userID, uploadID uint) error { session, err := uc.uploadSession(ctx, uploadID) if err != nil { return err } if session.UserID != userID { return errors.New("无权操作该上传") } _ = uc.DeleteChunks(ctx, uploadID) _ = uc.files.DeletePrefix(ctx, uc.chunkPrefix(uploadID)) return uc.DeleteUploadSession(ctx, uploadID) } func (uc *MediaUsecase) CleanupStale(ctx context.Context, ttlHours int) error { if ttlHours <= 0 { ttlHours = 24 } // The reference cleanup job is best-effort: a stale-session query or an // individual storage/database cleanup failure is not propagated to the // scheduler. Keep the endpoint-independent background behavior compatible. ids, _ := uc.StaleUploadSessionIDs(ctx, time.Now().Add(-time.Duration(ttlHours)*time.Hour)) for _, id := range ids { _ = uc.DeleteUploadData(ctx, id) _ = uc.files.DeletePrefix(ctx, uc.chunkPrefix(id)) } return nil }