junhong_cmp_fiber/internal/polling/lifecycle_service.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))
	}
	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) {
	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))
	}
}

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