diff --git a/internal/polling/initializer.go b/internal/polling/initializer.go index ded6b78..5507046 100644 --- a/internal/polling/initializer.go +++ b/internal/polling/initializer.go @@ -252,6 +252,13 @@ func (p *PollingInitializer) initBatch(ctx context.Context, cards []*model.IotCa }) cmdCount++ } + if cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0 { + nextCheck := calculateNextCheckTime(card.LastCardStatusCheckAt, *cfg.CardStatusCheckInterval, now) + pipe.ZAdd(ctx, constants.RedisPollingShardQueueKey(shardID, constants.TaskTypePollingCardStatus), redis.Z{ + Score: float64(nextCheck.Unix()), Member: cardIDStr, + }) + cmdCount++ + } cacheKey := constants.RedisPollingCardInfoKey(card.ID) cacheData := map[string]interface{}{ diff --git a/internal/polling/lifecycle_service.go b/internal/polling/lifecycle_service.go index 5f558b4..40bcd54 100644 --- a/internal/polling/lifecycle_service.go +++ b/internal/polling/lifecycle_service.go @@ -162,6 +162,9 @@ func getEnabledTaskTypes(cfg *model.PollingConfig) []string { if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 { types = append(types, constants.TaskTypePollingProtect) } + if cfg.CardStatusCheckInterval != nil && *cfg.CardStatusCheckInterval > 0 { + types = append(types, constants.TaskTypePollingCardStatus) + } return types } @@ -188,6 +191,10 @@ func calcInitialDelay(_ *model.IotCard, cfg *model.PollingConfig, taskType strin if cfg.ProtectCheckInterval != nil { interval = *cfg.ProtectCheckInterval } + case constants.TaskTypePollingCardStatus: + if cfg.CardStatusCheckInterval != nil { + interval = *cfg.CardStatusCheckInterval + } } if interval <= 0 { return now diff --git a/internal/polling/queue_manager.go b/internal/polling/queue_manager.go index d175f8e..7275893 100644 --- a/internal/polling/queue_manager.go +++ b/internal/polling/queue_manager.go @@ -18,6 +18,7 @@ var allTaskTypes = []string{ constants.TaskTypePollingCarddata, constants.TaskTypePollingPackage, constants.TaskTypePollingProtect, + constants.TaskTypePollingCardStatus, } // dequeueScript Lua 脚本:原子出队(ZRANGEBYSCORE + ZREM 服务端原子执行) @@ -97,7 +98,7 @@ func (m *PollingQueueManager) Requeue(ctx context.Context, cardID uint, taskType }).Err() } -// RemoveFromAllQueues 从所有分片的所有4个队列(含protect)移除指定卡 +// RemoveFromAllQueues 从所有分片的所有5个队列(realname/carddata/package/protect/card_status)移除指定卡 // 修复 Bug3:旧实现漏掉 protect 队列 func (m *PollingQueueManager) RemoveFromAllQueues(ctx context.Context, cardID uint) error { member := fmt.Sprintf("%d", cardID)