From 1d65c0e8d6526ba750e238be0dabb466fb280e28 Mon Sep 17 00:00:00 2001 From: seaheart Date: Sun, 8 Feb 2026 16:31:56 +0800 Subject: [PATCH] Support chunk upload and breakpoint resume feature --- code/rest/chunk_controller.go | 161 ++++++++ code/rest/chunk_dao.go | 212 +++++++++++ code/rest/chunk_model.go | 76 ++++ code/rest/chunk_service.go | 581 +++++++++++++++++++++++++++++ code/rest/install_controller.go | 2 + code/rest/matter_service.go | 4 +- code/rest/preference_controller.go | 7 + code/rest/preference_model.go | 1 + code/rest/user_service.go | 45 ++- code/support/tank_context.go | 6 + code/tool/util/util_params.go | 35 +- 11 files changed, 1103 insertions(+), 27 deletions(-) create mode 100644 code/rest/chunk_controller.go create mode 100644 code/rest/chunk_dao.go create mode 100644 code/rest/chunk_model.go create mode 100644 code/rest/chunk_service.go diff --git a/code/rest/chunk_controller.go b/code/rest/chunk_controller.go new file mode 100644 index 0000000..7859e9e --- /dev/null +++ b/code/rest/chunk_controller.go @@ -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") +} diff --git a/code/rest/chunk_dao.go b/code/rest/chunk_dao.go new file mode 100644 index 0000000..2065055 --- /dev/null +++ b/code/rest/chunk_dao.go @@ -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) +} diff --git a/code/rest/chunk_model.go b/code/rest/chunk_model.go new file mode 100644 index 0000000..3407eca --- /dev/null +++ b/code/rest/chunk_model.go @@ -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 +} diff --git a/code/rest/chunk_service.go b/code/rest/chunk_service.go new file mode 100644 index 0000000..5acf711 --- /dev/null +++ b/code/rest/chunk_service.go @@ -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) + } +} diff --git a/code/rest/install_controller.go b/code/rest/install_controller.go index 80ff8e5..f6bf757 100644 --- a/code/rest/install_controller.go +++ b/code/rest/install_controller.go @@ -81,6 +81,8 @@ func (this *InstallController) Init() { &Space{}, &SpaceMember{}, &UploadToken{}, + &UploadSession{}, + &UploadChunk{}, &User{}, } diff --git a/code/rest/matter_service.go b/code/rest/matter_service.go index c6d3721..e537e00 100644 --- a/code/rest/matter_service.go +++ b/code/rest/matter_service.go @@ -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 { diff --git a/code/rest/preference_controller.go b/code/rest/preference_controller.go index 4eabc29..3b4a216 100644 --- a/code/rest/preference_controller.go +++ b/code/rest/preference_controller.go @@ -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) diff --git a/code/rest/preference_model.go b/code/rest/preference_model.go index 973e28b..42cc39a 100644 --- a/code/rest/preference_model.go +++ b/code/rest/preference_model.go @@ -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"` diff --git a/code/rest/user_service.go b/code/rest/user_service.go index db7b64c..97e779a 100644 --- a/code/rest/user_service.go +++ b/code/rest/user_service.go @@ -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. diff --git a/code/support/tank_context.go b/code/support/tank_context.go index 8954a19..8b49249 100644 --- a/code/support/tank_context.go +++ b/code/support/tank_context.go @@ -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)) diff --git a/code/tool/util/util_params.go b/code/tool/util/util_params.go index 1bade3a..513598d 100644 --- a/code/tool/util/util_params.go +++ b/code/tool/util/util_params.go @@ -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 == "" {