okooooooo/team-api
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(bill):修复并发扣费问题

Browse Source
This commit is contained in:
qian committed 2026-05-22 00:01:48 +08:00
1 parent 555bddf5c3
commit e251787051
14 files changed
+306 -165
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/dao/internal/bil_transactions.go
+6
View File
@@ -36,6 +36,9 @@ type BilTransactionsColumns struct {
UserId string // 关联用户ID（consume 类型为实际消费用户，recharge 类型为操作用户，adjust 类型为空）
RequestId string // 关联请求ID（consume 类型对应 API 调用的 request_id，其他类型为空）
ModelName string // 关联模型名（consume 类型为调用的模型名，其他类型为空）
ProjectId string // 关联项目ID（consume 类型为 API Key 所属项目，个人密钥为空）
ApiKeyId string // 关联API密钥ID（consume 类型为发起请求的密钥）
TaskId string // 关联异步任务公开ID（consume+relay_mode=task 时关联 tsk_model_tasks.public_task_id）
}
// bilTransactionsColumns holds the columns for the table bil_transactions.
@@ -55,6 +58,9 @@ var bilTransactionsColumns = BilTransactionsColumns{
UserId: "user_id",
RequestId: "request_id",
ModelName: "model_name",
ProjectId: "project_id",
ApiKeyId: "api_key_id",
TaskId: "task_id",
}
// NewBilTransactionsDao creates and returns a new DAO object for table data access.
internal/logic/billing/provider.go
+4 -4
View File
@@ -28,13 +28,13 @@ func (b *BillingProviderImpl) PreDeduct(ctx context.Context, tenantID int64, mod
return amount, nil
}
func (b *BillingProviderImpl) Settle(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID, relayMode string, usage *common.Usage, preDeductAmount float64) error {
func (b *BillingProviderImpl) Settle(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID, relayMode string, usage *common.Usage, preDeductAmount float64, projectID int64) error {
inputTokens, outputTokens := 0, 0
if usage != nil {
inputTokens = usage.PromptTokens
outputTokens = usage.CompletionTokens
}
_, err := Settle(ctx, tenantID, userID, apiKeyID, channelID, modelName, requestID, relayMode, inputTokens, outputTokens, preDeductAmount)
_, err := Settle(ctx, tenantID, userID, apiKeyID, channelID, modelName, requestID, relayMode, inputTokens, outputTokens, preDeductAmount, projectID)
return err
}
@@ -77,13 +77,13 @@ func (b *BillingProviderImpl) SettleFailed(ctx context.Context, tenantID int64,
return SettleFailed(ctx, tenantID, requestID, preDeductAmount)
}
func (b *BillingProviderImpl) SettleStreamInterrupted(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID, relayMode string, usage *common.Usage, preDeductAmount float64) error {
func (b *BillingProviderImpl) SettleStreamInterrupted(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID, relayMode string, usage *common.Usage, preDeductAmount float64, projectID int64) error {
confirmedInput, confirmedOutput := 0, 0
if usage != nil {
confirmedInput = usage.PromptTokens
confirmedOutput = usage.CompletionTokens
}
_, err := SettleStreamInterrupted(ctx, tenantID, userID, apiKeyID, channelID, modelName, requestID, relayMode, confirmedInput, confirmedOutput, preDeductAmount)
_, err := SettleStreamInterrupted(ctx, tenantID, userID, apiKeyID, channelID, modelName, requestID, relayMode, confirmedInput, confirmedOutput, preDeductAmount, projectID)
return err
}
internal/logic/billing/settlement.go
+130 -70
View File
@@ -6,6 +6,7 @@ import (
do "github.com/qianfree/team-api/internal/model/do"
"time"
"github.com/gogf/gf/v2/database/gdb"
"github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/os/gtime"
@@ -35,7 +36,7 @@ type SettlementResult struct {
func Settle(ctx context.Context, tenantID, userID, apiKeyID, channelID int64,
modelName, requestID, relayMode string,
inputTokens, outputTokens int,
preDeductAmount float64) (*SettlementResult, error) {
preDeductAmount float64, projectID int64) (*SettlementResult, error) {
// 1. 计算实际费用
breakdown, err := CalculateCost(ctx, tenantID, modelName, inputTokens, outputTokens)
@@ -59,46 +60,73 @@ func Settle(ctx context.Context, tenantID, userID, apiKeyID, channelID int64,
return nil, gerror.Wrapf(err, "settle: get wallet")
}
// 4. 执行结算 DB 操作
now := time.Now()
_, err = g.DB().Exec(ctx,
"UPDATE bil_wallets SET frozen_balance = GREATEST(frozen_balance - ?, 0), balance = balance - ?, updated_at = ? WHERE id = ?",
preDeductAmount, actualCost, now, wallet.ID)
// 4. 获取定价信息（事务外只读）
pricingResult, _ := GetModelPrice(ctx, tenantID, modelName)
// 5. 事务内执行结算（钱包扣款 + 计费记录）
var balanceAfter, frozenAfter float64
var billingID int64
err = g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
// 5a. 更新钱包
now := time.Now()
_, err := tx.Ctx(ctx).Exec(
"UPDATE bil_wallets SET frozen_balance = GREATEST(frozen_balance - ?, 0), balance = balance - ?, updated_at = ? WHERE id = ?",
preDeductAmount, actualCost, now, wallet.ID)
if err != nil {
return gerror.Wrapf(err, "settle: update wallet")
}
// 5b. 事务内读取准确余额
type balRow struct {
Balance float64 `json:"balance"`
FrozenBalance float64 `json:"frozen_balance"`
}
var br *balRow
err = tx.Model("bil_wallets").Ctx(ctx).
Where("id", wallet.ID).
Fields("balance, frozen_balance").
Scan(&br)
if err == nil && br != nil {
balanceAfter = br.Balance
frozenAfter = br.FrozenBalance
}
// 5c. 创建计费记录
var inputSnapPrice, outputSnapPrice float64
var billingMode string
var discountRatio, billingInputMult, billingOutputMult float64
if pricingResult != nil {
inputSnapPrice = pricingResult.InputPrice * pricingResult.ModelMultiplier * pricingResult.TenantMultiplier
outputSnapPrice = pricingResult.OutputPrice * pricingResult.ModelMultiplier * pricingResult.TenantMultiplier
billingMode = pricingResult.BillingMode
discountRatio = pricingResult.DiscountRatio
billingInputMult = breakdown.InputMultiplier
billingOutputMult = breakdown.OutputMultiplier
}
billingID, err = createBillingRecord(ctx, tx, tenantID, userID, apiKeyID, channelID,
modelName, requestID, relayMode, inputTokens, outputTokens,
inputSnapPrice, outputSnapPrice, actualCost,
billingMode, discountRatio, billingInputMult, billingOutputMult)
if err != nil {
return gerror.Wrapf(err, "settle: create billing record")
}
return nil
})
if err != nil {
return nil, gerror.Wrapf(err, "settle: update wallet")
return nil, err
}
// 清除钱包缓存（内存 + Redis）
// 6. 清除缓存（事务提交后）
walletCache.Delete(ctx, fmt.Sprintf("%d", tenantID))
InvalidateWalletRedis(ctx, tenantID)
CleanupPreDeduct(ctx, tenantID, requestID)
// 5. 创建计费记录（获取定价快照）
pricingResult, _ := GetModelPrice(ctx, tenantID, modelName)
var inputSnapPrice, outputSnapPrice float64
var billingMode string
var discountRatio, billingInputMult, billingOutputMult float64
if pricingResult != nil {
inputSnapPrice = pricingResult.InputPrice * pricingResult.ModelMultiplier * pricingResult.TenantMultiplier
outputSnapPrice = pricingResult.OutputPrice * pricingResult.ModelMultiplier * pricingResult.TenantMultiplier
billingMode = pricingResult.BillingMode
discountRatio = pricingResult.DiscountRatio
billingInputMult = breakdown.InputMultiplier
billingOutputMult = breakdown.OutputMultiplier
}
billingID, err := createBillingRecord(ctx, tenantID, userID, apiKeyID, channelID,
modelName, requestID, relayMode, inputTokens, outputTokens,
inputSnapPrice, outputSnapPrice, actualCost,
billingMode, discountRatio, billingInputMult, billingOutputMult)
if err != nil {
g.Log().Errorf(ctx, "settle: create billing record failed: %v", err)
}
// 6. 记录消费流水（一条汇总）
// 7. 记录消费流水（事务外，best-effort）
recordTransaction(ctx, wallet.ID, tenantID, "consume", -actualCost,
fmt.Sprintf("consume: %s model=%s input=%d output=%d pre_deduct=%.4f actual=%.4f", requestID, modelName, inputTokens, outputTokens, preDeductAmount, actualCost),
userID, requestID, modelName)
userID, requestID, modelName, billingID, "billing_record", projectID, apiKeyID, "", balanceAfter, frozenAfter)
return &SettlementResult{
PreDeductAmount: preDeductAmount,
@@ -137,43 +165,69 @@ func SettleWithUsage(ctx context.Context, tenantID, userID, apiKeyID, channelID
return nil, gerror.Wrapf(err, "settle_with_usage: get wallet")
}
// 4. 执行结算 DB 操作
now := time.Now()
_, err = g.DB().Exec(ctx,
"UPDATE bil_wallets SET frozen_balance = GREATEST(frozen_balance - ?, 0), balance = balance - ?, updated_at = ? WHERE id = ?",
preDeductAmount, actualCost, now, wallet.ID)
// 4. 获取定价信息（事务外只读）
pricingResult, _ := GetModelPrice(ctx, tenantID, modelName)
// 5. 事务内执行结算（钱包扣款 + 计费记录）
var balanceAfter, frozenAfter float64
var billingID int64
err = g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
// 5a. 更新钱包
now := time.Now()
_, err := tx.Ctx(ctx).Exec(
"UPDATE bil_wallets SET frozen_balance = GREATEST(frozen_balance - ?, 0), balance = balance - ?, updated_at = ? WHERE id = ?",
preDeductAmount, actualCost, now, wallet.ID)
if err != nil {
return gerror.Wrapf(err, "settle_with_usage: update wallet")
}
// 5b. 事务内读取准确余额
type balRow struct {
Balance float64 `json:"balance"`
FrozenBalance float64 `json:"frozen_balance"`
}
var br *balRow
err = tx.Model("bil_wallets").Ctx(ctx).
Where("id", wallet.ID).
Fields("balance, frozen_balance").
Scan(&br)
if err == nil && br != nil {
balanceAfter = br.Balance
frozenAfter = br.FrozenBalance
}
// 5c. 创建计费记录（含快照）
var inputSnapPrice, outputSnapPrice float64
var billingMode string
var discountRatio, billingInputMult, billingOutputMult float64
if pricingResult != nil {
inputSnapPrice = pricingResult.InputPrice * pricingResult.ModelMultiplier * pricingResult.TenantMultiplier
outputSnapPrice = pricingResult.OutputPrice * pricingResult.ModelMultiplier * pricingResult.TenantMultiplier
billingMode = pricingResult.BillingMode
discountRatio = pricingResult.DiscountRatio
billingInputMult = breakdown.InputMultiplier
billingOutputMult = breakdown.OutputMultiplier
}
billingID, err = createBillingRecordWithSnapshot(ctx, tx, tenantID, userID, apiKeyID, channelID,
modelName, requestID, relayMode, breakdown,
inputSnapPrice, outputSnapPrice, actualCost,
billingMode, discountRatio, billingInputMult, billingOutputMult, pricingResult)
if err != nil {
return gerror.Wrapf(err, "settle_with_usage: create billing record")
}
return nil
})
if err != nil {
return nil, gerror.Wrapf(err, "settle_with_usage: update wallet")
return nil, err
}
// 6. 清除缓存（事务提交后）
walletCache.Delete(ctx, fmt.Sprintf("%d", tenantID))
InvalidateWalletRedis(ctx, tenantID)
CleanupPreDeduct(ctx, tenantID, requestID)
// 5. 获取定价信息
pricingResult, _ := GetModelPrice(ctx, tenantID, modelName)
// 6. 创建计费记录（含快照）
var inputSnapPrice, outputSnapPrice float64
var billingMode string
var discountRatio, billingInputMult, billingOutputMult float64
if pricingResult != nil {
inputSnapPrice = pricingResult.InputPrice * pricingResult.ModelMultiplier * pricingResult.TenantMultiplier
outputSnapPrice = pricingResult.OutputPrice * pricingResult.ModelMultiplier * pricingResult.TenantMultiplier
billingMode = pricingResult.BillingMode
discountRatio = pricingResult.DiscountRatio
billingInputMult = breakdown.InputMultiplier
billingOutputMult = breakdown.OutputMultiplier
}
billingID, err := createBillingRecordWithSnapshot(ctx, tenantID, userID, apiKeyID, channelID,
modelName, requestID, relayMode, breakdown,
inputSnapPrice, outputSnapPrice, actualCost,
billingMode, discountRatio, billingInputMult, billingOutputMult, pricingResult)
if err != nil {
g.Log().Errorf(ctx, "settle_with_usage: create billing record failed: %v", err)
}
// 7. 生成计费快照和摘要文本
settlementResult := &SettlementResult{
PreDeductAmount: preDeductAmount,
@@ -200,10 +254,16 @@ func SettleWithUsage(ctx context.Context, tenantID, userID, apiKeyID, channelID
settlementResult.BillingSnapshot = snapshotJSON
settlementResult.BillingSummary = summaryText
// 8. 记录消费流水（一条汇总）
// 8. 从 relayInfo 提取 projectID
var txProjectID int64
if relayInfo != nil {
txProjectID = relayInfo.ProjectID
}
// 9. 记录消费流水（事务外，best-effort）
recordTransaction(ctx, wallet.ID, tenantID, "consume", -actualCost,
fmt.Sprintf("consume: %s model=%s input=%d output=%d pre_deduct=%.4f actual=%.4f", requestID, modelName, breakdown.InputTokens, breakdown.OutputTokens, preDeductAmount, actualCost),
userID, requestID, modelName)
userID, requestID, modelName, billingID, "billing_record", txProjectID, apiKeyID, "", balanceAfter, frozenAfter)
return settlementResult, nil
}
@@ -221,17 +281,17 @@ func SettleFailed(ctx context.Context, tenantID int64, requestID string, preDedu
func SettleStreamInterrupted(ctx context.Context, tenantID, userID, apiKeyID, channelID int64,
modelName, requestID, relayMode string,
confirmedInput, confirmedOutput int,
preDeductAmount float64) (*SettlementResult, error) {
preDeductAmount float64, projectID int64) (*SettlementResult, error) {
// 流式中断：按已确认的 token 计费
return Settle(ctx, tenantID, userID, apiKeyID, channelID,
modelName, requestID, relayMode,
confirmedInput, confirmedOutput,
preDeductAmount)
preDeductAmount, projectID)
}
// createBillingRecord 创建计费记录（含快照字段）
func createBillingRecord(ctx context.Context, tenantID, userID, apiKeyID, channelID int64,
func createBillingRecord(ctx context.Context, tx gdb.TX, tenantID, userID, apiKeyID, channelID int64,
modelName, requestID, relayMode string,
inputTokens, outputTokens int,
inputPrice, outputPrice, totalCost float64,
@@ -269,7 +329,7 @@ func createBillingRecord(ctx context.Context, tenantID, userID, apiKeyID, channe
data.BillingInputMultiplier = billingInputMult
data.BillingOutputMultiplier = billingOutputMult
result, err := dao.BilRecords.Ctx(ctx).Insert(data)
result, err := tx.Model("bil_records").Ctx(ctx).Data(data).Insert()
if err != nil {
return 0, err
}
@@ -316,7 +376,7 @@ func UpdateUsageLogCostWithSnapshot(ctx context.Context, requestID string, break
}
// createBillingRecordWithSnapshot 创建计费记录（含 cache token 和完整快照）
func createBillingRecordWithSnapshot(ctx context.Context, tenantID, userID, apiKeyID, channelID int64,
func createBillingRecordWithSnapshot(ctx context.Context, tx gdb.TX, tenantID, userID, apiKeyID, channelID int64,
modelName, requestID, relayMode string,
breakdown *CostBreakdown,
inputPrice, outputPrice, totalCost float64,
@@ -368,7 +428,7 @@ func createBillingRecordWithSnapshot(ctx context.Context, tenantID, userID, apiK
data.BaseOutputPrice = pricing.BaseOutputPrice
}
result, err := dao.BilRecords.Ctx(ctx).Insert(data)
result, err := tx.Model("bil_records").Ctx(ctx).Data(data).Insert()
if err != nil {
return 0, err
}
internal/logic/billing/task_billing.go
+95 -60
View File
@@ -2,13 +2,14 @@ package billing
import (
"context"
"database/sql"
"fmt"
"time"
"github.com/gogf/gf/v2/database/gdb"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/os/gtime"
"github.com/qianfree/team-api/internal/dao"
do "github.com/qianfree/team-api/internal/model/do"
"github.com/qianfree/team-api/relay/common"
)
@@ -78,33 +79,19 @@ func (b *TaskBillingProviderImpl) PreDeductTask(ctx context.Context, tenantID in
// SettleTaskSuccess 任务成功结算（含计费快照）
// totalTokens/completionTokens: 上游返回的 token 用量
// ratios: 提交时保存的计费比率（如 video_input 折扣）
func (b *TaskBillingProviderImpl) SettleTaskSuccess(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID string, actualCost, preDeductAmount float64, totalTokens, completionTokens int, ratios map[string]float64) (*common.SettlementResult, error) {
func (b *TaskBillingProviderImpl) SettleTaskSuccess(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID string, actualCost, preDeductAmount float64, totalTokens, completionTokens int, ratios map[string]float64, taskID string) (*common.SettlementResult, error) {
diff := actualCost - preDeductAmount
// 1. 钱包操作
// 1. 获取钱包
wallet, err := GetWallet(ctx, tenantID)
if err != nil {
return nil, fmt.Errorf("settle task: get wallet: %w", err)
}
now := time.Now()
_, err = g.DB().Exec(ctx,
"UPDATE bil_wallets SET frozen_balance = GREATEST(frozen_balance - ?, 0), balance = balance - ?, updated_at = ? WHERE id = ?",
preDeductAmount, actualCost, now, wallet.ID)
if err != nil {
return nil, fmt.Errorf("settle task: update wallet: %w", err)
}
walletCache.Delete(ctx, fmt.Sprintf("%d", tenantID))
InvalidateWalletRedis(ctx, tenantID)
CleanupPreDeduct(ctx, tenantID, requestID)
CleanupPreDeduct(ctx, tenantID, requestID+"_adjust")
// 2. 获取定价 + 构建 CostBreakdown
// 2. 获取定价（事务外只读）
pricing, _ := GetModelPrice(ctx, tenantID, modelName)
breakdown := buildTaskCostBreakdown(pricing, actualCost, totalTokens, completionTokens)
// 3. 创建计费记录（含快照字段）
var billingMode string
var discountRatio, effectiveOutputPrice float64
if pricing != nil {
@@ -113,40 +100,95 @@ func (b *TaskBillingProviderImpl) SettleTaskSuccess(ctx context.Context, tenantI
effectiveOutputPrice = pricing.OutputPrice
}
_, _ = dao.BilRecords.Ctx(ctx).Insert(do.BilRecords{
TenantId: tenantID,
UserId: userID,
ApiKeyId: apiKeyID,
ChannelId: channelID,
ModelName: modelName,
RequestId: requestID,
RelayMode: "task",
InputTokens: 0,
OutputTokens: totalTokens,
InputPrice: 0,
OutputPrice: effectiveOutputPrice,
TotalCost: actualCost,
Currency: "USD",
Status: "settled",
SettledAt: gtime.NewFromTime(now),
BillingMode: billingMode,
DiscountRatio: func() float64 {
if discountRatio > 0 {
return discountRatio
}
return 0
}(),
EffectiveInputPrice: 0,
EffectiveOutputPrice: effectiveOutputPrice,
BillingInputMultiplier: 0,
BillingOutputMultiplier: 0,
CacheCreationTokens: 0,
CacheReadTokens: 0,
CacheCreationCost: 0,
CacheReadCost: 0,
})
// 3. 事务内执行结算（钱包扣款 + 计费记录）
var balanceAfter, frozenAfter float64
var taskBillingID int64
err = g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
// 3a. 更新钱包
now := time.Now()
_, err := tx.Ctx(ctx).Exec(
"UPDATE bil_wallets SET frozen_balance = GREATEST(frozen_balance - ?, 0), balance = balance - ?, updated_at = ? WHERE id = ?",
preDeductAmount, actualCost, now, wallet.ID)
if err != nil {
return fmt.Errorf("settle task: update wallet: %w", err)
}
// 4. 生成计费快照 + 摘要
// 3b. 事务内读取准确余额
type balRow struct {
Balance float64 `json:"balance"`
FrozenBalance float64 `json:"frozen_balance"`
}
var br *balRow
err = tx.Model("bil_wallets").Ctx(ctx).
Where("id", wallet.ID).
Fields("balance, frozen_balance").
Scan(&br)
if err == nil && br != nil {
balanceAfter = br.Balance
frozenAfter = br.FrozenBalance
}
// 3c. 创建计费记录
var billingResult sql.Result
billingResult, err = tx.Model("bil_records").Ctx(ctx).Data(do.BilRecords{
TenantId: tenantID,
UserId: userID,
ApiKeyId: apiKeyID,
ChannelId: channelID,
ModelName: modelName,
RequestId: requestID,
RelayMode: "task",
InputTokens: 0,
OutputTokens: totalTokens,
InputPrice: 0,
OutputPrice: effectiveOutputPrice,
TotalCost: actualCost,
Currency: "USD",
Status: "settled",
SettledAt: gtime.NewFromTime(now),
BillingMode: billingMode,
DiscountRatio: func() float64 {
if discountRatio > 0 {
return discountRatio
}
return 0
}(),
EffectiveInputPrice: 0,
EffectiveOutputPrice: effectiveOutputPrice,
BillingInputMultiplier: 0,
BillingOutputMultiplier: 0,
CacheCreationTokens: 0,
CacheReadTokens: 0,
CacheCreationCost: 0,
CacheReadCost: 0,
}).Insert()
if err != nil {
return fmt.Errorf("settle task: create billing record: %w", err)
}
if billingResult != nil {
taskBillingID, _ = billingResult.LastInsertId()
}
return nil
})
if err != nil {
return nil, err
}
// 4. 清除缓存（事务提交后）
walletCache.Delete(ctx, fmt.Sprintf("%d", tenantID))
InvalidateWalletRedis(ctx, tenantID)
CleanupPreDeduct(ctx, tenantID, requestID)
CleanupPreDeduct(ctx, tenantID, requestID+"_adjust")
// 5. 差额处理：实际 < 预扣时退还差额（实际 > 预扣时已在步骤 3 扣完，无需额外操作）
if diff < -0.001 {
if err := SettleFailed(ctx, tenantID, requestID+"_adjust", -diff); err != nil {
g.Log().Warningf(ctx, "settle task adjust refund failed for %s: %v", requestID, err)
}
}
// 6. 生成计费快照 + 摘要
result := &common.SettlementResult{
PreDeductAmount: preDeductAmount,
ActualCost: actualCost,
@@ -178,17 +220,10 @@ func (b *TaskBillingProviderImpl) SettleTaskSuccess(ctx context.Context, tenantI
result.RateMultiplier = pricing.DiscountRatio
}
// 5. 差额处理：实际 < 预扣时退还差额（实际 > 预扣时已在步骤 1 扣完，无需额外操作）
if diff < -0.001 {
if err := SettleFailed(ctx, tenantID, requestID+"_adjust", -diff); err != nil {
return result, fmt.Errorf("settle task adjust refund: %w", err)
}
}
// 6. 记录消费流水（一条汇总）
// 7. 记录消费流水（事务外，best-effort）
recordTransaction(ctx, wallet.ID, tenantID, "consume", -actualCost,
fmt.Sprintf("consume: %s model=%s pre_deduct=%.4f actual=%.4f", requestID, modelName, preDeductAmount, actualCost),
userID, requestID, modelName)
userID, requestID, modelName, taskBillingID, "billing_record", 0, apiKeyID, taskID, balanceAfter, frozenAfter)
return result, nil
}
internal/logic/billing/wallet.go
+7 -17
View File
@@ -317,23 +317,8 @@ func GetPreDeductAmount(ctx context.Context, requestID string) (float64, bool) {
}
// recordTransaction 记录钱包流水
func recordTransaction(ctx context.Context, walletID, tenantID int64, txnType string, amount float64, description string, userID int64, requestID string, modelName string) {
type walletRow struct {
Balance float64 `json:"balance"`
FrozenBalance float64 `json:"frozen_balance"`
}
var w *walletRow
dao.BilWallets.Ctx(ctx).
Where("id", walletID).
Fields("balance, frozen_balance").
Scan(&w)
var balanceAfter, frozenAfter float64
if w != nil {
balanceAfter = w.Balance
frozenAfter = w.FrozenBalance
}
// balanceAfter/frozenAfter 由调用方在事务内读取后传入，确保并发场景下流水记录准确
func recordTransaction(ctx context.Context, walletID, tenantID int64, txnType string, amount float64, description string, userID int64, requestID string, modelName string, relatedID int64, relatedType string, projectID int64, apiKeyID int64, taskID string, balanceAfter, frozenAfter float64) {
_, err := dao.BilTransactions.Ctx(ctx).Insert(do.BilTransactions{
TenantId: tenantID,
WalletId: walletID,
@@ -341,10 +326,15 @@ func recordTransaction(ctx context.Context, walletID, tenantID int64, txnType st
Amount: amount,
BalanceAfter: balanceAfter,
FrozenAfter: frozenAfter,
RelatedId: relatedID,
RelatedType: relatedType,
Description: description,
UserId: userID,
RequestId: requestID,
ModelName: modelName,
ProjectId: projectID,
ApiKeyId: apiKeyID,
TaskId: taskID,
})
if err != nil {
g.Log().Errorf(ctx, "record transaction failed: %v", err)
internal/logic/task/async_polling.go
+2 -2
View File
@@ -156,7 +156,7 @@ func handleUnsettledTasks(ctx context.Context) {
taskBilling := billing.NewTaskBillingProvider()
_, err := taskBilling.SettleTaskSuccess(ctx, t.TenantID, t.UserID, t.ApiKeyID, t.ChannelID,
t.ModelName, t.RequestID, actualCost, t.PreDeductAmount,
0, 0, pd.BillingContext.Ratios)
0, 0, pd.BillingContext.Ratios, t.PublicTaskID)
if err != nil {
g.Log().Warningf(ctx, "poll: retry settle task %s: %v", t.PublicTaskID, err)
} else {
@@ -318,7 +318,7 @@ func pollSingleTask(ctx context.Context, adaptor common.TaskAdaptor, channel *co
}
task.ActualCost = actualCost
settleResult, err = taskBilling.SettleTaskSuccess(ctx, task.TenantID, task.UserID, task.ApiKeyID, task.ChannelID, task.ModelName, task.RequestID, actualCost, task.PreDeductAmount, taskInfo.TotalTokens, taskInfo.CompletionTokens, pd.BillingContext.Ratios)
settleResult, err = taskBilling.SettleTaskSuccess(ctx, task.TenantID, task.UserID, task.ApiKeyID, task.ChannelID, task.ModelName, task.RequestID, actualCost, task.PreDeductAmount, taskInfo.TotalTokens, taskInfo.CompletionTokens, pd.BillingContext.Ratios, task.PublicTaskID)
if err != nil {
g.Log().Warningf(ctx, "poll: settle task %s: %v", task.PublicTaskID, err)
} else {
internal/logic/tenant/billing.go
+28 -6
View File
@@ -81,7 +81,7 @@ func (s *sTenant) WalletTransactions(ctx context.Context, req *v1.TenantWalletTr
page, pageSize := common.NormalizePagination(req.Page, req.PageSize)
query := dao.BilTransactions.Ctx(ctx).
Where("tenant_id", tenantID)
Where("bil_transactions.tenant_id", tenantID)
type transactionRow struct {
Id int64 `json:"id"`
@@ -89,19 +89,23 @@ func (s *sTenant) WalletTransactions(ctx context.Context, req *v1.TenantWalletTr
Amount float64 `json:"amount"`
BalanceAfter float64 `json:"balance_after"`
FrozenAfter float64 `json:"frozen_after"`
RelatedId int64 `json:"related_id"`
RelatedType string `json:"related_type"`
Description string `json:"description"`
UserId int64 `json:"user_id"`
Username string `json:"username"`
RequestId string `json:"request_id"`
ModelName string `json:"model_name"`
ProjectId int64 `json:"project_id"`
ApiKeyId int64 `json:"api_key_id"`
TaskId string `json:"task_id"`
CreatedAt *gtime.Time `json:"created_at"`
}
var records []transactionRow
var err error
var total int
err = query.Fields("bil_transactions.id, bil_transactions.type, bil_transactions.amount, bil_transactions.balance_after, bil_transactions.frozen_after, bil_transactions.related_type, bil_transactions.description, bil_transactions.user_id, COALESCE(tu.username, '') AS username, bil_transactions.request_id, bil_transactions.model_name, bil_transactions.created_at").
err = query.Fields("bil_transactions.id, bil_transactions.type, bil_transactions.amount, bil_transactions.balance_after, bil_transactions.frozen_after, bil_transactions.related_id, bil_transactions.related_type, bil_transactions.description, bil_transactions.user_id, COALESCE(tu.username, '') AS username, bil_transactions.request_id, bil_transactions.model_name, bil_transactions.project_id, bil_transactions.api_key_id, bil_transactions.task_id, bil_transactions.created_at").
LeftJoin("tnt_users tu", "bil_transactions.user_id = tu.id AND bil_transactions.tenant_id = tu.tenant_id").
OrderDesc("bil_transactions.created_at").
Page(page, pageSize).
@@ -121,12 +125,16 @@ func (s *sTenant) WalletTransactions(ctx context.Context, req *v1.TenantWalletTr
"amount": r.Amount,
"balance_after": r.BalanceAfter,
"frozen_after": r.FrozenAfter,
"related_id": r.RelatedId,
"related_type": r.RelatedType,
"description": r.Description,
"user_id": r.UserId,
"username": r.Username,
"request_id": r.RequestId,
"model_name": r.ModelName,
"project_id": r.ProjectId,
"api_key_id": r.ApiKeyId,
"task_id": r.TaskId,
"created_at": r.CreatedAt,
})
}
@@ -147,11 +155,11 @@ func (s *sTenant) BillingRecords(ctx context.Context, req *v1.TenantBillingRecor
page, pageSize := common.NormalizePagination(req.Page, req.PageSize)
query := dao.BilRecords.Ctx(ctx).
Where("tenant_id", tenantID)
Where("bil_records.tenant_id", tenantID)
// member 角色只能查看自己的账单记录
if role == "member" {
query = query.Where("user_id", userID)
query = query.Where("bil_records.user_id", userID)
}
type billingRecordRow struct {
@@ -167,13 +175,22 @@ func (s *sTenant) BillingRecords(ctx context.Context, req *v1.TenantBillingRecor
Status string `json:"status"`
SettledAt *gtime.Time `json:"settled_at"`
CreatedAt *gtime.Time `json:"created_at"`
RequestId string `json:"request_id"`
UserId int64 `json:"user_id"`
Username string `json:"username"`
ProjectId int64 `json:"project_id"`
ProjectName string `json:"project_name"`
ApiKeyId int64 `json:"api_key_id"`
ApiKeyName string `json:"api_key_name"`
}
var records []billingRecordRow
var err error
var total int
err = query.Fields("id, model_name, relay_mode, input_tokens, output_tokens, input_price, output_price, total_cost, currency, status, settled_at, created_at").
OrderDesc("created_at").
err = query.Fields("bil_records.id, bil_records.model_name, bil_records.relay_mode, bil_records.input_tokens, bil_records.output_tokens, bil_records.input_price, bil_records.output_price, bil_records.total_cost, bil_records.currency, bil_records.status, bil_records.settled_at, bil_records.created_at, bil_records.request_id, bil_records.user_id, COALESCE(tu.username, '') AS username, bil_records.api_key_id, COALESCE(ak.name, '') AS api_key_name").
LeftJoin("tnt_users tu", "bil_records.user_id = tu.id AND bil_records.tenant_id = tu.tenant_id").
LeftJoin("api_keys ak", "bil_records.api_key_id = ak.id AND bil_records.tenant_id = ak.tenant_id").
OrderDesc("bil_records.created_at").
Page(page, pageSize).
ScanAndCount(&records, &total, false)
if err != nil {
@@ -198,6 +215,11 @@ func (s *sTenant) BillingRecords(ctx context.Context, req *v1.TenantBillingRecor
"status": r.Status,
"settled_at": r.SettledAt,
"created_at": r.CreatedAt,
"request_id": r.RequestId,
"user_id": r.UserId,
"username": r.Username,
"api_key_id": r.ApiKeyId,
"api_key_name": r.ApiKeyName,
})
}
internal/model/do/bil_transactions.go
+3
View File
@@ -27,4 +27,7 @@ type BilTransactions struct {
UserId any // 关联用户ID（consume 类型为实际消费用户，recharge 类型为操作用户，adjust 类型为空）
RequestId any // 关联请求ID（consume 类型对应 API 调用的 request_id，其他类型为空）
ModelName any // 关联模型名（consume 类型为调用的模型名，其他类型为空）
ProjectId any // 关联项目ID（consume 类型为 API Key 所属项目，个人密钥为空）
ApiKeyId any // 关联API密钥ID（consume 类型为发起请求的密钥）
TaskId any // 关联异步任务公开ID（consume+relay_mode=task 时关联 tsk_model_tasks.public_task_id）
}
internal/model/entity/bil_transactions.go
+3
View File
@@ -25,4 +25,7 @@ type BilTransactions struct {
UserId int64 `json:"user_id" orm:"user_id" description:"关联用户ID（consume 类型为实际消费用户，recharge 类型为操作用户，adjust 类型为空）"` // 关联用户ID（consume 类型为实际消费用户，recharge 类型为操作用户，adjust 类型为空）
RequestId string `json:"request_id" orm:"request_id" description:"关联请求ID（consume 类型对应 API 调用的 request_id，其他类型为空）"` // 关联请求ID（consume 类型对应 API 调用的 request_id，其他类型为空）
ModelName string `json:"model_name" orm:"model_name" description:"关联模型名（consume 类型为调用的模型名，其他类型为空）"` // 关联模型名（consume 类型为调用的模型名，其他类型为空）
ProjectId int64 `json:"project_id" orm:"project_id" description:"关联项目ID（consume 类型为 API Key 所属项目，个人密钥为空）"` // 关联项目ID（consume 类型为 API Key 所属项目，个人密钥为空）
ApiKeyId int64 `json:"api_key_id" orm:"api_key_id" description:"关联API密钥ID（consume 类型为发起请求的密钥）"` // 关联API密钥ID（consume 类型为发起请求的密钥）
TaskId string `json:"task_id" orm:"task_id" description:"关联异步任务公开ID（consume+relay_mode=task 时关联 tsk_model_tasks.public_task_id）"` // 关联异步任务公开ID（consume+relay_mode=task 时关联 tsk_model_tasks.public_task_id）
}
migrations/000007_add_transaction_project_task_fields.sql
+22
View File
@@ -0,0 +1,22 @@
-- +goose Up
-- bil_transactions 增加项目和任务关联字段，支持按项目、密钥、任务维度追溯费用
ALTER TABLE bil_transactions ADD COLUMN IF NOT EXISTS project_id BIGINT;
ALTER TABLE bil_transactions ADD COLUMN IF NOT EXISTS api_key_id BIGINT;
ALTER TABLE bil_transactions ADD COLUMN IF NOT EXISTS task_id VARCHAR(64);
COMMENT ON COLUMN bil_transactions.project_id IS '关联项目ID（consume 类型为 API Key 所属项目，个人密钥为空）';
COMMENT ON COLUMN bil_transactions.api_key_id IS '关联API密钥ID（consume 类型为发起请求的密钥）';
COMMENT ON COLUMN bil_transactions.task_id IS '关联异步任务公开ID（consume+relay_mode=task 时关联 tsk_model_tasks.public_task_id）';
-- 按项目查询消费记录
CREATE INDEX IF NOT EXISTS idx_bil_transactions_project_created ON bil_transactions (tenant_id, project_id, created_at DESC) WHERE project_id IS NOT NULL;
-- 按 API Key 查询消费记录
CREATE INDEX IF NOT EXISTS idx_bil_transactions_apikey_created ON bil_transactions (tenant_id, api_key_id, created_at DESC) WHERE api_key_id IS NOT NULL;
-- +goose Down
DROP INDEX IF EXISTS idx_bil_transactions_apikey_created;
DROP INDEX IF EXISTS idx_bil_transactions_project_created;
ALTER TABLE bil_transactions DROP COLUMN IF EXISTS task_id;
ALTER TABLE bil_transactions DROP COLUMN IF EXISTS api_key_id;
ALTER TABLE bil_transactions DROP COLUMN IF EXISTS project_id;
relay/common/billing.go
+2 -2
View File
@@ -10,7 +10,7 @@ type BillingProvider interface {
PreDeduct(ctx context.Context, tenantID int64, modelName string, inputTokens, maxTokens int, isStream bool, requestID string) (preDeductAmount float64, err error)
// Settle 结算费用（成功请求）
Settle(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID, relayMode string, usage *Usage, preDeductAmount float64) error
Settle(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID, relayMode string, usage *Usage, preDeductAmount float64, projectID int64) error
// SettleWithUsage 完整 Usage 结算（含 cache token + 计费快照）
SettleWithUsage(ctx context.Context, tenantID, userID, apiKeyID, channelID int64,
@@ -21,7 +21,7 @@ type BillingProvider interface {
SettleFailed(ctx context.Context, tenantID int64, requestID string, preDeductAmount float64) error
// SettleStreamInterrupted 流式中断结算
SettleStreamInterrupted(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID, relayMode string, usage *Usage, preDeductAmount float64) error
SettleStreamInterrupted(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID, relayMode string, usage *Usage, preDeductAmount float64, projectID int64) error
// CheckRateLimit QPS 限流检查
CheckRateLimit(ctx context.Context, tenantID, userID, apiKeyID int64) (allowed bool, limitLevel string, remaining int, resetAt int64)
relay/common/task_billing.go
+1 -1
View File
@@ -12,7 +12,7 @@ type TaskBillingProvider interface {
PreDeductTask(ctx context.Context, tenantID int64, requestID string, estimatedCost float64, modelName string) (float64, error)
// SettleTaskSuccess 任务成功结算（含计费快照）
SettleTaskSuccess(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID string, actualCost, preDeductAmount float64, totalTokens, completionTokens int, ratios map[string]float64) (*SettlementResult, error)
SettleTaskSuccess(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, modelName, requestID string, actualCost, preDeductAmount float64, totalTokens, completionTokens int, ratios map[string]float64, taskID string) (*SettlementResult, error)
// SettleTaskFailed 任务失败退还预扣
SettleTaskFailed(ctx context.Context, tenantID int64, requestID string, preDeductAmount float64) error
relay/handler/realtime_handler.go
+2 -2
View File
@@ -152,10 +152,10 @@ func HandleRealtime(w http.ResponseWriter, r *http.Request, rc *RealtimeContext,
streamUsage = &common.Usage{}
}
_ = billing.SettleStreamInterrupted(ctx, rc.TenantID, rc.UserID, rc.ApiKeyID, selection.ChannelID,
modelName, rc.RequestID, "realtime", streamUsage, preDeductAmount)
modelName, rc.RequestID, "realtime", streamUsage, preDeductAmount, rc.ProjectID)
} else if usage != nil {
_ = billing.Settle(ctx, rc.TenantID, rc.UserID, rc.ApiKeyID, selection.ChannelID,
modelName, rc.RequestID, "realtime", usage, preDeductAmount)
modelName, rc.RequestID, "realtime", usage, preDeductAmount, rc.ProjectID)
}
}
relay/handler/relay_handler.go
+1 -1
View File
@@ -427,7 +427,7 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He
}
if billing != nil && preDeductAmount > 0 {
_ = billing.SettleStreamInterrupted(settleCtx, rc.TenantID, rc.UserID, rc.ApiKeyID, selection.ChannelID,
modelName, rc.RequestID, relayModeStr, streamUsage, preDeductAmount)
modelName, rc.RequestID, relayModeStr, streamUsage, preDeductAmount, rc.ProjectID)
}
recordFailedUsage(provider, rc, selection.ChannelID, modelName, relayMode, isStream, err)
hop.Success = false