break
c64f3d8b80
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 8m31s
181 lines
7.1 KiB
Go
181 lines
7.1 KiB
Go
package polling
|
||
|
||
import (
|
||
"context"
|
||
stderrors "errors"
|
||
"strconv"
|
||
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
|
||
auditinfra "github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"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/auditcontext"
|
||
"github.com/break/junhong_cmp_fiber/pkg/auditfailure"
|
||
"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 {
|
||
db *gorm.DB
|
||
deviceStore *postgres.DeviceStore
|
||
deviceBindingStore *postgres.DeviceSimBindingStore
|
||
iotCardService *iotCardSvc.Service
|
||
queueMgr *polling.PollingQueueManager
|
||
logger *zap.Logger
|
||
auditWriter *auditinfra.Writer
|
||
}
|
||
|
||
// NewAssetPollingService 创建资产轮询管控服务
|
||
func NewAssetPollingService(
|
||
db *gorm.DB,
|
||
deviceStore *postgres.DeviceStore,
|
||
deviceBindingStore *postgres.DeviceSimBindingStore,
|
||
iotCardService *iotCardSvc.Service,
|
||
queueMgr *polling.PollingQueueManager,
|
||
logger *zap.Logger,
|
||
auditWriter *auditinfra.Writer,
|
||
) *AssetPollingService {
|
||
return &AssetPollingService{
|
||
db: db,
|
||
deviceStore: deviceStore,
|
||
deviceBindingStore: deviceBindingStore,
|
||
iotCardService: iotCardService,
|
||
queueMgr: queueMgr,
|
||
logger: logger,
|
||
auditWriter: auditWriter,
|
||
}
|
||
}
|
||
|
||
// 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:
|
||
device, getErr := s.deviceStore.GetByID(ctx, assetID)
|
||
if getErr != nil {
|
||
appErr := errors.Wrap(errors.CodeDatabaseError, getErr, "查询设备失败")
|
||
result := constants.AuditResultFailed
|
||
if stderrors.Is(getErr, gorm.ErrRecordNotFound) {
|
||
appErr = errors.New(errors.CodeNotFound, "设备不存在")
|
||
result = constants.AuditResultDenied
|
||
}
|
||
s.recordDevicePollingFailure(ctx, &model.Device{Model: gorm.Model{ID: assetID}}, enablePolling, result, appErr)
|
||
return appErr
|
||
}
|
||
if err := s.updateDevicePollingStatus(ctx, device, enablePolling); err != nil {
|
||
s.recordDevicePollingFailure(ctx, device, enablePolling, constants.AuditResultFailed, err)
|
||
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")
|
||
}
|
||
}
|
||
|
||
func (s *AssetPollingService) updateDevicePollingStatus(ctx context.Context, device *model.Device, enablePolling bool) error {
|
||
if s.db == nil || s.auditWriter == nil {
|
||
return errors.New(errors.CodeInvalidStatus, "设备统一审计接缝未配置")
|
||
}
|
||
return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
if device.EnablePolling != enablePolling {
|
||
result := tx.Model(&model.Device{}).Where("id = ? AND enable_polling = ?", device.ID, device.EnablePolling).Update("enable_polling", enablePolling)
|
||
if result.Error != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, result.Error, "更新设备轮询状态失败")
|
||
}
|
||
if result.RowsAffected == 0 {
|
||
return errors.New(errors.CodeConflict, "轮询状态已变更，请刷新后重试")
|
||
}
|
||
}
|
||
return s.appendDevicePollingAudit(ctx, tx, device, enablePolling, constants.AuditResultSuccess, nil)
|
||
})
|
||
}
|
||
|
||
func (s *AssetPollingService) appendDevicePollingAudit(ctx context.Context, tx *gorm.DB, device *model.Device, enablePolling bool, result string, businessErr error) error {
|
||
if device == nil || device.ID == 0 {
|
||
return errors.New(errors.CodeInvalidStatus, "设备审计资源不完整")
|
||
}
|
||
id := strconv.FormatUint(uint64(device.ID), 10)
|
||
errorCode, errorSummary := auditErrorInfo(businessErr)
|
||
return s.auditWriter.Append(ctx, tx, auditinfra.AppendInput{
|
||
ActionCode: constants.AuditActionDevicePollingStatusUpdated,
|
||
Summary: "更新设备轮询开关",
|
||
Result: result,
|
||
ErrorCode: errorCode,
|
||
ErrorSummary: errorSummary,
|
||
ScopeType: constants.AuditScopePlatform,
|
||
Resources: []auditinfra.ResourceInput{{
|
||
Type: constants.AuditResourceDevice, ID: &id, Key: auditinfra.DeviceResourceKey(device), DisplayName: device.VirtualNo,
|
||
Relation: constants.AuditResourceRelationPrimary, Role: constants.AuditResourceRoleDeviceTarget,
|
||
IdentitySnapshot: auditinfra.DeviceIdentitySnapshot(device),
|
||
BeforeData: map[string]any{"enable_polling": device.EnablePolling}, AfterData: map[string]any{"enable_polling": enablePolling},
|
||
SubjectVisibility: constants.AuditSubjectResult, SubjectSummary: "设备轮询开关已更新",
|
||
}},
|
||
})
|
||
}
|
||
|
||
func (s *AssetPollingService) recordDevicePollingFailure(ctx context.Context, device *model.Device, enablePolling bool, result string, businessErr error) {
|
||
if s == nil || s.db == nil || s.auditWriter == nil || device == nil || device.ID == 0 {
|
||
return
|
||
}
|
||
err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
return s.appendDevicePollingAudit(ctx, tx, device, enablePolling, result, businessErr)
|
||
})
|
||
if err == nil {
|
||
return
|
||
}
|
||
linkage := auditcontext.From(ctx)
|
||
errorCode, _ := auditErrorInfo(businessErr)
|
||
auditfailure.RecordSecondaryWriteFailure(constants.AuditActionDevicePollingStatusUpdated, strconv.FormatUint(uint64(device.ID), 10), linkage.RequestID, linkage.CorrelationID, errorCode, err)
|
||
}
|
||
|
||
func auditErrorInfo(err error) (string, string) {
|
||
if err == nil {
|
||
return "", ""
|
||
}
|
||
var appErr *errors.AppError
|
||
if stderrors.As(err, &appErr) {
|
||
return strconv.Itoa(appErr.Code), appErr.Error()
|
||
}
|
||
return "", err.Error()
|
||
}
|