perf: 运营商账户轮询初始化按批写入 Redis

初始化每批 2000 张卡，原先逐张写入账户队列，每张卡约 4 次 Redis 往返；
测试环境 Redis 走外网，约 750 张/分钟，9.7 万张卡要 2 小时以上，期间未加载的卡不轮询。
新增 Store.UpsertBatch：一次往返读出各卡原账户与优先队列状态，一次往返写入整批；
单卡 Upsert 复用同一逻辑，写入规则不变。批量写入失败时本轮初始化失败、下一轮重试。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/polling/carrier_bridge.go

@@ -92,6 +92,21 @@ func (c *CarrierScheduling) UpsertCard(ctx context.Context, card *model.IotCard,
 	}
 }
 
+// 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 {
@@ -181,10 +196,11 @@ func (i *CarrierInitializer) Run(ctx context.Context) (int64, error) {
 		if tier := i.scheduling.configMgr.TierResolver(); tier != nil {
 			tier.Prefill(ctx, cards)
 		}
-		for _, card := range cards {
-			i.scheduling.UpsertCard(ctx, card, true)
-			written++
+		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() })

internal/polling/carrierqueue/store.go

@@ -173,27 +173,63 @@ func (s *Store) Remove(ctx context.Context, cardID, carrierID uint) error {
 // Upsert 写入卡各检查类型到期时间并按最早到期放入账户队列（已在优先队列中的保留在优先队列）。
 // 卡所属账户变化时先移出旧账户队列。dues 为空表示该卡没有任何需要执行的检查类型，移出队列。
 func (s *Store) Upsert(ctx context.Context, cardID, carrierID uint, dues map[string]time.Time) error {
-	dueKey := constants.RedisCardDueKey(cardID)
-	previous, err := s.redis.HGet(ctx, dueKey, fieldCarrier).Uint64()
-	if err != nil && !stderrors.Is(err, redis.Nil) {
+	return s.UpsertBatch(ctx, []UpsertItem{{CardID: cardID, CarrierID: carrierID, Dues: dues}})
+}
+
+// UpsertItem 是批量写入的一张卡：所属账户与各检查类型到期时间（含义同 Upsert）。
+type UpsertItem struct {
+	CardID    uint
+	CarrierID uint
+	Dues      map[string]time.Time
+}
+
+// UpsertBatch 按 Upsert 的规则批量写入：先一次往返读出各卡原账户与是否在优先队列，再一次往返写入全部卡。
+// 初始化按批调用，避免每张卡多次往返 Redis。
+func (s *Store) UpsertBatch(ctx context.Context, items []UpsertItem) error {
+	if len(items) == 0 {
+		return nil
+	}
+	read := s.redis.Pipeline()
+	previousCmds := make([]*redis.StringCmd, len(items))
+	priorityCmds := make([]*redis.FloatCmd, len(items))
+	for i, item := range items {
+		previousCmds[i] = read.HGet(ctx, constants.RedisCardDueKey(item.CardID), fieldCarrier)
+		priorityCmds[i] = read.ZScore(ctx, constants.RedisCarrierPriorityQueueKey(item.CarrierID), strconv.FormatUint(uint64(item.CardID), 10))
+	}
+	if _, err := read.Exec(ctx); err != nil && !stderrors.Is(err, redis.Nil) {
 		return err
 	}
-	pipe := s.redis.TxPipeline()
-	if previous != 0 && uint(previous) != carrierID {
-		pipe.ZRem(ctx, constants.RedisCarrierPriorityQueueKey(uint(previous)), cardID)
-		pipe.ZRem(ctx, constants.RedisCarrierQueueKey(uint(previous)), cardID)
+	write := s.redis.Pipeline()
+	for i, item := range items {
+		previous, err := previousCmds[i].Uint64()
+		if err != nil && !stderrors.Is(err, redis.Nil) {
+			return err
+		}
+		inPriority, _ := priorityCmds[i].Result()
+		queueUpsert(ctx, write, item, uint(previous), inPriority > 0)
 	}
-	if len(dues) == 0 {
-		pipe.ZRem(ctx, constants.RedisCarrierPriorityQueueKey(carrierID), cardID)
-		pipe.ZRem(ctx, constants.RedisCarrierQueueKey(carrierID), cardID)
+	_, err := write.Exec(ctx)
+	return err
+}
+
+// queueUpsert 把一张卡的写入命令加入管道：账户变化时移出旧账户队列；dues 为空移出队列；
+// 否则写到期 Hash，并按最早到期放入账户队列（已在优先队列中的保留在优先队列）。
+func queueUpsert(ctx context.Context, pipe redis.Pipeliner, item UpsertItem, previous uint, inPriority bool) {
+	dueKey := constants.RedisCardDueKey(item.CardID)
+	if previous != 0 && previous != item.CarrierID {
+		pipe.ZRem(ctx, constants.RedisCarrierPriorityQueueKey(previous), item.CardID)
+		pipe.ZRem(ctx, constants.RedisCarrierQueueKey(previous), item.CardID)
+	}
+	if len(item.Dues) == 0 {
+		pipe.ZRem(ctx, constants.RedisCarrierPriorityQueueKey(item.CarrierID), item.CardID)
+		pipe.ZRem(ctx, constants.RedisCarrierQueueKey(item.CarrierID), item.CardID)
 		pipe.Del(ctx, dueKey)
-		_, err := pipe.Exec(ctx)
-		return err
+		return
 	}
-	values := map[string]any{fieldCarrier: carrierID}
+	values := map[string]any{fieldCarrier: item.CarrierID}
 	var earliest time.Time
 	for _, checkType := range constants.AllCheckTypes() {
-		due, ok := dues[checkType]
+		due, ok := item.Dues[checkType]
 		if !ok {
 			pipe.HDel(ctx, dueKey, fieldDue+checkType)
 			continue
@@ -204,15 +240,11 @@ func (s *Store) Upsert(ctx context.Context, cardID, carrierID uint, dues map[str
 		}
 	}
 	pipe.HSet(ctx, dueKey, values)
-	_, err = pipe.Exec(ctx)
-	if err != nil {
-		return err
+	queueKey := constants.RedisCarrierQueueKey(item.CarrierID)
+	if inPriority {
+		queueKey = constants.RedisCarrierPriorityQueueKey(item.CarrierID)
 	}
-	priorityKey := constants.RedisCarrierPriorityQueueKey(carrierID)
-	if inPriority, _ := s.redis.ZScore(ctx, priorityKey, strconv.FormatUint(uint64(cardID), 10)).Result(); inPriority > 0 {
-		return s.redis.ZAdd(ctx, priorityKey, redis.Z{Score: float64(earliest.UnixMilli()), Member: cardID}).Err()
-	}
-	return s.redis.ZAdd(ctx, constants.RedisCarrierQueueKey(carrierID), redis.Z{Score: float64(earliest.UnixMilli()), Member: cardID}).Err()
+	pipe.ZAdd(ctx, queueKey, redis.Z{Score: float64(earliest.UnixMilli()), Member: item.CardID})
 }
 
 // CardDue 是卡到期信息的解析结果。