package fwhQueue import ( "encoding/json" "fmt" "fwh-toolbox-server/domain" "time" "github.com/hibiken/asynq" "go.uber.org/zap" ) // ==================== 对外暴露的入队函数 ==================== // EnqueueBatchDownloadTask 入队批量数据下载任务 func EnqueueBatchDownloadTask(payload domain.TaskBatchDownload) error { data, err := json.Marshal(payload) if err != nil { return fmt.Errorf("marshal BatchDownload payload: %w", err) } taskId := newTaskID(TaskTypeBatchDownload, payload.Id.Hex()) task := asynq.NewTask(TaskTypeBatchDownload, data, asynq.MaxRetry(0), // 主任务不重试(内部自行管理状态) asynq.Timeout(120*time.Minute), // 2 小时超时 asynq.Queue(QueueMain), // 主队列 asynq.TaskID(taskId), // 固定 ID 防止重复提交 asynq.Retention(7*24*time.Hour), // 完成后保留 7 天 ) info, err := TaskClient.Enqueue(task) if err != nil { zap.L().Error("[Enqueue] BatchDownload 入队失败", zap.String("id", payload.Id.Hex()), zap.String("err", err.Error()), ) return err } zap.L().Info("[Enqueue] BatchDownload 入队成功", zap.String("id", payload.Id.Hex()), zap.String("taskId", info.ID), zap.String("queue", info.Queue), ) return nil } // EnqueueSiteAnalysisTask 入队站点分析任务 func EnqueueSiteAnalysisTask(payload domain.TaskSiteAnalysis) error { data, err := json.Marshal(payload) if err != nil { return fmt.Errorf("marshal SiteAnalysis payload: %w", err) } taskId := newTaskID(TaskTypeSiteAnalysis, payload.Id.Hex()) task := asynq.NewTask(TaskTypeSiteAnalysis, data, asynq.MaxRetry(0), // 不重试 asynq.Timeout(60*time.Minute), // 1 小时超时 asynq.Queue(QueueMain), asynq.TaskID(taskId), asynq.Retention(7*24*time.Hour), ) info, err := TaskClient.Enqueue(task) if err != nil { zap.L().Error("[Enqueue] SiteAnalysis 入队失败", zap.String("id", payload.Id.Hex()), zap.String("err", err.Error()), ) return err } zap.L().Info("[Enqueue] SiteAnalysis 入队成功", zap.String("id", payload.Id.Hex()), zap.String("taskId", info.ID), ) return nil } // EnqueueShutdownStatisticsTask 入队 Shutdown 统计任务 func EnqueueShutdownStatisticsTask(payload domain.TaskShutdownStatistics) error { data, err := json.Marshal(payload) if err != nil { return fmt.Errorf("marshal ShutdownStatistics payload: %w", err) } taskId := newTaskID(TaskTypeShutdownStats, payload.Date) task := asynq.NewTask(TaskTypeShutdownStats, data, asynq.MaxRetry(0), asynq.Timeout(30*time.Minute), asynq.Queue(QueueMain), asynq.TaskID(taskId), asynq.Unique(10*time.Minute), // 10 分钟内去重 asynq.Retention(7*24*time.Hour), ) info, err := TaskClient.Enqueue(task) if err != nil { zap.L().Error("[Enqueue] ShutdownStatistics 入队失败", zap.String("date", payload.Date), zap.String("err", err.Error()), ) return err } zap.L().Info("[Enqueue] ShutdownStatistics 入队成功", zap.String("date", payload.Date), zap.String("taskId", info.ID), ) return nil } // EnqueueSiteInfoUpdateTask 入队站点信息更新任务(唯一任务) func EnqueueSiteInfoUpdateTask(payload domain.TaskSiteInfoUpdate) error { data, err := json.Marshal(payload) if err != nil { return fmt.Errorf("marshal SiteInfoUpdate payload: %w", err) } taskId := newTaskID(TaskTypeSiteInfoUpdate, payload.Id.Hex()) task := asynq.NewTask(TaskTypeSiteInfoUpdate, data, asynq.MaxRetry(0), asynq.Timeout(60*time.Minute), asynq.Queue(QueueMain), asynq.TaskID(taskId), asynq.Unique(30*time.Minute), // 30 分钟内去重(保证唯一性) asynq.Retention(7*24*time.Hour), ) info, err := TaskClient.Enqueue(task) if err != nil { zap.L().Error("[Enqueue] SiteInfoUpdate 入队失败", zap.String("id", payload.Id.Hex()), zap.String("err", err.Error()), ) return err } zap.L().Info("[Enqueue] SiteInfoUpdate 入队成功", zap.String("id", payload.Id.Hex()), zap.String("taskId", info.ID), ) return nil }