package service import ( "errors" "fmt" "math" "strconv" "strings" "time" "gorm.io/gorm" "gorm.io/gorm/clause" "laic-backend/cache" "laic-backend/common" "laic-backend/logger" "laic-backend/model" "laic-backend/tool" "laic-backend/vo" ) const ( gbBytes = 1024 * 1024 * 1024 // 1 GB 字节数 defaultTrafficUnitPrice = 10.0 // 云媒体流量单价(元/GB) ) // trafficDeductLua 原子扣减:key 不存在时用 MySQL 快照兜底初始化,再判断并扣减。 // 返回 {status, before, after}:status=1 成功,0 余额不足。 const trafficDeductLua = ` local key = KEYS[1] local delta = tonumber(ARGV[1]) local fallback = tonumber(ARGV[2]) local before = 0 if redis.call('EXISTS', key) == 0 then redis.call('SET', key, fallback) before = fallback else before = tonumber(redis.call('GET', key)) end if before < delta then return {0, before, before} end local after = redis.call('DECRBY', key, delta) return {1, before, after} ` type BillingService struct{} var DefaultBillingService = &BillingService{} // CarrierAPI 运营商接口抽象(查询用量 / 充值) type CarrierAPI interface { QueryUsage(iccid string) (usedGb float64, carrierStatus string, err error) Recharge(iccid string, amountGb int) (orderNo string, err error) } // MockCarrierAPI 运营商 Mock(仅供开发注入,不作为默认生产通道) type MockCarrierAPI struct{} func (m *MockCarrierAPI) QueryUsage(iccid string) (float64, string, error) { return 0.0, "normal", nil } func (m *MockCarrierAPI) Recharge(iccid string, amountGb int) (string, error) { return fmt.Sprintf("MOCK-%d", time.Now().UnixNano()), nil } // carrier 默认禁用,未接入真实运营商前不产生虚假用量或充值成功。 var carrier CarrierAPI // mysqlBalance 读取用户 MySQL 流量快照 func (b *BillingService) mysqlBalance(userID int64) (int64, error) { var u model.User if err := common.DB.Select("traffic_balance").First(&u, userID).Error; err != nil { return 0, err } return u.TrafficBalance, nil } // GetBalance 查询流量余额(Redis 权威,缺 key 时用 MySQL 快照兜底) func (b *BillingService) GetBalance(userID int64) (*vo.TrafficBalanceVO, *common.BusiError) { balance, err := b.currentBalance(userID) if err != nil { logger.ERROR("查询流量余额失败", err) return nil, common.ErrInternal } return &vo.TrafficBalanceVO{ BalanceBytes: balance, BalanceGb: float64(balance) / float64(gbBytes), }, nil } // currentBalance 返回 Redis 权威余额;缺 key 时先用 MySQL 快照 SetNX 初始化 func (b *BillingService) currentBalance(userID int64) (int64, error) { key := cache.TrafficKeyOf(userID) bal, err := common.GetInt64(key) if err == nil { return bal, nil } fallback, err := b.mysqlBalance(userID) if err != nil { return 0, err } if _, err := common.SetNX(key, fallback); err != nil { return 0, err } // SetNX 失败说明并发下已有值,重读一次 bal, err = common.GetInt64(key) if err != nil { return fallback, nil } return bal, nil } // Deduct 在一个事务内扣减用户余额和平台共享资源,并写入统一账务流水。 // idempotencyKey 必须由调用方按业务事件稳定生成。 func (b *BillingService) Deduct(userID int64, bytes int64, sourceType string, sourceID int64, idempotencyKey string) (bool, *common.BusiError) { if bytes <= 0 { return true, nil } if idempotencyKey == "" { return false, common.ErrParam } now := time.Now() var busiErr *common.BusiError var alreadyDeducted bool err := common.DB.Transaction(func(tx *gorm.DB) error { var existing model.TrafficLedger if err := tx.Where("idempotency_key = ?", idempotencyKey+":user").First(&existing).Error; err == nil { alreadyDeducted = true return nil } else if !errors.Is(err, gorm.ErrRecordNotFound) { return err } var user model.User if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id", "traffic_balance").First(&user, userID).Error; err != nil { return err } if user.TrafficBalance < bytes { busiErr = common.ErrTrafficNotEnough return busiErr } var pool model.PlatformResourcePool if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Order("id ASC").First(&pool).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { busiErr = common.ErrPlatformTrafficExhausted } return err } if pool.AvailableBytes < bytes || pool.Status != "active" { busiErr = common.ErrPlatformTrafficExhausted return busiErr } userAfter := user.TrafficBalance - bytes platformAfter := pool.AvailableBytes - bytes if err := tx.Model(&model.User{}).Where("id = ?", userID).UpdateColumn("traffic_balance", userAfter).Error; err != nil { return err } if err := tx.Model(&pool).Updates(map[string]any{ "available_bytes": platformAfter, "consumed_bytes": gorm.Expr("consumed_bytes + ?", bytes), "version": gorm.Expr("version + 1"), "updated_at": now, }).Error; err != nil { return err } if err := tx.Create(&model.TrafficLedger{ ID: mustID(), AccountType: "user", AccountID: userID, Direction: "debit", AmountBytes: bytes, BalanceBefore: user.TrafficBalance, BalanceAfter: userAfter, SourceType: sourceType, SourceID: strconv.FormatInt(sourceID, 10), IdempotencyKey: idempotencyKey + ":user", CreatedAt: now, }).Error; err != nil { return err } if err := tx.Create(&model.TrafficLedger{ ID: mustID(), AccountType: "platform", AccountID: pool.ID, Direction: "debit", AmountBytes: bytes, BalanceBefore: pool.AvailableBytes, BalanceAfter: platformAfter, SourceType: sourceType, SourceID: strconv.FormatInt(sourceID, 10), IdempotencyKey: idempotencyKey + ":platform", CreatedAt: now, }).Error; err != nil { return err } return tx.Create(&model.TrafficUsageLog{ ID: mustID(), UserID: userID, SourceType: sourceType, SourceID: sourceID, BytesUsed: bytes, BalanceBefore: user.TrafficBalance, BalanceAfter: userAfter, IdempotencyKey: idempotencyKey, CreatedAt: now, }).Error }) if err != nil { if busiErr != nil { return false, busiErr } logger.ERROR("流量扣费事务失败", err) return false, common.ErrInternal } if alreadyDeducted { return true, nil } if err := common.Delete(cache.TrafficKeyOf(userID)); err != nil { logger.WARN("删除流量余额缓存失败", err) } return true, nil } // credit 充值入账:Redis 权威自增 + MySQL 快照同步 func (b *BillingService) credit(userID int64, bytes int64) error { key := cache.TrafficKeyOf(userID) if _, err := b.currentBalance(userID); err != nil { return err } _, err := common.IncrBy(key, bytes) return err } // GetUsagePage 流量消费流水分页 func (b *BillingService) GetUsagePage(userID int64, req *vo.UsagePageReq) (*common.PageResponse[model.TrafficUsageLog], *common.BusiError) { db := common.DB.Model(&model.TrafficUsageLog{}).Where("user_id = ?", userID) if req.SourceType != "" { db = db.Where("source_type = ?", req.SourceType) } var total int64 if err := db.Count(&total).Error; err != nil { return nil, common.ErrInternal } var list []model.TrafficUsageLog if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { return nil, common.ErrInternal } return common.Page(req.Pagination, total, list), nil } // GetOrderPage 流量订单分页 func (b *BillingService) GetOrderPage(userID int64, req *vo.OrderPageReq) (*common.PageResponse[model.TrafficOrder], *common.BusiError) { db := common.DB.Model(&model.TrafficOrder{}).Where("user_id = ?", userID) if req.PayStatus != "" { db = db.Where("pay_status = ?", req.PayStatus) } var total int64 if err := db.Count(&total).Error; err != nil { return nil, common.ErrInternal } var list []model.TrafficOrder if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { return nil, common.ErrInternal } return common.Page(req.Pagination, total, list), nil } func (b *BillingService) CreateOrder(userID int64, amountGb int) (*model.TrafficOrder, *common.BusiError) { return b.CreateOrderWithPackage(userID, 0, amountGb) } func (b *BillingService) CreateOrderWithPackage(userID, packageID int64, amountGb int) (*model.TrafficOrder, *common.BusiError) { order := &model.TrafficOrder{ID: mustID(), UserID: userID, PayStatus: "unpaid", CreditStatus: "pending", CreatedAt: time.Now(), UpdatedAt: time.Now()} var busiErr *common.BusiError err := common.DB.Transaction(func(tx *gorm.DB) error { var user model.User if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id").First(&user, userID).Error; err != nil { return err } var pending model.TrafficOrder if err := tx.Where("user_id = ? AND pay_status IN ?", userID, []string{"unpaid", "processing"}).First(&pending).Error; err == nil { busiErr = common.ErrPendingOrderExists return busiErr } else if !errors.Is(err, gorm.ErrRecordNotFound) { return err } if packageID > 0 { var pkg model.TrafficPackage if err := tx.Where("id = ? AND status = ?", packageID, "active").First(&pkg).Error; err != nil { return err } order.PackageID, order.PackageCode = pkg.ID, pkg.Code order.AmountBytes, order.UnitPrice, order.TotalPrice = pkg.AmountBytes, pkg.Price/float64(pkg.AmountBytes)*float64(gbBytes), pkg.Price order.TotalFeeFen = int64(math.Round(pkg.Price * 100)) order.AmountGb = int(pkg.AmountBytes / gbBytes) } else { if amountGb <= 0 { busiErr = common.ErrParam return busiErr } order.AmountGb, order.AmountBytes = amountGb, int64(amountGb)*gbBytes order.UnitPrice, order.TotalPrice = defaultTrafficUnitPrice, float64(amountGb)*defaultTrafficUnitPrice order.TotalFeeFen = int64(math.Round(float64(amountGb) * defaultTrafficUnitPrice * 100)) } return tx.Create(order).Error }) if err != nil { if busiErr != nil { return nil, busiErr } if errors.Is(err, gorm.ErrRecordNotFound) { return nil, common.ErrUserNotFound } if code, _ := common.ParseError(err); code == 1062 { return nil, common.ErrPendingOrderExists } logger.ERROR("创建流量订单失败", err) return nil, common.ErrInternal } DefaultOperationLogService.RecordEvent(userID, "充值与履约", "创建流量充值订单", fmt.Sprintf("orderId=%d amountGb=%d feeFen=%d", order.ID, order.AmountGb, order.TotalFeeFen), "success", "user_api") return order, nil } func (b *BillingService) CancelOrder(userID, orderID int64) *common.BusiError { now := time.Now() var busiErr *common.BusiError err := common.DB.Transaction(func(tx *gorm.DB) error { var order model.TrafficOrder if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("id = ? AND user_id = ?", orderID, userID).First(&order).Error; err != nil { return err } if order.PayStatus != "unpaid" && order.PayStatus != "processing" { busiErr = common.ErrOrderNotCancellable return busiErr } result := tx.Model(&order).Where("id = ? AND pay_status IN ?", orderID, []string{"unpaid", "processing"}).Updates(map[string]any{"pay_status": "cancelled", "closed_at": now, "updated_at": now}) if result.Error != nil { return result.Error } if result.RowsAffected != 1 { busiErr = common.ErrOrderNotCancellable return busiErr } return tx.Model(&model.PaymentTransaction{}).Where("business_type = ? AND business_order_id = ? AND status IN ?", "traffic_order", orderID, []string{"unpaid", "processing"}).Updates(map[string]any{"status": "cancelled", "updated_at": now}).Error }) if err != nil { if busiErr != nil { return busiErr } if errors.Is(err, gorm.ErrRecordNotFound) { return common.ErrOrderNotFound } logger.ERROR("取消流量订单失败", err) return common.ErrInternal } DefaultOperationLogService.RecordEvent(userID, "充值与履约", "取消流量充值订单", fmt.Sprintf("orderId=%d status=cancelled", orderID), "success", "user_api") return nil } func (b *BillingService) CloseExpiredOrders() { cutoff := time.Now().Add(-30 * time.Minute) var orders []model.TrafficOrder if err := common.DB.Where("pay_status IN ? AND created_at <= ?", []string{"unpaid", "processing"}, cutoff).Limit(100).Find(&orders).Error; err != nil { logger.ERROR("扫描超时流量订单失败", err) } else { for i := range orders { b.closeTrafficOrder(orders[i].ID, cutoff) } } DefaultSimRechargeService.CloseExpiredOrders(cutoff) } func (b *BillingService) closeTrafficOrder(orderID int64, cutoff time.Time) { now := time.Now() closed := false var order model.TrafficOrder err := common.DB.Transaction(func(tx *gorm.DB) error { if err := tx.Where("id = ?", orderID).First(&order).Error; err != nil { return err } result := tx.Model(&model.TrafficOrder{}).Where("id = ? AND pay_status IN ? AND created_at <= ?", orderID, []string{"unpaid", "processing"}, cutoff).Updates(map[string]any{"pay_status": "closed", "closed_at": now, "updated_at": now}) if result.Error != nil { return result.Error } if result.RowsAffected != 1 { return nil } closed = true return tx.Model(&model.PaymentTransaction{}).Where("business_type = ? AND business_order_id = ? AND status IN ?", "traffic_order", orderID, []string{"unpaid", "processing"}).Updates(map[string]any{"status": "closed", "updated_at": now}).Error }) if err != nil { logger.ERROR("关闭超时流量订单失败", err) return } if closed && order.UserID != 0 { DefaultOperationLogService.RecordEvent(order.UserID, "充值与履约", "自动关闭流量充值订单", fmt.Sprintf("orderId=%d status=closed", orderID), "success", "system") } } // ListSimCards SIM 卡列表 func (b *BillingService) ListSimCards(userID int64, req *vo.SimCardPageReq) (*common.PageResponse[model.SimCard], *common.BusiError) { db := common.DB.Model(&model.SimCard{}).Where("user_id = ?", userID) if req.Status != "" { db = db.Where("status = ?", req.Status) } var total int64 if err := db.Count(&total).Error; err != nil { return nil, common.ErrInternal } var list []model.SimCard if err := db.Scopes(req.Paginate).Order("updated_at DESC, id DESC").Find(&list).Error; err != nil { return nil, common.ErrInternal } return common.Page(req.Pagination, total, list), nil } // GetSimRechargeLogPage SIM 卡充值记录分页 func (b *BillingService) GetSimRechargeLogPage(userID int64, req *vo.SimRechargeLogPageReq) (*common.PageResponse[model.SimRechargeLog], *common.BusiError) { db := common.DB.Model(&model.SimRechargeLog{}).Where("user_id = ?", userID) if req.SimCardID > 0 { db = db.Where("sim_card_id = ?", req.SimCardID) } if req.RechargeStatus != "" { db = db.Where("recharge_status = ?", req.RechargeStatus) } var total int64 if err := db.Count(&total).Error; err != nil { return nil, common.ErrInternal } var list []model.SimRechargeLog if err := db.Scopes(req.Paginate).Order("created_at DESC, id DESC").Find(&list).Error; err != nil { return nil, common.ErrInternal } return common.Page(req.Pagination, total, list), nil } // RechargeSimCard 当前未接入支付确认流程,不向运营商提交充值订单。 func (b *BillingService) RechargeSimCard(userID, simCardID int64, amountGb int) (*model.SimRechargeLog, *common.BusiError) { return nil, common.ErrCarrierUnavailable } // mustID 生成雪花 ID(失败返回 0,由 DB 约束兜底) func mustID() int64 { id, err := tool.NextID() if err != nil { logger.ERROR("生成雪花 ID 失败", err) return 0 } return id } // chargeLiveSessions 只结算设备与云端均确认在线的 streaming 会话。 func (b *BillingService) chargeLiveSessions() { var sessions []model.LiveSession if err := common.DB.Where("phase = 'streaming' AND device_streaming = ? AND cloud_online = ? AND started_at IS NOT NULL", true, true).Find(&sessions).Error; err != nil { logger.ERROR("扫描直播会话失败", err) return } now := time.Now().Truncate(time.Second) for i := range sessions { b.chargeLiveSession(&sessions[i], now) } } // chargeLiveSession 通过会话行锁和唯一直播计费段避免多实例重复结算。 func (b *BillingService) chargeLiveSession(session *model.LiveSession, now time.Time) { var userID int64 var needStop bool var charged bool err := common.DB.Transaction(func(tx *gorm.DB) error { var current model.LiveSession if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(¤t, "id = ?", session.ID).Error; err != nil { return err } if current.Phase != "streaming" || !current.DeviceStreaming || !current.CloudOnline || current.StartedAt == nil { return nil } var dock model.Dock if err := tx.Select("user_id").Where("dock_id = ?", current.DockID).First(&dock).Error; err != nil { return err } userID = dock.UserID periodStart := *current.StartedAt var last model.LiveBillingSegment if err := tx.Where("session_id = ?", current.ID).Order("period_end DESC").First(&last).Error; err == nil { periodStart = last.PeriodEnd } else if !errors.Is(err, gorm.ErrRecordNotFound) { return err } if !now.After(periodStart) { return nil } amount := int64(now.Sub(periodStart).Seconds() * float64(current.MaxBitrateBps) / 8) if amount <= 0 { return nil } idempotencyKey := fmt.Sprintf("live:%s:%d:%d", current.ID, periodStart.Unix(), now.Unix()) segment := model.LiveBillingSegment{ ID: mustID(), SessionID: current.ID, UserID: userID, PeriodStart: periodStart, PeriodEnd: now, Bytes: amount, Status: "pending", IdempotencyKey: idempotencyKey, CreatedAt: now, } if err := tx.Create(&segment).Error; err != nil { if code, _ := common.ParseError(err); code == 1062 { return nil } return err } var user model.User if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id", "traffic_balance").First(&user, userID).Error; err != nil { return err } var pool model.PlatformResourcePool if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Order("id ASC").First(&pool).Error; err != nil { return err } if user.TrafficBalance < amount || pool.Status != "active" || pool.AvailableBytes < amount { if err := tx.Model(&segment).Update("status", "insufficient").Error; err != nil { return err } result := tx.Model(&model.LiveSession{}).Where("id = ? AND phase = 'streaming'", current.ID).Updates(map[string]any{ "phase": "stopping", "stop_reason": "no_balance", "error_code": "TRAFFIC_NOT_ENOUGH", "updated_at": now, }) if result.Error != nil { return result.Error } needStop = result.RowsAffected == 1 return nil } userAfter, platformAfter := user.TrafficBalance-amount, pool.AvailableBytes-amount if err := tx.Model(&model.User{}).Where("id = ?", user.ID).UpdateColumn("traffic_balance", userAfter).Error; err != nil { return err } if err := tx.Model(&pool).Updates(map[string]any{ "available_bytes": platformAfter, "consumed_bytes": gorm.Expr("consumed_bytes + ?", amount), "version": gorm.Expr("version + 1"), "updated_at": now, }).Error; err != nil { return err } if err := tx.Create(&model.TrafficLedger{ID: mustID(), AccountType: "user", AccountID: user.ID, Direction: "debit", AmountBytes: amount, BalanceBefore: user.TrafficBalance, BalanceAfter: userAfter, SourceType: "live", SourceID: current.ID, IdempotencyKey: idempotencyKey + ":user", CreatedAt: now}).Error; err != nil { return err } if err := tx.Create(&model.TrafficLedger{ID: mustID(), AccountType: "platform", AccountID: pool.ID, Direction: "debit", AmountBytes: amount, BalanceBefore: pool.AvailableBytes, BalanceAfter: platformAfter, SourceType: "live", SourceID: current.ID, IdempotencyKey: idempotencyKey + ":platform", CreatedAt: now}).Error; err != nil { return err } if err := tx.Create(&model.TrafficUsageLog{ID: mustID(), UserID: user.ID, SourceType: "live", SourceID: 0, BytesUsed: amount, BalanceBefore: user.TrafficBalance, BalanceAfter: userAfter, IdempotencyKey: idempotencyKey, CreatedAt: now}).Error; err != nil { return err } if err := tx.Model(&segment).Updates(map[string]any{"status": "settled", "settled_at": now}).Error; err != nil { return err } charged = true return nil }) if err != nil { logger.ERROR("直播计费事务失败", err) return } if needStop { if _, busiErr := DefaultLiveService.stopForBilling(session.DockID, session.ID); busiErr != nil { logger.ERROR("直播余额不足停止推流失败", busiErr) } return } if charged { if err := common.Delete(cache.TrafficKeyOf(userID)); err != nil { logger.WARN("删除直播余额缓存失败", err) } } } // flushBalanceSnapshot 余额快照刷新(定时任务,每 30min):Redis 权威余额回写 MySQL func (b *BillingService) flushBalanceSnapshot() { keys, err := common.GetKeysWithPrefix(cache.TrafficKeyPrefix) if err != nil { logger.ERROR("扫描流量余额 key 失败", err) return } for _, key := range keys { idStr := strings.TrimPrefix(key, cache.TrafficKeyPrefix) userID, err := strconv.ParseInt(idStr, 10, 64) if err != nil { continue } bal, err := common.GetInt64(key) if err != nil { continue } if err := common.DB.Model(&model.User{}).Where("id = ?", userID). UpdateColumn("traffic_balance", bal).Error; err != nil { logger.ERROR("回写余额快照失败", err) } } } // syncSimUsage SIM 卡用量同步(定时任务,每 1h):运营商查询回填 + 用量快照 + 阈值告警 func (b *BillingService) syncSimUsage() { var cards []model.SimCard if err := common.DB.Where("status = ?", "active").Find(&cards).Error; err != nil { logger.ERROR("扫描 SIM 卡失败", err) return } now := time.Now() for i := range cards { b.syncSimCard(&cards[i], now) } } func (b *BillingService) syncSimCard(card *model.SimCard, now time.Time) { if carrier == nil { logger.INFO("SIM 运营商通道未接入:", card.Iccid) return } usedGb, carrierStatus, err := carrier.QueryUsage(card.Iccid) if err != nil { logger.WARN("查询 SIM 用量失败:", card.Iccid, err) return } updates := map[string]any{ "used_gb": usedGb, "carrier_status": carrierStatus, "last_sync_at": now, } if carrierStatus == "cancelled" { updates["status"] = "expired" } if err := common.DB.Model(&model.SimCard{}).Where("id = ?", card.ID).Updates(updates).Error; err != nil { logger.ERROR("回填 SIM 用量失败", err) } remainGb := float64(card.PlanGb) - usedGb record := &model.SimUsageRecord{ ID: mustID(), SimCardID: card.ID, PlanGb: card.PlanGb, UsedGb: usedGb, RemainGb: remainGb, CarrierStatus: carrierStatus, SyncedAt: now, } if err := common.DB.Create(record).Error; err != nil { logger.ERROR("写入 SIM 用量快照失败", err) } if remainGb < float64(card.PlanGb)*0.1 { logger.WARN("SIM 卡流量即将耗尽:", card.Iccid, "剩余", remainGb, "GB") } if carrierStatus == "suspended" || carrierStatus == "arrears" { logger.WARN("SIM 卡运营商状态异常:", card.Iccid, carrierStatus) } }