mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-03 15:06:36 +08:00
refactor(api): unify error handling and split upload/oauth god modules
Replace c.JSON(200, response.Err) and middleware gin.H bypasses with response.Abort* helpers so errors flow through Gin Error chain and ErrorHandlerMiddleware for OTel trace correlation. Split oauth/sources.go into domain-focused files and decompose upload into handler/filesrv/stats/task/cache/storage/util subpackages with a root facade preserving existing import paths.
This commit is contained in:
ryan
@@ -0,0 +1,148 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
// Package task provides upload-related async background task handlers.
|
||||
package task
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/apps/upload/shared"
|
||||
uploadstats "github.com/Rain-kl/Wavelet/internal/apps/upload/stats"
|
||||
uploadstorage "github.com/Rain-kl/Wavelet/internal/apps/upload/storage"
|
||||
"github.com/Rain-kl/Wavelet/internal/db"
|
||||
"github.com/Rain-kl/Wavelet/internal/model"
|
||||
"github.com/Rain-kl/Wavelet/internal/storage"
|
||||
"github.com/Rain-kl/Wavelet/internal/task"
|
||||
"github.com/Rain-kl/Wavelet/pkg/logger"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const (
|
||||
// SystemCleanupTask 系统定期垃圾清理任务标识
|
||||
SystemCleanupTask = "system:cleanup"
|
||||
// TaskTypeSystemCleanup 系统定期垃圾清理管理类型
|
||||
TaskTypeSystemCleanup = "system_cleanup"
|
||||
)
|
||||
|
||||
// SystemCleanupMeta represents the task metadata.
|
||||
var SystemCleanupMeta = task.TaskMeta{
|
||||
Type: TaskTypeSystemCleanup,
|
||||
AsynqTask: SystemCleanupTask,
|
||||
Name: "系统垃圾清理",
|
||||
Description: "定期清理未使用上传文件、历史推送记录和过期任务执行日志",
|
||||
SupportsTime: false,
|
||||
MaxRetry: task.DefaultMaxRetry,
|
||||
Queue: task.QueueDefault,
|
||||
Retryable: true,
|
||||
}
|
||||
|
||||
// SystemCleanupHandler 系统定期垃圾清理异步任务处理器
|
||||
type SystemCleanupHandler struct{}
|
||||
|
||||
// Execute 执行系统清理（包含文件清理、历史推送日志和任务执行日志清理）
|
||||
func (h *SystemCleanupHandler) Execute(ctx context.Context, _ []byte) (*task.TaskResult, error) {
|
||||
if uploadstorage.ReadOnly(ctx) {
|
||||
return nil, errors.New(shared.ErrStorageReadOnly)
|
||||
}
|
||||
const batchSize = 100
|
||||
var lastID uint64
|
||||
var totalProcessed int
|
||||
var totalDeleted int
|
||||
|
||||
oneHourAgo := time.Now().Add(-1 * time.Hour)
|
||||
|
||||
task.AppendLog(ctx, "开始扫描未使用上传文件，阈值: %s", oneHourAgo.Format(time.RFC3339))
|
||||
|
||||
for {
|
||||
var unusedUploads []model.Upload
|
||||
if err := db.DB(ctx).
|
||||
Where("id > ? AND status = ? AND created_at < ?", lastID, model.UploadStatusPending, oneHourAgo).
|
||||
Order("id ASC").
|
||||
Limit(batchSize).
|
||||
Find(&unusedUploads).Error; err != nil {
|
||||
task.AppendLog(ctx, "查询未使用的上传文件失败: %v", err)
|
||||
return nil, fmt.Errorf(shared.ErrQueryUnusedUploadsFailed, err)
|
||||
}
|
||||
|
||||
if len(unusedUploads) == 0 {
|
||||
break
|
||||
}
|
||||
|
||||
task.AppendLog(ctx, "本批次找到 %d 个需要清理的上传文件", len(unusedUploads))
|
||||
|
||||
for _, u := range unusedUploads {
|
||||
totalProcessed++
|
||||
|
||||
if err := db.DB(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Model(&model.Upload{}).
|
||||
Where("id = ? AND status = ?", u.ID, model.UploadStatusPending).
|
||||
Update("status", model.UploadStatusDeleted).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
driver := storage.Driver(u.StorageDriver)
|
||||
if driver == "" {
|
||||
driver = storage.DriverLocal
|
||||
}
|
||||
backend, err := storage.ForDriver(ctx, driver)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := backend.Delete(ctx, u.FilePath); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}); err != nil {
|
||||
task.AppendLog(ctx, "清理上传文件失败 [ID:%d]: %v", u.ID, err)
|
||||
lastID = u.ID
|
||||
continue
|
||||
}
|
||||
|
||||
uploadstats.RecordUploadStatsRemove(ctx, &u)
|
||||
totalDeleted++
|
||||
lastID = u.ID
|
||||
}
|
||||
}
|
||||
|
||||
task.AppendLog(ctx, "开始清理历史推送审计日志，只保留最近7天数据...")
|
||||
cutoff := time.Now().AddDate(0, 0, -7)
|
||||
var pushHistoryCount int64
|
||||
if err := db.DB(ctx).Model(&model.PushHistory{}).Where("created_at < ?", cutoff).Count(&pushHistoryCount).Error; err != nil {
|
||||
task.AppendLog(ctx, "统计待清理的历史推送记录失败: %v", err)
|
||||
} else if pushHistoryCount > 0 {
|
||||
if err := db.DB(ctx).Where("created_at < ?", cutoff).Delete(&model.PushHistory{}).Error; err != nil {
|
||||
task.AppendLog(ctx, "删除历史推送记录失败: %v", err)
|
||||
} else {
|
||||
task.AppendLog(ctx, "成功删除 %d 条历史推送记录 (截止时间: %s)", pushHistoryCount, cutoff.Format("2006-01-02 15:04:05"))
|
||||
}
|
||||
} else {
|
||||
task.AppendLog(ctx, "没有需要清理的历史推送记录 (截止时间: %s)", cutoff.Format("2006-01-02 15:04:05"))
|
||||
}
|
||||
|
||||
task.AppendLog(ctx, "开始清理任务执行日志：高频任务保留最近3天，低频任务保留最近30天...")
|
||||
taskLogStats, err := model.CleanupTaskExecutionLogs(ctx, time.Now())
|
||||
if err != nil {
|
||||
task.AppendLog(ctx, "清理任务执行日志失败: %v", err)
|
||||
logger.ErrorF(ctx, "清理任务执行日志失败: %v", err)
|
||||
} else {
|
||||
task.AppendLog(ctx, "成功清理任务执行日志 %d 条（高频 %d 条，低频 %d 条）",
|
||||
taskLogStats.HighFrequencyDeleted+taskLogStats.LowFrequencyDeleted,
|
||||
taskLogStats.HighFrequencyDeleted,
|
||||
taskLogStats.LowFrequencyDeleted,
|
||||
)
|
||||
}
|
||||
|
||||
msg := fmt.Sprintf("系统清理完成。成功清理未使用的上传文件 %d/%d 个；清理历史推送审计日志 %d 条；清理任务执行日志 %d 条。",
|
||||
totalDeleted,
|
||||
totalProcessed,
|
||||
pushHistoryCount,
|
||||
taskLogStats.HighFrequencyDeleted+taskLogStats.LowFrequencyDeleted,
|
||||
)
|
||||
task.AppendLog(ctx, "%s", msg)
|
||||
return &task.TaskResult{Message: msg}, nil
|
||||
}
|
||||
@@ -0,0 +1,399 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package task
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
uploadstats "github.com/Rain-kl/Wavelet/internal/apps/upload/stats"
|
||||
uploadstorage "github.com/Rain-kl/Wavelet/internal/apps/upload/storage"
|
||||
"github.com/Rain-kl/Wavelet/internal/db"
|
||||
"github.com/Rain-kl/Wavelet/internal/model"
|
||||
"github.com/Rain-kl/Wavelet/internal/storage"
|
||||
"github.com/Rain-kl/Wavelet/internal/task"
|
||||
"golang.org/x/sync/errgroup"
|
||||
)
|
||||
|
||||
const (
|
||||
// StorageMigrationTask is the Asynq task name for storage migration.
|
||||
StorageMigrationTask = uploadstorage.StorageMigrationTask
|
||||
// TaskTypeStorageMigration is the task metadata type for storage migration.
|
||||
TaskTypeStorageMigration = "storage_migration"
|
||||
|
||||
colStorageDriver = "storage_driver"
|
||||
)
|
||||
|
||||
// StorageMigrationMeta describes the manually dispatchable migration task.
|
||||
var StorageMigrationMeta = task.TaskMeta{
|
||||
Type: TaskTypeStorageMigration,
|
||||
AsynqTask: StorageMigrationTask,
|
||||
Name: "迁移文件存储",
|
||||
Description: "将活动存储中的文件迁移到待切换的目标存储，迁移期间文件系统保持只读",
|
||||
SupportsTime: false,
|
||||
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.
|
||||
type MigrationHandler struct{}
|
||||
|
||||
// ValidatePayload rejects duplicate active migrations through the task framework.
|
||||
func (h *MigrationHandler) ValidatePayload(payload []byte) ([]byte, error) {
|
||||
normalized, _, err := uploadstorage.NormalizeMigrationPayload(context.Background(), payload)
|
||||
if err != nil {
|
||||
return payload, err
|
||||
}
|
||||
active, err := hasUnresolvedMigrationTask(context.Background())
|
||||
if err != nil {
|
||||
return payload, err
|
||||
}
|
||||
if active {
|
||||
return payload, fmt.Errorf("storage migration task is already unresolved")
|
||||
}
|
||||
return normalized, nil
|
||||
}
|
||||
|
||||
// Execute migrates all unique active-storage objects to the pending backend.
|
||||
func (h *MigrationHandler) Execute(ctx context.Context, payload []byte) (*task.TaskResult, error) {
|
||||
if db.Redis != nil {
|
||||
const (
|
||||
cleanupTimeout = 5 * time.Second
|
||||
renewalInterval = 10 * time.Minute
|
||||
)
|
||||
|
||||
lockKey := db.PrefixedKey("lock:storage:migrate")
|
||||
ok, err := db.Redis.SetNX(ctx, lockKey, "locked", time.Hour).Result()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("acquire migration lock: %w", err)
|
||||
}
|
||||
if !ok {
|
||||
return nil, errors.New("另一个存储迁移任务正在运行中")
|
||||
}
|
||||
|
||||
stopRenewal := make(chan struct{})
|
||||
//nolint:contextcheck
|
||||
defer func() {
|
||||
close(stopRenewal)
|
||||
cleanupCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
|
||||
defer cancel()
|
||||
_ = db.Redis.Del(cleanupCtx, lockKey)
|
||||
}()
|
||||
|
||||
//nolint:contextcheck,gosec
|
||||
go func() {
|
||||
ticker := time.NewTicker(renewalInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
renewCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
|
||||
_ = db.Redis.Expire(renewCtx, lockKey, time.Hour).Err()
|
||||
cancel()
|
||||
case <-stopRenewal:
|
||||
return
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
active, err := storage.LoadConfig(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("load active storage config: %w", err)
|
||||
}
|
||||
target, err := uploadstorage.ParseMigrationTargetConfig(ctx, payload)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if target.Driver == active.Driver {
|
||||
if err := storage.SaveActiveConfig(ctx, target); err != nil {
|
||||
return nil, fmt.Errorf("activate same-driver storage config: %w", err)
|
||||
}
|
||||
message := fmt.Sprintf("存储配置已更新，活动存储保持为 %s", target.Driver)
|
||||
task.AppendLog(ctx, "%s", message)
|
||||
return &task.TaskResult{Message: message}, nil
|
||||
}
|
||||
|
||||
total, err := countStorageObjects(ctx, active.Driver)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("count source objects: %w", err)
|
||||
}
|
||||
if total == 0 {
|
||||
if err := storage.SaveActiveConfig(ctx, target); err != nil {
|
||||
return nil, fmt.Errorf("activate empty storage config: %w", err)
|
||||
}
|
||||
message := fmt.Sprintf("当前存储没有需要迁移的对象，活动存储已切换为 %s", target.Driver)
|
||||
task.AppendLog(ctx, "%s", message)
|
||||
return &task.TaskResult{Message: message}, nil
|
||||
}
|
||||
|
||||
sourceBackend, err := storage.NewBackend(ctx, active, active.Driver)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create source storage: %w", err)
|
||||
}
|
||||
targetBackend, err := storage.NewBackend(ctx, target, target.Driver)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create target storage: %w", err)
|
||||
}
|
||||
|
||||
task.AppendLog(ctx, "开始存储迁移: %s -> %s，总对象数: %d", active.Driver, target.Driver, total)
|
||||
migrated, err := migrateObjects(ctx, sourceBackend, targetBackend, active.Driver, target.Driver, total)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := storage.SaveActiveConfig(ctx, target); err != nil {
|
||||
return nil, fmt.Errorf("activate target storage: %w", err)
|
||||
}
|
||||
message := fmt.Sprintf("存储迁移完成，共迁移 %d 个对象，活动存储已切换为 %s", migrated, target.Driver)
|
||||
task.AppendLog(ctx, "%s", message)
|
||||
return &task.TaskResult{Message: message}, nil
|
||||
}
|
||||
|
||||
func countStorageObjects(ctx context.Context, driver storage.Driver) (int64, error) {
|
||||
var count int64
|
||||
err := db.DB(ctx).Model(&model.Upload{}).
|
||||
Where("storage_driver = ? AND status != ?", driver, model.UploadStatusDeleted).
|
||||
Distinct("file_path").
|
||||
Count(&count).Error
|
||||
return count, err
|
||||
}
|
||||
|
||||
func hasUnresolvedMigrationTask(ctx context.Context) (bool, error) {
|
||||
execution, ok, err := uploadstorage.LatestMigrationExecution(ctx)
|
||||
if err != nil || !ok {
|
||||
return false, err
|
||||
}
|
||||
return execution.Status == model.TaskExecutionStatusPending || execution.Status == model.TaskExecutionStatusRunning, nil
|
||||
}
|
||||
|
||||
func migrateObjects(
|
||||
ctx context.Context,
|
||||
sourceBackend storage.Backend,
|
||||
targetBackend storage.Backend,
|
||||
sourceDriver storage.Driver,
|
||||
targetDriver storage.Driver,
|
||||
total int64,
|
||||
) (int64, error) {
|
||||
const batchSize = 50
|
||||
const migrationConcurrency = 10
|
||||
const sha256HexLength = 64
|
||||
var migrated int64
|
||||
for {
|
||||
if err := ctx.Err(); err != nil {
|
||||
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"`
|
||||
MimeType string `gorm:"column:mime_type"`
|
||||
Hash string `gorm:"column:hash"`
|
||||
}
|
||||
if err := db.DB(ctx).Model(&model.Upload{}).
|
||||
Select("file_path, MAX(file_size) AS file_size, MAX(mime_type) AS mime_type, MAX(hash) AS hash").
|
||||
Where("storage_driver = ? AND status != ?", sourceDriver, model.UploadStatusDeleted).
|
||||
Group("file_path").
|
||||
Order("file_path ASC").
|
||||
Limit(batchSize).
|
||||
Scan(&objects).Error; err != nil {
|
||||
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
|
||||
g.Go(func() error {
|
||||
if err := migrateSingleObject(ctx, sourceBackend, targetBackend, sourceDriver, targetDriver, obj, sha256HexLength); err != nil {
|
||||
return err
|
||||
}
|
||||
atomic.AddInt64(&migrated, 1)
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
if err := g.Wait(); err != nil {
|
||||
return atomic.LoadInt64(&migrated), err
|
||||
}
|
||||
|
||||
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 {
|
||||
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) {
|
||||
return markMissingMigrationObjectDeleted(ctx, sourceDriver, targetDriver, obj.FilePath, err)
|
||||
}
|
||||
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)
|
||||
}
|
||||
|
||||
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 markMissingMigrationObjectDeleted(
|
||||
ctx context.Context,
|
||||
sourceDriver storage.Driver,
|
||||
targetDriver storage.Driver,
|
||||
filePath string,
|
||||
sourceErr error,
|
||||
) error {
|
||||
task.AppendLog(ctx, "警告: 源存储中物理文件不存在，标记为已删除并跳过: %s (错误: %v)", filePath, sourceErr)
|
||||
|
||||
var affectedUploads []model.Upload
|
||||
if err := db.DB(ctx).
|
||||
Where("storage_driver = ? AND file_path = ? AND status != ?", sourceDriver, filePath, model.UploadStatusDeleted).
|
||||
Find(&affectedUploads).Error; err != nil {
|
||||
return fmt.Errorf("load missing object uploads %q: %w", filePath, err)
|
||||
}
|
||||
if err := db.DB(ctx).Model(&model.Upload{}).
|
||||
Where("storage_driver = ? AND file_path = ?", sourceDriver, filePath).
|
||||
Updates(map[string]any{
|
||||
"status": model.UploadStatusDeleted,
|
||||
colStorageDriver: targetDriver,
|
||||
}).Error; err != nil {
|
||||
return fmt.Errorf("update missing object %q: %w", filePath, err)
|
||||
}
|
||||
for i := range affectedUploads {
|
||||
uploadstats.RecordUploadStatsRemove(ctx, &affectedUploads[i])
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func isNotFoundError(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return true
|
||||
}
|
||||
errStr := strings.ToLower(err.Error())
|
||||
for _, sub := range []string{"not found", "nosuchkey", "nosuchbucket", "404", "does not exist"} {
|
||||
if strings.Contains(errStr, sub) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,291 @@
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package task
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/db"
|
||||
"github.com/Rain-kl/Wavelet/internal/model"
|
||||
"github.com/Rain-kl/Wavelet/internal/storage"
|
||||
"github.com/Rain-kl/Wavelet/internal/testhelper"
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
func TestMigrationHandlerExecute(t *testing.T) {
|
||||
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
|
||||
defer cleanup()
|
||||
|
||||
sourceRoot := t.TempDir()
|
||||
sourcePath := filepath.Join(sourceRoot, "uploads", "test.txt")
|
||||
if err := os.MkdirAll(filepath.Dir(sourcePath), 0755); err != nil {
|
||||
t.Fatalf("MkdirAll(%q) returned error: %v", sourcePath, err)
|
||||
}
|
||||
const content = "storage migration"
|
||||
if err := os.WriteFile(sourcePath, []byte(content), 0644); err != nil {
|
||||
t.Fatalf("WriteFile(%q) returned error: %v", sourcePath, err)
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
active := storage.DefaultConfig()
|
||||
active.Local.Root = sourceRoot
|
||||
if err := storage.SaveActiveConfig(ctx, active); err != nil {
|
||||
t.Fatalf("SaveActiveConfig() returned error: %v", err)
|
||||
}
|
||||
target := storage.DefaultConfig()
|
||||
target.Driver = storage.DriverS3
|
||||
target.S3 = storage.ObjectConfig{
|
||||
Region: "us-east-1",
|
||||
Bucket: "target",
|
||||
AccessKeyID: "key",
|
||||
SecretAccessKey: "secret",
|
||||
}
|
||||
payload, err := json.Marshal(struct {
|
||||
Target storage.Config `json:"target"`
|
||||
}{Target: target})
|
||||
if err != nil {
|
||||
t.Fatalf("Marshal(storageMigrationPayload) returned error: %v", err)
|
||||
}
|
||||
|
||||
upload := model.Upload{
|
||||
ID: 99101,
|
||||
UserID: 1,
|
||||
FileName: "test.txt",
|
||||
FilePath: "uploads/test.txt",
|
||||
FileSize: int64(len(content)),
|
||||
MimeType: "text/plain",
|
||||
Extension: "txt",
|
||||
Hash: "hash",
|
||||
StorageDriver: string(storage.DriverLocal),
|
||||
Type: "attachment",
|
||||
Status: model.UploadStatusUsed,
|
||||
}
|
||||
if err := dbConn.Create(&upload).Error; err != nil {
|
||||
t.Fatalf("Create(upload) returned error: %v", err)
|
||||
}
|
||||
|
||||
var copied bytes.Buffer
|
||||
restore := storage.MockStorage(
|
||||
func(_ context.Context, _ string, body io.Reader, _ int64, _ string) error {
|
||||
_, err := io.Copy(&copied, body)
|
||||
return err
|
||||
},
|
||||
func(context.Context, string) (*storage.Object, error) {
|
||||
return nil, nil
|
||||
},
|
||||
func(context.Context, string) error {
|
||||
return nil
|
||||
},
|
||||
)
|
||||
defer restore()
|
||||
|
||||
result, err := (&MigrationHandler{}).Execute(ctx, payload)
|
||||
if err != nil {
|
||||
t.Fatalf("Execute() returned error: %v", err)
|
||||
}
|
||||
if result == nil {
|
||||
t.Fatal("Execute() result = nil, want non-nil")
|
||||
}
|
||||
if copied.String() != content {
|
||||
t.Errorf("migrated content = %q, want %q", copied.String(), content)
|
||||
}
|
||||
|
||||
var migrated model.Upload
|
||||
if err := dbConn.First(&migrated, upload.ID).Error; err != nil {
|
||||
t.Fatalf("First(upload) returned error: %v", err)
|
||||
}
|
||||
if migrated.StorageDriver != string(storage.DriverS3) {
|
||||
t.Errorf("StorageDriver = %q, want %q", migrated.StorageDriver, storage.DriverS3)
|
||||
}
|
||||
current, err := storage.LoadConfig(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("LoadConfig() returned error: %v", err)
|
||||
}
|
||||
if current.Driver != storage.DriverS3 {
|
||||
t.Errorf("active driver = %q, want %q", current.Driver, storage.DriverS3)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMigrationHandlerExecuteWithHashValidation(t *testing.T) {
|
||||
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
|
||||
defer cleanup()
|
||||
|
||||
sourceRoot := t.TempDir()
|
||||
sourcePath := filepath.Join(sourceRoot, "uploads", "test-hash.txt")
|
||||
if err := os.MkdirAll(filepath.Dir(sourcePath), 0755); err != nil {
|
||||
t.Fatalf("MkdirAll(%q) returned error: %v", sourcePath, err)
|
||||
}
|
||||
const content = "storage migration integrity check content"
|
||||
if err := os.WriteFile(sourcePath, []byte(content), 0644); err != nil {
|
||||
t.Fatalf("WriteFile(%q) returned error: %v", sourcePath, err)
|
||||
}
|
||||
|
||||
// Calculate correct SHA-256 hash
|
||||
h := sha256.New()
|
||||
h.Write([]byte(content))
|
||||
correctHash := hex.EncodeToString(h.Sum(nil))
|
||||
|
||||
ctx := context.Background()
|
||||
active := storage.DefaultConfig()
|
||||
active.Local.Root = sourceRoot
|
||||
if err := storage.SaveActiveConfig(ctx, active); err != nil {
|
||||
t.Fatalf("SaveActiveConfig() returned error: %v", err)
|
||||
}
|
||||
|
||||
target := storage.DefaultConfig()
|
||||
target.Driver = storage.DriverS3
|
||||
target.S3 = storage.ObjectConfig{
|
||||
Region: "us-east-1",
|
||||
Bucket: "target",
|
||||
AccessKeyID: "key",
|
||||
SecretAccessKey: "secret",
|
||||
}
|
||||
payload, err := json.Marshal(struct {
|
||||
Target storage.Config `json:"target"`
|
||||
}{Target: target})
|
||||
if err != nil {
|
||||
t.Fatalf("Marshal(storageMigrationPayload) returned error: %v", err)
|
||||
}
|
||||
|
||||
// Case 1: Incorrect Hash (should fail validation)
|
||||
uploadIncorrect := model.Upload{
|
||||
ID: 99102,
|
||||
UserID: 1,
|
||||
FileName: "test-hash.txt",
|
||||
FilePath: "uploads/test-hash.txt",
|
||||
FileSize: int64(len(content)),
|
||||
MimeType: "text/plain",
|
||||
Extension: "txt",
|
||||
Hash: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", // Invalid hash
|
||||
StorageDriver: string(storage.DriverLocal),
|
||||
Type: "attachment",
|
||||
Status: model.UploadStatusUsed,
|
||||
}
|
||||
if err := dbConn.Create(&uploadIncorrect).Error; err != nil {
|
||||
t.Fatalf("Create(uploadIncorrect) returned error: %v", err)
|
||||
}
|
||||
|
||||
var copied bytes.Buffer
|
||||
restore := storage.MockStorage(
|
||||
func(_ context.Context, _ string, body io.Reader, _ int64, _ string) error {
|
||||
copied.Reset()
|
||||
_, err := io.Copy(&copied, body)
|
||||
return err
|
||||
},
|
||||
func(context.Context, string) (*storage.Object, error) {
|
||||
return &storage.Object{
|
||||
Body: io.NopCloser(bytes.NewBuffer(copied.Bytes())),
|
||||
ContentLength: int64(copied.Len()),
|
||||
ContentType: "text/plain",
|
||||
}, nil
|
||||
},
|
||||
func(context.Context, string) error {
|
||||
return nil
|
||||
},
|
||||
)
|
||||
defer restore()
|
||||
|
||||
// Running execution with incorrect hash should fail with integrity error
|
||||
_, err = (&MigrationHandler{}).Execute(ctx, payload)
|
||||
if err == nil {
|
||||
t.Fatal("Execute() succeeded with incorrect hash, want error")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "integrity check failed") {
|
||||
t.Errorf("expected integrity check failed error, got: %v", err)
|
||||
}
|
||||
|
||||
// Case 2: Correct Hash (should succeed)
|
||||
if err := dbConn.Model(&model.Upload{}).Where("id = ?", uploadIncorrect.ID).Update("hash", correctHash).Error; err != nil {
|
||||
t.Fatalf("Update hash to correct value returned error: %v", err)
|
||||
}
|
||||
|
||||
// Run execution with correct hash should succeed
|
||||
result, err := (&MigrationHandler{}).Execute(ctx, payload)
|
||||
if err != nil {
|
||||
t.Fatalf("Execute() with correct hash failed: %v", err)
|
||||
}
|
||||
if result == nil {
|
||||
t.Fatal("Execute() result = nil, want non-nil")
|
||||
}
|
||||
|
||||
var migrated model.Upload
|
||||
if err := dbConn.First(&migrated, uploadIncorrect.ID).Error; err != nil {
|
||||
t.Fatalf("First(upload) returned error: %v", err)
|
||||
}
|
||||
if migrated.StorageDriver != string(storage.DriverS3) {
|
||||
t.Errorf("StorageDriver = %q, want %q", migrated.StorageDriver, storage.DriverS3)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMigrationHandlerExecuteWithRedisLock(t *testing.T) {
|
||||
_, _, cleanup := testhelper.SetupTestEnvironment(t)
|
||||
defer cleanup()
|
||||
|
||||
mr, err := miniredis.Run()
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to run miniredis: %v", err)
|
||||
}
|
||||
defer mr.Close()
|
||||
|
||||
rdb := redis.NewClient(&redis.Options{
|
||||
Addr: mr.Addr(),
|
||||
})
|
||||
defer rdb.Close()
|
||||
|
||||
oldRedis := db.Redis
|
||||
db.Redis = rdb
|
||||
defer func() {
|
||||
db.Redis = oldRedis
|
||||
}()
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
// Acquire lock manually
|
||||
lockKey := db.PrefixedKey("lock:storage:migrate")
|
||||
if err := rdb.Set(ctx, lockKey, "locked", time.Hour).Err(); err != nil {
|
||||
t.Fatalf("Failed to set manual lock in Redis: %v", err)
|
||||
}
|
||||
|
||||
active := storage.DefaultConfig()
|
||||
if err := storage.SaveActiveConfig(ctx, active); err != nil {
|
||||
t.Fatalf("SaveActiveConfig() returned error: %v", err)
|
||||
}
|
||||
|
||||
payload, err := json.Marshal(struct {
|
||||
Target storage.Config `json:"target"`
|
||||
}{Target: active})
|
||||
if err != nil {
|
||||
t.Fatalf("Marshal payload failed: %v", err)
|
||||
}
|
||||
|
||||
// Execution should fail because lock is already acquired
|
||||
_, err = (&MigrationHandler{}).Execute(ctx, payload)
|
||||
if err == nil {
|
||||
t.Fatal("Execute() succeeded when lock was held, want error")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "另一个存储迁移任务正在运行中") {
|
||||
t.Errorf("expected lock warning, got: %v", err)
|
||||
}
|
||||
|
||||
// Release lock and run again, should succeed
|
||||
if err := rdb.Del(ctx, lockKey).Err(); err != nil {
|
||||
t.Fatalf("Failed to delete lock: %v", err)
|
||||
}
|
||||
|
||||
_, err = (&MigrationHandler{}).Execute(ctx, payload)
|
||||
if err != nil {
|
||||
t.Fatalf("Execute() failed after lock released: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,184 @@
|
||||
// Copyright 2025 linux.do
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package task
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/apps/upload/filesrv"
|
||||
"github.com/Rain-kl/Wavelet/internal/apps/upload/shared"
|
||||
"github.com/Rain-kl/Wavelet/internal/db"
|
||||
"github.com/Rain-kl/Wavelet/internal/model"
|
||||
"github.com/Rain-kl/Wavelet/internal/task"
|
||||
)
|
||||
|
||||
const (
|
||||
// WarmImageCacheTask 图片压缩缓存预热任务标识
|
||||
WarmImageCacheTask = "upload:warm_image_cache"
|
||||
// TaskTypeWarmImageCache 图片压缩缓存预热管理类型
|
||||
TaskTypeWarmImageCache = "warm_image_cache"
|
||||
)
|
||||
|
||||
var warmImageCacheMu sync.Mutex
|
||||
|
||||
// WarmImageCacheMeta represents the image cache warmup task metadata.
|
||||
var WarmImageCacheMeta = task.TaskMeta{
|
||||
Type: TaskTypeWarmImageCache,
|
||||
AsynqTask: WarmImageCacheTask,
|
||||
Name: "预热图片压缩缓存",
|
||||
Description: "串行将文件管理中的图片转换为指定质量的 WebP 并写入永久缓存",
|
||||
SupportsTime: false,
|
||||
MaxRetry: task.DefaultMaxRetry,
|
||||
Queue: task.QueueDefault,
|
||||
Retryable: true,
|
||||
Params: []task.TaskParam{
|
||||
{
|
||||
Name: "quality",
|
||||
Label: "图片质量",
|
||||
Type: "string",
|
||||
Required: true,
|
||||
Placeholder: "low / medium / high",
|
||||
Description: "WebP 压缩质量，仅支持 low、medium、high",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
// WarmImageCachePayload is the image cache warmup task payload.
|
||||
type WarmImageCachePayload struct {
|
||||
Quality string `json:"quality"`
|
||||
}
|
||||
|
||||
// WarmImageCacheHandler serially warms compressed image cache entries.
|
||||
type WarmImageCacheHandler struct{}
|
||||
|
||||
// ValidatePayload validates and normalizes image cache warmup parameters.
|
||||
func (h *WarmImageCacheHandler) ValidatePayload(payload []byte) ([]byte, error) {
|
||||
if len(payload) == 0 {
|
||||
return nil, errors.New(shared.ErrImageCacheWarmupPayloadRequired)
|
||||
}
|
||||
|
||||
var req WarmImageCachePayload
|
||||
if err := json.Unmarshal(payload, &req); err != nil {
|
||||
return nil, fmt.Errorf(shared.ErrInvalidImageCacheWarmupPayload, err)
|
||||
}
|
||||
|
||||
req.Quality = strings.ToLower(strings.TrimSpace(req.Quality))
|
||||
if req.Quality != shared.ImageQualityLow &&
|
||||
req.Quality != shared.ImageQualityMedium &&
|
||||
req.Quality != shared.ImageQualityHigh {
|
||||
return nil, errors.New(shared.ErrInvalidImageCacheWarmupQuality)
|
||||
}
|
||||
|
||||
return json.Marshal(req)
|
||||
}
|
||||
|
||||
// Execute serially converts all managed images to WebP cache entries.
|
||||
func (h *WarmImageCacheHandler) Execute(ctx context.Context, payload []byte) (*task.TaskResult, error) {
|
||||
normalizedPayload, err := h.ValidatePayload(payload)
|
||||
if err != nil {
|
||||
task.AppendLog(ctx, "图片缓存预热参数无效: %v", err)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
var req WarmImageCachePayload
|
||||
if err := json.Unmarshal(normalizedPayload, &req); err != nil {
|
||||
return nil, fmt.Errorf(shared.ErrParseImageCacheWarmupPayload, err)
|
||||
}
|
||||
|
||||
task.AppendLog(ctx, "等待获取图片缓存预热执行锁，质量: %s", req.Quality)
|
||||
warmImageCacheMu.Lock()
|
||||
defer warmImageCacheMu.Unlock()
|
||||
|
||||
const (
|
||||
batchSize = 50
|
||||
maxFailureLogs = 5
|
||||
)
|
||||
var lastID uint64
|
||||
var totalProcessed int
|
||||
var totalCached int
|
||||
var totalGenerated int
|
||||
var totalFailed int
|
||||
|
||||
task.AppendLog(ctx, "开始串行预热图片压缩缓存，质量: %s，每批: %d", req.Quality, batchSize)
|
||||
|
||||
for {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, fmt.Errorf("image cache warmup canceled: %w", err)
|
||||
}
|
||||
|
||||
var uploads []model.Upload
|
||||
if err := db.DB(ctx).
|
||||
Where("id > ? AND status != ? AND (LOWER(mime_type) LIKE ? OR LOWER(extension) IN ?)",
|
||||
lastID,
|
||||
model.UploadStatusDeleted,
|
||||
"image/%",
|
||||
[]string{"jpg", "jpeg", "png", "webp", "gif"},
|
||||
).
|
||||
Order("id ASC").
|
||||
Limit(batchSize).
|
||||
Find(&uploads).Error; err != nil {
|
||||
task.AppendLog(ctx, "查询图片上传记录失败: %v", err)
|
||||
return nil, fmt.Errorf(shared.ErrQueryImagesForCacheWarmup, err)
|
||||
}
|
||||
|
||||
if len(uploads) == 0 {
|
||||
break
|
||||
}
|
||||
|
||||
batchGenerated := 0
|
||||
batchCached := 0
|
||||
batchFailed := 0
|
||||
for i := range uploads {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, fmt.Errorf("image cache warmup canceled: %w", err)
|
||||
}
|
||||
|
||||
upload := &uploads[i]
|
||||
totalProcessed++
|
||||
lastID = upload.ID
|
||||
|
||||
_, cacheHit, err := filesrv.EnsureCompressedImageCache(ctx, upload, req.Quality)
|
||||
if err != nil {
|
||||
totalFailed++
|
||||
batchFailed++
|
||||
if totalFailed <= maxFailureLogs {
|
||||
task.AppendLog(ctx, "图片处理失败 [ID:%d]: %v", upload.ID, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if cacheHit {
|
||||
totalCached++
|
||||
batchCached++
|
||||
continue
|
||||
}
|
||||
totalGenerated++
|
||||
batchGenerated++
|
||||
}
|
||||
|
||||
task.AppendLog(
|
||||
ctx,
|
||||
"批次完成，末尾 ID: %d，生成: %d，命中: %d，失败: %d",
|
||||
lastID,
|
||||
batchGenerated,
|
||||
batchCached,
|
||||
batchFailed,
|
||||
)
|
||||
}
|
||||
|
||||
msg := fmt.Sprintf(
|
||||
"图片缓存预热完成，共处理 %d 张，生成 %d 张，命中 %d 张，失败 %d 张",
|
||||
totalProcessed,
|
||||
totalGenerated,
|
||||
totalCached,
|
||||
totalFailed,
|
||||
)
|
||||
task.AppendLog(ctx, "%s", msg)
|
||||
return &task.TaskResult{Message: msg}, nil
|
||||
}
|
||||
@@ -0,0 +1,386 @@
|
||||
// Copyright 2025 linux.do
|
||||
// Copyright 2026 Arctel.net
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
|
||||
package task
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"image"
|
||||
"image/color"
|
||||
"image/png"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/Rain-kl/Wavelet/internal/apps/upload/filesrv"
|
||||
"github.com/Rain-kl/Wavelet/internal/apps/upload/shared"
|
||||
"github.com/Rain-kl/Wavelet/internal/db"
|
||||
"github.com/Rain-kl/Wavelet/internal/diskcache"
|
||||
"github.com/Rain-kl/Wavelet/internal/model"
|
||||
"github.com/Rain-kl/Wavelet/internal/storage"
|
||||
"github.com/Rain-kl/Wavelet/internal/task"
|
||||
"github.com/Rain-kl/Wavelet/internal/testhelper"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestSystemCleanupHandler_Execute(t *testing.T) {
|
||||
_, _, cleanup := testhelper.SetupTestEnvironment(t)
|
||||
defer cleanup()
|
||||
|
||||
// Mock S3 存储（让 DeleteObject 总是成功）
|
||||
storageMock := storage.MockStorage(
|
||||
func(ctx context.Context, key string, body io.Reader, size int64, contentType string) error {
|
||||
return nil
|
||||
},
|
||||
func(ctx context.Context, key string) (*storage.Object, error) { return nil, nil },
|
||||
func(ctx context.Context, key string) error { return nil },
|
||||
)
|
||||
defer storageMock()
|
||||
|
||||
ctx := context.Background()
|
||||
err := db.DB(ctx).AutoMigrate(&model.PushHistory{})
|
||||
require.NoError(t, err)
|
||||
|
||||
// 准备测试数据：创建一些上传记录
|
||||
now := time.Now()
|
||||
twoHoursAgo := now.Add(-2 * time.Hour)
|
||||
|
||||
records := []*model.Upload{
|
||||
// 超过1小时且状态为 pending 的记录 —— 应被清理
|
||||
{
|
||||
UserID: 1001, FileName: "old_file_1.jpg", FilePath: "uploads/old_1.jpg",
|
||||
FileSize: 1024, MimeType: "image/jpeg", Extension: "jpg", Hash: "hash1",
|
||||
StorageDriver: "s3", Type: "attachment", Status: model.UploadStatusPending,
|
||||
CreatedAt: twoHoursAgo,
|
||||
},
|
||||
{
|
||||
UserID: 1001, FileName: "old_file_2.png", FilePath: "uploads/old_2.png",
|
||||
FileSize: 2048, MimeType: "image/png", Extension: "png", Hash: "hash2",
|
||||
StorageDriver: "s3", Type: "attachment", Status: model.UploadStatusPending,
|
||||
CreatedAt: twoHoursAgo,
|
||||
},
|
||||
// 状态为 used 的记录 —— 不应被清理
|
||||
{
|
||||
UserID: 1001, FileName: "used_file.jpg", FilePath: "uploads/used.jpg",
|
||||
FileSize: 512, MimeType: "image/jpeg", Extension: "jpg", Hash: "hash3",
|
||||
StorageDriver: "s3", Type: "attachment", Status: model.UploadStatusUsed,
|
||||
CreatedAt: twoHoursAgo,
|
||||
},
|
||||
// 不到1小时的 pending 记录 —— 不应被清理
|
||||
{
|
||||
UserID: 1001, FileName: "recent_file.jpg", FilePath: "uploads/recent.jpg",
|
||||
FileSize: 256, MimeType: "image/jpeg", Extension: "jpg", Hash: "hash4",
|
||||
StorageDriver: "s3", Type: "attachment", Status: model.UploadStatusPending,
|
||||
CreatedAt: now.Add(-10 * time.Minute),
|
||||
},
|
||||
}
|
||||
for _, r := range records {
|
||||
err := db.DB(ctx).Create(r).Error
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
// 准备推送历史测试数据：1个旧的（应删除），1个新的（应保留）
|
||||
oldPush := &model.PushHistory{
|
||||
EventKey: "admin_login",
|
||||
Channel: "email",
|
||||
Target: "admin@test.com",
|
||||
Title: "Old Login",
|
||||
Content: "Old Content",
|
||||
Level: "INFO",
|
||||
Status: "success",
|
||||
CreatedAt: now.AddDate(0, 0, -10),
|
||||
}
|
||||
newPush := &model.PushHistory{
|
||||
EventKey: "admin_login",
|
||||
Channel: "lark",
|
||||
Target: "http://webhook.com",
|
||||
Title: "New Login",
|
||||
Content: "New Content",
|
||||
Level: "INFO",
|
||||
Status: "success",
|
||||
CreatedAt: now,
|
||||
}
|
||||
err = db.DB(ctx).Create(oldPush).Error
|
||||
require.NoError(t, err)
|
||||
err = db.DB(ctx).Create(newPush).Error
|
||||
require.NoError(t, err)
|
||||
|
||||
oldTaskLog := &model.TaskExecution{
|
||||
TaskID: "old_low_frequency_task_log",
|
||||
TaskType: "low:frequency",
|
||||
TaskName: "低频任务",
|
||||
Status: model.TaskExecutionStatusSucceeded,
|
||||
CreatedAt: now.AddDate(0, 0, -31),
|
||||
UpdatedAt: now.AddDate(0, 0, -31),
|
||||
TriggeredBy: "system",
|
||||
}
|
||||
err = model.CreateTaskExecution(ctx, oldTaskLog)
|
||||
require.NoError(t, err)
|
||||
|
||||
// 执行 handler
|
||||
handler := &SystemCleanupHandler{}
|
||||
result, err := handler.Execute(ctx, nil)
|
||||
|
||||
// 验证结果
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, result)
|
||||
assert.Contains(t, result.Message, "系统清理完成。成功清理未使用的上传文件 2/2 个；清理历史推送审计日志 1 条；清理任务执行日志 1 条。")
|
||||
|
||||
// 验证数据库状态：pending 且超过1小时的应被标记为 deleted
|
||||
var pendingCount int64
|
||||
db.DB(ctx).Model(&model.Upload{}).Where("status = ?", model.UploadStatusPending).Count(&pendingCount)
|
||||
assert.Equal(t, int64(1), pendingCount, "应只剩1条 pending 记录（最近的文件）")
|
||||
|
||||
var deletedCount int64
|
||||
db.DB(ctx).Model(&model.Upload{}).Where("status = ?", model.UploadStatusDeleted).Count(&deletedCount)
|
||||
assert.Equal(t, int64(2), deletedCount, "应有2条被标记为 deleted")
|
||||
|
||||
var usedCount int64
|
||||
db.DB(ctx).Model(&model.Upload{}).Where("status = ?", model.UploadStatusUsed).Count(&usedCount)
|
||||
assert.Equal(t, int64(1), usedCount, "used 状态的文件不应受影响")
|
||||
|
||||
// 验证推送历史数据状态：10天前的应被删除，今天的应保留
|
||||
var pushCount int64
|
||||
db.DB(ctx).Model(&model.PushHistory{}).Count(&pushCount)
|
||||
assert.Equal(t, int64(1), pushCount, "应只剩1条推送历史记录")
|
||||
|
||||
var remainingPush model.PushHistory
|
||||
err = db.DB(ctx).First(&remainingPush).Error
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "New Login", remainingPush.Title)
|
||||
|
||||
var taskLogCount int64
|
||||
err = db.DB(ctx).Model(&model.TaskExecution{}).Where("task_id = ?", "old_low_frequency_task_log").Count(&taskLogCount).Error
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, int64(0), taskLogCount, "过期低频任务日志应被清理")
|
||||
}
|
||||
|
||||
func TestSystemCleanupHandler_ExecuteNoFiles(t *testing.T) {
|
||||
_, _, cleanup := testhelper.SetupTestEnvironment(t)
|
||||
defer cleanup()
|
||||
|
||||
// Mock S3 存储
|
||||
storageMock := storage.MockStorage(
|
||||
func(ctx context.Context, key string, body io.Reader, size int64, contentType string) error {
|
||||
return nil
|
||||
},
|
||||
func(ctx context.Context, key string) (*storage.Object, error) { return nil, nil },
|
||||
func(ctx context.Context, key string) error { return nil },
|
||||
)
|
||||
defer storageMock()
|
||||
|
||||
ctx := context.Background()
|
||||
err := db.DB(ctx).AutoMigrate(&model.PushHistory{})
|
||||
require.NoError(t, err)
|
||||
|
||||
// 没有任何上传记录
|
||||
handler := &SystemCleanupHandler{}
|
||||
result, err := handler.Execute(ctx, nil)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, result)
|
||||
assert.Contains(t, result.Message, "系统清理完成。成功清理未使用的上传文件 0/0 个；清理历史推送审计日志 0 条；清理任务执行日志 0 条。")
|
||||
}
|
||||
|
||||
func TestSystemCleanupHandler_ImplementsTaskHandler(t *testing.T) {
|
||||
// 编译期验证 SystemCleanupHandler 实现了 TaskHandler 接口
|
||||
var _ task.TaskHandler = (*SystemCleanupHandler)(nil)
|
||||
}
|
||||
|
||||
func TestWarmImageCacheHandlerValidatePayload(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
payload []byte
|
||||
wantQuality string
|
||||
wantErr bool
|
||||
}{
|
||||
{
|
||||
name: "normalizes quality",
|
||||
payload: []byte(`{"quality":" HIGH "}`),
|
||||
wantQuality: shared.ImageQualityHigh,
|
||||
},
|
||||
{
|
||||
name: "empty payload",
|
||||
wantErr: true,
|
||||
},
|
||||
{
|
||||
name: "invalid json",
|
||||
payload: []byte(`{`),
|
||||
wantErr: true,
|
||||
},
|
||||
{
|
||||
name: "origin is not a compressed quality",
|
||||
payload: []byte(`{"quality":"origin"}`),
|
||||
wantErr: true,
|
||||
},
|
||||
{
|
||||
name: "unsupported quality",
|
||||
payload: []byte(`{"quality":"maximum"}`),
|
||||
wantErr: true,
|
||||
},
|
||||
}
|
||||
|
||||
handler := &WarmImageCacheHandler{}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
gotPayload, err := handler.ValidatePayload(tt.payload)
|
||||
if gotErr := err != nil; gotErr != tt.wantErr {
|
||||
t.Fatalf("ValidatePayload(%s) error = %v, want error presence = %t", tt.payload, err, tt.wantErr)
|
||||
}
|
||||
if tt.wantErr {
|
||||
return
|
||||
}
|
||||
|
||||
var got WarmImageCachePayload
|
||||
if err := json.Unmarshal(gotPayload, &got); err != nil {
|
||||
t.Fatalf("json.Unmarshal(%s) returned error: %v", gotPayload, err)
|
||||
}
|
||||
if got.Quality != tt.wantQuality {
|
||||
t.Errorf("ValidatePayload(%s).Quality = %q, want %q", tt.payload, got.Quality, tt.wantQuality)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestWarmImageCacheHandlerExecute(t *testing.T) {
|
||||
dbConn, _, cleanup := testhelper.SetupTestEnvironment(t)
|
||||
defer cleanup()
|
||||
|
||||
cache := diskcache.GetGlobalCache()
|
||||
if err := cache.Clear(); err != nil {
|
||||
t.Fatalf("Clear() before test returned error: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
if err := cache.Clear(); err != nil {
|
||||
t.Errorf("Clear() after test returned error: %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
testDir := t.TempDir()
|
||||
ctx := context.Background()
|
||||
active := storage.DefaultConfig()
|
||||
active.Local.Root = testDir
|
||||
if err := storage.SaveActiveConfig(ctx, active); err != nil {
|
||||
t.Fatalf("SaveActiveConfig() returned error: %v", err)
|
||||
}
|
||||
|
||||
firstPath := filepath.Join(testDir, "first.png")
|
||||
secondPath := filepath.Join(testDir, "second.jpg")
|
||||
writeTaskTestPNG(t, firstPath, color.RGBA{R: 255, A: 255})
|
||||
writeTaskTestPNG(t, secondPath, color.RGBA{G: 255, A: 255})
|
||||
|
||||
records := []model.Upload{
|
||||
{
|
||||
ID: 4101,
|
||||
UserID: 1001,
|
||||
FileName: "first.png",
|
||||
FilePath: firstPath,
|
||||
MimeType: "image/png",
|
||||
Extension: "png",
|
||||
StorageDriver: shared.StorageDriverLocal,
|
||||
Status: model.UploadStatusUsed,
|
||||
},
|
||||
{
|
||||
ID: 4102,
|
||||
UserID: 1001,
|
||||
FileName: "second.jpg",
|
||||
FilePath: secondPath,
|
||||
MimeType: "application/octet-stream",
|
||||
Extension: "jpg",
|
||||
StorageDriver: shared.StorageDriverLocal,
|
||||
Status: model.UploadStatusPending,
|
||||
},
|
||||
{
|
||||
ID: 4103,
|
||||
UserID: 1001,
|
||||
FileName: "notes.txt",
|
||||
FilePath: filepath.Join(testDir, "notes.txt"),
|
||||
MimeType: "text/plain",
|
||||
Extension: "txt",
|
||||
StorageDriver: shared.StorageDriverLocal,
|
||||
Status: model.UploadStatusUsed,
|
||||
},
|
||||
{
|
||||
ID: 4104,
|
||||
UserID: 1001,
|
||||
FileName: "deleted.png",
|
||||
FilePath: firstPath,
|
||||
MimeType: "image/png",
|
||||
Extension: "png",
|
||||
StorageDriver: shared.StorageDriverLocal,
|
||||
Status: model.UploadStatusDeleted,
|
||||
},
|
||||
}
|
||||
for i := range records {
|
||||
if info, err := os.Stat(records[i].FilePath); err == nil {
|
||||
records[i].FileSize = info.Size()
|
||||
}
|
||||
if err := dbConn.Create(&records[i]).Error; err != nil {
|
||||
t.Fatalf("failed to create upload %d: %v", records[i].ID, err)
|
||||
}
|
||||
}
|
||||
|
||||
handler := &WarmImageCacheHandler{}
|
||||
payload := []byte(`{"quality":"low"}`)
|
||||
|
||||
result, err := handler.Execute(context.Background(), payload)
|
||||
if err != nil {
|
||||
t.Fatalf("Execute(%s) returned error: %v", payload, err)
|
||||
}
|
||||
if result == nil {
|
||||
t.Fatal("Execute() result = nil, want non-nil")
|
||||
}
|
||||
if result.Message != "图片缓存预热完成，共处理 2 张，生成 2 张，命中 0 张，失败 0 张" {
|
||||
t.Errorf("Execute() message = %q, want generated summary", result.Message)
|
||||
}
|
||||
|
||||
for i := range records[:2] {
|
||||
key := filesrv.ImageCompressionCacheKey(&records[i], shared.ImageQualityLow)
|
||||
got, err := cache.Get(key)
|
||||
if err != nil {
|
||||
t.Errorf("cache.Get(%q) returned error: %v", key, err)
|
||||
continue
|
||||
}
|
||||
if len(got) == 0 {
|
||||
t.Errorf("cache.Get(%q) returned empty WebP data", key)
|
||||
}
|
||||
}
|
||||
|
||||
secondResult, err := handler.Execute(context.Background(), payload)
|
||||
if err != nil {
|
||||
t.Fatalf("second Execute(%s) returned error: %v", payload, err)
|
||||
}
|
||||
if secondResult.Message != "图片缓存预热完成，共处理 2 张，生成 0 张，命中 2 张，失败 0 张" {
|
||||
t.Errorf("second Execute() message = %q, want cache-hit summary", secondResult.Message)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWarmImageCacheHandlerImplementsTaskInterfaces(t *testing.T) {
|
||||
var _ task.TaskHandler = (*WarmImageCacheHandler)(nil)
|
||||
var _ task.PayloadValidator = (*WarmImageCacheHandler)(nil)
|
||||
}
|
||||
|
||||
func writeTaskTestPNG(t *testing.T, path string, fill color.RGBA) {
|
||||
t.Helper()
|
||||
|
||||
img := image.NewRGBA(image.Rect(0, 0, 2, 2))
|
||||
for y := 0; y < 2; y++ {
|
||||
for x := 0; x < 2; x++ {
|
||||
img.Set(x, y, fill)
|
||||
}
|
||||
}
|
||||
|
||||
var buf bytes.Buffer
|
||||
if err := png.Encode(&buf, img); err != nil {
|
||||
t.Fatalf("png.Encode() returned error: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(path, buf.Bytes(), 0o600); err != nil {
|
||||
t.Fatalf("os.WriteFile(%q) returned error: %v", path, err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user