You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

140 lines
5.4 KiB

  1. package service
  2. import (
  3. "fmt"
  4. "github.com/QuantumNous/new-api/common"
  5. "github.com/QuantumNous/new-api/model"
  6. "github.com/QuantumNous/new-api/setting/system_setting"
  7. )
  8. const BillingSourceSyncedWallet = "synced_wallet"
  9. // SyncedUserFunding 同步用户的资金来源实现
  10. // 同步用户的余额存储在 master 节点,本地通过原子 SQL 管理 synced_quota 快照
  11. // 预扣时原子扣减 synced_quota 并记录到 PendingSyncRecord,结算时同步到 master
  12. type SyncedUserFunding struct {
  13. userId int
  14. remoteUserId int
  15. requestId string
  16. consumed int // 实际预扣额度
  17. syncedQuota int // master 同步过来的余额快照(PreConsume 后为最新值)
  18. }
  19. func NewSyncedUserFunding(userId, remoteUserId int, requestId string, syncedQuota int) *SyncedUserFunding {
  20. return &SyncedUserFunding{
  21. userId: userId,
  22. remoteUserId: remoteUserId,
  23. requestId: requestId,
  24. syncedQuota: syncedQuota,
  25. }
  26. }
  27. func (s *SyncedUserFunding) Source() string { return BillingSourceSyncedWallet }
  28. func (s *SyncedUserFunding) PreConsume(amount int) error {
  29. if amount <= 0 {
  30. common.SysLog(fmt.Sprintf("[RegionSync] PreConsume skipped: userId=%d, amount=%d (<=0)", s.userId, amount))
  31. return nil
  32. }
  33. settings := system_setting.GetRegionSyncSettings()
  34. common.SysLog(fmt.Sprintf("[RegionSync] PreConsume: userId=%d, remoteUserId=%d, amount=%d, syncedQuota=%d, threshold=%d, requestId=%s",
  35. s.userId, s.remoteUserId, amount, s.syncedQuota, settings.MinBalanceThreshold, s.requestId))
  36. // 原子 SQL:检查余额 + 扣减 synced_quota
  37. newQuota, ok, err := model.AtomicDecreaseSyncedQuota(s.userId, amount, settings.MinBalanceThreshold)
  38. if err != nil {
  39. return fmt.Errorf("扣减同步额度失败: %w", err)
  40. }
  41. if !ok {
  42. return fmt.Errorf("同步用户余额不足 (userId=%d, syncedQuota=%d, need=%d, threshold=%d)",
  43. s.userId, newQuota, amount, settings.MinBalanceThreshold)
  44. }
  45. // 创建待同步记录(status=pending)
  46. err = model.CreatePendingSyncRecord(s.userId, s.remoteUserId, s.requestId, amount, amount)
  47. if err != nil {
  48. // 回滚:退还已扣减的额度
  49. common.SysError(fmt.Sprintf("[RegionSync] PreConsume failed to create pending record, rolling back: userId=%d, requestId=%s, err=%v", s.userId, s.requestId, err))
  50. model.IncreaseSyncedQuota(s.userId, amount)
  51. return fmt.Errorf("创建同步扣费记录失败: %w", err)
  52. }
  53. s.consumed = amount
  54. s.syncedQuota = newQuota
  55. common.SysLog(fmt.Sprintf("[RegionSync] PreConsume success: userId=%d, remoteUserId=%d, consumed=%d, syncedQuota=%d, requestId=%s", s.userId, s.remoteUserId, s.consumed, s.syncedQuota, s.requestId))
  56. return nil
  57. }
  58. func (s *SyncedUserFunding) Settle(delta int) error {
  59. if delta == 0 {
  60. return nil
  61. }
  62. // 结算时更新待同步记录的实际扣费额度
  63. if delta > 0 {
  64. s.consumed += delta
  65. } else {
  66. s.consumed += delta // delta 是负数
  67. }
  68. // 更新本地同步余额快照
  69. s.syncedQuota -= delta
  70. common.SysLog(fmt.Sprintf("[RegionSync] Settle: userId=%d, delta=%d, consumed=%d, syncedQuota=%d, requestId=%s", s.userId, delta, s.consumed, s.syncedQuota, s.requestId))
  71. // 将 pending record 结算为 settled(原子更新 quota + status)
  72. settled, err := model.SettlePendingSyncRecord(s.requestId, s.consumed)
  73. if err != nil {
  74. common.SysError(fmt.Sprintf("[RegionSync] Settle: failed to settle record, requestId=%s, err=%v", s.requestId, err))
  75. } else if !settled {
  76. // 记录不存在(PreConsume 跳过 amount=0),直接创建 settled 记录
  77. if err := model.CreateSettledSyncRecord(s.userId, s.remoteUserId, s.requestId, s.consumed); err != nil {
  78. common.SysError(fmt.Sprintf("[RegionSync] Settle: failed to create settled record, requestId=%s, err=%v", s.requestId, err))
  79. } else {
  80. common.SysLog(fmt.Sprintf("[RegionSync] Settle: created settled record (PreConsume skipped), userId=%d, quota=%d, requestId=%s", s.userId, s.consumed, s.requestId))
  81. }
  82. }
  83. // 原子增量更新 DB synced_quota(避免并发请求覆盖彼此的更新)
  84. if delta != 0 {
  85. if err := model.AdjustSyncedQuota(s.userId, delta); err != nil {
  86. common.SysError(fmt.Sprintf("[RegionSync] Settle: failed to adjust synced_quota, userId=%d, delta=%d, err=%v", s.userId, delta, err))
  87. }
  88. }
  89. return nil
  90. }
  91. func (s *SyncedUserFunding) Refund() error {
  92. if s.consumed <= 0 {
  93. return nil
  94. }
  95. common.SysLog(fmt.Sprintf("[RegionSync] Refund: userId=%d, consumed=%d, requestId=%s", s.userId, s.consumed, s.requestId))
  96. // 取消待同步记录
  97. var record model.PendingSyncRecord
  98. err := model.DB.Where("request_id = ? AND status = ?", s.requestId, model.PendingSyncStatusPending).First(&record).Error
  99. if err != nil {
  100. common.SysLog(fmt.Sprintf("[RegionSync] Refund: no pending record found for requestId=%s", s.requestId))
  101. return nil // 记录不存在,无需处理
  102. }
  103. // 标记为已同步(实际上已取消)
  104. if err := model.MarkRecordSynced(record.Id); err != nil {
  105. common.SysError(fmt.Sprintf("[RegionSync] Refund: failed to mark record synced, recordId=%d, err=%v", record.Id, err))
  106. return err
  107. }
  108. // 原子退还 DB synced_quota
  109. if err := model.IncreaseSyncedQuota(s.userId, s.consumed); err != nil {
  110. common.SysError(fmt.Sprintf("[RegionSync] Refund: failed to increase synced_quota, userId=%d, err=%v", s.userId, err))
  111. return err
  112. }
  113. common.SysLog(fmt.Sprintf("[RegionSync] Refund success: userId=%d, recordId=%d, refunded=%d, requestId=%s", s.userId, record.Id, s.consumed, s.requestId))
  114. return nil
  115. }