kaifabu/SeeyonFileSystem
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

初始化2

Browse Source
This commit is contained in:
RuicyWu
2026-07-10 17:33:33 +08:00
parent b51ee98afa
commit 94f6ecf901
59 changed files with 3893 additions and 980 deletions
Download Patch File Download Diff File Expand all files Collapse all files

170
server/service/chunk_upload_service.go
View File

@@ -11,6 +11,7 @@ import (
"sort"
"strconv"
"strings"
"time"
"seeyon-filesystem/config"
"seeyon-filesystem/dto"
@@ -34,6 +35,7 @@ func NewChunkUploadService() *ChunkUploadService {
}
// InitUpload 初始化分片上传
// 根据存储策略自动选择: 本地模式(服务端中转) 或 MinIO模式(客户端直传)
func (s *ChunkUploadService) InitUpload(fileName string, fileSize int64, md5 string, folderID, ownerID, policyID uint64) (*dto.ChunkUploadInitVO, error) {
// 秒传检查
if md5 != "" {
@@ -49,19 +51,28 @@ func (s *ChunkUploadService) InitUpload(fileName string, fileSize int64, md5 str
}
}
// 计算分片参数(5MB每片)
// 计算分片参数
chunkSize := int64(5 * 1024 * 1024)
if fileSize < chunkSize {
chunkSize = fileSize
}
totalChunks := int((fileSize + chunkSize - 1) / chunkSize)
// 创建临时目录
uploadID := uuid.New().String()
// 生成存储路径
ext := utils.GetExtension(fileName)
var storageKey string
if md5 != "" {
storageKey = storage.GenerateStorageKey(md5, ext)
} else {
storageKey = storage.GenerateStorageKey(fmt.Sprintf("%d-%d", ownerID, time.Now().UnixNano()), ext)
}
// 创建上传会话
tempDir := filepath.Join(config.AppConfig.Storage.LocalBasePath, ".tmp", uploadID)
os.MkdirAll(tempDir, 0755)
// 创建上传会话
session := &model.UploadSession{
ID: uploadID,
FileName: fileName,
@@ -73,22 +84,51 @@ func (s *ChunkUploadService) InitUpload(fileName string, fileSize int64, md5 str
OwnerID: ownerID,
PolicyID: policyID,
Status: 1,
TempDir: tempDir,
}
if err := config.DB.Create(session).Error; err != nil {
return nil, fmt.Errorf("创建上传会话失败: %w", err)
TempDir: storageKey, // 复用字段存储目标storageKey
}
config.DB.Create(session)
// 检查是否为MinIO存储
engine, err := storage.GetEngineByPolicyID(policyID)
if err == nil {
if minioEngine, ok := engine.(*storage.MinIOEngine); ok {
return s.initMinIOUpload(minioEngine, session, storageKey)
}
}
// 本地存储模式
return &dto.ChunkUploadInitVO{
UploadID: uploadID,
Instant: false,
ChunkSize: chunkSize,
Total: totalChunks,
Mode: "local",
}, nil
}
// UploadChunk 上传单个分片
// initMinIOUpload 初始化MinIO直传模式
func (s *ChunkUploadService) initMinIOUpload(minioEngine *storage.MinIOEngine, session *model.UploadSession, storageKey string) (*dto.ChunkUploadInitVO, error) {
// 为每个分片生成预签名上传URL
parts := make([]dto.ChunkPresignURL, session.TotalChunks)
for i := 0; i < session.TotalChunks; i++ {
chunkKey := fmt.Sprintf("%s.chunk_%06d", storageKey, i)
presignedURL, err := minioEngine.GetPresignedChunkURL(chunkKey, 24*time.Hour)
if err != nil {
presignedURL = ""
}
parts[i] = dto.ChunkPresignURL{Index: i, URL: presignedURL}
}
return &dto.ChunkUploadInitVO{
UploadID: session.ID,
ChunkSize: session.ChunkSize,
Total: session.TotalChunks,
Mode: "minio",
Parts: parts,
}, nil
}
// UploadChunk 上传单个分片(本地模式)
func (s *ChunkUploadService) UploadChunk(uploadID string, index int, reader io.Reader, checksum string) (*dto.ChunkUploadChunkVO, error) {
var session model.UploadSession
if err := config.DB.First(&session, uploadID).Error; err != nil {
@@ -101,14 +141,12 @@ func (s *ChunkUploadService) UploadChunk(uploadID string, index int, reader io.R
return nil, errors.New("分片索引无效")
}
// 检查是否已上传
uploaded := parseChunkIndices(session.UploadedChunks)
if uploaded[index] {
return &dto.ChunkUploadChunkVO{Index: index, Status: "exists"}, nil
}
// 写入临时文件
chunkPath := filepath.Join(session.TempDir, fmt.Sprintf("chunk_%06d", index))
chunkPath := filepath.Join(config.AppConfig.Storage.LocalBasePath, ".tmp", uploadID, fmt.Sprintf("chunk_%06d", index))
dst, err := os.Create(chunkPath)
if err != nil {
return nil, fmt.Errorf("创建分片文件失败: %w", err)
@@ -123,14 +161,12 @@ func (s *ChunkUploadService) UploadChunk(uploadID string, index int, reader io.R
return nil, fmt.Errorf("写入分片失败: %w", err)
}
// checksum校验
actualChecksum := hex.EncodeToString(hash.Sum(nil))
if checksum != "" && checksum != actualChecksum {
os.Remove(chunkPath)
return nil, fmt.Errorf("分片校验失败: 期望 %s, 实际 %s", checksum, actualChecksum)
}
// 更新进度
uploaded[index] = true
chunkStr := buildChunkIndices(uploaded)
config.DB.Model(&session).Update("uploaded_chunks", chunkStr)
@@ -138,6 +174,24 @@ func (s *ChunkUploadService) UploadChunk(uploadID string, index int, reader io.R
return &dto.ChunkUploadChunkVO{Index: index, Size: size, Status: "ok"}, nil
}
// UploadChunkMinIO 确认MinIO分片上传完成(客户端直传后调用)
func (s *ChunkUploadService) UploadChunkMinIO(uploadID string, index int, etag string) (*dto.ChunkUploadChunkVO, error) {
var session model.UploadSession
if err := config.DB.First(&session, uploadID).Error; err != nil {
return nil, errors.New("上传会话不存在")
}
if session.Status != 1 {
return nil, errors.New("上传会话已结束")
}
uploaded := parseChunkIndices(session.UploadedChunks)
uploaded[index] = true
chunkStr := buildChunkIndices(uploaded)
config.DB.Model(&session).Update("uploaded_chunks", chunkStr)
return &dto.ChunkUploadChunkVO{Index: index, Status: "ok", ETag: etag}, nil
}
// MergeChunks 合并分片完成上传
func (s *ChunkUploadService) MergeChunks(uploadID string) (*dto.FileVO, error) {
var session model.UploadSession
@@ -148,7 +202,6 @@ func (s *ChunkUploadService) MergeChunks(uploadID string) (*dto.FileVO, error) {
return nil, errors.New("上传会话已结束")
}
// 检查完整性
uploaded := parseChunkIndices(session.UploadedChunks)
for i := 0; i < session.TotalChunks; i++ {
if !uploaded[i] {
@@ -156,14 +209,30 @@ func (s *ChunkUploadService) MergeChunks(uploadID string) (*dto.FileVO, error) {
}
}
// 合并分片
mergedPath := filepath.Join(session.TempDir, "merged")
storageKey := session.TempDir // 目标storageKey
// 检测MinIO模式
engine, _ := storage.GetEngineByPolicyID(session.PolicyID)
if minioEngine, ok := engine.(*storage.MinIOEngine); ok {
return s.mergeMinIO(minioEngine, &session, storageKey)
}
// 本地模式: 合并分片文件
return s.mergeLocal(&session, storageKey)
}
// mergeLocal 本地模式合并
func (s *ChunkUploadService) mergeLocal(session *model.UploadSession, storageKey string) (*dto.FileVO, error) {
tmpDir := filepath.Join(config.AppConfig.Storage.LocalBasePath, ".tmp", session.ID)
mergedPath := filepath.Join(tmpDir, "merged")
merged, err := os.Create(mergedPath)
if err != nil {
return nil, fmt.Errorf("创建合并文件失败: %w", err)
}
for i := 0; i < session.TotalChunks; i++ {
chunkPath := filepath.Join(session.TempDir, fmt.Sprintf("chunk_%06d", i))
chunkPath := filepath.Join(tmpDir, fmt.Sprintf("chunk_%06d", i))
chunk, err := os.Open(chunkPath)
if err != nil {
merged.Close()
@@ -174,22 +243,20 @@ func (s *ChunkUploadService) MergeChunks(uploadID string) (*dto.FileVO, error) {
}
merged.Close()
// 计算MD5
fileMD5, _ := utils.FileMD5(mergedPath)
fileSHA256, _ := utils.FileSHA256(mergedPath)
// 验证MD5
if session.MD5 != "" && session.MD5 != fileMD5 {
os.RemoveAll(session.TempDir)
return nil, fmt.Errorf("文件MD5校验失败: 期望 %s, 实际 %s", session.MD5, fileMD5)
os.RemoveAll(tmpDir)
return nil, fmt.Errorf("MD5校验失败: 期望 %s, 实际 %s", session.MD5, fileMD5)
}
// 秒传检查
existing, _ := s.fileRepo.FindByMD5(fileMD5)
if existing != nil {
os.RemoveAll(session.TempDir)
os.RemoveAll(tmpDir)
newFile := &model.File{
Name: session.FileName, Extension: utils.GetExtension(session.FileName),
ID: utils.GenID(), Name: session.FileName, Extension: utils.GetExtension(session.FileName),
FolderID: session.FolderID, OwnerID: session.OwnerID, Size: session.FileSize,
MD5: fileMD5, StorageKey: existing.StorageKey, StoragePolicyID: existing.StoragePolicyID,
MimeType: utils.GetMimeType(utils.GetExtension(session.FileName)), Status: 1,
@@ -200,36 +267,60 @@ func (s *ChunkUploadService) MergeChunks(uploadID string) (*dto.FileVO, error) {
// 上传到存储引擎
ext := utils.GetExtension(session.FileName)
storageKey := storage.GenerateStorageKey(fileMD5, ext)
engine, err := storage.GetEngineByPolicyID(session.PolicyID)
if err != nil {
engine = storage.GetDefaultEngine()
}
engine, _ := storage.GetEngineByPolicyID(session.PolicyID)
reader, _ := os.Open(mergedPath)
defer reader.Close()
if err := engine.Upload(storageKey, reader, session.FileSize); err != nil {
os.RemoveAll(session.TempDir)
os.RemoveAll(tmpDir)
return nil, fmt.Errorf("存储文件失败: %w", err)
}
// 保存记录
newFile := &model.File{
Name: session.FileName, Extension: ext, FolderID: session.FolderID, OwnerID: session.OwnerID,
ID: utils.GenID(), Name: session.FileName, Extension: ext, FolderID: session.FolderID, OwnerID: session.OwnerID,
Size: session.FileSize, MD5: fileMD5, SHA256: fileSHA256, StorageKey: storageKey,
StoragePolicyID: session.PolicyID, MimeType: utils.GetMimeType(ext), Status: 1,
}
if err := s.fileRepo.Create(newFile); err != nil {
engine.Delete(storageKey)
os.RemoveAll(session.TempDir)
os.RemoveAll(tmpDir)
return nil, fmt.Errorf("保存文件记录失败: %w", err)
}
config.DB.Model(&session).Updates(map[string]interface{}{"status": 2, "file_id": newFile.ID, "md5": fileMD5, "sha256": fileSHA256})
os.RemoveAll(session.TempDir)
config.DB.Model(session).Updates(map[string]interface{}{"status": 2, "file_id": newFile.ID, "md5": fileMD5, "sha256": fileSHA256})
os.RemoveAll(tmpDir)
return fileToVO(newFile), nil
}
// GetUploadProgress 查询上传进度(断点续传)
// mergeMinIO MinIO模式合并
func (s *ChunkUploadService) mergeMinIO(minioEngine *storage.MinIOEngine, session *model.UploadSession, storageKey string) (*dto.FileVO, error) {
// 构建分片key列表
chunkKeys := make([]string, session.TotalChunks)
for i := 0; i < session.TotalChunks; i++ {
chunkKeys[i] = fmt.Sprintf("%s.chunk_%06d", storageKey, i)
}
// 合并分片
if err := minioEngine.MergeChunks(chunkKeys, storageKey); err != nil {
return nil, fmt.Errorf("MinIO合并失败: %w", err)
}
ext := utils.GetExtension(session.FileName)
fileMD5 := session.MD5 // 使用客户端提供的MD5
newFile := &model.File{
ID: utils.GenID(), Name: session.FileName, Extension: ext, FolderID: session.FolderID, OwnerID: session.OwnerID,
Size: session.FileSize, MD5: fileMD5, StorageKey: storageKey,
StoragePolicyID: session.PolicyID, MimeType: utils.GetMimeType(ext), Status: 1,
}
if err := s.fileRepo.Create(newFile); err != nil {
return nil, fmt.Errorf("保存文件记录失败: %w", err)
}
config.DB.Model(session).Updates(map[string]interface{}{"status": 2, "file_id": newFile.ID, "md5": fileMD5})
return fileToVO(newFile), nil
}
// GetUploadProgress 查询上传进度
func (s *ChunkUploadService) GetUploadProgress(uploadID string) (*dto.ChunkUploadProgressVO, error) {
var session model.UploadSession
if err := config.DB.First(&session, uploadID).Error; err != nil {
@@ -255,7 +346,7 @@ func (s *ChunkUploadService) CancelUpload(uploadID string) error {
if err := config.DB.First(&session, uploadID).Error; err != nil {
return errors.New("上传会话不存在")
}
os.RemoveAll(session.TempDir)
os.RemoveAll(filepath.Join(config.AppConfig.Storage.LocalBasePath, ".tmp", uploadID))
config.DB.Model(&session).Update("status", 3)
return nil
}
@@ -288,7 +379,8 @@ func buildChunkIndices(m map[int]bool) string {
func fileToVO(file *model.File) *dto.FileVO {
vo := &dto.FileVO{
ID: file.ID, Name: file.Name, Extension: file.Extension, FolderID: file.FolderID,
ID: dto.StringID(file.ID), Name: file.Name, Extension: file.Extension, FolderID: dto.StringID(file.FolderID),
OwnerID: dto.StringID(file.OwnerID),
Size: file.Size, StorageKey: file.StorageKey, StoragePolicyID: file.StoragePolicyID,
MimeType: file.MimeType, IsFavorite: file.IsFavorite, DownloadCount: file.DownloadCount,
Version: file.Version, CreatedAt: file.CreatedAt, UpdatedAt: file.UpdatedAt,
@@ -299,6 +391,8 @@ func fileToVO(file *model.File) *dto.FileVO {
if localEngine, ok := engine.(*storage.LocalEngine); ok {
vo.DownloadURL = localEngine.GetURL(file.StorageKey)
vo.PreviewURL = localEngine.GetURL(file.StorageKey)
} else if minioEngine, ok := engine.(*storage.MinIOEngine); ok {
vo.PreviewURL = minioEngine.GetPreviewURL(file.StorageKey)
}
}
}