FranklinWH-Toolbox/service/task/queue/fwhQueue/fwhEnqueue.go

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
}