280 lines
9.0 KiB
Go
280 lines
9.0 KiB
Go
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}
|
|
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 || 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)
|
|
}
|
|
_ = uc.CompleteUploadSession(ctx, uploadID, key, media.ID)
|
|
_ = 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
|
|
}
|