FranklinWH-Toolbox/service/task/queue/faultRecordQueue/queue.go

// Package faultRecordQueue 故障录波 MQTT 操作的异步执行器（asynq）
package faultRecordQueue

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	mongoRepo "fwh-toolbox-server/repositories/mongo"
	task_core "fwh-toolbox-server/task/core"

	"github.com/hibiken/asynq"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.uber.org/zap"
)

const (
	TaskTypeRun = "fault-record:run"
	QueueName   = "fault-record"
)

var client *asynq.Client

type runPayload struct {
	OperationID string `json:"operationId"`
}

func Init(tc *task_core.TaskCore) {
	mux := asynq.NewServeMux()
	mux.HandleFunc(TaskTypeRun, handleRun)
	client = tc.CreateTask(task_core.TaskConfig{
		Name: "FaultRecord",
		SvrCfg: asynq.Config{
			Concurrency: 10, // 不同网关的操作可并行；同一网关由网关锁串行
			Queues:      map[string]int{QueueName: 1},
			ErrorHandler: asynq.ErrorHandlerFunc(func(_ context.Context, task *asynq.Task, err error) {
				zap.L().Error("fault-record task failed", zap.String("type", task.Type()), zap.String("err", err.Error()))
			}),
		},
		Mux: mux,
	})
}

// Enqueue 提交操作。TaskID 固定为操作 ID，重复提交（如启动恢复）不会重复执行。
// 业务失败由执行器写入操作状态，不依赖 asynq 重试；MaxRetry 只用于进程崩溃后重新投递。
func Enqueue(id primitive.ObjectID) error {
	if client == nil {
		return fmt.Errorf("fault record queue not initialized")
	}
	payload, _ := json.Marshal(runPayload{OperationID: id.Hex()})
	_, err := client.Enqueue(asynq.NewTask(TaskTypeRun, payload),
		asynq.Queue(QueueName),
		asynq.MaxRetry(3),
		asynq.Timeout(15*time.Minute),
		asynq.TaskID("fault-record-"+id.Hex()),
		asynq.Retention(24*time.Hour),
	)
	if errors.Is(err, asynq.ErrTaskIDConflict) {
		return nil
	}
	return err
}

// EnqueueDownloadRetry 提交手动重试下载；每次重试使用新的 TaskID（原任务的 TaskID 仍在保留期内）
func EnqueueDownloadRetry(id primitive.ObjectID, attempt int) error {
	if client == nil {
		return fmt.Errorf("fault record queue not initialized")
	}
	payload, _ := json.Marshal(runPayload{OperationID: id.Hex()})
	_, err := client.Enqueue(asynq.NewTask(TaskTypeRun, payload),
		asynq.Queue(QueueName),
		asynq.MaxRetry(3),
		asynq.Timeout(15*time.Minute),
		asynq.TaskID(fmt.Sprintf("fault-record-%s-retry-%d", id.Hex(), attempt)),
		asynq.Retention(24*time.Hour),
	)
	if errors.Is(err, asynq.ErrTaskIDConflict) {
		return nil
	}
	return err
}

func handleRun(ctx context.Context, t *asynq.Task) error {
	var p runPayload
	if err := json.Unmarshal(t.Payload(), &p); err != nil {
		return fmt.Errorf("%w: %v", asynq.SkipRetry, err)
	}
	id, err := primitive.ObjectIDFromHex(p.OperationID)
	if err != nil {
		return fmt.Errorf("%w: %v", asynq.SkipRetry, err)
	}
	return defaultDeps().run(ctx, id)
}

// RecoverActive 启动时把未结束的操作重新提交；仍在队列中的会因 TaskID 冲突被忽略
func RecoverActive(ctx context.Context) {
	list, err := mongoRepo.ListActiveFaultRecordOps(ctx)
	if err != nil {
		zap.L().Error("fault record recover: list active failed", zap.Error(err))
		return
	}
	for _, op := range list {
		enqueue := func() error { return Enqueue(op.ID) }
		if op.DownloadRetryCount > 0 {
			// 手动重试下载使用独立的 TaskID，按当前重试序号重新提交
			enqueue = func() error { return EnqueueDownloadRetry(op.ID, op.DownloadRetryCount) }
		}
		if err := enqueue(); err != nil {
			zap.L().Error("fault record recover: enqueue failed", zap.String("id", op.ID.Hex()), zap.Error(err))
		}
	}
	if len(list) > 0 {
		zap.L().Info("fault record recover: resubmitted active operations", zap.Int("count", len(list)))
	}
}