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 }