fix: 补全三处轮询覆盖缺口
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 7m11s
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>
This commit is contained in:
@@ -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{
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user