Support chunk upload and breakpoint resume feature

This commit is contained in:
seaheart
2026-02-08 16:31:56 +08:00
parent 6b09c1ffac
commit 1d65c0e8d6
11 changed files with 1103 additions and 27 deletions
+161
View File
@@ -0,0 +1,161 @@
package rest
import (
"github.com/eyebluecn/tank/code/core"
"github.com/eyebluecn/tank/code/tool/result"
"github.com/eyebluecn/tank/code/tool/util"
"net/http"
)
// ChunkController handles chunked upload API endpoints
type ChunkController struct {
BaseController
chunkService *ChunkService
matterDao *MatterDao
uploadSessionDao *UploadSessionDao
uploadChunkDao *UploadChunkDao
}
func (this *ChunkController) Init() {
this.BaseController.Init()
b := core.CONTEXT.GetBean(this.chunkService)
if b, ok := b.(*ChunkService); ok {
this.chunkService = b
}
b = core.CONTEXT.GetBean(this.matterDao)
if b, ok := b.(*MatterDao); ok {
this.matterDao = b
}
b = core.CONTEXT.GetBean(this.uploadSessionDao)
if b, ok := b.(*UploadSessionDao); ok {
this.uploadSessionDao = b
}
b = core.CONTEXT.GetBean(this.uploadChunkDao)
if b, ok := b.(*UploadChunkDao); ok {
this.uploadChunkDao = b
}
}
func (this *ChunkController) RegisterRoutes() map[string]func(writer http.ResponseWriter, request *http.Request) {
routeMap := make(map[string]func(writer http.ResponseWriter, request *http.Request))
// Chunked upload endpoints
routeMap["/api/chunk/create/session"] = this.Wrap(this.CreateSession, USER_ROLE_USER)
routeMap["/api/chunk/upload"] = this.Wrap(this.Upload, USER_ROLE_USER)
routeMap["/api/chunk/merge"] = this.Wrap(this.Merge, USER_ROLE_USER)
routeMap["/api/chunk/session/info"] = this.Wrap(this.SessionInfo, USER_ROLE_USER)
routeMap["/api/chunk/cancel"] = this.Wrap(this.Cancel, USER_ROLE_USER)
routeMap["/api/chunk/clean/expired"] = this.Wrap(this.CleanExpired, USER_ROLE_ADMINISTRATOR)
return routeMap
}
// CreateSession creates a new chunked upload session
// Parameters:
// - puuid: parent directory uuid
// - filename: target filename
// - totalSize: total file size in bytes
// - chunkSize: size of each chunk (optional, default 5MB)
// - privacy: whether the file is private (optional, default true)
// - fileMd5: MD5 of the whole file (optional, for verification)
// - spaceUuid: space uuid (optional, default user's private space)
func (this *ChunkController) CreateSession(writer http.ResponseWriter, request *http.Request) *result.WebResult {
puuid := util.ExtractRequestString(request, "puuid")
filename := util.ExtractRequestString(request, "filename")
totalSize := util.ExtractRequestInt64(request, "totalSize")
chunkSize := util.ExtractRequestOptionalInt64(request, "chunkSize", DEFAULT_CHUNK_SIZE)
privacy := util.ExtractRequestOptionalBool(request, "privacy", true)
fileMd5 := util.ExtractRequestOptionalString(request, "fileMd5", "")
user := this.checkUser(request)
spaceUuid := util.ExtractRequestOptionalString(request, "spaceUuid", user.SpaceUuid)
space := this.spaceService.CheckWritableByUuid(request, user, spaceUuid)
dirMatter := this.matterDao.CheckWithRootByUuid(puuid, space)
session := this.chunkService.CreateSession(request, user, space, dirMatter, filename, totalSize, chunkSize, privacy, fileMd5)
return this.Success(session)
}
// Upload uploads a single chunk
// Parameters:
// - sessionUuid: upload session uuid
// - chunkIndex: chunk index (0-based)
// - file: chunk file data
// - chunkMd5: MD5 of the chunk (optional, for verification)
// - spaceUuid: space uuid (optional)
func (this *ChunkController) Upload(writer http.ResponseWriter, request *http.Request) *result.WebResult {
sessionUuid := util.ExtractRequestString(request, "sessionUuid")
chunkIndex := util.ExtractRequestInt(request, "chunkIndex")
chunkMd5 := util.ExtractRequestOptionalString(request, "chunkMd5", "")
user := this.checkUser(request)
spaceUuid := util.ExtractRequestOptionalString(request, "spaceUuid", user.SpaceUuid)
space := this.spaceService.CheckWritableByUuid(request, user, spaceUuid)
file, _, err := request.FormFile("file")
if err != nil {
panic(result.BadRequest("file is required: %s", err.Error()))
}
defer file.Close()
chunk := this.chunkService.UploadChunk(request, sessionUuid, chunkIndex, file, chunkMd5, user, space)
return this.Success(chunk)
}
// Merge merges all chunks into the final file
// Parameters:
// - sessionUuid: upload session uuid
// - spaceUuid: space uuid (optional)
func (this *ChunkController) Merge(writer http.ResponseWriter, request *http.Request) *result.WebResult {
sessionUuid := util.ExtractRequestString(request, "sessionUuid")
user := this.checkUser(request)
spaceUuid := util.ExtractRequestOptionalString(request, "spaceUuid", user.SpaceUuid)
space := this.spaceService.CheckWritableByUuid(request, user, spaceUuid)
matter := this.chunkService.MergeChunks(request, sessionUuid, user, space)
return this.Success(matter)
}
// SessionInfo gets the upload session info including uploaded chunks
// Parameters:
// - sessionUuid: upload session uuid
func (this *ChunkController) SessionInfo(writer http.ResponseWriter, request *http.Request) *result.WebResult {
sessionUuid := util.ExtractRequestString(request, "sessionUuid")
user := this.checkUser(request)
sessionInfo := this.chunkService.GetSessionInfo(sessionUuid, user)
return this.Success(sessionInfo)
}
// Cancel cancels an upload session
// Parameters:
// - sessionUuid: upload session uuid
// - spaceUuid: space uuid (optional)
func (this *ChunkController) Cancel(writer http.ResponseWriter, request *http.Request) *result.WebResult {
sessionUuid := util.ExtractRequestString(request, "sessionUuid")
user := this.checkUser(request)
spaceUuid := util.ExtractRequestOptionalString(request, "spaceUuid", user.SpaceUuid)
space := this.spaceService.CheckWritableByUuid(request, user, spaceUuid)
this.chunkService.CancelSession(sessionUuid, user, space)
return this.Success("OK")
}
// CleanExpired cleans up expired upload sessions (admin only)
func (this *ChunkController) CleanExpired(writer http.ResponseWriter, request *http.Request) *result.WebResult {
this.chunkService.CleanExpiredSessions()
return this.Success("OK")
}
+212
View File
@@ -0,0 +1,212 @@
package rest
import (
"github.com/eyebluecn/tank/code/core"
"github.com/eyebluecn/tank/code/tool/result"
"github.com/eyebluecn/tank/code/tool/uuid"
"time"
)
// UploadSessionDao handles database operations for UploadSession
type UploadSessionDao struct {
BaseDao
}
// FindByUuid finds an upload session by uuid
func (this *UploadSessionDao) FindByUuid(sessionUuid string) *UploadSession {
var entity = &UploadSession{}
db := core.CONTEXT.GetDB().Where("uuid = ?", sessionUuid).First(entity)
if db.Error != nil {
if db.Error.Error() == result.DB_ERROR_NOT_FOUND {
return nil
} else {
panic(db.Error)
}
}
return entity
}
// CheckByUuid checks if session exists, panics if not found
func (this *UploadSessionDao) CheckByUuid(sessionUuid string) *UploadSession {
entity := this.FindByUuid(sessionUuid)
if entity == nil {
panic(result.NotFound("upload session not found with uuid = %s", sessionUuid))
}
return entity
}
// FindByUserUuidAndFileMd5 finds session by user and file MD5 (for quick upload)
func (this *UploadSessionDao) FindByUserUuidAndFileMd5(userUuid string, fileMd5 string) *UploadSession {
var entity = &UploadSession{}
db := core.CONTEXT.GetDB().Where("user_uuid = ? AND file_md5 = ? AND status = ?",
userUuid, fileMd5, UPLOAD_SESSION_STATUS_COMPLETED).First(entity)
if db.Error != nil {
if db.Error.Error() == result.DB_ERROR_NOT_FOUND {
return nil
} else {
panic(db.Error)
}
}
return entity
}
// FindBySpaceUuidAndFileMd5 finds completed session by space and file MD5
func (this *UploadSessionDao) FindBySpaceUuidAndFileMd5(spaceUuid string, fileMd5 string) *UploadSession {
var entity = &UploadSession{}
db := core.CONTEXT.GetDB().Where("space_uuid = ? AND file_md5 = ? AND status = ?",
spaceUuid, fileMd5, UPLOAD_SESSION_STATUS_COMPLETED).First(entity)
if db.Error != nil {
if db.Error.Error() == result.DB_ERROR_NOT_FOUND {
return nil
} else {
panic(db.Error)
}
}
return entity
}
// Create creates a new upload session
func (this *UploadSessionDao) Create(session *UploadSession) *UploadSession {
timeUUID, _ := uuid.NewV4()
session.Uuid = timeUUID.String()
session.CreateTime = time.Now()
session.UpdateTime = time.Now()
session.Sort = time.Now().UnixNano() / 1e6
db := core.CONTEXT.GetDB().Create(session)
this.PanicError(db.Error)
return session
}
// Save updates an upload session
func (this *UploadSessionDao) Save(session *UploadSession) *UploadSession {
session.UpdateTime = time.Now()
db := core.CONTEXT.GetDB().Save(session)
this.PanicError(db.Error)
return session
}
// Delete deletes an upload session
func (this *UploadSessionDao) Delete(session *UploadSession) {
db := core.CONTEXT.GetDB().Delete(session)
this.PanicError(db.Error)
}
// DeleteByUuid deletes session by uuid
func (this *UploadSessionDao) DeleteByUuid(sessionUuid string) {
db := core.CONTEXT.GetDB().Where("uuid = ?", sessionUuid).Delete(UploadSession{})
this.PanicError(db.Error)
}
// FindExpiredSessions finds all expired sessions
func (this *UploadSessionDao) FindExpiredSessions() []*UploadSession {
var sessions []*UploadSession
db := core.CONTEXT.GetDB().Where("expire_time < ? AND status = ?",
time.Now(), UPLOAD_SESSION_STATUS_UPLOADING).Find(&sessions)
this.PanicError(db.Error)
return sessions
}
// FindByUserUuid finds all sessions for a user
func (this *UploadSessionDao) FindByUserUuid(userUuid string) []*UploadSession {
var sessions []*UploadSession
db := core.CONTEXT.GetDB().Where("user_uuid = ?", userUuid).Find(&sessions)
this.PanicError(db.Error)
return sessions
}
// DeleteByUserUuid deletes all sessions for a user
func (this *UploadSessionDao) DeleteByUserUuid(userUuid string) {
db := core.CONTEXT.GetDB().Where("user_uuid = ?", userUuid).Delete(UploadSession{})
this.PanicError(db.Error)
}
// UploadChunkDao handles database operations for UploadChunk
type UploadChunkDao struct {
BaseDao
}
// FindByUuid finds a chunk by uuid
func (this *UploadChunkDao) FindByUuid(chunkUuid string) *UploadChunk {
var entity = &UploadChunk{}
db := core.CONTEXT.GetDB().Where("uuid = ?", chunkUuid).First(entity)
if db.Error != nil {
if db.Error.Error() == result.DB_ERROR_NOT_FOUND {
return nil
} else {
panic(db.Error)
}
}
return entity
}
// FindBySessionUuidAndChunkIndex finds a chunk by session and index
func (this *UploadChunkDao) FindBySessionUuidAndChunkIndex(sessionUuid string, chunkIndex int) *UploadChunk {
var entity = &UploadChunk{}
db := core.CONTEXT.GetDB().Where("session_uuid = ? AND chunk_index = ?", sessionUuid, chunkIndex).First(entity)
if db.Error != nil {
if db.Error.Error() == result.DB_ERROR_NOT_FOUND {
return nil
} else {
panic(db.Error)
}
}
return entity
}
// FindBySessionUuid finds all chunks for a session
func (this *UploadChunkDao) FindBySessionUuid(sessionUuid string) []*UploadChunk {
var chunks []*UploadChunk
db := core.CONTEXT.GetDB().Where("session_uuid = ?", sessionUuid).Order("chunk_index asc").Find(&chunks)
this.PanicError(db.Error)
return chunks
}
// FindUploadedBySessionUuid finds all uploaded chunks for a session
func (this *UploadChunkDao) FindUploadedBySessionUuid(sessionUuid string) []*UploadChunk {
var chunks []*UploadChunk
db := core.CONTEXT.GetDB().Where("session_uuid = ? AND uploaded = ?", sessionUuid, true).Order("chunk_index asc").Find(&chunks)
this.PanicError(db.Error)
return chunks
}
// CountUploadedBySessionUuid counts uploaded chunks for a session
func (this *UploadChunkDao) CountUploadedBySessionUuid(sessionUuid string) int64 {
var count int64
db := core.CONTEXT.GetDB().Model(&UploadChunk{}).Where("session_uuid = ? AND uploaded = ?", sessionUuid, true).Count(&count)
this.PanicError(db.Error)
return count
}
// Create creates a new chunk record
func (this *UploadChunkDao) Create(chunk *UploadChunk) *UploadChunk {
timeUUID, _ := uuid.NewV4()
chunk.Uuid = timeUUID.String()
chunk.CreateTime = time.Now()
chunk.UpdateTime = time.Now()
chunk.Sort = time.Now().UnixNano() / 1e6
db := core.CONTEXT.GetDB().Create(chunk)
this.PanicError(db.Error)
return chunk
}
// Save updates a chunk record
func (this *UploadChunkDao) Save(chunk *UploadChunk) *UploadChunk {
chunk.UpdateTime = time.Now()
db := core.CONTEXT.GetDB().Save(chunk)
this.PanicError(db.Error)
return chunk
}
// Delete deletes a chunk record
func (this *UploadChunkDao) Delete(chunk *UploadChunk) {
db := core.CONTEXT.GetDB().Delete(chunk)
this.PanicError(db.Error)
}
// DeleteBySessionUuid deletes all chunks for a session
func (this *UploadChunkDao) DeleteBySessionUuid(sessionUuid string) {
db := core.CONTEXT.GetDB().Where("session_uuid = ?", sessionUuid).Delete(UploadChunk{})
this.PanicError(db.Error)
}
+76
View File
@@ -0,0 +1,76 @@
package rest
import (
"time"
)
const (
// Upload session status
UPLOAD_SESSION_STATUS_UPLOADING = "UPLOADING"
UPLOAD_SESSION_STATUS_MERGING = "MERGING"
UPLOAD_SESSION_STATUS_COMPLETED = "COMPLETED"
UPLOAD_SESSION_STATUS_ERROR = "ERROR"
// Chunk directory name
CHUNK_DIR = "chunk"
// Default chunk size: 50MB
DEFAULT_CHUNK_SIZE = 50 * 1024 * 1024
// Max chunk size: 100MB
MAX_CHUNK_SIZE = 100 * 1024 * 1024
// Min chunk size: 5MB (aligned with AWS S3 standard)
MIN_CHUNK_SIZE = 5 * 1024 * 1024
)
// UploadSession represents a chunked upload session
type UploadSession struct {
Uuid string `json:"uuid" gorm:"type:char(36);primary_key;unique"`
Sort int64 `json:"sort" gorm:"type:bigint(20) not null"`
UpdateTime time.Time `json:"updateTime" gorm:"type:timestamp not null;default:CURRENT_TIMESTAMP"`
CreateTime time.Time `json:"createTime" gorm:"type:timestamp not null;default:'2018-01-01 00:00:00'"`
UserUuid string `json:"userUuid" gorm:"type:char(36) not null;index:idx_upload_session_uu"`
SpaceUuid string `json:"spaceUuid" gorm:"type:char(36) not null;index:idx_upload_session_su"`
MatterUuid string `json:"matterUuid" gorm:"type:char(36)"` // Final matter uuid after merge
Puuid string `json:"puuid" gorm:"type:char(36) not null"` // Parent directory uuid
Filename string `json:"filename" gorm:"type:varchar(255) not null"` // Target filename
TotalSize int64 `json:"totalSize" gorm:"type:bigint(20) not null;default:0"` // Total file size
ChunkSize int64 `json:"chunkSize" gorm:"type:bigint(20) not null;default:52428800"` // Size of each chunk (default 50MB)
TotalChunks int `json:"totalChunks" gorm:"type:int not null;default:0"` // Total number of chunks
Privacy bool `json:"privacy" gorm:"type:tinyint(1) not null;default:1"` // Privacy flag
FileMd5 string `json:"fileMd5" gorm:"type:varchar(45)"` // MD5 of the whole file
Status string `json:"status" gorm:"type:varchar(20) not null;default:'UPLOADING'"` // Session status
ExpireTime time.Time `json:"expireTime" gorm:"type:timestamp not null;default:'2018-01-01 00:00:00'"` // Expiration time
}
// UploadChunk represents a single chunk in an upload session
type UploadChunk struct {
Uuid string `json:"uuid" gorm:"type:char(36);primary_key;unique"`
Sort int64 `json:"sort" gorm:"type:bigint(20) not null"`
UpdateTime time.Time `json:"updateTime" gorm:"type:timestamp not null;default:CURRENT_TIMESTAMP"`
CreateTime time.Time `json:"createTime" gorm:"type:timestamp not null;default:'2018-01-01 00:00:00'"`
SessionUuid string `json:"sessionUuid" gorm:"type:char(36) not null;index:idx_upload_chunk_su"` // Reference to UploadSession
ChunkIndex int `json:"chunkIndex" gorm:"type:int not null"` // Chunk sequence number (0-based)
ChunkSize int64 `json:"chunkSize" gorm:"type:bigint(20) not null;default:0"` // Actual size of this chunk
ChunkMd5 string `json:"chunkMd5" gorm:"type:varchar(45)"` // MD5 of this chunk
ChunkFilename string `json:"chunkFilename" gorm:"type:varchar(255) not null"` // Temp filename for this chunk
Uploaded bool `json:"uploaded" gorm:"type:tinyint(1) not null;default:0"` // Whether this chunk is uploaded
}
// Get the chunk temp directory for a space
func GetSpaceChunkDir(spaceName string) string {
return GetUserSpaceRootDir(spaceName) + "/" + CHUNK_DIR
}
// Get the session chunk directory
func GetSessionChunkDir(spaceName string, sessionUuid string) string {
return GetSpaceChunkDir(spaceName) + "/" + sessionUuid
}
// UploadSessionInfo is the response structure for session queries
type UploadSessionInfo struct {
Session *UploadSession `json:"session"`
UploadedChunks []int `json:"uploadedChunks"` // List of uploaded chunk indices
MissingChunks []int `json:"missingChunks"` // List of missing chunk indices
}
+581
View File
@@ -0,0 +1,581 @@
package rest
import (
"crypto/md5"
"encoding/hex"
"fmt"
"io"
"net/http"
"os"
"sort"
"sync"
"time"
"github.com/eyebluecn/tank/code/core"
"github.com/eyebluecn/tank/code/tool/i18n"
"github.com/eyebluecn/tank/code/tool/result"
"github.com/eyebluecn/tank/code/tool/util"
)
// ChunkService handles chunked upload operations
//@Service
type ChunkService struct {
BaseBean
uploadSessionDao *UploadSessionDao
uploadChunkDao *UploadChunkDao
matterDao *MatterDao
matterService *MatterService
spaceDao *SpaceDao
userService *UserService
// chunkLocker is used to prevent concurrent uploads of the same chunk
chunkLocker sync.Map
// mergeLocker is used to prevent concurrent merge of the same session
mergeLocker sync.Map
}
func (this *ChunkService) Init() {
this.BaseBean.Init()
b := core.CONTEXT.GetBean(this.uploadSessionDao)
if b, ok := b.(*UploadSessionDao); ok {
this.uploadSessionDao = b
}
b = core.CONTEXT.GetBean(this.uploadChunkDao)
if b, ok := b.(*UploadChunkDao); ok {
this.uploadChunkDao = b
}
b = core.CONTEXT.GetBean(this.matterDao)
if b, ok := b.(*MatterDao); ok {
this.matterDao = b
}
b = core.CONTEXT.GetBean(this.matterService)
if b, ok := b.(*MatterService); ok {
this.matterService = b
}
b = core.CONTEXT.GetBean(this.spaceDao)
if b, ok := b.(*SpaceDao); ok {
this.spaceDao = b
}
b = core.CONTEXT.GetBean(this.userService)
if b, ok := b.(*UserService); ok {
this.userService = b
}
}
// CreateSession creates a new upload session
func (this *ChunkService) CreateSession(
request *http.Request,
user *User,
space *Space,
dirMatter *Matter,
filename string,
totalSize int64,
chunkSize int64,
privacy bool,
fileMd5 string,
) *UploadSession {
if user == nil {
panic(result.BadRequest("user cannot be nil."))
}
if dirMatter == nil {
panic(result.BadRequest("dirMatter cannot be nil."))
}
if dirMatter.Deleted {
panic(result.BadRequest("Dir has been deleted. Cannot upload under it."))
}
// Validate filename
if len(filename) > MATTER_NAME_MAX_LENGTH {
panic(result.BadRequestI18n(request, i18n.MatterNameLengthExceedLimit, len(filename), MATTER_NAME_MAX_LENGTH))
}
CheckMatterName(request, filename)
// Validate chunk size
if chunkSize < MIN_CHUNK_SIZE {
chunkSize = MIN_CHUNK_SIZE
}
if chunkSize > MAX_CHUNK_SIZE {
chunkSize = MAX_CHUNK_SIZE
}
// Validate total size
if totalSize <= 0 {
panic(result.BadRequest("totalSize must be positive"))
}
// Check space size limit
if space.SizeLimit >= 0 {
if totalSize > space.SizeLimit {
panic(result.BadRequestI18n(request, i18n.MatterSizeExceedLimit, util.HumanFileSize(totalSize), util.HumanFileSize(space.SizeLimit)))
}
}
// Check total space limit
if space.TotalSizeLimit >= 0 {
if space.TotalSize+totalSize > space.TotalSizeLimit {
panic(result.BadRequestI18n(request, i18n.MatterSizeExceedTotalLimit, util.HumanFileSize(space.TotalSize), util.HumanFileSize(space.TotalSizeLimit)))
}
}
// Check if file already exists
dbMatter := this.matterDao.FindBySpaceUuidAndPuuidAndDirAndName(space.Uuid, dirMatter.Uuid, false, filename)
if dbMatter != nil {
if dbMatter.Deleted {
panic(result.BadRequestI18n(request, i18n.MatterRecycleBinExist, filename))
} else {
panic(result.BadRequestI18n(request, i18n.MatterExist, filename))
}
}
// Calculate total chunks
totalChunks := int((totalSize + chunkSize - 1) / chunkSize)
// Create session
session := &UploadSession{
UserUuid: user.Uuid,
SpaceUuid: space.Uuid,
Puuid: dirMatter.Uuid,
Filename: filename,
TotalSize: totalSize,
ChunkSize: chunkSize,
TotalChunks: totalChunks,
Privacy: privacy,
FileMd5: fileMd5,
Status: UPLOAD_SESSION_STATUS_UPLOADING,
ExpireTime: time.Now().Add(24 * time.Hour), // 24 hours expiration
}
session = this.uploadSessionDao.Create(session)
// Create chunk directory
chunkDir := GetSessionChunkDir(space.Name, session.Uuid)
util.MakeDirAll(chunkDir)
this.logger.Info("Created upload session %s for file %s, total chunks: %d", session.Uuid, filename, totalChunks)
return session
}
// UploadChunk uploads a single chunk with optional MD5 verification
func (this *ChunkService) UploadChunk(
request *http.Request,
sessionUuid string,
chunkIndex int,
file io.Reader,
clientChunkMd5 string,
user *User,
space *Space,
) *UploadChunk {
// Check session
session := this.uploadSessionDao.CheckByUuid(sessionUuid)
// Validate session ownership
if session.UserUuid != user.Uuid {
panic(result.BadRequest("session does not belong to you"))
}
if session.Status != UPLOAD_SESSION_STATUS_UPLOADING {
panic(result.BadRequest("session status is not uploading"))
}
// Check expiration
if time.Now().After(session.ExpireTime) {
panic(result.BadRequest("upload session has expired"))
}
// Validate chunk index
if chunkIndex < 0 || chunkIndex >= session.TotalChunks {
panic(result.BadRequest("invalid chunk index: %d, total chunks: %d", chunkIndex, session.TotalChunks))
}
// Acquire chunk-level lock to prevent concurrent upload of the same chunk
chunkLockKey := fmt.Sprintf("%s_%d", sessionUuid, chunkIndex)
if _, loaded := this.chunkLocker.LoadOrStore(chunkLockKey, true); loaded {
panic(result.BadRequest("chunk %d is being uploaded by another request, please retry later", chunkIndex))
}
defer this.chunkLocker.Delete(chunkLockKey)
// Check if chunk already exists
existingChunk := this.uploadChunkDao.FindBySessionUuidAndChunkIndex(sessionUuid, chunkIndex)
if existingChunk != nil && existingChunk.Uploaded {
this.logger.Info("Chunk %d already uploaded for session %s", chunkIndex, sessionUuid)
return existingChunk
}
// Prepare chunk file path
chunkFilename := fmt.Sprintf("chunk_%d", chunkIndex)
chunkDir := GetSessionChunkDir(space.Name, sessionUuid)
chunkFilePath := chunkDir + "/" + chunkFilename
// Write chunk to disk (use 0600 for security - only owner can read/write)
destFile, err := os.OpenFile(chunkFilePath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
if err != nil {
panic(result.BadRequest("failed to create chunk file: %s", err.Error()))
}
// Calculate MD5 while writing
hash := md5.New()
multiWriter := io.MultiWriter(destFile, hash)
chunkSize, err := io.Copy(multiWriter, file)
destFile.Close()
if err != nil {
os.Remove(chunkFilePath)
panic(result.BadRequest("failed to write chunk: %s", err.Error()))
}
// Validate chunk size to prevent malicious oversized uploads
isLastChunk := chunkIndex == session.TotalChunks-1
if isLastChunk {
// Last chunk: should be <= session.ChunkSize and match expected remainder
expectedLastChunkSize := session.TotalSize - int64(session.TotalChunks-1)*session.ChunkSize
if chunkSize != expectedLastChunkSize {
os.Remove(chunkFilePath)
panic(result.BadRequest("last chunk size mismatch: expected %d, got %d", expectedLastChunkSize, chunkSize))
}
} else {
// Non-last chunks: should be exactly session.ChunkSize
if chunkSize != session.ChunkSize {
os.Remove(chunkFilePath)
panic(result.BadRequest("chunk size mismatch: expected %d, got %d", session.ChunkSize, chunkSize))
}
}
chunkMd5 := hex.EncodeToString(hash.Sum(nil))
// Verify chunk MD5 if client provided one
if clientChunkMd5 != "" && clientChunkMd5 != chunkMd5 {
os.Remove(chunkFilePath)
this.logger.Error("Chunk MD5 verification failed for chunk %d: client=%s, server=%s", chunkIndex, clientChunkMd5, chunkMd5)
panic(result.BadRequest("chunk MD5 verification failed: data may be corrupted during upload"))
}
this.logger.Info("Uploaded chunk %d for session %s, size: %d, md5: %s", chunkIndex, sessionUuid, chunkSize, chunkMd5)
// Create or update chunk record
var chunk *UploadChunk
if existingChunk != nil {
existingChunk.ChunkSize = chunkSize
existingChunk.ChunkMd5 = chunkMd5
existingChunk.ChunkFilename = chunkFilename
existingChunk.Uploaded = true
chunk = this.uploadChunkDao.Save(existingChunk)
} else {
chunk = &UploadChunk{
SessionUuid: sessionUuid,
ChunkIndex: chunkIndex,
ChunkSize: chunkSize,
ChunkMd5: chunkMd5,
ChunkFilename: chunkFilename,
Uploaded: true,
}
chunk = this.uploadChunkDao.Create(chunk)
}
return chunk
}
// GetSessionInfo gets the upload session info including uploaded chunks
func (this *ChunkService) GetSessionInfo(sessionUuid string, user *User) *UploadSessionInfo {
session := this.uploadSessionDao.CheckByUuid(sessionUuid)
// Validate ownership
if session.UserUuid != user.Uuid {
panic(result.BadRequest("session does not belong to you"))
}
// Get uploaded chunks
uploadedChunks := this.uploadChunkDao.FindUploadedBySessionUuid(sessionUuid)
uploadedIndices := make([]int, 0)
uploadedMap := make(map[int]bool)
for _, chunk := range uploadedChunks {
uploadedIndices = append(uploadedIndices, chunk.ChunkIndex)
uploadedMap[chunk.ChunkIndex] = true
}
// Calculate missing chunks
missingIndices := make([]int, 0)
for i := 0; i < session.TotalChunks; i++ {
if !uploadedMap[i] {
missingIndices = append(missingIndices, i)
}
}
return &UploadSessionInfo{
Session: session,
UploadedChunks: uploadedIndices,
MissingChunks: missingIndices,
}
}
// MergeChunks merges all chunks into final file
func (this *ChunkService) MergeChunks(
request *http.Request,
sessionUuid string,
user *User,
space *Space,
) *Matter {
// Use session-level lock instead of user-level lock to allow concurrent uploads of different files
if _, loaded := this.mergeLocker.LoadOrStore(sessionUuid, true); loaded {
panic(result.BadRequest("merge operation is already in progress for this session"))
}
defer this.mergeLocker.Delete(sessionUuid)
session := this.uploadSessionDao.CheckByUuid(sessionUuid)
// Validate ownership
if session.UserUuid != user.Uuid {
panic(result.BadRequest("session does not belong to you"))
}
// Check session expiration
if time.Now().After(session.ExpireTime) {
panic(result.BadRequest("upload session has expired, please start a new upload"))
}
if session.Status == UPLOAD_SESSION_STATUS_COMPLETED {
// Already completed, return the matter
if session.MatterUuid != "" {
return this.matterDao.CheckByUuid(session.MatterUuid)
}
panic(result.BadRequest("session completed but matter not found"))
}
// Allow retry from ERROR status (e.g., previous merge failed)
if session.Status != UPLOAD_SESSION_STATUS_UPLOADING && session.Status != UPLOAD_SESSION_STATUS_ERROR {
panic(result.BadRequest("session status is not uploading or error, current: %s", session.Status))
}
// If retrying from ERROR status, reset to UPLOADING first
if session.Status == UPLOAD_SESSION_STATUS_ERROR {
this.logger.Info("Retrying merge for session %s from ERROR status", sessionUuid)
session.Status = UPLOAD_SESSION_STATUS_UPLOADING
this.uploadSessionDao.Save(session)
}
// Check if all chunks are uploaded
uploadedCount := this.uploadChunkDao.CountUploadedBySessionUuid(sessionUuid)
if int(uploadedCount) < session.TotalChunks {
panic(result.BadRequest("not all chunks uploaded: %d/%d", uploadedCount, session.TotalChunks))
}
// Update status to merging
session.Status = UPLOAD_SESSION_STATUS_MERGING
this.uploadSessionDao.Save(session)
// Get parent directory
var dirMatter *Matter
if session.Puuid == MATTER_ROOT {
dirMatter = NewRootMatter(space)
} else {
dirMatter = this.matterDao.CheckByUuid(session.Puuid)
}
// Prepare final file path
dirAbsolutePath := dirMatter.AbsolutePath()
util.MakeDirAll(dirAbsolutePath)
fileAbsolutePath := dirAbsolutePath + "/" + session.Filename
// Check if file exists
if util.PathExists(fileAbsolutePath) {
this.logger.Error("%s exists, removing it.", fileAbsolutePath)
os.Remove(fileAbsolutePath)
}
// Create final file (use 0644 for normal file permissions)
destFile, err := os.OpenFile(fileAbsolutePath, os.O_WRONLY|os.O_CREATE, 0644)
if err != nil {
session.Status = UPLOAD_SESSION_STATUS_ERROR
this.uploadSessionDao.Save(session)
panic(result.BadRequest("failed to create destination file: %s", err.Error()))
}
// Get all chunks sorted by index
chunks := this.uploadChunkDao.FindBySessionUuid(sessionUuid)
sort.Slice(chunks, func(i, j int) bool {
return chunks[i].ChunkIndex < chunks[j].ChunkIndex
})
chunkDir := GetSessionChunkDir(space.Name, sessionUuid)
// Merge chunks
var totalWritten int64 = 0
for _, chunk := range chunks {
chunkPath := chunkDir + "/" + chunk.ChunkFilename
chunkFile, err := os.Open(chunkPath)
if err != nil {
destFile.Close()
os.Remove(fileAbsolutePath)
session.Status = UPLOAD_SESSION_STATUS_ERROR
this.uploadSessionDao.Save(session)
panic(result.BadRequest("failed to open chunk %d: %s", chunk.ChunkIndex, err.Error()))
}
written, err := io.Copy(destFile, chunkFile)
chunkFile.Close()
if err != nil {
destFile.Close()
os.Remove(fileAbsolutePath)
session.Status = UPLOAD_SESSION_STATUS_ERROR
this.uploadSessionDao.Save(session)
panic(result.BadRequest("failed to write chunk %d: %s", chunk.ChunkIndex, err.Error()))
}
totalWritten += written
}
destFile.Close()
this.logger.Info("Merged %d chunks into file %s, total size: %d", len(chunks), session.Filename, totalWritten)
// Verify file size
if totalWritten != session.TotalSize {
os.Remove(fileAbsolutePath)
session.Status = UPLOAD_SESSION_STATUS_ERROR
this.uploadSessionDao.Save(session)
panic(result.BadRequest("merged file size mismatch: expected %d, got %d", session.TotalSize, totalWritten))
}
// Verify file MD5 if provided
if session.FileMd5 != "" {
mergedFile, err := os.Open(fileAbsolutePath)
if err != nil {
os.Remove(fileAbsolutePath)
session.Status = UPLOAD_SESSION_STATUS_ERROR
this.uploadSessionDao.Save(session)
panic(result.BadRequest("failed to open merged file for MD5 verification: %s", err.Error()))
}
hash := md5.New()
_, err = io.Copy(hash, mergedFile)
mergedFile.Close()
if err != nil {
os.Remove(fileAbsolutePath)
session.Status = UPLOAD_SESSION_STATUS_ERROR
this.uploadSessionDao.Save(session)
panic(result.BadRequest("failed to calculate MD5 of merged file: %s", err.Error()))
}
calculatedMd5 := hex.EncodeToString(hash.Sum(nil))
if calculatedMd5 != session.FileMd5 {
os.Remove(fileAbsolutePath)
session.Status = UPLOAD_SESSION_STATUS_ERROR
this.uploadSessionDao.Save(session)
this.logger.Error("MD5 verification failed for session %s: expected %s, got %s", sessionUuid, session.FileMd5, calculatedMd5)
panic(result.BadRequest("MD5 verification failed: file may be corrupted during upload"))
}
this.logger.Info("MD5 verification passed for session %s", sessionUuid)
}
// Create matter record
dirRelativePath := dirMatter.Path
fileRelativePath := dirRelativePath + "/" + session.Filename
matter := &Matter{
Puuid: dirMatter.Uuid,
UserUuid: user.Uuid,
SpaceName: space.Name,
SpaceUuid: space.Uuid,
Dir: false,
Name: session.Filename,
Md5: session.FileMd5,
Size: totalWritten,
Privacy: session.Privacy,
Path: fileRelativePath,
Prop: EMPTY_JSON_MAP,
VisitTime: time.Now(),
}
matter = this.matterDao.Create(matter)
// Update session status
session.Status = UPLOAD_SESSION_STATUS_COMPLETED
session.MatterUuid = matter.Uuid
this.uploadSessionDao.Save(session)
// Clean up chunk files
go core.RunWithRecovery(func() {
this.cleanupChunks(space.Name, sessionUuid)
this.matterService.ComputeRouteSize(dirMatter.Uuid, user, space)
})
this.logger.Info("Upload session %s completed, matter uuid: %s", sessionUuid, matter.Uuid)
return matter
}
// cleanupChunks removes chunk files and records with improved atomicity
// First delete database records, then delete files (safer approach)
func (this *ChunkService) cleanupChunks(spaceName string, sessionUuid string) {
chunkDir := GetSessionChunkDir(spaceName, sessionUuid)
// First, remove chunk records from database
// This ensures that even if file deletion fails, we don't have orphaned DB records
this.uploadChunkDao.DeleteBySessionUuid(sessionUuid)
// Then remove chunk directory with retry mechanism
maxRetries := 3
for i := 0; i < maxRetries; i++ {
err := os.RemoveAll(chunkDir)
if err == nil {
this.logger.Info("Cleaned up chunks for session %s", sessionUuid)
return
}
this.logger.Error("failed to remove chunk dir %s (attempt %d/%d): %s", chunkDir, i+1, maxRetries, err.Error())
// Wait before retry (exponential backoff)
if i < maxRetries-1 {
time.Sleep(time.Duration(100*(i+1)) * time.Millisecond)
}
}
// If all retries failed, log a critical warning
this.logger.Error("CRITICAL: Failed to remove chunk dir %s after %d attempts. Manual cleanup may be required.", chunkDir, maxRetries)
}
// CancelSession cancels an upload session
func (this *ChunkService) CancelSession(sessionUuid string, user *User, space *Space) {
session := this.uploadSessionDao.CheckByUuid(sessionUuid)
// Validate ownership
if session.UserUuid != user.Uuid {
panic(result.BadRequest("session does not belong to you"))
}
// Clean up
this.cleanupChunks(space.Name, sessionUuid)
// Delete session
this.uploadSessionDao.Delete(session)
this.logger.Info("Cancelled upload session %s", sessionUuid)
}
// CleanExpiredSessions cleans up expired sessions
func (this *ChunkService) CleanExpiredSessions() {
sessions := this.uploadSessionDao.FindExpiredSessions()
for _, session := range sessions {
space := this.spaceDao.FindByUuid(session.SpaceUuid)
if space != nil {
this.cleanupChunks(space.Name, session.Uuid)
}
this.uploadSessionDao.Delete(session)
this.logger.Info("Cleaned expired session %s", session.Uuid)
}
}
+2
View File
@@ -81,6 +81,8 @@ func (this *InstallController) Init() {
&Space{},
&SpaceMember{},
&UploadToken{},
&UploadSession{},
&UploadChunk{},
&User{},
}
+1 -3
View File
@@ -483,9 +483,7 @@ func (this *MatterService) Upload(request *http.Request, file io.Reader, fileHea
panic(result.BadRequest("Dir has been deleted. Cannot upload under it."))
}
if len(filename) > MATTER_NAME_MAX_LENGTH {
panic(result.BadRequestI18n(request, i18n.MatterNameLengthExceedLimit, len(filename), MATTER_NAME_MAX_LENGTH))
}
CheckMatterName(request, filename)
//if fileHeader.Size not nill . check size in advance.
if fileHeader != nil {
+7
View File
@@ -89,6 +89,7 @@ func (this *PreferenceController) Edit(writer http.ResponseWriter, request *http
downloadDirMaxNumStr := request.FormValue("downloadDirMaxNum")
defaultTotalSizeLimitStr := request.FormValue("defaultTotalSizeLimit")
allowRegisterStr := request.FormValue("allowRegister")
allowChunkUploadStr := request.FormValue("allowChunkUpload")
deletedKeepDaysStr := request.FormValue("deletedKeepDays")
if name == "" {
@@ -140,6 +141,11 @@ func (this *PreferenceController) Edit(writer http.ResponseWriter, request *http
allowRegister = true
}
var allowChunkUpload = false
if allowChunkUploadStr == TRUE {
allowChunkUpload = true
}
preference := this.preferenceDao.Fetch()
oldDeletedKeepDays := preference.DeletedKeepDays
preference.Name = name
@@ -151,6 +157,7 @@ func (this *PreferenceController) Edit(writer http.ResponseWriter, request *http
preference.DownloadDirMaxNum = downloadDirMaxNum
preference.DefaultTotalSizeLimit = defaultTotalSizeLimit
preference.AllowRegister = allowRegister
preference.AllowChunkUpload = allowChunkUpload
preference.DeletedKeepDays = deletedKeepDays
preference = this.preferenceService.Save(preference)
+1
View File
@@ -19,6 +19,7 @@ type Preference struct {
DownloadDirMaxNum int64 `json:"downloadDirMaxNum" gorm:"type:bigint(20) not null;default:-1"`
DefaultTotalSizeLimit int64 `json:"defaultTotalSizeLimit" gorm:"type:bigint(20) not null;default:-1"`
AllowRegister bool `json:"allowRegister" gorm:"type:tinyint(1) not null;default:0"`
AllowChunkUpload bool `json:"allowChunkUpload" gorm:"type:tinyint(1) not null;default:0"`
PreviewConfig string `json:"previewConfig" gorm:"type:text"`
ScanConfig string `json:"scanConfig" gorm:"type:text"`
DeletedKeepDays int64 `json:"deletedKeepDays" gorm:"type:bigint(20) not null;default:7"`
+25 -20
View File
@@ -1,6 +1,9 @@
package rest
import (
"sync"
"time"
"github.com/eyebluecn/tank/code/core"
"github.com/eyebluecn/tank/code/tool/cache"
"github.com/eyebluecn/tank/code/tool/result"
@@ -8,7 +11,6 @@ import (
"github.com/eyebluecn/tank/code/tool/uuid"
"net/http"
"os"
"time"
)
// @Service
@@ -19,8 +21,10 @@ type UserService struct {
spaceService *SpaceService
//file lock
//file lock (legacy, kept for compatibility)
locker *cache.Table
// matterLocker is used for atomic matter operations with retry support
matterLocker sync.Map
matterDao *MatterDao
matterService *MatterService
@@ -100,32 +104,33 @@ func (this *UserService) Init() {
this.locker = cache.NewTable()
}
// lock a user's operation. If lock, user cannot operate file.
// lock a user's operation with retry mechanism.
// This allows concurrent operations to wait for each other instead of immediately failing.
// Uses sync.Map for thread-safe atomic operations.
func (this *UserService) MatterLock(userUuid string) {
maxRetries := 30 // Maximum retry attempts (30 * 200ms = 6 seconds max wait)
retryInterval := 200 // Retry interval in milliseconds
cacheItem, err := this.locker.Value(userUuid)
if err != nil {
this.logger.Error("error while get cache" + err.Error())
for i := 0; i < maxRetries; i++ {
// Try to acquire lock atomically
if _, loaded := this.matterLocker.LoadOrStore(userUuid, true); !loaded {
// Successfully acquired the lock
return
}
// Lock is held by another operation, wait and retry
if i < maxRetries-1 {
time.Sleep(time.Duration(retryInterval) * time.Millisecond)
}
}
if cacheItem != nil && cacheItem.Data() != nil {
panic(result.BadRequest("file is being operating, retry later"))
}
duration := 12 * time.Hour
this.locker.Add(userUuid, duration, true)
// After all retries, still cannot acquire lock
panic(result.BadRequest("file is being operating, retry later"))
}
// unlock
func (this *UserService) MatterUnlock(userUuid string) {
exist := this.locker.Exists(userUuid)
if exist {
_, err := this.locker.Delete(userUuid)
this.PanicError(err)
} else {
this.logger.Error("unlock error. %s has no matter lock ", userUuid)
}
this.matterLocker.Delete(userUuid)
}
// load session to SessionCache. This method will be invoked in every request.
+6
View File
@@ -217,6 +217,12 @@ func (this *TankContext) registerBeans() {
//uploadToken
this.registerBean(new(rest.UploadTokenDao))
//chunk upload
this.registerBean(new(rest.ChunkController))
this.registerBean(new(rest.ChunkService))
this.registerBean(new(rest.UploadSessionDao))
this.registerBean(new(rest.UploadChunkDao))
//task
this.registerBean(new(rest.TaskService))
+31 -4
View File
@@ -46,16 +46,14 @@ func ExtractRequestArray(request *http.Request, key string) []string {
func ExtractRequestInt64(request *http.Request, key string) int64 {
keyStr := request.FormValue(key)
var num int64 = 0
if keyStr == "" {
panic(fmt.Sprintf("%s is required", key))
} else {
intVal, err := strconv.Atoi(keyStr)
intVal, err := strconv.ParseInt(keyStr, 10, 64)
if err != nil {
panic(err)
}
num = int64(intVal)
return num
return intVal
}
}
@@ -76,7 +74,36 @@ func ExtractRequestInt64WithErrorHint(request *http.Request, key string, errorHi
}
}
// param is optional. when missing, use default.
func ExtractRequestOptionalInt64(request *http.Request, key string, defaultValue int64) int64 {
str := request.FormValue(key)
if str == "" {
return defaultValue
} else {
intVal, err := strconv.ParseInt(str, 10, 64)
if err != nil {
panic(err)
}
return intVal
}
}
// param is required. when missing, panic error.
func ExtractRequestInt(request *http.Request, key string) int {
keyStr := request.FormValue(key)
if keyStr == "" {
panic(fmt.Sprintf("%s is required", key))
} else {
intVal, err := strconv.Atoi(keyStr)
if err != nil {
panic(err)
}
return intVal
}
}
// param is optional. when missing, use default.
func ExtractRequestOptionalInt(request *http.Request, key string, defaultValue int) int {
str := request.FormValue(key)
if str == "" {