junhong_cmp_fiber/internal/service/polling/asset_polling_service.go

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")
	}
}