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

// 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))
	}
	// 清理轮询卡缓存，确保下次轮询从 DB 读取最新状态
	s.queueMgr.InvalidateCardCache(ctx, cardID)

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

// OnBatchCardsStatusChanged 批量卡状态变化：单次 Redis pipeline 清理 + 单次 DB 批量查询
// 替代 N 次 OnCardStatusChanged 调用，避免批量分配/回收时打满 DB 连接池
func (s *PollingLifecycleService) OnBatchCardsStatusChanged(ctx context.Context, cardIDs []uint) {
	if len(cardIDs) == 0 {
		return
	}
	// 单次 pipeline：批量从当前分片队列移除并清理缓存
	if err := s.queueMgr.RemoveBatchFromCurrentShardQueues(ctx, cardIDs); err != nil {
		s.logger.Warn("批量卡状态变化：从队列移除失败", zap.Int("count", len(cardIDs)), zap.Error(err))
	}
	// 单次 DB 查询：批量加载所有卡信息
	cards, err := s.iotCardStore.GetByIDs(ctx, cardIDs)
	if err != nil {
		s.logger.Error("批量卡状态变化：加载卡信息失败", zap.Int("count", len(cardIDs)), zap.Error(err))
		return
	}
	for _, card := range cards {
		if !s.shouldEnqueue(ctx, card) {
			continue
		}
		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) {
	s.queueMgr.InvalidateCardCache(ctx, cardID)
	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))
	}
	s.queueMgr.InvalidateCardCache(ctx, cardID)
}

// 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) {
	intervals := s.configMgr.MergedTaskIntervals(card)
	if len(intervals) == 0 {
		return
	}
	for taskType, info := range intervals {
		if err := s.queueMgr.Requeue(ctx, card.ID, taskType, calcInitialDelay(info.Interval)); err != nil {
			s.logger.Warn("卡入队失败",
				zap.Uint("card_id", card.ID), zap.String("task_type", taskType), zap.Error(err))
		}
	}
}

// calcInitialDelay 计算新入队卡的初始检查时间（加随机抖动避免惊群）
// 抖动范围 [0, max(interval/10, 2)) 秒；下界取 2 保证 rand.Intn 有实际随机性
// （rand.Intn(1) 永远返回 0，对 interval≤10 的配置会导致零抖动）
func calcInitialDelay(interval int) time.Time {
	now := time.Now()
	if interval <= 0 {
		return now
	}
	jitterMax := interval / 10
	if jitterMax < 2 {
		jitterMax = 2
	}
	return now.Add(time.Duration(rand.Intn(jitterMax)) * time.Second)
}