package packagepkg import ( "context" "time" "github.com/break/junhong_cmp_fiber/internal/model" "github.com/break/junhong_cmp_fiber/internal/store/postgres" "github.com/break/junhong_cmp_fiber/pkg/constants" "github.com/break/junhong_cmp_fiber/pkg/errors" "github.com/redis/go-redis/v9" "go.uber.org/zap" "gorm.io/gorm" ) // StopResumeCallback 任务 24.6: 停复机回调接口 // 用于在流量用完时触发停机操作 type StopResumeCallback interface { // CheckAndStopCard 检查流量耗尽并停机 CheckAndStopCard(ctx context.Context, cardID uint) error } type UsageService struct { db *gorm.DB redis *redis.Client packageUsageStore *postgres.PackageUsageStore packageUsageDailyRecord *postgres.PackageUsageDailyRecordStore logger *zap.Logger stopResumeCallback StopResumeCallback // 停复机回调,可选 } func NewUsageService( db *gorm.DB, redis *redis.Client, packageUsageStore *postgres.PackageUsageStore, packageUsageDailyRecord *postgres.PackageUsageDailyRecordStore, logger *zap.Logger, ) *UsageService { return &UsageService{ db: db, redis: redis, packageUsageStore: packageUsageStore, packageUsageDailyRecord: packageUsageDailyRecord, logger: logger, } } // SetStopResumeCallback 任务 24.6: 设置停复机回调 // 在应用启动时由 bootstrap 调用,注入停复机服务 func (s *UsageService) SetStopResumeCallback(callback StopResumeCallback) { s.stopResumeCallback = callback } // DeductDataUsage 任务 10.2-10.6: 按优先级扣减流量 // 扣减顺序:加油包(按 priority ASC) → 主套餐 // 流量用完时自动标记 status=2,所有套餐用完时触发停机 func (s *UsageService) DeductDataUsage(ctx context.Context, carrierType string, carrierID uint, usageMB int64) error { if usageMB <= 0 { return errors.New(errors.CodeInvalidParam, "扣减流量必须大于0") } 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, "查询生效套餐失败") } if len(packages) == 0 { return errors.New(errors.CodeNoAvailablePackage, "没有可用套餐") } // 按优先级扣减流量 remainingUsage := usageMB today := time.Now().Format("2006-01-02") for _, pkg := range packages { if remainingUsage <= 0 { break } // 计算当前套餐剩余额度 remainingQuota := pkg.DataLimitMB - pkg.DataUsageMB if remainingQuota <= 0 { // 套餐已用完,标记为已用完 if err := tx.Model(pkg).Update("status", constants.PackageUsageStatusDepleted).Error; err != nil { return errors.Wrap(errors.CodeDatabaseError, err, "更新套餐状态失败") } continue } // 本次从该套餐扣减的流量 var deductFromPkg int64 if remainingUsage <= remainingQuota { deductFromPkg = remainingUsage } else { deductFromPkg = remainingQuota } // 更新套餐使用量 newUsage := pkg.DataUsageMB + deductFromPkg updates := map[string]interface{}{ "data_usage_mb": newUsage, } // 检查是否用完 if newUsage >= pkg.DataLimitMB { updates["status"] = constants.PackageUsageStatusDepleted } if err := tx.Model(pkg).Updates(updates).Error; err != nil { return errors.Wrap(errors.CodeDatabaseError, err, "更新套餐使用量失败") } // 任务 10.6: 写入日记录 if err := s.updateDailyRecord(ctx, tx, pkg.ID, today, deductFromPkg, newUsage); err != nil { return err } remainingUsage -= deductFromPkg s.logger.Info("扣减套餐流量", zap.Uint("usage_id", pkg.ID), zap.Int64("deduct_mb", deductFromPkg), zap.Int64("new_usage_mb", newUsage), zap.Int64("data_limit_mb", pkg.DataLimitMB)) } // 如果流量扣减未完成,说明所有套餐都不够 if remainingUsage > 0 { s.logger.Warn("流量不足", zap.String("carrier_type", carrierType), zap.Uint("carrier_id", carrierID), zap.Int64("requested_mb", usageMB), zap.Int64("remaining_mb", remainingUsage)) return errors.New(errors.CodeInsufficientQuota, "流量不足") } // 任务 10.5: 检查是否所有套餐都用完(触发停机) if err := s.checkAndTriggerSuspension(ctx, tx, carrierType, carrierID); err != nil { return err } return nil }) } // updateDailyRecord 任务 10.6: 更新日流量记录 func (s *UsageService) updateDailyRecord(ctx context.Context, tx *gorm.DB, packageUsageID uint, dateStr string, dailyUsageMB, cumulativeUsageMB int64) error { // 解析日期字符串 date, err := time.Parse("2006-01-02", dateStr) if err != nil { return errors.Wrap(errors.CodeInvalidParam, err, "日期格式错误") } // 查询是否已有今日记录 var record model.PackageUsageDailyRecord err = tx.Where("package_usage_id = ? AND date = ?", packageUsageID, date). First(&record).Error if err == gorm.ErrRecordNotFound { // 创建新记录 record = model.PackageUsageDailyRecord{ PackageUsageID: packageUsageID, Date: date, DailyUsageMB: int(dailyUsageMB), CumulativeUsageMB: cumulativeUsageMB, } if err := tx.Create(&record).Error; err != nil { return errors.Wrap(errors.CodeDatabaseError, err, "创建日流量记录失败") } } else if err != nil { return errors.Wrap(errors.CodeDatabaseError, err, "查询日流量记录失败") } else { // 更新现有记录 updates := map[string]interface{}{ "daily_usage_mb": record.DailyUsageMB + int(dailyUsageMB), "cumulative_usage_mb": cumulativeUsageMB, } if err := tx.Model(&record).Updates(updates).Error; err != nil { return errors.Wrap(errors.CodeDatabaseError, err, "更新日流量记录失败") } } return nil } // checkAndTriggerSuspension 任务 10.5: 检查停机条件 func (s *UsageService) checkAndTriggerSuspension(ctx context.Context, tx *gorm.DB, carrierType string, carrierID uint) error { // 查询是否还有生效中的套餐 var activeCount int64 query := tx.Model(&model.PackageUsage{}). 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) } if err := query.Count(&activeCount).Error; err != nil { return errors.Wrap(errors.CodeDatabaseError, err, "查询生效套餐数量失败") } // 如果没有生效中的套餐,触发停机操作 if activeCount == 0 { s.logger.Warn("所有套餐已用完,触发停机", zap.String("carrier_type", carrierType), zap.Uint("carrier_id", carrierID)) // 任务 24.6: 调用停复机服务执行停机 if s.stopResumeCallback != nil && carrierType == "iot_card" { // 在事务外异步执行停机,避免长事务 go func() { stopCtx := context.Background() if err := s.stopResumeCallback.CheckAndStopCard(stopCtx, carrierID); err != nil { s.logger.Error("调用停机服务失败", zap.Uint("card_id", carrierID), zap.Error(err)) } }() } } return nil }