From 69ee754c193ba6e77947f8ec2923db024e6604d3 Mon Sep 17 00:00:00 2001 From: huang Date: Tue, 31 Mar 2026 19:15:45 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=A1=A5=E5=85=A8=E7=BD=91=E5=85=B3?= =?UTF-8?q?=E5=81=9C=E5=A4=8D=E6=9C=BA=E8=B0=83=E7=94=A8=E7=9A=84=E5=85=B3?= =?UTF-8?q?=E9=94=AE=E6=97=A5=E5=BF=97=EF=BC=88=E8=B0=83=E7=94=A8=E5=89=8D?= =?UTF-8?q?/=E6=88=90=E5=8A=9F=E5=90=8E=E5=9D=87=E8=BE=93=E5=87=BA=20INFO?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/service/device/service.go | 12 ++ internal/service/iot_card/gateway_service.go | 21 ++- .../service/iot_card/stop_resume_service.go | 16 ++ internal/task/polling_handler.go | 164 +++++++++++++++--- 4 files changed, 189 insertions(+), 24 deletions(-) diff --git a/internal/service/device/service.go b/internal/service/device/service.go index 430baee..48ba070 100644 --- a/internal/service/device/service.go +++ b/internal/service/device/service.go @@ -835,6 +835,9 @@ func (s *Service) StopDevice(ctx context.Context, deviceID uint) (*dto.DeviceSus } if s.gatewayClient != nil { + log.Info("调用网关停机（设备）", + zap.Uint("device_id", deviceID), + zap.String("iccid", card.ICCID)) if gwErr := s.gatewayClient.StopCard(ctx, &gateway.CardOperationReq{CardNo: card.ICCID}); gwErr != nil { log.Error("设备停机-调网关停机失败", zap.Uint("device_id", deviceID), @@ -846,6 +849,9 @@ func (s *Service) StopDevice(ctx context.Context, deviceID uint) (*dto.DeviceSus }) continue } + log.Info("网关停机成功（设备）", + zap.Uint("device_id", deviceID), + zap.String("iccid", card.ICCID)) } now := time.Now() @@ -936,6 +942,9 @@ func (s *Service) StartDevice(ctx context.Context, deviceID uint) error { } if s.gatewayClient != nil { + log.Info("调用网关复机（设备）", + zap.Uint("device_id", deviceID), + zap.String("iccid", card.ICCID)) if gwErr := s.gatewayClient.StartCard(ctx, &gateway.CardOperationReq{CardNo: card.ICCID}); gwErr != nil { log.Error("设备复机-调网关复机失败", zap.Uint("device_id", deviceID), @@ -944,6 +953,9 @@ func (s *Service) StartDevice(ctx context.Context, deviceID uint) error { lastErr = gwErr continue } + log.Info("网关复机成功（设备）", + zap.Uint("device_id", deviceID), + zap.String("iccid", card.ICCID)) } now := time.Now() diff --git a/internal/service/iot_card/gateway_service.go b/internal/service/iot_card/gateway_service.go index 41a51f5..ba8b6fd 100644 --- a/internal/service/iot_card/gateway_service.go +++ b/internal/service/iot_card/gateway_service.go @@ -5,6 +5,7 @@ import ( "github.com/break/junhong_cmp_fiber/internal/gateway" "github.com/break/junhong_cmp_fiber/pkg/errors" + "go.uber.org/zap" ) // validateCardAccess 通过 ICCID 验证卡存在且当前用户有权限访问 @@ -62,9 +63,15 @@ func (s *Service) GatewayStopCard(ctx context.Context, iccid string) error { if err := s.validateCardAccess(ctx, iccid); err != nil { return err } - return s.gatewayClient.StopCard(ctx, &gateway.CardOperationReq{ + s.logger.Info("调用网关停机（手动）", zap.String("iccid", iccid)) + if err := s.gatewayClient.StopCard(ctx, &gateway.CardOperationReq{ CardNo: iccid, - }) + }); err != nil { + s.logger.Error("网关停机失败（手动）", zap.String("iccid", iccid), zap.Error(err)) + return err + } + s.logger.Info("网关停机成功（手动）", zap.String("iccid", iccid)) + return nil } // GatewayStartCard 恢复卡服务（通过 Gateway API） @@ -72,7 +79,13 @@ func (s *Service) GatewayStartCard(ctx context.Context, iccid string) error { if err := s.validateCardAccess(ctx, iccid); err != nil { return err } - return s.gatewayClient.StartCard(ctx, &gateway.CardOperationReq{ + s.logger.Info("调用网关复机（手动）", zap.String("iccid", iccid)) + if err := s.gatewayClient.StartCard(ctx, &gateway.CardOperationReq{ CardNo: iccid, - }) + }); err != nil { + s.logger.Error("网关复机失败（手动）", zap.String("iccid", iccid), zap.Error(err)) + return err + } + s.logger.Info("网关复机成功（手动）", zap.String("iccid", iccid)) + return nil } diff --git a/internal/service/iot_card/stop_resume_service.go b/internal/service/iot_card/stop_resume_service.go index bd3addb..0405171 100644 --- a/internal/service/iot_card/stop_resume_service.go +++ b/internal/service/iot_card/stop_resume_service.go @@ -237,6 +237,10 @@ func (s *StopResumeService) stopCardWithRetry(ctx context.Context, card *model.I return errors.New(errors.CodeInternalError, "Gateway 未配置，停复机操作不可用") } + s.logger.Info("调用网关停机", + zap.Uint("card_id", card.ID), + zap.String("iccid", card.ICCID)) + var lastErr error for i := 0; i < s.maxRetries; i++ { if i > 0 { @@ -250,12 +254,16 @@ func (s *StopResumeService) stopCardWithRetry(ctx context.Context, card *model.I CardNo: card.ICCID, }) if err == nil { + s.logger.Info("网关停机成功", + zap.Uint("card_id", card.ID), + zap.String("iccid", card.ICCID)) return nil } lastErr = err s.logger.Warn("调用停机接口失败，准备重试", zap.Int("attempt", i+1), + zap.String("iccid", card.ICCID), zap.Error(err)) } @@ -268,6 +276,10 @@ func (s *StopResumeService) resumeCardWithRetry(ctx context.Context, card *model return errors.New(errors.CodeInternalError, "Gateway 未配置，停复机操作不可用") } + s.logger.Info("调用网关复机", + zap.Uint("card_id", card.ID), + zap.String("iccid", card.ICCID)) + var lastErr error for i := 0; i < s.maxRetries; i++ { if i > 0 { @@ -281,12 +293,16 @@ func (s *StopResumeService) resumeCardWithRetry(ctx context.Context, card *model CardNo: card.ICCID, }) if err == nil { + s.logger.Info("网关复机成功", + zap.Uint("card_id", card.ID), + zap.String("iccid", card.ICCID)) return nil } lastErr = err s.logger.Warn("调用复机接口失败，准备重试", zap.Int("attempt", i+1), + zap.String("iccid", card.ICCID), zap.Error(err)) } diff --git a/internal/task/polling_handler.go b/internal/task/polling_handler.go index fb24d97..0db6661 100644 --- a/internal/task/polling_handler.go +++ b/internal/task/polling_handler.go @@ -2,6 +2,7 @@ package task import ( "context" + "sort" "strconv" "strings" "time" @@ -34,6 +35,7 @@ type PollingHandler struct { gatewayClient *gateway.Client iotCardStore *postgres.IotCardStore concurrencyStore *postgres.PollingConcurrencyConfigStore + pollingConfigStore *postgres.PollingConfigStore // 用于 requeueCard 读取配置间隔 deviceSimBindingStore *postgres.DeviceSimBindingStore packageUsageStore *postgres.PackageUsageStore usageService *packagepkg.UsageService @@ -56,6 +58,7 @@ func NewPollingHandler( 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, @@ -350,6 +353,9 @@ func (h *PollingHandler) stopCardByUsageExhausted(ctx context.Context, card *mod // 调用 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 { @@ -359,12 +365,15 @@ func (h *PollingHandler) stopCardByUsageExhausted(ctx context.Context, card *mod 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, // 停机 + "network_status": 0, "stopped_at": now, "stop_reason": constants.StopReasonTrafficExhausted, "updated_at": now, @@ -670,6 +679,9 @@ func (h *PollingHandler) stopCards(ctx context.Context, cards []*model.IotCard, 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, }) @@ -678,7 +690,6 @@ func (h *PollingHandler) stopCards(ctx context.Context, cards []*model.IotCard, zap.Uint("card_id", card.ID), zap.String("iccid", card.ICCID), zap.Error(err)) - // 继续处理其他卡，不中断 continue } @@ -764,27 +775,24 @@ func (h *PollingHandler) releaseConcurrency(ctx context.Context, taskType string } // requeueCard 重新将卡加入队列 +// 从 PollingConfig 读取实际配置间隔；无匹配配置或间隔为 NULL 时不入队（停止轮询该卡） func (h *PollingHandler) requeueCard(ctx context.Context, cardID uint, taskType string) error { - // 获取配置中的检查间隔 - var intervalSeconds int - - switch taskType { - case constants.TaskTypePollingRealname: - intervalSeconds = 300 // 默认 5 分钟 - case constants.TaskTypePollingCarddata: - intervalSeconds = 600 // 默认 10 分钟 - case constants.TaskTypePollingPackage: - intervalSeconds = 600 // 默认 10 分钟 - case constants.TaskTypePollingProtect: - intervalSeconds = 300 // 默认 5 分钟 - default: + 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 } - // 计算下次检查时间 - nextCheck := time.Now().Add(time.Duration(intervalSeconds) * time.Second) - - // 确定队列 key var queueKey string switch taskType { case constants.TaskTypePollingRealname: @@ -795,15 +803,117 @@ func (h *PollingHandler) requeueCard(ctx context.Context, cardID uint, taskType queueKey = constants.RedisPollingQueuePackageKey() case constants.TaskTypePollingProtect: queueKey = constants.RedisPollingQueueProtectKey() + default: + return nil } - // 添加到队列 + nextCheck := time.Now().Add(time.Duration(intervalSeconds) * time.Second) return h.redis.ZAdd(ctx, queueKey, redis.Z{ Score: float64(nextCheck.Unix()), Member: strconv.FormatUint(uint64(cardID), 10), }).Err() } +// 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: + // protect 任务暂无专用间隔字段，使用 realname 间隔兜底 + if cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0 { + return *cfg.RealnameCheckInterval, 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) @@ -1009,12 +1119,19 @@ func (h *PollingHandler) HandleProtectConsistencyCheck(ctx context.Context, t *a 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 nil } + 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{ @@ -1035,12 +1152,19 @@ func (h *PollingHandler) HandleProtectConsistencyCheck(ctx context.Context, t *a 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 nil } + 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{