junhong_cmp_fiber/internal/task/polling_package_handler.go

package task

import (
	"context"
	"time"

	"github.com/hibiken/asynq"
	"go.uber.org/zap"

	priorityapp "github.com/break/junhong_cmp_fiber/internal/application/prioritypolling"
	iot_card_svc "github.com/break/junhong_cmp_fiber/internal/service/iot_card"
	"github.com/break/junhong_cmp_fiber/internal/store/postgres"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
)

// PollingPackageHandler 套餐检查任务处理器
// 职责：触发 EvaluateAndAct 检查套餐状态并判断停复机
// 不做：直接停复机决策，委托给 StopResumeService.EvaluateAndAct
// 注：套餐状态由 PackageActivationHandler/DataResetHandler/UsageService 维护，
// 本 Handler 无需查 Gateway，只负责周期性触发再评估。
type PollingPackageHandler struct {
	base          *PollingBase
	iotCardStore  *postgres.IotCardStore
	stopResumeSvc iot_card_svc.StopResumeServiceInterface
}

// NewPollingPackageHandler 创建套餐检查任务处理器
func NewPollingPackageHandler(
	base *PollingBase,
	iotCardStore *postgres.IotCardStore,
	stopResumeSvc iot_card_svc.StopResumeServiceInterface,
) *PollingPackageHandler {
	return &PollingPackageHandler{
		base:          base,
		iotCardStore:  iotCardStore,
		stopResumeSvc: stopResumeSvc,
	}
}

// Handle 处理套餐检查任务（旧分片调度）。按运营商账户调度时直接结束。
func (h *PollingPackageHandler) Handle(ctx context.Context, t *asynq.Task) error {
	cardID, ok := parseTaskPayload(t.Payload(), h.base.logger)
	if !ok {
		return nil
	}
	if h.base.carrierSchedulerActive(ctx) {
		return nil
	}
	ctx = withPollingWorkerAuditContext(ctx, constants.TaskTypePollingPackage, "套餐状态轮询任务", t.ResultWriter().TaskID())

	if !h.base.acquireConcurrency(ctx, constants.TaskTypePollingPackage) {
		h.base.logger.Debug("并发已满，重新入队", zap.Uint("card_id", cardID))
		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage)
	}
	defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingPackage)
	return h.base.applyLegacyOutcome(ctx, cardID, constants.TaskTypePollingPackage, h.Step(ctx, cardID))
}

// Step 执行一次套餐检查步骤（按规则重新判断停复机），返回结果分类；不负责重入队。
// 调用方负责在 ctx 中设置 Worker 审计上下文。旧分片调度与按运营商账户调度共用。
func (h *PollingPackageHandler) Step(ctx context.Context, cardID uint) StepOutcome {
	startTime := time.Now()
	// 优先轮询互斥接缝：存在活动优先项时本轮即优先执行或被跳过；无活动项时行为与既有实现一致。
	priority, skip, priorityErr := h.base.beginPollingPriority(ctx, cardID, constants.TaskTypePollingPackage)
	if priorityErr != nil {
		h.base.logger.Error("查询活动优先轮询项失败", zap.Uint("card_id", cardID), zap.Error(priorityErr))
		h.base.updateStats(ctx, constants.TaskTypePollingPackage, false, time.Since(startTime))
		return h.base.stepFail(ctx, "", "查询活动优先轮询项失败", priorityErr)
	}
	if skip {
		h.base.logger.Debug("卡轮询优先项已被其它执行持有，本轮跳过并延后",
			zap.Uint("card_id", cardID), zap.String("task_type", constants.TaskTypePollingPackage))
		return stepDefer(0)
	}

	card, cacheErr := h.base.getCardWithCache(ctx, cardID)
	if cacheErr != nil {
		if isNotFound(cacheErr) {
			priority.FailFinal(ctx, "卡不存在")
			return StepOutcome{Result: StepCardGone}
		}
		h.base.logger.Error("获取卡信息失败", zap.Uint("card_id", cardID), zap.Error(cacheErr))
		// 本地读卡失败属可恢复失败：与其余三个轮询 handler 统一按尝试上限收敛，避免加急事实被静默丢弃。
		priority.MarkExecution(ctx)
		priority.FailRetryable(ctx, "获取卡信息失败")
		h.base.updateStats(ctx, constants.TaskTypePollingPackage, false, time.Since(startTime))
		return h.base.stepFail(ctx, "", "获取卡信息失败", cacheErr)
	}
	// 执行前轮询范围校验：优先通道不得重启已停止轮询的卡。
	if !h.base.priorityInPollingScope(ctx, priority, card) {
		return stepSkip()
	}

	if h.stopResumeSvc == nil {
		priority.FailFinal(ctx, "套餐评估能力未配置")
		h.base.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime))
		return stepSkip()
	}

	// 停复机评估就是本任务的上游执行动作：先累加尝试次数再发起。
	priority.MarkExecution(ctx)
	freshCard, loadErr := h.iotCardStore.GetByID(ctx, cardID)
	if loadErr != nil {
		h.base.logger.Warn("套餐检查：加载卡信息失败",
			zap.Uint("card_id", cardID), zap.Error(loadErr))
		priority.FailRetryable(ctx, "获取卡信息失败")
		h.base.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime))
		return h.base.stepFail(ctx, "", "获取卡信息失败", loadErr)
	}
	h.base.logger.Info("套餐检查：执行停复机评估",
		zap.Uint("card_id", cardID),
		zap.Int("network_status", freshCard.NetworkStatus),
		zap.String("stop_reason", freshCard.StopReason))
	if h.base.verboseLog {
		h.base.logger.Info("套餐检查详情",
			zap.Uint("card_id", cardID),
			zap.String("iccid", freshCard.ICCID),
			zap.Int("network_status", freshCard.NetworkStatus),
			zap.String("stop_reason", freshCard.StopReason))
	}
	// 标记本次评估来自普通套餐轮询：无有效套餐场景据此才允许追加优先轮询请求，
	// 其余调用链（观测消费者、手动实名等）不打标即不写。
	evaluationCtx := priorityapp.WithPollingPriorityTrigger(ctx)
	if evalErr := h.stopResumeSvc.EvaluateAndAct(evaluationCtx, freshCard); evalErr != nil {
		h.base.logger.Warn("套餐检查后停复机评估失败",
			zap.Uint("card_id", cardID), zap.Error(evalErr))
		priority.FailRetryable(ctx, "停复机评估失败")
		h.base.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime))
		outcome := h.base.stepFail(ctx, constants.GatewayOpStart, "停复机评估失败", evalErr)
		return outcome
	}

	priority.Complete(ctx)
	h.base.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime))
	return stepOK()
}