From 50e5377ea5feec326c416450e4e8bcc0bdfe7749 Mon Sep 17 00:00:00 2001 From: CaIon Date: Mon, 10 Aug 2026 22:12:20 +0800 Subject: [PATCH] fix(topup): settle recharge orders atomically --- common/quota_math.go | 6 ++ controller/topup.go | 71 ++++++-------- model/payment_method_guard_test.go | 138 ++++++++++++++++++++++++++- model/quota_reserve.go | 68 ++++++++++++++ model/redemption.go | 1 + model/topup.go | 146 +++++++++++++++++++++-------- model/user_cache.go | 19 +++- 7 files changed, 367 insertions(+), 82 deletions(-) create mode 100644 model/quota_reserve.go diff --git a/common/quota_math.go b/common/quota_math.go index 86c72492..66d62093 100644 --- a/common/quota_math.go +++ b/common/quota_math.go @@ -146,3 +146,9 @@ func QuotaFromDecimalChecked(d decimal.Decimal) (int, *QuotaClamp) { f, _ := d.Round(0).Float64() return saturateQuota(f, "QuotaFromDecimal") } + +// QuotaFromDecimalStrict converts an in-range decimal quota and rejects a +// value that would otherwise be saturated at the database's int32 boundary. +func QuotaFromDecimalStrict(d decimal.Decimal) (int, error) { + return strictQuota(QuotaFromDecimalChecked(d)) +} diff --git a/controller/topup.go b/controller/topup.go index 390f53f7..e60312c7 100644 --- a/controller/topup.go +++ b/controller/topup.go @@ -1,6 +1,7 @@ package controller import ( + "errors" "fmt" "net/http" "net/url" @@ -181,7 +182,7 @@ func getMinTopup() int64 { if operation_setting.GetQuotaDisplayType() == operation_setting.QuotaDisplayTypeTokens { dMinTopup := decimal.NewFromInt(int64(minTopup)) dQuotaPerUnit := decimal.NewFromFloat(common.QuotaPerUnit) - minTopup = int(dMinTopup.Mul(dQuotaPerUnit).IntPart()) + minTopup = common.QuotaFromDecimal(dMinTopup.Mul(dQuotaPerUnit)) } return int64(minTopup) } @@ -351,16 +352,9 @@ func EpayNotify(c *gin.Context) { return } verifyInfo, err := client.Verify(params) - if err == nil && verifyInfo.VerifyStatus { - logger.LogInfo(c.Request.Context(), fmt.Sprintf("易支付 webhook 验签成功 trade_no=%s callback_type=%s trade_status=%s client_ip=%s verify_info=%q", verifyInfo.ServiceTradeNo, verifyInfo.Type, verifyInfo.TradeStatus, c.ClientIP(), common.GetJsonString(verifyInfo))) - _, err := c.Writer.Write([]byte("success")) - if err != nil { - logger.LogError(c.Request.Context(), fmt.Sprintf("易支付 webhook 响应写入失败 trade_no=%s client_ip=%s error=%q", verifyInfo.ServiceTradeNo, c.ClientIP(), err.Error())) - } - } else { - _, err := c.Writer.Write([]byte("fail")) - if err != nil { - logger.LogError(c.Request.Context(), fmt.Sprintf("易支付 webhook 响应写入失败 path=%q client_ip=%s error=%q", c.Request.RequestURI, c.ClientIP(), err.Error())) + if err != nil || !verifyInfo.VerifyStatus { + if _, writeErr := c.Writer.Write([]byte("fail")); writeErr != nil { + logger.LogError(c.Request.Context(), fmt.Sprintf("易支付 webhook 响应写入失败 path=%q client_ip=%s error=%q", c.Request.RequestURI, c.ClientIP(), writeErr.Error())) } if err != nil { logger.LogWarn(c.Request.Context(), fmt.Sprintf("易支付 webhook 验签失败 path=%q client_ip=%s verify_error=%q", c.Request.RequestURI, c.ClientIP(), err.Error())) @@ -369,46 +363,41 @@ func EpayNotify(c *gin.Context) { } return } + logger.LogInfo(c.Request.Context(), fmt.Sprintf("易支付 webhook 验签成功 trade_no=%s callback_type=%s trade_status=%s client_ip=%s verify_info=%q", verifyInfo.ServiceTradeNo, verifyInfo.Type, verifyInfo.TradeStatus, c.ClientIP(), common.GetJsonString(verifyInfo))) if verifyInfo.TradeStatus == epay.StatusTradeSuccess { + // 进程内锁只是优化；重复/并发回调的正确性由 RechargeEpay 的 + // 数据库行锁 + 事务内状态校验保证（多实例部署下同样安全）。 LockOrder(verifyInfo.ServiceTradeNo) defer UnlockOrder(verifyInfo.ServiceTradeNo) - topUp := model.GetTopUpByTradeNo(verifyInfo.ServiceTradeNo) - if topUp == nil { - logger.LogWarn(c.Request.Context(), fmt.Sprintf("易支付 回调订单不存在 trade_no=%s callback_type=%s client_ip=%s verify_info=%q", verifyInfo.ServiceTradeNo, verifyInfo.Type, c.ClientIP(), common.GetJsonString(verifyInfo))) + alreadyDone, err := model.RechargeEpay(verifyInfo.ServiceTradeNo, verifyInfo.Type, c.ClientIP()) + if err != nil { + switch { + case errors.Is(err, model.ErrTopUpNotFound): + logger.LogWarn(c.Request.Context(), fmt.Sprintf("易支付 回调订单不存在 trade_no=%s callback_type=%s client_ip=%s verify_info=%q", verifyInfo.ServiceTradeNo, verifyInfo.Type, c.ClientIP(), common.GetJsonString(verifyInfo))) + case errors.Is(err, model.ErrPaymentMethodMismatch): + logger.LogWarn(c.Request.Context(), fmt.Sprintf("易支付 订单支付网关不匹配 trade_no=%s callback_type=%s client_ip=%s", verifyInfo.ServiceTradeNo, verifyInfo.Type, c.ClientIP())) + case errors.Is(err, model.ErrTopUpStatusInvalid): + logger.LogWarn(c.Request.Context(), fmt.Sprintf("易支付 订单状态非法 trade_no=%s callback_type=%s client_ip=%s", verifyInfo.ServiceTradeNo, verifyInfo.Type, c.ClientIP())) + default: + logger.LogError(c.Request.Context(), fmt.Sprintf("易支付 充值处理失败 trade_no=%s client_ip=%s error=%q", verifyInfo.ServiceTradeNo, c.ClientIP(), err.Error())) + } + if _, writeErr := c.Writer.Write([]byte("fail")); writeErr != nil { + logger.LogError(c.Request.Context(), fmt.Sprintf("易支付 webhook 响应写入失败 trade_no=%s client_ip=%s error=%q", verifyInfo.ServiceTradeNo, c.ClientIP(), writeErr.Error())) + } return } - if topUp.PaymentProvider != model.PaymentProviderEpay { - logger.LogWarn(c.Request.Context(), fmt.Sprintf("易支付 订单支付网关不匹配 trade_no=%s order_provider=%s callback_type=%s client_ip=%s", verifyInfo.ServiceTradeNo, topUp.PaymentProvider, verifyInfo.Type, c.ClientIP())) - return - } - if topUp.Status == common.TopUpStatusPending { - if topUp.PaymentMethod != verifyInfo.Type { - logger.LogInfo(c.Request.Context(), fmt.Sprintf("易支付 实际支付方式与订单不同 trade_no=%s order_payment_method=%s actual_type=%s client_ip=%s", verifyInfo.ServiceTradeNo, topUp.PaymentMethod, verifyInfo.Type, c.ClientIP())) - topUp.PaymentMethod = verifyInfo.Type - } - topUp.Status = common.TopUpStatusSuccess - err := topUp.Update() - if err != nil { - logger.LogError(c.Request.Context(), fmt.Sprintf("易支付 更新充值订单失败 trade_no=%s user_id=%d client_ip=%s error=%q topup=%q", topUp.TradeNo, topUp.UserId, c.ClientIP(), err.Error(), common.GetJsonString(topUp))) - return - } - //user, _ := model.GetUserById(topUp.UserId, false) - //user.Quota += topUp.Amount * 500000 - dAmount := decimal.NewFromInt(int64(topUp.Amount)) - dQuotaPerUnit := decimal.NewFromFloat(common.QuotaPerUnit) - quotaToAdd := int(dAmount.Mul(dQuotaPerUnit).IntPart()) - err = model.IncreaseUserQuota(topUp.UserId, quotaToAdd, true) - if err != nil { - logger.LogError(c.Request.Context(), fmt.Sprintf("易支付 更新用户额度失败 trade_no=%s user_id=%d client_ip=%s quota_to_add=%d error=%q topup=%q", topUp.TradeNo, topUp.UserId, c.ClientIP(), quotaToAdd, err.Error(), common.GetJsonString(topUp))) - return - } - logger.LogInfo(c.Request.Context(), fmt.Sprintf("易支付 充值成功 trade_no=%s user_id=%d client_ip=%s quota_to_add=%d money=%.2f topup=%q", topUp.TradeNo, topUp.UserId, c.ClientIP(), quotaToAdd, topUp.Money, common.GetJsonString(topUp))) - model.RecordTopupLog(topUp.UserId, fmt.Sprintf("使用在线充值成功，充值金额: %v，支付金额：%f", logger.LogQuota(quotaToAdd), topUp.Money), c.ClientIP(), topUp.PaymentMethod, "epay") + if alreadyDone { + logger.LogInfo(c.Request.Context(), fmt.Sprintf("易支付 重复回调幂等忽略 trade_no=%s callback_type=%s client_ip=%s", verifyInfo.ServiceTradeNo, verifyInfo.Type, c.ClientIP())) + } else { + logger.LogInfo(c.Request.Context(), fmt.Sprintf("易支付 充值成功 trade_no=%s callback_type=%s client_ip=%s", verifyInfo.ServiceTradeNo, verifyInfo.Type, c.ClientIP())) } } else { logger.LogInfo(c.Request.Context(), fmt.Sprintf("易支付 webhook 忽略事件 trade_no=%s callback_type=%s trade_status=%s client_ip=%s verify_info=%q", verifyInfo.ServiceTradeNo, verifyInfo.Type, verifyInfo.TradeStatus, c.ClientIP(), common.GetJsonString(verifyInfo))) } + if _, writeErr := c.Writer.Write([]byte("success")); writeErr != nil { + logger.LogError(c.Request.Context(), fmt.Sprintf("易支付 webhook 响应写入失败 trade_no=%s client_ip=%s error=%q", verifyInfo.ServiceTradeNo, c.ClientIP(), writeErr.Error())) + } } func RequestAmount(c *gin.Context) { diff --git a/model/payment_method_guard_test.go b/model/payment_method_guard_test.go index 7f4f15cc..b400ad50 100644 --- a/model/payment_method_guard_test.go +++ b/model/payment_method_guard_test.go @@ -9,7 +9,7 @@ import ( "github.com/stretchr/testify/require" ) -func insertUserForPaymentGuardTest(t *testing.T, id int, quota int) { +func insertUserForPaymentGuardTest(t *testing.T, id int, quota int) *User { t.Helper() user := &User{ Id: id, @@ -18,6 +18,7 @@ func insertUserForPaymentGuardTest(t *testing.T, id int, quota int) { Quota: quota, } require.NoError(t, DB.Create(user).Error) + return user } func insertSubscriptionPlanForPaymentGuardTest(t *testing.T, id int) *SubscriptionPlan { @@ -172,3 +173,138 @@ func TestExpireSubscriptionOrder_RejectsMismatchedPaymentProvider(t *testing.T) require.NotNil(t, order) assert.Equal(t, common.TopUpStatusPending, order.Status) } + +func createEpayTestOrder(t *testing.T, userId int, tradeNo string, provider string, status string) TopUp { + t.Helper() + topUp := TopUp{ + UserId: userId, + Amount: 2, + Money: 10.0, + TradeNo: tradeNo, + PaymentMethod: "alipay", + PaymentProvider: provider, + CreateTime: common.GetTimestamp(), + Status: status, + } + require.NoError(t, DB.Create(&topUp).Error) + return topUp +} + +func TestRechargeEpayCreditsQuotaExactlyOnce(t *testing.T) { + truncateTables(t) + + oldQuotaPerUnit := common.QuotaPerUnit + common.QuotaPerUnit = 500000 + t.Cleanup(func() { common.QuotaPerUnit = oldQuotaPerUnit }) + + user := insertUserForPaymentGuardTest(t, 501, 0) + order := createEpayTestOrder(t, user.Id, "EPAYTESTONCE", PaymentProviderEpay, common.TopUpStatusPending) + + alreadyDone, err := RechargeEpay(order.TradeNo, "alipay", "127.0.0.1") + require.NoError(t, err) + assert.False(t, alreadyDone) + assert.Equal(t, 2*500000, getUserQuotaForPaymentGuardTest(t, user.Id)) + + reloaded := GetTopUpByTradeNo(order.TradeNo) + require.NotNil(t, reloaded) + assert.Equal(t, common.TopUpStatusSuccess, reloaded.Status) + assert.NotZero(t, reloaded.CompleteTime) + + alreadyDone, err = RechargeEpay(order.TradeNo, "alipay", "127.0.0.1") + require.NoError(t, err) + assert.True(t, alreadyDone) + assert.Equal(t, 2*500000, getUserQuotaForPaymentGuardTest(t, user.Id)) +} + +func TestRechargeEpayKeepsRedisAndDatabaseCreditInSync(t *testing.T) { + truncateTables(t) + useUserCacheMiniRedis(t) + + oldQuotaPerUnit := common.QuotaPerUnit + common.QuotaPerUnit = 5 + t.Cleanup(func() { common.QuotaPerUnit = oldQuotaPerUnit }) + + user := insertUserForPaymentGuardTest(t, 502, 7) + require.NoError(t, populateUserCache(*user)) + order := createEpayTestOrder(t, user.Id, "EPAYTESTREDISSYNC", PaymentProviderEpay, common.TopUpStatusPending) + + alreadyDone, err := RechargeEpay(order.TradeNo, "alipay", "127.0.0.1") + require.NoError(t, err) + assert.False(t, alreadyDone) + assert.Equal(t, 17, getUserQuotaForPaymentGuardTest(t, user.Id)) + cached, err := cacheGetUserBase(user.Id) + require.NoError(t, err) + assert.Equal(t, 17, cached.Quota) + + alreadyDone, err = RechargeEpay(order.TradeNo, "alipay", "127.0.0.1") + require.NoError(t, err) + assert.True(t, alreadyDone) + cached, err = cacheGetUserBase(user.Id) + require.NoError(t, err) + assert.Equal(t, 17, cached.Quota) +} + +func TestRechargeEpayUpdatesPaymentMethodToActual(t *testing.T) { + truncateTables(t) + + oldQuotaPerUnit := common.QuotaPerUnit + common.QuotaPerUnit = 500000 + t.Cleanup(func() { common.QuotaPerUnit = oldQuotaPerUnit }) + + user := insertUserForPaymentGuardTest(t, 503, 0) + order := createEpayTestOrder(t, user.Id, "EPAYTESTMETHOD", PaymentProviderEpay, common.TopUpStatusPending) + + alreadyDone, err := RechargeEpay(order.TradeNo, "wxpay", "127.0.0.1") + require.NoError(t, err) + assert.False(t, alreadyDone) + + reloaded := GetTopUpByTradeNo(order.TradeNo) + require.NotNil(t, reloaded) + assert.Equal(t, "wxpay", reloaded.PaymentMethod) + assert.Equal(t, 2*500000, getUserQuotaForPaymentGuardTest(t, user.Id)) +} + +func TestRechargeEpayRejectsForeignAndNonPendingOrders(t *testing.T) { + truncateTables(t) + + oldQuotaPerUnit := common.QuotaPerUnit + common.QuotaPerUnit = 500000 + t.Cleanup(func() { common.QuotaPerUnit = oldQuotaPerUnit }) + + user := insertUserForPaymentGuardTest(t, 504, 7) + + t.Run("order from another payment provider", func(t *testing.T) { + order := createEpayTestOrder(t, user.Id, "EPAYTESTSTRIPE", PaymentProviderStripe, common.TopUpStatusPending) + _, err := RechargeEpay(order.TradeNo, "alipay", "127.0.0.1") + assert.ErrorIs(t, err, ErrPaymentMethodMismatch) + assert.Equal(t, 7, getUserQuotaForPaymentGuardTest(t, user.Id)) + }) + + t.Run("order that is not pending", func(t *testing.T) { + order := createEpayTestOrder(t, user.Id, "EPAYTESTEXPIRED", PaymentProviderEpay, common.TopUpStatusExpired) + _, err := RechargeEpay(order.TradeNo, "alipay", "127.0.0.1") + assert.ErrorIs(t, err, ErrTopUpStatusInvalid) + assert.Equal(t, 7, getUserQuotaForPaymentGuardTest(t, user.Id)) + }) + + t.Run("missing order", func(t *testing.T) { + _, err := RechargeEpay("EPAYTESTMISSING", "alipay", "127.0.0.1") + assert.ErrorIs(t, err, ErrTopUpNotFound) + }) +} + +func TestRechargeEpayRejectsQuotaOverflowBeforeCompletingOrder(t *testing.T) { + truncateTables(t) + + oldQuotaPerUnit := common.QuotaPerUnit + common.QuotaPerUnit = float64(common.MaxQuota) + t.Cleanup(func() { common.QuotaPerUnit = oldQuotaPerUnit }) + + user := insertUserForPaymentGuardTest(t, 505, 3) + order := createEpayTestOrder(t, user.Id, "EPAYTESTOVERFLOW", PaymentProviderEpay, common.TopUpStatusPending) + + _, err := RechargeEpay(order.TradeNo, "alipay", "127.0.0.1") + require.Error(t, err) + assert.Equal(t, 3, getUserQuotaForPaymentGuardTest(t, user.Id)) + assert.Equal(t, common.TopUpStatusPending, getTopUpStatusForPaymentGuardTest(t, order.TradeNo)) +} diff --git a/model/quota_reserve.go b/model/quota_reserve.go new file mode 100644 index 00000000..352377d1 --- /dev/null +++ b/model/quota_reserve.go @@ -0,0 +1,68 @@ +package model + +import ( + "context" + + "github.com/QuantumNous/new-api/common" +) + +type cacheQuotaResult int + +const ( + cacheQuotaInsufficient cacheQuotaResult = iota + cacheQuotaOK + cacheQuotaMiss +) + +// 下列脚本都是守卫式的：只在完整哈希（Id 匹配且配额字段存在）上操作， +// 哈希缺失时返回 miss 而不是创建残缺哈希。脚本不修改 TTL（HINCRBY 天然保留 +// 水合时设置的 TTL），因此即使某个写库路径绕过了缓存，偏差也会在一个 TTL +// 窗口内随缓存过期而自愈。 + +const userQuotaReserveScript = ` +if tonumber(redis.call('HGET', KEYS[1], 'Id') or '0') ~= tonumber(ARGV[2]) + or tonumber(redis.call('HGET', KEYS[1], 'CacheSchema') or '0') ~= tonumber(ARGV[3]) + or redis.call('HEXISTS', KEYS[1], 'Quota') == 0 then + return -1 +end +local quota = tonumber(redis.call('HGET', KEYS[1], 'Quota')) +if quota == nil or quota < tonumber(ARGV[1]) then + return 0 +end +redis.call('HINCRBY', KEYS[1], 'Quota', -tonumber(ARGV[1])) +return 1` + +const userQuotaDeltaScript = ` +if tonumber(redis.call('HGET', KEYS[1], 'Id') or '0') ~= tonumber(ARGV[2]) + or tonumber(redis.call('HGET', KEYS[1], 'CacheSchema') or '0') ~= tonumber(ARGV[3]) + or redis.call('HEXISTS', KEYS[1], 'Quota') == 0 then + return -1 +end +redis.call('HINCRBY', KEYS[1], 'Quota', tonumber(ARGV[1])) +return 1` + +func quotaResultFromLua(result int, err error) (cacheQuotaResult, error) { + if err != nil { + return cacheQuotaMiss, err + } + switch result { + case 1: + return cacheQuotaOK, nil + case 0: + return cacheQuotaInsufficient, nil + default: + return cacheQuotaMiss, nil + } +} + +func cacheTryReserveUserQuota(userID int, amount int64) (cacheQuotaResult, error) { + result, err := common.RDB.Eval(context.Background(), userQuotaReserveScript, + []string{getUserCacheKey(userID)}, amount, userID, userCacheSchemaVersion).Int() + return quotaResultFromLua(result, err) +} + +func cacheApplyUserQuotaDelta(userID int, delta int64) (cacheQuotaResult, error) { + result, err := common.RDB.Eval(context.Background(), userQuotaDeltaScript, + []string{getUserCacheKey(userID)}, delta, userID, userCacheSchemaVersion).Int() + return quotaResultFromLua(result, err) +} diff --git a/model/redemption.go b/model/redemption.go index 23985ef4..a7751d90 100644 --- a/model/redemption.go +++ b/model/redemption.go @@ -181,6 +181,7 @@ func Redeem(key string, userId int) (quota int, err error) { common.SysError("redemption failed: " + err.Error()) return 0, ErrRedeemFailed } + syncCreditUserQuotaCache(userId, redemption.Quota, "redemption") RecordLog(userId, LogTypeTopup, fmt.Sprintf("通过兑换码充值 %s，兑换码ID %d", logger.LogQuota(redemption.Quota), redemption.Id)) return redemption.Quota, nil } diff --git a/model/topup.go b/model/topup.go index 92cb276b..821c77ce 100644 --- a/model/topup.go +++ b/model/topup.go @@ -106,12 +106,82 @@ func UpdatePendingTopUpStatus(tradeNo string, expectedPaymentProvider string, ta }) } +// RechargeEpay 原子完成易支付订单：订单行锁、状态校验、成功更新与用户额度增加 +// 在同一个事务内完成，因此同一订单的并发/重复回调（包括多实例部署下）最多充值一次。 +// alreadyDone=true 表示订单此前已完成，本次为幂等重复回调。 +// 进程内的 LockOrder 只是优化，正确性由本函数的数据库行锁保证。 +func RechargeEpay(tradeNo string, actualPaymentMethod string, callerIp string) (alreadyDone bool, err error) { + if tradeNo == "" { + return false, errors.New("未提供支付单号") + } + + refCol := "`trade_no`" + if common.UsingMainDatabase(common.DatabaseTypePostgreSQL) { + refCol = `"trade_no"` + } + + var quotaToAdd int + topUp := &TopUp{} + err = DB.Transaction(func(tx *gorm.DB) error { + if err := lockForUpdate(tx).Where(refCol+" = ?", tradeNo).First(topUp).Error; err != nil { + return ErrTopUpNotFound + } + if topUp.PaymentProvider != PaymentProviderEpay { + return ErrPaymentMethodMismatch + } + if topUp.Status == common.TopUpStatusSuccess { + alreadyDone = true + return nil + } + if topUp.Status != common.TopUpStatusPending { + return ErrTopUpStatusInvalid + } + if actualPaymentMethod != "" && topUp.PaymentMethod != actualPaymentMethod { + topUp.PaymentMethod = actualPaymentMethod + } + var quotaErr error + quotaToAdd, quotaErr = common.QuotaFromDecimalStrict( + decimal.NewFromInt(topUp.Amount).Mul(decimal.NewFromFloat(common.QuotaPerUnit)), + ) + if quotaErr != nil || quotaToAdd <= 0 { + return errors.New("无效的充值额度") + } + topUp.CompleteTime = common.GetTimestamp() + topUp.Status = common.TopUpStatusSuccess + if err := tx.Save(topUp).Error; err != nil { + return err + } + result := tx.Model(&User{}).Where("id = ?", topUp.UserId).Update("quota", gorm.Expr("quota + ?", quotaToAdd)) + if result.Error != nil { + return result.Error + } + if result.RowsAffected != 1 { + return gorm.ErrRecordNotFound + } + return nil + }) + if err != nil { + if !errors.Is(err, ErrTopUpNotFound) && !errors.Is(err, ErrPaymentMethodMismatch) && !errors.Is(err, ErrTopUpStatusInvalid) { + common.SysError("epay topup failed: " + err.Error()) + } + return false, err + } + if alreadyDone { + return true, nil + } + syncCreditUserQuotaCache(topUp.UserId, quotaToAdd, "epay topup") + + common.SysLog(fmt.Sprintf("易支付充值成功 trade_no=%s user_id=%d quota_to_add=%d money=%.2f", topUp.TradeNo, topUp.UserId, quotaToAdd, topUp.Money)) + RecordTopupLog(topUp.UserId, fmt.Sprintf("使用在线充值成功，充值金额: %v，支付金额：%f", logger.LogQuota(quotaToAdd), topUp.Money), callerIp, topUp.PaymentMethod, PaymentProviderEpay) + return false, nil +} + func Recharge(referenceId string, customerId string, callerIp string) (err error) { if referenceId == "" { return errors.New("未提供支付单号") } - var quota float64 + var quota int topUp := &TopUp{} refCol := "`trade_no`" @@ -140,21 +210,23 @@ func Recharge(referenceId string, customerId string, callerIp string) (err error return err } - quota = topUp.Money * common.QuotaPerUnit - err = tx.Model(&User{}).Where("id = ?", topUp.UserId).Updates(map[string]interface{}{"stripe_customer": customerId, "quota": gorm.Expr("quota + ?", quota)}).Error - if err != nil { - return err + quota, err = common.QuotaFromDecimalStrict( + decimal.NewFromFloat(topUp.Money).Mul(decimal.NewFromFloat(common.QuotaPerUnit)), + ) + if err != nil || quota <= 0 { + return errors.New("无效的充值额度") } - - return nil + return tx.Model(&User{}).Where("id = ?", topUp.UserId). + Updates(map[string]interface{}{"stripe_customer": customerId, "quota": gorm.Expr("quota + ?", quota)}).Error }) if err != nil { common.SysError("topup failed: " + err.Error()) return errors.New("充值失败，请稍后重试") } + syncCreditUserQuotaCache(topUp.UserId, quota, "stripe topup") - RecordTopupLog(topUp.UserId, fmt.Sprintf("使用在线充值成功，充值金额: %v，支付金额：%d", logger.FormatQuota(int(quota)), topUp.Amount), callerIp, topUp.PaymentMethod, PaymentMethodStripe) + RecordTopupLog(topUp.UserId, fmt.Sprintf("使用在线充值成功，充值金额: %v，支付金额：%d", logger.FormatQuota(quota), topUp.Amount), callerIp, topUp.PaymentMethod, PaymentMethodStripe) return nil } @@ -351,15 +423,17 @@ func ManualCompleteTopUp(tradeNo string, callerIp string) error { // 计算应充值额度： // - Stripe 订单：Money 代表经分组倍率换算后的美元数量，直接 * QuotaPerUnit // - 其他订单（如易支付）：Amount 为美元数量，* QuotaPerUnit + var quotaErr error if topUp.PaymentProvider == PaymentProviderStripe { - dQuotaPerUnit := decimal.NewFromFloat(common.QuotaPerUnit) - quotaToAdd = int(decimal.NewFromFloat(topUp.Money).Mul(dQuotaPerUnit).IntPart()) + quotaToAdd, quotaErr = common.QuotaFromDecimalStrict( + decimal.NewFromFloat(topUp.Money).Mul(decimal.NewFromFloat(common.QuotaPerUnit)), + ) } else { - dAmount := decimal.NewFromInt(topUp.Amount) - dQuotaPerUnit := decimal.NewFromFloat(common.QuotaPerUnit) - quotaToAdd = int(dAmount.Mul(dQuotaPerUnit).IntPart()) + quotaToAdd, quotaErr = common.QuotaFromDecimalStrict( + decimal.NewFromInt(topUp.Amount).Mul(decimal.NewFromFloat(common.QuotaPerUnit)), + ) } - if quotaToAdd <= 0 { + if quotaErr != nil || quotaToAdd <= 0 { return errors.New("无效的充值额度") } @@ -386,6 +460,7 @@ func ManualCompleteTopUp(tradeNo string, callerIp string) error { } // 事务外记录日志，避免阻塞 + syncCreditUserQuotaCache(userId, quotaToAdd, "manual topup") RecordTopupLog(userId, fmt.Sprintf("管理员补单成功，充值金额: %v，支付金额：%f", logger.FormatQuota(quotaToAdd), payMoney), callerIp, paymentMethod, "admin") return nil } @@ -394,7 +469,7 @@ func RechargeCreem(referenceId string, customerEmail string, customerName string return errors.New("未提供支付单号") } - var quota int64 + var quota int topUp := &TopUp{} refCol := "`trade_no`" @@ -424,7 +499,10 @@ func RechargeCreem(referenceId string, customerEmail string, customerName string } // Creem 直接使用 Amount 作为充值额度（整数） - quota = topUp.Amount + quota, err = common.QuotaFromDecimalStrict(decimal.NewFromInt(topUp.Amount)) + if err != nil || quota <= 0 { + return errors.New("无效的充值额度") + } // 构建更新字段，优先使用邮箱，如果邮箱为空则使用用户名 updateFields := map[string]interface{}{ @@ -446,18 +524,14 @@ func RechargeCreem(referenceId string, customerEmail string, customerName string } } - err = tx.Model(&User{}).Where("id = ?", topUp.UserId).Updates(updateFields).Error - if err != nil { - return err - } - - return nil + return tx.Model(&User{}).Where("id = ?", topUp.UserId).Updates(updateFields).Error }) if err != nil { common.SysError("creem topup failed: " + err.Error()) return errors.New("充值失败，请稍后重试") } + syncCreditUserQuotaCache(topUp.UserId, quota, "creem topup") RecordTopupLog(topUp.UserId, fmt.Sprintf("使用Creem充值成功，充值额度: %v，支付金额：%.2f", quota, topUp.Money), callerIp, topUp.PaymentMethod, PaymentMethodCreem) @@ -495,10 +569,10 @@ func RechargeWaffo(tradeNo string, callerIp string) (err error) { return errors.New("充值订单状态错误") } - dAmount := decimal.NewFromInt(topUp.Amount) - dQuotaPerUnit := decimal.NewFromFloat(common.QuotaPerUnit) - quotaToAdd = int(dAmount.Mul(dQuotaPerUnit).IntPart()) - if quotaToAdd <= 0 { + quotaToAdd, err = common.QuotaFromDecimalStrict( + decimal.NewFromInt(topUp.Amount).Mul(decimal.NewFromFloat(common.QuotaPerUnit)), + ) + if err != nil || quotaToAdd <= 0 { return errors.New("无效的充值额度") } @@ -508,17 +582,14 @@ func RechargeWaffo(tradeNo string, callerIp string) (err error) { return err } - if err := tx.Model(&User{}).Where("id = ?", topUp.UserId).Update("quota", gorm.Expr("quota + ?", quotaToAdd)).Error; err != nil { - return err - } - - return nil + return tx.Model(&User{}).Where("id = ?", topUp.UserId).Update("quota", gorm.Expr("quota + ?", quotaToAdd)).Error }) if err != nil { common.SysError("waffo topup failed: " + err.Error()) return errors.New("充值失败，请稍后重试") } + syncCreditUserQuotaCache(topUp.UserId, quotaToAdd, "waffo topup") if quotaToAdd > 0 { RecordTopupLog(topUp.UserId, fmt.Sprintf("Waffo充值成功，充值额度: %v，支付金额: %.2f", logger.FormatQuota(quotaToAdd), topUp.Money), callerIp, topUp.PaymentMethod, PaymentMethodWaffo) @@ -558,8 +629,10 @@ func RechargeWaffoPancake(tradeNo string) (err error) { return errors.New("充值订单状态错误") } - quotaToAdd = int(decimal.NewFromInt(topUp.Amount).Mul(decimal.NewFromFloat(common.QuotaPerUnit)).IntPart()) - if quotaToAdd <= 0 { + quotaToAdd, err = common.QuotaFromDecimalStrict( + decimal.NewFromInt(topUp.Amount).Mul(decimal.NewFromFloat(common.QuotaPerUnit)), + ) + if err != nil || quotaToAdd <= 0 { return errors.New("无效的充值额度") } @@ -569,17 +642,14 @@ func RechargeWaffoPancake(tradeNo string) (err error) { return err } - if err := tx.Model(&User{}).Where("id = ?", topUp.UserId).Update("quota", gorm.Expr("quota + ?", quotaToAdd)).Error; err != nil { - return err - } - - return nil + return tx.Model(&User{}).Where("id = ?", topUp.UserId).Update("quota", gorm.Expr("quota + ?", quotaToAdd)).Error }) if err != nil { common.SysError("waffo pancake topup failed: " + err.Error()) return errors.New("充值失败，请稍后重试") } + syncCreditUserQuotaCache(topUp.UserId, quotaToAdd, "waffo pancake topup") if quotaToAdd > 0 { RecordLog(topUp.UserId, LogTypeTopup, fmt.Sprintf("Waffo Pancake充值成功，充值额度: %v，支付金额: %.2f", logger.FormatQuota(quotaToAdd), topUp.Money)) diff --git a/model/user_cache.go b/model/user_cache.go index aa72fc26..7ca31497 100644 --- a/model/user_cache.go +++ b/model/user_cache.go @@ -143,18 +143,33 @@ func cacheGetUserBase(userId int) (*UserBase, error) { return &userCache, nil } -// Add atomic quota operations using hash fields +// Add atomic quota operations using hash fields. +// 通过守卫式 Lua 脚本执行：哈希不存在时直接跳过（下次读取会从数据库水合）， +// 不会像裸 HINCRBY 那样创建只含 Quota 字段的残缺哈希。 func cacheIncrUserQuota(userId int, delta int64) error { if !common.RedisEnabled { return nil } - return common.RedisHIncrBy(getUserCacheKey(userId), "Quota", delta) + _, err := cacheApplyUserQuotaDelta(userId, delta) + return err } func cacheDecrUserQuota(userId int, delta int64) error { return cacheIncrUserQuota(userId, -delta) } +// syncCreditUserQuotaCache 在授信事务（充值/兑换等）提交后同步把增量补进缓存 +// 余额。预扣以缓存值为准（存在期间），授信不能绕过它，否则新到账的额度在 +// 缓存过期前不可用；缓存未命中无需处理，下次读取会从已提交的数据库余额水合。 +func syncCreditUserQuotaCache(userId int, quota int, operation string) { + if quota <= 0 { + return + } + if err := cacheIncrUserQuota(userId, int64(quota)); err != nil { + common.SysLog(fmt.Sprintf("failed to sync %s credit to user quota cache: %s", operation, err.Error())) + } +} + // Helper functions to get individual fields if needed func getUserGroupCache(userId int) (string, error) { cache, err := GetUserCache(userId)