package polling import ( "context" "go.uber.org/zap" "github.com/break/junhong_cmp_fiber/internal/polling" iotCardSvc "github.com/break/junhong_cmp_fiber/internal/service/iot_card" "github.com/break/junhong_cmp_fiber/internal/store/postgres" "github.com/break/junhong_cmp_fiber/pkg/constants" "github.com/break/junhong_cmp_fiber/pkg/errors" ) // AssetPollingService 资产轮询管控服务 // 管理 IoT 卡和设备的轮询启用状态 // S2 修复:card 类型委托给 IotCardService(含 DB 写入 + callback 通知),避免绕过生命周期 type AssetPollingService struct { deviceStore *postgres.DeviceStore deviceBindingStore *postgres.DeviceSimBindingStore iotCardService *iotCardSvc.Service queueMgr *polling.PollingQueueManager logger *zap.Logger } // NewAssetPollingService 创建资产轮询管控服务 func NewAssetPollingService( deviceStore *postgres.DeviceStore, deviceBindingStore *postgres.DeviceSimBindingStore, iotCardService *iotCardSvc.Service, queueMgr *polling.PollingQueueManager, logger *zap.Logger, ) *AssetPollingService { return &AssetPollingService{ deviceStore: deviceStore, deviceBindingStore: deviceBindingStore, iotCardService: iotCardService, queueMgr: queueMgr, logger: logger, } } // UpdatePollingStatus 更新资产轮询状态 // assetType: "card" 或 "device" // assetID: 资产ID // enablePolling: 是否启用轮询 func (s *AssetPollingService) UpdatePollingStatus(ctx context.Context, assetType string, assetID uint, enablePolling bool) error { switch assetType { case constants.AssetTypeIotCard: // S2 修复:委托给 IotCardService,确保 DB 写入 + PollingCallback 回调一并触发 return s.iotCardService.UpdatePollingStatus(ctx, assetID, enablePolling) case constants.AssetTypeDevice: // 1. 更新设备的 enable_polling 字段 if err := s.deviceStore.UpdatePollingStatus(ctx, assetID, enablePolling); err != nil { return err } bindings, err := s.deviceBindingStore.ListByDeviceID(ctx, assetID) if err != nil { s.logger.Warn("查询设备绑定卡失败,绑定卡轮询状态同步可能不完整", zap.Uint("device_id", assetID), zap.Error(err)) return nil } if len(bindings) == 0 { return nil } cardIDs := make([]uint, len(bindings)) for i, b := range bindings { cardIDs[i] = b.IotCardID } // 2. M3 修复:级联同步绑定卡的 enable_polling,防止生命周期事件绕过设备级设置 if syncErr := s.iotCardService.BatchUpdatePollingStatus(ctx, cardIDs, enablePolling); syncErr != nil { s.logger.Warn("批量同步绑定卡轮询状态失败", zap.Uint("device_id", assetID), zap.Error(syncErr)) } if !enablePolling { for _, b := range bindings { if rmErr := s.queueMgr.RemoveFromAllQueues(ctx, b.IotCardID); rmErr != nil { s.logger.Warn("从队列移除卡失败", zap.Uint("card_id", b.IotCardID), zap.Error(rmErr)) } } } return nil default: return errors.New(errors.CodeInvalidParam, "资产类型无效,支持 card 或 device") } }