huang
8980e6d999
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 7m11s
问题一：protect 轮询任务从未被调度器初始化入队 PollingConfig 新增 protect_check_interval 字段（NULL=不参与）， 调度器 initCardsBatch/initCardPolling/requeueCard 补全 protect 队列 初始化逻辑，IotCard 新增 last_protect_check_at 记录上次检查时间。 迁移文件：000102_add_polling_protect_fields 问题二：设备套餐流量耗尽时绑定卡未被停机 UsageService.checkAndTriggerSuspension 新增 carrier_type=device 分支， 通过 DeviceSimBindingStore.ListByDeviceID 查询绑定卡，对每张卡异步 触发 CheckAndStopCard，同步注入 Bootstrap 依赖。 问题三：主套餐过期后不立即停机，依赖下次轮询兜底 PackageActivationHandler.processExpiredPackage 在 updateCarrierSuspended Status 后，若载体无后续生效套餐，立即异步调用 CheckAndStopCard（iot_card 类型直接触发，device 类型遍历绑定卡），消除停机延迟窗口。 Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
1361 lines
45 KiB
Go
1361 lines
45 KiB
Go
package task
|
||
|
||
import (
|
||
"context"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/bytedance/sonic"
|
||
"github.com/hibiken/asynq"
|
||
"github.com/redis/go-redis/v9"
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/gateway"
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
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"
|
||
)
|
||
|
||
// PollingTaskPayload 轮询任务载荷
|
||
type PollingTaskPayload struct {
|
||
CardID string `json:"card_id"`
|
||
IsManual bool `json:"is_manual"`
|
||
Timestamp int64 `json:"timestamp"`
|
||
}
|
||
|
||
// PollingHandler 轮询任务处理器
|
||
type PollingHandler struct {
|
||
db *gorm.DB
|
||
redis *redis.Client
|
||
asynqClient *asynq.Client // 用于提交 Asynq 任务（如首次实名激活）
|
||
gatewayClient *gateway.Client
|
||
iotCardStore *postgres.IotCardStore
|
||
concurrencyStore *postgres.PollingConcurrencyConfigStore
|
||
pollingConfigStore *postgres.PollingConfigStore // 用于 requeueCard 读取配置间隔
|
||
deviceSimBindingStore *postgres.DeviceSimBindingStore
|
||
packageUsageStore *postgres.PackageUsageStore
|
||
usageService *packagepkg.UsageService
|
||
logger *zap.Logger
|
||
}
|
||
|
||
// NewPollingHandler 创建轮询任务处理器
|
||
func NewPollingHandler(
|
||
db *gorm.DB,
|
||
redis *redis.Client,
|
||
asynqClient *asynq.Client,
|
||
gatewayClient *gateway.Client,
|
||
usageService *packagepkg.UsageService,
|
||
logger *zap.Logger,
|
||
) *PollingHandler {
|
||
return &PollingHandler{
|
||
db: db,
|
||
redis: redis,
|
||
asynqClient: asynqClient,
|
||
gatewayClient: gatewayClient,
|
||
iotCardStore: postgres.NewIotCardStore(db, redis),
|
||
concurrencyStore: postgres.NewPollingConcurrencyConfigStore(db),
|
||
pollingConfigStore: postgres.NewPollingConfigStore(db),
|
||
deviceSimBindingStore: postgres.NewDeviceSimBindingStore(db, redis),
|
||
packageUsageStore: postgres.NewPackageUsageStore(db, redis),
|
||
usageService: usageService,
|
||
logger: logger,
|
||
}
|
||
}
|
||
|
||
// HandleRealnameCheck 处理实名检查任务
|
||
func (h *PollingHandler) HandleRealnameCheck(ctx context.Context, t *asynq.Task) error {
|
||
startTime := time.Now()
|
||
|
||
var payload PollingTaskPayload
|
||
if err := sonic.Unmarshal(t.Payload(), &payload); err != nil {
|
||
h.logger.Error("解析任务载荷失败", zap.Error(err))
|
||
return nil // 不重试
|
||
}
|
||
|
||
cardID, err := strconv.ParseUint(payload.CardID, 10, 64)
|
||
if err != nil {
|
||
h.logger.Error("解析卡ID失败", zap.String("card_id", payload.CardID), zap.Error(err))
|
||
return nil
|
||
}
|
||
|
||
// 获取并发信号量
|
||
if !h.acquireConcurrency(ctx, constants.TaskTypePollingRealname) {
|
||
h.logger.Debug("并发已满，任务稍后重试", zap.Uint64("card_id", cardID))
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingRealname)
|
||
}
|
||
defer h.releaseConcurrency(ctx, constants.TaskTypePollingRealname)
|
||
|
||
h.logger.Debug("开始实名检查",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.Bool("is_manual", payload.IsManual))
|
||
|
||
// 获取卡信息
|
||
card, err := h.getCardWithCache(ctx, uint(cardID))
|
||
if err != nil {
|
||
if err == gorm.ErrRecordNotFound {
|
||
h.logger.Warn("卡不存在", zap.Uint64("card_id", cardID))
|
||
return nil
|
||
}
|
||
h.logger.Error("获取卡信息失败", zap.Uint64("card_id", cardID), zap.Error(err))
|
||
h.updateStats(ctx, constants.TaskTypePollingRealname, false, time.Since(startTime))
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingRealname)
|
||
}
|
||
|
||
// 行业卡跳过实名检查
|
||
if card.CardCategory == "industry" {
|
||
h.logger.Debug("行业卡跳过实名检查", zap.Uint64("card_id", cardID))
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingRealname)
|
||
}
|
||
|
||
// 调用 Gateway API 查询实名状态
|
||
var newRealnameStatus int
|
||
if h.gatewayClient != nil {
|
||
result, err := h.gatewayClient.QueryRealnameStatus(ctx, &gateway.CardStatusReq{
|
||
CardNo: card.ICCID,
|
||
})
|
||
if err != nil {
|
||
h.logger.Warn("查询实名状态失败",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Error(err))
|
||
h.updateStats(ctx, constants.TaskTypePollingRealname, false, time.Since(startTime))
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingRealname)
|
||
}
|
||
|
||
// 解析实名状态
|
||
newRealnameStatus = h.parseRealnameStatus(result.RealStatus)
|
||
h.logger.Info("实名检查完成",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Bool("gateway_real_status", result.RealStatus),
|
||
zap.Int("new_status", newRealnameStatus),
|
||
zap.Int("old_status", card.RealNameStatus))
|
||
} else {
|
||
// Gateway 未配置，模拟检查
|
||
newRealnameStatus = card.RealNameStatus
|
||
h.logger.Debug("实名检查完成（模拟，Gateway未配置）",
|
||
zap.Uint64("card_id", cardID))
|
||
}
|
||
|
||
// 检测状态变化
|
||
statusChanged := newRealnameStatus != card.RealNameStatus
|
||
|
||
// 提前检测首次实名（之前未实名 → 现在已实名），用于同步记录 first_realname_at
|
||
isFirstRealname := statusChanged &&
|
||
card.RealNameStatus != constants.RealNameStatusVerified &&
|
||
newRealnameStatus == constants.RealNameStatusVerified
|
||
|
||
// 更新数据库
|
||
now := time.Now()
|
||
updates := map[string]any{
|
||
"last_real_name_check_at": now,
|
||
}
|
||
if statusChanged {
|
||
updates["real_name_status"] = newRealnameStatus
|
||
}
|
||
// 首次实名时记录实名时间，后续实名状态不覆盖
|
||
if isFirstRealname {
|
||
updates["first_realname_at"] = now
|
||
}
|
||
|
||
if err := h.db.Model(&model.IotCard{}).
|
||
Where("id = ?", cardID).
|
||
Updates(updates).Error; err != nil {
|
||
h.logger.Error("更新卡信息失败", zap.Uint64("card_id", cardID), zap.Error(err))
|
||
}
|
||
|
||
// 如果状态变化，更新 Redis 缓存并重新匹配配置
|
||
if statusChanged {
|
||
h.updateCardCache(ctx, uint(cardID), map[string]any{
|
||
"real_name_status": newRealnameStatus,
|
||
})
|
||
// 状态变化后需要重新匹配配置（通过调度器回调）
|
||
h.logger.Info("实名状态已变化，需要重新匹配配置",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.Int("old_status", card.RealNameStatus),
|
||
zap.Int("new_status", newRealnameStatus))
|
||
|
||
if isFirstRealname {
|
||
h.triggerFirstRealnameActivation(ctx, uint(cardID))
|
||
}
|
||
}
|
||
|
||
// 更新监控统计
|
||
h.updateStats(ctx, constants.TaskTypePollingRealname, true, time.Since(startTime))
|
||
|
||
// 重新入队
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingRealname)
|
||
}
|
||
|
||
// HandleCarddataCheck 处理卡流量检查任务
|
||
// 任务 18.2-18.4: 改造为支持流量扣减优先级和新停机条件
|
||
func (h *PollingHandler) HandleCarddataCheck(ctx context.Context, t *asynq.Task) error {
|
||
startTime := time.Now()
|
||
|
||
var payload PollingTaskPayload
|
||
if err := sonic.Unmarshal(t.Payload(), &payload); err != nil {
|
||
h.logger.Error("解析任务载荷失败", zap.Error(err))
|
||
return nil
|
||
}
|
||
|
||
cardID, err := strconv.ParseUint(payload.CardID, 10, 64)
|
||
if err != nil {
|
||
h.logger.Error("解析卡ID失败", zap.String("card_id", payload.CardID), zap.Error(err))
|
||
return nil
|
||
}
|
||
|
||
// 获取并发信号量
|
||
if !h.acquireConcurrency(ctx, constants.TaskTypePollingCarddata) {
|
||
h.logger.Debug("并发已满，任务稍后重试", zap.Uint64("card_id", cardID))
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingCarddata)
|
||
}
|
||
defer h.releaseConcurrency(ctx, constants.TaskTypePollingCarddata)
|
||
|
||
h.logger.Debug("开始流量检查",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.Bool("is_manual", payload.IsManual))
|
||
|
||
// 获取卡信息
|
||
card, err := h.getCardWithCache(ctx, uint(cardID))
|
||
if err != nil {
|
||
if err == gorm.ErrRecordNotFound {
|
||
h.logger.Warn("卡不存在", zap.Uint64("card_id", cardID))
|
||
return nil
|
||
}
|
||
h.logger.Error("获取卡信息失败", zap.Uint64("card_id", cardID), zap.Error(err))
|
||
h.updateStats(ctx, constants.TaskTypePollingCarddata, false, time.Since(startTime))
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingCarddata)
|
||
}
|
||
|
||
// 必须在 gateway 查询前执行：gateway 报错会提前 return，checkStopResume 放在后面时无套餐在线卡无法被自动停机
|
||
h.checkStopResume(ctx, card)
|
||
|
||
// 调用 Gateway API 查询流量
|
||
var gatewayFlowMB float64
|
||
if h.gatewayClient != nil {
|
||
result, err := h.gatewayClient.QueryFlow(ctx, &gateway.FlowQueryReq{
|
||
CardNo: card.ICCID,
|
||
})
|
||
if err != nil {
|
||
h.logger.Warn("查询流量失败",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Error(err))
|
||
h.updateStats(ctx, constants.TaskTypePollingCarddata, false, time.Since(startTime))
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingCarddata)
|
||
}
|
||
|
||
// Gateway 返回的是 MB 单位的流量
|
||
gatewayFlowMB = result.Used
|
||
h.logger.Info("流量检查完成",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Float64("gateway_flow_mb", gatewayFlowMB),
|
||
zap.String("unit", result.Unit))
|
||
} else {
|
||
// Gateway 未配置，使用当前值
|
||
gatewayFlowMB = card.CurrentMonthUsageMB
|
||
h.logger.Debug("流量检查完成（模拟，Gateway未配置）",
|
||
zap.Uint64("card_id", cardID))
|
||
}
|
||
|
||
now := time.Now()
|
||
updates, flowIncrementMB := h.calculateFlowUpdates(card, gatewayFlowMB, now)
|
||
updates["last_data_check_at"] = now
|
||
|
||
// 更新数据库
|
||
if err := h.db.Model(&model.IotCard{}).
|
||
Where("id = ?", cardID).
|
||
Updates(updates).Error; err != nil {
|
||
h.logger.Error("更新卡流量信息失败", zap.Uint64("card_id", cardID), zap.Error(err))
|
||
}
|
||
|
||
go h.insertDataUsageRecord(context.Background(), uint(cardID), flowIncrementMB, now)
|
||
|
||
// 更新 Redis 缓存
|
||
h.updateCardCache(ctx, uint(cardID), map[string]any{
|
||
"current_month_usage_mb": updates["current_month_usage_mb"],
|
||
"last_gateway_reading_mb": gatewayFlowMB,
|
||
})
|
||
|
||
// 调用 UsageService.DeductDataUsage 进行流量扣减
|
||
if flowIncrementMB > 0 && h.usageService != nil {
|
||
if err := h.usageService.DeductDataUsage(ctx, "iot_card", uint(cardID), int64(flowIncrementMB)); err != nil {
|
||
h.logger.Warn("套餐流量扣减失败",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.Float64("increment_mb", flowIncrementMB),
|
||
zap.Error(err))
|
||
} else {
|
||
h.logger.Info("套餐流量扣减成功",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.Float64("increment_mb", flowIncrementMB))
|
||
}
|
||
}
|
||
|
||
// 更新监控统计
|
||
h.updateStats(ctx, constants.TaskTypePollingCarddata, true, time.Since(startTime))
|
||
|
||
// 触发套餐检查（加入手动队列优先处理）
|
||
h.triggerPackageCheck(ctx, uint(cardID))
|
||
|
||
// 重新入队
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingCarddata)
|
||
}
|
||
|
||
// getCarrierResetDay 读取运营商的上游流量重置日
|
||
func (h *PollingHandler) getCarrierResetDay(carrierID uint) int {
|
||
var carrier model.Carrier
|
||
if err := h.db.Select("data_reset_day").First(&carrier, carrierID).Error; err != nil {
|
||
return 1
|
||
}
|
||
if carrier.DataResetDay == 0 {
|
||
return 1
|
||
}
|
||
return carrier.DataResetDay
|
||
}
|
||
|
||
// isResetWindow 判断今天是否在运营商重置日窗口内
|
||
// 窗口 = 重置日当天 + 前一天（容错网关数据延迟）
|
||
func isResetWindow(now time.Time, resetDay int) bool {
|
||
today := now.Day()
|
||
if today == resetDay {
|
||
return true
|
||
}
|
||
resetDate := time.Date(now.Year(), now.Month(), resetDay, 0, 0, 0, 0, now.Location())
|
||
prevDay := resetDate.AddDate(0, 0, -1).Day()
|
||
return today == prevDay
|
||
}
|
||
|
||
// shouldStopCard 检查是否应该停机（无有效套餐）
|
||
// 有效套餐 = 待生效(0) 或 生效中(1)，与 StopResumeService.hasAvailablePackage 对齐
|
||
func (h *PollingHandler) shouldStopCard(ctx context.Context, cardID uint) bool {
|
||
var count int64
|
||
if err := h.db.WithContext(ctx).Model(&model.PackageUsage{}).
|
||
Where("iot_card_id = ? AND status IN ?", cardID, []int{
|
||
constants.PackageUsageStatusPending, // 待生效
|
||
constants.PackageUsageStatusActive, // 生效中
|
||
}).
|
||
Count(&count).Error; err != nil {
|
||
h.logger.Warn("查询有效套餐失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||
return false
|
||
}
|
||
|
||
return count == 0
|
||
}
|
||
|
||
// hasAvailablePackage 检查卡是否有可用套餐（待生效或生效中）
|
||
func (h *PollingHandler) hasAvailablePackage(ctx context.Context, cardID uint) bool {
|
||
var count int64
|
||
if err := h.db.WithContext(ctx).Model(&model.PackageUsage{}).
|
||
Where("iot_card_id = ? AND status IN ?", cardID, []int{
|
||
constants.PackageUsageStatusPending,
|
||
constants.PackageUsageStatusActive,
|
||
}).
|
||
Count(&count).Error; err != nil {
|
||
h.logger.Warn("查询可用套餐失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||
return false
|
||
}
|
||
|
||
return count > 0
|
||
}
|
||
|
||
// stopCardByUsageExhausted 任务 18.4: 流量耗尽停机
|
||
func (h *PollingHandler) stopCardByUsageExhausted(ctx context.Context, card *model.IotCard) {
|
||
// 只有在线的卡才需要停机
|
||
if card.NetworkStatus != 1 {
|
||
return
|
||
}
|
||
|
||
// 调用 Gateway 停机
|
||
if h.gatewayClient != nil {
|
||
h.logger.Info("调用网关停机",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
if err := h.gatewayClient.StopCard(ctx, &gateway.CardOperationReq{
|
||
CardNo: card.ICCID,
|
||
}); err != nil {
|
||
h.logger.Error("停机失败",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Error(err))
|
||
return
|
||
}
|
||
h.logger.Info("网关停机成功",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
}
|
||
|
||
// 更新数据库：卡的网络状态
|
||
now := time.Now()
|
||
updates := map[string]any{
|
||
"network_status": 0,
|
||
"stopped_at": now,
|
||
"stop_reason": constants.StopReasonTrafficExhausted,
|
||
"updated_at": now,
|
||
}
|
||
if err := h.db.Model(&model.IotCard{}).
|
||
Where("id = ?", card.ID).
|
||
Updates(updates).Error; err != nil {
|
||
h.logger.Error("更新卡状态失败", zap.Uint("card_id", card.ID), zap.Error(err))
|
||
}
|
||
|
||
h.updateCardCache(ctx, card.ID, map[string]any{
|
||
"network_status": 0,
|
||
"stop_reason": constants.StopReasonTrafficExhausted,
|
||
})
|
||
|
||
h.logger.Warn("卡已停机（套餐流量耗尽）",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
}
|
||
|
||
// checkStopResume 独立停机/复机检查
|
||
// 停机条件：卡在线 + 无有效套餐(pending/active) → 停机
|
||
// 复机条件：卡停机 + 停机原因为流量耗尽 + 有有效套餐 → 复机
|
||
func (h *PollingHandler) checkStopResume(ctx context.Context, card *model.IotCard) {
|
||
if card.NetworkStatus == constants.NetworkStatusOnline {
|
||
if h.shouldStopCard(ctx, card.ID) {
|
||
h.logger.Warn("无有效套餐，触发停机",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
h.stopCardByUsageExhausted(ctx, card)
|
||
}
|
||
return
|
||
}
|
||
|
||
// 停机卡：只有流量耗尽停机的才自动复机，手动停机不管
|
||
if card.NetworkStatus == constants.NetworkStatusOffline &&
|
||
card.StopReason == constants.StopReasonTrafficExhausted &&
|
||
h.hasAvailablePackage(ctx, card.ID) {
|
||
h.logger.Info("检测到有效套餐，触发自动复机",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
h.resumeCardByPackageAvailable(ctx, card)
|
||
}
|
||
}
|
||
|
||
// resumeCardByPackageAvailable 有效套餐恢复后自动复机
|
||
func (h *PollingHandler) resumeCardByPackageAvailable(ctx context.Context, card *model.IotCard) {
|
||
if card.RealNameStatus != constants.RealNameStatusVerified {
|
||
h.logger.Warn("未实名卡跳过自动复机",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
return
|
||
}
|
||
|
||
// 分布式锁防止并发复机
|
||
lockKey := constants.RedisCardResumeLockKey(card.ID)
|
||
locked, err := h.redis.SetNX(ctx, lockKey, "1", 30*time.Second).Result()
|
||
if err != nil || !locked {
|
||
h.logger.Debug("复机操作进行中，跳过", zap.Uint("card_id", card.ID))
|
||
return
|
||
}
|
||
defer h.redis.Del(ctx, lockKey)
|
||
|
||
if h.gatewayClient != nil {
|
||
h.logger.Info("调用网关复机",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
if err := h.gatewayClient.StartCard(ctx, &gateway.CardOperationReq{
|
||
CardNo: card.ICCID,
|
||
}); err != nil {
|
||
h.logger.Error("复机失败",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Error(err))
|
||
return
|
||
}
|
||
h.logger.Info("网关复机成功",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
}
|
||
|
||
now := time.Now()
|
||
if err := h.db.Model(&model.IotCard{}).
|
||
Where("id = ?", card.ID).
|
||
Updates(map[string]any{
|
||
"network_status": constants.NetworkStatusOnline,
|
||
"resumed_at": now,
|
||
"stop_reason": "",
|
||
"updated_at": now,
|
||
}).Error; err != nil {
|
||
h.logger.Error("复机更新卡状态失败", zap.Uint("card_id", card.ID), zap.Error(err))
|
||
return
|
||
}
|
||
|
||
h.updateCardCache(ctx, card.ID, map[string]any{
|
||
"network_status": constants.NetworkStatusOnline,
|
||
"stop_reason": "",
|
||
})
|
||
|
||
h.logger.Info("卡已自动复机（检测到有效套餐）",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
}
|
||
|
||
// calculateFlowUpdates 计算流量更新值和增量（合并原 calculateFlowUpdates 和 calculateFlowIncrement）
|
||
// 增量算法：increment = 当前网关读数 - 上次网关读数，检测上游重置和跨自然月
|
||
func (h *PollingHandler) calculateFlowUpdates(card *model.IotCard, gatewayFlowMB float64, now time.Time) (map[string]any, float64) {
|
||
updates := make(map[string]any)
|
||
|
||
// 增量 = 当前读数 - 上次读数
|
||
increment := gatewayFlowMB - card.LastGatewayReadingMB
|
||
|
||
if increment < 0 {
|
||
// 当前值比上次小，检查是否为上游运营商正常重置
|
||
resetDay := h.getCarrierResetDay(card.CarrierID)
|
||
if isResetWindow(now, resetDay) {
|
||
// 在重置日窗口内，视为正常上游重置，本次原始值即为增量
|
||
increment = gatewayFlowMB
|
||
h.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.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.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
|
||
// 跨月重置为 0，再加上本次增量
|
||
if increment > 0 {
|
||
updates["current_month_usage_mb"] = increment
|
||
} else {
|
||
updates["current_month_usage_mb"] = float64(0)
|
||
}
|
||
} else if increment > 0 {
|
||
// 同月内，增量累加（使用 gorm.Expr 保证原子性）
|
||
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
|
||
}
|
||
|
||
// HandlePackageCheck 处理套餐检查任务
|
||
func (h *PollingHandler) HandlePackageCheck(ctx context.Context, t *asynq.Task) error {
|
||
startTime := time.Now()
|
||
|
||
var payload PollingTaskPayload
|
||
if err := sonic.Unmarshal(t.Payload(), &payload); err != nil {
|
||
h.logger.Error("解析任务载荷失败", zap.Error(err))
|
||
return nil
|
||
}
|
||
|
||
cardID, err := strconv.ParseUint(payload.CardID, 10, 64)
|
||
if err != nil {
|
||
h.logger.Error("解析卡ID失败", zap.String("card_id", payload.CardID), zap.Error(err))
|
||
return nil
|
||
}
|
||
|
||
// 获取并发信号量
|
||
if !h.acquireConcurrency(ctx, constants.TaskTypePollingPackage) {
|
||
h.logger.Debug("并发已满，任务稍后重试", zap.Uint64("card_id", cardID))
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingPackage)
|
||
}
|
||
defer h.releaseConcurrency(ctx, constants.TaskTypePollingPackage)
|
||
|
||
h.logger.Debug("开始套餐检查",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.Bool("is_manual", payload.IsManual))
|
||
|
||
// 获取卡信息
|
||
card, err := h.getCardWithCache(ctx, uint(cardID))
|
||
if err != nil {
|
||
if err == gorm.ErrRecordNotFound {
|
||
h.logger.Warn("卡不存在", zap.Uint64("card_id", cardID))
|
||
return nil
|
||
}
|
||
h.logger.Error("获取卡信息失败", zap.Uint64("card_id", cardID), zap.Error(err))
|
||
h.updateStats(ctx, constants.TaskTypePollingPackage, false, time.Since(startTime))
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingPackage)
|
||
}
|
||
|
||
// 检查套餐配置
|
||
if card.SeriesID == nil {
|
||
h.logger.Debug("卡无关联套餐系列，跳过检查", zap.Uint64("card_id", cardID))
|
||
h.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime))
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingPackage)
|
||
}
|
||
|
||
// 查询套餐信息（获取虚流量限制）
|
||
var pkg model.Package
|
||
if err := h.db.Where("series_id = ? AND status = 1", *card.SeriesID).
|
||
Order("created_at ASC").First(&pkg).Error; err != nil {
|
||
if err == gorm.ErrRecordNotFound {
|
||
h.logger.Debug("套餐系列无可用套餐", zap.Uint("series_id", *card.SeriesID))
|
||
} else {
|
||
h.logger.Warn("查询套餐失败", zap.Uint("series_id", *card.SeriesID), zap.Error(err))
|
||
}
|
||
h.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime))
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingPackage)
|
||
}
|
||
|
||
// 检查是否启用虚流量
|
||
if !pkg.EnableVirtualData || pkg.VirtualDataMB <= 0 {
|
||
h.logger.Debug("套餐未启用虚流量或虚流量为0，跳过检查",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.Uint("package_id", pkg.ID),
|
||
zap.Bool("enable_virtual", pkg.EnableVirtualData),
|
||
zap.Int64("virtual_data_mb", pkg.VirtualDataMB))
|
||
h.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime))
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingPackage)
|
||
}
|
||
|
||
// 计算流量使用：支持设备级套餐流量汇总
|
||
usedMB, deviceCards, isDeviceLevel := h.calculatePackageUsage(ctx, card)
|
||
limitMB := float64(pkg.VirtualDataMB)
|
||
usagePercent := (usedMB / limitMB) * 100
|
||
|
||
h.logger.Info("套餐流量检查",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Float64("used_mb", usedMB),
|
||
zap.Float64("limit_mb", limitMB),
|
||
zap.Float64("usage_percent", usagePercent),
|
||
zap.Bool("is_device_level", isDeviceLevel),
|
||
zap.Int("device_card_count", len(deviceCards)))
|
||
|
||
// 判断状态
|
||
var needStop bool
|
||
var statusMsg string
|
||
|
||
switch {
|
||
case usedMB > limitMB:
|
||
// 已超额，需要停机
|
||
needStop = true
|
||
statusMsg = "已超额"
|
||
h.logger.Warn("套餐已超额，准备停机",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.Float64("used_mb", usedMB),
|
||
zap.Float64("limit_mb", limitMB))
|
||
case usagePercent >= 95:
|
||
// 临近超额（预警）
|
||
statusMsg = "临近超额"
|
||
h.logger.Warn("套餐流量临近超额",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.Float64("usage_percent", usagePercent))
|
||
default:
|
||
// 正常
|
||
statusMsg = "正常"
|
||
}
|
||
|
||
// 执行停机
|
||
if needStop {
|
||
// 设备级套餐需要停机设备下所有卡
|
||
cardsToStop := []*model.IotCard{card}
|
||
if isDeviceLevel && len(deviceCards) > 0 {
|
||
cardsToStop = deviceCards
|
||
}
|
||
|
||
if h.gatewayClient != nil {
|
||
h.stopCards(ctx, cardsToStop, &pkg, usedMB)
|
||
} else {
|
||
h.logger.Debug("停机跳过（Gateway未配置）",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.Int("cards_to_stop", len(cardsToStop)))
|
||
}
|
||
}
|
||
|
||
h.logger.Info("套餐检查完成",
|
||
zap.Uint64("card_id", cardID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Float64("used_mb", usedMB),
|
||
zap.Float64("limit_mb", limitMB),
|
||
zap.String("status", statusMsg),
|
||
zap.Bool("stopped", needStop && card.NetworkStatus == 1),
|
||
zap.Duration("duration", time.Since(startTime)))
|
||
|
||
// 更新监控统计
|
||
h.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime))
|
||
|
||
// 重新入队
|
||
return h.requeueCard(ctx, uint(cardID), constants.TaskTypePollingPackage)
|
||
}
|
||
|
||
// logStopOperation 记录停机操作日志
|
||
func (h *PollingHandler) logStopOperation(_ context.Context, card *model.IotCard, pkg *model.Package, usedMB float64) {
|
||
// 记录详细的停机操作日志（应用日志级别）
|
||
h.logger.Info("停机操作记录",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Uint("package_id", pkg.ID),
|
||
zap.String("package_name", pkg.PackageName),
|
||
zap.Float64("used_mb", usedMB),
|
||
zap.Int64("virtual_data_mb", pkg.VirtualDataMB),
|
||
zap.String("reason", "套餐超额自动停机"),
|
||
zap.String("operator", "系统自动"),
|
||
zap.Int("before_network_status", 1),
|
||
zap.Int("after_network_status", 0))
|
||
}
|
||
|
||
// calculatePackageUsage 计算套餐流量使用（支持设备级套餐汇总）
|
||
// 返回：总流量MB、设备下所有卡（如果是设备级套餐）、是否为设备级套餐
|
||
func (h *PollingHandler) calculatePackageUsage(ctx context.Context, card *model.IotCard) (float64, []*model.IotCard, bool) {
|
||
// 检查卡是否绑定到设备
|
||
binding, err := h.deviceSimBindingStore.GetActiveBindingByCardID(ctx, card.ID)
|
||
if err != nil {
|
||
// 卡未绑定到设备，使用单卡流量
|
||
return card.CurrentMonthUsageMB, nil, false
|
||
}
|
||
|
||
// 卡绑定到设备，获取设备下所有卡
|
||
bindings, err := h.deviceSimBindingStore.ListByDeviceID(ctx, binding.DeviceID)
|
||
if err != nil {
|
||
h.logger.Warn("查询设备下所有绑定失败，使用单卡流量",
|
||
zap.Uint("device_id", binding.DeviceID),
|
||
zap.Error(err))
|
||
return card.CurrentMonthUsageMB, nil, false
|
||
}
|
||
|
||
if len(bindings) == 0 {
|
||
return card.CurrentMonthUsageMB, nil, false
|
||
}
|
||
|
||
// 获取设备下所有卡的 ID
|
||
cardIDs := make([]uint, len(bindings))
|
||
for i, b := range bindings {
|
||
cardIDs[i] = b.IotCardID
|
||
}
|
||
|
||
// 批量查询卡信息
|
||
var cards []*model.IotCard
|
||
if err := h.db.WithContext(ctx).Where("id IN ?", cardIDs).Find(&cards).Error; err != nil {
|
||
h.logger.Warn("查询设备下所有卡失败，使用单卡流量",
|
||
zap.Uint("device_id", binding.DeviceID),
|
||
zap.Error(err))
|
||
return card.CurrentMonthUsageMB, nil, false
|
||
}
|
||
|
||
// 汇总流量
|
||
var totalUsedMB float64
|
||
for _, c := range cards {
|
||
totalUsedMB += c.CurrentMonthUsageMB
|
||
}
|
||
|
||
h.logger.Debug("设备级套餐流量汇总",
|
||
zap.Uint("device_id", binding.DeviceID),
|
||
zap.Int("card_count", len(cards)),
|
||
zap.Float64("total_used_mb", totalUsedMB))
|
||
|
||
return totalUsedMB, cards, true
|
||
}
|
||
|
||
// stopCards 批量停机卡
|
||
func (h *PollingHandler) stopCards(ctx context.Context, cards []*model.IotCard, pkg *model.Package, usedMB float64) {
|
||
for _, card := range cards {
|
||
// 跳过已停机的卡
|
||
if card.NetworkStatus != 1 {
|
||
continue
|
||
}
|
||
|
||
h.logger.Info("调用网关停机",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
err := h.gatewayClient.StopCard(ctx, &gateway.CardOperationReq{
|
||
CardNo: card.ICCID,
|
||
})
|
||
if err != nil {
|
||
h.logger.Error("停机失败",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Error(err))
|
||
continue
|
||
}
|
||
|
||
h.logger.Warn("停机成功",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.String("reason", "套餐超额"))
|
||
|
||
now := time.Now()
|
||
if err := h.db.Model(&model.IotCard{}).
|
||
Where("id = ?", card.ID).
|
||
Updates(map[string]any{
|
||
"network_status": 0,
|
||
"stopped_at": now,
|
||
"stop_reason": constants.StopReasonTrafficExhausted,
|
||
"updated_at": now,
|
||
}).Error; err != nil {
|
||
h.logger.Error("更新卡状态失败", zap.Uint("card_id", card.ID), zap.Error(err))
|
||
}
|
||
|
||
// 更新 Redis 缓存
|
||
h.updateCardCache(ctx, card.ID, map[string]any{
|
||
"network_status": 0,
|
||
})
|
||
|
||
// 记录操作日志
|
||
h.logStopOperation(ctx, card, pkg, usedMB)
|
||
}
|
||
}
|
||
|
||
// parseRealnameStatus 解析实名状态
|
||
// Gateway 返回 bool 类型：true=已实名, false=未实名
|
||
func (h *PollingHandler) parseRealnameStatus(realStatus bool) int {
|
||
if realStatus {
|
||
return constants.RealNameStatusVerified // = 1，已实名
|
||
}
|
||
return constants.RealNameStatusNotVerified // = 0，未实名
|
||
}
|
||
|
||
// extractTaskType 从完整的任务类型中提取简短的类型名
|
||
// 例如：polling:carddata -> carddata
|
||
func extractTaskType(fullTaskType string) string {
|
||
if idx := strings.LastIndex(fullTaskType, ":"); idx != -1 {
|
||
return fullTaskType[idx+1:]
|
||
}
|
||
return fullTaskType
|
||
}
|
||
|
||
// acquireConcurrency 获取并发信号量
|
||
func (h *PollingHandler) acquireConcurrency(ctx context.Context, taskType string) bool {
|
||
// 提取简短的任务类型（数据库中存的是 carddata，不是 polling:carddata）
|
||
shortType := extractTaskType(taskType)
|
||
configKey := constants.RedisPollingConcurrencyConfigKey(shortType)
|
||
currentKey := constants.RedisPollingConcurrencyCurrentKey(taskType) // current 保持原样
|
||
|
||
// 获取最大并发数
|
||
maxConcurrency, err := h.redis.Get(ctx, configKey).Int()
|
||
if err != nil {
|
||
maxConcurrency = 50 // 默认值
|
||
}
|
||
|
||
// 尝试获取信号量
|
||
current, err := h.redis.Incr(ctx, currentKey).Result()
|
||
if err != nil {
|
||
h.logger.Warn("获取并发计数失败", zap.Error(err))
|
||
return true // 出错时允许执行
|
||
}
|
||
|
||
if current > int64(maxConcurrency) {
|
||
h.redis.Decr(ctx, currentKey)
|
||
return false
|
||
}
|
||
|
||
return true
|
||
}
|
||
|
||
// releaseConcurrency 释放并发信号量
|
||
func (h *PollingHandler) releaseConcurrency(ctx context.Context, taskType string) {
|
||
currentKey := constants.RedisPollingConcurrencyCurrentKey(taskType)
|
||
if err := h.redis.Decr(ctx, currentKey).Err(); err != nil {
|
||
h.logger.Warn("释放并发计数失败", zap.Error(err))
|
||
}
|
||
}
|
||
|
||
// requeueCard 重新将卡加入队列
|
||
// 从 PollingConfig 读取实际配置间隔；无匹配配置或间隔为 NULL 时不入队（停止轮询该卡）
|
||
func (h *PollingHandler) requeueCard(ctx context.Context, cardID uint, taskType string) error {
|
||
intervalSeconds, err := h.getMatchedPollingInterval(ctx, cardID, taskType)
|
||
if err != nil {
|
||
h.logger.Warn("获取轮询配置间隔失败，跳过入队",
|
||
zap.Uint("card_id", cardID),
|
||
zap.String("task_type", taskType),
|
||
zap.Error(err))
|
||
return nil
|
||
}
|
||
if intervalSeconds <= 0 {
|
||
// 无匹配配置或该任务类型间隔为 NULL，不再轮询该卡
|
||
h.logger.Debug("无匹配轮询配置，跳过入队",
|
||
zap.Uint("card_id", cardID),
|
||
zap.String("task_type", taskType))
|
||
return nil
|
||
}
|
||
|
||
var queueKey string
|
||
switch taskType {
|
||
case constants.TaskTypePollingRealname:
|
||
queueKey = constants.RedisPollingQueueRealnameKey()
|
||
case constants.TaskTypePollingCarddata:
|
||
queueKey = constants.RedisPollingQueueCarddataKey()
|
||
case constants.TaskTypePollingPackage:
|
||
queueKey = constants.RedisPollingQueuePackageKey()
|
||
case constants.TaskTypePollingProtect:
|
||
queueKey = constants.RedisPollingQueueProtectKey()
|
||
default:
|
||
return nil
|
||
}
|
||
|
||
nextCheck := time.Now().Add(time.Duration(intervalSeconds) * time.Second)
|
||
if err := h.redis.ZAdd(ctx, queueKey, redis.Z{
|
||
Score: float64(nextCheck.Unix()),
|
||
Member: strconv.FormatUint(uint64(cardID), 10),
|
||
}).Err(); err != nil {
|
||
return err
|
||
}
|
||
|
||
// protect 任务重新入队时更新检查时间，供下次初始化计算队列延迟
|
||
if taskType == constants.TaskTypePollingProtect {
|
||
now := time.Now()
|
||
h.db.Model(&model.IotCard{}).Where("id = ?", cardID).Update("last_protect_check_at", now)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// getMatchedPollingInterval 读取该卡匹配的 PollingConfig，返回指定任务类型的间隔秒数
|
||
// 优先读取 Redis 缓存，miss 时回退到 DB；无匹配或间隔为 NULL 返回 0
|
||
func (h *PollingHandler) getMatchedPollingInterval(ctx context.Context, cardID uint, taskType string) (int, error) {
|
||
card, err := h.getCardWithCache(ctx, cardID)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
|
||
configs, err := h.getPollingConfigsFromCache(ctx)
|
||
if err != nil || len(configs) == 0 {
|
||
// Redis 缓存未命中，回退查 DB
|
||
configs, err = h.pollingConfigStore.ListEnabled(ctx)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
}
|
||
|
||
for _, cfg := range configs {
|
||
if !h.matchPollingConfig(cfg, card) {
|
||
continue
|
||
}
|
||
switch taskType {
|
||
case constants.TaskTypePollingRealname:
|
||
if cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0 {
|
||
return *cfg.RealnameCheckInterval, nil
|
||
}
|
||
case constants.TaskTypePollingCarddata:
|
||
if cfg.CarddataCheckInterval != nil && *cfg.CarddataCheckInterval > 0 {
|
||
return *cfg.CarddataCheckInterval, nil
|
||
}
|
||
case constants.TaskTypePollingPackage:
|
||
if cfg.PackageCheckInterval != nil && *cfg.PackageCheckInterval > 0 {
|
||
return *cfg.PackageCheckInterval, nil
|
||
}
|
||
case constants.TaskTypePollingProtect:
|
||
if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 {
|
||
return *cfg.ProtectCheckInterval, nil
|
||
}
|
||
}
|
||
// 匹配到配置但该任务类型间隔为 NULL，说明此卡不需要此类轮询
|
||
return 0, nil
|
||
}
|
||
|
||
return 0, nil
|
||
}
|
||
|
||
// getPollingConfigsFromCache 从 Redis HASH 缓存读取已启用的轮询配置列表，按 priority 升序排列
|
||
// 与 Scheduler.syncConfigsToRedis 写入格式对应
|
||
func (h *PollingHandler) getPollingConfigsFromCache(ctx context.Context) ([]*model.PollingConfig, error) {
|
||
key := constants.RedisPollingConfigsCacheKey()
|
||
result, err := h.redis.HGetAll(ctx, key).Result()
|
||
if err != nil || len(result) == 0 {
|
||
return nil, err
|
||
}
|
||
|
||
configs := make([]*model.PollingConfig, 0, len(result))
|
||
for _, jsonStr := range result {
|
||
var cfg model.PollingConfig
|
||
if err := sonic.UnmarshalString(jsonStr, &cfg); err != nil {
|
||
h.logger.Warn("解析轮询配置缓存失败", zap.Error(err))
|
||
continue
|
||
}
|
||
configs = append(configs, &cfg)
|
||
}
|
||
|
||
sort.Slice(configs, func(i, j int) bool {
|
||
return configs[i].Priority < configs[j].Priority
|
||
})
|
||
|
||
return configs, nil
|
||
}
|
||
|
||
// matchPollingConfig 判断卡是否满足配置的匹配条件
|
||
// 逻辑与 polling.Scheduler.matchConfigConditions 保持一致
|
||
func (h *PollingHandler) matchPollingConfig(cfg *model.PollingConfig, card *model.IotCard) bool {
|
||
if cfg.CardCondition != "" && cfg.CardCondition != h.cardCondition(card) {
|
||
return false
|
||
}
|
||
if cfg.CardCategory != "" && cfg.CardCategory != card.CardCategory {
|
||
return false
|
||
}
|
||
if cfg.CarrierID != nil && *cfg.CarrierID != card.CarrierID {
|
||
return false
|
||
}
|
||
return true
|
||
}
|
||
|
||
// cardCondition 根据卡的实名和网络状态返回条件字符串
|
||
// 逻辑与 polling.Scheduler.getCardCondition 保持一致
|
||
func (h *PollingHandler) cardCondition(card *model.IotCard) string {
|
||
if card.RealNameStatus != constants.RealNameStatusVerified {
|
||
return "not_real_name"
|
||
}
|
||
if card.NetworkStatus == constants.NetworkStatusOnline {
|
||
return "activated"
|
||
}
|
||
return "real_name"
|
||
}
|
||
|
||
// triggerPackageCheck 触发套餐检查
|
||
func (h *PollingHandler) triggerPackageCheck(ctx context.Context, cardID uint) {
|
||
key := constants.RedisPollingManualQueueKey(constants.TaskTypePollingPackage)
|
||
if err := h.redis.RPush(ctx, key, strconv.FormatUint(uint64(cardID), 10)).Err(); err != nil {
|
||
h.logger.Warn("触发套餐检查失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||
}
|
||
}
|
||
|
||
// updateStats 更新监控统计
|
||
func (h *PollingHandler) updateStats(ctx context.Context, taskType string, success bool, duration time.Duration) {
|
||
key := constants.RedisPollingStatsKey(taskType)
|
||
|
||
pipe := h.redis.Pipeline()
|
||
if success {
|
||
pipe.HIncrBy(ctx, key, "success_count_1h", 1)
|
||
} else {
|
||
pipe.HIncrBy(ctx, key, "failure_count_1h", 1)
|
||
}
|
||
pipe.HIncrBy(ctx, key, "total_duration_1h", duration.Milliseconds())
|
||
pipe.Expire(ctx, key, 2*time.Hour)
|
||
_, _ = pipe.Exec(ctx)
|
||
}
|
||
|
||
// insertDataUsageRecord 写入日流量缓冲到 Redis
|
||
// 增量 <= 0 时直接跳过；增量 > 0 时 INCRBYFLOAT 写 Redis + 48h TTL
|
||
func (h *PollingHandler) insertDataUsageRecord(ctx context.Context, cardID uint, incrementMB float64, checkTime time.Time) {
|
||
if incrementMB <= 0 {
|
||
return
|
||
}
|
||
|
||
dateStr := checkTime.Format("2006-01-02")
|
||
key := constants.RedisCardDailyTrafficKey(cardID, dateStr)
|
||
|
||
if err := h.redis.IncrByFloat(ctx, key, incrementMB).Err(); err != nil {
|
||
h.logger.Warn("写入日流量缓冲失败",
|
||
zap.Uint("card_id", cardID),
|
||
zap.Float64("increment_mb", incrementMB),
|
||
zap.Error(err))
|
||
return
|
||
}
|
||
h.redis.Expire(ctx, key, 48*time.Hour)
|
||
}
|
||
|
||
// updateCardCache 更新卡缓存
|
||
func (h *PollingHandler) updateCardCache(ctx context.Context, cardID uint, updates map[string]any) {
|
||
key := constants.RedisPollingCardInfoKey(cardID)
|
||
|
||
// 转换为 []any 用于 HSet
|
||
args := make([]any, 0, len(updates)*2)
|
||
for k, v := range updates {
|
||
args = append(args, k, v)
|
||
}
|
||
|
||
if len(args) > 0 {
|
||
if err := h.redis.HSet(ctx, key, args...).Err(); err != nil {
|
||
h.logger.Warn("更新卡缓存失败",
|
||
zap.Uint("card_id", cardID),
|
||
zap.Error(err))
|
||
}
|
||
}
|
||
}
|
||
|
||
// getCardWithCache 优先从 Redis 缓存获取卡信息，miss 时查 DB
|
||
// 大幅减少数据库查询，避免高并发时连接池耗尽
|
||
func (h *PollingHandler) getCardWithCache(ctx context.Context, cardID uint) (*model.IotCard, error) {
|
||
key := constants.RedisPollingCardInfoKey(cardID)
|
||
|
||
// 尝试从 Redis 读取
|
||
result, err := h.redis.HGetAll(ctx, key).Result()
|
||
if err == nil && len(result) > 0 {
|
||
// 缓存命中，构建卡对象
|
||
card := &model.IotCard{}
|
||
card.ID = cardID
|
||
|
||
if v, ok := result["iccid"]; ok {
|
||
card.ICCID = v
|
||
}
|
||
if v, ok := result["card_category"]; ok {
|
||
card.CardCategory = v
|
||
}
|
||
if v, ok := result["real_name_status"]; ok {
|
||
if status, err := strconv.Atoi(v); err == nil {
|
||
card.RealNameStatus = status
|
||
}
|
||
}
|
||
if v, ok := result["network_status"]; ok {
|
||
if status, err := strconv.Atoi(v); err == nil {
|
||
card.NetworkStatus = status
|
||
}
|
||
}
|
||
if v, ok := result["carrier_id"]; ok {
|
||
if id, err := strconv.ParseUint(v, 10, 64); err == nil {
|
||
card.CarrierID = uint(id)
|
||
}
|
||
}
|
||
if v, ok := result["current_month_usage_mb"]; ok {
|
||
if usage, err := strconv.ParseFloat(v, 64); err == nil {
|
||
card.CurrentMonthUsageMB = usage
|
||
}
|
||
}
|
||
if v, ok := result["last_gateway_reading_mb"]; ok {
|
||
if reading, err := strconv.ParseFloat(v, 64); err == nil {
|
||
card.LastGatewayReadingMB = reading
|
||
}
|
||
}
|
||
if v, ok := result["data_usage_mb"]; ok {
|
||
if usage, err := strconv.ParseInt(v, 10, 64); err == nil {
|
||
card.DataUsageMB = usage
|
||
}
|
||
}
|
||
if v, ok := result["series_id"]; ok {
|
||
if id, err := strconv.ParseUint(v, 10, 64); err == nil {
|
||
seriesID := uint(id)
|
||
card.SeriesID = &seriesID
|
||
}
|
||
}
|
||
if v, ok := result["stop_reason"]; ok {
|
||
card.StopReason = v
|
||
}
|
||
|
||
return card, nil
|
||
}
|
||
|
||
// 缓存 miss，查询数据库
|
||
card, err := h.iotCardStore.GetByID(ctx, cardID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 异步写入缓存
|
||
go func() {
|
||
cacheCtx := context.Background()
|
||
cacheData := map[string]any{
|
||
"id": card.ID,
|
||
"iccid": card.ICCID,
|
||
"card_category": card.CardCategory,
|
||
"real_name_status": card.RealNameStatus,
|
||
"network_status": card.NetworkStatus,
|
||
"carrier_id": card.CarrierID,
|
||
"current_month_usage_mb": card.CurrentMonthUsageMB,
|
||
"last_gateway_reading_mb": card.LastGatewayReadingMB,
|
||
"data_usage_mb": card.DataUsageMB,
|
||
"stop_reason": card.StopReason,
|
||
"cached_at": time.Now().Unix(),
|
||
}
|
||
if card.SeriesID != nil {
|
||
cacheData["series_id"] = *card.SeriesID
|
||
}
|
||
pipe := h.redis.Pipeline()
|
||
pipe.HSet(cacheCtx, key, cacheData)
|
||
pipe.Expire(cacheCtx, key, 7*24*time.Hour)
|
||
_, _ = pipe.Exec(cacheCtx)
|
||
}()
|
||
|
||
return card, nil
|
||
}
|
||
|
||
// HandleProtectConsistencyCheck 保护期一致性检查
|
||
// 检查绑定设备（is_standalone=false）且已实名（real_name_status=1）的卡
|
||
// stop 保护期 + 开机 → 调网关停机；start 保护期 + 停机 → 调网关复机
|
||
func (h *PollingHandler) HandleProtectConsistencyCheck(ctx context.Context, t *asynq.Task) error {
|
||
var payload PollingTaskPayload
|
||
if err := sonic.Unmarshal(t.Payload(), &payload); err != nil {
|
||
h.logger.Error("解析保护期检查任务载荷失败", zap.Error(err))
|
||
return nil
|
||
}
|
||
|
||
cardID, err := strconv.ParseUint(payload.CardID, 10, 64)
|
||
if err != nil {
|
||
h.logger.Error("解析卡ID失败", zap.String("card_id", payload.CardID), zap.Error(err))
|
||
return nil
|
||
}
|
||
|
||
// 查询卡信息
|
||
var card model.IotCard
|
||
if err := h.db.WithContext(ctx).Where("id = ?", cardID).First(&card).Error; err != nil {
|
||
if err == gorm.ErrRecordNotFound {
|
||
return nil
|
||
}
|
||
h.logger.Error("查询卡信息失败", zap.Uint64("card_id", cardID), zap.Error(err))
|
||
return nil
|
||
}
|
||
|
||
// 未绑设备（独立卡），跳过
|
||
if card.IsStandalone {
|
||
return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect)
|
||
}
|
||
|
||
// 未实名，跳过
|
||
if card.RealNameStatus != constants.RealNameStatusVerified {
|
||
return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect)
|
||
}
|
||
|
||
// 查绑定设备
|
||
binding, err := h.deviceSimBindingStore.GetActiveBindingByCardID(ctx, card.ID)
|
||
if err != nil || binding == nil {
|
||
return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect)
|
||
}
|
||
deviceID := binding.DeviceID
|
||
|
||
// 检查 stop 保护期：设备处于停机保护期，但卡是开机状态 → 需要停机
|
||
stopProtect := h.redis.Exists(ctx, constants.RedisDeviceProtectKey(deviceID, "stop")).Val() > 0
|
||
if stopProtect && card.NetworkStatus == constants.NetworkStatusOnline {
|
||
h.logger.Info("保护期一致性检查：停机保护期内发现开机卡，执行停机",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Uint("device_id", deviceID))
|
||
|
||
if h.gatewayClient != nil {
|
||
h.logger.Info("调用网关停机（保护期）",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
if err := h.gatewayClient.StopCard(ctx, &gateway.CardOperationReq{CardNo: card.ICCID}); err != nil {
|
||
h.logger.Error("保护期一致性停机失败",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Error(err))
|
||
return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect)
|
||
}
|
||
h.logger.Info("网关停机成功（保护期）",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
}
|
||
|
||
h.db.Model(&model.IotCard{}).Where("id = ?", card.ID).Updates(map[string]any{
|
||
"network_status": constants.NetworkStatusOffline,
|
||
"stopped_at": time.Now(),
|
||
"stop_reason": constants.StopReasonProtectPeriod,
|
||
})
|
||
h.updateCardCache(ctx, card.ID, map[string]any{"network_status": constants.NetworkStatusOffline})
|
||
return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect)
|
||
}
|
||
|
||
// 检查 start 保护期：设备处于复机保护期，但卡是停机状态 → 需要复机
|
||
startProtect := h.redis.Exists(ctx, constants.RedisDeviceProtectKey(deviceID, "start")).Val() > 0
|
||
if startProtect && card.NetworkStatus == constants.NetworkStatusOffline {
|
||
h.logger.Info("保护期一致性检查：复机保护期内发现停机卡，执行复机",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Uint("device_id", deviceID))
|
||
|
||
if h.gatewayClient != nil {
|
||
h.logger.Info("调用网关复机（保护期）",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
if err := h.gatewayClient.StartCard(ctx, &gateway.CardOperationReq{CardNo: card.ICCID}); err != nil {
|
||
h.logger.Error("保护期一致性复机失败",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Error(err))
|
||
return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect)
|
||
}
|
||
h.logger.Info("网关复机成功（保护期）",
|
||
zap.Uint("card_id", card.ID),
|
||
zap.String("iccid", card.ICCID))
|
||
}
|
||
|
||
h.db.Model(&model.IotCard{}).Where("id = ?", card.ID).Updates(map[string]any{
|
||
"network_status": constants.NetworkStatusOnline,
|
||
"resumed_at": time.Now(),
|
||
"stop_reason": "",
|
||
})
|
||
h.updateCardCache(ctx, card.ID, map[string]any{"network_status": constants.NetworkStatusOnline})
|
||
}
|
||
|
||
return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect)
|
||
}
|
||
|
||
// triggerFirstRealnameActivation 任务 21.3-21.4: 首次实名后触发套餐激活
|
||
func (h *PollingHandler) triggerFirstRealnameActivation(ctx context.Context, cardID uint) {
|
||
// 任务 21.3: 查询该卡是否有待激活套餐
|
||
// WHERE pending_realname_activation=true AND status=0 AND iot_card_id=?
|
||
var pendingPackages []model.PackageUsage
|
||
err := h.db.WithContext(ctx).
|
||
Where("iot_card_id = ?", cardID).
|
||
Where("pending_realname_activation = ?", true).
|
||
Where("status = ?", constants.PackageUsageStatusPending).
|
||
Find(&pendingPackages).Error
|
||
|
||
if err != nil {
|
||
h.logger.Warn("查询待激活套餐失败",
|
||
zap.Uint("card_id", cardID),
|
||
zap.Error(err))
|
||
return
|
||
}
|
||
|
||
if len(pendingPackages) == 0 {
|
||
h.logger.Debug("无待激活套餐",
|
||
zap.Uint("card_id", cardID))
|
||
return
|
||
}
|
||
|
||
h.logger.Info("发现待激活套餐",
|
||
zap.Uint("card_id", cardID),
|
||
zap.Int("count", len(pendingPackages)))
|
||
|
||
// 任务 21.4: 提交 Asynq 任务激活套餐
|
||
for _, pkg := range pendingPackages {
|
||
payload := map[string]any{
|
||
"package_usage_id": pkg.ID,
|
||
"carrier_type": "iot_card",
|
||
"carrier_id": cardID,
|
||
"activation_type": "realname",
|
||
"timestamp": time.Now().Unix(),
|
||
}
|
||
|
||
payloadBytes, err := sonic.Marshal(payload)
|
||
if err != nil {
|
||
h.logger.Warn("序列化激活任务载荷失败",
|
||
zap.Uint("package_usage_id", pkg.ID),
|
||
zap.Error(err))
|
||
continue
|
||
}
|
||
|
||
task := asynq.NewTask(constants.TaskTypePackageFirstActivation, payloadBytes,
|
||
asynq.MaxRetry(3),
|
||
asynq.Timeout(30*time.Second),
|
||
asynq.Queue(constants.QueueDefault),
|
||
)
|
||
|
||
if _, err := h.asynqClient.EnqueueContext(ctx, task); err != nil {
|
||
h.logger.Warn("提交首次实名激活任务失败",
|
||
zap.Uint("package_usage_id", pkg.ID),
|
||
zap.Error(err))
|
||
continue
|
||
}
|
||
|
||
h.logger.Info("已提交首次实名激活任务",
|
||
zap.Uint("package_usage_id", pkg.ID),
|
||
zap.Uint("card_id", cardID))
|
||
}
|
||
}
|