mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-27 16:07:02 +00:00
Merge pull request #5780 from Randark-JMT/fix/channel-monitor-p2
fix(monitor): P2 audit follow-ups — credential/balance semantics, capability checks, single account load, quota placeholder UI
This commit is contained in:
@@ -337,7 +337,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) {
|
||||
batchImageWorkerRuntime := service.ProvideBatchImageWorkerRuntime(batchImageRepository, accountRepository, batchImageQueue, usageBillingRepository, usageLogRepository, batchImageModelPricingResolver, apiKeyAuthCacheInvalidator, configConfig)
|
||||
scheduledTestRunnerService := service.ProvideScheduledTestRunnerService(scheduledTestPlanRepository, scheduledTestService, accountTestService, rateLimitService, configConfig)
|
||||
paymentOrderExpiryService := service.ProvidePaymentOrderExpiryService(paymentService, leaderLockCache, db)
|
||||
channelMonitorQuotaFetcher := service.NewChannelMonitorQuotaFetcher(accountUsageService, cnProviderQuotaService, cnProviderBalanceService, accountRepository)
|
||||
channelMonitorQuotaFetcher := service.NewChannelMonitorQuotaFetcher(accountUsageService, cnProviderQuotaService, cnProviderBalanceService, accountRepository, configConfig)
|
||||
channelMonitorRunner := service.ProvideChannelMonitorRunner(channelMonitorService, settingService, channelMonitorQuotaFetcher)
|
||||
channelMonitorV2Aggregator := service.ProvideChannelMonitorV2Aggregator(channelMonitorV2Repository, db, settingService)
|
||||
userPlatformQuotaUsageFlusher := service.ProvideUserPlatformQuotaUsageFlusher(configConfig, billingCache, serviceUserPlatformQuotaRepository, timingWheelService)
|
||||
|
||||
@@ -50,6 +50,10 @@ type MonitorQuotaSnapshot struct {
|
||||
Balances []MonitorBalance `json:"balances,omitempty"` // 多币种余额(如 DeepSeek CNY+USD)
|
||||
Currency string `json:"currency,omitempty"` // 主余额币种
|
||||
PlanLevel string `json:"plan_level,omitempty"` // 套餐等级(如智谱 level)
|
||||
// BalanceLow 余额低于阈值或账号被上游标记不可用(仅 cn_balance 来源)。
|
||||
// 抓取器按 Gateway.CNProviders.BalanceThreshold 判定,口径与账号停调
|
||||
// (CNProviderBalanceCheckService.checkOne)一致:任一币种达标即健康。
|
||||
BalanceLow bool `json:"balance_low,omitempty"`
|
||||
// CredentialInvalid 上游 401/403 鉴权失败(区别于网络/解析错误),
|
||||
// 检测状态据此推导 failed 而非 error。
|
||||
CredentialInvalid bool `json:"credential_invalid,omitempty"`
|
||||
|
||||
@@ -503,6 +503,13 @@ func (s *AccountUsageService) GetUsage(ctx context.Context, accountID int64, for
|
||||
return s.getUsageForAccount(ctx, account, forceProbe)
|
||||
}
|
||||
|
||||
// GetUsageForAccount 已加载账号的使用量直通入口(配额监控 fetcher 复用,
|
||||
// 避免缓存未命中时账号被加载两次——每次 GetByID 含 proxies/groups 联查)。
|
||||
func (s *AccountUsageService) GetUsageForAccount(ctx context.Context, account *Account, force ...bool) (*UsageInfo, error) {
|
||||
forceProbe := len(force) > 0 && force[0]
|
||||
return s.getUsageForAccount(ctx, account, forceProbe)
|
||||
}
|
||||
|
||||
// GetUsageBatch 批量获取账号使用量。
|
||||
// Anthropic OAuth/SetupToken 统一走 passive 链路,其他账号复用现有主动查询逻辑。
|
||||
// 单个账号失败不会中断整批请求,错误会按账号返回。
|
||||
|
||||
@@ -162,6 +162,9 @@ var (
|
||||
ErrChannelMonitorProviderIncompatible = infraerrors.BadRequest(
|
||||
"CHANNEL_MONITOR_PROVIDER_INCOMPATIBLE", "monitor provider must match the linked account platform",
|
||||
)
|
||||
ErrChannelMonitorAccountNotSupportable = infraerrors.BadRequest(
|
||||
"CHANNEL_MONITOR_ACCOUNT_NOT_SUPPORTABLE", "linked account cannot serve as a quota data source (cn coding plan must be kimi/zhipu, cn payg must be kimi/deepseek, openai requires an oauth account, anthropic requires oauth or setup-token)",
|
||||
)
|
||||
ErrChannelMonitorInvalidAPIMode = infraerrors.BadRequest(
|
||||
"CHANNEL_MONITOR_INVALID_API_MODE", "api_mode must be chat_completions or responses; responses is only supported for openai",
|
||||
)
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/config"
|
||||
"github.com/Wei-Shaw/sub2api/internal/domain"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/xai"
|
||||
"golang.org/x/sync/singleflight"
|
||||
@@ -18,10 +19,12 @@ import (
|
||||
// 渠道监控「配额模式」的配额抓取器。
|
||||
//
|
||||
// 不直接对接上游,而是把账号侧现成的用量服务归一成 domain.MonitorQuotaSnapshot:
|
||||
// - 海外 5 家(anthropic/openai/gemini/antigravity/grok)→ AccountUsageService.GetUsage
|
||||
// - 国产 coding plan(kimi/zhipu/deepseek)→ CNProviderQuotaService.QueryUsage
|
||||
// - 国产 payg(kimi/deepseek)→ CNProviderBalanceService.QueryBalance
|
||||
// (zhipu payg 无公开余额端点,QueryBalance 会返回该错误,原样透出)
|
||||
// - 海外 5 家(anthropic/openai/gemini/antigravity/grok)→ AccountUsageService.GetUsageForAccount
|
||||
// - 国产 coding plan(kimi/zhipu/deepseek)→ CNProviderQuotaService.QueryUsageForAccount
|
||||
// - 国产 payg(kimi/deepseek)→ CNProviderBalanceService.QueryBalanceForAccount
|
||||
// (zhipu payg 无公开余额端点,探测会返回该错误,原样透出)
|
||||
// 数据源统一接受已加载的 *Account:fetchUncached 路由前 GetByID 一次并传下去,
|
||||
// 下游服务不再各自重载(每次 GetByID 含 proxies/groups 联查)。
|
||||
//
|
||||
// Fetch 永不返回 error:所有失败都降级为 Success=false 的快照照常入库,
|
||||
// 由 deriveQuotaCheckResult 推导为 failed/error 状态。
|
||||
@@ -32,18 +35,19 @@ import (
|
||||
// 抓取由 singleflight 合并为一次上游查询。
|
||||
|
||||
// monitorUsageSource 海外平台账号用量查询(AccountUsageService 天然满足)。
|
||||
// 传已加载的 *Account:fetchUncached 只 GetByID 一次,下游不再重复加载。
|
||||
type monitorUsageSource interface {
|
||||
GetUsage(ctx context.Context, accountID int64, force ...bool) (*UsageInfo, error)
|
||||
GetUsageForAccount(ctx context.Context, account *Account, force ...bool) (*UsageInfo, error)
|
||||
}
|
||||
|
||||
// monitorCNQuotaSource 国产 coding plan 滚动窗口额度探测(CNProviderQuotaService 天然满足)。
|
||||
type monitorCNQuotaSource interface {
|
||||
QueryUsage(ctx context.Context, accountID int64) (*CNProviderQuotaProbeResult, error)
|
||||
QueryUsageForAccount(ctx context.Context, account *Account) (*CNProviderQuotaProbeResult, error)
|
||||
}
|
||||
|
||||
// monitorCNBalanceSource 国产 payg 余额探测(CNProviderBalanceService 天然满足)。
|
||||
type monitorCNBalanceSource interface {
|
||||
QueryBalance(ctx context.Context, accountID int64) (*CNProviderBalanceResult, error)
|
||||
QueryBalanceForAccount(ctx context.Context, account *Account) (*CNProviderBalanceResult, error)
|
||||
}
|
||||
|
||||
// monitorAccountSource 账号加载(AccountRepository 天然满足)。
|
||||
@@ -58,6 +62,8 @@ type ChannelMonitorQuotaFetcher struct {
|
||||
cnQuota monitorCNQuotaSource
|
||||
cnBalance monitorCNBalanceSource
|
||||
accounts monitorAccountSource
|
||||
// balanceThreshold cn_balance 余额告警阈值(与账号停调共用配置,见 monitorBalanceThreshold)。
|
||||
balanceThreshold float64
|
||||
|
||||
mu sync.Mutex
|
||||
cache map[int64]monitorQuotaCacheEntry
|
||||
@@ -76,8 +82,12 @@ func NewChannelMonitorQuotaFetcher(
|
||||
cnQuota *CNProviderQuotaService,
|
||||
cnBalance *CNProviderBalanceService,
|
||||
accounts AccountRepository,
|
||||
cfg *config.Config,
|
||||
) *ChannelMonitorQuotaFetcher {
|
||||
f := &ChannelMonitorQuotaFetcher{cache: make(map[int64]monitorQuotaCacheEntry)}
|
||||
f := &ChannelMonitorQuotaFetcher{
|
||||
cache: make(map[int64]monitorQuotaCacheEntry),
|
||||
balanceThreshold: monitorBalanceThreshold(cfg),
|
||||
}
|
||||
if usage != nil {
|
||||
f.usage = usage
|
||||
}
|
||||
@@ -93,6 +103,17 @@ func NewChannelMonitorQuotaFetcher(
|
||||
return f
|
||||
}
|
||||
|
||||
// monitorBalanceThreshold 余额告警阈值,与账号停调(CNProviderBalanceCheckService)
|
||||
// 共用 gateway.cn_providers.balance_threshold,保证监控 degraded 与调度器停调
|
||||
// 口径一致(任一币种达标即健康)。未配置/非正值时回退 viper 默认 0.5(config.go),
|
||||
// 避免 0 阈值下「余额=0 也不告警」相对旧 `<=0` 判定的回归。
|
||||
func monitorBalanceThreshold(cfg *config.Config) float64 {
|
||||
if cfg != nil && cfg.Gateway.CNProviders.BalanceThreshold > 0 {
|
||||
return cfg.Gateway.CNProviders.BalanceThreshold
|
||||
}
|
||||
return 0.5
|
||||
}
|
||||
|
||||
// LoadAccount 加载账号(不走缓存)。供 Create/Update 时校验
|
||||
// provider 与 account.platform 一致;账号不存在时返回错误。
|
||||
func (f *ChannelMonitorQuotaFetcher) LoadAccount(ctx context.Context, id int64) (*Account, error) {
|
||||
@@ -173,23 +194,26 @@ func (f *ChannelMonitorQuotaFetcher) fetchUncached(ctx context.Context, accountI
|
||||
return quotaErrorSnapshot("usage", "linked account not found", now)
|
||||
}
|
||||
|
||||
// 账号只在路由前加载这一次;已加载的 account 直接传给数据源
|
||||
// (GetUsageForAccount / QueryUsageForAccount / QueryBalanceForAccount),
|
||||
// 下游服务不再各自 GetByID(每次含 proxies/groups 联查)。
|
||||
switch account.Platform {
|
||||
case domain.PlatformKimi, domain.PlatformZhipu, domain.PlatformDeepseek:
|
||||
if account.IsCodingPlan() {
|
||||
return f.fetchCNQuota(ctx, accountID, now)
|
||||
return f.fetchCNQuota(ctx, account, now)
|
||||
}
|
||||
return f.fetchCNBalance(ctx, accountID, now)
|
||||
return f.fetchCNBalance(ctx, account, now)
|
||||
default:
|
||||
return f.fetchUsage(ctx, accountID, now)
|
||||
return f.fetchUsage(ctx, account, now)
|
||||
}
|
||||
}
|
||||
|
||||
// fetchUsage 海外平台:AccountUsageService.GetUsage → 快照。
|
||||
func (f *ChannelMonitorQuotaFetcher) fetchUsage(ctx context.Context, accountID int64, now time.Time) *domain.MonitorQuotaSnapshot {
|
||||
// fetchUsage 海外平台:AccountUsageService.GetUsageForAccount → 快照。
|
||||
func (f *ChannelMonitorQuotaFetcher) fetchUsage(ctx context.Context, account *Account, now time.Time) *domain.MonitorQuotaSnapshot {
|
||||
if f.usage == nil {
|
||||
return quotaErrorSnapshot("usage", "usage service is not configured", now)
|
||||
}
|
||||
usage, err := f.usage.GetUsage(ctx, accountID)
|
||||
usage, err := f.usage.GetUsageForAccount(ctx, account)
|
||||
if err != nil {
|
||||
msg := truncateMessage(sanitizeErrorMessage(err.Error()))
|
||||
return &domain.MonitorQuotaSnapshot{
|
||||
@@ -318,12 +342,12 @@ func sortedQuotaModelNames(quotas map[string]*AntigravityModelQuota) []string {
|
||||
return names
|
||||
}
|
||||
|
||||
// fetchCNQuota 国产 coding plan:CNProviderQuotaService.QueryUsage → 快照。
|
||||
func (f *ChannelMonitorQuotaFetcher) fetchCNQuota(ctx context.Context, accountID int64, now time.Time) *domain.MonitorQuotaSnapshot {
|
||||
// fetchCNQuota 国产 coding plan:CNProviderQuotaService.QueryUsageForAccount → 快照。
|
||||
func (f *ChannelMonitorQuotaFetcher) fetchCNQuota(ctx context.Context, account *Account, now time.Time) *domain.MonitorQuotaSnapshot {
|
||||
if f.cnQuota == nil {
|
||||
return quotaErrorSnapshot("cn_quota", "cn quota service is not configured", now)
|
||||
}
|
||||
result, err := f.cnQuota.QueryUsage(ctx, accountID)
|
||||
result, err := f.cnQuota.QueryUsageForAccount(ctx, account)
|
||||
if err != nil {
|
||||
msg := truncateMessage(sanitizeErrorMessage(err.Error()))
|
||||
return &domain.MonitorQuotaSnapshot{
|
||||
@@ -341,7 +365,10 @@ func (f *ChannelMonitorQuotaFetcher) fetchCNQuota(ctx context.Context, accountID
|
||||
Error: result.Error,
|
||||
FetchedAt: now,
|
||||
}
|
||||
if !result.Success && !result.CredentialValid {
|
||||
// 只有 401/403 判凭据失效(与 fetchCNBalance 口径一致):CN quota 服务的
|
||||
// CredentialValid 仅在成功路径置 true,若按 `!Success && !CredentialValid`
|
||||
// 推导,500/429/智谱业务错误全会被误判为 failed。
|
||||
if !result.Success && (result.StatusCode == 401 || result.StatusCode == 403) {
|
||||
snapshot.CredentialInvalid = true
|
||||
}
|
||||
if len(result.Tiers) > 0 {
|
||||
@@ -360,12 +387,12 @@ func (f *ChannelMonitorQuotaFetcher) fetchCNQuota(ctx context.Context, accountID
|
||||
return snapshot
|
||||
}
|
||||
|
||||
// fetchCNBalance 国产 payg:CNProviderBalanceService.QueryBalance → 快照。
|
||||
func (f *ChannelMonitorQuotaFetcher) fetchCNBalance(ctx context.Context, accountID int64, now time.Time) *domain.MonitorQuotaSnapshot {
|
||||
// fetchCNBalance 国产 payg:CNProviderBalanceService.QueryBalanceForAccount → 快照。
|
||||
func (f *ChannelMonitorQuotaFetcher) fetchCNBalance(ctx context.Context, account *Account, now time.Time) *domain.MonitorQuotaSnapshot {
|
||||
if f.cnBalance == nil {
|
||||
return quotaErrorSnapshot("cn_balance", "cn balance service is not configured", now)
|
||||
}
|
||||
result, err := f.cnBalance.QueryBalance(ctx, accountID)
|
||||
result, err := f.cnBalance.QueryBalanceForAccount(ctx, account)
|
||||
if err != nil {
|
||||
msg := truncateMessage(sanitizeErrorMessage(err.Error()))
|
||||
return &domain.MonitorQuotaSnapshot{
|
||||
@@ -386,6 +413,10 @@ func (f *ChannelMonitorQuotaFetcher) fetchCNBalance(ctx context.Context, account
|
||||
if result.Success {
|
||||
balance := result.Balance
|
||||
snapshot.Balance = &balance
|
||||
// 与账号停调(checkOne)同口径:上游标记不可用或全部币种低于阈值
|
||||
// 才告警,任一币种达标即健康(余额 5 元/阈值 10 元的账号调度器已
|
||||
// 停调,监控不能仍绿灯)。
|
||||
snapshot.BalanceLow = !result.Available || allCNBalancesBelowThreshold(result, f.balanceThreshold)
|
||||
} else if result.StatusCode == 401 || result.StatusCode == 403 {
|
||||
snapshot.CredentialInvalid = true
|
||||
}
|
||||
@@ -454,7 +485,7 @@ func usageFailureInfo(usage *UsageInfo) (failed, credentialInvalid bool, msg str
|
||||
// deriveQuotaCheckResult 把配额快照推导为检测状态(复用既有 status 枚举,
|
||||
// 时间线/可用率机制自动生效):
|
||||
// - 查询成功且无告警 → operational
|
||||
// - 任一窗口使用率 >= 阈值或余额耗尽 → degraded
|
||||
// - 任一窗口使用率 >= 阈值或余额低于阈值/不可用 → degraded
|
||||
// - 账号未关联(配置问题) → degraded
|
||||
// - 凭据失效(401/403) → failed
|
||||
// - 网络/解析等其他错误 → error
|
||||
@@ -499,8 +530,11 @@ func quotaDegradedHint(snapshot *domain.MonitorQuotaSnapshot) string {
|
||||
return fmt.Sprintf("quota high: %s at %s%%", name, strconv.FormatFloat(tier.UsedPercent, 'f', 1, 64))
|
||||
}
|
||||
}
|
||||
if snapshot.Balance != nil && *snapshot.Balance <= 0 {
|
||||
return fmt.Sprintf("balance depleted (%s)", firstNonEmpty(snapshot.Currency, "?"))
|
||||
if snapshot.BalanceLow {
|
||||
if snapshot.Balance != nil {
|
||||
return fmt.Sprintf("balance low: %s %s", strconv.FormatFloat(*snapshot.Balance, 'f', -1, 64), firstNonEmpty(snapshot.Currency, "?"))
|
||||
}
|
||||
return fmt.Sprintf("balance low (%s)", firstNonEmpty(snapshot.Currency, "?"))
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/Wei-Shaw/sub2api/internal/config"
|
||||
"github.com/Wei-Shaw/sub2api/internal/domain"
|
||||
"github.com/Wei-Shaw/sub2api/internal/pkg/xai"
|
||||
"github.com/stretchr/testify/require"
|
||||
@@ -19,18 +20,20 @@ import (
|
||||
type stubMonitorUsageSource struct {
|
||||
usage *UsageInfo
|
||||
err error
|
||||
// block 非 nil 时 GetUsage 阻塞在该 channel 上,用于并发/singleflight 测试。
|
||||
// block 非 nil 时 GetUsageForAccount 阻塞在该 channel 上,用于并发/singleflight 测试。
|
||||
block chan struct{}
|
||||
|
||||
mu sync.Mutex
|
||||
calls int
|
||||
lastCtx context.Context
|
||||
mu sync.Mutex
|
||||
calls int
|
||||
lastCtx context.Context
|
||||
lastAccount *Account
|
||||
}
|
||||
|
||||
func (s *stubMonitorUsageSource) GetUsage(ctx context.Context, accountID int64, force ...bool) (*UsageInfo, error) {
|
||||
func (s *stubMonitorUsageSource) GetUsageForAccount(ctx context.Context, account *Account, force ...bool) (*UsageInfo, error) {
|
||||
s.mu.Lock()
|
||||
s.calls++
|
||||
s.lastCtx = ctx
|
||||
s.lastAccount = account
|
||||
s.mu.Unlock()
|
||||
if s.block != nil {
|
||||
<-s.block
|
||||
@@ -44,25 +47,35 @@ func (s *stubMonitorUsageSource) getCalls() int {
|
||||
return s.calls
|
||||
}
|
||||
|
||||
type stubMonitorCNQuotaSource struct {
|
||||
result *CNProviderQuotaProbeResult
|
||||
err error
|
||||
calls int
|
||||
func (s *stubMonitorUsageSource) getLastAccount() *Account {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.lastAccount
|
||||
}
|
||||
|
||||
func (s *stubMonitorCNQuotaSource) QueryUsage(ctx context.Context, accountID int64) (*CNProviderQuotaProbeResult, error) {
|
||||
type stubMonitorCNQuotaSource struct {
|
||||
result *CNProviderQuotaProbeResult
|
||||
err error
|
||||
calls int
|
||||
lastAccount *Account
|
||||
}
|
||||
|
||||
func (s *stubMonitorCNQuotaSource) QueryUsageForAccount(ctx context.Context, account *Account) (*CNProviderQuotaProbeResult, error) {
|
||||
s.calls++
|
||||
s.lastAccount = account
|
||||
return s.result, s.err
|
||||
}
|
||||
|
||||
type stubMonitorCNBalanceSource struct {
|
||||
result *CNProviderBalanceResult
|
||||
err error
|
||||
calls int
|
||||
result *CNProviderBalanceResult
|
||||
err error
|
||||
calls int
|
||||
lastAccount *Account
|
||||
}
|
||||
|
||||
func (s *stubMonitorCNBalanceSource) QueryBalance(ctx context.Context, accountID int64) (*CNProviderBalanceResult, error) {
|
||||
func (s *stubMonitorCNBalanceSource) QueryBalanceForAccount(ctx context.Context, account *Account) (*CNProviderBalanceResult, error) {
|
||||
s.calls++
|
||||
s.lastAccount = account
|
||||
return s.result, s.err
|
||||
}
|
||||
|
||||
@@ -87,11 +100,12 @@ func newQuotaFetcherTestSetup(t *testing.T) (*ChannelMonitorQuotaFetcher, *stubM
|
||||
cnBalance := &stubMonitorCNBalanceSource{}
|
||||
accounts := &stubMonitorAccountSource{accounts: make(map[int64]*Account)}
|
||||
fetcher := &ChannelMonitorQuotaFetcher{
|
||||
usage: usage,
|
||||
cnQuota: cnQuota,
|
||||
cnBalance: cnBalance,
|
||||
accounts: accounts,
|
||||
cache: make(map[int64]monitorQuotaCacheEntry),
|
||||
usage: usage,
|
||||
cnQuota: cnQuota,
|
||||
cnBalance: cnBalance,
|
||||
accounts: accounts,
|
||||
balanceThreshold: monitorBalanceThreshold(nil),
|
||||
cache: make(map[int64]monitorQuotaCacheEntry),
|
||||
}
|
||||
return fetcher, usage, cnQuota, cnBalance, accounts
|
||||
}
|
||||
@@ -167,9 +181,10 @@ func TestQuotaFetcher_PayGAccountUsesCNBalance(t *testing.T) {
|
||||
Credentials: map[string]any{"account_mode": AccountModePayG},
|
||||
}
|
||||
cnBalance.result = &CNProviderBalanceResult{
|
||||
Success: true,
|
||||
Balance: 12.34,
|
||||
Currency: "CNY",
|
||||
Success: true,
|
||||
Available: true,
|
||||
Balance: 12.34,
|
||||
Currency: "CNY",
|
||||
Balances: []CNProviderBalanceEntry{
|
||||
{Currency: "CNY", Balance: 12.34},
|
||||
{Currency: "USD", Balance: 1.5},
|
||||
@@ -185,9 +200,53 @@ func TestQuotaFetcher_PayGAccountUsesCNBalance(t *testing.T) {
|
||||
require.Equal(t, "CNY", snapshot.Currency)
|
||||
require.Len(t, snapshot.Balances, 2)
|
||||
require.Equal(t, "USD", snapshot.Balances[1].Currency)
|
||||
require.False(t, snapshot.BalanceLow)
|
||||
require.Empty(t, snapshot.Error)
|
||||
}
|
||||
|
||||
// P2-6:fetchUncached 只 GetByID 一次,已加载的 account 指针直传数据源,
|
||||
// 三条路由都不能让下游重载账号。
|
||||
func TestQuotaFetcher_LoadsAccountOnceAndPassesItThrough(t *testing.T) {
|
||||
t.Run("overseas usage", func(t *testing.T) {
|
||||
fetcher, usage, _, _, accounts := newQuotaFetcherTestSetup(t)
|
||||
acc := &Account{ID: 21, Platform: domain.PlatformAnthropic}
|
||||
accounts.accounts[21] = acc
|
||||
usage.usage = &UsageInfo{}
|
||||
|
||||
fetcher.Fetch(context.Background(), 21)
|
||||
|
||||
require.Equal(t, 1, accounts.calls)
|
||||
require.Same(t, acc, usage.getLastAccount())
|
||||
require.Equal(t, 1, usage.getCalls())
|
||||
})
|
||||
|
||||
t.Run("cn coding plan", func(t *testing.T) {
|
||||
fetcher, _, cnQuota, _, accounts := newQuotaFetcherTestSetup(t)
|
||||
acc := &Account{ID: 22, Platform: domain.PlatformKimi, Credentials: map[string]any{"account_mode": AccountModeCoding}}
|
||||
accounts.accounts[22] = acc
|
||||
cnQuota.result = &CNProviderQuotaProbeResult{Success: true}
|
||||
|
||||
fetcher.Fetch(context.Background(), 22)
|
||||
|
||||
require.Equal(t, 1, accounts.calls)
|
||||
require.Same(t, acc, cnQuota.lastAccount)
|
||||
require.Equal(t, 1, cnQuota.calls)
|
||||
})
|
||||
|
||||
t.Run("cn payg", func(t *testing.T) {
|
||||
fetcher, _, _, cnBalance, accounts := newQuotaFetcherTestSetup(t)
|
||||
acc := &Account{ID: 23, Platform: domain.PlatformDeepseek, Credentials: map[string]any{"account_mode": AccountModePayG}}
|
||||
accounts.accounts[23] = acc
|
||||
cnBalance.result = &CNProviderBalanceResult{Success: true, Available: true, Balance: 1, Currency: "CNY"}
|
||||
|
||||
fetcher.Fetch(context.Background(), 23)
|
||||
|
||||
require.Equal(t, 1, accounts.calls)
|
||||
require.Same(t, acc, cnBalance.lastAccount)
|
||||
require.Equal(t, 1, cnBalance.calls)
|
||||
})
|
||||
}
|
||||
|
||||
// --- 失败路径(Fetch 永不返回 error) ---
|
||||
|
||||
func TestQuotaFetcher_AccountMissingYieldsLinkedAccountSnapshot(t *testing.T) {
|
||||
@@ -278,20 +337,44 @@ func TestUsageFailureInfo_ClassificationMatrix(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestQuotaFetcher_CNQuotaCredentialInvalidFlagPropagates(t *testing.T) {
|
||||
fetcher, _, cnQuota, _, accounts := newQuotaFetcherTestSetup(t)
|
||||
accounts.accounts[5] = &Account{
|
||||
ID: 5,
|
||||
Platform: domain.PlatformZhipu,
|
||||
Credentials: map[string]any{"account_mode": AccountModeCoding},
|
||||
// 凭据失效只认 401/403(与 fetchCNBalance 口径一致):CN quota 服务的
|
||||
// CredentialValid 仅成功路径置 true,500/429/智谱业务错误须推导为 error 而非 failed。
|
||||
func TestQuotaFetcher_CNQuotaCredentialInvalidByStatusCode(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
accountID int64
|
||||
statusCode int
|
||||
credentialBad bool
|
||||
expectedStatus string
|
||||
}{
|
||||
{name: "401 unauthorized", accountID: 5, statusCode: 401, credentialBad: true, expectedStatus: MonitorStatusFailed},
|
||||
{name: "403 forbidden", accountID: 15, statusCode: 403, credentialBad: true, expectedStatus: MonitorStatusFailed},
|
||||
{name: "500 server error", accountID: 16, statusCode: 500, expectedStatus: MonitorStatusError},
|
||||
{name: "429 rate limited", accountID: 17, statusCode: 429, expectedStatus: MonitorStatusError},
|
||||
// 智谱 2xx 但业务级失败:StatusCode=200,非凭据问题。
|
||||
{name: "200 business error", accountID: 18, statusCode: 200, expectedStatus: MonitorStatusError},
|
||||
}
|
||||
cnQuota.result = &CNProviderQuotaProbeResult{Success: false, CredentialValid: false, Error: "api key expired"}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
fetcher, _, cnQuota, _, accounts := newQuotaFetcherTestSetup(t)
|
||||
accounts.accounts[tc.accountID] = &Account{
|
||||
ID: tc.accountID,
|
||||
Platform: domain.PlatformZhipu,
|
||||
Credentials: map[string]any{"account_mode": AccountModeCoding},
|
||||
}
|
||||
cnQuota.result = &CNProviderQuotaProbeResult{
|
||||
Success: false,
|
||||
StatusCode: tc.statusCode,
|
||||
Error: "api key expired",
|
||||
}
|
||||
|
||||
snapshot := fetcher.Fetch(context.Background(), 5)
|
||||
snapshot := fetcher.Fetch(context.Background(), tc.accountID)
|
||||
|
||||
require.False(t, snapshot.Success)
|
||||
require.True(t, snapshot.CredentialInvalid)
|
||||
require.Equal(t, "api key expired", snapshot.Error)
|
||||
require.False(t, snapshot.Success)
|
||||
require.Equal(t, tc.credentialBad, snapshot.CredentialInvalid)
|
||||
require.Equal(t, tc.expectedStatus, deriveQuotaCheckResult(snapshot, "quota", time.Now()).Status)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestQuotaFetcher_CNBalanceHTTP403MarksCredentialInvalid(t *testing.T) {
|
||||
@@ -305,6 +388,86 @@ func TestQuotaFetcher_CNBalanceHTTP403MarksCredentialInvalid(t *testing.T) {
|
||||
require.True(t, snapshot.CredentialInvalid)
|
||||
}
|
||||
|
||||
// 余额告警口径与账号停调(CNProviderBalanceCheckService.checkOne)一致:
|
||||
// 上游标记不可用或全部币种低于阈值 → BalanceLow → degraded;任一币种达标即健康。
|
||||
func TestQuotaFetcher_CNBalanceLowMarksDegraded(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
accountID int64
|
||||
result *CNProviderBalanceResult
|
||||
balanceLow bool
|
||||
wantStatus string
|
||||
wantMessage string
|
||||
}{
|
||||
{
|
||||
// 审查例:余额 5/阈值 10 的账号调度器已停调,监控不能仍绿灯。
|
||||
name: "balance below threshold",
|
||||
accountID: 21,
|
||||
result: &CNProviderBalanceResult{Success: true, Available: true, Balance: 5, Currency: "CNY"},
|
||||
balanceLow: true,
|
||||
wantStatus: MonitorStatusDegraded, wantMessage: "balance low: 5 CNY",
|
||||
},
|
||||
{
|
||||
name: "upstream marked unavailable",
|
||||
accountID: 22,
|
||||
result: &CNProviderBalanceResult{Success: true, Available: false, Balance: 20, Currency: "CNY"},
|
||||
balanceLow: true,
|
||||
wantStatus: MonitorStatusDegraded, wantMessage: "balance low: 20 CNY",
|
||||
},
|
||||
{
|
||||
// deepseek 双币种:任一币种(USD 20)达标即健康。
|
||||
name: "any currency above threshold is healthy",
|
||||
accountID: 23,
|
||||
result: &CNProviderBalanceResult{
|
||||
Success: true, Available: true, Balance: 5, Currency: "CNY",
|
||||
Balances: []CNProviderBalanceEntry{{Currency: "CNY", Balance: 5}, {Currency: "USD", Balance: 20}},
|
||||
},
|
||||
wantStatus: MonitorStatusOperational,
|
||||
},
|
||||
{
|
||||
name: "single currency above threshold",
|
||||
accountID: 24,
|
||||
result: &CNProviderBalanceResult{Success: true, Available: true, Balance: 20, Currency: "CNY"},
|
||||
wantStatus: MonitorStatusOperational,
|
||||
},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
fetcher, _, _, cnBalance, accounts := newQuotaFetcherTestSetup(t)
|
||||
fetcher.balanceThreshold = 10
|
||||
accounts.accounts[tc.accountID] = &Account{
|
||||
ID: tc.accountID,
|
||||
Platform: domain.PlatformKimi,
|
||||
Credentials: map[string]any{"account_mode": AccountModePayG},
|
||||
}
|
||||
cnBalance.result = tc.result
|
||||
|
||||
snapshot := fetcher.Fetch(context.Background(), tc.accountID)
|
||||
|
||||
require.True(t, snapshot.Success)
|
||||
require.Equal(t, tc.balanceLow, snapshot.BalanceLow)
|
||||
res := deriveQuotaCheckResult(snapshot, "quota", time.Now())
|
||||
require.Equal(t, tc.wantStatus, res.Status)
|
||||
if tc.wantMessage != "" {
|
||||
require.Contains(t, res.Message, tc.wantMessage)
|
||||
} else {
|
||||
require.Empty(t, res.Message)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewChannelMonitorQuotaFetcher_ThresholdFromConfig(t *testing.T) {
|
||||
require.InDelta(t, 0.5, NewChannelMonitorQuotaFetcher(nil, nil, nil, nil, nil).balanceThreshold, 0.0001)
|
||||
|
||||
cfg10 := &config.Config{Gateway: config.GatewayConfig{CNProviders: config.GatewayCNProvidersConfig{BalanceThreshold: 10}}}
|
||||
require.InDelta(t, 10, NewChannelMonitorQuotaFetcher(nil, nil, nil, nil, cfg10).balanceThreshold, 0.0001)
|
||||
|
||||
// 非正值(含显式 0)回退默认,避免 0 阈值下「余额=0 也不告警」。
|
||||
cfg0 := &config.Config{Gateway: config.GatewayConfig{CNProviders: config.GatewayCNProvidersConfig{BalanceThreshold: 0}}}
|
||||
require.InDelta(t, 0.5, NewChannelMonitorQuotaFetcher(nil, nil, nil, nil, cfg0).balanceThreshold, 0.0001)
|
||||
}
|
||||
|
||||
func TestQuotaFetcher_NilDependenciesProduceErrorSnapshots(t *testing.T) {
|
||||
// fetcher 本体为 nil:直接降级为错误快照,不 panic。
|
||||
var nilFetcher *ChannelMonitorQuotaFetcher
|
||||
@@ -495,10 +658,10 @@ func TestDeriveQuotaCheckResult_StatusMatrix(t *testing.T) {
|
||||
require.Contains(t, res.Message, "95.0%")
|
||||
|
||||
balance := -0.5
|
||||
depleted := &domain.MonitorQuotaSnapshot{Success: true, Balance: &balance, Currency: "CNY"}
|
||||
res = deriveQuotaCheckResult(depleted, "quota", now)
|
||||
lowBalance := &domain.MonitorQuotaSnapshot{Success: true, BalanceLow: true, Balance: &balance, Currency: "CNY"}
|
||||
res = deriveQuotaCheckResult(lowBalance, "quota", now)
|
||||
require.Equal(t, MonitorStatusDegraded, res.Status)
|
||||
require.Contains(t, res.Message, "balance depleted")
|
||||
require.Contains(t, res.Message, "balance low")
|
||||
|
||||
invalid := &domain.MonitorQuotaSnapshot{Success: false, CredentialInvalid: true, Error: "401 unauthorized"}
|
||||
res = deriveQuotaCheckResult(invalid, "quota", now)
|
||||
|
||||
@@ -325,6 +325,16 @@ func TestValidateCreateParams_CheckModeMatrix(t *testing.T) {
|
||||
},
|
||||
wantErr: ErrChannelMonitorInvalidCheckMode,
|
||||
},
|
||||
{
|
||||
// quota_probe 仍要打真实探活请求:空模型必须报错,不再用 "quota" 占位。
|
||||
name: "quota_probe requires primary model",
|
||||
params: ChannelMonitorCreateParams{
|
||||
Provider: MonitorProviderKimi, CheckMode: MonitorCheckModeQuotaProbe,
|
||||
Endpoint: "https://api.kimi.com", APIKey: "sk",
|
||||
IntervalSeconds: 60, AccountID: &accountID,
|
||||
},
|
||||
wantErr: ErrChannelMonitorMissingPrimaryModel,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
@@ -342,8 +352,15 @@ func TestValidateCreateParams_CheckModeMatrix(t *testing.T) {
|
||||
func TestNormalizeMonitorPrimaryModel_QuotaDefault(t *testing.T) {
|
||||
require.Equal(t, "quota", normalizeMonitorPrimaryModel(MonitorProviderKimi, MonitorCheckModeQuota, ""))
|
||||
require.Equal(t, "quota", normalizeMonitorPrimaryModel(MonitorProviderAntigravity, MonitorCheckModeQuota, " "))
|
||||
// 探活模式沿用原语义:grok 默认模型,其余必填(空串报错在 validateCreateParams)。
|
||||
// quota_probe 仍要打真实探活请求:空模型返回 ""(由上层报 MissingPrimaryModel),
|
||||
// 不再用 "quota" 占位打 model="quota" 的请求。
|
||||
require.Equal(t, "", normalizeMonitorPrimaryModel(MonitorProviderKimi, MonitorCheckModeQuotaProbe, ""))
|
||||
// grok 分支在纯 quota 占位之后:grok+quota 占位 "quota",
|
||||
// grok 探活(probe/quota_probe)默认轻量测活模型。
|
||||
require.Equal(t, "quota", normalizeMonitorPrimaryModel(MonitorProviderGrok, MonitorCheckModeQuota, ""))
|
||||
require.Equal(t, MonitorDefaultGrokModel, normalizeMonitorPrimaryModel(MonitorProviderGrok, MonitorCheckModeProbe, ""))
|
||||
require.Equal(t, MonitorDefaultGrokModel, normalizeMonitorPrimaryModel(MonitorProviderGrok, MonitorCheckModeQuotaProbe, ""))
|
||||
// 探活模式沿用原语义:其余必填(空串报错在 validateCreateParams)。
|
||||
require.Equal(t, "kimi-k2", normalizeMonitorPrimaryModel(MonitorProviderKimi, MonitorCheckModeQuotaProbe, "kimi-k2"))
|
||||
}
|
||||
|
||||
@@ -414,6 +431,150 @@ func TestRevalidateLinkedAccount_PlatformMismatch(t *testing.T) {
|
||||
require.Nil(t, probe.AccountID)
|
||||
}
|
||||
|
||||
// 能力矩阵:与 fetchUncached 路由一一对应,创建期拦截注定运行期永久 error 的组合。
|
||||
func TestMonitorAccountQuotaCapability_Matrix(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
account *Account
|
||||
wantErr error
|
||||
}{
|
||||
{
|
||||
name: "deepseek coding has no quota endpoint",
|
||||
account: &Account{ID: 1, Platform: domain.PlatformDeepseek, Credentials: map[string]any{"account_mode": AccountModeCoding}},
|
||||
wantErr: ErrChannelMonitorAccountNotSupportable,
|
||||
},
|
||||
{
|
||||
// 自定义域名 kimi coding:GetCodingPlanProvider 识别不到 → 无额度端点。
|
||||
name: "custom-domain kimi coding unsupported",
|
||||
account: &Account{ID: 2, Platform: domain.PlatformKimi, Type: AccountTypeAPIKey,
|
||||
Credentials: map[string]any{"account_mode": AccountModeCoding, "base_url": "https://cw.example.com"}},
|
||||
wantErr: ErrChannelMonitorAccountNotSupportable,
|
||||
},
|
||||
{
|
||||
name: "kimi coding default endpoint ok",
|
||||
account: &Account{ID: 3, Platform: domain.PlatformKimi, Credentials: map[string]any{"account_mode": AccountModeCoding}},
|
||||
},
|
||||
{
|
||||
name: "zhipu coding default endpoint ok",
|
||||
account: &Account{ID: 4, Platform: domain.PlatformZhipu, Credentials: map[string]any{"account_mode": AccountModeCoding}},
|
||||
},
|
||||
{
|
||||
name: "zhipu payg has no balance endpoint",
|
||||
account: &Account{ID: 5, Platform: domain.PlatformZhipu},
|
||||
wantErr: ErrChannelMonitorAccountNotSupportable,
|
||||
},
|
||||
{
|
||||
name: "kimi payg ok",
|
||||
account: &Account{ID: 6, Platform: domain.PlatformKimi},
|
||||
},
|
||||
{
|
||||
name: "deepseek payg ok",
|
||||
account: &Account{ID: 7, Platform: domain.PlatformDeepseek},
|
||||
},
|
||||
{
|
||||
name: "anthropic api key cannot query usage",
|
||||
account: &Account{ID: 8, Platform: domain.PlatformAnthropic, Type: AccountTypeAPIKey},
|
||||
wantErr: ErrChannelMonitorAccountNotSupportable,
|
||||
},
|
||||
{
|
||||
name: "anthropic oauth ok",
|
||||
account: &Account{ID: 9, Platform: domain.PlatformAnthropic, Type: AccountTypeOAuth},
|
||||
},
|
||||
{
|
||||
name: "anthropic setup token ok (local estimation)",
|
||||
account: &Account{ID: 10, Platform: domain.PlatformAnthropic, Type: AccountTypeSetupToken},
|
||||
},
|
||||
{
|
||||
name: "openai api key cannot query usage",
|
||||
account: &Account{ID: 11, Platform: domain.PlatformOpenAI, Type: AccountTypeAPIKey},
|
||||
wantErr: ErrChannelMonitorAccountNotSupportable,
|
||||
},
|
||||
{
|
||||
name: "openai oauth ok",
|
||||
account: &Account{ID: 12, Platform: domain.PlatformOpenAI, Type: AccountTypeOAuth},
|
||||
},
|
||||
{
|
||||
// 防过度拦截:gemini/grok/antigravity 走本地统计/值通道降级,不会永久 error。
|
||||
name: "gemini api key ok",
|
||||
account: &Account{ID: 13, Platform: domain.PlatformGemini, Type: AccountTypeAPIKey},
|
||||
},
|
||||
{
|
||||
name: "grok ok",
|
||||
account: &Account{ID: 14, Platform: domain.PlatformGrok},
|
||||
},
|
||||
{
|
||||
name: "antigravity ok",
|
||||
account: &Account{ID: 15, Platform: domain.PlatformAntigravity},
|
||||
},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
err := monitorAccountQuotaCapability(tc.account)
|
||||
if tc.wantErr == nil {
|
||||
require.NoError(t, err)
|
||||
} else {
|
||||
require.ErrorIs(t, err, tc.wantErr)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateLinkedAccount_CapabilityRejected(t *testing.T) {
|
||||
svc := NewChannelMonitorService(nil, nil)
|
||||
svc.SetQuotaFetcher(newQuotaModeFetcher(map[int64]*Account{
|
||||
1: {ID: 1, Platform: domain.PlatformDeepseek, Credentials: map[string]any{"account_mode": AccountModeCoding}},
|
||||
}, nil))
|
||||
|
||||
err := svc.validateLinkedAccount(context.Background(), MonitorProviderDeepseek, int64Ptr(1))
|
||||
require.ErrorIs(t, err, ErrChannelMonitorAccountNotSupportable)
|
||||
}
|
||||
|
||||
func TestRevalidateLinkedAccount_Capability(t *testing.T) {
|
||||
svc := NewChannelMonitorService(nil, nil)
|
||||
svc.SetQuotaFetcher(newQuotaModeFetcher(map[int64]*Account{
|
||||
2: {ID: 2, Platform: domain.PlatformDeepseek, Credentials: map[string]any{"account_mode": AccountModeCoding}},
|
||||
}, nil))
|
||||
|
||||
quota := &ChannelMonitor{Provider: MonitorProviderDeepseek, CheckMode: MonitorCheckModeQuota, AccountID: int64Ptr(2)}
|
||||
require.ErrorIs(t, svc.revalidateLinkedAccount(context.Background(), quota), ErrChannelMonitorAccountNotSupportable)
|
||||
require.NotNil(t, quota.AccountID, "quota mode keeps the binding for the admin to fix")
|
||||
|
||||
probe := &ChannelMonitor{Provider: MonitorProviderDeepseek, CheckMode: MonitorCheckModeProbe, AccountID: int64Ptr(2)}
|
||||
require.NoError(t, svc.revalidateLinkedAccount(context.Background(), probe))
|
||||
require.Nil(t, probe.AccountID, "probe mode should silently unbind unusable account")
|
||||
}
|
||||
|
||||
// provider-only 更新不得绕过 provider × check_mode 组合校验。
|
||||
func TestApplyMonitorUpdate_ProviderOnlyRevalidatesCheckMode(t *testing.T) {
|
||||
probeKimi := func() *ChannelMonitor {
|
||||
return &ChannelMonitor{
|
||||
Provider: MonitorProviderKimi, APIMode: MonitorAPIModeChatCompletions,
|
||||
Endpoint: "https://api.kimi.com", PrimaryModel: "kimi-k2",
|
||||
CheckMode: MonitorCheckModeProbe,
|
||||
}
|
||||
}
|
||||
|
||||
provider := MonitorProviderAntigravity
|
||||
err := applyMonitorUpdate(probeKimi(), ChannelMonitorUpdateParams{Provider: &provider})
|
||||
require.ErrorIs(t, err, ErrChannelMonitorInvalidCheckMode)
|
||||
|
||||
// 带上 check_mode/account_id 的完整切换合法。
|
||||
accountID := int64(3)
|
||||
err = applyMonitorUpdate(probeKimi(), ChannelMonitorUpdateParams{
|
||||
Provider: &provider, CheckMode: strPtr(MonitorCheckModeQuota), AccountID: &accountID,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// 存量非法行(antigravity+probe)仅改名/停用不被砖化。
|
||||
legacy := &ChannelMonitor{
|
||||
Provider: MonitorProviderAntigravity, APIMode: MonitorAPIModeChatCompletions,
|
||||
Endpoint: "https://example.com", PrimaryModel: "gemini-3-pro",
|
||||
CheckMode: MonitorCheckModeProbe,
|
||||
}
|
||||
newName := "renamed"
|
||||
require.NoError(t, applyMonitorUpdate(legacy, ChannelMonitorUpdateParams{Name: &newName}))
|
||||
}
|
||||
|
||||
// --- quota → probe 切换的 key 管控(validateProbeAPIKey) ---
|
||||
|
||||
func TestValidateProbeAPIKey_QuotaToProbeRequiresFreshKey(t *testing.T) {
|
||||
|
||||
@@ -406,7 +406,8 @@ func validateCreateParams(p ChannelMonitorCreateParams) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// validateLinkedAccount 校验关联账号存在且平台与监控 provider 一致。
|
||||
// validateLinkedAccount 校验关联账号存在、平台与监控 provider 一致、且能充当
|
||||
// 配额数据源(能力拦截,见 monitorAccountQuotaCapability)。
|
||||
// fetcher 未注入时 fail-closed(拒绝创建配额监控,而不是创建后静默坏)。
|
||||
func (s *ChannelMonitorService) validateLinkedAccount(ctx context.Context, provider string, accountID *int64) error {
|
||||
if accountID == nil || *accountID <= 0 {
|
||||
@@ -422,7 +423,7 @@ func (s *ChannelMonitorService) validateLinkedAccount(ctx context.Context, provi
|
||||
if account.Platform != provider {
|
||||
return ErrChannelMonitorProviderIncompatible
|
||||
}
|
||||
return nil
|
||||
return monitorAccountQuotaCapability(account)
|
||||
}
|
||||
|
||||
// Update 更新监控。APIKey 字段:nil 或空字符串 = 不修改;非空 = 加密后覆盖。
|
||||
@@ -533,6 +534,15 @@ func (s *ChannelMonitorService) revalidateLinkedAccount(ctx context.Context, m *
|
||||
m.AccountID = nil
|
||||
return nil
|
||||
}
|
||||
// 能力失配(如 deepseek coding / zhipu payg / API-Key 型海外账号):
|
||||
// quota 模式显式报错(有该类存量监控时编辑会被拦,出路是换账号或切 probe),
|
||||
// probe 模式账号无用途,静默解绑。
|
||||
if err := monitorAccountQuotaCapability(account); err != nil {
|
||||
if usesQuota {
|
||||
return err
|
||||
}
|
||||
m.AccountID = nil
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -886,11 +896,16 @@ func applyMonitorUpdate(existing *ChannelMonitor, p ChannelMonitorUpdateParams)
|
||||
existing.Provider = *p.Provider
|
||||
}
|
||||
if p.CheckMode != nil {
|
||||
mode := defaultCheckMode(*p.CheckMode)
|
||||
if err := validateCheckMode(existing.Provider, mode); err != nil {
|
||||
existing.CheckMode = defaultCheckMode(*p.CheckMode)
|
||||
}
|
||||
// provider 与 check_mode 任一变化后统一复核组合矩阵:provider-only 更新
|
||||
// (如把 probe 监控的 provider 改成 antigravity)也不得落库非法组合,否则
|
||||
// 运行期恒 error。条件限定避免把存量非法行的 name/enabled-only 更新也判死
|
||||
// (否则连改名/停用都无法操作)。
|
||||
if p.Provider != nil || p.CheckMode != nil {
|
||||
if err := validateCheckMode(existing.Provider, defaultCheckMode(existing.CheckMode)); err != nil {
|
||||
return err
|
||||
}
|
||||
existing.CheckMode = mode
|
||||
}
|
||||
if p.AccountID != nil {
|
||||
if *p.AccountID > 0 {
|
||||
|
||||
@@ -190,21 +190,61 @@ func normalizeModels(in []string) []string {
|
||||
return out
|
||||
}
|
||||
|
||||
// normalizeMonitorPrimaryModel applies provider/check_mode defaults while
|
||||
// preserving the existing required-model behavior:
|
||||
// - Grok 探活默认轻量测活模型
|
||||
// - quota 模式占位 "quota"(primary_model 列 NotEmpty;历史行/时间线机制无需特判)
|
||||
// normalizeMonitorPrimaryModel applies provider/check_mode defaults:
|
||||
// - pure quota mode never sends requests: placeholder "quota" keeps
|
||||
// primary_model NOT NULL (history rows / timeline need no special-casing)
|
||||
// - quota_probe still sends a real probe request: empty model returns ""
|
||||
// so validateCreateParams / applyMonitorUpdate report
|
||||
// ErrChannelMonitorMissingPrimaryModel instead of probing model="quota"
|
||||
// - Grok probing (probe/quota_probe) defaults to the lightweight check model
|
||||
func normalizeMonitorPrimaryModel(provider, checkMode, model string) string {
|
||||
model = strings.TrimSpace(model)
|
||||
if model == "" && defaultCheckMode(checkMode) == MonitorCheckModeQuota {
|
||||
return MonitorDefaultQuotaModel
|
||||
}
|
||||
if model == "" && provider == MonitorProviderGrok {
|
||||
return MonitorDefaultGrokModel
|
||||
}
|
||||
if model == "" && monitorCheckModeUsesQuota(defaultCheckMode(checkMode)) {
|
||||
return MonitorDefaultQuotaModel
|
||||
}
|
||||
return model
|
||||
}
|
||||
|
||||
// monitorAccountQuotaCapability 校验关联账号能否充当配额数据源,与
|
||||
// fetchUncached 的路由一一对应(coding→CN 额度端点 / payg→CN 余额端点 /
|
||||
// 其余→AccountUsageService)。在创建/更新期拦截注定运行期永久 error 的组合:
|
||||
// - kimi/zhipu/deepseek coding:GetCodingPlanProvider 须识别为 kimi/zhipu
|
||||
// (deepseek coding、自定义域名 kimi coding 无法路由额度端点)
|
||||
// - kimi/zhipu/deepseek payg:仅 kimi/deepseek 有公开余额端点(zhipu payg 无)
|
||||
// - anthropic:OAuth / Setup Token(API-Key 型无 usage 通道,永久 error)
|
||||
// - openai:OAuth(API-Key 型无 usage 通道)
|
||||
// - gemini/grok/antigravity:本地统计/值通道降级,不会永久 error,放行
|
||||
func monitorAccountQuotaCapability(account *Account) error {
|
||||
switch account.Platform {
|
||||
case PlatformKimi, PlatformZhipu, PlatformDeepseek:
|
||||
if account.IsCodingPlan() {
|
||||
if p := account.GetCodingPlanProvider(); p != PlatformKimi && p != PlatformZhipu {
|
||||
return ErrChannelMonitorAccountNotSupportable
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if account.Platform == PlatformZhipu {
|
||||
return ErrChannelMonitorAccountNotSupportable
|
||||
}
|
||||
return nil
|
||||
case PlatformAnthropic:
|
||||
if account.Type == AccountTypeOAuth || account.Type == AccountTypeSetupToken {
|
||||
return nil
|
||||
}
|
||||
return ErrChannelMonitorAccountNotSupportable
|
||||
case PlatformOpenAI:
|
||||
if account.Type == AccountTypeOAuth {
|
||||
return nil
|
||||
}
|
||||
return ErrChannelMonitorAccountNotSupportable
|
||||
default:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// defaultAPIMode 空串归一为 chat_completions,保证历史数据与旧客户端兼容。
|
||||
func defaultAPIMode(apiMode string) string {
|
||||
if strings.TrimSpace(apiMode) == "" {
|
||||
|
||||
@@ -84,14 +84,27 @@ func NewCNProviderBalanceService(
|
||||
|
||||
// QueryBalance 探测指定 payg 账号的余额并落 Extra 快照。
|
||||
func (s *CNProviderBalanceService) QueryBalance(ctx context.Context, accountID int64) (*CNProviderBalanceResult, error) {
|
||||
account, err := s.loadPayGAccount(ctx, accountID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s.QueryBalanceForAccount(ctx, account)
|
||||
}
|
||||
|
||||
// QueryBalanceForAccount 探测已加载账号(配额监控 fetcher / 周期余额检测复用,
|
||||
// 避免二次 GetByID)。singleflight key 与 QueryBalance 相同,按账号 ID 合并。
|
||||
func (s *CNProviderBalanceService) QueryBalanceForAccount(ctx context.Context, account *Account) (*CNProviderBalanceResult, error) {
|
||||
if s == nil || s.accountRepo == nil || s.httpUpstream == nil {
|
||||
return nil, infraerrors.New(http.StatusInternalServerError, "CN_BALANCE_NOT_CONFIGURED", "cn provider balance service is not configured")
|
||||
}
|
||||
key := "cn_balance:" + strconv.FormatInt(accountID, 10)
|
||||
if err := validatePayGAccount(account); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
key := "cn_balance:" + strconv.FormatInt(account.ID, 10)
|
||||
resultCh := s.flight.DoChan(key, func() (any, error) {
|
||||
probeCtx, cancel := context.WithTimeout(context.Background(), cnBalanceUpstreamTimeout+5*time.Second)
|
||||
defer cancel()
|
||||
return s.queryBalance(probeCtx, accountID)
|
||||
return s.queryBalanceForAccount(probeCtx, account)
|
||||
})
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
@@ -109,11 +122,7 @@ func (s *CNProviderBalanceService) QueryBalance(ctx context.Context, accountID i
|
||||
}
|
||||
}
|
||||
|
||||
func (s *CNProviderBalanceService) queryBalance(ctx context.Context, accountID int64) (*CNProviderBalanceResult, error) {
|
||||
account, err := s.loadPayGAccount(ctx, accountID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
func (s *CNProviderBalanceService) queryBalanceForAccount(ctx context.Context, account *Account) (*CNProviderBalanceResult, error) {
|
||||
provider := account.Platform
|
||||
if provider != PlatformKimi && provider != PlatformDeepseek {
|
||||
return nil, infraerrors.New(http.StatusBadRequest, "CN_BALANCE_NO_ENDPOINT", "account provider has no balance endpoint")
|
||||
@@ -230,17 +239,26 @@ func (s *CNProviderBalanceService) loadPayGAccount(ctx context.Context, accountI
|
||||
if err != nil {
|
||||
return nil, infraerrors.Newf(http.StatusNotFound, "CN_BALANCE_ACCOUNT_NOT_FOUND", "account not found: %v", err)
|
||||
}
|
||||
if err := validatePayGAccount(account); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return account, nil
|
||||
}
|
||||
|
||||
// validatePayGAccount 加载后的非 DB 校验(ForAccount 入口同样复用,
|
||||
// 保证直传 account 也不绕过平台/模式检查)。
|
||||
func validatePayGAccount(account *Account) error {
|
||||
if account == nil {
|
||||
return nil, infraerrors.New(http.StatusNotFound, "CN_BALANCE_ACCOUNT_NOT_FOUND", "account not found")
|
||||
return infraerrors.New(http.StatusNotFound, "CN_BALANCE_ACCOUNT_NOT_FOUND", "account not found")
|
||||
}
|
||||
if !account.IsCNProvider() {
|
||||
return nil, infraerrors.New(http.StatusBadRequest, "CN_BALANCE_INVALID_PLATFORM", "account is not a CN provider account")
|
||||
return infraerrors.New(http.StatusBadRequest, "CN_BALANCE_INVALID_PLATFORM", "account is not a CN provider account")
|
||||
}
|
||||
// coding 账号走额度探测,余额端点不适用。
|
||||
if account.IsCodingPlan() {
|
||||
return nil, infraerrors.New(http.StatusBadRequest, "CN_BALANCE_CODING_PLAN", "coding plan account has no balance endpoint; use quota probe")
|
||||
return infraerrors.New(http.StatusBadRequest, "CN_BALANCE_CODING_PLAN", "coding plan account has no balance endpoint; use quota probe")
|
||||
}
|
||||
return account, nil
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *CNProviderBalanceService) resolveProxyURL(ctx context.Context, account *Account) string {
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
package service
|
||||
|
||||
// ForAccount 直传入口(P2-6)的校验回归测试:
|
||||
// QueryUsageForAccount / QueryBalanceForAccount 接受已加载的 *Account,
|
||||
// 但必须复用与 ID 入口相同的加载后校验——直传不能绕过平台/模式检查,
|
||||
// 且校验在 singleflight 之前完成(无效账号不得发起任何上游请求)。
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func codingAccount(platform string) *Account {
|
||||
return &Account{
|
||||
ID: 1, Platform: platform, Type: AccountTypeAPIKey, Status: StatusActive,
|
||||
Credentials: map[string]any{"account_mode": AccountModeCoding, "api_key": "sk-test"},
|
||||
}
|
||||
}
|
||||
|
||||
func paygAccount(platform string) *Account {
|
||||
return &Account{
|
||||
ID: 2, Platform: platform, Type: AccountTypeAPIKey, Status: StatusActive,
|
||||
Credentials: map[string]any{"account_mode": AccountModePayG, "api_key": "sk-test"},
|
||||
}
|
||||
}
|
||||
|
||||
func requireReason(t *testing.T, err error, reason string) {
|
||||
t.Helper()
|
||||
require.Error(t, err)
|
||||
var appErr *infraerrors.ApplicationError
|
||||
require.ErrorAs(t, err, &appErr)
|
||||
require.Equal(t, reason, appErr.Reason)
|
||||
}
|
||||
|
||||
func TestValidateCodingPlanAccount_Matrix(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
account *Account
|
||||
wantReason string
|
||||
}{
|
||||
{name: "nil", account: nil, wantReason: "CN_QUOTA_ACCOUNT_NOT_FOUND"},
|
||||
{name: "non cn provider", account: &Account{ID: 3, Platform: PlatformAnthropic}, wantReason: "CN_QUOTA_INVALID_PLATFORM"},
|
||||
{name: "payg has no quota endpoint", account: paygAccount(PlatformKimi), wantReason: "CN_QUOTA_NOT_CODING_PLAN"},
|
||||
{name: "kimi coding ok", account: codingAccount(PlatformKimi)},
|
||||
{name: "zhipu coding ok", account: codingAccount(PlatformZhipu)},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
err := validateCodingPlanAccount(tc.account)
|
||||
if tc.wantReason == "" {
|
||||
require.NoError(t, err)
|
||||
return
|
||||
}
|
||||
requireReason(t, err, tc.wantReason)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidatePayGAccount_Matrix(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
account *Account
|
||||
wantReason string
|
||||
}{
|
||||
{name: "nil", account: nil, wantReason: "CN_BALANCE_ACCOUNT_NOT_FOUND"},
|
||||
{name: "non cn provider", account: &Account{ID: 3, Platform: PlatformAnthropic}, wantReason: "CN_BALANCE_INVALID_PLATFORM"},
|
||||
{name: "coding has no balance endpoint", account: codingAccount(PlatformKimi), wantReason: "CN_BALANCE_CODING_PLAN"},
|
||||
{name: "kimi payg ok", account: paygAccount(PlatformKimi)},
|
||||
{name: "deepseek payg ok", account: paygAccount(PlatformDeepseek)},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
err := validatePayGAccount(tc.account)
|
||||
if tc.wantReason == "" {
|
||||
require.NoError(t, err)
|
||||
return
|
||||
}
|
||||
requireReason(t, err, tc.wantReason)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 直传入口的校验在 singleflight/上游请求之前:无效账号必须零出站请求。
|
||||
func TestCNProviderQuotaService_QueryUsageForAccount_RejectsInvalidAccount(t *testing.T) {
|
||||
repo := &fakeCNProbeAccountRepo{}
|
||||
upstream := &recordingHTTPUpstream{}
|
||||
svc := NewCNProviderQuotaService(repo, nil, upstream, nil)
|
||||
|
||||
_, err := svc.QueryUsageForAccount(context.Background(), paygAccount(PlatformKimi))
|
||||
requireReason(t, err, "CN_QUOTA_NOT_CODING_PLAN")
|
||||
require.Zero(t, upstream.calls)
|
||||
|
||||
_, err = svc.QueryUsageForAccount(context.Background(), nil)
|
||||
requireReason(t, err, "CN_QUOTA_ACCOUNT_NOT_FOUND")
|
||||
require.Zero(t, upstream.calls)
|
||||
}
|
||||
|
||||
func TestCNProviderBalanceService_QueryBalanceForAccount_RejectsInvalidAccount(t *testing.T) {
|
||||
repo := &fakeCNProbeAccountRepo{}
|
||||
upstream := &recordingHTTPUpstream{}
|
||||
svc := NewCNProviderBalanceService(repo, nil, upstream, nil)
|
||||
|
||||
_, err := svc.QueryBalanceForAccount(context.Background(), codingAccount(PlatformKimi))
|
||||
requireReason(t, err, "CN_BALANCE_CODING_PLAN")
|
||||
require.Zero(t, upstream.calls)
|
||||
|
||||
_, err = svc.QueryBalanceForAccount(context.Background(), &Account{ID: 9, Platform: PlatformAnthropic})
|
||||
requireReason(t, err, "CN_BALANCE_INVALID_PLATFORM")
|
||||
require.Zero(t, upstream.calls)
|
||||
}
|
||||
|
||||
// ID 入口与 ForAccount 入口对同一账号的行为一致(loadCodingPlanAccount 的
|
||||
// 加载后校验 = validateCodingPlanAccount;余额侧对称)。
|
||||
func TestCNProviderServices_IDEntryAppliesSameValidation(t *testing.T) {
|
||||
repo := &fakeCNProbeAccountRepo{account: paygAccount(PlatformKimi)}
|
||||
upstream := &recordingHTTPUpstream{}
|
||||
svc := NewCNProviderQuotaService(repo, nil, upstream, nil)
|
||||
|
||||
_, err := svc.QueryUsage(context.Background(), 2)
|
||||
requireReason(t, err, "CN_QUOTA_NOT_CODING_PLAN")
|
||||
require.Zero(t, upstream.calls)
|
||||
}
|
||||
@@ -89,14 +89,27 @@ func NewCNProviderQuotaService(
|
||||
// QueryUsage 探测指定账号的 Coding Plan 滚动窗口用量并落 Extra 快照。
|
||||
// 同一账号的并发探测会被 singleflight 合并。
|
||||
func (s *CNProviderQuotaService) QueryUsage(ctx context.Context, accountID int64) (*CNProviderQuotaProbeResult, error) {
|
||||
account, err := s.loadCodingPlanAccount(ctx, accountID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s.QueryUsageForAccount(ctx, account)
|
||||
}
|
||||
|
||||
// QueryUsageForAccount 探测已加载账号(配额监控 fetcher 复用,避免二次 GetByID)。
|
||||
// singleflight key 与 QueryUsage 相同,按账号 ID 与 admin 侧并发探测合并。
|
||||
func (s *CNProviderQuotaService) QueryUsageForAccount(ctx context.Context, account *Account) (*CNProviderQuotaProbeResult, error) {
|
||||
if s == nil || s.accountRepo == nil || s.httpUpstream == nil {
|
||||
return nil, infraerrors.New(http.StatusInternalServerError, "CN_QUOTA_NOT_CONFIGURED", "cn provider quota service is not configured")
|
||||
}
|
||||
key := "cn_quota:" + strconv.FormatInt(accountID, 10)
|
||||
if err := validateCodingPlanAccount(account); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
key := "cn_quota:" + strconv.FormatInt(account.ID, 10)
|
||||
resultCh := s.flight.DoChan(key, func() (any, error) {
|
||||
probeCtx, cancel := context.WithTimeout(context.Background(), cnQuotaUpstreamTimeout+5*time.Second)
|
||||
defer cancel()
|
||||
return s.queryUsage(probeCtx, accountID)
|
||||
return s.queryUsageForAccount(probeCtx, account)
|
||||
})
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
@@ -114,12 +127,7 @@ func (s *CNProviderQuotaService) QueryUsage(ctx context.Context, accountID int64
|
||||
}
|
||||
}
|
||||
|
||||
func (s *CNProviderQuotaService) queryUsage(ctx context.Context, accountID int64) (*CNProviderQuotaProbeResult, error) {
|
||||
account, err := s.loadCodingPlanAccount(ctx, accountID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
func (s *CNProviderQuotaService) queryUsageForAccount(ctx context.Context, account *Account) (*CNProviderQuotaProbeResult, error) {
|
||||
provider := account.GetCodingPlanProvider()
|
||||
if provider != PlatformKimi && provider != PlatformZhipu {
|
||||
return nil, infraerrors.New(http.StatusBadRequest, "CN_QUOTA_NOT_CODING_PLAN", "account is not a kimi/zhipu coding plan account")
|
||||
@@ -231,18 +239,27 @@ func (s *CNProviderQuotaService) loadCodingPlanAccount(ctx context.Context, acco
|
||||
if err != nil {
|
||||
return nil, infraerrors.Newf(http.StatusNotFound, "CN_QUOTA_ACCOUNT_NOT_FOUND", "account not found: %v", err)
|
||||
}
|
||||
if account == nil {
|
||||
return nil, infraerrors.New(http.StatusNotFound, "CN_QUOTA_ACCOUNT_NOT_FOUND", "account not found")
|
||||
}
|
||||
if !account.IsCNProvider() {
|
||||
return nil, infraerrors.New(http.StatusBadRequest, "CN_QUOTA_INVALID_PLATFORM", "account is not a CN provider account")
|
||||
}
|
||||
if !account.IsCodingPlan() {
|
||||
return nil, infraerrors.New(http.StatusBadRequest, "CN_QUOTA_NOT_CODING_PLAN", "account is not a coding plan account")
|
||||
if err := validateCodingPlanAccount(account); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return account, nil
|
||||
}
|
||||
|
||||
// validateCodingPlanAccount 加载后的非 DB 校验(ForAccount 入口同样复用,
|
||||
// 保证直传 account 也不绕过平台/模式检查)。
|
||||
func validateCodingPlanAccount(account *Account) error {
|
||||
if account == nil {
|
||||
return infraerrors.New(http.StatusNotFound, "CN_QUOTA_ACCOUNT_NOT_FOUND", "account not found")
|
||||
}
|
||||
if !account.IsCNProvider() {
|
||||
return infraerrors.New(http.StatusBadRequest, "CN_QUOTA_INVALID_PLATFORM", "account is not a CN provider account")
|
||||
}
|
||||
if !account.IsCodingPlan() {
|
||||
return infraerrors.New(http.StatusBadRequest, "CN_QUOTA_NOT_CODING_PLAN", "account is not a coding plan account")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *CNProviderQuotaService) resolveProxyURL(ctx context.Context, account *Account) string {
|
||||
if account == nil || account.ProxyID == nil {
|
||||
return ""
|
||||
|
||||
Reference in New Issue
Block a user