9c3c258f16
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 9m27s
- 每卡 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>
119 lines
5.2 KiB
Go
119 lines
5.2 KiB
Go
package cardobservation
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
|
||
cardflow "github.com/break/junhong_cmp_fiber/internal/domain/cardflow"
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
)
|
||
|
||
// flowOutcome 是一个写入点的日快照计量结果。
|
||
type flowOutcome struct {
|
||
// Written 为 true 表示本次写入了日快照（没套餐且读数为 0 时为 false）。
|
||
Written bool
|
||
// Decision 是写入判断：已写入或建起点。
|
||
Decision string
|
||
// EmitMB 是本次写入流量事件的增量，0 表示不写事件。
|
||
EmitMB float64
|
||
// DeltaMB 是本次按月对账计入已结算的增量（含没套餐、建起点时只记已结算的部分）。
|
||
DeltaMB float64
|
||
// MonthUsedMB 是本次读数换算后的本月累计；MonthMaxMB 是写入后的本月最大快照。
|
||
MonthUsedMB float64
|
||
MonthMaxMB float64
|
||
// StaleGuarded 为 true 表示本次读数在重置日触发防旧值、按 0 计。
|
||
StaleGuarded bool
|
||
// CrossMonth 为 true 表示卡表本月用量跨入新的自然月。
|
||
CrossMonth bool
|
||
// EventID 是本次写入的流量事件 ID。
|
||
EventID string
|
||
}
|
||
|
||
// flowWrite 是一次快照写入与按月对账的结果。
|
||
type flowWrite struct {
|
||
flowDate time.Time
|
||
conversion cardflow.Conversion
|
||
monthMax float64
|
||
delta float64
|
||
}
|
||
|
||
// writeFlow 按卡所属通道的重置日换算并原子写入当天快照，再按月对账把增量计入当天已结算。
|
||
func (s *Service) writeFlow(ctx context.Context, tx *gorm.DB, card *model.IotCard, readingMB float64, observedAt time.Time, hasPackage bool) (flowWrite, error) {
|
||
resetDay := s.strictResetDay(ctx, tx, card)
|
||
flowDate := cardflow.ShanghaiDate(observedAt)
|
||
days := postgres.NewCardFlowDayStore(tx)
|
||
monthCtx, err := days.MonthContext(ctx, card.ID, flowDate, resetDay)
|
||
if err != nil {
|
||
return flowWrite{}, errors.Wrap(errors.CodeDatabaseError, err, "查询卡日流量快照失败")
|
||
}
|
||
conversion := cardflow.Convert(readingMB, flowDate.Day(), resetDay, monthCtx)
|
||
if err := days.Upsert(ctx, postgres.CardFlowSnapshotWrite{
|
||
CardID: card.ID, FlowDate: flowDate, MonthUsedMB: conversion.MonthUsedMB,
|
||
VendorUsedMB: readingMB, StaleGuarded: conversion.StaleGuarded,
|
||
}); err != nil {
|
||
return flowWrite{}, errors.Wrap(errors.CodeDatabaseError, err, "写入卡日流量快照失败")
|
||
}
|
||
monthMax, settled, err := days.MonthTotals(ctx, card.ID, flowDate)
|
||
if err != nil {
|
||
return flowWrite{}, errors.Wrap(errors.CodeDatabaseError, err, "查询卡本月流量结算失败")
|
||
}
|
||
delta := cardflow.MonthlyDelta(monthMax, settled)
|
||
if err := days.AddSettled(ctx, card.ID, flowDate, delta); err != nil {
|
||
return flowWrite{}, errors.Wrap(errors.CodeDatabaseError, err, "结算卡流量增量失败")
|
||
}
|
||
return flowWrite{flowDate: flowDate, conversion: conversion, monthMax: monthMax, delta: delta}, nil
|
||
}
|
||
|
||
// settleExisting 用卡上已有的本月最大快照结算（激活前结算本次没写入时）：差额全部记为已结算，不新写快照。
|
||
func (s *Service) settleExisting(ctx context.Context, tx *gorm.DB, card *model.IotCard, observedAt time.Time) error {
|
||
flowDate := cardflow.ShanghaiDate(observedAt)
|
||
days := postgres.NewCardFlowDayStore(tx)
|
||
monthMax, settled, err := days.MonthTotals(ctx, card.ID, flowDate)
|
||
if err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "查询卡本月流量结算失败")
|
||
}
|
||
if err := days.AddSettledToLatest(ctx, card.ID, flowDate, cardflow.MonthlyDelta(monthMax, settled)); err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "激活前结算已有快照失败")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ensureDailyUsageRows 为采集时刻每个有效套餐确保当天一行套餐日用量（没扣也写 0）。
|
||
func ensureDailyUsageRows(ctx context.Context, tx *gorm.DB, packages []*model.PackageUsage, flowDate time.Time) error {
|
||
ids := make([]uint, 0, len(packages))
|
||
cumulative := make(map[uint]int64, len(packages))
|
||
for _, pkg := range packages {
|
||
ids = append(ids, pkg.ID)
|
||
cumulative[pkg.ID] = pkg.DataUsageMB
|
||
}
|
||
if err := postgres.EnsureDailyUsageRows(ctx, tx, ids, flowDate, cumulative); err != nil {
|
||
return errors.Wrap(errors.CodeDatabaseError, err, "写入套餐当天日用量失败")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// strictResetDay 在当前连接内按卡的 carrier_id 读取上游流量重置日，不使用调用方预先计算的值；
|
||
// 查询失败、为空或非法时按卡的运营商类型兜底（联通 27，其余 1）并记录日志。
|
||
func (s *Service) strictResetDay(ctx context.Context, db *gorm.DB, card *model.IotCard) int {
|
||
var days []int
|
||
queryErr := db.WithContext(ctx).Model(&model.Carrier{}).
|
||
Where("id = ?", card.CarrierID).Limit(1).Pluck("data_reset_day", &days).Error
|
||
queried := 0
|
||
if len(days) > 0 {
|
||
queried = days[0]
|
||
}
|
||
resetDay, fallback := cardflow.ResolveResetDay(queried, queryErr == nil && len(days) > 0, card.CarrierType)
|
||
if fallback {
|
||
s.warnLogger().Warn("通道上游流量重置日不可用，按运营商类型兜底",
|
||
zap.Uint("card_id", card.ID), zap.Uint("carrier_id", card.CarrierID),
|
||
zap.String("carrier_type", card.CarrierType), zap.Int("queried", queried),
|
||
zap.Int("reset_day", resetDay), zap.Error(queryErr))
|
||
}
|
||
return resetDay
|
||
}
|