From 58eb0474334e89f6415372a388e55a4662762497 Mon Sep 17 00:00:00 2001 From: huang Date: Tue, 31 Mar 2026 19:50:17 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E6=89=8B=E5=8A=A8?= =?UTF-8?q?=E5=88=B7=E6=96=B0=E8=B5=84=E4=BA=A7=E6=97=B6=E6=B5=81=E9=87=8F?= =?UTF-8?q?=E7=9B=B4=E6=8E=A5=E8=A6=86=E7=9B=96=E4=B8=BA=E8=BF=90=E8=90=A5?= =?UTF-8?q?=E5=95=86=E5=8E=9F=E5=A7=8B=E8=AF=BB=E6=95=B0=E7=9A=84=E9=97=AE?= =?UTF-8?q?=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RefreshCardDataFromGateway 原先直接将网关返回的累计读数写入 current_month_usage_mb，导致本月已用流量从 312MB 跳变为 29GB。 修复内容： - 流量计算改为增量逻辑（与轮询 calculateFlowUpdates 一致） - 补充更新 last_gateway_reading_mb 基准，防止后续轮询增量异常 - 补充跨月检测和运营商重置日窗口判断 - 通过 DataDeductor 回调触发套餐流量扣减（手动刷新也能更新套餐） --- internal/bootstrap/services.go | 5 +- internal/service/iot_card/service.go | 123 ++++++++++++++++++++++++++- 2 files changed, 124 insertions(+), 4 deletions(-) diff --git a/internal/bootstrap/services.go b/internal/bootstrap/services.go index 5ee0dd6..ef65037 100644 --- a/internal/bootstrap/services.go +++ b/internal/bootstrap/services.go @@ -103,9 +103,12 @@ func initServices(s *stores, deps *Dependencies) *services { accountAudit := accountAuditSvc.NewService(s.AccountOperationLog) 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.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 依赖） wechatConfig := wechatConfigSvc.New(s.WechatConfig, s.Order, s.AssetRecharge, s.AgentRecharge, accountAudit, deps.Redis, deps.Logger) diff --git a/internal/service/iot_card/service.go b/internal/service/iot_card/service.go index 09c8c29..ebb66fd 100644 --- a/internal/service/iot_card/service.go +++ b/internal/service/iot_card/service.go @@ -31,6 +31,13 @@ type PollingCallback interface { 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 { db *gorm.DB iotCardStore *postgres.IotCardStore @@ -42,6 +49,7 @@ type Service struct { gatewayClient *gateway.Client logger *zap.Logger pollingCallback PollingCallback // 轮询回调，可选 + dataDeductor DataDeductor // 流量扣减回调，可选 } func New( @@ -74,6 +82,12 @@ func (s *Service) SetPollingCallback(callback PollingCallback) { 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) { page := req.Page pageSize := req.PageSize @@ -796,6 +810,7 @@ func (s *Service) buildCardNotFoundFailedItems(iccids []string) []dto.CardSeries // RefreshCardDataFromGateway 从 Gateway 完整同步卡数据 // 调用网关查询网络状态、实名状态、本月流量，并写回数据库 +// 流量采用增量计算（与轮询逻辑一致）：increment = 当前网关读数 - 上次网关读数 func (s *Service) RefreshCardDataFromGateway(ctx context.Context, iccid string) error { card, err := s.iotCardStore.GetByICCID(ctx, iccid) if err != nil { @@ -809,6 +824,7 @@ func (s *Service) RefreshCardDataFromGateway(ctx context.Context, iccid string) updates := map[string]any{ "last_sync_time": syncTime, } + var flowIncrementMB float64 if s.gatewayClient != nil { // 1. 查询网络状态（卡的开/停机状态） @@ -833,14 +849,14 @@ func (s *Service) RefreshCardDataFromGateway(ctx context.Context, iccid string) updates["real_name_status"] = realNameStatus } - // 3. 查询本月流量用量 + // 3. 查询本月流量用量 — 使用增量计算（与轮询 calculateFlowUpdates 逻辑一致） flowResp, err := s.gatewayClient.QueryFlow(ctx, &gateway.FlowQueryReq{ CardNo: iccid, }) if err != nil { s.logger.Warn("刷新卡数据：查询流量失败", zap.String("iccid", iccid), zap.Error(err)) } 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, "更新卡数据失败") } - 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 } +// 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 数值 // 停机→0，其他（准备/正常）→1 func parseNetworkStatus(cardStatus string) int {