diff --git a/internal/polling/scheduler.go b/internal/polling/scheduler.go index 386043d..29b41b6 100644 --- a/internal/polling/scheduler.go +++ b/internal/polling/scheduler.go @@ -538,6 +538,7 @@ func (s *Scheduler) initCardsBatch(ctx context.Context, cards []*model.IotCard) "real_name_status": card.RealNameStatus, "network_status": card.NetworkStatus, "carrier_id": card.CarrierID, + "stop_reason": card.StopReason, "cached_at": now.Unix(), } pipe.HSet(ctx, cacheKey, cacheData) @@ -689,6 +690,7 @@ func (s *Scheduler) cacheCardInfo(ctx context.Context, card *model.IotCard, cach "real_name_status": card.RealNameStatus, "network_status": card.NetworkStatus, "carrier_id": card.CarrierID, + "stop_reason": card.StopReason, "cached_at": cachedAt.Unix(), } diff --git a/internal/task/polling_handler.go b/internal/task/polling_handler.go index 0db6661..d46f1e4 100644 --- a/internal/task/polling_handler.go +++ b/internal/task/polling_handler.go @@ -273,21 +273,13 @@ func (h *PollingHandler) HandleCarddataCheck(ctx context.Context, t *asynq.Task) "last_gateway_reading_mb": gatewayFlowMB, }) - // 任务 18.3: 调用 UsageService.DeductDataUsage 进行流量扣减 + // 调用 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)) - - // 任务 18.4: 检查是否需要停机(所有套餐用完) - if h.shouldStopCard(ctx, uint(cardID)) { - h.logger.Warn("所有套餐流量已用完,触发停机", - zap.Uint64("card_id", cardID)) - h.stopCardByUsageExhausted(ctx, card) - } } else { h.logger.Info("套餐流量扣减成功", zap.Uint64("card_id", cardID), @@ -295,6 +287,9 @@ func (h *PollingHandler) HandleCarddataCheck(ctx context.Context, t *asynq.Task) } } + // 独立停机/复机检查(不依赖流量增量,每次轮询都执行) + h.checkStopResume(ctx, card) + // 更新监控统计 h.updateStats(ctx, constants.TaskTypePollingCarddata, true, time.Since(startTime)) @@ -329,19 +324,37 @@ func isResetWindow(now time.Time, resetDay int) bool { return today == prevDay } -// shouldStopCard 任务 18.4: 检查是否应该停机(所有套餐用完) +// shouldStopCard 检查是否应该停机(无有效套餐) +// 有效套餐 = 待生效(0) 或 生效中(1),与 StopResumeService.hasAvailablePackage 对齐 func (h *PollingHandler) shouldStopCard(ctx context.Context, cardID uint) bool { - // 查询是否还有生效中的套餐 - var activeCount int64 + var count int64 if err := h.db.WithContext(ctx).Model(&model.PackageUsage{}). - Where("iot_card_id = ? AND status = ?", cardID, constants.PackageUsageStatusActive). - Count(&activeCount).Error; err != nil { - h.logger.Warn("查询生效套餐失败", zap.Uint("card_id", cardID), zap.Error(err)) + 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 activeCount == 0 + 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: 流量耗尽停机 @@ -384,9 +397,9 @@ func (h *PollingHandler) stopCardByUsageExhausted(ctx context.Context, card *mod h.logger.Error("更新卡状态失败", zap.Uint("card_id", card.ID), zap.Error(err)) } - // 更新 Redis 缓存 h.updateCardCache(ctx, card.ID, map[string]any{ "network_status": 0, + "stop_reason": constants.StopReasonTrafficExhausted, }) h.logger.Warn("卡已停机(套餐流量耗尽)", @@ -394,6 +407,90 @@ func (h *PollingHandler) stopCardByUsageExhausted(ctx context.Context, card *mod 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) { @@ -1030,6 +1127,9 @@ func (h *PollingHandler) getCardWithCache(ctx context.Context, cardID uint) (*mo card.SeriesID = &seriesID } } + if v, ok := result["stop_reason"]; ok { + card.StopReason = v + } return card, nil } @@ -1053,6 +1153,7 @@ func (h *PollingHandler) getCardWithCache(ctx context.Context, cardID uint) (*mo "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 { diff --git a/migrations/000101_add_activated_polling_config.down.sql b/migrations/000101_add_activated_polling_config.down.sql new file mode 100644 index 0000000..69ac25a --- /dev/null +++ b/migrations/000101_add_activated_polling_config.down.sql @@ -0,0 +1,2 @@ +DELETE FROM tb_polling_config +WHERE config_name = '已激活在线卡' AND card_condition = 'activated'; diff --git a/migrations/000101_add_activated_polling_config.up.sql b/migrations/000101_add_activated_polling_config.up.sql new file mode 100644 index 0000000..21d2784 --- /dev/null +++ b/migrations/000101_add_activated_polling_config.up.sql @@ -0,0 +1,12 @@ +-- 新增 activated 轮询配置:覆盖已实名+在线的卡 +-- 修复问题:activated 状态的卡无匹配配置导致不被轮询,套餐过期后无法触发停机 +INSERT INTO tb_polling_config ( + config_name, card_condition, card_category, carrier_id, + priority, realname_check_interval, carddata_check_interval, package_check_interval, + status, created_at, updated_at +) VALUES ( + '已激活在线卡', 'activated', 'normal', 1566, + 2, 30, 60, 60, + 1, NOW(), NOW() +) +ON CONFLICT DO NOTHING;