diff --git a/internal/model/dto/asset_dto.go b/internal/model/dto/asset_dto.go index de18829..9bd47b0 100644 --- a/internal/model/dto/asset_dto.go +++ b/internal/model/dto/asset_dto.go @@ -79,16 +79,17 @@ type BoundCardInfo struct { // AssetRealtimeStatusResponse 资产实时状态（只读DB/Redis） type AssetRealtimeStatusResponse struct { - AssetType string `json:"asset_type" description:"资产类型：card 或 device"` - AssetID uint `json:"asset_id" description:"资产ID"` - NetworkStatus int `json:"network_status,omitempty" description:"网络状态（asset_type=card时有效）：0停机 1开机"` - RealNameStatus int `json:"real_name_status,omitempty" description:"实名状态（asset_type=card时有效）：0未实名 1已实名"` - RealnamePolicy string `json:"realname_policy,omitempty" description:"实名认证策略 (none:无需实名, before_order:先实名后充值/购买, after_order:先充值/购买后实名)"` - CurrentMonthUsageMB float64 `json:"current_month_usage_mb,omitempty" description:"系统累计的自然月流量MB（asset_type=card时有效）"` - LastSyncTime *time.Time `json:"last_sync_time,omitempty" description:"最后同步时间（asset_type=card时有效）"` - DeviceProtectStatus string `json:"device_protect_status,omitempty" description:"保护期状态（asset_type=device时有效）：none/stop/start"` - Cards []BoundCardInfo `json:"cards,omitempty" description:"绑定卡状态列表（asset_type=device时有效）"` - DeviceRealtime *DeviceGatewayInfo `json:"device_realtime" description:"设备实时状态（来自 Gateway sync-info，始终返回，失败时仅含 gateway_msg）"` + AssetType string `json:"asset_type" description:"资产类型：card 或 device"` + AssetID uint `json:"asset_id" description:"资产ID"` + NetworkStatus int `json:"network_status,omitempty" description:"网络状态（asset_type=card时有效）：0停机 1开机"` + RealNameStatus int `json:"real_name_status,omitempty" description:"实名状态（asset_type=card时有效）：0未实名 1已实名"` + RealnamePolicy string `json:"realname_policy,omitempty" description:"实名认证策略 (none:无需实名, before_order:先实名后充值/购买, after_order:先充值/购买后实名)"` + CurrentMonthUsageMB float64 `json:"current_month_usage_mb,omitempty" description:"系统自然月累计流量(MB，asset_type=card时有效，不等于运营商周期口径)"` + LastGatewayReadingMB float64 `json:"last_gateway_reading_mb,omitempty" description:"运营商当前周期累计流量读数(MB，asset_type=card时有效，运营商口径展示建议读取该字段)"` + LastSyncTime *time.Time `json:"last_sync_time,omitempty" description:"最后同步时间（asset_type=card时有效）"` + DeviceProtectStatus string `json:"device_protect_status,omitempty" description:"保护期状态（asset_type=device时有效）：none/stop/start"` + Cards []BoundCardInfo `json:"cards,omitempty" description:"绑定卡状态列表（asset_type=device时有效）"` + DeviceRealtime *DeviceGatewayInfo `json:"device_realtime" description:"设备实时状态（来自 Gateway sync-info，始终返回，失败时仅含 gateway_msg）"` } // AssetPackageResponse 资产套餐信息 diff --git a/internal/model/dto/iot_card_dto.go b/internal/model/dto/iot_card_dto.go index 9367a7e..92dd8a9 100644 --- a/internal/model/dto/iot_card_dto.go +++ b/internal/model/dto/iot_card_dto.go @@ -45,7 +45,8 @@ type StandaloneIotCardResponse struct { NetworkStatus int `json:"network_status" description:"网络状态 (0:停机, 1:开机)"` NetworkStatusName string `json:"network_status_name" description:"网络状态名称（中文）"` DataUsageMB int64 `json:"data_usage_mb" description:"累计流量使用(MB)"` - CurrentMonthUsageMB float64 `json:"current_month_usage_mb" description:"本月已用流量(MB)"` + CurrentMonthUsageMB float64 `json:"current_month_usage_mb" description:"系统自然月累计流量(MB，不等于运营商周期口径)"` + LastGatewayReadingMB float64 `json:"last_gateway_reading_mb" description:"运营商当前周期累计流量读数(MB，展示运营商口径时使用)"` CurrentMonthStartDate *time.Time `json:"current_month_start_date" description:"本月开始日期"` LastMonthTotalMB float64 `json:"last_month_total_mb" description:"上月流量总量(MB)"` LastDataCheckAt *time.Time `json:"last_data_check_at" description:"最后流量检查时间"` diff --git a/internal/model/iot_card.go b/internal/model/iot_card.go index 38f1606..b9565b6 100644 --- a/internal/model/iot_card.go +++ b/internal/model/iot_card.go @@ -30,7 +30,7 @@ type IotCard struct { RealNameStatus int `gorm:"column:real_name_status;type:int;default:0;not null;comment:实名状态 0-未实名 1-已实名" json:"real_name_status"` NetworkStatus int `gorm:"column:network_status;type:int;default:0;not null;comment:网络状态 0-停机 1-开机" json:"network_status"` DataUsageMB int64 `gorm:"column:data_usage_mb;type:bigint;default:0;comment:累计流量使用(MB)" json:"data_usage_mb"` - CurrentMonthUsageMB float64 `gorm:"column:current_month_usage_mb;type:decimal(10,2);default:0;comment:系统累计的自然月流量(MB)" json:"current_month_usage_mb"` + CurrentMonthUsageMB float64 `gorm:"column:current_month_usage_mb;type:decimal(10,2);default:0;comment:系统自然月累计流量(MB，不等于运营商周期口径)" json:"current_month_usage_mb"` CurrentMonthStartDate *time.Time `gorm:"column:current_month_start_date;type:date;comment:本月开始日期 - 用于检测跨月流量重置" json:"current_month_start_date"` LastMonthTotalMB float64 `gorm:"column:last_month_total_mb;type:decimal(10,2);default:0;comment:上月结束时的总流量(MB) - 用于跨月流量计算" json:"last_month_total_mb"` EnablePolling bool `gorm:"column:enable_polling;type:boolean;default:true;comment:是否参与轮询 true-参与 false-不参与" json:"enable_polling"` @@ -53,7 +53,7 @@ type IotCard struct { IsStandalone bool `gorm:"column:is_standalone;type:boolean;default:true;not null;comment:是否为独立卡(未绑定设备) 由触发器自动维护" json:"is_standalone"` VirtualNo string `gorm:"column:virtual_no;type:varchar(50);uniqueIndex:idx_iot_card_virtual_no,where:deleted_at IS NULL AND virtual_no IS NOT NULL AND virtual_no <> '';comment:虚拟号(可选，非空时全局唯一)" json:"virtual_no,omitempty"` DeviceVirtualNo string `gorm:"column:device_virtual_no;type:varchar(100);not null;default:'';comment:绑定设备的虚拟号快照(未绑定时为空字符串)" json:"device_virtual_no"` - LastGatewayReadingMB float64 `gorm:"column:last_gateway_reading_mb;type:float;not null;default:0;comment:上次轮询时网关返回的流量读数(MB)" json:"last_gateway_reading_mb"` + LastGatewayReadingMB float64 `gorm:"column:last_gateway_reading_mb;type:float;not null;default:0;comment:运营商当前周期累计流量读数(MB，来自网关)" json:"last_gateway_reading_mb"` RealnamePolicy string `gorm:"column:realname_policy;type:varchar(20);default:'after_order';not null;comment:实名认证策略(none=无需实名,before_order=先实名后充值/购买,after_order=先充值/购买后实名)" json:"realname_policy"` // ICCID 双列存储，用于支持 19/20 位 ICCID 精确路由查询 diff --git a/internal/service/asset/service.go b/internal/service/asset/service.go index a25b654..295edd8 100644 --- a/internal/service/asset/service.go +++ b/internal/service/asset/service.go @@ -347,6 +347,7 @@ func (s *Service) GetRealtimeStatus(ctx context.Context, assetType string, id ui resp.NetworkStatus = card.NetworkStatus resp.RealNameStatus = card.RealNameStatus resp.CurrentMonthUsageMB = card.CurrentMonthUsageMB + resp.LastGatewayReadingMB = card.LastGatewayReadingMB resp.LastSyncTime = card.LastSyncTime case "device": diff --git a/internal/service/iot_card/service.go b/internal/service/iot_card/service.go index 607e0cf..36f01e4 100644 --- a/internal/service/iot_card/service.go +++ b/internal/service/iot_card/service.go @@ -35,7 +35,7 @@ type PollingCallback interface { // 用于在手动刷新流量后触发套餐扣减，避免循环依赖 type DataDeductor interface { // DeductDataUsage 按优先级扣减套餐流量 - DeductDataUsage(ctx context.Context, carrierType string, carrierID uint, usageMB int64) error + DeductDataUsage(ctx context.Context, carrierType string, carrierID uint, usageMB float64) error } type Service struct { @@ -279,6 +279,7 @@ func (s *Service) toStandaloneResponse(card *model.IotCard, shopMap map[uint]str NetworkStatusName: constants.GetNetworkStatusName(card.NetworkStatus), DataUsageMB: card.DataUsageMB, CurrentMonthUsageMB: card.CurrentMonthUsageMB, + LastGatewayReadingMB: card.LastGatewayReadingMB, CurrentMonthStartDate: card.CurrentMonthStartDate, LastMonthTotalMB: card.LastMonthTotalMB, LastDataCheckAt: card.LastDataCheckAt, @@ -884,7 +885,7 @@ func (s *Service) RefreshCardDataFromGateway(ctx context.Context, iccid string) // 有增量时触发套餐流量扣减 if flowIncrementMB > 0 && s.dataDeductor != nil { - if err := s.dataDeductor.DeductDataUsage(ctx, "iot_card", card.ID, int64(flowIncrementMB)); err != nil { + if err := s.dataDeductor.DeductDataUsage(ctx, constants.AssetTypeIotCard, card.ID, flowIncrementMB); err != nil { s.logger.Warn("手动刷新：套餐流量扣减失败", zap.Uint("card_id", card.ID), zap.Float64("increment_mb", flowIncrementMB), diff --git a/internal/service/package/usage_service.go b/internal/service/package/usage_service.go index a256faf..0090172 100644 --- a/internal/service/package/usage_service.go +++ b/internal/service/package/usage_service.go @@ -2,6 +2,8 @@ package packagepkg import ( "context" + "math" + "strconv" "time" "github.com/break/junhong_cmp_fiber/internal/model" @@ -13,6 +15,13 @@ import ( "gorm.io/gorm" ) +const ( + // usageRemainderTTL 流量小数余量缓存时长 + usageRemainderTTL = 7 * 24 * time.Hour + // usageRemainderEpsilon 浮点运算容差 + usageRemainderEpsilon = 1e-9 +) + // StopResumeCallback 任务 24.6: 停复机回调接口 // 用于在流量用完时触发停机操作 type StopResumeCallback interface { @@ -57,28 +66,20 @@ func (s *UsageService) SetStopResumeCallback(callback StopResumeCallback) { // DeductDataUsage 任务 10.2-10.6: 按优先级扣减流量 // 扣减顺序：加油包（按 priority ASC） → 主套餐 // 流量用完时自动标记 status=2，所有套餐用完时触发停机 -func (s *UsageService) DeductDataUsage(ctx context.Context, carrierType string, carrierID uint, usageMB int64) error { +func (s *UsageService) DeductDataUsage(ctx context.Context, carrierType string, carrierID uint, usageMB float64) error { if usageMB <= 0 { return errors.New(errors.CodeInvalidParam, "扣减流量必须大于0") } + deductUsageMB := s.convertToDeductMB(ctx, carrierType, carrierID, usageMB) + if deductUsageMB <= 0 { + return nil + } + return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { - // 查询所有生效中的套餐（按优先级排序） - var packages []*model.PackageUsage - query := tx.Where("status = ?", constants.PackageUsageStatusActive) - - if carrierType == "iot_card" { - query = query.Where("iot_card_id = ?", carrierID) - } else if carrierType == "device" { - query = query.Where("device_id = ?", carrierID) - } else { - return errors.New(errors.CodeInvalidParam, "无效的载体类型") - } - - // 加油包按 priority ASC 排序，主套餐在后 - if err := query.Order("CASE WHEN master_usage_id IS NOT NULL THEN 0 ELSE 1 END, priority ASC"). - Find(&packages).Error; err != nil { - return errors.Wrap(errors.CodeDatabaseError, err, "查询生效套餐失败") + targetCarrierType, targetCarrierID, packages, err := s.resolveActivePackages(ctx, tx, carrierType, carrierID) + if err != nil { + return err } if len(packages) == 0 { @@ -86,7 +87,7 @@ func (s *UsageService) DeductDataUsage(ctx context.Context, carrierType string, } // 按优先级扣减流量 - remainingUsage := usageMB + remainingUsage := deductUsageMB today := time.Now().Format("2006-01-02") for _, pkg := range packages { @@ -144,15 +145,15 @@ func (s *UsageService) DeductDataUsage(ctx context.Context, carrierType string, // 如果流量扣减未完成，说明所有套餐都不够 if remainingUsage > 0 { s.logger.Warn("流量不足", - zap.String("carrier_type", carrierType), - zap.Uint("carrier_id", carrierID), - zap.Int64("requested_mb", usageMB), + zap.String("carrier_type", targetCarrierType), + zap.Uint("carrier_id", targetCarrierID), + zap.Int64("requested_mb", deductUsageMB), zap.Int64("remaining_mb", remainingUsage)) return errors.New(errors.CodeInsufficientQuota, "流量不足") } // 任务 10.5: 检查是否所有套餐都用完（触发停机） - if err := s.checkAndTriggerSuspension(ctx, tx, carrierType, carrierID); err != nil { + if err := s.checkAndTriggerSuspension(ctx, tx, targetCarrierType, targetCarrierID); err != nil { return err } @@ -160,6 +161,132 @@ func (s *UsageService) DeductDataUsage(ctx context.Context, carrierType string, }) } +// convertToDeductMB 将浮点增量转换为可扣减的整数 MB +// 规则：先合并 Redis 中的小数余量，再按整数 MB 扣减；余数回写 Redis +func (s *UsageService) convertToDeductMB(ctx context.Context, carrierType string, carrierID uint, usageMB float64) int64 { + deductUsageMB := int64(math.Floor(usageMB + usageRemainderEpsilon)) + remainder := usageMB - float64(deductUsageMB) + + if s.redis == nil { + return deductUsageMB + } + + remainderKey := constants.RedisPackageUsageRemainderKey(carrierType, carrierID) + cachedRemainder, err := s.getUsageRemainder(ctx, remainderKey) + if err != nil { + s.logger.Warn("读取流量小数余量失败，降级为本次直接取整扣减", + zap.String("carrier_type", carrierType), + zap.Uint("carrier_id", carrierID), + zap.Float64("usage_mb", usageMB), + zap.Error(err)) + return deductUsageMB + } + + totalUsage := usageMB + cachedRemainder + deductUsageMB = int64(math.Floor(totalUsage + usageRemainderEpsilon)) + remainder = totalUsage - float64(deductUsageMB) + if remainder < usageRemainderEpsilon { + remainder = 0 + } + + if err := s.persistUsageRemainder(ctx, remainderKey, remainder); err != nil { + s.logger.Warn("写入流量小数余量失败，本次仅按整数扣减", + zap.String("carrier_type", carrierType), + zap.Uint("carrier_id", carrierID), + zap.Float64("remainder_mb", remainder), + zap.Error(err)) + } + + return deductUsageMB +} + +// getUsageRemainder 读取载体流量小数余量 +func (s *UsageService) getUsageRemainder(ctx context.Context, key string) (float64, error) { + value, err := s.redis.Get(ctx, key).Result() + if err == redis.Nil { + return 0, nil + } + if err != nil { + return 0, err + } + remainder, parseErr := strconv.ParseFloat(value, 64) + if parseErr != nil { + // 脏值直接删除，避免持续影响后续扣减 + _ = s.redis.Del(ctx, key).Err() + return 0, parseErr + } + return remainder, nil +} + +// persistUsageRemainder 持久化载体流量小数余量 +func (s *UsageService) persistUsageRemainder(ctx context.Context, key string, remainder float64) error { + if remainder <= usageRemainderEpsilon { + return s.redis.Del(ctx, key).Err() + } + value := strconv.FormatFloat(remainder, 'f', 6, 64) + return s.redis.Set(ctx, key, value, usageRemainderTTL).Err() +} + +// resolveActivePackages 统一解析可扣减载体 +// 优先按请求载体查找；当 iot_card 无生效套餐且存在设备绑定时，回退到 device 载体 +func (s *UsageService) resolveActivePackages(ctx context.Context, tx *gorm.DB, carrierType string, carrierID uint) (string, uint, []*model.PackageUsage, error) { + packages, err := s.queryActivePackages(tx, carrierType, carrierID) + if err != nil { + return "", 0, nil, err + } + if len(packages) > 0 { + return carrierType, carrierID, packages, nil + } + + if carrierType != constants.AssetTypeIotCard { + return carrierType, carrierID, packages, nil + } + + binding, err := s.deviceSimBindingStore.GetActiveBindingByCardID(ctx, carrierID) + if err == gorm.ErrRecordNotFound { + return carrierType, carrierID, packages, nil + } + if err != nil { + return "", 0, nil, errors.Wrap(errors.CodeDatabaseError, err, "查询卡绑定设备失败") + } + + devicePackages, queryErr := s.queryActivePackages(tx, constants.AssetTypeDevice, binding.DeviceID) + if queryErr != nil { + return "", 0, nil, queryErr + } + if len(devicePackages) == 0 { + return carrierType, carrierID, packages, nil + } + + s.logger.Info("卡载体无生效套餐，回退设备载体扣减流量", + zap.Uint("card_id", carrierID), + zap.Uint("device_id", binding.DeviceID)) + + return constants.AssetTypeDevice, binding.DeviceID, devicePackages, nil +} + +// queryActivePackages 查询指定载体下的生效套餐（按扣减优先级排序） +func (s *UsageService) queryActivePackages(tx *gorm.DB, carrierType string, carrierID uint) ([]*model.PackageUsage, error) { + query := tx.Where("status = ?", constants.PackageUsageStatusActive) + + switch carrierType { + case constants.AssetTypeIotCard: + query = query.Where("iot_card_id = ?", carrierID) + case constants.AssetTypeDevice: + query = query.Where("device_id = ?", carrierID) + default: + return nil, errors.New(errors.CodeInvalidParam, "无效的载体类型") + } + + var packages []*model.PackageUsage + if err := query. + Order("CASE WHEN master_usage_id IS NOT NULL THEN 0 ELSE 1 END, priority ASC"). + Find(&packages).Error; err != nil { + return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询生效套餐失败") + } + return packages, nil +} + // updateDailyRecord 任务 10.6: 更新日流量记录 func (s *UsageService) updateDailyRecord(ctx context.Context, tx *gorm.DB, packageUsageID uint, dateStr string, dailyUsageMB, cumulativeUsageMB int64) error { // 解析日期字符串 @@ -207,10 +334,12 @@ func (s *UsageService) checkAndTriggerSuspension(ctx context.Context, tx *gorm.D query := tx.Model(&model.PackageUsage{}). Where("status = ?", constants.PackageUsageStatusActive) - if carrierType == "iot_card" { + if carrierType == constants.AssetTypeIotCard { query = query.Where("iot_card_id = ?", carrierID) - } else if carrierType == "device" { + } else if carrierType == constants.AssetTypeDevice { query = query.Where("device_id = ?", carrierID) + } else { + return errors.New(errors.CodeInvalidParam, "无效的载体类型") } if err := query.Count(&activeCount).Error; err != nil { @@ -223,7 +352,7 @@ func (s *UsageService) checkAndTriggerSuspension(ctx context.Context, tx *gorm.D zap.String("carrier_type", carrierType), zap.Uint("carrier_id", carrierID)) - if carrierType == "iot_card" { + if carrierType == constants.AssetTypeIotCard { if s.stopResumeCallback != nil { go func() { stopCtx := context.Background() @@ -234,7 +363,7 @@ func (s *UsageService) checkAndTriggerSuspension(ctx context.Context, tx *gorm.D } }() } - } else if carrierType == "device" { + } else if carrierType == constants.AssetTypeDevice { if s.stopResumeCallback == nil { s.logger.Warn("停复机回调未注入，跳过设备绑定卡停机", zap.Uint("device_id", carrierID)) diff --git a/internal/task/polling_carddata_handler.go b/internal/task/polling_carddata_handler.go index df06d13..66ba229 100644 --- a/internal/task/polling_carddata_handler.go +++ b/internal/task/polling_carddata_handler.go @@ -120,7 +120,7 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, t *asynq.Task) erro h.base.updateCardCache(ctx, cardID, cacheUpdates) if flowIncrementMB > 0 && h.usageService != nil { - if deductErr := h.usageService.DeductDataUsage(ctx, "iot_card", cardID, int64(flowIncrementMB)); deductErr != nil { + if deductErr := h.usageService.DeductDataUsage(ctx, constants.AssetTypeIotCard, cardID, flowIncrementMB); deductErr != nil { h.base.logger.Warn("套餐流量扣减失败", zap.Uint("card_id", cardID), zap.Float64("increment_mb", flowIncrementMB), zap.Error(deductErr)) } diff --git a/pkg/constants/redis.go b/pkg/constants/redis.go index 152f35d..d8f3590 100644 --- a/pkg/constants/redis.go +++ b/pkg/constants/redis.go @@ -408,6 +408,13 @@ func RedisCardDailyTrafficKey(cardID uint, date string) string { return fmt.Sprintf("traffic:daily:%d:%s", cardID, date) } +// RedisPackageUsageRemainderKey 生成套餐扣减流量小数余量缓存键 +// 用途：缓存载体维度的 <1MB 小数余量，累计后再触发整数 MB 扣减 +// 过期时间：7 天 +func RedisPackageUsageRemainderKey(carrierType string, carrierID uint) string { + return fmt.Sprintf("package:usage:remainder:%s:%d", carrierType, carrierID) +} + // ======================================== // 轮询分片队列相关 Redis Key // ========================================