暂存一下,防止丢失

internal/task/polling_carddata_handler.go

@@ -2,211 +2,110 @@ package task
 
 import (
 	"context"
+	"strings"
 	"time"
 
 	"github.com/hibiken/asynq"
 	"go.uber.org/zap"
-	"gorm.io/gorm"
 
+	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/model"
-	iot_card_svc "github.com/break/junhong_cmp_fiber/internal/service/iot_card"
-	packagepkg "github.com/break/junhong_cmp_fiber/internal/service/package"
+	"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 流量检查任务处理器
-// 职责：调 Gateway 查流量 → 写 DB → 扣减套餐流量 → 调 EvaluateAndAct 判断停复机
-// 不做：直接停复机决策，委托给 StopResumeService.EvaluateAndAct
+// PollingCarddataHandler 负责流量轮询编排，查询结果统一交给卡观测应用服务。
 type PollingCarddataHandler struct {
-	base          *PollingBase
-	gateway       *gateway.Client
-	iotCardStore  *postgres.IotCardStore
-	carrierStore  *postgres.CarrierStore
-	usageService  *packagepkg.UsageService
-	stopResumeSvc iot_card_svc.StopResumeServiceInterface
+	base        *PollingBase
+	gateway     *gateway.Client
+	carrier     *postgres.CarrierStore
+	observation *cardapp.Service
+	integration *integrationlog.Repository
 }
 
-// NewPollingCarddataHandler 创建流量检查任务处理器
-func NewPollingCarddataHandler(
-	base *PollingBase,
-	gw *gateway.Client,
-	iotCardStore *postgres.IotCardStore,
-	carrierStore *postgres.CarrierStore,
-	usageService *packagepkg.UsageService,
-	stopResumeSvc iot_card_svc.StopResumeServiceInterface,
-) *PollingCarddataHandler {
-	return &PollingCarddataHandler{
-		base:          base,
-		gateway:       gw,
-		iotCardStore:  iotCardStore,
-		carrierStore:  carrierStore,
-		usageService:  usageService,
-		stopResumeSvc: stopResumeSvc,
-	}
+// 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, t *asynq.Task) error {
-	startTime := time.Now()
-
-	cardID, ok := parseTaskPayload(t.Payload(), h.base.logger)
+// Handle 处理流量检查任务。
+func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) error {
+	startedAt := time.Now()
+	cardID, ok := parseTaskPayload(task.Payload(), h.base.logger)
 	if !ok {
 		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)
-
-	locked, lockErr := h.base.acquireCardTrafficSyncLock(ctx, cardID)
-	if lockErr != nil {
-		h.base.logger.Warn("获取卡流量同步锁失败，延迟重入队", zap.Uint("card_id", cardID), zap.Error(lockErr))
-		return h.base.queueMgr.Requeue(ctx, cardID, constants.TaskTypePollingCarddata, time.Now().Add(5*time.Second))
-	}
-	if !locked {
-		h.base.logger.Info("卡流量同步正在执行，延迟重入队", zap.Uint("card_id", cardID))
+	locked, err := h.base.acquireCardTrafficSyncLock(ctx, cardID)
+	if err != nil || !locked {
+		h.base.logger.Warn("卡流量同步锁不可用，延迟重入队", zap.Uint("card_id", cardID), zap.Error(err))
 		return h.base.queueMgr.Requeue(ctx, cardID, constants.TaskTypePollingCarddata, time.Now().Add(5*time.Second))
 	}
 	defer h.base.releaseCardTrafficSyncLock(ctx, cardID)
 
 	h.base.invalidateCardCache(ctx, cardID)
-	card, err := h.iotCardStore.GetByID(ctx, cardID)
+	card, err := h.base.iotCardStore.GetByID(ctx, cardID)
 	if err != nil {
 		if isNotFound(err) {
 			return nil
 		}
-		h.base.logger.Error("获取卡信息失败", zap.Uint("card_id", cardID), zap.Error(err))
-		h.base.updateStats(ctx, constants.TaskTypePollingCarddata, false, time.Since(startTime))
+		return h.failAndRequeue(ctx, cardID, startedAt, "获取卡信息失败", err)
+	}
+	if h.gateway == nil {
 		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata)
 	}
-
-	var gatewayFlowMB float64
-	if h.gateway != nil {
-		result, gwErr := h.gateway.QueryFlow(ctx, &gateway.FlowQueryReq{CardNo: card.ICCID})
-		if gwErr != nil {
-			h.base.logger.Warn("查询流量失败",
-				zap.Uint("card_id", cardID), zap.String("iccid", card.ICCID), zap.Error(gwErr))
-			h.base.updateStats(ctx, constants.TaskTypePollingCarddata, false, time.Since(startTime))
-			return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata)
-		}
-		gatewayFlowMB = float64(result.Used)
-	} else {
-		gatewayFlowMB = card.CurrentMonthUsageMB
+	attemptStartedAt := time.Now()
+	attempt, err := startGatewayAttempt(ctx, h.integration, cardID, constants.IntegrationOperationGatewayTraffic, constants.CardObservationSceneTrafficPolling)
+	if err != nil {
+		return h.failAndRequeue(ctx, cardID, startedAt, "建立 Gateway 流量 Integration Log 失败", err)
+	}
+	result, err := h.gateway.QueryFlow(ctx, &gateway.FlowQueryReq{CardNo: card.ICCID})
+	if err != nil {
+		_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
+		return h.failAndRequeue(ctx, cardID, startedAt, "查询流量失败", err)
+	}
+	if strings.TrimSpace(result.ICCID) == "" {
+		_ = completeGatewayAttempt(ctx, h.integration, attempt, false, attemptStartedAt)
+		return h.failAndRequeue(ctx, cardID, startedAt, "流量查询响应缺少 ICCID", nil)
+	}
+	if logErr := completeGatewayAttempt(ctx, h.integration, attempt, true, attemptStartedAt); logErr != nil {
+		return h.failAndRequeue(ctx, cardID, startedAt, "完成 Gateway 流量 Integration Log 失败", logErr)
+	}
+	if h.observation == nil || h.carrier == nil {
+		return h.failAndRequeue(ctx, cardID, startedAt, "卡流量观测能力未配置", nil)
 	}
-
 	now := time.Now()
-	resetDay := h.carrierStore.GetDataResetDay(ctx, card.CarrierID)
-	updates, flowIncrementMB, isCrossMonth := h.calculateFlowUpdates(card, gatewayFlowMB, now, resetDay)
-	updates["last_data_check_at"] = now
-	if h.gateway != nil {
-		updates["last_sync_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 {
+		return h.failAndRequeue(ctx, cardID, startedAt, "应用流量观测失败", err)
 	}
-
-	if h.base.verboseLog && h.gateway != nil {
-		h.base.logger.Info("流量轮询详情",
-			zap.Uint("card_id", cardID),
-			zap.String("iccid", card.ICCID),
-			zap.Float64("gateway_flow_mb", gatewayFlowMB),
-			zap.Float64("increment_mb", flowIncrementMB),
-			zap.Bool("is_cross_month", isCrossMonth))
+	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))
 	}
-
-	if updateErr := h.iotCardStore.UpdateFields(ctx, cardID, updates); updateErr != nil {
-		h.base.logger.Error("更新卡流量信息失败", zap.Uint("card_id", cardID), zap.Error(updateErr))
-		h.base.updateStats(ctx, constants.TaskTypePollingCarddata, false, time.Since(startTime))
-		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata)
-	}
-
-	go h.base.insertDataUsageRecord(context.Background(), cardID, flowIncrementMB, now)
-
-	if refreshedCard, loadErr := h.iotCardStore.GetByID(ctx, cardID); loadErr == nil {
-		writeCardToCache(h.base, constants.RedisPollingCardInfoKey(cardID), refreshedCard)
-	} else {
-		h.base.invalidateCardCache(ctx, cardID)
-		h.base.logger.Warn("刷新卡缓存失败", zap.Uint("card_id", cardID), zap.Error(loadErr))
-	}
-
-	if flowIncrementMB > 0 && h.usageService != nil {
-		if deductErr := h.usageService.DeductDataUsage(ctx, constants.AssetTypeIotCard, cardID, flowIncrementMB); deductErr != nil {
-			h.base.logger.Warn("套餐流量扣减失败",
-				zap.Uint("card_id", cardID), zap.Float64("increment_mb", flowIncrementMB), zap.Error(deductErr))
-		}
-	}
-
-	if h.stopResumeSvc != nil {
-		freshCard, loadErr := h.iotCardStore.GetByID(ctx, cardID)
-		if loadErr == nil {
-			if evalErr := h.stopResumeSvc.EvaluateAndAct(ctx, freshCard); evalErr != nil {
-				h.base.logger.Warn("流量检查后停复机评估失败",
-					zap.Uint("card_id", cardID), zap.Error(evalErr))
-			}
-		}
-	}
-
-	h.base.updateStats(ctx, constants.TaskTypePollingCarddata, true, time.Since(startTime))
+	h.base.updateStats(ctx, constants.TaskTypePollingCarddata, true, time.Since(startedAt))
 	return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata)
 }
 
-// calculateFlowUpdates 计算流量更新字段、增量和是否跨月
-// 决策13：完整迁移跨月流量边界检测逻辑（月份切换检测、上月总量保存、当月计数器重置）
-// 第三个返回值 isCrossMonth 供调用方同步更新 Redis 缓存，避免下次误判跨月
-func (h *PollingCarddataHandler) calculateFlowUpdates(card *model.IotCard, gatewayFlowMB float64, now time.Time, resetDay int) (map[string]any, float64, bool) {
-	updates := make(map[string]any)
-
-	increment := gatewayFlowMB - card.LastGatewayReadingMB
-	shouldUpdateGatewayReading := true
-	if increment < 0 {
-		if isResetWindow(now, resetDay) {
-			increment = gatewayFlowMB
-			h.base.logger.Info("检测到上游运营商重置",
-				zap.Uint("card_id", card.ID),
-				zap.Float64("gateway_flow", gatewayFlowMB),
-				zap.Float64("last_reading", card.LastGatewayReadingMB),
-				zap.Int("reset_day", resetDay))
-		} else {
-			h.base.logger.Warn("流量异常：非重置日出现值下降",
-				zap.Uint("card_id", card.ID),
-				zap.Float64("gateway_flow", gatewayFlowMB),
-				zap.Float64("last_reading", card.LastGatewayReadingMB))
-			increment = 0
-			shouldUpdateGatewayReading = false
-		}
+func (h *PollingCarddataHandler) failAndRequeue(ctx context.Context, cardID uint, startedAt time.Time, message string, err error) error {
+	fields := []zap.Field{zap.Uint("card_id", cardID)}
+	if err != nil {
+		fields = append(fields, zap.Error(err))
 	}
-
-	if shouldUpdateGatewayReading {
-		updates["last_gateway_reading_mb"] = gatewayFlowMB
-	}
-
-	currentMonthStart := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, now.Location())
-	isCrossMonth := card.CurrentMonthStartDate == nil ||
-		card.CurrentMonthStartDate.Before(currentMonthStart)
-
-	if isCrossMonth {
-		h.base.logger.Info("检测到跨月，重置流量计数",
-			zap.Uint("card_id", card.ID),
-			zap.Float64("last_month_total", card.CurrentMonthUsageMB))
-		updates["last_month_total_mb"] = card.CurrentMonthUsageMB
-		updates["current_month_start_date"] = currentMonthStart
-		if increment > 0 {
-			updates["current_month_usage_mb"] = increment
-		} else {
-			updates["current_month_usage_mb"] = float64(0)
-		}
-	} else if increment > 0 {
-		updates["current_month_usage_mb"] = gorm.Expr("current_month_usage_mb + ?", increment)
-	}
-
-	if increment > 0 {
-		updates["data_usage_mb"] = gorm.Expr("data_usage_mb + ?", int64(increment))
-	}
-
-	if card.CurrentMonthStartDate == nil {
-		updates["current_month_start_date"] = currentMonthStart
-	}
-
-	return updates, increment, isCrossMonth
+	h.base.logger.Warn(message, fields...)
+	h.base.updateStats(ctx, constants.TaskTypePollingCarddata, false, time.Since(startedAt))
+	return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata)
 }