1214181e78
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 16m2s
- 轮询步骤异步模式下只调网关并发布 card.ingest 入库消息，入库消费者固定并发 30、按采集 ID 去重、可重试错误不限次数退避重投 - 采集记录经 card.collection 攒批写入，失败整批重投，新增 (observation_id, observed_at) 唯一索引 - 网络、实名观测增加乱序保护；轮询读查询走缓存，套餐状态查询各自计时且失败不缓存 - 入库积压告警与反压；系统配置 polling.ingest.mode 默认 sync 可回退；测试环境 nsqd 消息全部落盘 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
260 lines
12 KiB
Go
260 lines
12 KiB
Go
package cardobservation
|
||
|
||
import (
|
||
"context"
|
||
"strconv"
|
||
"time"
|
||
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
"gorm.io/gorm/clause"
|
||
|
||
cardflow "github.com/break/junhong_cmp_fiber/internal/domain/cardflow"
|
||
domain "github.com/break/junhong_cmp_fiber/internal/domain/cardobservation"
|
||
carrierthresholddomain "github.com/break/junhong_cmp_fiber/internal/domain/carrierthreshold"
|
||
"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"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
"github.com/break/junhong_cmp_fiber/pkg/outboxid"
|
||
)
|
||
|
||
// TrafficIncrementedEvent 是卡流量正增量的可靠领域事实。
|
||
type TrafficIncrementedEvent struct {
|
||
EventID string `json:"event_id"`
|
||
CardID uint `json:"card_id"`
|
||
IncrementMB float64 `json:"increment_mb"`
|
||
ObservedAt time.Time `json:"observed_at"`
|
||
// ObservationID 是采集 ID，扣费审计据此追溯到采集记录。
|
||
ObservationID string `json:"observation_id,omitempty"`
|
||
Source string `json:"source"`
|
||
Scene string `json:"scene"`
|
||
RequestID string `json:"request_id,omitempty"`
|
||
CorrelationID string `json:"correlation_id,omitempty"`
|
||
}
|
||
|
||
// TrafficQueuePort 是卡流量采集队列：原子入队并返回入队后的队列与新点是否在队尾。
|
||
type TrafficQueuePort interface {
|
||
Push(ctx context.Context, cardID uint, point cardflow.QueuePoint, settings TrafficQueueSettings) ([]cardflow.QueuePoint, bool, error)
|
||
}
|
||
|
||
// TrafficQueueSettingsSource 返回当前采集队列配置。
|
||
type TrafficQueueSettingsSource interface {
|
||
Current(ctx context.Context) TrafficQueueSettings
|
||
}
|
||
|
||
// ApplyTrafficObservation 应用一次流量采集：读数先进采集队列，满足写入条件才写日快照、按月对账、
|
||
// 在采集时刻有套餐时同事务写流量事件；有套餐的卡每次采集写一条采集记录。
|
||
func (s *Service) ApplyTrafficObservation(ctx context.Context, observation domain.TrafficObservation) (domain.TrafficDecision, error) {
|
||
if s == nil || s.db == nil || s.eventWriter == nil || s.trafficQueue == nil {
|
||
return domain.TrafficDecision{}, errors.New(errors.CodeInternalError, "卡流量观测能力未完整配置")
|
||
}
|
||
if err := domain.ValidateTrafficObservation(observation); err != nil {
|
||
return domain.TrafficDecision{}, err
|
||
}
|
||
record := newCollectionRecord(observation)
|
||
decision := domain.TrafficDecision{Result: record.Result}
|
||
if record.Result != constants.TrafficCollectionResultOK {
|
||
s.writeCollectionRecord(ctx, record, observation.Metadata.ObservedAt)
|
||
return decision, nil
|
||
}
|
||
reading := observation.GatewayReadingMB
|
||
decision.ReadingMB = &reading
|
||
settings := s.queueSettings()
|
||
points, tail, err := s.trafficQueue.Push(ctx, observation.CardID,
|
||
cardflow.QueuePoint{ObservedAt: observation.Metadata.ObservedAt, ReadingMB: reading}, settings)
|
||
if err != nil {
|
||
return decision, err
|
||
}
|
||
record.setQueue(points)
|
||
judge := cardflow.JudgeQueue(points, tail, settings.Size, settings.DropToleranceMB)
|
||
record.Decision = judge
|
||
var outcome *flowOutcome
|
||
err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
var card model.IotCard
|
||
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ?", observation.CardID).First(&card).Error; err != nil {
|
||
if err == gorm.ErrRecordNotFound {
|
||
return errors.New(errors.CodeNotFound, "IoT卡不存在")
|
||
}
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "锁定IoT卡流量事实失败")
|
||
}
|
||
decision.CurrentMonthUsageMB, decision.LastGatewayReadingMB = card.CurrentMonthUsageMB, card.LastGatewayReadingMB
|
||
packages, err := postgres.FindEffectivePackagesAt(ctx, tx, card.ID, observation.Metadata.ObservedAt, false)
|
||
if err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "查询卡采集时刻有效套餐失败")
|
||
}
|
||
record.HasPackage = len(packages.Packages) > 0
|
||
record.CarrierID = card.CarrierID
|
||
if judge != cardflow.QueueWrite {
|
||
if observation.Metadata.Scene == constants.CardObservationScenePreActivationSettlement {
|
||
// 激活前结算没写入：用已有本月最大快照结算，不新写快照。
|
||
if err := s.settleExisting(ctx, tx, &card, observation.Metadata.ObservedAt); err != nil {
|
||
return err
|
||
}
|
||
record.Decision = constants.TrafficDecisionSettledExisting
|
||
}
|
||
return touchTrafficCheck(tx, card.ID, observation.Metadata.ObservedAt)
|
||
}
|
||
next, err := s.applyWrittenPoint(ctx, tx, &card, observation, packages)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
outcome = next
|
||
return nil
|
||
})
|
||
if err != nil {
|
||
s.writeTrafficFailureAudit(ctx, observation, err)
|
||
return domain.TrafficDecision{}, err
|
||
}
|
||
if outcome != nil {
|
||
record.applyOutcome(outcome)
|
||
decision.ReadingAccepted = outcome.Written
|
||
decision.IncrementMB = outcome.EmitMB
|
||
decision.CrossMonth = outcome.CrossMonth
|
||
decision.CurrentMonthUsageMB = outcome.MonthMaxMB
|
||
}
|
||
decision.Decision = record.Decision
|
||
s.writeCollectionRecord(ctx, record, observation.Metadata.ObservedAt)
|
||
if s.cache != nil {
|
||
s.cache.Invalidate(ctx, observation.CardID)
|
||
}
|
||
return decision, nil
|
||
}
|
||
|
||
// applyWrittenPoint 在观测事务内处理一个通过队列判断的写入点。
|
||
func (s *Service) applyWrittenPoint(ctx context.Context, tx *gorm.DB, card *model.IotCard, observation domain.TrafficObservation, packages postgres.EffectivePackages) (*flowOutcome, error) {
|
||
hasPackage := len(packages.Packages) > 0
|
||
reading := observation.GatewayReadingMB
|
||
observedAt := observation.Metadata.ObservedAt
|
||
outcome := &flowOutcome{MonthMaxMB: card.CurrentMonthUsageMB, Decision: constants.TrafficDecisionWritten}
|
||
if reading == 0 && !hasPackage {
|
||
// 没套餐的卡读数为 0 不写快照。
|
||
return outcome, touchTrafficCheck(tx, card.ID, observedAt)
|
||
}
|
||
flow, err := s.writeFlow(ctx, tx, card, reading, observedAt, hasPackage)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
outcome.Written, outcome.MonthMaxMB, outcome.StaleGuarded = true, flow.monthMax, flow.conversion.StaleGuarded
|
||
outcome.MonthUsedMB, outcome.DeltaMB = flow.conversion.MonthUsedMB, flow.delta
|
||
if card.TrafficBaselinePending {
|
||
outcome.Decision = constants.TrafficDecisionBaseline
|
||
} else if hasPackage && flow.delta > 0 {
|
||
outcome.EmitMB = flow.delta
|
||
}
|
||
if hasPackage {
|
||
if err := ensureDailyUsageRows(ctx, tx, packages.Packages, flow.flowDate); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
monthStart := cardflow.MonthStart(flow.flowDate)
|
||
outcome.CrossMonth = card.CurrentMonthStartDate == nil || card.CurrentMonthStartDate.Before(monthStart)
|
||
updates := map[string]any{
|
||
"last_data_check_at": observedAt, "last_sync_time": observedAt,
|
||
"current_month_usage_mb": flow.monthMax, "current_month_start_date": monthStart,
|
||
"data_usage_mb": card.DataUsageMB + int64(flow.delta), "traffic_baseline_pending": false,
|
||
}
|
||
if outcome.CrossMonth {
|
||
updates["last_month_total_mb"] = card.CurrentMonthUsageMB
|
||
}
|
||
if err := tx.Model(&model.IotCard{}).Where("id = ?", card.ID).Updates(updates).Error; err != nil {
|
||
return nil, errors.Wrap(errors.CodeDatabaseError, err, "更新卡流量事实失败")
|
||
}
|
||
if outcome.EmitMB > 0 {
|
||
outcome.EventID = outboxid.Stable("card-traffic:", strconv.FormatUint(uint64(card.ID), 10)+":"+observation.Metadata.ObservationID+":incremented")
|
||
if err := s.eventWriter.AppendTraffic(ctx, tx, TrafficIncrementedEvent{
|
||
EventID: outcome.EventID, CardID: card.ID, IncrementMB: outcome.EmitMB,
|
||
ObservedAt: observedAt, ObservationID: observation.Metadata.ObservationID,
|
||
Source: observation.Metadata.Source, Scene: observation.Metadata.Scene,
|
||
RequestID: observation.Metadata.RequestID, CorrelationID: observation.Metadata.CorrelationID,
|
||
}); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
// 通道阈值只在写入点判断：读数大于 0 且未被防旧值按 0 处理；激活前结算不触发停复机评估。
|
||
if reading > 0 && !outcome.StaleGuarded && s.channelThreshold != nil &&
|
||
observation.Metadata.Scene != constants.CardObservationScenePreActivationSettlement {
|
||
if err := s.channelThreshold.EvaluateInTx(ctx, tx, carrierthresholddomain.Evaluation{
|
||
CardID: card.ID, CarrierID: card.CarrierID, ReadingMB: reading, ObservedAt: observedAt,
|
||
// 观测来源是判定端获取触发来源的唯一位置：命中留痕据此区分自动判定、人工同步与回调。
|
||
Source: observation.Metadata.Source,
|
||
}); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
if err := s.writeTrafficStateAudit(ctx, tx, card, observation, outcome); err != nil {
|
||
return nil, err
|
||
}
|
||
return outcome, nil
|
||
}
|
||
|
||
// touchTrafficCheck 只刷新卡的最近流量检查时间（本次没有写入快照）。
|
||
func touchTrafficCheck(tx *gorm.DB, cardID uint, observedAt time.Time) error {
|
||
if err := tx.Model(&model.IotCard{}).Where("id = ?", cardID).
|
||
Updates(map[string]any{"last_data_check_at": observedAt, "last_sync_time": observedAt}).Error; err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "更新卡流量检查时间失败")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// writeTrafficStateAudit 在本次写入改变了卡流量事实时写统一审计（人工刷新或 Worker 观测）。
|
||
func (s *Service) writeTrafficStateAudit(ctx context.Context, tx *gorm.DB, card *model.IotCard, observation domain.TrafficObservation, outcome *flowOutcome) error {
|
||
if outcome.DeltaMB <= 0 && !outcome.CrossMonth {
|
||
return nil
|
||
}
|
||
before := map[string]any{
|
||
"data_usage_mb": card.DataUsageMB, "current_month_usage_mb": card.CurrentMonthUsageMB,
|
||
"current_month_start_date": card.CurrentMonthStartDate, "last_month_total_mb": card.LastMonthTotalMB,
|
||
}
|
||
after := map[string]any{
|
||
"data_usage_mb": card.DataUsageMB + int64(outcome.DeltaMB), "current_month_usage_mb": outcome.MonthMaxMB,
|
||
"increment_mb": outcome.EmitMB, "delta_mb": outcome.DeltaMB, "decision": outcome.Decision,
|
||
"observation_id": observation.Metadata.ObservationID,
|
||
}
|
||
input := StateAudit{Card: card, BeforeData: before, AfterData: after}
|
||
if actionCode, audited := manualRefreshAuditAction(ctx); observation.Metadata.Source == constants.CardObservationSourceManualSync && audited {
|
||
input.ActionCode, input.Summary = actionCode, "人工刷新 IoT 卡流量"
|
||
} else if workerObservationAudited(ctx) {
|
||
input.ActionCode, input.Summary = constants.AuditActionIotCardWorkerTrafficSynced, "Worker 同步 IoT 卡流量事实"
|
||
input.IntegrationID = observation.Metadata.ObservationID
|
||
} else {
|
||
return nil
|
||
}
|
||
if s.auditWriter == nil {
|
||
return errors.New(errors.CodeInternalError, "卡状态统一审计能力未配置")
|
||
}
|
||
return s.auditWriter.WriteCardStateAudit(ctx, tx, input)
|
||
}
|
||
|
||
// writeTrafficFailureAudit 记录 Worker 观测事务失败的审计。
|
||
func (s *Service) writeTrafficFailureAudit(ctx context.Context, observation domain.TrafficObservation, err error) {
|
||
if !workerObservationAudited(ctx) || s.auditWriter == nil {
|
||
return
|
||
}
|
||
// 观测失败常因上下文已超时，失败审计改用独立短超时上下文，避免连带报错。
|
||
auditCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
||
defer cancel()
|
||
ctx = auditCtx
|
||
var card model.IotCard
|
||
if loadErr := s.db.WithContext(ctx).Where("id = ?", observation.CardID).First(&card).Error; loadErr != nil {
|
||
return
|
||
}
|
||
s.auditWriter.WriteCardStateFailure(ctx, StateAudit{
|
||
ActionCode: constants.AuditActionIotCardWorkerTrafficSynced,
|
||
Summary: "Worker 同步 IoT 卡流量事实失败", Card: &card,
|
||
IntegrationID: observation.Metadata.ObservationID,
|
||
}, err)
|
||
}
|
||
|
||
func (s *Service) queueSettings() TrafficQueueSettings {
|
||
if s.trafficQueueSettings == nil {
|
||
return DefaultTrafficQueueSettings()
|
||
}
|
||
return s.trafficQueueSettings.Current(context.Background())
|
||
}
|
||
|
||
// logTrafficWarn 记录不影响观测结果的告警。
|
||
func (s *Service) logTrafficWarn(message string, fields ...zap.Field) {
|
||
s.warnLogger().Warn(message, fields...)
|
||
}
|