feat: 卡流量读数先过采集队列再入账，扣减按采集时刻确定套餐

- 每卡 Redis 采集队列（默认 N=3、跨天清空、回退容差 1 MB），攒满且不下降才写最后一个点
- 测试环境全量切换日快照计量，下线灰度名单、切换水位与卡表基线扣费；全部卡置待建起点
- 读数区分没给值与 0；有套餐的卡读数 0 也写快照与日用量 0 行；重置日比较基准为 0 照常防旧值
- 新增采集记录（按月分区保留 90 天）、人工冲正流水与后台接口、采集告警
- 扣减按采集时刻有效套餐查找，只对生效中套餐做状态迁移，查不到套餐按成功跳过
- 换货新卡（新设备下的卡）打待建起点；采集间隔上限只校验启用中的轮询配置

Change: add-card-traffic-queue-and-observed-deduction

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

internal/infrastructure/cardobservation/event.go

@@ -14,6 +14,7 @@ import (
 	"gorm.io/gorm/clause"
 
 	cardapp "github.com/break/junhong_cmp_fiber/internal/application/cardobservation"
+	"github.com/break/junhong_cmp_fiber/internal/domain/cardflow"
 	"github.com/break/junhong_cmp_fiber/internal/infrastructure/messaging/outbox"
 	"github.com/break/junhong_cmp_fiber/internal/model"
 	"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
@@ -111,7 +112,7 @@ type RealnameActivator interface {
 
 // TrafficDeductor 按卡流量正增量扣减套餐流量。
 type TrafficDeductor interface {
-	DeductDataUsage(ctx context.Context, carrierType string, carrierID uint, usageMB float64) error
+	DeductDataUsage(ctx context.Context, cardID uint, usageMB float64, observedAt time.Time, observationID string) error
 }
 
 // StopResumeEvaluator 根据卡的当前事实执行停复机评估。
@@ -258,7 +259,7 @@ func (c *TrafficIncrementedConsumer) Consume(ctx context.Context, envelope outbo
 		return nil
 	}
 	if receipt.Status == constants.CardObservationEffectStatusProcessing {
-		dailyKey := constants.RedisCardDailyTrafficKey(event.CardID, event.ObservedAt.Format("2006-01-02"))
+		dailyKey := constants.RedisCardDailyTrafficKey(event.CardID, cardflow.ShanghaiDate(event.ObservedAt).Format("2006-01-02"))
 		pipe := c.redis.TxPipeline()
 		pipe.IncrByFloat(ctx, dailyKey, event.IncrementMB)
 		pipe.Expire(ctx, dailyKey, 48*time.Hour)
@@ -274,7 +275,7 @@ func (c *TrafficIncrementedConsumer) Consume(ctx context.Context, envelope outbo
 		if err := c.updateTrafficEffectStatus(ctx, receipt.ID, constants.CardObservationEffectStatusDailySaved, constants.CardObservationEffectStatusProcessing); err != nil {
 			return err
 		}
-		if err := c.deductor.DeductDataUsage(ctx, constants.AssetTypeIotCard, event.CardID, event.IncrementMB); err != nil {
+		if err := c.deductor.DeductDataUsage(ctx, event.CardID, event.IncrementMB, event.ObservedAt, event.ObservationID); err != nil {
 			return err
 		}
 		if err := c.updateTrafficEffectStatus(ctx, receipt.ID, constants.CardObservationEffectStatusProcessing, constants.CardObservationEffectStatusDeducted); err != nil {

internal/infrastructure/cardobservation/pre_activation_settlement.go

@@ -55,9 +55,6 @@ func (s *PreActivationSettler) SettleBeforeActivation(ctx context.Context, carri
 	}
 	deadline := time.Now().Add(constants.PreActivationSettlementTotalTimeoutSeconds * time.Second)
 	for _, card := range cards {
-		if !s.observation.TrafficSwitched(card.ID) {
-			continue
-		}
 		if time.Now().After(deadline) {
 			s.alert(card, carrierType, carrierID, "整次结算达到总超时，跳过该卡")
 			continue
@@ -88,7 +85,7 @@ func (s *PreActivationSettler) settleCard(ctx context.Context, card *model.IotCa
 	}
 	observedAt := time.Now().UTC()
 	decision, err := s.observation.ApplyTrafficObservation(stepCtx, carddomain.TrafficObservation{
-		CardID: card.ID, GatewayReadingMB: float64(response.Used), ResetDay: resetDay,
+		CardID: card.ID, GatewayReadingMB: float64(response.Used), ReadingResult: flowReadingResult(response), ResetDay: resetDay,
 		Metadata: carddomain.ObservationMetadata{
 			ObservationID: "pre-activation-" + card.ICCID + "-" + observedAt.Format("20060102150405.000"),
 			Source:        constants.CardObservationSourceBusinessEvent, Scene: constants.CardObservationScenePreActivationSettlement,
@@ -102,6 +99,7 @@ func (s *PreActivationSettler) settleCard(ctx context.Context, card *model.IotCa
 	s.logger.Info("激活前流量结算完成", zap.Uint("card_id", card.ID), zap.String("iccid", card.ICCID),
 		zap.String("carrier_type", carrierType), zap.Uint("asset_id", carrierID),
 		zap.Float64("gateway_flow_mb", float64(response.Used)), zap.Float64("increment_mb", decision.IncrementMB),
+		zap.String("result", decision.Result), zap.String("decision", decision.Decision),
 		zap.String("metric", "package.pre_activation_settlement.done"))
 }
 

internal/infrastructure/cardobservation/series_runner.go

@@ -558,7 +558,8 @@ func (r *SeriesRunner) queryAndApply(ctx context.Context, payload cardapp.Series
 			return cardapp.RunResult{}, err
 		}
 		decision, err := r.observation.ApplyTrafficObservation(ctx, carddomain.TrafficObservation{
-			CardID: card.ID, GatewayReadingMB: float64(response.Used), ResetDay: r.carrier.GetDataResetDay(ctx, card.CarrierID), Metadata: metadata,
+			CardID: card.ID, GatewayReadingMB: float64(response.Used), ReadingResult: flowReadingResult(response),
+			ResetDay: r.carrier.GetDataResetDay(ctx, card.CarrierID), Metadata: metadata,
 		})
 		return cardapp.RunResult{StateChanged: decision.IncrementMB > 0 || decision.CrossMonth}, err
 	case constants.CardObservationSyncTypeNetwork:
@@ -640,3 +641,11 @@ func initialResult(result string) string {
 
 var _ cardapp.SeriesAttemptLogger = (*SeriesAttemptLogger)(nil)
 var _ cardapp.SeriesRunner = (*SeriesRunner)(nil)
+
+// flowReadingResult 把运营商流量响应映射为读数结果：没给出可解析读数时为无读数。
+func flowReadingResult(response *gateway.FlowUsageResp) string {
+	if !response.HasReading() {
+		return constants.TrafficCollectionResultNoReading
+	}
+	return constants.TrafficCollectionResultOK
+}

internal/infrastructure/cardobservation/traffic_alert.go

@@ -0,0 +1,120 @@
+package cardobservation
+
+import (
+	"context"
+	"time"
+
+	"go.uber.org/zap"
+	"gorm.io/gorm"
+
+	"github.com/break/junhong_cmp_fiber/internal/domain/cardflow"
+	"github.com/break/junhong_cmp_fiber/pkg/constants"
+)
+
+// 采集告警阈值。
+const (
+	// trafficAlertSlopeRatio 是单卡当天"读数下降"拦截占比告警阈值。
+	trafficAlertSlopeRatio = 0.5
+	// trafficAlertSlopeMinRecords 是单卡当天参与拦截占比统计的最少采集记录数。
+	trafficAlertSlopeMinRecords = 6
+	// trafficAlertSampleLimit 是每类告警日志里列出的卡数上限。
+	trafficAlertSampleLimit = 50
+)
+
+// TrafficAlertScanner 按天扫描采集记录与日快照，输出卡流量采集告警日志（只看开启轮询的卡）。
+type TrafficAlertScanner struct {
+	db     *gorm.DB
+	logger *zap.Logger
+}
+
+// NewTrafficAlertScanner 创建卡流量采集告警扫描器。
+func NewTrafficAlertScanner(db *gorm.DB, logger *zap.Logger) *TrafficAlertScanner {
+	if logger == nil {
+		logger = zap.NewNop()
+	}
+	return &TrafficAlertScanner{db: db, logger: logger}
+}
+
+// Scan 扫描 day（上海日期）一整天：有套餐的卡全天没有写入、待建起点仍未清除、单卡读数下降占比过高、
+// 乱序数量与重置日防旧值命中数。当天没有写入时不补写快照，只告警。
+func (s *TrafficAlertScanner) Scan(ctx context.Context, day time.Time) error {
+	from := cardflow.ShanghaiDate(day)
+	to := from.AddDate(0, 0, 1)
+	db := s.db.WithContext(ctx)
+
+	var notWritten []uint
+	if err := db.Raw(`
+SELECT c.iot_card_id FROM tb_card_traffic_collection c
+JOIN tb_iot_card card ON card.id = c.iot_card_id AND card.enable_polling = TRUE AND card.deleted_at IS NULL
+WHERE c.observed_at >= ? AND c.observed_at < ?
+GROUP BY c.iot_card_id
+HAVING COUNT(*) FILTER (WHERE c.decision IN (?, ?)) = 0
+LIMIT ?`, from, to, constants.TrafficDecisionWritten, constants.TrafficDecisionBaseline, trafficAlertSampleLimit).
+		Scan(&notWritten).Error; err != nil {
+		return err
+	}
+	if len(notWritten) > 0 {
+		s.logger.Error("有套餐的卡一整天没有写入流量快照，请核对运营商读数与采集间隔",
+			zap.String("date", from.Format("2006-01-02")), zap.Uints("card_ids", notWritten),
+			zap.String("metric", "card_traffic.collection.day_not_written"))
+	}
+
+	var baselinePending []uint
+	if err := db.Raw(`
+SELECT id FROM tb_iot_card
+WHERE traffic_baseline_pending = TRUE AND enable_polling = TRUE AND deleted_at IS NULL
+  AND last_data_check_at IS NOT NULL AND last_data_check_at < ? AND last_data_check_at >= ?
+LIMIT ?`, to, from, trafficAlertSampleLimit).Scan(&baselinePending).Error; err != nil {
+		return err
+	}
+	if len(baselinePending) > 0 {
+		s.logger.Error("开启轮询的卡已被采集一天仍未建起点，请核对采集队列是否一直不满足写入条件",
+			zap.String("date", from.Format("2006-01-02")), zap.Uints("card_ids", baselinePending),
+			zap.String("metric", "card_traffic.collection.baseline_pending"))
+	}
+
+	var slopeCards []uint
+	if err := db.Raw(`
+SELECT c.iot_card_id FROM tb_card_traffic_collection c
+JOIN tb_iot_card card ON card.id = c.iot_card_id AND card.enable_polling = TRUE AND card.deleted_at IS NULL
+WHERE c.observed_at >= ? AND c.observed_at < ? AND c.result = ?
+GROUP BY c.iot_card_id
+HAVING COUNT(*) >= ? AND COUNT(*) FILTER (WHERE c.decision = ?)::float / COUNT(*) > ?
+LIMIT ?`, from, to, constants.TrafficCollectionResultOK, trafficAlertSlopeMinRecords,
+		constants.TrafficDecisionSlopeNegative, trafficAlertSlopeRatio, trafficAlertSampleLimit).Scan(&slopeCards).Error; err != nil {
+		return err
+	}
+	if len(slopeCards) > 0 {
+		s.logger.Error("卡当天读数下降拦截占比过高，请核对运营商读数",
+			zap.String("date", from.Format("2006-01-02")), zap.Uints("card_ids", slopeCards),
+			zap.Float64("ratio_threshold", trafficAlertSlopeRatio), zap.String("metric", "card_traffic.collection.slope_negative_high"))
+	}
+
+	var totals struct {
+		Total      int64
+		Slope      int64
+		OutOfOrder int64
+	}
+	if err := db.Raw(`
+SELECT COUNT(*) AS total,
+       COUNT(*) FILTER (WHERE decision = ?) AS slope,
+       COUNT(*) FILTER (WHERE decision = ?) AS out_of_order
+FROM tb_card_traffic_collection WHERE observed_at >= ? AND observed_at < ?`,
+		constants.TrafficDecisionSlopeNegative, constants.TrafficDecisionOutOfOrder, from, to).Scan(&totals).Error; err != nil {
+		return err
+	}
+	var staleHits int64
+	if err := db.Raw(`SELECT COUNT(*) FROM tb_card_flow_day WHERE flow_date = ?::date AND stale_guarded = TRUE`,
+		from.Format("2006-01-02")).Scan(&staleHits).Error; err != nil {
+		return err
+	}
+	level := s.logger.Info
+	if totals.OutOfOrder > 0 {
+		level = s.logger.Warn
+	}
+	level("卡流量采集日统计",
+		zap.String("date", from.Format("2006-01-02")), zap.Int64("collections", totals.Total),
+		zap.Int64("slope_negative", totals.Slope), zap.Int64("out_of_order", totals.OutOfOrder),
+		zap.Int64("stale_guarded_cards", staleHits), zap.String("metric", "card_traffic.collection.daily"))
+	return nil
+}

internal/infrastructure/cardobservation/traffic_queue.go

@@ -0,0 +1,130 @@
+package cardobservation
+
+import (
+	"context"
+	"encoding/json"
+	"strconv"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+
+	cardapp "github.com/break/junhong_cmp_fiber/internal/application/cardobservation"
+	"github.com/break/junhong_cmp_fiber/internal/domain/cardflow"
+	"github.com/break/junhong_cmp_fiber/pkg/constants"
+	"github.com/break/junhong_cmp_fiber/pkg/errors"
+)
+
+// trafficQueueScript 原子更新一张卡的采集队列：跨天清空（新点日期晚于队尾日期时）、丢弃过期点、
+// 按采集时间插入并对同一时间去重、只保留最后 N 个、设置过期，返回队列与新点是否在队尾。
+// 新点日期早于队尾日期（迟到的前一天点）时不改队列，按乱序处理。日期由调用方按上海时区算好传入。
+//
+// KEYS[1] 队列 key；ARGV: 采集时间毫秒、读数、日期(YYYYMMDD)、N、是否跨天清空(1/0)、最长点龄毫秒(0 不限)、过期毫秒。
+var trafficQueueScript = redis.NewScript(`
+local raw = redis.call('GET', KEYS[1])
+local pts = {}
+if raw then pts = cjson.decode(raw) end
+local t = tonumber(ARGV[1])
+local v = tonumber(ARGV[2])
+local d = ARGV[3]
+local n = tonumber(ARGV[4])
+local clear = ARGV[5] == '1'
+local maxAge = tonumber(ARGV[6])
+local ttl = tonumber(ARGV[7])
+if clear and #pts > 0 then
+  local lastDay = pts[#pts].d
+  if d < lastDay then
+    return {cjson.encode(pts), 0}
+  end
+  if d > lastDay then pts = {} end
+end
+if maxAge > 0 then
+  local kept = {}
+  for _, p in ipairs(pts) do
+    if t - p.t <= maxAge then table.insert(kept, p) end
+  end
+  pts = kept
+end
+local out = {}
+local inserted = false
+for _, p in ipairs(pts) do
+  if not inserted and t <= p.t then
+    table.insert(out, {t = t, v = v, d = d})
+    inserted = true
+  end
+  if p.t ~= t then table.insert(out, p) end
+end
+if not inserted then table.insert(out, {t = t, v = v, d = d}) end
+while #out > n do table.remove(out, 1) end
+local tail = 0
+if #out > 0 and out[#out].t == t then tail = 1 end
+redis.call('SET', KEYS[1], cjson.encode(out), 'PX', ttl)
+return {cjson.encode(out), tail}
+`)
+
+// TrafficQueue 是基于 Redis 的卡流量采集队列。队列只是写入前的过滤器：key 丢失时从空开始，只晚写不错账。
+type TrafficQueue struct {
+	redis *redis.Client
+}
+
+// NewTrafficQueue 创建卡流量采集队列。
+func NewTrafficQueue(redisClient *redis.Client) *TrafficQueue {
+	return &TrafficQueue{redis: redisClient}
+}
+
+type queuedPoint struct {
+	T int64   `json:"t"`
+	V float64 `json:"v"`
+	D string  `json:"d"`
+}
+
+// Push 把新点放入卡的采集队列，返回入队后的队列（按采集时间升序）与新点是否在队尾。
+func (q *TrafficQueue) Push(ctx context.Context, cardID uint, point cardflow.QueuePoint, settings cardapp.TrafficQueueSettings) ([]cardflow.QueuePoint, bool, error) {
+	if q == nil || q.redis == nil {
+		return nil, false, errors.New(errors.CodeInternalError, "卡流量采集队列未配置")
+	}
+	day := cardflow.ShanghaiDate(point.ObservedAt)
+	clear := "0"
+	if settings.ClearOnNewDay {
+		clear = "1"
+	}
+	maxAgeMS := int64(settings.MaxPointAgeHours) * int64(time.Hour/time.Millisecond)
+	result, err := trafficQueueScript.Run(ctx, q.redis, []string{constants.RedisCardTrafficQueueKey(cardID)},
+		point.ObservedAt.UnixMilli(), strconv.FormatFloat(point.ReadingMB, 'f', -1, 64), day.Format("20060102"),
+		settings.Size, clear, maxAgeMS, queueTTL(point.ObservedAt, settings).Milliseconds(),
+	).Slice()
+	if err != nil {
+		return nil, false, errors.Wrap(errors.CodeRedisError, err, "更新卡流量采集队列失败")
+	}
+	if len(result) != 2 {
+		return nil, false, errors.New(errors.CodeRedisError, "卡流量采集队列返回格式异常")
+	}
+	encoded, _ := result[0].(string)
+	var stored []queuedPoint
+	if encoded != "" && encoded != "{}" {
+		if err := json.Unmarshal([]byte(encoded), &stored); err != nil {
+			return nil, false, errors.Wrap(errors.CodeRedisError, err, "解析卡流量采集队列失败")
+		}
+	}
+	points := make([]cardflow.QueuePoint, 0, len(stored))
+	for _, p := range stored {
+		points = append(points, cardflow.QueuePoint{ObservedAt: time.UnixMilli(p.T), ReadingMB: p.V})
+	}
+	tail, _ := result[1].(int64)
+	return points, tail == 1, nil
+}
+
+// queueTTL 返回队列 key 的过期时长：跨天清空开启时到次日 01:00（上海时间），关闭时为最长点龄（不限时 7 天）。
+func queueTTL(observedAt time.Time, settings cardapp.TrafficQueueSettings) time.Duration {
+	if settings.ClearOnNewDay {
+		expireAt := cardflow.ShanghaiDate(observedAt).AddDate(0, 0, 1).Add(time.Hour)
+		if ttl := time.Until(expireAt); ttl > time.Minute {
+			return ttl
+		}
+		return time.Hour
+	}
+	hours := settings.MaxPointAgeHours
+	if hours <= 0 {
+		hours = constants.TrafficQueueUnlimitedTTLHours
+	}
+	return time.Duration(hours) * time.Hour
+}