package polling import ( "context" "go.uber.org/zap" "github.com/break/junhong_cmp_fiber/internal/polling" assetAuditSvc "github.com/break/junhong_cmp_fiber/internal/service/asset_audit" 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 assetAuditService assetAuditSvc.OperationLogger } // NewAssetPollingService 创建资产轮询管控服务 func NewAssetPollingService( deviceStore *postgres.DeviceStore, deviceBindingStore *postgres.DeviceSimBindingStore, iotCardService *iotCardSvc.Service, queueMgr *polling.PollingQueueManager, logger *zap.Logger, assetAuditService assetAuditSvc.OperationLogger, ) *AssetPollingService { return &AssetPollingService{ deviceStore: deviceStore, deviceBindingStore: deviceBindingStore, iotCardService: iotCardService, queueMgr: queueMgr, logger: logger, assetAuditService: assetAuditService, } } func (s *AssetPollingService) logAssetPollingAudit(ctx context.Context, p assetAuditSvc.BuildLogParams) { if s == nil || s.assetAuditService == nil { return } if p.OperationType == "" { p.OperationType = constants.AssetAuditOpAssetPollingStatus } if p.Operator.Type == "" { p.Operator = assetAuditSvc.OperatorFromContext(ctx) } s.assetAuditService.LogOperation(ctx, assetAuditSvc.BuildLog(ctx, p)) } // 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: beforeData := map[string]any{ "asset_type": constants.AssetTypeIotCard, "asset_id": assetID, "enable_polling": "unknown", "source_service": "asset_polling", } afterData := map[string]any{ "asset_type": constants.AssetTypeIotCard, "asset_id": assetID, "enable_polling": enablePolling, } // S2 修复:委托给 IotCardService,确保 DB 写入 + PollingCallback 回调一并触发 if err := s.iotCardService.UpdatePollingStatus(ctx, assetID, enablePolling); err != nil { errorCode, errorMsg := assetAuditSvc.BuildErrorInfo(err) s.logAssetPollingAudit(ctx, assetAuditSvc.BuildLogParams{ AssetType: constants.AssetTypeIotCard, AssetID: assetID, OperationDesc: "统一入口更新轮询状态失败", ResultStatus: constants.AssetAuditResultFailed, ErrorCode: errorCode, ErrorMsg: errorMsg, BeforeData: beforeData, AfterData: afterData, }) return err } s.logAssetPollingAudit(ctx, assetAuditSvc.BuildLogParams{ AssetType: constants.AssetTypeIotCard, AssetID: assetID, OperationDesc: "统一入口更新轮询状态", ResultStatus: constants.AssetAuditResultSuccess, BeforeData: beforeData, AfterData: afterData, }) return nil case constants.AssetTypeDevice: device, getErr := s.deviceStore.GetByID(ctx, assetID) if getErr != nil { errorCode, errorMsg := assetAuditSvc.BuildErrorInfo(getErr) s.logAssetPollingAudit(ctx, assetAuditSvc.BuildLogParams{ AssetType: constants.AssetTypeDevice, AssetID: assetID, OperationDesc: "统一入口更新轮询状态失败", ResultStatus: constants.AssetAuditResultFailed, ErrorCode: errorCode, ErrorMsg: errorMsg, AfterData: map[string]any{ "asset_type": constants.AssetTypeDevice, "asset_id": assetID, "enable_polling": enablePolling, }, }) return getErr } beforeData := map[string]any{ "asset_type": constants.AssetTypeDevice, "asset_id": device.ID, "asset_identifier": device.VirtualNo, "enable_polling": device.EnablePolling, } afterData := map[string]any{ "asset_type": constants.AssetTypeDevice, "asset_id": device.ID, "asset_identifier": device.VirtualNo, "enable_polling": enablePolling, } // 1. 更新设备的 enable_polling 字段 if err := s.deviceStore.UpdatePollingStatus(ctx, assetID, enablePolling); err != nil { errorCode, errorMsg := assetAuditSvc.BuildErrorInfo(err) s.logAssetPollingAudit(ctx, assetAuditSvc.BuildLogParams{ AssetType: constants.AssetTypeDevice, AssetID: device.ID, AssetIdentifier: device.VirtualNo, OperationDesc: "统一入口更新轮询状态失败", ResultStatus: constants.AssetAuditResultFailed, ErrorCode: errorCode, ErrorMsg: errorMsg, BeforeData: beforeData, AfterData: afterData, }) 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)) } } } s.logAssetPollingAudit(ctx, assetAuditSvc.BuildLogParams{ AssetType: constants.AssetTypeDevice, AssetID: device.ID, AssetIdentifier: device.VirtualNo, OperationDesc: "统一入口更新轮询状态", ResultStatus: constants.AssetAuditResultSuccess, BeforeData: beforeData, AfterData: afterData, }) return nil default: err := errors.New(errors.CodeInvalidParam, "资产类型无效,支持 card 或 device") errorCode, errorMsg := assetAuditSvc.BuildErrorInfo(err) s.logAssetPollingAudit(ctx, assetAuditSvc.BuildLogParams{ AssetType: assetType, AssetID: assetID, OperationDesc: "统一入口更新轮询状态被拒绝", ResultStatus: constants.AssetAuditResultDenied, ErrorCode: errorCode, ErrorMsg: errorMsg, AfterData: map[string]any{ "asset_type": assetType, "asset_id": assetID, "enable_polling": enablePolling, }, }) return err } }