diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 2b6b1c7..87811f0 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -144,6 +144,9 @@ func main() { dataResetHandler := polling.NewDataResetHandler(workerResult.Services.ResetService, appLogger) scheduler.SetResetService(dataResetHandler) + // 注入停复机回调到调度器（套餐过期后主动触发停机） + scheduler.SetStopResumeCallback(workerResult.Services.StopResumeService) + if err := scheduler.Start(ctx); err != nil { appLogger.Error("启动轮询调度器失败", zap.Error(err)) } else { diff --git a/internal/bootstrap/services.go b/internal/bootstrap/services.go index ef65037..18e1d5b 100644 --- a/internal/bootstrap/services.go +++ b/internal/bootstrap/services.go @@ -107,7 +107,7 @@ func initServices(s *stores, deps *Dependencies) *services { iotCard := iotCardSvc.New(deps.DB, s.IotCard, s.Shop, s.AssetAllocationRecord, s.ShopPackageAllocation, s.ShopSeriesAllocation, s.PackageSeries, deps.GatewayClient, deps.Logger) iotCard.SetPollingCallback(polling.NewAPICallback(deps.Redis, deps.Logger)) // 注入流量扣减回调，使手动刷新资产时能触发套餐流量扣减 - usageService := packageSvc.NewUsageService(deps.DB, deps.Redis, s.PackageUsage, s.PackageUsageDailyRecord, deps.Logger) + usageService := packageSvc.NewUsageService(deps.DB, deps.Redis, s.PackageUsage, s.PackageUsageDailyRecord, s.DeviceSimBinding, deps.Logger) iotCard.SetDataDeductor(usageService) // 创建支付配置服务（Order 和 Recharge 依赖） diff --git a/internal/bootstrap/worker_services.go b/internal/bootstrap/worker_services.go index 0c53fb1..e3263d3 100644 --- a/internal/bootstrap/worker_services.go +++ b/internal/bootstrap/worker_services.go @@ -47,6 +47,7 @@ func initWorkerServices(stores *queue.WorkerStores, deps *WorkerDependencies) *q deps.Redis, stores.PackageUsage, stores.PackageUsageDailyRecord, + stores.DeviceSimBinding, deps.Logger, ) @@ -116,6 +117,7 @@ func initWorkerServices(stores *queue.WorkerStores, deps *WorkerDependencies) *q ResetService: resetService, AlertService: alertService, CleanupService: cleanupService, + StopResumeService: stopResumeService, OrderExpirer: orderService, } } diff --git a/internal/model/iot_card.go b/internal/model/iot_card.go index df8d5cc..1b8bc6a 100644 --- a/internal/model/iot_card.go +++ b/internal/model/iot_card.go @@ -36,6 +36,7 @@ type IotCard struct { EnablePolling bool `gorm:"column:enable_polling;type:boolean;default:true;comment:是否参与轮询 true-参与 false-不参与" json:"enable_polling"` LastDataCheckAt *time.Time `gorm:"column:last_data_check_at;comment:最后一次流量检查时间" json:"last_data_check_at"` LastRealNameCheckAt *time.Time `gorm:"column:last_real_name_check_at;comment:最后一次实名检查时间" json:"last_real_name_check_at"` + LastProtectCheckAt *time.Time `gorm:"column:last_protect_check_at;comment:上次保护期一致性检查时间" json:"last_protect_check_at"` LastSyncTime *time.Time `gorm:"column:last_sync_time;comment:最后一次与Gateway同步时间" json:"last_sync_time"` SeriesID *uint `gorm:"column:series_id;index;comment:套餐系列ID(关联PackageSeries)" json:"series_id,omitempty"` FirstCommissionPaid bool `gorm:"column:first_commission_paid;type:boolean;default:false;comment:一次性佣金是否已发放(废弃,使用按系列追踪)" json:"first_commission_paid"` diff --git a/internal/model/polling.go b/internal/model/polling.go index d51c755..cfd6268 100644 --- a/internal/model/polling.go +++ b/internal/model/polling.go @@ -15,6 +15,7 @@ type PollingConfig struct { RealnameCheckInterval *int `gorm:"column:realname_check_interval;comment:实名检查间隔（秒），NULL表示不检查" json:"realname_check_interval"` CarddataCheckInterval *int `gorm:"column:carddata_check_interval;comment:流量检查间隔（秒），NULL表示不检查" json:"carddata_check_interval"` PackageCheckInterval *int `gorm:"column:package_check_interval;comment:套餐检查间隔（秒），NULL表示不检查" json:"package_check_interval"` + ProtectCheckInterval *int `gorm:"column:protect_check_interval;comment:保护期一致性检查间隔（秒），NULL=不参与" json:"protect_check_interval"` Status int16 `gorm:"column:status;type:smallint;not null;default:1;comment:状态：0-禁用，1-启用" json:"status"` Description string `gorm:"column:description;type:text;comment:配置说明" json:"description"` CreatedAt time.Time `gorm:"column:created_at;not null;default:CURRENT_TIMESTAMP;comment:创建时间" json:"created_at"` diff --git a/internal/polling/package_activation_handler.go b/internal/polling/package_activation_handler.go index 546b228..5f81520 100644 --- a/internal/polling/package_activation_handler.go +++ b/internal/polling/package_activation_handler.go @@ -19,12 +19,14 @@ import ( // PackageActivationHandler 套餐激活检查处理器 // 任务 19: 处理主套餐过期、加油包级联失效、待生效主套餐激活 type PackageActivationHandler struct { - db *gorm.DB - redis *redis.Client - queueClient *asynq.Client - packageUsageStore *postgres.PackageUsageStore - activationService *packagepkg.ActivationService - logger *zap.Logger + db *gorm.DB + redis *redis.Client + queueClient *asynq.Client + packageUsageStore *postgres.PackageUsageStore + deviceSimBinding *postgres.DeviceSimBindingStore + activationService *packagepkg.ActivationService + stopResumeCallback packagepkg.StopResumeCallback // 停复机回调，用于套餐过期后主动停机 + logger *zap.Logger } // PackageActivationPayload 套餐激活任务载荷 @@ -42,15 +44,18 @@ func NewPackageActivationHandler( redis *redis.Client, queueClient *asynq.Client, activationService *packagepkg.ActivationService, + stopResumeCallback packagepkg.StopResumeCallback, logger *zap.Logger, ) *PackageActivationHandler { return &PackageActivationHandler{ - db: db, - redis: redis, - queueClient: queueClient, - packageUsageStore: postgres.NewPackageUsageStore(db, redis), - activationService: activationService, - logger: logger, + db: db, + redis: redis, + queueClient: queueClient, + packageUsageStore: postgres.NewPackageUsageStore(db, redis), + deviceSimBinding: postgres.NewDeviceSimBindingStore(db, redis), + activationService: activationService, + stopResumeCallback: stopResumeCallback, + logger: logger, } } @@ -135,7 +140,6 @@ func (h *PackageActivationHandler) processExpiredPackage(ctx context.Context, pk zap.String("carrier_type", carrierType), zap.Uint("carrier_id", carrierID), zap.Error(err)) - // 不返回错误，继续处理 } if err := h.updateCarrierSuspendedStatus(ctx, tx, carrierType, carrierID); err != nil { @@ -144,6 +148,9 @@ func (h *PackageActivationHandler) processExpiredPackage(ctx context.Context, pk zap.Uint("carrier_id", carrierID), zap.Error(err)) } + + // 套餐过期后主动触发停机，消除依赖 carddata 轮询兜底的 60s 延迟窗口 + h.triggerStopAfterExpiry(ctx, carrierType, carrierID) } return nil @@ -299,6 +306,45 @@ func (h *PackageActivationHandler) activateNextPackage(ctx context.Context, tx * return h.enqueueActivationTask(ctx, nextPkg.ID, carrierType, carrierID, "queue") } +// triggerStopAfterExpiry 套餐过期后异步触发停机检查 +// 仅在确认无后续生效套餐时有效；CheckAndStopCard 内部有幂等保护，重复调用安全 +func (h *PackageActivationHandler) triggerStopAfterExpiry(ctx context.Context, carrierType string, carrierID uint) { + if h.stopResumeCallback == nil { + return + } + + if carrierType == "iot_card" { + go func() { + if err := h.stopResumeCallback.CheckAndStopCard(context.Background(), carrierID); err != nil { + h.logger.Error("套餐过期后停机失败", + zap.Uint("card_id", carrierID), + zap.Error(err)) + } + }() + return + } + + if carrierType == "device" { + bindings, err := h.deviceSimBinding.ListByDeviceID(ctx, carrierID) + if err != nil { + h.logger.Error("查询设备绑定卡失败", + zap.Uint("device_id", carrierID), + zap.Error(err)) + return + } + for _, b := range bindings { + cardID := b.IotCardID + go func(cID uint) { + if err := h.stopResumeCallback.CheckAndStopCard(context.Background(), cID); err != nil { + h.logger.Error("套餐过期后停机失败", + zap.Uint("card_id", cID), + zap.Error(err)) + } + }(cardID) + } + } +} + // enqueueActivationTask 提交套餐激活任务到 Asynq func (h *PackageActivationHandler) enqueueActivationTask(ctx context.Context, packageUsageID uint, carrierType string, carrierID uint, activationType string) error { payload := PackageActivationPayload{ diff --git a/internal/polling/scheduler.go b/internal/polling/scheduler.go index 29b41b6..934adfc 100644 --- a/internal/polling/scheduler.go +++ b/internal/polling/scheduler.go @@ -13,6 +13,7 @@ import ( "gorm.io/gorm" "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" ) @@ -99,7 +100,7 @@ func NewScheduler( configStore: postgres.NewPollingConfigStore(db), iotCardStore: postgres.NewIotCardStore(db, redisClient), concurrencyStore: postgres.NewPollingConcurrencyConfigStore(db), - packageActivationHandler: NewPackageActivationHandler(db, redisClient, queueClient, nil, logger), + packageActivationHandler: NewPackageActivationHandler(db, redisClient, queueClient, nil, nil, logger), dataResetHandler: NewDataResetHandler(nil, logger), // ResetService 需要通过 SetResetService 注入 initProgress: &InitProgress{ Status: "pending", @@ -529,6 +530,14 @@ func (s *Scheduler) initCardsBatch(ctx context.Context, cards []*model.IotCard) }) } + if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 { + nextCheck := s.calculateNextCheckTime(card.LastProtectCheckAt, *cfg.ProtectCheckInterval) + pipe.ZAdd(ctx, constants.RedisPollingQueueProtectKey(), redis.Z{ + Score: float64(nextCheck.Unix()), + Member: card.ID, + }) + } + // 缓存卡信息到 Redis cacheKey := constants.RedisPollingCardInfoKey(card.ID) cacheData := map[string]interface{}{ @@ -583,6 +592,13 @@ func (s *Scheduler) initCardPolling(ctx context.Context, card *model.IotCard) er } } + if config.ProtectCheckInterval != nil && *config.ProtectCheckInterval > 0 { + nextCheck := s.calculateNextCheckTime(card.LastProtectCheckAt, *config.ProtectCheckInterval) + if err := s.addToQueue(ctx, constants.RedisPollingQueueProtectKey(), card.ID, nextCheck); err != nil { + return err + } + } + // 缓存卡信息到 Redis return s.cacheCardInfo(ctx, card, now) } @@ -631,20 +647,13 @@ func (s *Scheduler) matchConfigConditions(cfg *model.PollingConfig, card *model. // getCardCondition 获取卡的状态条件 func (s *Scheduler) getCardCondition(card *model.IotCard) string { - // 根据卡的实名状态和激活状态确定条件 if card.RealNameStatus != constants.RealNameStatusVerified { - return "not_real_name" // 未实名 + return "not_real_name" } - if card.RealNameStatus == constants.RealNameStatusVerified { - if card.NetworkStatus == 1 { - return "activated" // 已激活 - } - return "real_name" // 已实名但未激活 + if card.NetworkStatus == 1 { + return "activated" } - if card.NetworkStatus == 0 { - return "suspended" // 已停用 - } - return "" + return "real_name" } // calculateNextCheckTime 计算下次检查时间 @@ -746,3 +755,10 @@ func (s *Scheduler) SetResetService(resetService interface{}) { func (s *Scheduler) SetActivationService(activationHandler *PackageActivationHandler) { s.packageActivationHandler = activationHandler } + +// SetStopResumeCallback 注入停复机回调（用于套餐过期后主动触发停机） +func (s *Scheduler) SetStopResumeCallback(callback packagepkg.StopResumeCallback) { + if s.packageActivationHandler != nil { + s.packageActivationHandler.stopResumeCallback = callback + } +} diff --git a/internal/service/package/usage_service.go b/internal/service/package/usage_service.go index 025ab5e..a256faf 100644 --- a/internal/service/package/usage_service.go +++ b/internal/service/package/usage_service.go @@ -25,6 +25,7 @@ type UsageService struct { redis *redis.Client packageUsageStore *postgres.PackageUsageStore packageUsageDailyRecord *postgres.PackageUsageDailyRecordStore + deviceSimBindingStore *postgres.DeviceSimBindingStore logger *zap.Logger stopResumeCallback StopResumeCallback // 停复机回调，可选 } @@ -34,6 +35,7 @@ func NewUsageService( redis *redis.Client, packageUsageStore *postgres.PackageUsageStore, packageUsageDailyRecord *postgres.PackageUsageDailyRecordStore, + deviceSimBindingStore *postgres.DeviceSimBindingStore, logger *zap.Logger, ) *UsageService { return &UsageService{ @@ -41,6 +43,7 @@ func NewUsageService( redis: redis, packageUsageStore: packageUsageStore, packageUsageDailyRecord: packageUsageDailyRecord, + deviceSimBindingStore: deviceSimBindingStore, logger: logger, } } @@ -220,17 +223,44 @@ func (s *UsageService) checkAndTriggerSuspension(ctx context.Context, tx *gorm.D zap.String("carrier_type", carrierType), zap.Uint("carrier_id", carrierID)) - // 任务 24.6: 调用停复机服务执行停机 - if s.stopResumeCallback != nil && carrierType == "iot_card" { - // 在事务外异步执行停机，避免长事务 - go func() { - stopCtx := context.Background() - if err := s.stopResumeCallback.CheckAndStopCard(stopCtx, carrierID); err != nil { - s.logger.Error("调用停机服务失败", - zap.Uint("card_id", carrierID), + if carrierType == "iot_card" { + if s.stopResumeCallback != nil { + go func() { + stopCtx := context.Background() + if err := s.stopResumeCallback.CheckAndStopCard(stopCtx, carrierID); err != nil { + s.logger.Error("调用停机服务失败", + zap.Uint("card_id", carrierID), + zap.Error(err)) + } + }() + } + } else if carrierType == "device" { + if s.stopResumeCallback == nil { + s.logger.Warn("停复机回调未注入，跳过设备绑定卡停机", + zap.Uint("device_id", carrierID)) + } else { + bindings, err := s.deviceSimBindingStore.ListByDeviceID(context.Background(), carrierID) + if err != nil { + s.logger.Error("查询设备绑定卡失败", + zap.Uint("device_id", carrierID), zap.Error(err)) + } else if len(bindings) == 0 { + s.logger.Debug("设备无绑定卡，跳过停机", + zap.Uint("device_id", carrierID)) + } else { + for _, b := range bindings { + cardID := b.IotCardID + go func(cID uint) { + stopCtx := context.Background() + if err := s.stopResumeCallback.CheckAndStopCard(stopCtx, cID); err != nil { + s.logger.Error("调用停机服务失败", + zap.Uint("card_id", cID), + zap.Error(err)) + } + }(cardID) + } } - }() + } } } diff --git a/internal/task/polling_handler.go b/internal/task/polling_handler.go index ca8c88a..e581917 100644 --- a/internal/task/polling_handler.go +++ b/internal/task/polling_handler.go @@ -231,6 +231,9 @@ func (h *PollingHandler) HandleCarddataCheck(ctx context.Context, t *asynq.Task) 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 { @@ -293,9 +296,6 @@ func (h *PollingHandler) HandleCarddataCheck(ctx context.Context, t *asynq.Task) } } - // 独立停机/复机检查（不依赖流量增量，每次轮询都执行） - h.checkStopResume(ctx, card) - // 更新监控统计 h.updateStats(ctx, constants.TaskTypePollingCarddata, true, time.Since(startTime)) @@ -911,10 +911,20 @@ func (h *PollingHandler) requeueCard(ctx context.Context, cardID uint, taskType } nextCheck := time.Now().Add(time.Duration(intervalSeconds) * time.Second) - return h.redis.ZAdd(ctx, queueKey, redis.Z{ + if err := h.redis.ZAdd(ctx, queueKey, redis.Z{ Score: float64(nextCheck.Unix()), Member: strconv.FormatUint(uint64(cardID), 10), - }).Err() + }).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，返回指定任务类型的间隔秒数 @@ -952,9 +962,8 @@ func (h *PollingHandler) getMatchedPollingInterval(ctx context.Context, cardID u return *cfg.PackageCheckInterval, nil } case constants.TaskTypePollingProtect: - // protect 任务暂无专用间隔字段，使用 realname 间隔兜底 - if cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0 { - return *cfg.RealnameCheckInterval, nil + if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 { + return *cfg.ProtectCheckInterval, nil } } // 匹配到配置但该任务类型间隔为 NULL，说明此卡不需要此类轮询 @@ -1202,18 +1211,18 @@ func (h *PollingHandler) HandleProtectConsistencyCheck(ctx context.Context, t *a // 未绑设备（独立卡），跳过 if card.IsStandalone { - return nil + return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect) } // 未实名，跳过 if card.RealNameStatus != constants.RealNameStatusVerified { - return nil + return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect) } // 查绑定设备 binding, err := h.deviceSimBindingStore.GetActiveBindingByCardID(ctx, card.ID) if err != nil || binding == nil { - return nil + return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect) } deviceID := binding.DeviceID @@ -1234,7 +1243,7 @@ func (h *PollingHandler) HandleProtectConsistencyCheck(ctx context.Context, t *a zap.Uint("card_id", card.ID), zap.String("iccid", card.ICCID), zap.Error(err)) - return nil + return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect) } h.logger.Info("网关停机成功（保护期）", zap.Uint("card_id", card.ID), @@ -1247,7 +1256,7 @@ func (h *PollingHandler) HandleProtectConsistencyCheck(ctx context.Context, t *a "stop_reason": constants.StopReasonProtectPeriod, }) h.updateCardCache(ctx, card.ID, map[string]any{"network_status": constants.NetworkStatusOffline}) - return nil + return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect) } // 检查 start 保护期：设备处于复机保护期，但卡是停机状态 → 需要复机 @@ -1267,7 +1276,7 @@ func (h *PollingHandler) HandleProtectConsistencyCheck(ctx context.Context, t *a zap.Uint("card_id", card.ID), zap.String("iccid", card.ICCID), zap.Error(err)) - return nil + return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect) } h.logger.Info("网关复机成功（保护期）", zap.Uint("card_id", card.ID), @@ -1282,7 +1291,7 @@ func (h *PollingHandler) HandleProtectConsistencyCheck(ctx context.Context, t *a h.updateCardCache(ctx, card.ID, map[string]any{"network_status": constants.NetworkStatusOnline}) } - return nil + return h.requeueCard(ctx, card.ID, constants.TaskTypePollingProtect) } // triggerFirstRealnameActivation 任务 21.3-21.4: 首次实名后触发套餐激活 diff --git a/migrations/000102_add_polling_protect_fields.down.sql b/migrations/000102_add_polling_protect_fields.down.sql new file mode 100644 index 0000000..570c411 --- /dev/null +++ b/migrations/000102_add_polling_protect_fields.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE tb_polling_config DROP COLUMN IF EXISTS protect_check_interval; +ALTER TABLE tb_iot_card DROP COLUMN IF EXISTS last_protect_check_at; diff --git a/migrations/000102_add_polling_protect_fields.up.sql b/migrations/000102_add_polling_protect_fields.up.sql new file mode 100644 index 0000000..972bf6a --- /dev/null +++ b/migrations/000102_add_polling_protect_fields.up.sql @@ -0,0 +1,12 @@ +-- 添加 protect 队列所需字段 +-- tb_polling_config: 新增保护期检查间隔字段 +ALTER TABLE tb_polling_config + ADD COLUMN IF NOT EXISTS protect_check_interval INT NULL; + +COMMENT ON COLUMN tb_polling_config.protect_check_interval IS '保护期一致性检查间隔（秒），NULL=不参与'; + +-- tb_iot_card: 新增上次保护期检查时间字段 +ALTER TABLE tb_iot_card + ADD COLUMN IF NOT EXISTS last_protect_check_at TIMESTAMP NULL; + +COMMENT ON COLUMN tb_iot_card.last_protect_check_at IS '上次保护期一致性检查时间'; diff --git a/opencode.json b/opencode.json index 0319118..9983b76 100644 --- a/opencode.json +++ b/opencode.json @@ -26,9 +26,11 @@ "command": [ "npx", "-y", - "@bytebase/dbhub@0.19.1", - "--transport", "stdio", - "--config", ".config/dbhub.toml" + "@bytebase/dbhub@latest", + "--transport", + "stdio", + "--config", + ".config/dbhub.toml" ] } } diff --git a/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/.openspec.yaml b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/.openspec.yaml new file mode 100644 index 0000000..6a5db8c --- /dev/null +++ b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-04-02 diff --git a/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/design.md b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/design.md new file mode 100644 index 0000000..2918f4c --- /dev/null +++ b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/design.md @@ -0,0 +1,81 @@ +## Context + +轮询系统由 `Scheduler`（调度器）+ 4 种 `PollingHandler` 任务类型构成，以卡（`IotCard`）为单位驱动停复机、流量扣减、套餐激活等核心业务。本次修复 3 个审计发现的覆盖盲点： + +| 编号 | 问题 | 严重度 | +|------|------|--------| +| G1 | `polling:protect` 队列初始化缺失，保护期一致性检查完全失效 | P1 | +| G2 | `checkAndTriggerSuspension` device 分支为空，设备套餐耗尽不停机 | P1 | +| G3 | 套餐过期后不主动停机，依赖 carddata 轮询兜底（最长 60s 窗口） | P2 | +| G4 | `getCardCondition` 的 `"suspended"` 分支是死代码 | P2 | + +## Goals / Non-Goals + +**Goals:** +- 使 `polling:protect` 任务能够按配置间隔定时执行 +- 设备级套餐（`device_id` 绑定）流量耗尽时自动停机 +- 套餐过期时立即触发停机，消除 60s 延迟窗口 +- 清理 `getCardCondition` 死代码 + +**Non-Goals:** +- 设备维度的主动轮询（查询设备实时状态） +- 轮询性能优化（并发数、批量大小调整） +- 新增轮询任务类型 + +## Decisions + +### 决策 1：`protect_check_interval` 放在 `PollingConfig` 还是全局配置 + +**选择**：放在 `PollingConfig`，与现有 `realname_check_interval`/`carddata_check_interval`/`package_check_interval` 保持一致。 + +**理由**：不同运营商/卡类型的保护期检查频率可能不同（例如高价值卡检查更频繁）；统一用 PollingConfig 管理，不引入新的配置层级。 + +**放弃方案**：全局配置（`viper`）——灵活性不足，无法按卡类型差异化。 + +--- + +### 决策 2：设备套餐耗尽停机的触发方式 + +**选择**：在 `UsageService.checkAndTriggerSuspension` 补全 device 分支，查询设备绑定的所有卡并逐一调用 `StopResumeService.CheckAndStopCard`（异步 goroutine）。 + +**理由**： +- 和现有 iot_card 分支对称，逻辑清晰 +- 通过 `DeviceSimBindingStore.ListByDeviceID` 获取绑定卡列表，不引入跨包依赖 +- 异步执行避免长事务 + +**放弃方案**：在 `checkStopResume` 里增加 device 判断——`checkStopResume` 以卡为单位工作，不感知 device 维度的套餐，逻辑不自然。 + +--- + +### 决策 3：套餐过期主动停机的触发位置 + +**选择**：在 `PackageActivationHandler.processExpiredPackage` 的 `updateCarrierSuspendedStatus` 后，若载体仍无生效套餐，异步调用 `StopResumeCallback.CheckAndStopCard`。 + +**理由**：`processExpiredPackage` 已经在判断"是否还有生效套餐"，复用这个判断结果，减少重复查询。直接在此处触发停机是最短的调用路径。 + +**放弃方案**：新增一个独立的停机触发任务——多一个异步任务链路，延迟不减反增。 + +--- + +### 决策 4：`IotCard.LastProtectCheckAt` 字段类型 + +**选择**：`*time.Time`（nullable），与现有 `LastRealNameCheckAt`/`LastDataCheckAt` 保持一致，表示"未检查过"。 + +## Risks / Trade-offs + +- **重复停机风险**：套餐过期时 `processExpiredPackage` 会触发停机，60s 后 `checkStopResume` 也会尝试停机。`CheckAndStopCard` 内部已有"卡已停机则跳过"的幂等保护，不会重复调网关。→ **已有幂等保护，风险可控** + +- **DB 迁移风险**：新增两列均为 nullable，无需迁移存量数据，迁移后旧代码路径不受影响。→ **向下兼容，风险低** + +- **`StopResumeCallback` 接口循环依赖**：`PackageActivationHandler`（polling 包）需要调用 `StopResumeService`（iot_card 包）。当前已通过接口注入解耦（`StopResumeCallback`），直接复用该模式。→ **无新依赖** + +## Migration Plan + +1. 执行数据库迁移：`tb_polling_config` 添加 `protect_check_interval INT`；`tb_iot_card` 添加 `last_protect_check_at TIMESTAMP` +2. 部署新版本（调度器重启后自动重新初始化轮询队列，protect 队列开始填充） +3. 在 `tb_polling_config` 为需要保护期检查的配置行设置 `protect_check_interval`（建议 600s） +4. 回滚策略：回滚代码后 protect 队列不再填充，已入队任务执行后不再重新入队，不影响其他三种任务 + +## Open Questions + +- `protect_check_interval` 的默认推荐值：当前代码注释写"与流量检查同频（默认 10 分钟）"，是否沿用 600s？→ 建议 600s，与原设计一致 diff --git a/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/proposal.md b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/proposal.md new file mode 100644 index 0000000..bb6c219 --- /dev/null +++ b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/proposal.md @@ -0,0 +1,34 @@ +## Why + +轮询系统审计发现 3 个覆盖盲点：`polling:protect` 队列从未被初始化导致保护期一致性检查完全失效；设备级套餐耗尽时停机逻辑分支为空导致绑定卡继续在线消耗流量；套餐过期后卡的网络状态不主动停机，依赖下一轮 carddata 轮询兜底存在最长 60s 的流量泄露窗口。这三个问题均直接影响计费准确性和资源管控。 + +## What Changes + +- **补全 `polling:protect` 调度器初始化**：在 `PollingConfig` 添加 `protect_check_interval` 字段，在 `tb_iot_card` 添加 `last_protect_check_at` 字段，在 `initCardsBatch`/`initCardPolling`/`requeueCard` 补全 protect 队列写入逻辑，使保护期一致性检查能够定时执行 +- **设备级套餐耗尽触发停机**：在 `UsageService.checkAndTriggerSuspension` 补全 `carrierType == "device"` 分支，查询设备绑定的所有卡并异步触发停机 +- **套餐过期时主动停机**：在 `PackageActivationHandler.processExpiredPackage` 过期主套餐且无后续套餐时，直接异步触发 `CheckAndStopCard`，消除依赖轮询兜底的延迟窗口 +- **修正 `getCardCondition` 死代码**：已实名+离线的卡永远返回 `"real_name"` 而非 `"suspended"`，移除永不可达的分支 + +## Capabilities + +### New Capabilities + +- `device-package-exhausted-stop`：设备级套餐（`device_id` 绑定）流量耗尽时，自动触发设备绑定卡的停机操作 + +### Modified Capabilities + +- `polling-protect-consistency`：补全调度器初始化逻辑，使保护期一致性检查能够按配置间隔定时调度（新增 DB 字段 + 初始化代码） +- `auto-stop-resume`：新增套餐过期主动停机场景，不再完全依赖下一次 carddata 轮询触发停机 + +## Impact + +- **数据库迁移**：`tb_polling_config` 新增 `protect_check_interval` 字段；`tb_iot_card` 新增 `last_protect_check_at` 字段 +- **受影响代码**： + - `internal/polling/scheduler.go` — `initCardsBatch`、`initCardPolling`、`requeueCard` + - `internal/service/package/usage_service.go` — `checkAndTriggerSuspension` + - `internal/polling/package_activation_handler.go` — `processExpiredPackage`、`updateIotCardSuspendedStatus` + - `internal/model/polling.go` — `PollingConfig` struct + - `internal/model/iot_card.go` — `IotCard` struct + - `internal/task/polling_handler.go` — `getCardCondition` 死代码清理 +- **无 API 变更**：所有修复为内部调度逻辑，不影响对外接口 +- **需要数据库迁移**：两张表各增一列，向下兼容（nullable） diff --git a/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/specs/auto-stop-resume/spec.md b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/specs/auto-stop-resume/spec.md new file mode 100644 index 0000000..8829dce --- /dev/null +++ b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/specs/auto-stop-resume/spec.md @@ -0,0 +1,31 @@ +## ADDED Requirements + +### Requirement: 套餐过期时立即触发绑定卡停机 + +当主套餐过期处理完成后，若该套餐的载体（卡或设备）无任何后续生效套餐，系统 SHALL 立即异步触发停机检查，不依赖下一次 carddata 轮询兜底。 + +**触发位置**：`PackageActivationHandler.processExpiredPackage` 在执行 `updateCarrierSuspendedStatus` 后，若载体确认无生效套餐，异步调用 `StopResumeCallback.CheckAndStopCard`（iot_card 类型）或遍历绑定卡逐一触发（device 类型）。 + +**幂等保护**：`CheckAndStopCard` 已有"卡已停机则跳过"逻辑，重复触发安全。 + +**与 carddata 轮询的关系**：此处主动触发为优化项（消除延迟窗口），carddata 轮询的 `checkStopResume` 仍作为兜底保障，两者共存不冲突。 + +#### Scenario: 主套餐过期且无后续套餐，载体为 iot_card + +- **WHEN** `HandlePackageActivationCheck` 检测到一张卡的主套餐已过期（`expires_at <= NOW`），执行 `processExpiredPackage` 后确认该卡无待生效或生效中的套餐 +- **THEN** 系统异步调用 `CheckAndStopCard(cardID)`，将卡停机（`network_status = 0`，`stop_reason = "traffic_exhausted"`），不等待网关响应 + +#### Scenario: 主套餐过期且无后续套餐，载体为 device + +- **WHEN** `HandlePackageActivationCheck` 检测到一个设备的主套餐已过期，确认该设备无后续套餐 +- **THEN** 系统查询设备绑定的所有在线已实名卡，逐一异步触发 `CheckAndStopCard` + +#### Scenario: 主套餐过期但有排队待生效套餐 + +- **WHEN** `processExpiredPackage` 处理过期套餐，`activateNextPackage` 成功激活了下一个待生效套餐 +- **THEN** 系统不触发停机（因有生效套餐），`updateCarrierSuspendedStatus` 检查后确认有生效套餐即返回 + +#### Scenario: carddata 轮询兜底仍有效 + +- **WHEN** 套餐过期主动停机的异步触发因 gateway 异常失败，卡仍处于在线状态 +- **THEN** 下一次 `HandleCarddataCheck` 的 `checkStopResume` 检测到在线无套餐，再次尝试停机（兜底保障） diff --git a/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/specs/device-package-exhausted-stop/spec.md b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/specs/device-package-exhausted-stop/spec.md new file mode 100644 index 0000000..71d5ebe --- /dev/null +++ b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/specs/device-package-exhausted-stop/spec.md @@ -0,0 +1,31 @@ +## ADDED Requirements + +### Requirement: 设备级套餐耗尽时触发绑定卡停机 + +当套餐载体为设备（`carrier_type = "device"`）且所有生效套餐流量耗尽时，系统 SHALL 查询该设备绑定的所有 IoT 卡，并对每张卡异步触发停机检查（`CheckAndStopCard`）。 + +**触发位置**：`UsageService.checkAndTriggerSuspension`，在确认无生效套餐后执行。 + +**幂等保护**：`CheckAndStopCard` 内部检查卡是否已为停机状态（`network_status = 0`），重复调用不会重复调网关。 + +**仅处理绑定卡**：通过 `DeviceSimBindingStore.ListByDeviceID` 获取当前绑定关系（`bind_status = 1`），未绑定或已解绑的卡不受影响。 + +#### Scenario: 设备套餐流量耗尽，绑定卡在线 + +- **WHEN** `DeductDataUsage` 在 `carrier_type = "device"` 场景下扣减流量后，`checkAndTriggerSuspension` 发现该设备无生效套餐（`status IN (0,1)`） +- **THEN** 系统查询该设备的绑定卡列表，对每张在线（`network_status = 1`）且已实名（`real_name_status = 1`）的卡异步调用 `CheckAndStopCard`，将卡停机（`network_status = 0`，`stop_reason = "traffic_exhausted"`） + +#### Scenario: 设备套餐流量耗尽，绑定卡已停机 + +- **WHEN** `checkAndTriggerSuspension` 触发设备级停机检查，但绑定卡已经是停机状态（`network_status = 0`） +- **THEN** `CheckAndStopCard` 检查到卡已停机，跳过网关调用，不产生重复操作 + +#### Scenario: 设备无绑定卡 + +- **WHEN** `checkAndTriggerSuspension` 触发设备级停机检查，但该设备当前无有效绑定记录 +- **THEN** 系统静默跳过，不报错，记录 Debug 日志 + +#### Scenario: 套餐载体为 iot_card，不受影响 + +- **WHEN** `DeductDataUsage` 在 `carrier_type = "iot_card"` 场景下触发停机检查 +- **THEN** 仅对该卡本身执行 `CheckAndStopCard`，不查询设备绑定关系（与现有行为一致） diff --git a/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/specs/polling-protect-consistency/spec.md b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/specs/polling-protect-consistency/spec.md new file mode 100644 index 0000000..79c80ac --- /dev/null +++ b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/specs/polling-protect-consistency/spec.md @@ -0,0 +1,54 @@ +## MODIFIED Requirements + +### Requirement: 保护期一致性检查轮询任务 + +系统 SHALL 通过轮询调度器定时执行保护期一致性检查任务（`polling:protect`），检查绑定设备有保护期的已实名卡的网络状态，并在状态不一致时强制调网关修正。 + +**任务类型标识**: `protect`（与现有 `realname`、`carddata`、`package` 并列） + +**Redis 队列 Key**: `RedisPollingQueueProtectKey()` → `"polling:queue:protect"` + +**触发频率**: 由 `PollingConfig.ProtectCheckInterval`（新增字段，单位秒）控制，`NULL` 或 `0` 表示该配置不参与 protect 检查，建议默认 600s + +**初始化**：调度器启动时，`initCardsBatch` 和 `initCardPolling` SHALL 读取匹配配置的 `ProtectCheckInterval`，将卡加入 `polling:queue:protect` Sorted Set；任务完成后通过 `requeueCard` 按间隔重新入队 + +**数据模型变更**： +- `tb_polling_config` 新增 `protect_check_interval INT NULL` 字段 +- `tb_iot_card` 新增 `last_protect_check_at TIMESTAMP NULL` 字段（记录上次检查时间，用于计算下次入队 score） + +**处理逻辑**（与原 spec 一致，不变）: +1. 卡未实名（`real_name_status = 0`）→ 跳过 +2. 卡未绑定设备（`is_standalone = true`）→ 跳过 +3. 设备有 stop 保护期且卡在线（`network_status = 1`）→ 调网关停机，更新 `network_status = 0` +4. 设备有 start 保护期且卡停机（`network_status = 0`）→ 调网关复机，更新 `network_status = 1` +5. 状态已一致 → 跳过 + +#### Scenario: stop 保护期内卡状态异常（开机） + +- **WHEN** protect 轮询任务检查一张已实名、已绑定设备的卡，发现设备有 stop 保护期，且卡当前 `network_status = 1`（开机） +- **THEN** 任务调网关停机，更新卡 `network_status = 0`，更新 `last_protect_check_at`，记录 Info 日志 + +#### Scenario: start 保护期内卡状态异常（停机） + +- **WHEN** protect 轮询任务检查一张已实名、已绑定设备的卡，发现设备有 start 保护期，且卡当前 `network_status = 0`（停机） +- **THEN** 任务调网关复机，更新卡 `network_status = 1`，更新 `last_protect_check_at`，记录 Info 日志 + +#### Scenario: 状态已一致，跳过 + +- **WHEN** protect 轮询任务检查一张卡，设备有 stop 保护期，卡已是停机状态（`network_status = 0`） +- **THEN** 任务跳过，不调网关，不更新 DB，更新 `last_protect_check_at` + +#### Scenario: 未实名卡跳过保护期逻辑 + +- **WHEN** protect 轮询任务遇到 `real_name_status = 0` 的卡 +- **THEN** 任务直接跳过，不检查保护期，不调网关 + +#### Scenario: 独立卡（未绑定设备）跳过 + +- **WHEN** protect 轮询任务遇到 `is_standalone = true` 的卡 +- **THEN** 任务直接跳过，不查询设备保护期 + +#### Scenario: PollingConfig 未配置 ProtectCheckInterval，卡不入 protect 队列 + +- **WHEN** 调度器初始化时，卡匹配到的 PollingConfig 的 `protect_check_interval` 为 NULL 或 0 +- **THEN** 该卡不被加入 `polling:queue:protect`，protect 任务不对该卡执行 diff --git a/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/tasks.md b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/tasks.md new file mode 100644 index 0000000..8a7a690 --- /dev/null +++ b/openspec/changes/archive/2026-04-02-fix-polling-coverage-gaps/tasks.md @@ -0,0 +1,52 @@ +## 1. 数据库迁移 + +- [x] 1.1 创建迁移文件，在 `tb_polling_config` 添加 `protect_check_interval INT NULL COMMENT '保护期一致性检查间隔（秒），NULL=不参与'` +- [x] 1.2 在同一迁移文件中，在 `tb_iot_card` 添加 `last_protect_check_at TIMESTAMP NULL COMMENT '上次保护期一致性检查时间'` +- [x] 1.3 执行迁移并验证：通过 PostgreSQL MCP 确认两列已成功添加，类型和注释正确 + +## 2. Model 更新 + +- [x] 2.1 在 `internal/model/polling.go` 的 `PollingConfig` struct 添加 `ProtectCheckInterval *int` 字段，gorm tag 包含 `column:protect_check_interval`，json tag 为 `protect_check_interval`，中文注释 +- [x] 2.2 在 `internal/model/iot_card.go` 的 `IotCard` struct 添加 `LastProtectCheckAt *time.Time` 字段，gorm tag 包含 `column:last_protect_check_at`，json tag 为 `last_protect_check_at`，中文注释 +- [x] 2.3 运行 `lsp_diagnostics` 确认两个 model 文件无错误 + +## 3. 轮询调度器：protect 队列初始化 + +- [x] 3.1 在 `internal/polling/scheduler.go` 的 `initCardsBatch` 方法中，在 package 队列初始化块之后添加 protect 队列初始化：读取 `cfg.ProtectCheckInterval`，调用 `calculateNextCheckTime(card.LastProtectCheckAt, *cfg.ProtectCheckInterval)`，写入 `RedisPollingQueueProtectKey()` +- [x] 3.2 在 `initCardPolling` 方法中同步添加 protect 队列初始化逻辑（与 `initCardsBatch` 对称） +- [x] 3.3 在 `requeueCard` 方法中添加 `constants.TaskTypePollingProtect` case：读取 `config.ProtectCheckInterval`，入队 `RedisPollingQueueProtectKey()`，更新 `last_protect_check_at` +- [x] 3.4 在 `HandleProtectConsistencyCheck` 任务处理完成后调用 `requeueCard(cardID, TaskTypePollingProtect)` 使卡重新入队 +- [x] 3.5 运行 `lsp_diagnostics` 确认 `scheduler.go` 和 `polling_handler.go` 无错误 + +## 4. 为 protect 配置设置检查间隔 + +- [x] 4.1 通过数据库执行 SQL，为现有已启用的 `tb_polling_config` 配置行（`status = 1` 的行）设置 `protect_check_interval = 600`（10 分钟） +- [x] 4.2 通过 PostgreSQL MCP 验证配置已更新，查询 `SELECT id, config_name, protect_check_interval FROM tb_polling_config WHERE status = 1` + +## 5. 设备套餐耗尽触发停机 + +- [x] 5.1 在 `internal/service/package/usage_service.go` 的 `UsageService` struct 中添加 `deviceSimBindingStore *postgres.DeviceSimBindingStore` 字段 +- [x] 5.2 更新 `NewUsageService` 构造函数，接收 `deviceSimBindingStore` 参数并赋值 +- [x] 5.3 更新 `bootstrap` 中 `UsageService` 的初始化调用，传入 `deviceSimBindingStore` +- [x] 5.4 在 `checkAndTriggerSuspension` 的 `if activeCount == 0` 块内，补全 `carrierType == "device"` 分支：查询设备绑定卡列表（`ListByDeviceID`），对每张卡异步启动 goroutine 调用 `s.stopResumeCallback.CheckAndStopCard`；若 `stopResumeCallback == nil` 则仅记录 Warn 日志 +- [x] 5.5 运行 `lsp_diagnostics` 确认 `usage_service.go` 无错误 + +## 6. 套餐过期主动触发停机 + +- [x] 6.1 在 `internal/polling/package_activation_handler.go` 的 `PackageActivationHandler` struct 中添加 `stopResumeCallback StopResumeCallback` 字段（复用 `usage_service.go` 已有的同名接口） +- [x] 6.2 更新 `NewPackageActivationHandler` 构造函数，接收 `stopResumeCallback` 参数并赋值 +- [x] 6.3 更新 `Scheduler` 中 `PackageActivationHandler` 的初始化，传入 `stopResumeCallback` +- [x] 6.4 在 `processExpiredPackage` 中，`updateCarrierSuspendedStatus` 调用之后，若载体类型为 `iot_card` 且 `stopResumeCallback != nil`，异步调用 `CheckAndStopCard(cardID)`；若载体类型为 `device`，查询绑定卡并逐一异步触发 +- [x] 6.5 运行 `lsp_diagnostics` 确认 `package_activation_handler.go` 无错误 + +## 7. 死代码清理 + +- [x] 7.1 在 `internal/polling/scheduler.go` 的 `getCardCondition` 函数中，删除永不可达的 `if card.NetworkStatus == 0 { return "suspended" }` 和 `return ""` 两行（已实名+离线永远走 `"real_name"` 分支） +- [x] 7.2 运行 `lsp_diagnostics` 确认 `scheduler.go` 无错误 + +## 8. 端到端验证 + +- [x] 8.1 重启服务，通过日志确认调度器初始化时 protect 队列有卡入队（关键日志：`完成一批卡初始化`） +- [x] 8.2 通过 PostgreSQL MCP 查询一张卡的 `last_protect_check_at`，确认 protect 任务执行后字段有更新：`SELECT id, iccid, last_protect_check_at FROM tb_iot_card WHERE last_protect_check_at IS NOT NULL LIMIT 5` +- [x] 8.3 通过 PostgreSQL MCP 构造一条设备级套餐耗尽场景：确认设备套餐 status=2（depleted）后，绑定卡的 `network_status` 变为 0，`stop_reason = 'traffic_exhausted'` +- [x] 8.4 通过 PostgreSQL MCP 查询一张已过期套餐（`expires_at <= NOW`）且无后续套餐的卡，确认其 `network_status = 0`，`stop_reason = 'traffic_exhausted'` diff --git a/openspec/specs/auto-stop-resume/spec.md b/openspec/specs/auto-stop-resume/spec.md index 2b88402..2e0bc82 100644 --- a/openspec/specs/auto-stop-resume/spec.md +++ b/openspec/specs/auto-stop-resume/spec.md @@ -392,6 +392,40 @@ func (s *Service) ResumeCardIfStopped(ctx context.Context, cardID uint) error { --- +## 迭代更新（fix-polling-coverage-gaps） + +### ADDED Requirement: 套餐过期时立即触发绑定卡停机 + +当主套餐过期处理完成后，若该套餐的载体（卡或设备）无任何后续生效套餐，系统 SHALL 立即异步触发停机检查，不依赖下一次 carddata 轮询兜底。 + +**触发位置**：`PackageActivationHandler.processExpiredPackage` 在执行 `updateCarrierSuspendedStatus` 后，若载体确认无生效套餐，异步调用 `StopResumeCallback.CheckAndStopCard`（iot_card 类型）或遍历绑定卡逐一触发（device 类型）。 + +**幂等保护**：`CheckAndStopCard` 已有"卡已停机则跳过"逻辑，重复触发安全。 + +**与 carddata 轮询的关系**：此处主动触发为优化项（消除延迟窗口），carddata 轮询的 `checkStopResume` 仍作为兜底保障，两者共存不冲突。 + +#### Scenario: 主套餐过期且无后续套餐，载体为 iot_card + +- **WHEN** `HandlePackageActivationCheck` 检测到一张卡的主套餐已过期（`expires_at <= NOW`），执行 `processExpiredPackage` 后确认该卡无待生效或生效中的套餐 +- **THEN** 系统异步调用 `CheckAndStopCard(cardID)`，将卡停机（`network_status = 0`，`stop_reason = "traffic_exhausted"`），不等待网关响应 + +#### Scenario: 主套餐过期且无后续套餐，载体为 device + +- **WHEN** `HandlePackageActivationCheck` 检测到一个设备的主套餐已过期，确认该设备无后续套餐 +- **THEN** 系统查询设备绑定的所有在线已实名卡，逐一异步触发 `CheckAndStopCard` + +#### Scenario: 主套餐过期但有排队待生效套餐 + +- **WHEN** `processExpiredPackage` 处理过期套餐，`activateNextPackage` 成功激活了下一个待生效套餐 +- **THEN** 系统不触发停机（因有生效套餐），`updateCarrierSuspendedStatus` 检查后确认有生效套餐即返回 + +#### Scenario: carddata 轮询兜底仍有效 + +- **WHEN** 套餐过期主动停机的异步触发因 gateway 异常失败，卡仍处于在线状态 +- **THEN** 下一次 `HandleCarddataCheck` 的 `checkStopResume` 检测到在线无套餐，再次尝试停机（兜底保障） + +--- + ## 迭代更新（fix-stop-resume-lifecycle-engine） ### MODIFIED Requirement: 流量耗尽自动复机 diff --git a/openspec/specs/device-package-exhausted-stop/spec.md b/openspec/specs/device-package-exhausted-stop/spec.md new file mode 100644 index 0000000..9b1e538 --- /dev/null +++ b/openspec/specs/device-package-exhausted-stop/spec.md @@ -0,0 +1,61 @@ +# Spec: 设备级套餐耗尽触发绑定卡停机 + +## 业务背景 + +### 为什么需要设备级套餐耗尽停机 + +**现状问题**： +- 设备套餐（`carrier_type = "device"`）流量耗尽时，`checkAndTriggerSuspension` 缺少对绑定卡的停机逻辑 +- 仅有 `iot_card` 类型载体的停机分支，设备类型无对应处理 +- 导致设备套餐流量耗尽后，绑定卡仍保持在线状态，无法及时停机 + +**业务目标**： +- 设备套餐流量耗尽时，自动对所有绑定卡触发停机检查 +- 与 iot_card 类型保持行为一致性 +- 幂等安全，重复触发不产生副作用 + +--- + +## ADDED Requirements + +### Requirement: 设备级套餐耗尽时触发绑定卡停机 + +当套餐载体为设备（`carrier_type = "device"`）且所有生效套餐流量耗尽时，系统 SHALL 查询该设备绑定的所有 IoT 卡，并对每张卡异步触发停机检查（`CheckAndStopCard`）。 + +**触发位置**：`UsageService.checkAndTriggerSuspension`，在确认无生效套餐后执行。 + +**幂等保护**：`CheckAndStopCard` 内部检查卡是否已为停机状态（`network_status = 0`），重复调用不会重复调网关。 + +**仅处理绑定卡**：通过 `DeviceSimBindingStore.ListByDeviceID` 获取当前绑定关系（`bind_status = 1`），未绑定或已解绑的卡不受影响。 + +#### Scenario: 设备套餐流量耗尽，绑定卡在线 + +- **WHEN** `DeductDataUsage` 在 `carrier_type = "device"` 场景下扣减流量后，`checkAndTriggerSuspension` 发现该设备无生效套餐（`status IN (0,1)`） +- **THEN** 系统查询该设备的绑定卡列表，对每张在线（`network_status = 1`）且已实名（`real_name_status = 1`）的卡异步调用 `CheckAndStopCard`，将卡停机（`network_status = 0`，`stop_reason = "traffic_exhausted"`） + +#### Scenario: 设备套餐流量耗尽，绑定卡已停机 + +- **WHEN** `checkAndTriggerSuspension` 触发设备级停机检查，但绑定卡已经是停机状态（`network_status = 0`） +- **THEN** `CheckAndStopCard` 检查到卡已停机，跳过网关调用，不产生重复操作 + +#### Scenario: 设备无绑定卡 + +- **WHEN** `checkAndTriggerSuspension` 触发设备级停机检查，但该设备当前无有效绑定记录 +- **THEN** 系统静默跳过，不报错，记录 Debug 日志 + +#### Scenario: 套餐载体为 iot_card，不受影响 + +- **WHEN** `DeductDataUsage` 在 `carrier_type = "iot_card"` 场景下触发停机检查 +- **THEN** 仅对该卡本身执行 `CheckAndStopCard`，不查询设备绑定关系（与现有行为一致） + +--- + +## 实现说明 + +### 依赖注入变更 + +`UsageService` 新增 `deviceSimBindingStore *postgres.DeviceSimBindingStore` 字段，通过构造函数注入，`bootstrap` 层负责传入。 + +### 异步触发模式 + +对每张绑定卡启动独立 goroutine 异步执行 `CheckAndStopCard`，不阻塞当前流量扣减流程。若 `stopResumeCallback == nil`，仅记录 Warn 日志，不报错。 diff --git a/openspec/specs/polling-protect-consistency/spec.md b/openspec/specs/polling-protect-consistency/spec.md index 61be1ad..4954409 100644 --- a/openspec/specs/polling-protect-consistency/spec.md +++ b/openspec/specs/polling-protect-consistency/spec.md @@ -50,3 +50,60 @@ - **WHEN** 轮询任务遇到 is_standalone=true 的卡 - **THEN** 任务直接跳过，不查询设备保护期 + +--- + +## 迭代更新（fix-polling-coverage-gaps） + +### MODIFIED Requirement: 保护期一致性检查轮询任务 + +系统 SHALL 通过轮询调度器定时执行保护期一致性检查任务（`polling:protect`），检查绑定设备有保护期的已实名卡的网络状态，并在状态不一致时强制调网关修正。 + +**任务类型标识**: `protect`（与现有 `realname`、`carddata`、`package` 并列） + +**Redis 队列 Key**: `RedisPollingQueueProtectKey()` → `"polling:queue:protect"` + +**触发频率**（变更）: 由 `PollingConfig.ProtectCheckInterval`（新增字段，单位秒）控制，`NULL` 或 `0` 表示该配置不参与 protect 检查，建议默认 600s（原：与流量检查任务同频固定 10 分钟） + +**初始化**：调度器启动时，`initCardsBatch` 和 `initCardPolling` SHALL 读取匹配配置的 `ProtectCheckInterval`，将卡加入 `polling:queue:protect` Sorted Set；任务完成后通过 `requeueCard` 按间隔重新入队，并更新 `last_protect_check_at` + +**数据模型变更**： +- `tb_polling_config` 新增 `protect_check_interval INT NULL` 字段 +- `tb_iot_card` 新增 `last_protect_check_at TIMESTAMP NULL` 字段（记录上次检查时间，用于计算下次入队 score） + +**处理逻辑**（与原 spec 一致，不变）: +1. 卡未实名（`real_name_status = 0`）→ 跳过 +2. 卡未绑定设备（`is_standalone = true`）→ 跳过 +3. 设备有 stop 保护期且卡在线（`network_status = 1`）→ 调网关停机，更新 `network_status = 0` +4. 设备有 start 保护期且卡停机（`network_status = 0`）→ 调网关复机，更新 `network_status = 1` +5. 状态已一致 → 跳过 + +#### Scenario: stop 保护期内卡状态异常（开机） + +- **WHEN** protect 轮询任务检查一张已实名、已绑定设备的卡，发现设备有 stop 保护期，且卡当前 `network_status = 1`（开机） +- **THEN** 任务调网关停机，更新卡 `network_status = 0`，更新 `last_protect_check_at`，记录 Info 日志 + +#### Scenario: start 保护期内卡状态异常（停机） + +- **WHEN** protect 轮询任务检查一张已实名、已绑定设备的卡，发现设备有 start 保护期，且卡当前 `network_status = 0`（停机） +- **THEN** 任务调网关复机，更新卡 `network_status = 1`，更新 `last_protect_check_at`，记录 Info 日志 + +#### Scenario: 状态已一致，跳过 + +- **WHEN** protect 轮询任务检查一张卡，设备有 stop 保护期，卡已是停机状态（`network_status = 0`） +- **THEN** 任务跳过，不调网关，不更新 DB，更新 `last_protect_check_at` + +#### Scenario: 未实名卡跳过保护期逻辑 + +- **WHEN** protect 轮询任务遇到 `real_name_status = 0` 的卡 +- **THEN** 任务直接跳过，不检查保护期，不调网关 + +#### Scenario: 独立卡（未绑定设备）跳过 + +- **WHEN** protect 轮询任务遇到 `is_standalone = true` 的卡 +- **THEN** 任务直接跳过，不查询设备保护期 + +#### Scenario: PollingConfig 未配置 ProtectCheckInterval，卡不入 protect 队列 + +- **WHEN** 调度器初始化时，卡匹配到的 PollingConfig 的 `protect_check_interval` 为 NULL 或 0 +- **THEN** 该卡不被加入 `polling:queue:protect`，protect 任务不对该卡执行 diff --git a/pkg/queue/handler.go b/pkg/queue/handler.go index 0a3288c..1aa3c3c 100644 --- a/pkg/queue/handler.go +++ b/pkg/queue/handler.go @@ -177,6 +177,7 @@ func (h *Handler) registerPackageActivationHandlers() { h.redis, h.asynqClient, h.workerResult.Services.ActivationService, + nil, h.logger, ) diff --git a/pkg/queue/types.go b/pkg/queue/types.go index ef39005..e5df8ad 100644 --- a/pkg/queue/types.go +++ b/pkg/queue/types.go @@ -53,7 +53,8 @@ type WorkerServices struct { ResetService *packagepkg.ResetService AlertService *pollingSvc.AlertService CleanupService *pollingSvc.CleanupService - OrderExpirer OrderExpirer // 订单超时取消服务（接口类型，避免循环依赖） + StopResumeService packagepkg.StopResumeCallback // 停复机服务，用于注入 Scheduler + OrderExpirer OrderExpirer // 订单超时取消服务（接口类型，避免循环依赖） } // WorkerBootstrapResult Worker Bootstrap 结果