From aee457093debfaf7ec53b5e5cd5c73a67e49ff33 Mon Sep 17 00:00:00 2001 From: ryan Date: Sat, 13 Jun 2026 16:46:05 +0800 Subject: [PATCH] chore(release): v1.1.0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### 🛠 修复 - 修复了 CAPTCHA 验证的逻辑和安全性。 ### ⚡️ 优化与改进 - 新增了动态存储配置与多后端迁移机制。 - 新增了手动触发存储迁移的 Web GUI 操作界面。 - 实现了存储迁移的并发处理（使用 Group）与上传后 SHA-256 完整性自动校验机制。 - 实现了基于 Redis 的分布式锁与集群多节点缓存失效广播机制，防止迁移任务冲突。 - 实现了 Redis 锁续期守护协程 (watchdog) 防止超长迁移任务锁过期。 - 优化了更新存储配置时的连通性自动校验机制，防止配置错误。 - 优化了存储配置的内存缓存机制，并引入了全局共享连接池以提高 TCP 复用率。 - 优化了上传前已有同名文件的比对逻辑，跳过下载阶段，仅比对 Content-Length 以实现零网络流量跳过。 - 新增了普通用户的独立文件管理 Dashboard 与控制接口。 - 优化了文件管理接口命名空间，将全局管理迁移至管理员级 API Namespace 隔离控制。 - 优化了管理员新建用户接口，增加了邮箱（Email）字段的必填要求。 ### 💄 其他/体验 - 重构并优化了文件管理页面，引入了多标签页多维度统计数据大屏。 - 基于 shadcn 原生 UI 组件重构并优化了文件详情和仪表盘组件。 - 抽取后端核心路由注册流程，降低主路由文件圈复杂度，使路由配置更易维护。 --- .../apps/upload/storage_migration_task.go | 217 +++++++++++++----- internal/task/executor.go | 8 +- 2 files changed, 168 insertions(+), 57 deletions(-) diff --git a/internal/apps/upload/storage_migration_task.go b/internal/apps/upload/storage_migration_task.go index c642b51d..2c40826c 100644 --- a/internal/apps/upload/storage_migration_task.go +++ b/internal/apps/upload/storage_migration_task.go @@ -29,6 +29,8 @@ const ( StorageMigrationTask = "storage:migrate" // TaskTypeStorageMigration is the task metadata type for storage migration. TaskTypeStorageMigration = "storage_migration" + + colStorageDriver = "storage_driver" ) // StorageMigrationMeta describes the manually dispatchable migration task. @@ -41,6 +43,16 @@ var StorageMigrationMeta = task.TaskMeta{ MaxRetry: task.DefaultMaxRetry, Queue: task.QueueDefault, Retryable: true, + Params: []task.TaskParam{ + { + Name: "target", + Label: "目标存储配置 (JSON)", + Type: "text", + Required: true, + Placeholder: `{"driver": "s3", "local": {"root": "."}, "s3": {"bucket": "my-bucket", ...}}`, + Description: "待迁移到的目标存储引擎完整配置 JSON 字符串", + }, + }, } // MigrationHandler copies stored objects and activates the target backend. @@ -82,7 +94,7 @@ func (h *MigrationHandler) Execute(ctx context.Context, payload []byte) (*task.T if !ok { return nil, errors.New("另一个存储迁移任务正在运行中") } - + // 任务结束时清理锁，使用 Background context 避免受任务 context 取消的影响 stopRenewal := make(chan struct{}) //nolint:contextcheck @@ -182,15 +194,40 @@ func parseMigrationTargetConfig(ctx context.Context, payload []byte) (storage.Co if strings.TrimSpace(string(payload)) == "" { return storage.Config{}, errors.New("storage migration target payload is required") } - var req storageMigrationPayload - if err := json.Unmarshal(payload, &req); err != nil { - return storage.Config{}, fmt.Errorf("parse storage migration payload: %w", err) + + // Try to parse using raw JSON message to handle both struct and string payload formats + var raw struct { + Target json.RawMessage `json:"target"` } + if err := json.Unmarshal(payload, &raw); err != nil { + return storage.Config{}, fmt.Errorf("parse storage migration payload envelope: %w", err) + } + + if len(raw.Target) == 0 { + return storage.Config{}, errors.New("storage migration target payload is required") + } + + var targetBytes []byte + var targetStr string + // Check if Target is a JSON string + if err := json.Unmarshal(raw.Target, &targetStr); err == nil { + // It is a string (e.g. from dynamic form input), parse its content as JSON + targetBytes = []byte(targetStr) + } else { + // It is a JSON object, use directly + targetBytes = raw.Target + } + + var target storage.Config + if err := json.Unmarshal(targetBytes, &target); err != nil { + return storage.Config{}, fmt.Errorf("parse target storage config: %w", err) + } + current, err := storage.LoadConfig(ctx) if err != nil { return storage.Config{}, fmt.Errorf("load active storage config: %w", err) } - target := storage.MergeMaskedSecrets(req.Target, current) + target = storage.MergeMaskedSecrets(target, current) if err := storage.ValidateConfig(target); err != nil { return storage.Config{}, fmt.Errorf("validate target storage config: %w", err) } @@ -246,6 +283,8 @@ func migrateObjects( return atomic.LoadInt64(&migrated), fmt.Errorf("storage migration canceled: %w", err) } + task.AppendLog(ctx, "正在查询待迁移对象批次，当前已完成迁移: %d/%d", atomic.LoadInt64(&migrated), total) + var objects []struct { FilePath string `gorm:"column:file_path"` FileSize int64 `gorm:"column:file_size"` @@ -262,65 +301,20 @@ func migrateObjects( return atomic.LoadInt64(&migrated), fmt.Errorf("query source objects: %w", err) } if len(objects) == 0 { + task.AppendLog(ctx, "所有对象迁移完毕") break } + task.AppendLog(ctx, "获取当前批次迁移对象，批次大小: %d，实际获取对象数: %d", batchSize, len(objects)) + var g errgroup.Group g.SetLimit(migrationConcurrency) for _, object := range objects { obj := object // Capture range variable g.Go(func() error { - source, err := sourceBackend.Get(ctx, obj.FilePath) - if err != nil { - if isNotFoundError(err) { - task.AppendLog(ctx, "警告: 源存储中物理文件不存在，标记为已删除并跳过: %s (错误: %v)", obj.FilePath, err) - if updateErr := db.DB(ctx).Model(&model.Upload{}). - Where("storage_driver = ? AND file_path = ?", sourceDriver, obj.FilePath). - Updates(map[string]any{ - "status": model.UploadStatusDeleted, - "storage_driver": targetDriver, - }).Error; updateErr != nil { - return fmt.Errorf("update missing object %q: %w", obj.FilePath, updateErr) - } - return nil - } - return fmt.Errorf("open source object %q: %w", obj.FilePath, err) - } - targetResult, putErr := targetBackend.Put(ctx, obj.FilePath, source.Body, obj.FileSize, obj.MimeType) - closeErr := source.Body.Close() - if putErr != nil { - return fmt.Errorf("copy object %q: %w", obj.FilePath, putErr) - } - if closeErr != nil { - return fmt.Errorf("close source object %q: %w", obj.FilePath, closeErr) - } - - // Data integrity check (SHA-256 hash verification) - if len(obj.Hash) == sha256HexLength { - targetObj, getErr := targetBackend.Get(ctx, targetResult.Key) - if getErr != nil { - return fmt.Errorf("retrieve target object for verification %q: %w", obj.FilePath, getErr) - } - h := sha256.New() - if _, copyErr := io.Copy(h, targetObj.Body); copyErr != nil { - _ = targetObj.Body.Close() - return fmt.Errorf("read target object for verification %q: %w", obj.FilePath, copyErr) - } - _ = targetObj.Body.Close() - computedHash := hex.EncodeToString(h.Sum(nil)) - if computedHash != obj.Hash { - return fmt.Errorf("integrity check failed for %q: got hash %s, want %s", obj.FilePath, computedHash, obj.Hash) - } - } - - if err := db.DB(ctx).Model(&model.Upload{}). - Where("storage_driver = ? AND file_path = ?", sourceDriver, obj.FilePath). - Updates(map[string]any{ - "storage_driver": targetDriver, - "file_path": targetResult.Key, - }).Error; err != nil { - return fmt.Errorf("update migrated object %q: %w", obj.FilePath, err) + if err := migrateSingleObject(ctx, sourceBackend, targetBackend, sourceDriver, targetDriver, obj, sha256HexLength); err != nil { + return err } atomic.AddInt64(&migrated, 1) return nil @@ -331,11 +325,122 @@ func migrateObjects( return atomic.LoadInt64(&migrated), err } - task.AppendLog(ctx, "迁移进度: %d/%d", atomic.LoadInt64(&migrated), total) + task.AppendLog(ctx, "当前批次迁移完成。迁移进度: %d/%d", atomic.LoadInt64(&migrated), total) } return atomic.LoadInt64(&migrated), nil } +func migrateSingleObject( + ctx context.Context, + sourceBackend storage.Backend, + targetBackend storage.Backend, + sourceDriver storage.Driver, + targetDriver storage.Driver, + obj struct { + FilePath string `gorm:"column:file_path"` + FileSize int64 `gorm:"column:file_size"` + MimeType string `gorm:"column:mime_type"` + Hash string `gorm:"column:hash"` + }, + sha256HexLength int, +) error { + // Check if the file already exists in target storage and has matching size + if shouldSkipMigration(ctx, targetBackend, obj) { + task.AppendLog(ctx, "[跳过迁移] 目标存储已存在相同文件且校验一致: %s", obj.FilePath) + if err := db.DB(ctx).Model(&model.Upload{}). + Where("storage_driver = ? AND file_path = ?", sourceDriver, obj.FilePath). + Updates(map[string]any{ + colStorageDriver: targetDriver, + }).Error; err != nil { + return fmt.Errorf("update migrated object %q: %w", obj.FilePath, err) + } + return nil + } + + task.AppendLog(ctx, "[迁移开始] 正在从源存储读取文件: %s", obj.FilePath) + source, err := sourceBackend.Get(ctx, obj.FilePath) + if err != nil { + if isNotFoundError(err) { + task.AppendLog(ctx, "警告: 源存储中物理文件不存在，标记为已删除并跳过: %s (错误: %v)", obj.FilePath, err) + if updateErr := db.DB(ctx).Model(&model.Upload{}). + Where("storage_driver = ? AND file_path = ?", sourceDriver, obj.FilePath). + Updates(map[string]any{ + "status": model.UploadStatusDeleted, + colStorageDriver: targetDriver, + }).Error; updateErr != nil { + return fmt.Errorf("update missing object %q: %w", obj.FilePath, updateErr) + } + return nil + } + return fmt.Errorf("open source object %q: %w", obj.FilePath, err) + } + task.AppendLog(ctx, "[传输中] 正在向目标存储上传文件: %s (大小: %d 字节, 类型: %s)", obj.FilePath, obj.FileSize, obj.MimeType) + targetResult, putErr := targetBackend.Put(ctx, obj.FilePath, source.Body, obj.FileSize, obj.MimeType) + closeErr := source.Body.Close() + if putErr != nil { + return fmt.Errorf("copy object %q: %w", obj.FilePath, putErr) + } + if closeErr != nil { + return fmt.Errorf("close source object %q: %w", obj.FilePath, closeErr) + } + + // Data integrity check (SHA-256 hash verification) + if len(obj.Hash) == sha256HexLength { + task.AppendLog(ctx, "[校验中] 正在对目标文件进行数据一致性校验 (SHA-256): %s", targetResult.Key) + targetObj, getErr := targetBackend.Get(ctx, targetResult.Key) + if getErr != nil { + return fmt.Errorf("retrieve target object for verification %q: %w", obj.FilePath, getErr) + } + if targetObj == nil || targetObj.Body == nil { + return fmt.Errorf("retrieve target object for verification %q: object or body is nil", obj.FilePath) + } + h := sha256.New() + if _, copyErr := io.Copy(h, targetObj.Body); copyErr != nil { + _ = targetObj.Body.Close() + return fmt.Errorf("read target object for verification %q: %w", obj.FilePath, copyErr) + } + _ = targetObj.Body.Close() + computedHash := hex.EncodeToString(h.Sum(nil)) + if computedHash != obj.Hash { + return fmt.Errorf("integrity check failed for %q: got hash %s, want %s", obj.FilePath, computedHash, obj.Hash) + } + task.AppendLog(ctx, "[校验通过] 文件一致性校验成功: %s", targetResult.Key) + } + + task.AppendLog(ctx, "[更新数据库] 正在更新文件 %s 的存储路径与驱动信息为: %s (%s)", obj.FilePath, targetResult.Key, targetDriver) + if err := db.DB(ctx).Model(&model.Upload{}). + Where("storage_driver = ? AND file_path = ?", sourceDriver, obj.FilePath). + Updates(map[string]any{ + colStorageDriver: targetDriver, + "file_path": targetResult.Key, + }).Error; err != nil { + return fmt.Errorf("update migrated object %q: %w", obj.FilePath, err) + } + task.AppendLog(ctx, "[迁移成功] 文件已完成迁移: %s -> %s", obj.FilePath, targetResult.Key) + return nil +} + +func shouldSkipMigration( + ctx context.Context, + targetBackend storage.Backend, + obj struct { + FilePath string `gorm:"column:file_path"` + FileSize int64 `gorm:"column:file_size"` + MimeType string `gorm:"column:mime_type"` + Hash string `gorm:"column:hash"` + }, +) bool { + targetObj, err := targetBackend.Get(ctx, obj.FilePath) + if err != nil || targetObj == nil || targetObj.Body == nil { + return false + } + defer func() { + _ = targetObj.Body.Close() + }() + + return targetObj.ContentLength == obj.FileSize +} + func isNotFoundError(err error) bool { if err == nil { return false diff --git a/internal/task/executor.go b/internal/task/executor.go index 21bff30b..60fb5308 100644 --- a/internal/task/executor.go +++ b/internal/task/executor.go @@ -174,13 +174,19 @@ func RetryTask(ctx context.Context, id uint64) (string, error) { return "", fmt.Errorf(errCreateRetryExecutionFailed, err) } + meta := GetTaskMeta(execution.TaskType) + queueName := QueueDefault + if meta != nil { + queueName = meta.Queue + } + // 入队 Asynq taskInfo := asynq.NewTask(execution.TaskType, []byte(execution.Payload)) if _, err := AsynqClient.Enqueue( taskInfo, asynq.TaskID(newTaskID), asynq.MaxRetry(execution.MaxRetry), - asynq.Queue(PrefixedQueue(QueueDefault)), + asynq.Queue(queueName), ); err != nil { newExecution.Status = model.TaskExecutionStatusFailed newExecution.ErrorMessage = fmt.Sprintf("重试入队失败: %v", err)