junhong_cmp_fiber/internal/polling/config_manager.go

package polling

import (
	"context"
	"sync"
	"time"

	"github.com/bytedance/sonic"
	"github.com/redis/go-redis/v9"
	"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"
)

// PollingConfigManager 轮询配置管理器
// 从 DB 加载 tb_polling_config，同步到 Redis Hash，内存缓存（读写锁）
// 5 分钟定时自动刷新；加载失败时保留原缓存不清空
type PollingConfigManager struct {
	configStore *postgres.PollingConfigStore
	redis       *redis.Client
	logger      *zap.Logger

	mu          sync.RWMutex
	configs     []*model.PollingConfig
	refreshOnce sync.Once
}

// NewPollingConfigManager 创建配置管理器
func NewPollingConfigManager(configStore *postgres.PollingConfigStore, redisClient *redis.Client, logger *zap.Logger) *PollingConfigManager {
	return &PollingConfigManager{
		configStore: configStore,
		redis:       redisClient,
		logger:      logger,
	}
}

// Load 加载配置到内存缓存并同步到 Redis
// 加载失败时保留原缓存不清空，确保可用性
func (m *PollingConfigManager) Load(ctx context.Context) error {
	configs, err := m.configStore.ListEnabled(ctx)
	if err != nil {
		m.logger.Error("加载轮询配置失败", zap.Error(err))
		return err
	}

	m.mu.Lock()
	m.configs = configs
	m.mu.Unlock()

	if syncErr := m.syncToRedis(ctx, configs); syncErr != nil {
		m.logger.Warn("同步配置到 Redis 失败", zap.Error(syncErr))
	}

	m.logger.Info("轮询配置已加载", zap.Int("count", len(configs)))
	return nil
}

// Start 启动定时刷新（每 5 分钟自动 Load 一次）
// 使用 sync.Once 确保只启动一个刷新 goroutine
func (m *PollingConfigManager) Start(ctx context.Context) {
	m.refreshOnce.Do(func() {
		go func() {
			ticker := time.NewTicker(5 * time.Minute)
			defer ticker.Stop()
			m.logger.Info("配置管理器已启动，5分钟自动刷新")
			for {
				select {
				case <-ctx.Done():
					return
				case <-ticker.C:
					if err := m.Load(ctx); err != nil {
						m.logger.Warn("定时刷新轮询配置失败，保留原配置", zap.Error(err))
					}
				}
			}
		}()
	})
}

// MatchConfig 按优先级匹配第一个符合条件的轮询配置
// 配置按 priority ASC 排序（DB 层保证），数字越小优先级越高
func (m *PollingConfigManager) MatchConfig(card *model.IotCard) *model.PollingConfig {
	m.mu.RLock()
	defer m.mu.RUnlock()

	for _, cfg := range m.configs {
		if matchConfigConditions(cfg, card) {
			return cfg
		}
	}
	return nil
}

// matchConfigConditions 检查卡是否满足配置的匹配条件
func matchConfigConditions(cfg *model.PollingConfig, card *model.IotCard) bool {
	if cfg.CardCondition != "" {
		if cfg.CardCondition != getCardCondition(card) {
			return false
		}
	}
	if cfg.CardCategory != "" {
		if cfg.CardCategory != card.CardCategory {
			return false
		}
	}
	if cfg.CarrierID != nil {
		if *cfg.CarrierID != card.CarrierID {
			return false
		}
	}
	return true
}

// getCardCondition 获取卡的状态条件（用于匹配轮询配置）
// ⚠️ 注意判断顺序：停机优先，避免停机卡错误匹配 activated 或 not_real_name 配置
// 停机卡需要继续轮询 carddata/package 以检测复机条件（套餐购买、流量重置、实名完成）
func getCardCondition(card *model.IotCard) string {
	if card.NetworkStatus == constants.NetworkStatusOffline {
		return "suspended"
	}
	if card.RealNameStatus != constants.RealNameStatusVerified {
		return "not_real_name"
	}
	return "activated"
}

// syncToRedis 将配置同步到 Redis Hash（TTL 24h）
func (m *PollingConfigManager) syncToRedis(ctx context.Context, configs []*model.PollingConfig) error {
	if len(configs) == 0 {
		return nil
	}
	key := constants.RedisPollingConfigsCacheKey()
	configData := make([]interface{}, 0, len(configs)*2)
	for _, cfg := range configs {
		jsonData, err := sonic.Marshal(cfg)
		if err != nil {
			m.logger.Warn("序列化轮询配置失败", zap.Uint("config_id", cfg.ID), zap.Error(err))
			continue
		}
		configData = append(configData, cfg.ID, string(jsonData))
	}
	if len(configData) == 0 {
		return nil
	}
	pipe := m.redis.Pipeline()
	pipe.HSet(ctx, key, configData...)
	pipe.Expire(ctx, key, 24*time.Hour)
	_, err := pipe.Exec(ctx)
	return err
}