201 lines
7.3 KiB
Go
201 lines
7.3 KiB
Go
package task
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
"github.com/hibiken/asynq"
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
|
||
"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/store/postgres"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
)
|
||
|
||
// PollingCarddataHandler 流量检查任务处理器
|
||
// 职责：调 Gateway 查流量 → 写 DB → 扣减套餐流量 → 调 EvaluateAndAct 判断停复机
|
||
// 不做：直接停复机决策，委托给 StopResumeService.EvaluateAndAct
|
||
type PollingCarddataHandler struct {
|
||
base *PollingBase
|
||
gateway *gateway.Client
|
||
iotCardStore *postgres.IotCardStore
|
||
carrierStore *postgres.CarrierStore
|
||
usageService *packagepkg.UsageService
|
||
stopResumeSvc iot_card_svc.StopResumeServiceInterface
|
||
}
|
||
|
||
// 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,
|
||
}
|
||
}
|
||
|
||
// Handle 处理流量检查任务
|
||
func (h *PollingCarddataHandler) Handle(ctx context.Context, t *asynq.Task) error {
|
||
startTime := time.Now()
|
||
|
||
cardID, ok := parseTaskPayload(t.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)
|
||
|
||
card, err := h.base.getCardWithCache(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.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
|
||
}
|
||
|
||
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
|
||
}
|
||
|
||
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 updateErr := h.iotCardStore.UpdateFields(ctx, cardID, updates); updateErr != nil {
|
||
h.base.logger.Error("更新卡流量信息失败", zap.Uint("card_id", cardID), zap.Error(updateErr))
|
||
}
|
||
|
||
go h.base.insertDataUsageRecord(context.Background(), cardID, flowIncrementMB, now)
|
||
|
||
cacheUpdates := map[string]any{"last_gateway_reading_mb": gatewayFlowMB}
|
||
if isCrossMonth {
|
||
// 跨月时同步更新缓存中的本月开始日期和本月用量，避免下次从缓存读取时再次误判跨月
|
||
currentMonthStart := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, now.Location())
|
||
cacheUpdates["current_month_start_date"] = currentMonthStart.Unix()
|
||
if flowIncrementMB > 0 {
|
||
cacheUpdates["current_month_usage_mb"] = flowIncrementMB
|
||
} else {
|
||
cacheUpdates["current_month_usage_mb"] = float64(0)
|
||
}
|
||
}
|
||
h.base.updateCardCache(ctx, cardID, cacheUpdates)
|
||
|
||
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))
|
||
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
|
||
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
|
||
}
|
||
}
|
||
|
||
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
|
||
}
|