修复设备套餐流量不同步的问题

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))