fix: 修复手动刷新资产时流量直接覆盖为运营商原始读数的问题
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 7m6s
All checks were successful
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Successful in 7m6s
RefreshCardDataFromGateway 原先直接将网关返回的累计读数写入 current_month_usage_mb,导致本月已用流量从 312MB 跳变为 29GB。 修复内容: - 流量计算改为增量逻辑(与轮询 calculateFlowUpdates 一致) - 补充更新 last_gateway_reading_mb 基准,防止后续轮询增量异常 - 补充跨月检测和运营商重置日窗口判断 - 通过 DataDeductor 回调触发套餐流量扣减(手动刷新也能更新套餐)
This commit is contained in:
@@ -103,9 +103,12 @@ func initServices(s *stores, deps *Dependencies) *services {
|
|||||||
accountAudit := accountAuditSvc.NewService(s.AccountOperationLog)
|
accountAudit := accountAuditSvc.NewService(s.AccountOperationLog)
|
||||||
account := accountSvc.New(s.Account, s.Role, s.AccountRole, s.ShopRole, s.Shop, s.Enterprise, accountAudit)
|
account := accountSvc.New(s.Account, s.Role, s.AccountRole, s.ShopRole, s.Shop, s.Enterprise, accountAudit)
|
||||||
|
|
||||||
// 创建 IotCard service 并设置 polling callback
|
// 创建 IotCard service 并设置回调
|
||||||
iotCard := iotCardSvc.New(deps.DB, s.IotCard, s.Shop, s.AssetAllocationRecord, s.ShopPackageAllocation, s.ShopSeriesAllocation, s.PackageSeries, deps.GatewayClient, deps.Logger)
|
iotCard := iotCardSvc.New(deps.DB, s.IotCard, s.Shop, s.AssetAllocationRecord, s.ShopPackageAllocation, s.ShopSeriesAllocation, s.PackageSeries, deps.GatewayClient, deps.Logger)
|
||||||
iotCard.SetPollingCallback(polling.NewAPICallback(deps.Redis, deps.Logger))
|
iotCard.SetPollingCallback(polling.NewAPICallback(deps.Redis, deps.Logger))
|
||||||
|
// 注入流量扣减回调,使手动刷新资产时能触发套餐流量扣减
|
||||||
|
usageService := packageSvc.NewUsageService(deps.DB, deps.Redis, s.PackageUsage, s.PackageUsageDailyRecord, deps.Logger)
|
||||||
|
iotCard.SetDataDeductor(usageService)
|
||||||
|
|
||||||
// 创建支付配置服务(Order 和 Recharge 依赖)
|
// 创建支付配置服务(Order 和 Recharge 依赖)
|
||||||
wechatConfig := wechatConfigSvc.New(s.WechatConfig, s.Order, s.AssetRecharge, s.AgentRecharge, accountAudit, deps.Redis, deps.Logger)
|
wechatConfig := wechatConfigSvc.New(s.WechatConfig, s.Order, s.AssetRecharge, s.AgentRecharge, accountAudit, deps.Redis, deps.Logger)
|
||||||
|
|||||||
@@ -31,6 +31,13 @@ type PollingCallback interface {
|
|||||||
OnCardDisabled(ctx context.Context, cardID uint)
|
OnCardDisabled(ctx context.Context, cardID uint)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DataDeductor 流量扣减回调接口
|
||||||
|
// 用于在手动刷新流量后触发套餐扣减,避免循环依赖
|
||||||
|
type DataDeductor interface {
|
||||||
|
// DeductDataUsage 按优先级扣减套餐流量
|
||||||
|
DeductDataUsage(ctx context.Context, carrierType string, carrierID uint, usageMB int64) error
|
||||||
|
}
|
||||||
|
|
||||||
type Service struct {
|
type Service struct {
|
||||||
db *gorm.DB
|
db *gorm.DB
|
||||||
iotCardStore *postgres.IotCardStore
|
iotCardStore *postgres.IotCardStore
|
||||||
@@ -42,6 +49,7 @@ type Service struct {
|
|||||||
gatewayClient *gateway.Client
|
gatewayClient *gateway.Client
|
||||||
logger *zap.Logger
|
logger *zap.Logger
|
||||||
pollingCallback PollingCallback // 轮询回调,可选
|
pollingCallback PollingCallback // 轮询回调,可选
|
||||||
|
dataDeductor DataDeductor // 流量扣减回调,可选
|
||||||
}
|
}
|
||||||
|
|
||||||
func New(
|
func New(
|
||||||
@@ -74,6 +82,12 @@ func (s *Service) SetPollingCallback(callback PollingCallback) {
|
|||||||
s.pollingCallback = callback
|
s.pollingCallback = callback
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SetDataDeductor 设置流量扣减回调
|
||||||
|
// 在应用启动时由 bootstrap 调用,注入套餐扣减服务
|
||||||
|
func (s *Service) SetDataDeductor(deductor DataDeductor) {
|
||||||
|
s.dataDeductor = deductor
|
||||||
|
}
|
||||||
|
|
||||||
func (s *Service) ListStandalone(ctx context.Context, req *dto.ListStandaloneIotCardRequest) (*dto.ListStandaloneIotCardResponse, error) {
|
func (s *Service) ListStandalone(ctx context.Context, req *dto.ListStandaloneIotCardRequest) (*dto.ListStandaloneIotCardResponse, error) {
|
||||||
page := req.Page
|
page := req.Page
|
||||||
pageSize := req.PageSize
|
pageSize := req.PageSize
|
||||||
@@ -796,6 +810,7 @@ func (s *Service) buildCardNotFoundFailedItems(iccids []string) []dto.CardSeries
|
|||||||
|
|
||||||
// RefreshCardDataFromGateway 从 Gateway 完整同步卡数据
|
// RefreshCardDataFromGateway 从 Gateway 完整同步卡数据
|
||||||
// 调用网关查询网络状态、实名状态、本月流量,并写回数据库
|
// 调用网关查询网络状态、实名状态、本月流量,并写回数据库
|
||||||
|
// 流量采用增量计算(与轮询逻辑一致):increment = 当前网关读数 - 上次网关读数
|
||||||
func (s *Service) RefreshCardDataFromGateway(ctx context.Context, iccid string) error {
|
func (s *Service) RefreshCardDataFromGateway(ctx context.Context, iccid string) error {
|
||||||
card, err := s.iotCardStore.GetByICCID(ctx, iccid)
|
card, err := s.iotCardStore.GetByICCID(ctx, iccid)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -809,6 +824,7 @@ func (s *Service) RefreshCardDataFromGateway(ctx context.Context, iccid string)
|
|||||||
updates := map[string]any{
|
updates := map[string]any{
|
||||||
"last_sync_time": syncTime,
|
"last_sync_time": syncTime,
|
||||||
}
|
}
|
||||||
|
var flowIncrementMB float64
|
||||||
|
|
||||||
if s.gatewayClient != nil {
|
if s.gatewayClient != nil {
|
||||||
// 1. 查询网络状态(卡的开/停机状态)
|
// 1. 查询网络状态(卡的开/停机状态)
|
||||||
@@ -833,14 +849,14 @@ func (s *Service) RefreshCardDataFromGateway(ctx context.Context, iccid string)
|
|||||||
updates["real_name_status"] = realNameStatus
|
updates["real_name_status"] = realNameStatus
|
||||||
}
|
}
|
||||||
|
|
||||||
// 3. 查询本月流量用量
|
// 3. 查询本月流量用量 — 使用增量计算(与轮询 calculateFlowUpdates 逻辑一致)
|
||||||
flowResp, err := s.gatewayClient.QueryFlow(ctx, &gateway.FlowQueryReq{
|
flowResp, err := s.gatewayClient.QueryFlow(ctx, &gateway.FlowQueryReq{
|
||||||
CardNo: iccid,
|
CardNo: iccid,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
s.logger.Warn("刷新卡数据:查询流量失败", zap.String("iccid", iccid), zap.Error(err))
|
s.logger.Warn("刷新卡数据:查询流量失败", zap.String("iccid", iccid), zap.Error(err))
|
||||||
} else {
|
} else {
|
||||||
updates["current_month_usage_mb"] = flowResp.Used
|
flowIncrementMB = s.calculateRefreshFlowUpdates(card, flowResp.Used, syncTime, updates)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -850,10 +866,111 @@ func (s *Service) RefreshCardDataFromGateway(ctx context.Context, iccid string)
|
|||||||
return errors.Wrap(errors.CodeInternalError, err, "更新卡数据失败")
|
return errors.Wrap(errors.CodeInternalError, err, "更新卡数据失败")
|
||||||
}
|
}
|
||||||
|
|
||||||
s.logger.Info("刷新卡数据成功", zap.String("iccid", iccid), zap.Uint("card_id", card.ID))
|
// 有增量时触发套餐流量扣减
|
||||||
|
if flowIncrementMB > 0 && s.dataDeductor != nil {
|
||||||
|
if err := s.dataDeductor.DeductDataUsage(ctx, "iot_card", card.ID, int64(flowIncrementMB)); err != nil {
|
||||||
|
s.logger.Warn("手动刷新:套餐流量扣减失败",
|
||||||
|
zap.Uint("card_id", card.ID),
|
||||||
|
zap.Float64("increment_mb", flowIncrementMB),
|
||||||
|
zap.Error(err))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
s.logger.Info("刷新卡数据成功",
|
||||||
|
zap.String("iccid", iccid),
|
||||||
|
zap.Uint("card_id", card.ID),
|
||||||
|
zap.Float64("flow_increment_mb", flowIncrementMB))
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// calculateRefreshFlowUpdates 计算手动刷新时的流量增量并填充 updates
|
||||||
|
// 算法与轮询 calculateFlowUpdates 一致:增量 = 当前网关读数 - 上次网关读数
|
||||||
|
// 返回本次增量值(MB),用于触发套餐扣减
|
||||||
|
func (s *Service) calculateRefreshFlowUpdates(card *model.IotCard, gatewayFlowMB float64, now time.Time, updates map[string]any) float64 {
|
||||||
|
increment := gatewayFlowMB - card.LastGatewayReadingMB
|
||||||
|
|
||||||
|
if increment < 0 {
|
||||||
|
// 当前值比上次小,检查是否为上游运营商正常重置
|
||||||
|
resetDay := s.getCarrierResetDay(card.CarrierID)
|
||||||
|
if isRefreshResetWindow(now, resetDay) {
|
||||||
|
// 在重置日窗口内,本次原始值即为增量
|
||||||
|
increment = gatewayFlowMB
|
||||||
|
s.logger.Info("手动刷新:检测到上游运营商重置",
|
||||||
|
zap.Uint("card_id", card.ID),
|
||||||
|
zap.Float64("gateway_flow", gatewayFlowMB),
|
||||||
|
zap.Float64("last_reading", card.LastGatewayReadingMB),
|
||||||
|
zap.Int("reset_day", resetDay))
|
||||||
|
} else {
|
||||||
|
// 非重置日出现值下降,异常,不计入
|
||||||
|
s.logger.Warn("手动刷新:流量异常,非重置日出现值下降",
|
||||||
|
zap.Uint("card_id", card.ID),
|
||||||
|
zap.Float64("gateway_flow", gatewayFlowMB),
|
||||||
|
zap.Float64("last_reading", card.LastGatewayReadingMB))
|
||||||
|
increment = 0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 始终更新网关读数基准
|
||||||
|
updates["last_gateway_reading_mb"] = gatewayFlowMB
|
||||||
|
|
||||||
|
// 检测跨自然月
|
||||||
|
currentMonthStart := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, now.Location())
|
||||||
|
isCrossMonth := card.CurrentMonthStartDate == nil ||
|
||||||
|
card.CurrentMonthStartDate.Before(currentMonthStart)
|
||||||
|
|
||||||
|
if isCrossMonth {
|
||||||
|
s.logger.Info("手动刷新:检测到跨月,重置流量计数",
|
||||||
|
zap.Uint("card_id", card.ID),
|
||||||
|
zap.Float64("last_month_total", card.CurrentMonthUsageMB))
|
||||||
|
updates["last_month_total_mb"] = card.CurrentMonthUsageMB
|
||||||
|
updates["current_month_start_date"] = currentMonthStart
|
||||||
|
if increment > 0 {
|
||||||
|
updates["current_month_usage_mb"] = increment
|
||||||
|
} else {
|
||||||
|
updates["current_month_usage_mb"] = float64(0)
|
||||||
|
}
|
||||||
|
} else if increment > 0 {
|
||||||
|
// 同月内,增量累加(使用 gorm.Expr 保证原子性)
|
||||||
|
updates["current_month_usage_mb"] = gorm.Expr("current_month_usage_mb + ?", increment)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 更新卡生命周期总用量
|
||||||
|
if increment > 0 {
|
||||||
|
updates["data_usage_mb"] = gorm.Expr("data_usage_mb + ?", int64(increment))
|
||||||
|
}
|
||||||
|
|
||||||
|
// 首次流量查询初始化
|
||||||
|
if card.CurrentMonthStartDate == nil {
|
||||||
|
updates["current_month_start_date"] = currentMonthStart
|
||||||
|
}
|
||||||
|
|
||||||
|
return increment
|
||||||
|
}
|
||||||
|
|
||||||
|
// getCarrierResetDay 读取运营商的上游流量重置日
|
||||||
|
func (s *Service) getCarrierResetDay(carrierID uint) int {
|
||||||
|
var carrier model.Carrier
|
||||||
|
if err := s.db.Select("data_reset_day").First(&carrier, carrierID).Error; err != nil {
|
||||||
|
return 1
|
||||||
|
}
|
||||||
|
if carrier.DataResetDay == 0 {
|
||||||
|
return 1
|
||||||
|
}
|
||||||
|
return carrier.DataResetDay
|
||||||
|
}
|
||||||
|
|
||||||
|
// isRefreshResetWindow 判断今天是否在运营商重置日窗口内
|
||||||
|
// 窗口 = 重置日当天 + 前一天(容错网关数据延迟)
|
||||||
|
func isRefreshResetWindow(now time.Time, resetDay int) bool {
|
||||||
|
today := now.Day()
|
||||||
|
if today == resetDay {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
resetDate := time.Date(now.Year(), now.Month(), resetDay, 0, 0, 0, 0, now.Location())
|
||||||
|
prevDay := resetDate.AddDate(0, 0, -1).Day()
|
||||||
|
return today == prevDay
|
||||||
|
}
|
||||||
|
|
||||||
// parseNetworkStatus 将网关返回的卡状态字符串转换为 network_status 数值
|
// parseNetworkStatus 将网关返回的卡状态字符串转换为 network_status 数值
|
||||||
// 停机→0,其他(准备/正常)→1
|
// 停机→0,其他(准备/正常)→1
|
||||||
func parseNetworkStatus(cardStatus string) int {
|
func parseNetworkStatus(cardStatus string) int {
|
||||||
|
|||||||
Reference in New Issue
Block a user