junhong_cmp_fiber/internal/task/polling_carddata_handler.go

package task

import (
	"context"
	"strings"
	"time"

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

	cardapp "github.com/break/junhong_cmp_fiber/internal/application/cardobservation"
	carddomain "github.com/break/junhong_cmp_fiber/internal/domain/cardobservation"
	"github.com/break/junhong_cmp_fiber/internal/gateway"
	"github.com/break/junhong_cmp_fiber/internal/infrastructure/integrationlog"
	"github.com/break/junhong_cmp_fiber/internal/store/postgres"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
)

// PollingCarddataHandler 负责流量轮询编排，查询结果统一交给卡观测应用服务。
type PollingCarddataHandler struct {
	base        *PollingBase
	gateway     *gateway.Client
	carrier     *postgres.CarrierStore
	observation *cardapp.Service
	integration *integrationlog.Repository
}

// NewPollingCarddataHandler 创建流量检查任务处理器。
func NewPollingCarddataHandler(base *PollingBase, gw *gateway.Client, carrier *postgres.CarrierStore, observation *cardapp.Service, integration *integrationlog.Repository) *PollingCarddataHandler {
	return &PollingCarddataHandler{base: base, gateway: gw, carrier: carrier, observation: observation, integration: integration}
}

// Handle 处理流量检查任务（旧分片调度）。按运营商账户调度时直接结束。
func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) error {
	cardID, ok := parseTaskPayload(task.Payload(), h.base.logger)
	if !ok {
		return nil
	}
	if h.base.carrierSchedulerActive(ctx) {
		return nil
	}
	if !h.base.acquireConcurrency(ctx, constants.TaskTypePollingCarddata) {
		h.base.logger.Debug("并发已满，重新入队", zap.Uint("card_id", cardID))
		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata)
	}
	defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingCarddata)
	return h.base.applyLegacyOutcome(ctx, cardID, constants.TaskTypePollingCarddata, h.Step(ctx, cardID))
}

// Step 执行一次流量检查步骤，返回结果分类；不负责重入队。旧分片调度与按运营商账户调度共用。
func (h *PollingCarddataHandler) Step(ctx context.Context, cardID uint) StepOutcome {
	startedAt := time.Now()
	lockToken, locked, err := h.base.acquireCardTrafficSyncLock(ctx, cardID)
	if err != nil || !locked {
		h.base.logger.Warn("卡流量同步锁不可用，延迟重入队", zap.Uint("card_id", cardID), zap.Error(err))
		return stepDefer(5 * time.Second)
	}
	defer h.base.releaseCardTrafficSyncLock(ctx, cardID, lockToken)

	// 优先轮询互斥接缝：放在流量同步锁获取成功之后，避免认领后拿不到锁导致优先项空占。
	priority, skip, priorityErr := h.base.beginPollingPriority(ctx, cardID, constants.TaskTypePollingCarddata)
	if priorityErr != nil {
		return h.fail(ctx, cardID, startedAt, "查询活动优先轮询项失败", priorityErr)
	}
	if skip {
		h.base.logger.Debug("卡轮询优先项已被其它执行持有，本轮跳过并延后",
			zap.Uint("card_id", cardID), zap.String("task_type", constants.TaskTypePollingCarddata))
		return stepDefer(0)
	}

	h.base.invalidateCardCache(ctx, cardID)
	card, err := h.base.iotCardStore.GetByID(ctx, cardID)
	if err != nil {
		if isNotFound(err) {
			priority.FailFinal(ctx, "卡不存在")
			return StepOutcome{Result: StepCardGone}
		}
		// 本地读卡失败属可恢复失败：先按真实执行累加尝试次数，再按上限收敛，避免加急事实被静默丢弃。
		priority.MarkExecution(ctx)
		priority.FailRetryable(ctx, "获取卡信息失败")
		return h.fail(ctx, cardID, startedAt, "获取卡信息失败", err)
	}
	// 执行前轮询范围校验：优先通道不得重启已停止轮询的卡。
	if !h.base.priorityInPollingScope(ctx, priority, card) {
		return stepSkip()
	}
	if h.gateway == nil {
		priority.FailFinal(ctx, "流量查询能力未配置")
		return stepSkip()
	}
	priority.MarkExecution(ctx)
	attempt := newGatewayAttempt(cardID, constants.IntegrationOperationGatewayTraffic, constants.CardObservationSceneTrafficPolling)
	result, err := h.gateway.QueryFlow(gateway.WithCarrier(ctx, card.CarrierID), &gateway.FlowQueryReq{CardNo: card.ICCID})
	if gateway.IsRateLimited(err) {
		priority.FailRetryable(ctx, "运营商账户网关限流，稍后重试")
		return StepOutcome{Result: StepRateLimited}
	}
	if err != nil {
		_ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultFailed, false, "查询流量失败", priority)
		priority.FailRetryable(ctx, "查询流量失败")
		return h.fail(ctx, cardID, startedAt, "查询流量失败", err)
	}
	if strings.TrimSpace(result.ICCID) == "" {
		_ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultInvalidPayload, false, "流量查询响应缺少 ICCID", priority)
		priority.FailRetryable(ctx, "流量查询响应缺少 ICCID")
		return h.fail(ctx, cardID, startedAt, "流量查询响应缺少 ICCID", nil)
	}
	ctx = withPollingWorkerAuditContext(ctx, constants.TaskTypePollingCarddata, "卡流量轮询任务", attempt.IntegrationID)
	if h.observation == nil || h.carrier == nil {
		priority.FailFinal(ctx, "卡流量观测能力未配置")
		return h.fail(ctx, cardID, startedAt, "卡流量观测能力未配置", nil)
	}
	now := time.Now()
	decision, err := h.observation.ApplyTrafficObservation(ctx, carddomain.TrafficObservation{
		CardID: cardID, GatewayReadingMB: float64(result.Used), ResetDay: h.carrier.GetDataResetDay(ctx, card.CarrierID),
		Metadata: carddomain.ObservationMetadata{
			ObservationID: attempt.IntegrationID, Source: constants.CardObservationSourcePolling,
			Scene: constants.CardObservationSceneTrafficPolling, ObservedAt: now, CorrelationID: attempt.IntegrationID,
		},
	})
	if err != nil {
		_ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultUnknown, false, "应用流量观测失败", priority)
		priority.FailRetryable(ctx, "应用流量观测失败")
		return h.fail(ctx, cardID, startedAt, "应用流量观测失败", err)
	}
	h.base.configMgr.RecordPreResetCheck(ctx, card, now)
	changed := decision.IncrementMB > 0 || decision.CrossMonth
	if changed {
		_ = recordPriorityGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultSuccess, true, "流量业务事实发生变化", priority)
	}
	if h.base.verboseLog {
		h.base.logger.Info("流量轮询详情", zap.Uint("card_id", cardID), zap.String("iccid", card.ICCID),
			zap.Float64("gateway_flow_mb", float64(result.Used)), zap.Float64("increment_mb", decision.IncrementMB),
			zap.Bool("is_cross_month", decision.CrossMonth), zap.Bool("reading_accepted", decision.ReadingAccepted))
	}
	metric := "polling.observation.unchanged"
	if changed {
		metric = "polling.observation.persisted"
	}
	h.base.logger.Info("流量轮询观测完成", zap.Uint("card_id", cardID), zap.Bool("changed", changed), zap.String("metric", metric))
	priority.Complete(ctx)
	h.base.updateStats(ctx, constants.TaskTypePollingCarddata, true, time.Since(startedAt))
	return StepOutcome{Result: StepSucceeded, TriggeredEvaluation: changed}
}

func (h *PollingCarddataHandler) fail(ctx context.Context, cardID uint, startedAt time.Time, message string, err error) StepOutcome {
	fields := []zap.Field{zap.Uint("card_id", cardID)}
	if err != nil {
		fields = append(fields, zap.Error(err))
	}
	h.base.logger.Warn(message, fields...)
	h.base.updateStats(ctx, constants.TaskTypePollingCarddata, false, time.Since(startedAt))
	return h.base.stepFail(ctx, constants.GatewayOpQueryFlow, message, err)
}