junhong_cmp_fiber/internal/polling/carrier_bridge.go

package polling

import (
	"context"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"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/polling/carrierqueue"
	"github.com/break/junhong_cmp_fiber/internal/store/postgres"
	"github.com/break/junhong_cmp_fiber/pkg/constants"
)

// CarrierScheduling 是按运营商账户调度在轮询包内的接缝：判断当前模式，并按卡计算各检查类型到期时间写入账户队列。
// 所有写入轮询结构的入口（初始化、生命周期、优先提示、手动触发）按当前模式写新结构或旧分片结构。
type CarrierScheduling struct {
	store     *carrierqueue.Store
	settings  *carrierqueue.SettingsProvider
	configMgr *PollingConfigManager
	logger    *zap.Logger
}

// NewCarrierScheduling 创建按运营商账户调度接缝。
func NewCarrierScheduling(store *carrierqueue.Store, settings *carrierqueue.SettingsProvider, configMgr *PollingConfigManager, logger *zap.Logger) *CarrierScheduling {
	if logger == nil {
		logger = zap.NewNop()
	}
	return &CarrierScheduling{store: store, settings: settings, configMgr: configMgr, logger: logger}
}

// Active 判断当前是否为按运营商账户调度。
func (c *CarrierScheduling) Active(ctx context.Context) bool {
	return c != nil && c.store != nil && c.settings != nil && c.settings.Current(ctx).CarrierMode()
}

// Store 返回账户队列存储。
func (c *CarrierScheduling) Store() *carrierqueue.Store {
	if c == nil {
		return nil
	}
	return c.store
}

// Settings 返回调度参数提供者。
func (c *CarrierScheduling) Settings() *carrierqueue.SettingsProvider {
	if c == nil {
		return nil
	}
	return c.settings
}

// DuesForCard 按卡匹配的轮询配置计算各检查类型下次到期时间。
// fromLastCheck 为 true 时按各类型上次检查时间计算（初始化）；否则按当前时间加抖动（新卡、恢复轮询、状态变化）。
// 间隔为 0 或未配置的类型不返回；重置日前加密窗口规则算出的时间更早时以它为准。
func (c *CarrierScheduling) DuesForCard(ctx context.Context, card *model.IotCard, fromLastCheck bool) map[string]time.Time {
	dues := map[string]time.Time{}
	if c == nil || c.configMgr == nil || card == nil {
		return dues
	}
	now := time.Now()
	for taskType, info := range c.configMgr.MergedTaskIntervals(card) {
		checkType := constants.CheckTypeFromTaskType(taskType)
		if checkType == "" || info.Interval <= 0 {
			continue
		}
		var due time.Time
		if fromLastCheck {
			due = calculateNextCheckTime(lastCheckAtByTaskType(card, taskType), info.Interval, now)
			if windowDue := c.configMgr.NextCheckAt(ctx, card, taskType, info, now); windowDue.Before(due) && windowDue.After(now) {
				due = windowDue
			}
		} else {
			due = calcInitialDelay(info.Interval)
		}
		dues[checkType] = due
	}
	return dues
}

// UpsertCard 把卡写入所属账户队列；fromLastCheck 含义同 DuesForCard。
func (c *CarrierScheduling) UpsertCard(ctx context.Context, card *model.IotCard, fromLastCheck bool) {
	if c == nil || c.store == nil || card == nil {
		return
	}
	if err := c.store.Upsert(ctx, card.ID, card.CarrierID, c.DuesForCard(ctx, card, fromLastCheck)); err != nil {
		c.logger.Warn("写入运营商账户轮询队列失败", zap.Uint("card_id", card.ID), zap.Uint("carrier_id", card.CarrierID), zap.Error(err))
	}
}

// UpsertCards 批量把卡写入所属账户队列（初始化按批调用）；fromLastCheck 含义同 DuesForCard。
func (c *CarrierScheduling) UpsertCards(ctx context.Context, cards []*model.IotCard, fromLastCheck bool) error {
	if c == nil || c.store == nil || len(cards) == 0 {
		return nil
	}
	items := make([]carrierqueue.UpsertItem, 0, len(cards))
	for _, card := range cards {
		if card == nil {
			continue
		}
		items = append(items, carrierqueue.UpsertItem{CardID: card.ID, CarrierID: card.CarrierID, Dues: c.DuesForCard(ctx, card, fromLastCheck)})
	}
	return c.store.UpsertBatch(ctx, items)
}

// RemoveCard 把卡移出所属账户队列；carrierID 为 0 时按到期信息中记录的账户移出。
func (c *CarrierScheduling) RemoveCard(ctx context.Context, cardID, carrierID uint) {
	if c == nil || c.store == nil {
		return
	}
	if carrierID == 0 {
		due, err := c.store.ReadDue(ctx, cardID)
		if err != nil || due.CarrierID == 0 {
			return
		}
		carrierID = due.CarrierID
	}
	if err := c.store.Remove(ctx, cardID, carrierID); err != nil {
		c.logger.Warn("移出运营商账户轮询队列失败", zap.Uint("card_id", cardID), zap.Error(err))
	}
}

// RequestTaskType 把卡的某检查类型（旧任务类型 polling:xxx）设为立即到期放入账户优先队列（优先提示、手动触发）。
// 卡不在轮询中时不处理。
func (c *CarrierScheduling) RequestTaskType(ctx context.Context, cardID, carrierID uint, taskType string) {
	checkType := constants.CheckTypeFromTaskType(taskType)
	if c == nil || c.store == nil || checkType == "" {
		return
	}
	if _, err := c.store.RequestCheck(ctx, carrierqueue.CheckRequest{CardID: cardID, CarrierID: carrierID, CheckTypes: []string{checkType}}); err != nil {
		c.logger.Warn("写入运营商账户优先队列失败", zap.Uint("card_id", cardID), zap.String("task_type", taskType), zap.Error(err))
	}
}

// CarrierInitializer 从数据库按批初始化全部运营商账户队列与卡到期信息；进度与旧初始化器同结构，供 init-progress 展示。
type CarrierInitializer struct {
	scheduling   *CarrierScheduling
	iotCardStore *postgres.IotCardStore
	logger       *zap.Logger

	mu       sync.RWMutex
	progress InitProgress
	running  atomic.Bool
}

// NewCarrierInitializer 创建新结构初始化器。
func NewCarrierInitializer(scheduling *CarrierScheduling, iotCardStore *postgres.IotCardStore, logger *zap.Logger) *CarrierInitializer {
	if logger == nil {
		logger = zap.NewNop()
	}
	return &CarrierInitializer{scheduling: scheduling, iotCardStore: iotCardStore, logger: logger}
}

// Progress 返回初始化进度快照。
func (i *CarrierInitializer) Progress() InitProgress {
	i.mu.RLock()
	defer i.mu.RUnlock()
	return i.progress
}

// Run 同步执行一次全量初始化：按批读取启用轮询的卡，计算各类型到期并写入账户队列。返回写入的卡数。
func (i *CarrierInitializer) Run(ctx context.Context) (int64, error) {
	if !i.running.CompareAndSwap(false, true) {
		return 0, nil
	}
	defer i.running.Store(false)
	started := time.Now()
	total, err := i.iotCardStore.CountForPolling(ctx)
	if err != nil {
		i.setProgress(func(p *InitProgress) { p.Status, p.ErrorMessage = "failed", err.Error() })
		return 0, err
	}
	i.setProgress(func(p *InitProgress) {
		*p = InitProgress{TotalCards: total, StartTime: started, Status: "running"}
	})
	const batchSize = 2000
	var lastID uint
	var written int64
	for {
		if err := ctx.Err(); err != nil {
			i.setProgress(func(p *InitProgress) { p.Status, p.ErrorMessage = "failed", err.Error() })
			return written, err
		}
		cards, fetchErr := i.iotCardStore.ListForPollingBatch(ctx, lastID, batchSize)
		if fetchErr != nil {
			i.setProgress(func(p *InitProgress) { p.Status, p.ErrorMessage = "failed", fetchErr.Error() })
			return written, fetchErr
		}
		if len(cards) == 0 {
			break
		}
		if tier := i.scheduling.configMgr.TierResolver(); tier != nil {
			tier.Prefill(ctx, cards)
		}
		if err := i.scheduling.UpsertCards(ctx, cards, true); err != nil {
			i.setProgress(func(p *InitProgress) { p.Status, p.ErrorMessage = "failed", err.Error() })
			return written, err
		}
		written += int64(len(cards))
		lastID = cards[len(cards)-1].ID
		loaded := written
		i.setProgress(func(p *InitProgress) { p.LoadedCards, p.LastBatchTime = loaded, time.Now() })
	}
	i.setProgress(func(p *InitProgress) { p.Status = "completed" })
	i.logger.Info("运营商账户轮询结构初始化完成", zap.Int64("cards", written), zap.Duration("duration", time.Since(started)))
	return written, nil
}

func (i *CarrierInitializer) setProgress(update func(*InitProgress)) {
	i.mu.Lock()
	update(&i.progress)
	i.mu.Unlock()
}

// AdvanceDue 把卡某检查类型的到期时间提前到 due（只提前不推后），并同步把账户队列中的 score 提前；返回是否提前。
// 卡不在轮询中（无到期信息）时不处理。
func (c *CarrierScheduling) AdvanceDue(ctx context.Context, card *model.IotCard, checkType string, due time.Time) (bool, error) {
	if c == nil || c.store == nil || card == nil {
		return false, nil
	}
	current, err := c.store.ReadDue(ctx, card.ID)
	if err != nil || current.CarrierID == 0 {
		return false, err
	}
	item, ok := current.Types[checkType]
	if !ok || !item.HasDue || !due.Before(item.Due) {
		return false, nil
	}
	client := c.store.Client()
	member := strconv.FormatUint(uint64(card.ID), 10)
	pipe := client.TxPipeline()
	pipe.HSet(ctx, constants.RedisCardDueKey(card.ID), carrierqueue.FieldDue(checkType), due.UnixMilli())
	score := redis.ZAddArgs{XX: true, LT: true, Members: []redis.Z{{Score: float64(due.UnixMilli()), Member: member}}}
	pipe.ZAddArgs(ctx, constants.RedisCarrierQueueKey(current.CarrierID), score)
	pipe.ZAddArgs(ctx, constants.RedisCarrierPriorityQueueKey(current.CarrierID), score)
	_, err = pipe.Exec(ctx)
	return err == nil, err
}