|
- package service
-
- import (
- "fmt"
-
- "github.com/QuantumNous/new-api/common"
- "github.com/QuantumNous/new-api/model"
- "github.com/QuantumNous/new-api/setting/system_setting"
- )
-
- const BillingSourceSyncedWallet = "synced_wallet"
-
- // SyncedUserFunding 同步用户的资金来源实现
- // 同步用户的余额存储在 master 节点,本地通过原子 SQL 管理 synced_quota 快照
- // 预扣时原子扣减 synced_quota 并记录到 PendingSyncRecord,结算时同步到 master
- type SyncedUserFunding struct {
- userId int
- remoteUserId int
- requestId string
- consumed int // 实际预扣额度
- syncedQuota int // master 同步过来的余额快照(PreConsume 后为最新值)
- }
-
- func NewSyncedUserFunding(userId, remoteUserId int, requestId string, syncedQuota int) *SyncedUserFunding {
- return &SyncedUserFunding{
- userId: userId,
- remoteUserId: remoteUserId,
- requestId: requestId,
- syncedQuota: syncedQuota,
- }
- }
-
- func (s *SyncedUserFunding) Source() string { return BillingSourceSyncedWallet }
-
- func (s *SyncedUserFunding) PreConsume(amount int) error {
- if amount <= 0 {
- common.SysLog(fmt.Sprintf("[RegionSync] PreConsume skipped: userId=%d, amount=%d (<=0)", s.userId, amount))
- return nil
- }
-
- settings := system_setting.GetRegionSyncSettings()
- common.SysLog(fmt.Sprintf("[RegionSync] PreConsume: userId=%d, remoteUserId=%d, amount=%d, syncedQuota=%d, threshold=%d, requestId=%s",
- s.userId, s.remoteUserId, amount, s.syncedQuota, settings.MinBalanceThreshold, s.requestId))
-
- // 原子 SQL:检查余额 + 扣减 synced_quota
- newQuota, ok, err := model.AtomicDecreaseSyncedQuota(s.userId, amount, settings.MinBalanceThreshold)
- if err != nil {
- return fmt.Errorf("扣减同步额度失败: %w", err)
- }
- if !ok {
- return fmt.Errorf("同步用户余额不足 (userId=%d, syncedQuota=%d, need=%d, threshold=%d)",
- s.userId, newQuota, amount, settings.MinBalanceThreshold)
- }
-
- // 创建待同步记录(status=pending)
- err = model.CreatePendingSyncRecord(s.userId, s.remoteUserId, s.requestId, amount, amount)
- if err != nil {
- // 回滚:退还已扣减的额度
- common.SysError(fmt.Sprintf("[RegionSync] PreConsume failed to create pending record, rolling back: userId=%d, requestId=%s, err=%v", s.userId, s.requestId, err))
- model.IncreaseSyncedQuota(s.userId, amount)
- return fmt.Errorf("创建同步扣费记录失败: %w", err)
- }
-
- s.consumed = amount
- s.syncedQuota = newQuota
- 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))
- return nil
- }
-
- func (s *SyncedUserFunding) Settle(delta int) error {
- if delta == 0 {
- return nil
- }
-
- // 结算时更新待同步记录的实际扣费额度
- if delta > 0 {
- s.consumed += delta
- } else {
- s.consumed += delta // delta 是负数
- }
-
- // 更新本地同步余额快照
- s.syncedQuota -= delta
-
- 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))
-
- // 将 pending record 结算为 settled(原子更新 quota + status)
- settled, err := model.SettlePendingSyncRecord(s.requestId, s.consumed)
- if err != nil {
- common.SysError(fmt.Sprintf("[RegionSync] Settle: failed to settle record, requestId=%s, err=%v", s.requestId, err))
- } else if !settled {
- // 记录不存在(PreConsume 跳过 amount=0),直接创建 settled 记录
- if err := model.CreateSettledSyncRecord(s.userId, s.remoteUserId, s.requestId, s.consumed); err != nil {
- common.SysError(fmt.Sprintf("[RegionSync] Settle: failed to create settled record, requestId=%s, err=%v", s.requestId, err))
- } else {
- common.SysLog(fmt.Sprintf("[RegionSync] Settle: created settled record (PreConsume skipped), userId=%d, quota=%d, requestId=%s", s.userId, s.consumed, s.requestId))
- }
- }
-
- // 原子增量更新 DB synced_quota(避免并发请求覆盖彼此的更新)
- if delta != 0 {
- if err := model.AdjustSyncedQuota(s.userId, delta); err != nil {
- common.SysError(fmt.Sprintf("[RegionSync] Settle: failed to adjust synced_quota, userId=%d, delta=%d, err=%v", s.userId, delta, err))
- }
- }
-
- return nil
- }
-
- func (s *SyncedUserFunding) Refund() error {
- if s.consumed <= 0 {
- return nil
- }
-
- common.SysLog(fmt.Sprintf("[RegionSync] Refund: userId=%d, consumed=%d, requestId=%s", s.userId, s.consumed, s.requestId))
-
- // 取消待同步记录
- var record model.PendingSyncRecord
- err := model.DB.Where("request_id = ? AND status = ?", s.requestId, model.PendingSyncStatusPending).First(&record).Error
- if err != nil {
- common.SysLog(fmt.Sprintf("[RegionSync] Refund: no pending record found for requestId=%s", s.requestId))
- return nil // 记录不存在,无需处理
- }
-
- // 标记为已同步(实际上已取消)
- if err := model.MarkRecordSynced(record.Id); err != nil {
- common.SysError(fmt.Sprintf("[RegionSync] Refund: failed to mark record synced, recordId=%d, err=%v", record.Id, err))
- return err
- }
-
- // 原子退还 DB synced_quota
- if err := model.IncreaseSyncedQuota(s.userId, s.consumed); err != nil {
- common.SysError(fmt.Sprintf("[RegionSync] Refund: failed to increase synced_quota, userId=%d, err=%v", s.userId, err))
- return err
- }
-
- common.SysLog(fmt.Sprintf("[RegionSync] Refund success: userId=%d, recordId=%d, refunded=%d, requestId=%s", s.userId, record.Id, s.consumed, s.requestId))
- return nil
- }
|