huang
2a7ac3f86e
Some checks failed
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Has been cancelled
- OnCardStatusChanged/Enabled/Disabled 添加 InvalidateCardCache， 解决 Refresh API 更新 DB 后 polling 缓存仍为旧值的 bug - AssetResolveResponse 的 enable_polling/network_status 去掉 omitempty，解决 false/0 时字段从响应中消失的问题 - Scheduler/Initializer/PackageHandler 增加 INFO 级别日志， 可通过日志判断轮询是否工作、处理了哪些卡
201 lines
6.9 KiB
Go
201 lines
6.9 KiB
Go
package polling
|
||
|
||
import (
|
||
"context"
|
||
"math/rand"
|
||
"time"
|
||
|
||
"go.uber.org/zap"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
)
|
||
|
||
// PollingLifecycleService 卡生命周期轮询管理服务
|
||
// 替代 callbacks.go 和 api_callback.go 中的生命周期方法
|
||
// 实现 iot_card.PollingCallback 接口，两个进程（API 和 Worker）共享同一实现
|
||
// 依赖：PollingQueueManager（队列操作）+ PollingConfigManager（配置匹配）
|
||
type PollingLifecycleService struct {
|
||
queueMgr *PollingQueueManager
|
||
configMgr *PollingConfigManager
|
||
iotCardStore *postgres.IotCardStore
|
||
deviceBindingStore *postgres.DeviceSimBindingStore
|
||
deviceStore *postgres.DeviceStore
|
||
logger *zap.Logger
|
||
}
|
||
|
||
// NewPollingLifecycleService 创建卡生命周期轮询管理服务
|
||
func NewPollingLifecycleService(
|
||
queueMgr *PollingQueueManager,
|
||
configMgr *PollingConfigManager,
|
||
iotCardStore *postgres.IotCardStore,
|
||
deviceBindingStore *postgres.DeviceSimBindingStore,
|
||
deviceStore *postgres.DeviceStore,
|
||
logger *zap.Logger,
|
||
) *PollingLifecycleService {
|
||
return &PollingLifecycleService{
|
||
queueMgr: queueMgr,
|
||
configMgr: configMgr,
|
||
iotCardStore: iotCardStore,
|
||
deviceBindingStore: deviceBindingStore,
|
||
deviceStore: deviceStore,
|
||
logger: logger,
|
||
}
|
||
}
|
||
|
||
// OnCardCreated 新卡创建后初始化轮询
|
||
func (s *PollingLifecycleService) OnCardCreated(ctx context.Context, card *model.IotCard) {
|
||
if card == nil {
|
||
return
|
||
}
|
||
if !s.shouldEnqueue(ctx, card) {
|
||
s.logger.Debug("卡禁用轮询，跳过入队", zap.Uint("card_id", card.ID))
|
||
return
|
||
}
|
||
s.enqueueCard(ctx, card)
|
||
}
|
||
|
||
// OnBatchCardsCreated 批量卡创建后批量初始化轮询
|
||
func (s *PollingLifecycleService) OnBatchCardsCreated(ctx context.Context, cards []*model.IotCard) {
|
||
for _, card := range cards {
|
||
s.OnCardCreated(ctx, card)
|
||
}
|
||
}
|
||
|
||
// OnCardStatusChanged 卡状态变化后清理缓存、重新匹配配置并更新队列
|
||
func (s *PollingLifecycleService) OnCardStatusChanged(ctx context.Context, cardID uint) {
|
||
if err := s.queueMgr.RemoveFromAllQueues(ctx, cardID); err != nil {
|
||
s.logger.Warn("卡状态变化：从队列移除失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||
}
|
||
// 清理轮询卡缓存，确保下次轮询从 DB 读取最新状态
|
||
s.queueMgr.InvalidateCardCache(ctx, cardID)
|
||
|
||
card, err := s.iotCardStore.GetByID(ctx, cardID)
|
||
if err != nil {
|
||
s.logger.Error("卡状态变化：加载卡信息失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||
return
|
||
}
|
||
if !s.shouldEnqueue(ctx, card) {
|
||
return
|
||
}
|
||
s.enqueueCard(ctx, card)
|
||
}
|
||
|
||
// OnCardDeleted 卡删除后移除所有队列并清理缓存
|
||
func (s *PollingLifecycleService) OnCardDeleted(ctx context.Context, cardID uint) {
|
||
if err := s.queueMgr.OnCardDeleted(ctx, cardID); err != nil {
|
||
s.logger.Warn("卡删除：清理队列和缓存失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||
}
|
||
}
|
||
|
||
// OnCardEnabled 卡启用轮询后初始化
|
||
func (s *PollingLifecycleService) OnCardEnabled(ctx context.Context, cardID uint) {
|
||
s.queueMgr.InvalidateCardCache(ctx, cardID)
|
||
card, err := s.iotCardStore.GetByID(ctx, cardID)
|
||
if err != nil {
|
||
s.logger.Error("卡启用：加载卡信息失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||
return
|
||
}
|
||
if !s.shouldEnqueue(ctx, card) {
|
||
return
|
||
}
|
||
s.enqueueCard(ctx, card)
|
||
}
|
||
|
||
// OnCardDisabled 卡禁用轮询后移除所有队列
|
||
func (s *PollingLifecycleService) OnCardDisabled(ctx context.Context, cardID uint) {
|
||
if err := s.queueMgr.RemoveFromAllQueues(ctx, cardID); err != nil {
|
||
s.logger.Warn("卡禁用：从队列移除失败", zap.Uint("card_id", cardID), zap.Error(err))
|
||
}
|
||
s.queueMgr.InvalidateCardCache(ctx, cardID)
|
||
}
|
||
|
||
// shouldEnqueue M3 修复：检查卡或其绑定设备是否允许轮询
|
||
// 两层检查：先查卡级 enable_polling，再查设备级 enable_polling（IsStandalone=false 时）
|
||
// 直接查 device.enable_polling，不依赖级联同步正确性，防止同步失败时生命周期事件绕过设备级禁用
|
||
func (s *PollingLifecycleService) shouldEnqueue(ctx context.Context, card *model.IotCard) bool {
|
||
if !card.EnablePolling {
|
||
return false
|
||
}
|
||
if !card.IsStandalone && s.deviceBindingStore != nil && s.deviceStore != nil {
|
||
binding, err := s.deviceBindingStore.GetActiveBindingByCardID(ctx, card.ID)
|
||
if err == nil && binding != nil {
|
||
device, devErr := s.deviceStore.GetByID(ctx, binding.DeviceID)
|
||
if devErr == nil && !device.EnablePolling {
|
||
s.logger.Debug("设备已禁用轮询，跳过卡入队",
|
||
zap.Uint("card_id", card.ID), zap.Uint("device_id", binding.DeviceID))
|
||
return false
|
||
}
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
|
||
// enqueueCard 按匹配配置将卡入队分片队列
|
||
func (s *PollingLifecycleService) enqueueCard(ctx context.Context, card *model.IotCard) {
|
||
cfg := s.configMgr.MatchConfig(card)
|
||
if cfg == nil {
|
||
return
|
||
}
|
||
taskTypes := getEnabledTaskTypes(cfg)
|
||
for _, taskType := range taskTypes {
|
||
if err := s.queueMgr.Requeue(ctx, card.ID, taskType, calcInitialDelay(card, cfg, taskType)); err != nil {
|
||
s.logger.Warn("卡入队失败",
|
||
zap.Uint("card_id", card.ID), zap.String("task_type", taskType), zap.Error(err))
|
||
}
|
||
}
|
||
}
|
||
|
||
// getEnabledTaskTypes 返回配置中启用的任务类型列表
|
||
func getEnabledTaskTypes(cfg *model.PollingConfig) []string {
|
||
var types []string
|
||
if cfg.RealnameCheckInterval != nil && *cfg.RealnameCheckInterval > 0 {
|
||
types = append(types, constants.TaskTypePollingRealname)
|
||
}
|
||
if cfg.CarddataCheckInterval != nil && *cfg.CarddataCheckInterval > 0 {
|
||
types = append(types, constants.TaskTypePollingCarddata)
|
||
}
|
||
if cfg.PackageCheckInterval != nil && *cfg.PackageCheckInterval > 0 {
|
||
types = append(types, constants.TaskTypePollingPackage)
|
||
}
|
||
if cfg.ProtectCheckInterval != nil && *cfg.ProtectCheckInterval > 0 {
|
||
types = append(types, constants.TaskTypePollingProtect)
|
||
}
|
||
return types
|
||
}
|
||
|
||
// calcInitialDelay 计算新入队卡的初始检查时间（加随机抖动避免惊群）
|
||
// 抖动范围 [0, max(interval/10, 2)) 秒；下界取 2 保证 rand.Intn 有实际随机性
|
||
// （rand.Intn(1) 永远返回 0，对 interval≤10 的配置会导致零抖动）
|
||
func calcInitialDelay(_ *model.IotCard, cfg *model.PollingConfig, taskType string) time.Time {
|
||
now := time.Now()
|
||
var interval int
|
||
switch taskType {
|
||
case constants.TaskTypePollingRealname:
|
||
if cfg.RealnameCheckInterval != nil {
|
||
interval = *cfg.RealnameCheckInterval
|
||
}
|
||
case constants.TaskTypePollingCarddata:
|
||
if cfg.CarddataCheckInterval != nil {
|
||
interval = *cfg.CarddataCheckInterval
|
||
}
|
||
case constants.TaskTypePollingPackage:
|
||
if cfg.PackageCheckInterval != nil {
|
||
interval = *cfg.PackageCheckInterval
|
||
}
|
||
case constants.TaskTypePollingProtect:
|
||
if cfg.ProtectCheckInterval != nil {
|
||
interval = *cfg.ProtectCheckInterval
|
||
}
|
||
}
|
||
if interval <= 0 {
|
||
return now
|
||
}
|
||
jitterMax := interval / 10
|
||
if jitterMax < 2 {
|
||
jitterMax = 2
|
||
}
|
||
return now.Add(time.Duration(rand.Intn(jitterMax)) * time.Second)
|
||
}
|