diff --git a/backend/cmd/server/wire.go b/backend/cmd/server/wire.go index a38962ffdb..06720610ba 100644 --- a/backend/cmd/server/wire.go +++ b/backend/cmd/server/wire.go @@ -125,6 +125,7 @@ func provideCleanup( upstreamBillingProbe *service.UpstreamBillingProbeService, ollamaCloudUsage *service.OllamaCloudUsageService, auditLog *service.AuditLogService, + openAIAutoReset *service.OpenAIQuotaAutoResetService, promptAudit *securityaudit.PromptService, pluginManager *service.PluginManager, ) func() { @@ -145,6 +146,12 @@ func provideCleanup( } return nil }}, + {"OpenAIQuotaAutoResetService", func() error { + if openAIAutoReset != nil { + openAIAutoReset.Stop() + } + return nil + }}, {"OpsIngressRejectAggregator", func() error { if opsIngressReject != nil { opsIngressReject.Stop() diff --git a/backend/cmd/server/wire_gen.go b/backend/cmd/server/wire_gen.go index a51e13bb9d..b373606891 100644 --- a/backend/cmd/server/wire_gen.go +++ b/backend/cmd/server/wire_gen.go @@ -319,7 +319,8 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { batchImageHandler := handler.ProvideBatchImageHandler(batchImagePublicService, batchImageDownloadService, batchImageCleanupService, openAIGatewayHandler) idempotencyCoordinator := service.ProvideIdempotencyCoordinator(idempotencyRepository, configConfig) idempotencyCleanupService := service.ProvideIdempotencyCleanupService(idempotencyRepository, configConfig) - handlers := handler.ProvideHandlers(authHandler, userHandler, apiKeyHandler, usageHandler, redeemHandler, subscriptionHandler, announcementHandler, channelMonitorUserHandler, channelMonitorV2Handler, adminHandlers, gatewayHandler, openAIGatewayHandler, handlerSettingHandler, totpHandler, passkeyHandler, handlerPaymentHandler, paymentWebhookHandler, availableChannelHandler, modelPlazaHandler, asyncImageHandler, batchImageHandler, idempotencyCoordinator, idempotencyCleanupService) + openAIQuotaAutoResetService := service.ProvideOpenAIQuotaAutoResetService(accountRepository, openAIQuotaService, rateLimitService, idempotencyCoordinator, auditLogService, settingService, leaderLockCache) + handlers := handler.ProvideHandlers(authHandler, userHandler, apiKeyHandler, usageHandler, redeemHandler, subscriptionHandler, announcementHandler, channelMonitorUserHandler, channelMonitorV2Handler, adminHandlers, gatewayHandler, openAIGatewayHandler, handlerSettingHandler, totpHandler, passkeyHandler, handlerPaymentHandler, paymentWebhookHandler, availableChannelHandler, modelPlazaHandler, asyncImageHandler, batchImageHandler, idempotencyCoordinator, idempotencyCleanupService, openAIQuotaAutoResetService) jwtAuthMiddleware := middleware.NewJWTAuthMiddleware(authService, userService, settingService, auditLogService) optionalJWTAuthMiddleware := middleware.NewOptionalJWTAuthMiddleware(authService, userService, settingService, auditLogService) adminAuthMiddleware := middleware.NewAdminAuthMiddleware(authService, userService, settingService, auditLogService) @@ -346,7 +347,7 @@ func initializeApplication(buildInfo handler.BuildInfo) (*Application, error) { channelMonitorRunner := service.ProvideChannelMonitorRunner(channelMonitorService, settingService, channelMonitorQuotaFetcher) channelMonitorV2Aggregator := service.ProvideChannelMonitorV2Aggregator(channelMonitorV2Repository, db, settingService) userPlatformQuotaUsageFlusher := service.ProvideUserPlatformQuotaUsageFlusher(configConfig, billingCache, serviceUserPlatformQuotaRepository, timingWheelService) - v := provideCleanup(client, redisClient, opsMetricsCollector, opsAggregationService, opsAlertEvaluatorService, opsCleanupService, opsScheduledReportService, opsSystemLogSink, opsService, opsIngressRejectAggregator, apiKeyService, authCacheInvalidationWorker, schedulerSnapshotService, tokenRefreshService, accountExpiryService, cnProviderBalanceCheckService, openAICodexVersionSyncService, proxyExpiryService, subscriptionExpiryService, usageCleanupService, idempotencyCleanupService, batchImageCleanupService, batchImageWorkerRuntime, pricingService, emailQueueService, billingCacheService, usageRecordWorkerPool, subscriptionService, oAuthService, openAIOAuthService, geminiOAuthService, antigravityOAuthService, grokOAuthService, openAIGatewayService, scheduledTestRunnerService, backupService, paymentOrderExpiryService, channelMonitorRunner, channelMonitorV2Aggregator, userPlatformQuotaUsageFlusher, upstreamBillingProbeService, ollamaCloudUsageService, auditLogService, promptService, pluginManager) + v := provideCleanup(client, redisClient, opsMetricsCollector, opsAggregationService, opsAlertEvaluatorService, opsCleanupService, opsScheduledReportService, opsSystemLogSink, opsService, opsIngressRejectAggregator, apiKeyService, authCacheInvalidationWorker, schedulerSnapshotService, tokenRefreshService, accountExpiryService, cnProviderBalanceCheckService, openAICodexVersionSyncService, proxyExpiryService, subscriptionExpiryService, usageCleanupService, idempotencyCleanupService, batchImageCleanupService, batchImageWorkerRuntime, pricingService, emailQueueService, billingCacheService, usageRecordWorkerPool, subscriptionService, oAuthService, openAIOAuthService, geminiOAuthService, antigravityOAuthService, grokOAuthService, openAIGatewayService, scheduledTestRunnerService, backupService, paymentOrderExpiryService, channelMonitorRunner, channelMonitorV2Aggregator, userPlatformQuotaUsageFlusher, upstreamBillingProbeService, ollamaCloudUsageService, auditLogService, openAIQuotaAutoResetService, promptService, pluginManager) application := &Application{ Server: httpServer, PromptAudit: promptService, @@ -427,6 +428,7 @@ func provideCleanup( upstreamBillingProbe *service.UpstreamBillingProbeService, ollamaCloudUsage *service.OllamaCloudUsageService, auditLog *service.AuditLogService, + openAIAutoReset *service.OpenAIQuotaAutoResetService, promptAudit *securityaudit.PromptService, pluginManager *service.PluginManager, ) func() { @@ -446,6 +448,12 @@ func provideCleanup( } return nil }}, + {"OpenAIQuotaAutoResetService", func() error { + if openAIAutoReset != nil { + openAIAutoReset.Stop() + } + return nil + }}, {"OpsIngressRejectAggregator", func() error { if opsIngressReject != nil { opsIngressReject.Stop() diff --git a/backend/cmd/server/wire_gen_test.go b/backend/cmd/server/wire_gen_test.go index a4858cc363..b3581641ea 100644 --- a/backend/cmd/server/wire_gen_test.go +++ b/backend/cmd/server/wire_gen_test.go @@ -94,6 +94,7 @@ func TestProvideCleanup_WithMinimalDependencies_NoPanic(t *testing.T) { nil, // upstreamBillingProbe nil, // ollamaCloudUsage nil, // auditLog + nil, // openAIAutoReset nil, // promptAudit nil, // pluginManager ) diff --git a/backend/internal/handler/admin/openai_oauth_handler.go b/backend/internal/handler/admin/openai_oauth_handler.go index d7cfc8d2a6..779e4455cd 100644 --- a/backend/internal/handler/admin/openai_oauth_handler.go +++ b/backend/internal/handler/admin/openai_oauth_handler.go @@ -34,12 +34,6 @@ type openAIAccountStateRecoverer interface { RecoverAccountState(ctx context.Context, accountID int64, options service.AccountRecoveryOptions) (*service.SuccessfulTestRecoveryResult, error) } -const ( - openAIQuotaResetWarningCacheRefreshFailed = "reset_credit_cache_refresh_failed" - openAIQuotaResetWarningAccountRecoveryFailed = "account_state_recovery_failed" - openAIQuotaResetWarningAccountRefreshFailed = "account_state_refresh_failed" -) - // openAIQuotaResetPostProcessTimeout bounds the work performed AFTER the // (non-refundable) reset credit has already been consumed upstream. The whole // request must stay comfortably inside the panel HTTP client timeout, otherwise @@ -493,6 +487,7 @@ func (h *OpenAIOAuthHandler) QueryQuota(c *gin.Context) { response.ErrorFrom(c, err) return } + service.NotifyOpenAIAutoResetCredit(accountID) response.Success(c, usage) } @@ -523,6 +518,7 @@ func (h *OpenAIOAuthHandler) RefreshQuota(c *gin.Context) { response.Error(c, http.StatusInternalServerError, "openai quota query returned an empty result") return } + service.NotifyOpenAIAutoResetCredit(accountID) refreshResponse := openAIQuotaRefreshResponse{OpenAIQuotaUsage: *usage} // A failed snapshot write leaves the previous cache intact — report it as a @@ -600,54 +596,19 @@ func (h *OpenAIOAuthHandler) ResetQuota(c *gin.Context) { postCtx, cancelPost := openAIQuotaResetPostProcessContext(c.Request.Context()) defer cancelPost() - // Step 1 — unblocking the account is the whole point of consuming a credit - // (#3672 / #3740), so it runs FIRST and is never gated on the display cache. - // Recovery is DB-only and leaves the manual `schedulable` switch untouched. - if h.rateLimitService == nil { - resetResponse.WarningCode = openAIQuotaResetWarningAccountRecoveryFailed - response.Success(c, resetResponse) - return + postResult := service.RunOpenAIQuotaResetPostProcess( + postCtx, + accountID, + h.quotaService, + h.rateLimitService, + h.adminService.GetAccount, + ) + resetResponse.Quota = postResult.Quota + resetResponse.CacheRefreshed = postResult.CacheRefreshed + resetResponse.AccountStateRecovered = postResult.AccountStateRecovered + resetResponse.WarningCode = postResult.WarningCode + if postResult.Account != nil { + resetResponse.Account = dto.AccountFromService(postResult.Account) } - if _, err := h.rateLimitService.RecoverAccountState(postCtx, accountID, service.AccountRecoveryOptions{ - InvalidateToken: true, - }); err != nil { - // Recovery failures are almost always storage-level; the remaining steps - // share that dependency, so stop here instead of compounding the failure. - slog.Warn("openai_quota_reset_account_recovery_failed", "account_id", accountID, "error", err) - resetResponse.WarningCode = openAIQuotaResetWarningAccountRecoveryFailed - response.Success(c, resetResponse) - return - } - resetResponse.AccountStateRecovered = true - - // Step 2 — refresh the reset-credit display cache. A failure here is reported - // but must not hide the recovered account row produced by step 3. - usage, usageErr := h.quotaService.QueryUsage(postCtx, accountID) - switch { - case usageErr != nil || usage == nil: - slog.Warn("openai_quota_reset_cache_refresh_failed", "account_id", accountID, "error", usageErr) - resetResponse.WarningCode = openAIQuotaResetWarningCacheRefreshFailed - default: - if err := h.quotaService.CacheResetCreditsSnapshot(postCtx, accountID, usage.RateLimitResetCredits); err != nil { - slog.Warn("openai_quota_reset_cache_refresh_failed", "account_id", accountID, "error", err) - resetResponse.WarningCode = openAIQuotaResetWarningCacheRefreshFailed - } else { - resetResponse.Quota = usage - resetResponse.CacheRefreshed = true - } - } - - // Step 3 — hand back the post-recovery account row so the list drops the - // stale rate-limit badge without waiting for the next poll. - account, err := h.adminService.GetAccount(postCtx, accountID) - if err != nil { - slog.Warn("openai_quota_reset_account_refresh_failed", "account_id", accountID, "error", err) - if resetResponse.WarningCode == "" { - resetResponse.WarningCode = openAIQuotaResetWarningAccountRefreshFailed - } - response.Success(c, resetResponse) - return - } - resetResponse.Account = dto.AccountFromService(account) response.Success(c, resetResponse) } diff --git a/backend/internal/handler/admin/openai_oauth_handler_reset_quota_test.go b/backend/internal/handler/admin/openai_oauth_handler_reset_quota_test.go index c0e3ca161d..b9e8dac679 100644 --- a/backend/internal/handler/admin/openai_oauth_handler_reset_quota_test.go +++ b/backend/internal/handler/admin/openai_oauth_handler_reset_quota_test.go @@ -215,7 +215,7 @@ func TestOpenAIResetQuota_RecoveryFailureStopsWorkflow(t *testing.T) { status, envelope := performOpenAIQuotaResetRequest(t, handler) require.Equal(t, http.StatusOK, status) - require.Equal(t, openAIQuotaResetWarningAccountRecoveryFailed, envelope.Data.WarningCode) + require.Equal(t, service.OpenAIQuotaResetWarningAccountRecoveryFailed, envelope.Data.WarningCode) require.False(t, envelope.Data.AccountStateRecovered) require.False(t, envelope.Data.CacheRefreshed) require.Nil(t, envelope.Data.Quota) @@ -237,7 +237,7 @@ func TestOpenAIResetQuota_MissingRecovererReportsRecoveryFailure(t *testing.T) { status, envelope := performOpenAIQuotaResetRequest(t, handler) require.Equal(t, http.StatusOK, status) - require.Equal(t, openAIQuotaResetWarningAccountRecoveryFailed, envelope.Data.WarningCode) + require.Equal(t, service.OpenAIQuotaResetWarningAccountRecoveryFailed, envelope.Data.WarningCode) require.False(t, envelope.Data.AccountStateRecovered) require.Zero(t, quota.queryCalls) require.Zero(t, adminService.calls) @@ -260,7 +260,7 @@ func TestOpenAIResetQuota_QueryFailureStillRecoversAndReturnsAccount(t *testing. status, envelope := performOpenAIQuotaResetRequest(t, handler) require.Equal(t, http.StatusOK, status) - require.Equal(t, openAIQuotaResetWarningCacheRefreshFailed, envelope.Data.WarningCode) + require.Equal(t, service.OpenAIQuotaResetWarningCacheRefreshFailed, envelope.Data.WarningCode) require.True(t, envelope.Data.AccountStateRecovered) require.False(t, envelope.Data.CacheRefreshed) require.Nil(t, envelope.Data.Quota) @@ -285,7 +285,7 @@ func TestOpenAIResetQuota_CacheFailureStillRecoversAndReturnsAccount(t *testing. status, envelope := performOpenAIQuotaResetRequest(t, handler) require.Equal(t, http.StatusOK, status) - require.Equal(t, openAIQuotaResetWarningCacheRefreshFailed, envelope.Data.WarningCode) + require.Equal(t, service.OpenAIQuotaResetWarningCacheRefreshFailed, envelope.Data.WarningCode) require.True(t, envelope.Data.AccountStateRecovered) require.False(t, envelope.Data.CacheRefreshed) require.Nil(t, envelope.Data.Quota) @@ -307,7 +307,7 @@ func TestOpenAIResetQuota_AccountRefreshFailureReportsRecoveredState(t *testing. status, envelope := performOpenAIQuotaResetRequest(t, handler) require.Equal(t, http.StatusOK, status) - require.Equal(t, openAIQuotaResetWarningAccountRefreshFailed, envelope.Data.WarningCode) + require.Equal(t, service.OpenAIQuotaResetWarningAccountRefreshFailed, envelope.Data.WarningCode) require.True(t, envelope.Data.CacheRefreshed) require.True(t, envelope.Data.AccountStateRecovered) require.NotNil(t, envelope.Data.Quota) @@ -331,7 +331,7 @@ func TestOpenAIResetQuota_CacheAndAccountFailureKeepsFirstWarning(t *testing.T) status, envelope := performOpenAIQuotaResetRequest(t, handler) require.Equal(t, http.StatusOK, status) - require.Equal(t, openAIQuotaResetWarningCacheRefreshFailed, envelope.Data.WarningCode) + require.Equal(t, service.OpenAIQuotaResetWarningCacheRefreshFailed, envelope.Data.WarningCode) require.True(t, envelope.Data.AccountStateRecovered) require.Nil(t, envelope.Data.Account) } @@ -435,7 +435,7 @@ func TestOpenAIQuotaEmptyUsageIsHandledWithoutPanic(t *testing.T) { status, envelope := performOpenAIQuotaResetRequest(t, handler) require.Equal(t, http.StatusOK, status) - require.Equal(t, openAIQuotaResetWarningCacheRefreshFailed, envelope.Data.WarningCode) + require.Equal(t, service.OpenAIQuotaResetWarningCacheRefreshFailed, envelope.Data.WarningCode) require.True(t, envelope.Data.AccountStateRecovered) require.NotNil(t, envelope.Data.Account) require.Zero(t, quota.cacheCalls) diff --git a/backend/internal/handler/wire.go b/backend/internal/handler/wire.go index bd11f72f9a..0e75dc9258 100644 --- a/backend/internal/handler/wire.go +++ b/backend/internal/handler/wire.go @@ -196,6 +196,7 @@ func ProvideHandlers( batchImageHandler *BatchImageHandler, _ *service.IdempotencyCoordinator, _ *service.IdempotencyCleanupService, + _ *service.OpenAIQuotaAutoResetService, ) *Handlers { return &Handlers{ Auth: authHandler, diff --git a/backend/internal/service/account_usage_service.go b/backend/internal/service/account_usage_service.go index 1661eca7c6..f49d3aaa71 100644 --- a/backend/internal/service/account_usage_service.go +++ b/backend/internal/service/account_usage_service.go @@ -729,6 +729,9 @@ func (s *AccountUsageService) getOpenAIUsage(ctx context.Context, account *Accou if updates := buildCodexSparkWindowExtraUpdates(quotaUsage, now); len(updates) > 0 { mergeAccountExtra(account, updates) s.persistOpenAICodexProbeSnapshot(account.ID, updates) + if account.ParentAccountID != nil { + notifyOpenAIAutoReset(*account.ParentAccountID) + } if usage.UpdatedAt == nil { usage.UpdatedAt = &now } @@ -921,7 +924,9 @@ func (s *AccountUsageService) persistOpenAICodexProbeSnapshot(accountID int64, u go func() { updateCtx, updateCancel := context.WithTimeout(context.Background(), 5*time.Second) defer updateCancel() - _ = s.accountRepo.UpdateExtra(updateCtx, accountID, updates) + if err := s.accountRepo.UpdateExtra(updateCtx, accountID, updates); err == nil { + notifyOpenAIAutoReset(accountID) + } }() } diff --git a/backend/internal/service/admin_account.go b/backend/internal/service/admin_account.go index cc09fb808f..ae194f61e1 100644 --- a/backend/internal/service/admin_account.go +++ b/backend/internal/service/admin_account.go @@ -470,6 +470,10 @@ func (s *adminServiceImpl) CreateAccount(ctx context.Context, input *CreateAccou if err != nil { return nil, err } + accountExtra, err = normalizeOpenAIAutoResetCreditExtra(input.Platform, input.Type, false, accountExtra) + if err != nil { + return nil, err + } // 绑定分组 groupIDs := input.GroupIDs @@ -559,6 +563,14 @@ func (s *adminServiceImpl) UpdateAccount(ctx context.Context, id int64, input *U if err != nil { return nil, err } + effectiveType := account.Type + if input.Type != "" { + effectiveType = input.Type + } + normalizedExtra, err = normalizeOpenAIAutoResetCreditExtra(account.Platform, effectiveType, account.IsShadow(), normalizedExtra) + if err != nil { + return nil, err + } } previousProbeIdentity := upstreamBillingProbeIdentity(account) previousOllamaUsageIdentity := ollamaCloudUsageIdentity(account) @@ -649,6 +661,7 @@ func (s *adminServiceImpl) UpdateAccount(ctx context.Context, id int64, input *U OllamaCloudUsageSessionExtraKey, OllamaCloudUsageAutoRefreshExtraKey, OllamaCloudUsageSnapshotExtraKey, + OpenAIAutoResetCreditStateExtraKey, } { if v, ok := account.Extra[key]; ok { normalizedExtra[key] = v @@ -860,6 +873,7 @@ func (s *adminServiceImpl) UpdateAccount(ctx context.Context, id int64, input *U // （如 model_rate_limits / passive_usage_* 等）。 func (s *adminServiceImpl) UpdateAccountExtra(ctx context.Context, id int64, updates map[string]any) error { updates = sanitizedCodexFingerprintExtraUpdates(updates) + updates = stripOpenAIAutoResetCreditManagedExtra(updates, true) delete(updates, UpstreamBillingProbeEnabledExtraKey) delete(updates, UpstreamBillingRateSyncEnabledExtraKey) delete(updates, UpstreamBillingProbeExtraKey) @@ -886,6 +900,7 @@ func (s *adminServiceImpl) UpdateAccountExtra(ctx context.Context, id int64, upd func (s *adminServiceImpl) BulkUpdateAccounts(ctx context.Context, input *BulkUpdateAccountsInput) (*BulkUpdateAccountsResult, error) { // Managed probe/session state may only enter through dedicated typed endpoints. input.Extra = sanitizedCodexFingerprintExtraUpdates(input.Extra) + input.Extra = stripOpenAIAutoResetCreditManagedExtra(input.Extra, true) delete(input.Extra, UpstreamBillingProbeEnabledExtraKey) delete(input.Extra, UpstreamBillingRateSyncEnabledExtraKey) delete(input.Extra, UpstreamBillingProbeExtraKey) diff --git a/backend/internal/service/openai_account_scheduler.go b/backend/internal/service/openai_account_scheduler.go index 2f2fef563b..2986f6abeb 100644 --- a/backend/internal/service/openai_account_scheduler.go +++ b/backend/internal/service/openai_account_scheduler.go @@ -1780,6 +1780,9 @@ func (s *defaultOpenAIAccountScheduler) isAccountRequestCompatibleReason(ctx con // rechecks won't reach healthy accounts that fell outside TopK — manifesting as // "no available accounts" even though healthy ones exist. if paused, decision := shouldAutoPauseOpenAIAccountByQuota(ctx, account); paused { + if decision.reason != "" { + return false, decision.reason + } reason := "quota_auto_pause" if decision.window != "" { reason += "_" + decision.window diff --git a/backend/internal/service/openai_gateway_chat_completions.go b/backend/internal/service/openai_gateway_chat_completions.go index 7d9241a7ca..5699dc09e9 100644 --- a/backend/internal/service/openai_gateway_chat_completions.go +++ b/backend/internal/service/openai_gateway_chat_completions.go @@ -392,6 +392,8 @@ func (s *OpenAIGatewayService) ForwardAsChatCompletions( if snapshot := ParseCodexRateLimitHeaders(resp.Header); snapshot != nil { s.updateCodexUsageSnapshot(ctx, account.ID, snapshot) } + } else if handleErr == nil && account.IsShadow() && account.ParentAccountID != nil { + notifyOpenAIAutoReset(*account.ParentAccountID) } return result, handleErr diff --git a/backend/internal/service/openai_gateway_forward.go b/backend/internal/service/openai_gateway_forward.go index c567a00be7..eb2f5c55a6 100644 --- a/backend/internal/service/openai_gateway_forward.go +++ b/backend/internal/service/openai_gateway_forward.go @@ -1164,6 +1164,8 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco if snapshot := ParseCodexRateLimitHeaders(resp.Header); snapshot != nil { s.updateCodexUsageSnapshot(ctx, account.ID, snapshot) } + } else if account.IsShadow() && account.ParentAccountID != nil { + notifyOpenAIAutoReset(*account.ParentAccountID) } if usage == nil { diff --git a/backend/internal/service/openai_gateway_messages.go b/backend/internal/service/openai_gateway_messages.go index 75841f1061..734cde2cab 100644 --- a/backend/internal/service/openai_gateway_messages.go +++ b/backend/internal/service/openai_gateway_messages.go @@ -509,6 +509,8 @@ func (s *OpenAIGatewayService) ForwardAsAnthropic( if snapshot := ParseCodexRateLimitHeaders(resp.Header); snapshot != nil { s.updateCodexUsageSnapshot(ctx, account.ID, snapshot) } + } else if handleErr == nil && account.IsShadow() && account.ParentAccountID != nil { + notifyOpenAIAutoReset(*account.ParentAccountID) } return result, handleErr diff --git a/backend/internal/service/openai_gateway_passthrough.go b/backend/internal/service/openai_gateway_passthrough.go index a4375f681e..b2115357d7 100644 --- a/backend/internal/service/openai_gateway_passthrough.go +++ b/backend/internal/service/openai_gateway_passthrough.go @@ -500,6 +500,8 @@ func (s *OpenAIGatewayService) forwardOpenAIPassthrough( if snapshot := ParseCodexRateLimitHeaders(resp.Header); snapshot != nil { s.updateCodexUsageSnapshot(ctx, account.ID, snapshot) } + } else if account.ParentAccountID != nil { + notifyOpenAIAutoReset(*account.ParentAccountID) } if usage == nil { diff --git a/backend/internal/service/openai_gateway_scheduling.go b/backend/internal/service/openai_gateway_scheduling.go index 5a0de5cd65..2a4a4f3d09 100644 --- a/backend/internal/service/openai_gateway_scheduling.go +++ b/backend/internal/service/openai_gateway_scheduling.go @@ -450,6 +450,7 @@ type openAIQuotaAutoPauseDecision struct { window string threshold float64 utilization float64 + reason string } func shouldAutoPauseGrokAccountByQuota(account *Account) (bool, openAIQuotaAutoPauseDecision) { @@ -520,6 +521,38 @@ func shouldAutoPauseOpenAIAccountByQuota(ctx context.Context, account *Account) if account == nil || !account.IsOpenAI() { return false, openAIQuotaAutoPauseDecision{} } + // 自动用卡有独立阈值：达到消费阈值时必须先退出调度；仅达到普通暂停阈值时， + // 只有新鲜状态明确存在可用卡才继续放行到消费阈值。 + if config := ResolveOpenAIAutoResetCreditConfig(account); config.Enabled { + now := time.Now() + utilization5h, has5h := resolveOpenAIQuotaUtilization(account.Extra, "5h", now) + utilization7d, has7d := resolveOpenAIQuotaUtilization(account.Extra, "7d", now) + if has5h && utilization5h >= config.Threshold5h { + notifyOpenAIAutoReset(account.ID) + return true, openAIQuotaAutoPauseDecision{window: "5h", threshold: config.Threshold5h, utilization: utilization5h, reason: "quota_auto_reset_pending_5h"} + } + if has7d && utilization7d >= config.Threshold7d { + notifyOpenAIAutoReset(account.ID) + return true, openAIQuotaAutoPauseDecision{window: "7d", threshold: config.Threshold7d, utilization: utilization7d, reason: "quota_auto_reset_pending_7d"} + } + + disabled5h := resolveAccountExtraBool(account.Extra, "auto_pause_5h_disabled") + disabled7d := resolveAccountExtraBool(account.Extra, "auto_pause_7d_disabled") + pause5h, pause7d := resolveOpenAIQuotaAutoPauseThresholds(ctx, account) + pauseReached5h := !disabled5h && pause5h > 0 && has5h && utilization5h >= pause5h + pauseReached7d := !disabled7d && pause7d > 0 && has7d && utilization7d >= pause7d + if pauseReached5h || pauseReached7d { + state := openAIAutoResetStateFromExtra(account.Extra) + if state != nil && state.Status == OpenAIAutoResetStatusAvailable && state.AvailableCount > 0 && !openAIAutoResetStateStale(state, now) { + return false, openAIQuotaAutoPauseDecision{} + } + notifyOpenAIAutoReset(account.ID) + if pauseReached5h { + return true, openAIQuotaAutoPauseDecision{window: "5h", threshold: pause5h, utilization: utilization5h, reason: "quota_auto_reset_credit_check_5h"} + } + return true, openAIQuotaAutoPauseDecision{window: "7d", threshold: pause7d, utilization: utilization7d, reason: "quota_auto_reset_credit_check_7d"} + } + } // Per-account explicit-disable flags must take precedence over the global default. // Without these, leaving the account threshold blank means "use global default", // so an admin has no way to exempt a single account from auto-pause once a global diff --git a/backend/internal/service/openai_gateway_usage.go b/backend/internal/service/openai_gateway_usage.go index 25af28973a..7489456bb8 100644 --- a/backend/internal/service/openai_gateway_usage.go +++ b/backend/internal/service/openai_gateway_usage.go @@ -1075,7 +1075,9 @@ func (s *OpenAIGatewayService) updateCodexUsageSnapshot(ctx context.Context, acc go func() { updateCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - _ = s.accountRepo.UpdateExtra(updateCtx, accountID, updates) + if err := s.accountRepo.UpdateExtra(updateCtx, accountID, updates); err == nil { + notifyOpenAIAutoReset(accountID) + } }() } diff --git a/backend/internal/service/openai_quota_auto_reset.go b/backend/internal/service/openai_quota_auto_reset.go new file mode 100644 index 0000000000..aafc5991d7 --- /dev/null +++ b/backend/internal/service/openai_quota_auto_reset.go @@ -0,0 +1,809 @@ +package service + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "log/slog" + "net/http" + "sort" + "strings" + "sync" + "time" + + infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" + "github.com/Wei-Shaw/sub2api/internal/pkg/pagination" + "github.com/google/uuid" +) + +const ( + openAIAutoResetScanInterval = time.Minute + openAIAutoResetSnapshotTTL = openAIProbeCacheTTL + openAIAutoResetBatchSize = 100 + openAIAutoResetWorkerCount = 4 + openAIAutoResetQueueCapacity = 1024 + openAIAutoResetAttemptTTL = 8 * 24 * time.Hour + openAIAutoResetLeaderLockKey = "jobs:openai-auto-reset-credit" +) + +const ( + OpenAIAutoResetStatusChecking = "checking" + OpenAIAutoResetStatusAvailable = "available" + OpenAIAutoResetStatusResetting = "resetting" + OpenAIAutoResetStatusSuccess = "success" + OpenAIAutoResetStatusNoCredit = "no_credit" + OpenAIAutoResetStatusFailed = "failed" +) + +// OpenAIAutoResetCreditState 是可返回管理端的脱敏运行态。Attempt* 仅保存不可逆 +// 指纹，用于重启后拒绝切换到另一张卡；不会保存卡 ID 或兑换 ID。 +type OpenAIAutoResetCreditState struct { + Status string `json:"status"` + TriggerWindow string `json:"trigger_window,omitempty"` + AvailableCount int `json:"available_count"` + CheckedAt string `json:"checked_at,omitempty"` + LastResultAt string `json:"last_result_at,omitempty"` + ErrorCode string `json:"error_code,omitempty"` + AttemptCycleHash string `json:"attempt_cycle_hash,omitempty"` + AttemptCreditHash string `json:"attempt_credit_hash,omitempty"` +} + +type openAIAutoResetQuota interface { + QueryUsage(ctx context.Context, accountID int64) (*OpenAIQuotaUsage, error) + CacheResetCreditsSnapshot(ctx context.Context, accountID int64, credits *OpenAIRateLimitResetCredits) error + ResetCreditTargeted(ctx context.Context, accountID int64, creditID, redeemRequestID string) (*OpenAIQuotaResetResult, error) +} + +type openAIAutoResetContextKey struct{} + +func withOpenAIAutoResetContext(ctx context.Context) context.Context { + if ctx == nil { + ctx = context.Background() + } + return context.WithValue(ctx, openAIAutoResetContextKey{}, true) +} + +func isOpenAIAutoResetContext(ctx context.Context) bool { + if ctx == nil { + return false + } + value, _ := ctx.Value(openAIAutoResetContextKey{}).(bool) + return value +} + +type openAIAutoResetRecovery interface { + RecoverAccountState(ctx context.Context, accountID int64, options AccountRecoveryOptions) (*SuccessfulTestRecoveryResult, error) +} + +// OpenAIQuotaAutoResetService 通过小型去重队列承接实时信号，并用分钟扫描补偿 +// 重启、漏事件和多实例读取；真正消费仍由 PostgreSQL 幂等记录串行化。 +type OpenAIQuotaAutoResetService struct { + accountRepo AccountRepository + quota openAIAutoResetQuota + recoverer openAIAutoResetRecovery + idempotency *IdempotencyCoordinator + audit *AuditLogService + settings *SettingService + leaderLock LeaderLockCache + + ctx context.Context + cancel context.CancelFunc + queue chan int64 + pending sync.Map + owner string + start sync.Once + stop sync.Once + wg sync.WaitGroup +} + +func NewOpenAIQuotaAutoResetService( + accountRepo AccountRepository, + quota openAIAutoResetQuota, + recoverer openAIAutoResetRecovery, + idempotency *IdempotencyCoordinator, + audit *AuditLogService, + settings *SettingService, + leaderLock LeaderLockCache, +) *OpenAIQuotaAutoResetService { + ctx, cancel := context.WithCancel(context.Background()) + return &OpenAIQuotaAutoResetService{ + accountRepo: accountRepo, + quota: quota, + recoverer: recoverer, + idempotency: idempotency, + audit: audit, + settings: settings, + leaderLock: leaderLock, + ctx: ctx, + cancel: cancel, + queue: make(chan int64, openAIAutoResetQueueCapacity), + owner: uuid.NewString(), + } +} + +func (s *OpenAIQuotaAutoResetService) Start() { + if s == nil || s.accountRepo == nil || s.quota == nil || s.idempotency == nil { + return + } + s.start.Do(func() { + setOpenAIAutoResetNotifier(s) + for range openAIAutoResetWorkerCount { + s.wg.Add(1) + go s.runWorker() + } + s.wg.Add(1) + go s.runScanner() + }) +} + +func (s *OpenAIQuotaAutoResetService) Stop() { + if s == nil { + return + } + s.stop.Do(func() { + clearOpenAIAutoResetNotifier(s) + s.cancel() + s.wg.Wait() + }) +} + +// Notify 是请求热路径的非阻塞入口。同一账号尚在队列时只保留一个任务；队列 +// 满时丢弃本次信号，分钟扫描仍会补偿，因此不会反向拖慢网关请求。 +func (s *OpenAIQuotaAutoResetService) Notify(accountID int64) { + if s == nil || accountID <= 0 { + return + } + if _, loaded := s.pending.LoadOrStore(accountID, struct{}{}); loaded { + return + } + select { + case <-s.ctx.Done(): + s.pending.Delete(accountID) + case s.queue <- accountID: + default: + s.pending.Delete(accountID) + slog.Warn("openai_auto_reset_queue_full", "account_id", accountID) + } +} + +func (s *OpenAIQuotaAutoResetService) runWorker() { + defer s.wg.Done() + for { + select { + case <-s.ctx.Done(): + return + case accountID := <-s.queue: + ctx, cancel := context.WithTimeout(s.ctx, 50*time.Second) + if err := s.evaluateAccount(ctx, accountID); err != nil && !errors.Is(err, context.Canceled) { + slog.Warn("openai_auto_reset_evaluate_failed", "account_id", accountID, "error_code", infraerrors.Reason(err)) + } + cancel() + s.pending.Delete(accountID) + } + } +} + +func (s *OpenAIQuotaAutoResetService) runScanner() { + defer s.wg.Done() + timer := time.NewTimer(2 * time.Second) + defer timer.Stop() + select { + case <-s.ctx.Done(): + return + case <-timer.C: + s.scanEnabledAccounts(s.ctx) + } + ticker := time.NewTicker(openAIAutoResetScanInterval) + defer ticker.Stop() + for { + select { + case <-s.ctx.Done(): + return + case <-ticker.C: + s.scanEnabledAccounts(s.ctx) + } + } +} + +func (s *OpenAIQuotaAutoResetService) scanEnabledAccounts(ctx context.Context) { + release, scan := s.tryAcquireScanLock(ctx) + if !scan { + return + } + if release != nil { + defer release() + } + for page := 1; ; page++ { + accounts, pageInfo, err := s.accountRepo.ListWithFilters(ctx, pagination.PaginationParams{ + Page: page, PageSize: openAIAutoResetBatchSize, + }, PlatformOpenAI, AccountTypeOAuth, StatusActive, "", 0, "") + if err != nil { + slog.Warn("openai_auto_reset_scan_failed", "page", page, "error", err) + return + } + for i := range accounts { + account := &accounts[i] + if account.Schedulable && ResolveOpenAIAutoResetCreditConfig(account).Enabled { + s.Notify(account.ID) + } + } + if len(accounts) < openAIAutoResetBatchSize || pageInfo == nil || page >= pageInfo.Pages { + return + } + } +} + +// Redis 锁异常时允许重复扫描，避免协调设施故障导致所有实例同时停止补偿； +// 消费唯一性由数据库幂等记录负责，扫描锁只用于削减重复查询。 +func (s *OpenAIQuotaAutoResetService) tryAcquireScanLock(ctx context.Context) (func(), bool) { + if s.leaderLock == nil { + return func() {}, true + } + ok, err := s.leaderLock.TryAcquireLeaderLock(ctx, openAIAutoResetLeaderLockKey, s.owner, 55*time.Second) + if err != nil { + slog.Warn("openai_auto_reset_leader_lock_unavailable", "error", err) + return func() {}, true + } + if !ok { + return nil, false + } + return func() { + releaseCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _ = s.leaderLock.ReleaseLeaderLock(releaseCtx, openAIAutoResetLeaderLockKey, s.owner) + }, true +} + +type openAIAutoResetAssessment struct { + triggerWindow string + resetReached bool + pauseReached bool + utilization5h float64 + utilization7d float64 + threshold5h float64 + threshold7d float64 +} + +func (s *OpenAIQuotaAutoResetService) evaluateAccount(ctx context.Context, accountID int64) error { + ctx = withOpenAIAutoResetContext(ctx) + account, err := s.accountRepo.GetByID(ctx, accountID) + if err != nil || account == nil { + return err + } + if account.IsShadow() { + if account.ParentAccountID != nil { + s.Notify(*account.ParentAccountID) + } + return nil + } + config := ResolveOpenAIAutoResetCreditConfig(account) + if !config.Enabled || !account.IsActive() || !account.Schedulable { + return nil + } + + now := time.Now() + assessment := s.assessExtra(account, config, now) + state := openAIAutoResetStateFromExtra(account.Extra) + needsQuery := openAIAutoResetSnapshotStale(account.Extra, now) || assessment.resetReached + if assessment.pauseReached && !assessment.resetReached { + needsQuery = needsQuery || state == nil || state.Status == OpenAIAutoResetStatusChecking || state.Status == OpenAIAutoResetStatusFailed || openAIAutoResetStateStale(state, now) + } + if !needsQuery { + if !assessment.pauseReached && state != nil && state.TriggerWindow != "" { + state.TriggerWindow = "" + state.ErrorCode = "" + state.CheckedAt = now.UTC().Format(time.RFC3339) + if state.AvailableCount > 0 { + state.Status = OpenAIAutoResetStatusAvailable + } else { + state.Status = OpenAIAutoResetStatusNoCredit + } + return s.persistState(ctx, accountID, state) + } + return nil + } + + checking := &OpenAIAutoResetCreditState{ + Status: OpenAIAutoResetStatusChecking, + TriggerWindow: assessment.triggerWindow, + AvailableCount: stateAvailableCount(state), + CheckedAt: now.UTC().Format(time.RFC3339), + } + copyOpenAIAutoResetAttempt(checking, state) + if err := s.persistState(ctx, accountID, checking); err != nil { + return err + } + + usage, err := s.quota.QueryUsage(ctx, accountID) + if err != nil || usage == nil { + return s.failState(ctx, accountID, checking, "RESET_CREDIT_QUERY_FAILED", err) + } + if err := s.persistFreshUsage(ctx, accountID, usage, now); err != nil { + return s.failState(ctx, accountID, checking, "USAGE_SNAPSHOT_WRITE_FAILED", err) + } + if usage.RateLimitResetCredits == nil { + return s.failState(ctx, accountID, checking, "RESET_CREDIT_DETAILS_UNAVAILABLE", nil) + } + + // 查询期间管理员可能关闭开关；消费前重新读取账号，确保尚未发出的任务可取消。 + account, err = s.accountRepo.GetByID(ctx, accountID) + if err != nil || account == nil { + return err + } + config = ResolveOpenAIAutoResetCreditConfig(account) + if !config.Enabled { + return nil + } + assessment = s.assessUsage(usage, account, config, now) + available := usage.RateLimitResetCredits.AvailableCount + if !assessment.resetReached { + status := OpenAIAutoResetStatusNoCredit + if available > 0 { + status = OpenAIAutoResetStatusAvailable + } + return s.persistState(ctx, accountID, &OpenAIAutoResetCreditState{ + Status: status, + TriggerWindow: assessment.triggerWindow, + AvailableCount: available, + CheckedAt: now.UTC().Format(time.RFC3339), + }) + } + if available <= 0 { + return s.persistState(ctx, accountID, &OpenAIAutoResetCreditState{ + Status: OpenAIAutoResetStatusNoCredit, + TriggerWindow: assessment.triggerWindow, + AvailableCount: 0, + CheckedAt: now.UTC().Format(time.RFC3339), + LastResultAt: now.UTC().Format(time.RFC3339), + ErrorCode: "NO_RESET_CREDIT", + }) + } + + cycleSeed := openAIAutoResetCycleSeed(usage) + cycleHash := shortOpenAIAutoResetHash(cycleSeed) + candidate, selectErr := selectOpenAIAutoResetCandidate(usage.autoResetCandidates, available, state, cycleHash) + if selectErr != nil { + failed := checking + failed.AvailableCount = available + failed.TriggerWindow = assessment.triggerWindow + failed.AttemptCycleHash = cycleHash + return s.failState(ctx, accountID, failed, infraerrors.Reason(selectErr), selectErr) + } + creditHash := shortOpenAIAutoResetHash(candidate.ID) + stableKey := fmt.Sprintf("oarc:%d:%s:%s", accountID, creditHash, cycleHash) + redeemRequestID := uuid.NewSHA1(uuid.NameSpaceURL, []byte(stableKey)).String() + resetting := &OpenAIAutoResetCreditState{ + Status: OpenAIAutoResetStatusResetting, + TriggerWindow: assessment.triggerWindow, + AvailableCount: available, + CheckedAt: now.UTC().Format(time.RFC3339), + AttemptCycleHash: cycleHash, + AttemptCreditHash: creditHash, + } + if err := s.persistState(ctx, accountID, resetting); err != nil { + return err + } + + account, err = s.accountRepo.GetByID(ctx, accountID) + if err != nil || account == nil || !ResolveOpenAIAutoResetCreditConfig(account).Enabled { + return err + } + result, err := s.idempotency.Execute(ctx, IdempotencyExecuteOptions{ + Scope: "openai_auto_reset_credit", + ActorScope: fmt.Sprintf("account:%d", accountID), + Method: http.MethodPost, + Route: "/system/openai/reset-credit/auto", + IdempotencyKey: stableKey, + Payload: map[string]any{ + "account_id": accountID, + "credit_hash": creditHash, + "cycle_hash": cycleHash, + }, + TTL: openAIAutoResetAttemptTTL, + RequireKey: true, + }, func(execCtx context.Context) (any, error) { + resetResult, resetErr := s.quota.ResetCreditTargeted(execCtx, accountID, candidate.ID, redeemRequestID) + if resetErr != nil { + return nil, resetErr + } + if resetResult == nil { + return nil, infraerrors.InternalServer("OPENAI_AUTO_RESET_EMPTY_RESULT", "automatic reset returned an empty result") + } + // 幂等表只保存脱敏结果，避免上游返回的卡 ID 被持久化到响应体列。 + return openAIAutoResetConsumeResult{Code: resetResult.Code, WindowsReset: resetResult.WindowsReset}, nil + }) + if err != nil { + // 另一个实例已持有同一周期的兑换时保持 resetting，等待下一轮读取同一 + // 幂等结果；不能把并发冲突误报成上游消费失败，更不能改选下一张卡。 + reason := infraerrors.Reason(err) + if reason == infraerrors.Reason(ErrIdempotencyInProgress) || reason == infraerrors.Reason(ErrIdempotencyRetryBackoff) { + return nil + } + s.recordAudit(accountID, assessment, available, "failed", 0, infraerrors.Reason(err)) + return s.failState(ctx, accountID, resetting, infraerrors.Reason(err), err) + } + + consumeResult := decodeOpenAIAutoResetConsumeResult(result.Data) + if strings.EqualFold(strings.TrimSpace(consumeResult.Code), "no_credit") { + noCreditAt := time.Now().UTC().Format(time.RFC3339) + noCredit := &OpenAIAutoResetCreditState{ + Status: OpenAIAutoResetStatusNoCredit, + TriggerWindow: assessment.triggerWindow, + AvailableCount: 0, + CheckedAt: noCreditAt, + LastResultAt: noCreditAt, + ErrorCode: "NO_RESET_CREDIT", + AttemptCycleHash: cycleHash, + AttemptCreditHash: creditHash, + } + s.recordAudit(accountID, assessment, available, "no_credit", 0, noCredit.ErrorCode) + return s.persistState(ctx, accountID, noCredit) + } + postCtx, cancelPost := context.WithTimeout(context.WithoutCancel(ctx), 8*time.Second) + post := RunOpenAIQuotaResetPostProcess(postCtx, accountID, s.quota, s.recoverer, s.accountRepo.GetByID) + cancelPost() + if !post.AccountStateRecovered || post.WarningCode != "" { + code := post.WarningCode + if code == "" { + code = OpenAIQuotaResetWarningAccountRecoveryFailed + } + s.recordAudit(accountID, assessment, available, "recovery_failed", consumeResult.WindowsReset, code) + return s.failState(ctx, accountID, resetting, code, nil) + } + + successAt := time.Now().UTC().Format(time.RFC3339) + success := &OpenAIAutoResetCreditState{ + Status: OpenAIAutoResetStatusSuccess, + TriggerWindow: assessment.triggerWindow, + AvailableCount: max(0, available-1), + CheckedAt: successAt, + LastResultAt: successAt, + AttemptCycleHash: cycleHash, + AttemptCreditHash: creditHash, + } + if post.Quota != nil && post.Quota.RateLimitResetCredits != nil { + success.AvailableCount = post.Quota.RateLimitResetCredits.AvailableCount + } + if err := s.persistState(ctx, accountID, success); err != nil { + return err + } + s.recordAudit(accountID, assessment, available, "success", consumeResult.WindowsReset, "") + slog.Info("openai_auto_reset_credit_success", + "account_id", accountID, + "trigger_window", assessment.triggerWindow, + "threshold_5h", assessment.threshold5h, + "threshold_7d", assessment.threshold7d, + "utilization_5h", assessment.utilization5h, + "utilization_7d", assessment.utilization7d, + "windows_reset", consumeResult.WindowsReset, + ) + return nil +} + +type openAIAutoResetConsumeResult struct { + Code string `json:"code"` + WindowsReset int `json:"windows_reset"` +} + +func decodeOpenAIAutoResetConsumeResult(value any) openAIAutoResetConsumeResult { + if typed, ok := value.(openAIAutoResetConsumeResult); ok { + return typed + } + raw, _ := json.Marshal(value) + var decoded openAIAutoResetConsumeResult + _ = json.Unmarshal(raw, &decoded) + return decoded +} + +func (s *OpenAIQuotaAutoResetService) assessExtra(account *Account, config OpenAIAutoResetCreditConfig, now time.Time) openAIAutoResetAssessment { + utilization5h, _ := resolveOpenAIQuotaUtilization(account.Extra, "5h", now) + utilization7d, _ := resolveOpenAIQuotaUtilization(account.Extra, "7d", now) + return s.buildAssessment(account, config, utilization5h, utilization7d) +} + +func (s *OpenAIQuotaAutoResetService) assessUsage(usage *OpenAIQuotaUsage, account *Account, config OpenAIAutoResetCreditConfig, now time.Time) openAIAutoResetAssessment { + updates := buildOpenAIAutoResetUsageUpdates(usage, now) + utilization5h := readOpenAIQuotaUsedPercent(updates, "5h") / 100 + utilization7d := readOpenAIQuotaUsedPercent(updates, "7d") / 100 + return s.buildAssessment(account, config, utilization5h, utilization7d) +} + +func (s *OpenAIQuotaAutoResetService) buildAssessment(account *Account, config OpenAIAutoResetCreditConfig, utilization5h, utilization7d float64) openAIAutoResetAssessment { + assessment := openAIAutoResetAssessment{ + utilization5h: utilization5h, + utilization7d: utilization7d, + threshold5h: config.Threshold5h, + threshold7d: config.Threshold7d, + } + reset5h := utilization5h >= config.Threshold5h + reset7d := utilization7d >= config.Threshold7d + assessment.resetReached = reset5h || reset7d + assessment.triggerWindow = joinOpenAIAutoResetWindows(reset5h, reset7d) + + pause5h, pause7d := resolveOpenAIQuotaAutoPauseThresholds(context.Background(), account) + if s.settings != nil { + pause5h, pause7d = resolveOpenAIQuotaAutoPauseThresholds( + withOpenAIQuotaAutoPauseSettings(context.Background(), s.settings.GetOpenAIQuotaAutoPauseSettings(context.Background())), + account, + ) + } + pauseReached5h := !resolveAccountExtraBool(account.Extra, "auto_pause_5h_disabled") && pause5h > 0 && utilization5h >= pause5h + pauseReached7d := !resolveAccountExtraBool(account.Extra, "auto_pause_7d_disabled") && pause7d > 0 && utilization7d >= pause7d + assessment.pauseReached = pauseReached5h || pauseReached7d || assessment.resetReached + if assessment.triggerWindow == "" { + assessment.triggerWindow = joinOpenAIAutoResetWindows(pauseReached5h, pauseReached7d) + } + return assessment +} + +func joinOpenAIAutoResetWindows(fiveHour, sevenDay bool) string { + switch { + case fiveHour && sevenDay: + return "5h+7d" + case fiveHour: + return "5h" + case sevenDay: + return "7d" + default: + return "" + } +} + +func buildOpenAIAutoResetUsageUpdates(usage *OpenAIQuotaUsage, now time.Time) map[string]any { + if usage == nil || usage.RateLimit == nil { + return nil + } + rateLimit := usage.RateLimit + snapshot := &OpenAICodexUsageSnapshot{UpdatedAt: now.UTC().Format(time.RFC3339)} + applyWindow := func(window *OpenAIRateLimitWindow, primary bool) { + if window == nil { + return + } + used := window.UsedPercent + resetAfter := int(window.ResetAfterSeconds) + windowMinutes := int(window.LimitWindowSeconds / 60) + if primary { + snapshot.PrimaryUsedPercent = &used + snapshot.PrimaryResetAfterSeconds = &resetAfter + snapshot.PrimaryWindowMinutes = &windowMinutes + } else { + snapshot.SecondaryUsedPercent = &used + snapshot.SecondaryResetAfterSeconds = &resetAfter + snapshot.SecondaryWindowMinutes = &windowMinutes + } + } + applyWindow(rateLimit.PrimaryWindow, true) + applyWindow(rateLimit.SecondaryWindow, false) + return buildCodexUsageExtraUpdates(snapshot, now) +} + +func (s *OpenAIQuotaAutoResetService) persistFreshUsage(ctx context.Context, accountID int64, usage *OpenAIQuotaUsage, now time.Time) error { + updates := buildOpenAIAutoResetUsageUpdates(usage, now) + if len(updates) > 0 { + if err := s.accountRepo.UpdateExtra(ctx, accountID, updates); err != nil { + return err + } + } + return s.quota.CacheResetCreditsSnapshot(ctx, accountID, usage.RateLimitResetCredits) +} + +func selectOpenAIAutoResetCandidate(candidates []openAIAutoResetCreditCandidate, available int, previous *OpenAIAutoResetCreditState, cycleHash string) (openAIAutoResetCreditCandidate, error) { + if available <= 0 { + return openAIAutoResetCreditCandidate{}, infraerrors.Conflict("OPENAI_AUTO_RESET_NO_CREDIT", "no reset credit is available") + } + if len(candidates) < available { + return openAIAutoResetCreditCandidate{}, infraerrors.Conflict("OPENAI_AUTO_RESET_CREDIT_DETAILS_INCOMPLETE", "reset credit details are incomplete") + } + for _, candidate := range candidates { + if _, err := time.Parse(time.RFC3339, candidate.ExpiresAt); err != nil { + return openAIAutoResetCreditCandidate{}, infraerrors.Conflict("OPENAI_AUTO_RESET_CREDIT_EXPIRY_INVALID", "reset credit expiration is invalid") + } + } + sorted := append([]openAIAutoResetCreditCandidate(nil), candidates...) + sort.SliceStable(sorted, func(i, j int) bool { + left, leftErr := time.Parse(time.RFC3339, sorted[i].ExpiresAt) + right, rightErr := time.Parse(time.RFC3339, sorted[j].ExpiresAt) + if leftErr != nil { + return false + } + if rightErr != nil { + return true + } + return left.Before(right) + }) + if previous != nil && previous.AttemptCycleHash == cycleHash && previous.AttemptCreditHash != "" { + for _, candidate := range sorted { + if shortOpenAIAutoResetHash(candidate.ID) == previous.AttemptCreditHash { + if strings.TrimSpace(candidate.ID) == "" { + break + } + return candidate, nil + } + } + return openAIAutoResetCreditCandidate{}, infraerrors.Conflict("OPENAI_AUTO_RESET_ORIGINAL_CREDIT_UNAVAILABLE", "the original reset credit cannot be confirmed; refusing to switch credits") + } + if len(sorted) == 0 || strings.TrimSpace(sorted[0].ID) == "" { + return openAIAutoResetCreditCandidate{}, infraerrors.Conflict("OPENAI_AUTO_RESET_CREDIT_ID_MISSING", "the earliest reset credit has no official id") + } + return sorted[0], nil +} + +func openAIAutoResetCycleSeed(usage *OpenAIQuotaUsage) string { + if usage == nil || usage.RateLimit == nil { + return "5h:0|7d:0" + } + var fiveHour, sevenDay int64 + for _, window := range []*OpenAIRateLimitWindow{usage.RateLimit.PrimaryWindow, usage.RateLimit.SecondaryWindow} { + if window == nil { + continue + } + resetAt := window.ResetAt + if resetAt <= 0 { + resetAt = usage.FetchedAt + window.ResetAfterSeconds + } + if window.LimitWindowSeconds <= 6*60*60 { + fiveHour = resetAt + } else { + sevenDay = resetAt + } + } + return fmt.Sprintf("5h:%d|7d:%d", fiveHour, sevenDay) +} + +func shortOpenAIAutoResetHash(value string) string { + sum := sha256.Sum256([]byte(value)) + return hex.EncodeToString(sum[:12]) +} + +func openAIAutoResetSnapshotStale(extra map[string]any, now time.Time) bool { + if len(extra) == 0 { + return true + } + raw, ok := extra["codex_usage_updated_at"] + if !ok { + return true + } + updatedAt, err := parseTime(fmt.Sprint(raw)) + return err != nil || now.Sub(updatedAt) >= openAIAutoResetSnapshotTTL +} + +func openAIAutoResetStateFromExtra(extra map[string]any) *OpenAIAutoResetCreditState { + if len(extra) == 0 { + return nil + } + raw, ok := extra[OpenAIAutoResetCreditStateExtraKey] + if !ok || raw == nil { + return nil + } + encoded, err := json.Marshal(raw) + if err != nil { + return nil + } + var state OpenAIAutoResetCreditState + if err := json.Unmarshal(encoded, &state); err != nil || state.Status == "" { + return nil + } + return &state +} + +func openAIAutoResetStateStale(state *OpenAIAutoResetCreditState, now time.Time) bool { + if state == nil || state.CheckedAt == "" { + return true + } + checkedAt, err := time.Parse(time.RFC3339, state.CheckedAt) + return err != nil || now.Sub(checkedAt) >= openAIAutoResetSnapshotTTL +} + +func stateAvailableCount(state *OpenAIAutoResetCreditState) int { + if state == nil { + return 0 + } + return state.AvailableCount +} + +func copyOpenAIAutoResetAttempt(target, source *OpenAIAutoResetCreditState) { + if target == nil || source == nil { + return + } + target.AttemptCycleHash = source.AttemptCycleHash + target.AttemptCreditHash = source.AttemptCreditHash +} + +func (s *OpenAIQuotaAutoResetService) persistState(ctx context.Context, accountID int64, state *OpenAIAutoResetCreditState) error { + if state == nil { + return nil + } + return s.accountRepo.UpdateExtra(ctx, accountID, map[string]any{OpenAIAutoResetCreditStateExtraKey: state}) +} + +func (s *OpenAIQuotaAutoResetService) failState(ctx context.Context, accountID int64, state *OpenAIAutoResetCreditState, code string, cause error) error { + if state == nil { + state = &OpenAIAutoResetCreditState{} + } + if strings.TrimSpace(code) == "" { + code = "OPENAI_AUTO_RESET_FAILED" + } + state.Status = OpenAIAutoResetStatusFailed + state.ErrorCode = code + state.LastResultAt = time.Now().UTC().Format(time.RFC3339) + if err := s.persistState(ctx, accountID, state); err != nil { + return err + } + slog.Warn("openai_auto_reset_credit_failed", + "account_id", accountID, + "trigger_window", state.TriggerWindow, + "available_count", state.AvailableCount, + "error_code", code, + ) + if cause != nil { + return cause + } + return infraerrors.Conflict(code, "automatic reset credit operation failed") +} + +func (s *OpenAIQuotaAutoResetService) recordAudit(accountID int64, assessment openAIAutoResetAssessment, available int, resultCode string, windowsReset int, errorCode string) { + if s.audit == nil { + return + } + statusCode := http.StatusOK + if resultCode != "success" { + statusCode = http.StatusConflict + } + s.audit.Record(&AuditLog{ + ActorEmail: "system", + ActorRole: "system", + AuthMethod: "system", + Action: "system.openai.reset_credit.auto", + Method: "SYSTEM", + Path: fmt.Sprintf("/system/openai/accounts/%d/auto-reset-credit", accountID), + StatusCode: statusCode, + Extra: map[string]any{ + "account_id": accountID, + "trigger_window": assessment.triggerWindow, + "threshold_5h": assessment.threshold5h, + "threshold_7d": assessment.threshold7d, + "utilization_5h": assessment.utilization5h, + "utilization_7d": assessment.utilization7d, + "available_count": available, + "result_code": resultCode, + "windows_reset": windowsReset, + "error_code": errorCode, + }, + }) +} + +var openAIAutoResetNotifierRegistry struct { + sync.RWMutex + service *OpenAIQuotaAutoResetService +} + +func setOpenAIAutoResetNotifier(service *OpenAIQuotaAutoResetService) { + openAIAutoResetNotifierRegistry.Lock() + openAIAutoResetNotifierRegistry.service = service + openAIAutoResetNotifierRegistry.Unlock() +} + +func clearOpenAIAutoResetNotifier(service *OpenAIQuotaAutoResetService) { + openAIAutoResetNotifierRegistry.Lock() + if openAIAutoResetNotifierRegistry.service == service { + openAIAutoResetNotifierRegistry.service = nil + } + openAIAutoResetNotifierRegistry.Unlock() +} + +func notifyOpenAIAutoReset(accountID int64) { + openAIAutoResetNotifierRegistry.RLock() + service := openAIAutoResetNotifierRegistry.service + openAIAutoResetNotifierRegistry.RUnlock() + if service != nil { + service.Notify(accountID) + } +} + +// NotifyOpenAIAutoResetCredit 供额度查询入口发送轻量信号；不执行同步上游请求。 +func NotifyOpenAIAutoResetCredit(accountID int64) { + notifyOpenAIAutoReset(accountID) +} diff --git a/backend/internal/service/openai_quota_auto_reset_config.go b/backend/internal/service/openai_quota_auto_reset_config.go new file mode 100644 index 0000000000..e891049795 --- /dev/null +++ b/backend/internal/service/openai_quota_auto_reset_config.go @@ -0,0 +1,148 @@ +package service + +import ( + "encoding/json" + "math" + "net/http" + "strconv" + "strings" + + infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" +) + +const ( + OpenAIAutoResetCreditEnabledExtraKey = "auto_reset_credit_enabled" + OpenAIAutoResetCredit5hThresholdExtraKey = "auto_reset_credit_5h_threshold" + OpenAIAutoResetCredit7dThresholdExtraKey = "auto_reset_credit_7d_threshold" + OpenAIAutoResetCreditStateExtraKey = "codex_auto_reset_credit_state" + + openAIAutoResetCreditDefaultThreshold = 1.0 + openAIAutoResetCreditMinimumThreshold = 0.001 +) + +// OpenAIAutoResetCreditConfig 是账号级自动用卡配置。阈值采用 0-1 比例， +// 避免后端调度与前端百分比展示混用同一数值语义。 +type OpenAIAutoResetCreditConfig struct { + Enabled bool + Threshold5h float64 + Threshold7d float64 +} + +// ResolveOpenAIAutoResetCreditConfig 只接受 OpenAI OAuth 母账号；历史账号未配置时 +// 始终保持关闭，防止升级后产生意外消费。 +func ResolveOpenAIAutoResetCreditConfig(account *Account) OpenAIAutoResetCreditConfig { + config := OpenAIAutoResetCreditConfig{ + Threshold5h: openAIAutoResetCreditDefaultThreshold, + Threshold7d: openAIAutoResetCreditDefaultThreshold, + } + if !isOpenAIAutoResetCreditAccount(account) || account.Extra == nil { + return config + } + config.Enabled = resolveAccountExtraBool(account.Extra, OpenAIAutoResetCreditEnabledExtraKey) + if value, ok := resolveAccountExtraNumber(account.Extra, OpenAIAutoResetCredit5hThresholdExtraKey); ok && isValidOpenAIAutoResetThreshold(value) { + config.Threshold5h = value + } + if value, ok := resolveAccountExtraNumber(account.Extra, OpenAIAutoResetCredit7dThresholdExtraKey); ok && isValidOpenAIAutoResetThreshold(value) { + config.Threshold7d = value + } + return config +} + +func isOpenAIAutoResetCreditAccount(account *Account) bool { + return account != nil && account.Platform == PlatformOpenAI && account.Type == AccountTypeOAuth && !account.IsShadow() +} + +// normalizeOpenAIAutoResetCreditExtra 校验管理请求中的配置并剥离服务运行态。 +// enabled=true 时补齐两个 100% 默认阈值；关闭时保留已设置阈值，方便再次开启。 +func normalizeOpenAIAutoResetCreditExtra(platform, accountType string, isShadow bool, extra map[string]any) (map[string]any, error) { + if extra == nil { + return nil, nil + } + normalized := cloneOpenAIAutoResetExtra(extra) + delete(normalized, OpenAIAutoResetCreditStateExtraKey) + + _, hasEnabled := normalized[OpenAIAutoResetCreditEnabledExtraKey] + _, has5h := normalized[OpenAIAutoResetCredit5hThresholdExtraKey] + _, has7d := normalized[OpenAIAutoResetCredit7dThresholdExtraKey] + if !hasEnabled && !has5h && !has7d { + return normalized, nil + } + if platform != PlatformOpenAI || accountType != AccountTypeOAuth || isShadow { + return nil, infraerrors.New(http.StatusBadRequest, "OPENAI_AUTO_RESET_CREDIT_ACCOUNT_INVALID", "automatic reset credits are only supported for OpenAI OAuth parent accounts") + } + + enabled := false + if hasEnabled { + value, ok := normalized[OpenAIAutoResetCreditEnabledExtraKey].(bool) + if !ok { + return nil, infraerrors.New(http.StatusBadRequest, "OPENAI_AUTO_RESET_CREDIT_ENABLED_INVALID", "auto_reset_credit_enabled must be a boolean") + } + enabled = value + } + for key, present := range map[string]bool{ + OpenAIAutoResetCredit5hThresholdExtraKey: has5h, + OpenAIAutoResetCredit7dThresholdExtraKey: has7d, + } { + if !present { + if enabled { + normalized[key] = openAIAutoResetCreditDefaultThreshold + } + continue + } + value, ok := parseOpenAIAutoResetThreshold(normalized[key]) + if !ok || !isValidOpenAIAutoResetThreshold(value) { + return nil, infraerrors.Newf(http.StatusBadRequest, "OPENAI_AUTO_RESET_CREDIT_THRESHOLD_INVALID", "%s must be between 0.001 and 1.0", key) + } + normalized[key] = value + } + return normalized, nil +} + +func stripOpenAIAutoResetCreditManagedExtra(extra map[string]any, stripConfig bool) map[string]any { + if extra == nil { + return nil + } + delete(extra, OpenAIAutoResetCreditStateExtraKey) + if stripConfig { + delete(extra, OpenAIAutoResetCreditEnabledExtraKey) + delete(extra, OpenAIAutoResetCredit5hThresholdExtraKey) + delete(extra, OpenAIAutoResetCredit7dThresholdExtraKey) + } + return extra +} + +func parseOpenAIAutoResetThreshold(value any) (float64, bool) { + switch typed := value.(type) { + case float64: + return typed, true + case float32: + return float64(typed), true + case int: + return float64(typed), true + case int64: + return float64(typed), true + case json.Number: + parsed, err := typed.Float64() + return parsed, err == nil + case string: + parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64) + return parsed, err == nil + default: + return 0, false + } +} + +func isValidOpenAIAutoResetThreshold(value float64) bool { + return !math.IsNaN(value) && !math.IsInf(value, 0) && value >= openAIAutoResetCreditMinimumThreshold && value <= 1 +} + +func cloneOpenAIAutoResetExtra(source map[string]any) map[string]any { + if source == nil { + return nil + } + cloned := make(map[string]any, len(source)) + for key, value := range source { + cloned[key] = value + } + return cloned +} diff --git a/backend/internal/service/openai_quota_auto_reset_test.go b/backend/internal/service/openai_quota_auto_reset_test.go new file mode 100644 index 0000000000..ad2dc965aa --- /dev/null +++ b/backend/internal/service/openai_quota_auto_reset_test.go @@ -0,0 +1,332 @@ +package service + +import ( + "context" + "encoding/json" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestNormalizeOpenAIAutoResetCreditExtra(t *testing.T) { + t.Run("历史账号默认关闭", func(t *testing.T) { + account := &Account{Platform: PlatformOpenAI, Type: AccountTypeOAuth} + config := ResolveOpenAIAutoResetCreditConfig(account) + require.False(t, config.Enabled) + require.Equal(t, 1.0, config.Threshold5h) + require.Equal(t, 1.0, config.Threshold7d) + }) + + t.Run("开启时补齐两个百分百阈值并剥离运行态", func(t *testing.T) { + extra, err := normalizeOpenAIAutoResetCreditExtra(PlatformOpenAI, AccountTypeOAuth, false, map[string]any{ + OpenAIAutoResetCreditEnabledExtraKey: true, + OpenAIAutoResetCreditStateExtraKey: map[string]any{"status": "success"}, + }) + require.NoError(t, err) + require.Equal(t, 1.0, extra[OpenAIAutoResetCredit5hThresholdExtraKey]) + require.Equal(t, 1.0, extra[OpenAIAutoResetCredit7dThresholdExtraKey]) + require.NotContains(t, extra, OpenAIAutoResetCreditStateExtraKey) + }) + + t.Run("阈值和账号类型严格校验", func(t *testing.T) { + _, err := normalizeOpenAIAutoResetCreditExtra(PlatformOpenAI, AccountTypeOAuth, false, map[string]any{ + OpenAIAutoResetCreditEnabledExtraKey: true, + OpenAIAutoResetCredit5hThresholdExtraKey: 0.0009, + }) + require.Error(t, err) + + _, err = normalizeOpenAIAutoResetCreditExtra(PlatformOpenAI, AccountTypeOAuth, true, map[string]any{ + OpenAIAutoResetCreditEnabledExtraKey: true, + }) + require.Error(t, err) + }) +} + +func TestShouldAutoPauseOpenAIAccountByQuota_AutoResetCreditStates(t *testing.T) { + now := time.Now().UTC() + baseExtra := map[string]any{ + OpenAIAutoResetCreditEnabledExtraKey: true, + OpenAIAutoResetCredit5hThresholdExtraKey: 1.0, + OpenAIAutoResetCredit7dThresholdExtraKey: 1.0, + "auto_pause_5h_threshold": 0.8, + "auto_pause_7d_disabled": true, + "codex_5h_used_percent": 90.0, + "codex_usage_updated_at": now.Format(time.RFC3339), + "codex_5h_reset_at": now.Add(time.Hour).Format(time.RFC3339), + } + + t.Run("卡状态未知时暂停并触发异步查询", func(t *testing.T) { + account := &Account{ID: 1, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Extra: cloneOpenAIAutoResetExtra(baseExtra)} + paused, decision := shouldAutoPauseOpenAIAccountByQuota(context.Background(), account) + require.True(t, paused) + require.Equal(t, "quota_auto_reset_credit_check_5h", decision.reason) + }) + + t.Run("明确有卡时允许继续到用卡阈值", func(t *testing.T) { + extra := cloneOpenAIAutoResetExtra(baseExtra) + extra[OpenAIAutoResetCreditStateExtraKey] = OpenAIAutoResetCreditState{ + Status: OpenAIAutoResetStatusAvailable, AvailableCount: 1, CheckedAt: now.Format(time.RFC3339), + } + account := &Account{ID: 2, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Extra: extra} + paused, _ := shouldAutoPauseOpenAIAccountByQuota(context.Background(), account) + require.False(t, paused) + }) + + t.Run("达到用卡阈值后即使有卡也退出调度", func(t *testing.T) { + extra := cloneOpenAIAutoResetExtra(baseExtra) + extra["codex_5h_used_percent"] = 100.0 + extra[OpenAIAutoResetCreditStateExtraKey] = OpenAIAutoResetCreditState{ + Status: OpenAIAutoResetStatusAvailable, AvailableCount: 1, CheckedAt: now.Format(time.RFC3339), + } + account := &Account{ID: 3, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Extra: extra} + paused, decision := shouldAutoPauseOpenAIAccountByQuota(context.Background(), account) + require.True(t, paused) + require.Equal(t, "quota_auto_reset_pending_5h", decision.reason) + }) + + t.Run("自然窗口重置后清除动态阻塞", func(t *testing.T) { + extra := cloneOpenAIAutoResetExtra(baseExtra) + extra["codex_5h_used_percent"] = 100.0 + extra["codex_5h_reset_at"] = now.Add(-time.Second).Format(time.RFC3339) + extra[OpenAIAutoResetCreditStateExtraKey] = OpenAIAutoResetCreditState{ + Status: OpenAIAutoResetStatusFailed, TriggerWindow: "5h", ErrorCode: "RESET_FAILED", + } + account := &Account{ID: 4, Platform: PlatformOpenAI, Type: AccountTypeOAuth, Extra: extra} + paused, _ := shouldAutoPauseOpenAIAccountByQuota(context.Background(), account) + require.False(t, paused) + }) +} + +func TestSelectOpenAIAutoResetCandidate_FailsClosed(t *testing.T) { + candidates := []openAIAutoResetCreditCandidate{ + {ID: "later", ExpiresAt: "2026-09-02T00:00:00Z"}, + {ID: "earlier", ExpiresAt: "2026-09-01T00:00:00Z"}, + } + selected, err := selectOpenAIAutoResetCandidate(candidates, 2, nil, "cycle-a") + require.NoError(t, err) + require.Equal(t, "earlier", selected.ID) + + _, err = selectOpenAIAutoResetCandidate([]openAIAutoResetCreditCandidate{ + {ExpiresAt: "2026-09-01T00:00:00Z"}, + }, 1, nil, "cycle-a") + require.Error(t, err) + + _, err = selectOpenAIAutoResetCandidate(candidates, 2, &OpenAIAutoResetCreditState{ + AttemptCycleHash: "cycle-a", AttemptCreditHash: shortOpenAIAutoResetHash("missing"), + }, "cycle-a") + require.Error(t, err, "模糊结果后原卡消失时不得切换下一张卡") +} + +func TestOpenAIQuotaAutoResetService_AssessesIndependentWindows(t *testing.T) { + service := &OpenAIQuotaAutoResetService{} + account := &Account{Extra: map[string]any{ + "auto_pause_5h_disabled": true, + "auto_pause_7d_disabled": true, + }} + config := OpenAIAutoResetCreditConfig{Enabled: true, Threshold5h: 0.8, Threshold7d: 0.9} + tests := []struct { + name string + fiveHour float64 + sevenDay float64 + wantWindow string + }{ + {name: "5h", fiveHour: 0.8, sevenDay: 0.2, wantWindow: "5h"}, + {name: "7d", fiveHour: 0.2, sevenDay: 0.9, wantWindow: "7d"}, + {name: "同时触发", fiveHour: 0.95, sevenDay: 0.95, wantWindow: "5h+7d"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assessment := service.buildAssessment(account, config, test.fiveHour, test.sevenDay) + require.True(t, assessment.resetReached) + require.Equal(t, test.wantWindow, assessment.triggerWindow) + }) + } +} + +type autoResetTestAccountRepo struct { + AccountRepository + mu sync.Mutex + account *Account +} + +func (r *autoResetTestAccountRepo) GetByID(_ context.Context, id int64) (*Account, error) { + r.mu.Lock() + defer r.mu.Unlock() + copy := *r.account + copy.Extra = cloneOpenAIAutoResetExtra(r.account.Extra) + return ©, nil +} + +func (r *autoResetTestAccountRepo) UpdateExtra(_ context.Context, id int64, updates map[string]any) error { + r.mu.Lock() + defer r.mu.Unlock() + if r.account.Extra == nil { + r.account.Extra = make(map[string]any) + } + for key, value := range updates { + r.account.Extra[key] = value + } + return nil +} + +type autoResetTestQuota struct { + usage *OpenAIQuotaUsage + resetCalls atomic.Int32 + resetEntered chan struct{} + releaseReset chan struct{} + enterOnce sync.Once + mu sync.Mutex + resetArgs [][2]string + failFirst bool +} + +func (q *autoResetTestQuota) QueryUsage(context.Context, int64) (*OpenAIQuotaUsage, error) { + copy := *q.usage + return ©, nil +} + +func (q *autoResetTestQuota) CacheResetCreditsSnapshot(context.Context, int64, *OpenAIRateLimitResetCredits) error { + return nil +} + +func (q *autoResetTestQuota) ResetCreditTargeted(_ context.Context, _ int64, creditID, redeemRequestID string) (*OpenAIQuotaResetResult, error) { + if creditID == "" || redeemRequestID == "" { + panic("targeted reset identifiers must be present") + } + call := q.resetCalls.Add(1) + q.mu.Lock() + q.resetArgs = append(q.resetArgs, [2]string{creditID, redeemRequestID}) + q.mu.Unlock() + if q.failFirst && call == 1 { + return nil, context.DeadlineExceeded + } + if q.resetEntered != nil { + q.enterOnce.Do(func() { close(q.resetEntered) }) + } + if q.releaseReset != nil { + <-q.releaseReset + } + return &OpenAIQuotaResetResult{Code: "ok", WindowsReset: 2}, nil +} + +type autoResetTestRecoverer struct{} + +func (autoResetTestRecoverer) RecoverAccountState(context.Context, int64, AccountRecoveryOptions) (*SuccessfulTestRecoveryResult, error) { + return &SuccessfulTestRecoveryResult{ClearedRateLimit: true}, nil +} + +func TestOpenAIQuotaAutoResetService_ConcurrentInstancesConsumeOnce(t *testing.T) { + now := time.Now().UTC() + account := &Account{ + ID: 99, Platform: PlatformOpenAI, Type: AccountTypeOAuth, + Status: StatusActive, Schedulable: true, + Extra: map[string]any{ + OpenAIAutoResetCreditEnabledExtraKey: true, + OpenAIAutoResetCredit5hThresholdExtraKey: 1.0, + OpenAIAutoResetCredit7dThresholdExtraKey: 1.0, + "codex_5h_used_percent": 100.0, + "codex_7d_used_percent": 10.0, + "codex_usage_updated_at": now.Format(time.RFC3339), + "codex_5h_reset_at": now.Add(time.Hour).Format(time.RFC3339), + "codex_7d_reset_at": now.Add(24 * time.Hour).Format(time.RFC3339), + }, + } + repo := &autoResetTestAccountRepo{account: account} + usage := &OpenAIQuotaUsage{ + FetchedAt: now.Unix(), + RateLimit: &OpenAIRateLimit{ + PrimaryWindow: &OpenAIRateLimitWindow{UsedPercent: 100, LimitWindowSeconds: 5 * 60 * 60, ResetAfterSeconds: 3600, ResetAt: now.Add(time.Hour).Unix()}, + SecondaryWindow: &OpenAIRateLimitWindow{UsedPercent: 10, LimitWindowSeconds: 7 * 24 * 60 * 60, ResetAfterSeconds: 86400, ResetAt: now.Add(24 * time.Hour).Unix()}, + }, + RateLimitResetCredits: &OpenAIRateLimitResetCredits{ + AvailableCount: 1, + Credits: []OpenAIRateLimitResetCreditDetail{{ExpiresAt: now.Add(48 * time.Hour).Format(time.RFC3339)}}, + }, + autoResetCandidates: []openAIAutoResetCreditCandidate{{ID: "credit-sensitive-id", ExpiresAt: now.Add(48 * time.Hour).Format(time.RFC3339)}}, + } + quota := &autoResetTestQuota{usage: usage, resetEntered: make(chan struct{}), releaseReset: make(chan struct{})} + idempotencyRepo := newInMemoryIdempotencyRepo() + config := DefaultIdempotencyConfig() + config.ObserveOnly = false + config.ProcessingTimeout = time.Second + serviceA := NewOpenAIQuotaAutoResetService(repo, quota, autoResetTestRecoverer{}, NewIdempotencyCoordinator(idempotencyRepo, config), nil, nil, nil) + serviceB := NewOpenAIQuotaAutoResetService(repo, quota, autoResetTestRecoverer{}, NewIdempotencyCoordinator(idempotencyRepo, config), nil, nil, nil) + + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + _ = serviceA.evaluateAccount(context.Background(), account.ID) + }() + <-quota.resetEntered + go func() { + defer wg.Done() + _ = serviceB.evaluateAccount(context.Background(), account.ID) + }() + time.Sleep(50 * time.Millisecond) + close(quota.releaseReset) + wg.Wait() + + require.Equal(t, int32(1), quota.resetCalls.Load()) + repo.mu.Lock() + state := openAIAutoResetStateFromExtra(repo.account.Extra) + repo.mu.Unlock() + require.NotNil(t, state) + require.Equal(t, OpenAIAutoResetStatusSuccess, state.Status) + encodedState, err := json.Marshal(state) + require.NoError(t, err) + require.NotContains(t, string(encodedState), "credit-sensitive-id") +} + +func TestOpenAIQuotaAutoResetService_TimeoutRetryReusesRequestBody(t *testing.T) { + now := time.Now().UTC() + account := &Account{ + ID: 100, Platform: PlatformOpenAI, Type: AccountTypeOAuth, + Status: StatusActive, Schedulable: true, + Extra: map[string]any{ + OpenAIAutoResetCreditEnabledExtraKey: true, + OpenAIAutoResetCredit5hThresholdExtraKey: 1.0, + OpenAIAutoResetCredit7dThresholdExtraKey: 1.0, + "codex_5h_used_percent": 100.0, + "codex_usage_updated_at": now.Format(time.RFC3339), + "codex_5h_reset_at": now.Add(time.Hour).Format(time.RFC3339), + }, + } + repo := &autoResetTestAccountRepo{account: account} + expiresAt := now.Add(48 * time.Hour).Format(time.RFC3339) + quota := &autoResetTestQuota{ + failFirst: true, + usage: &OpenAIQuotaUsage{ + FetchedAt: now.Unix(), + RateLimit: &OpenAIRateLimit{ + PrimaryWindow: &OpenAIRateLimitWindow{UsedPercent: 100, LimitWindowSeconds: 5 * 60 * 60, ResetAfterSeconds: 3600, ResetAt: now.Add(time.Hour).Unix()}, + }, + RateLimitResetCredits: &OpenAIRateLimitResetCredits{ + AvailableCount: 1, + Credits: []OpenAIRateLimitResetCreditDetail{{ExpiresAt: expiresAt}}, + }, + autoResetCandidates: []openAIAutoResetCreditCandidate{{ID: "retry-credit", ExpiresAt: expiresAt}}, + }, + } + idempotencyConfig := DefaultIdempotencyConfig() + idempotencyConfig.ObserveOnly = false + idempotencyConfig.FailedRetryBackoff = 0 + service := NewOpenAIQuotaAutoResetService( + repo, + quota, + autoResetTestRecoverer{}, + NewIdempotencyCoordinator(newInMemoryIdempotencyRepo(), idempotencyConfig), + nil, nil, nil, + ) + + require.Error(t, service.evaluateAccount(context.Background(), account.ID)) + require.NoError(t, service.evaluateAccount(context.Background(), account.ID)) + quota.mu.Lock() + args := append([][2]string(nil), quota.resetArgs...) + quota.mu.Unlock() + require.Len(t, args, 2) + require.Equal(t, args[0], args[1], "超时重试必须复用相同 credit_id 与 redeem_request_id") +} diff --git a/backend/internal/service/openai_quota_reset_credits.go b/backend/internal/service/openai_quota_reset_credits.go index 9ff7e70f16..b87e3fd0d1 100644 --- a/backend/internal/service/openai_quota_reset_credits.go +++ b/backend/internal/service/openai_quota_reset_credits.go @@ -8,6 +8,9 @@ import ( ) type openAIRateLimitResetCreditDetailPayload struct { + ID string `json:"id,omitempty"` + CreditID string `json:"credit_id,omitempty"` + CreditIDCamel string `json:"creditId,omitempty"` ExpiresAt string `json:"expires_at,omitempty"` ExpiresAtCamel string `json:"expiresAt,omitempty"` ResetType string `json:"reset_type,omitempty"` @@ -29,6 +32,14 @@ type openAIRateLimitResetCreditDetails struct { AvailableCreditCount int CreditListPresent bool Credits []OpenAIRateLimitResetCreditDetail + AutoResetCandidates []openAIAutoResetCreditCandidate +} + +// openAIAutoResetCreditCandidate 仅在服务内部流转。上游卡 ID 不进入 API DTO、 +// 账号 extra 或日志，避免管理端响应扩大敏感标识暴露面。 +type openAIAutoResetCreditCandidate struct { + ID string + ExpiresAt string } func parseOpenAIRateLimitResetCreditDetails(body []byte) (openAIRateLimitResetCreditDetails, error) { @@ -64,6 +75,7 @@ func parseOpenAIRateLimitResetCreditDetails(body []byte) (openAIRateLimitResetCr } credits := make([]OpenAIRateLimitResetCreditDetail, 0, len(rawCredits)) + autoResetCandidates := make([]openAIAutoResetCreditCandidate, 0, len(rawCredits)) availableCreditCount := 0 for _, raw := range rawCredits { if raw == nil { @@ -88,12 +100,24 @@ func parseOpenAIRateLimitResetCreditDetails(body []byte) (openAIRateLimitResetCr continue } credits = append(credits, OpenAIRateLimitResetCreditDetail{ExpiresAt: expiresAt}) + creditID := strings.TrimSpace(raw.ID) + if creditID == "" { + creditID = strings.TrimSpace(raw.CreditID) + } + if creditID == "" { + creditID = strings.TrimSpace(raw.CreditIDCamel) + } + autoResetCandidates = append(autoResetCandidates, openAIAutoResetCreditCandidate{ + ID: creditID, + ExpiresAt: expiresAt, + }) } return openAIRateLimitResetCreditDetails{ AvailableCount: availableCount, AvailableCreditCount: availableCreditCount, CreditListPresent: creditListPresent, Credits: credits, + AutoResetCandidates: autoResetCandidates, }, nil } diff --git a/backend/internal/service/openai_quota_reset_credits_test.go b/backend/internal/service/openai_quota_reset_credits_test.go index 34f87f9a2f..2deb01ed89 100644 --- a/backend/internal/service/openai_quota_reset_credits_test.go +++ b/backend/internal/service/openai_quota_reset_credits_test.go @@ -14,8 +14,8 @@ func TestParseOpenAIRateLimitResetCreditDetails_PreservesAvailableCreditOrder(t "availableCount":"2", "credits":[ {"reset_type":"codex_rate_limits","status":"redeemed","expires_at":"2026-07-01T04:05:06Z"}, - {"reset_type":"codex_rate_limits","status":"available","expires_at":"2026-07-04T04:05:06Z"}, - {"resetType":"codex_rate_limits","status":"available","expiresAt":"2026-07-03T04:05:06Z"}, + {"id":"credit-later","reset_type":"codex_rate_limits","status":"available","expires_at":"2026-07-04T04:05:06Z"}, + {"creditId":"credit-earlier","resetType":"codex_rate_limits","status":"available","expiresAt":"2026-07-03T04:05:06Z"}, {"reset_type":"other","status":"available","expires_at":"2026-07-02T04:05:06Z"} ] }`) @@ -28,6 +28,10 @@ func TestParseOpenAIRateLimitResetCreditDetails_PreservesAvailableCreditOrder(t {ExpiresAt: "2026-07-04T04:05:06Z"}, {ExpiresAt: "2026-07-03T04:05:06Z"}, }, details.Credits) + require.Equal(t, []openAIAutoResetCreditCandidate{ + {ID: "credit-later", ExpiresAt: "2026-07-04T04:05:06Z"}, + {ID: "credit-earlier", ExpiresAt: "2026-07-03T04:05:06Z"}, + }, details.AutoResetCandidates) } func TestQueryUsageResetCreditCountPrecedence(t *testing.T) { diff --git a/backend/internal/service/openai_quota_reset_workflow.go b/backend/internal/service/openai_quota_reset_workflow.go new file mode 100644 index 0000000000..430b29618b --- /dev/null +++ b/backend/internal/service/openai_quota_reset_workflow.go @@ -0,0 +1,86 @@ +package service + +import ( + "context" + "log/slog" + + infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" +) + +const ( + OpenAIQuotaResetWarningCacheRefreshFailed = "reset_credit_cache_refresh_failed" + OpenAIQuotaResetWarningAccountRecoveryFailed = "account_state_recovery_failed" + OpenAIQuotaResetWarningAccountRefreshFailed = "account_state_refresh_failed" +) + +type openAIQuotaResetWorkflowQuota interface { + QueryUsage(ctx context.Context, accountID int64) (*OpenAIQuotaUsage, error) + CacheResetCreditsSnapshot(ctx context.Context, accountID int64, credits *OpenAIRateLimitResetCredits) error +} + +type openAIQuotaResetWorkflowRecoverer interface { + RecoverAccountState(ctx context.Context, accountID int64, options AccountRecoveryOptions) (*SuccessfulTestRecoveryResult, error) +} + +// OpenAIQuotaResetPostProcessResult 汇总手动和自动用卡后的共享恢复结果。 +// WarningCode 允许上游消费成功但本地恢复部分失败时仍然准确呈现状态。 +type OpenAIQuotaResetPostProcessResult struct { + Quota *OpenAIQuotaUsage + Account *Account + CacheRefreshed bool + AccountStateRecovered bool + WarningCode string +} + +// RunOpenAIQuotaResetPostProcess 按“解除限流、刷新额度缓存、刷新账号行”的固定顺序 +// 执行消费后恢复，避免手动和自动入口在失败语义上逐渐分叉。 +func RunOpenAIQuotaResetPostProcess( + ctx context.Context, + accountID int64, + quota openAIQuotaResetWorkflowQuota, + recoverer openAIQuotaResetWorkflowRecoverer, + loadAccount func(context.Context, int64) (*Account, error), +) OpenAIQuotaResetPostProcessResult { + result := OpenAIQuotaResetPostProcessResult{} + if recoverer == nil { + result.WarningCode = OpenAIQuotaResetWarningAccountRecoveryFailed + return result + } + if _, err := recoverer.RecoverAccountState(ctx, accountID, AccountRecoveryOptions{InvalidateToken: true}); err != nil { + slog.Warn("openai_quota_reset_account_recovery_failed", "account_id", accountID, "error_code", infraerrors.Reason(err)) + result.WarningCode = OpenAIQuotaResetWarningAccountRecoveryFailed + return result + } + result.AccountStateRecovered = true + + if quota != nil { + usage, usageErr := quota.QueryUsage(ctx, accountID) + switch { + case usageErr != nil || usage == nil: + slog.Warn("openai_quota_reset_cache_refresh_failed", "account_id", accountID, "error_code", infraerrors.Reason(usageErr)) + result.WarningCode = OpenAIQuotaResetWarningCacheRefreshFailed + default: + if err := quota.CacheResetCreditsSnapshot(ctx, accountID, usage.RateLimitResetCredits); err != nil { + slog.Warn("openai_quota_reset_cache_refresh_failed", "account_id", accountID, "error_code", infraerrors.Reason(err)) + result.WarningCode = OpenAIQuotaResetWarningCacheRefreshFailed + } else { + result.Quota = usage + result.CacheRefreshed = true + } + } + } + + if loadAccount == nil { + return result + } + account, err := loadAccount(ctx, accountID) + if err != nil { + slog.Warn("openai_quota_reset_account_refresh_failed", "account_id", accountID, "error_code", infraerrors.Reason(err)) + if result.WarningCode == "" { + result.WarningCode = OpenAIQuotaResetWarningAccountRefreshFailed + } + return result + } + result.Account = account + return result +} diff --git a/backend/internal/service/openai_quota_service.go b/backend/internal/service/openai_quota_service.go index 58bf4659f5..6e40c497fc 100644 --- a/backend/internal/service/openai_quota_service.go +++ b/backend/internal/service/openai_quota_service.go @@ -87,6 +87,7 @@ type OpenAIQuotaUsage struct { AdditionalRateLimits []OpenAIAdditionalRateLimit `json:"additional_rate_limits,omitempty"` RateLimitResetCredits *OpenAIRateLimitResetCredits `json:"rate_limit_reset_credits,omitempty"` FetchedAt int64 `json:"fetched_at"` + autoResetCandidates []openAIAutoResetCreditCandidate } // OpenAIQuotaResetCredit captures the redeemed credit metadata returned by the @@ -180,6 +181,10 @@ func (s *OpenAIQuotaService) QueryUsage(ctx context.Context, accountID int64) (* continue } status := resp.StatusCode + if isOpenAIAutoResetContext(ctx) { + slog.Warn("openai_quota_query_failed", "account_id", accountID, "status", status, "source", "auto_reset") + return nil, infraerrors.Newf(mapUpstreamStatus(status), "OPENAI_QUOTA_UPSTREAM_ERROR", "upstream returned %d", status) + } body := truncate(s.redactQuotaErrorBody(ctx, accountID, resp.String()), 240) slog.Warn("openai_quota_query_failed", "account_id", accountID, "status", status, "body", body) return nil, infraerrors.Newf(mapUpstreamStatus(status), "OPENAI_QUOTA_UPSTREAM_ERROR", "upstream returned %d: %s", status, body) @@ -190,6 +195,7 @@ func (s *OpenAIQuotaService) QueryUsage(ctx context.Context, accountID int64) (* payload.FetchedAt = time.Now().Unix() details := s.queryResetCreditDetails(callCtx, client, accessToken, chatGPTAccountID, fedRAMP, accountID) if details != nil { + payload.autoResetCandidates = details.AutoResetCandidates hasDetailCount := details.AvailableCount != nil if payload.RateLimitResetCredits == nil { payload.RateLimitResetCredits = &OpenAIRateLimitResetCredits{} @@ -274,6 +280,25 @@ func (s *OpenAIQuotaService) queryResetCreditDetails(ctx context.Context, client // The redeem_request_id is auto-generated (uuid-like) — upstream uses it for // idempotency. Returns the consumed credit metadata so the UI can refresh. func (s *OpenAIQuotaService) ResetCredit(ctx context.Context, accountID int64) (*OpenAIQuotaResetResult, error) { + redeemRequestID, err := generateRedeemRequestID() + if err != nil { + return nil, infraerrors.Newf(http.StatusInternalServerError, "OPENAI_QUOTA_REDEEM_ID_FAILED", "failed to generate redeem id: %v", err) + } + return s.resetCredit(ctx, accountID, "", redeemRequestID, false) +} + +// ResetCreditTargeted 使用固定卡 ID 与兑换 ID执行自动消费。调用方必须在重试时 +// 复用同一组参数；本方法不会回退到不带 credit_id 的旧消费方式。 +func (s *OpenAIQuotaService) ResetCreditTargeted(ctx context.Context, accountID int64, creditID, redeemRequestID string) (*OpenAIQuotaResetResult, error) { + creditID = strings.TrimSpace(creditID) + redeemRequestID = strings.TrimSpace(redeemRequestID) + if creditID == "" || redeemRequestID == "" { + return nil, infraerrors.New(http.StatusBadRequest, "OPENAI_QUOTA_TARGETED_RESET_INVALID", "credit_id and redeem_request_id are required") + } + return s.resetCredit(ctx, accountID, creditID, redeemRequestID, true) +} + +func (s *OpenAIQuotaService) resetCredit(ctx context.Context, accountID int64, creditID, redeemRequestID string, targeted bool) (*OpenAIQuotaResetResult, error) { // Shadow guard: resetting credits via a shadow account would silently // operate on the parent's quota; that is surprising and unwanted. Callers // must reset the parent account directly. @@ -297,11 +322,6 @@ func (s *OpenAIQuotaService) ResetCredit(ctx context.Context, accountID int64) ( return nil, err } - redeemRequestID, err := generateRedeemRequestID() - if err != nil { - return nil, infraerrors.Newf(http.StatusInternalServerError, "OPENAI_QUOTA_REDEEM_ID_FAILED", "failed to generate redeem id: %v", err) - } - client, err := s.privacyClientFactory(proxyURL) if err != nil { return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_QUOTA_CLIENT_ERROR", "failed to build upstream client: %v", err) @@ -318,10 +338,14 @@ func (s *OpenAIQuotaService) ResetCredit(ctx context.Context, accountID int64) ( return nil, infraerrors.Newf(http.StatusBadGateway, "OPENAI_QUOTA_AUTH_FAILED", "failed to build upstream authentication: %v", headerErr) } headers["content-type"] = "application/json" + body := map[string]string{"redeem_request_id": redeemRequestID} + if targeted { + body["credit_id"] = creditID + } resp, err := client.R(). SetContext(callCtx). SetHeaders(headers). - SetBody(map[string]string{"redeem_request_id": redeemRequestID}). + SetBody(body). SetSuccessResult(&payload). Post(chatGPTRateLimitResetURL) if err != nil { @@ -336,6 +360,10 @@ func (s *OpenAIQuotaService) ResetCredit(ctx context.Context, accountID int64) ( continue } status := resp.StatusCode + if targeted { + slog.Warn("openai_quota_targeted_reset_failed", "account_id", accountID, "status", status) + return nil, infraerrors.Newf(mapUpstreamStatus(status), "OPENAI_QUOTA_RESET_UPSTREAM_ERROR", "upstream returned %d", status) + } body := truncate(s.redactQuotaErrorBody(callCtx, accountID, resp.String()), 240) slog.Warn("openai_quota_reset_failed", "account_id", accountID, "status", status, "body", body) return nil, infraerrors.Newf(mapUpstreamStatus(status), "OPENAI_QUOTA_RESET_UPSTREAM_ERROR", "upstream returned %d: %s", status, body) diff --git a/backend/internal/service/openai_quota_spark_window_test.go b/backend/internal/service/openai_quota_spark_window_test.go index a351381cb4..cb47a0c0c5 100644 --- a/backend/internal/service/openai_quota_spark_window_test.go +++ b/backend/internal/service/openai_quota_spark_window_test.go @@ -189,6 +189,32 @@ func TestResetCreditShadowRejected(t *testing.T) { "shadow ResetCredit 应映射为 409 Conflict 而非 500") } +func TestResetCreditTargetedSendsStableCreditAndRedeemIDs(t *testing.T) { + account := &Account{ + ID: 203, Platform: PlatformOpenAI, Type: AccountTypeOAuth, + Credentials: map[string]any{"chatgpt_account_id": "account-targeted"}, + } + repo := &stubQuotaAccountRepo{accounts: map[int64]*Account{account.ID: account}} + tokenCache := &stubQuotaTokenCache{tokens: map[string]string{OpenAITokenCacheKey(account): "fake-token"}} + tokenProvider := NewOpenAITokenProvider(repo, tokenCache, nil) + var body map[string]string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.Equal(t, "/backend-api/wham/rate-limit-reset-credits/consume", r.URL.Path) + require.NoError(t, json.NewDecoder(r.Body).Decode(&body)) + w.Header().Set("content-type", "application/json") + _, _ = w.Write([]byte(`{"code":"ok","windows_reset":1}`)) + })) + defer srv.Close() + + svc := NewOpenAIQuotaService(repo, nil, tokenProvider, newQuotaRedirectingFactory(srv)) + result, err := svc.ResetCreditTargeted(context.Background(), account.ID, "credit-123", "redeem-456") + require.NoError(t, err) + require.Equal(t, "ok", result.Code) + require.Equal(t, map[string]string{ + "credit_id": "credit-123", "redeem_request_id": "redeem-456", + }, body) +} + func TestResetCreditAgentIdentityUsesAssertionAndRecoversInvalidTaskOnce(t *testing.T) { _, privateKey, err := ed25519.GenerateKey(rand.Reader) require.NoError(t, err) diff --git a/backend/internal/service/ratelimit_service.go b/backend/internal/service/ratelimit_service.go index 2c42b810b6..f5ded9ccf0 100644 --- a/backend/internal/service/ratelimit_service.go +++ b/backend/internal/service/ratelimit_service.go @@ -1092,6 +1092,9 @@ func (s *RateLimitService) handle429(ctx context.Context, account *Account, head // 单影子场景直接变成无可用账号(外审第8轮 P1)。整段跳过;影子的 codex_* 仅由 account_usage 的 // QueryUsage→persistOpenAICodexProbeSnapshot 维护,枯竭由调度守卫处理。 if account.IsShadow() { + if account.ParentAccountID != nil { + notifyOpenAIAutoReset(*account.ParentAccountID) + } return } // 国产供应商（kimi/zhipu/deepseek）的 429 走专用可恢复路径：余额不足 → 临时停调， @@ -1105,6 +1108,7 @@ func (s *RateLimitService) handle429(ctx context.Context, account *Account, head if account.Platform == PlatformOpenAI { persistOpenAI429PlanType(ctx, s.accountRepo, account, responseBody) s.persistOpenAICodexSnapshot(ctx, account, headers) + notifyOpenAIAutoReset(account.ID) if resetAt := s.calculateOpenAI429ResetTime(headers); resetAt != nil { s.notifyAccountSchedulingBlocked(account, *resetAt, "429") if err := s.accountRepo.SetRateLimited(ctx, account.ID, *resetAt); err != nil { @@ -1620,7 +1624,9 @@ func (s *RateLimitService) persistOpenAICodexSnapshot(ctx context.Context, accou } if err := s.accountRepo.UpdateExtra(ctx, account.ID, updates); err != nil { slog.Warn("openai_codex_snapshot_persist_failed", "account_id", account.ID, "error", err) + return } + notifyOpenAIAutoReset(account.ID) } // parseOpenAIRateLimitResetTime 解析 OpenAI 格式的 429 响应，返回重置时间的 Unix 时间戳 diff --git a/backend/internal/service/wire.go b/backend/internal/service/wire.go index 17a68799c4..6e708de1d9 100644 --- a/backend/internal/service/wire.go +++ b/backend/internal/service/wire.go @@ -189,6 +189,29 @@ func ProvideOpenAIQuotaService( return service } +// ProvideOpenAIQuotaAutoResetService 启动账号级自动用卡队列与补偿扫描。 +func ProvideOpenAIQuotaAutoResetService( + accountRepo AccountRepository, + quotaService *OpenAIQuotaService, + rateLimitService *RateLimitService, + idempotency *IdempotencyCoordinator, + audit *AuditLogService, + settingService *SettingService, + leaderLock LeaderLockCache, +) *OpenAIQuotaAutoResetService { + service := NewOpenAIQuotaAutoResetService( + accountRepo, + quotaService, + rateLimitService, + idempotency, + audit, + settingService, + leaderLock, + ) + service.Start() + return service +} + func ProvideAccountUsageService( accountRepo AccountRepository, usageLogRepo UsageLogRepository, @@ -842,6 +865,7 @@ var ProviderSet = wire.NewSet( ProvideGrokTokenProvider, ProvideOpenAITokenProvider, ProvideOpenAIQuotaService, + ProvideOpenAIQuotaAutoResetService, ProvideGrokQuotaService, ProvideCNProviderQuotaService, ProvideCNProviderBalanceService, diff --git a/frontend/src/components/account/EditAccountModal.vue b/frontend/src/components/account/EditAccountModal.vue index d1cb92df59..e93d4726e1 100644 --- a/frontend/src/components/account/EditAccountModal.vue +++ b/frontend/src/components/account/EditAccountModal.vue @@ -2266,6 +2266,66 @@ +
+
+
+ +

+ {{ t('admin.accounts.autoResetCredit.hint') }} +

+
+ +
+
+
+ + +
+
+ + +
+
+

{{ t('admin.accounts.autoResetCredit.thresholdHint') }}

+
+
(null) const autoPause7dThreshold = ref(null) const autoPause5hDisabled = ref(false) const autoPause7dDisabled = ref(false) +const autoResetCreditEnabled = ref(false) +const autoResetCredit5hThreshold = ref(100) +const autoResetCredit7dThreshold = ref(100) const upstreamBillingAutoProbeEnabled = ref(false) const upstreamBillingRateSyncEnabled = ref(false) const mixedScheduling = ref(false) // For antigravity accounts: enable mixed scheduling @@ -3626,6 +3689,11 @@ const syncFormFromAccount = (newAccount: Account | null) => { autoPause7dThreshold.value = typeof extra?.auto_pause_7d_threshold === 'number' ? extra.auto_pause_7d_threshold * 100 : null autoPause5hDisabled.value = extra?.auto_pause_5h_disabled === true autoPause7dDisabled.value = extra?.auto_pause_7d_disabled === true + autoResetCreditEnabled.value = extra?.auto_reset_credit_enabled === true + autoResetCredit5hThreshold.value = + typeof extra?.auto_reset_credit_5h_threshold === 'number' ? extra.auto_reset_credit_5h_threshold * 100 : 100 + autoResetCredit7dThreshold.value = + typeof extra?.auto_reset_credit_7d_threshold === 'number' ? extra.auto_reset_credit_7d_threshold * 100 : 100 upstreamBillingAutoProbeEnabled.value = extra?.upstream_billing_probe_enabled === true upstreamBillingRateSyncEnabled.value = upstreamBillingAutoProbeEnabled.value && extra?.upstream_billing_rate_sync_enabled === true @@ -4518,6 +4586,13 @@ const handleSubmit = async () => { appStore.showError(t('admin.accounts.pleaseSelectStatus')) return } + if (autoResetCreditEnabled.value) { + const thresholds = [autoResetCredit5hThreshold.value, autoResetCredit7dThreshold.value] + if (thresholds.some((value) => !Number.isFinite(value) || value < 0.1 || value > 100)) { + appStore.showError(t('admin.accounts.autoResetCredit.thresholdInvalid')) + return + } + } const updatePayload: Record = { ...form } try { @@ -5082,6 +5157,13 @@ const handleSubmit = async () => { } else { delete newExtra.auto_pause_7d_disabled } + if (props.account.type === 'oauth' && !isSparkShadow.value) { + newExtra.auto_reset_credit_enabled = autoResetCreditEnabled.value + newExtra.auto_reset_credit_5h_threshold = autoResetCredit5hThreshold.value / 100 + newExtra.auto_reset_credit_7d_threshold = autoResetCredit7dThreshold.value / 100 + } + // 运行态只允许后端服务更新，账号编辑不得回写旧状态。 + delete newExtra.codex_auto_reset_credit_state delete newExtra.codex_image_generation_bridge_enabled switch (codexImageToolMode.value) { diff --git a/frontend/src/components/account/OpenAIQuotaResetCell.vue b/frontend/src/components/account/OpenAIQuotaResetCell.vue index cb81bcfd9d..6a3225b8bf 100644 --- a/frontend/src/components/account/OpenAIQuotaResetCell.vue +++ b/frontend/src/components/account/OpenAIQuotaResetCell.vue @@ -63,6 +63,32 @@
+
+ + {{ autoResetStateLabel }} + + {{ autoResetState.trigger_window }} + + + + {{ formatResetCreditExpiry(autoResetState.checked_at, 'short') }} + + + {{ autoResetState.error_code }} + +
+
(null) const showResetConfirm = ref(false) const showResetCreditDetails = ref(false) +type AutoResetCreditState = NonNullable['codex_auto_reset_credit_state']> +const validAutoResetStatuses = new Set(['checking', 'available', 'resetting', 'success', 'no_credit', 'failed']) +const autoResetState = computed(() => { + if (props.account.extra?.auto_reset_credit_enabled !== true) return null + const state = props.account.extra?.codex_auto_reset_credit_state + if (!state || typeof state !== 'object' || !validAutoResetStatuses.has(String(state.status))) return null + return state +}) +const autoResetStateLabel = computed(() => { + if (!autoResetState.value?.status) return '' + const keyByStatus: Record = { + checking: 'checking', + available: 'available', + resetting: 'resetting', + success: 'success', + no_credit: 'noCredit', + failed: 'failed' + } + return t(`admin.accounts.openaiQuotaReset.autoStatus.${keyByStatus[autoResetState.value.status]}`) +}) +const autoResetStateClass = computed(() => { + switch (autoResetState.value?.status) { + case 'available': + return 'bg-blue-50 text-blue-700 dark:bg-blue-900/30 dark:text-blue-300' + case 'success': + return 'bg-emerald-50 text-emerald-700 dark:bg-emerald-900/30 dark:text-emerald-300' + case 'no_credit': + case 'failed': + return 'bg-red-50 text-red-700 dark:bg-red-900/30 dark:text-red-300' + case 'resetting': + return 'bg-orange-50 text-orange-700 dark:bg-orange-900/30 dark:text-orange-300' + default: + return 'bg-gray-100 text-gray-600 dark:bg-dark-800 dark:text-gray-300' + } +}) + // Rehydrate the card from the persisted snapshot. Credits that already expired // are dropped and the count is clamped to what remains: the snapshot has no // freshness signal, so an unfiltered read would offer to consume credits that no diff --git a/frontend/src/components/account/__tests__/EditAccountModal.spec.ts b/frontend/src/components/account/__tests__/EditAccountModal.spec.ts index acd0436f6a..eceb670c92 100644 --- a/frontend/src/components/account/__tests__/EditAccountModal.spec.ts +++ b/frontend/src/components/account/__tests__/EditAccountModal.spec.ts @@ -290,6 +290,18 @@ function buildOpenAISetupTokenAccount() { } as any } +function buildOpenAIOAuthParentAccount() { + return { + ...buildAccount(), + id: 7, + name: 'OpenAI OAuth Parent', + type: 'oauth', + parent_account_id: null, + credentials: { access_token: 'oauth-token' }, + extra: {} + } as any +} + function mountModal(account = buildAccount()) { return mount(EditAccountModal, { props: { @@ -1351,3 +1363,63 @@ describe('EditAccountModal', () => { ) }) }) + +describe('EditAccountModal OpenAI 自动使用重置卡', () => { + beforeEach(() => { + authIsSimpleMode.value = true + updateAccountMock.mockReset() + checkMixedChannelRiskMock.mockReset().mockResolvedValue({ has_risk: false }) + }) + + it('仅对 OpenAI OAuth 母账号显示，默认关闭且阈值为 100/100', () => { + const parent = mountModal(buildOpenAIOAuthParentAccount()) + expect(parent.find('[data-testid="auto-reset-credit-settings"]').exists()).toBe(true) + expect((parent.get('[data-testid="auto-reset-credit-5h-threshold"]').element as HTMLInputElement).value).toBe('100') + expect((parent.get('[data-testid="auto-reset-credit-7d-threshold"]').element as HTMLInputElement).value).toBe('100') + expect(parent.get('[data-testid="auto-reset-credit-5h-threshold"]').attributes('disabled')).toBeDefined() + parent.unmount() + + for (const account of [buildAccount(), buildOpenAISetupTokenAccount(), buildOpenAISparkShadowAccount()]) { + const wrapper = mountModal(account) + expect(wrapper.find('[data-testid="auto-reset-credit-settings"]').exists()).toBe(false) + wrapper.unmount() + } + }) + + it('独立保存两个阈值，并禁止把运行态回写到管理请求', async () => { + const account = buildOpenAIOAuthParentAccount() + account.extra = { + codex_auto_reset_credit_state: { + status: 'success', + trigger_window: '5h', + available_count: 1 + } + } + updateAccountMock.mockResolvedValue(account) + const wrapper = mountModal(account) + + await wrapper.get('[data-testid="auto-reset-credit-enabled"]').trigger('click') + await wrapper.get('[data-testid="auto-reset-credit-5h-threshold"]').setValue('75.5') + await wrapper.get('[data-testid="auto-reset-credit-7d-threshold"]').setValue('92') + await wrapper.get('form#edit-account-form').trigger('submit.prevent') + + expect(updateAccountMock).toHaveBeenCalledTimes(1) + const extra = updateAccountMock.mock.calls[0]?.[1]?.extra + expect(extra).toMatchObject({ + auto_reset_credit_enabled: true, + auto_reset_credit_5h_threshold: 0.755, + auto_reset_credit_7d_threshold: 0.92 + }) + expect(extra).not.toHaveProperty('codex_auto_reset_credit_state') + wrapper.unmount() + }) + + it('开启后拒绝超出 0.1–100 范围的任一阈值', async () => { + const wrapper = mountModal(buildOpenAIOAuthParentAccount()) + await wrapper.get('[data-testid="auto-reset-credit-enabled"]').trigger('click') + await wrapper.get('[data-testid="auto-reset-credit-5h-threshold"]').setValue('0') + await wrapper.get('form#edit-account-form').trigger('submit.prevent') + expect(updateAccountMock).not.toHaveBeenCalled() + wrapper.unmount() + }) +}) diff --git a/frontend/src/components/account/__tests__/OpenAIQuotaResetCell.spark_shadow.spec.ts b/frontend/src/components/account/__tests__/OpenAIQuotaResetCell.spark_shadow.spec.ts index f888cbee44..280725bc5b 100644 --- a/frontend/src/components/account/__tests__/OpenAIQuotaResetCell.spark_shadow.spec.ts +++ b/frontend/src/components/account/__tests__/OpenAIQuotaResetCell.spark_shadow.spec.ts @@ -344,3 +344,45 @@ describe('OpenAIQuotaResetCell — 外审 F6:影子禁用重置', () => { wrapper.unmount() }) }) + +describe('OpenAIQuotaResetCell 自动用卡运行态', () => { + it.each([ + ['checking', 'checking'], + ['available', 'available'], + ['resetting', 'resetting'], + ['success', 'success'], + ['no_credit', 'noCredit'], + ['failed', 'failed'], + ] as const)('展示 %s 状态且不需要暴露卡标识', (status, labelKey) => { + const account = makeAccount({ + extra: { + auto_reset_credit_enabled: true, + codex_auto_reset_credit_state: { + status, + trigger_window: '5h', + available_count: 1, + checked_at: '2099-07-03T04:05:06Z', + error_code: status === 'failed' ? 'RESET_FAILED' : undefined, + }, + }, + }) + const wrapper = mount(OpenAIQuotaResetCell, { props: { account } }) + const state = wrapper.get('[data-testid="auto-reset-credit-state"]') + expect(state.text()).toContain(`admin.accounts.openaiQuotaReset.autoStatus.${labelKey}`) + expect(state.text()).toContain('5h') + expect(state.text()).not.toContain('credit_id') + wrapper.unmount() + }) + + it('开关关闭时不显示历史运行态', () => { + const account = makeAccount({ + extra: { + auto_reset_credit_enabled: false, + codex_auto_reset_credit_state: { status: 'success', available_count: 1 }, + }, + }) + const wrapper = mount(OpenAIQuotaResetCell, { props: { account } }) + expect(wrapper.find('[data-testid="auto-reset-credit-state"]').exists()).toBe(false) + wrapper.unmount() + }) +}) diff --git a/frontend/src/i18n/locales/en/admin/accounts.ts b/frontend/src/i18n/locales/en/admin/accounts.ts index 3c126b78b0..9a2b1739b0 100644 --- a/frontend/src/i18n/locales/en/admin/accounts.ts +++ b/frontend/src/i18n/locales/en/admin/accounts.ts @@ -821,6 +821,14 @@ export default { autoPause5hDisabled: 'Disable 5h auto-pause', autoPause7dDisabled: 'Disable 7d auto-pause', autoPauseDisabledHint: 'When enabled, this account is never auto-paused (even if a global default threshold is configured).', + autoResetCredit: { + title: 'Automatically use reset credits', + hint: 'Uses the earliest-expiring available credit only when actual usage reaches a threshold. Off by default; the account remains paused if no credit is available or reset fails.', + threshold5h: '5h auto-reset threshold (%)', + threshold7d: '7d auto-reset threshold (%)', + thresholdHint: 'Each window is evaluated independently. Enter 0.1–100; both default to 100.', + thresholdInvalid: 'Automatic reset-credit thresholds must be between 0.1% and 100%.' + }, // Quota control (Anthropic OAuth/SetupToken only) quotaControl: { title: 'Quota Control', @@ -1515,6 +1523,14 @@ export default { resetAccountRecoveryFailed: 'The window was reset, but account state recovery failed. Recover the account state manually.', resetAccountRefreshFailed: 'The window, account state, and reset-credit cache were updated, but the latest account display could not be loaded.', refreshCachePersistFailed: 'Showing the live count, but its expiration details were unavailable, so the cached details were kept.', + autoStatus: { + checking: 'Checking', + available: 'Credit available', + resetting: 'Auto-resetting', + success: 'Auto-reset succeeded', + noCredit: 'No credit', + failed: 'Auto-reset failed' + }, confirmTitle: 'Confirm Weekly Limit Reset', confirmMessage: 'This will consume 1 reset credit to immediately restore the current window ({count} remaining). This action cannot be undone. Continue?' }, diff --git a/frontend/src/i18n/locales/zh/admin/accounts.ts b/frontend/src/i18n/locales/zh/admin/accounts.ts index 923776d259..4c44d93392 100644 --- a/frontend/src/i18n/locales/zh/admin/accounts.ts +++ b/frontend/src/i18n/locales/zh/admin/accounts.ts @@ -486,6 +486,14 @@ export default { resetAccountRecoveryFailed: '窗口已重置，但账号状态恢复失败，请手动恢复账号状态。', resetAccountRefreshFailed: '窗口、账号状态和重置次数缓存已更新，但无法加载最新账号显示。', refreshCachePersistFailed: '已显示实时次数，但到期明细获取失败，仍保留原有缓存明细。', + autoStatus: { + checking: '检测中', + available: '卡可用', + resetting: '自动重置中', + success: '自动重置成功', + noCredit: '无卡', + failed: '自动重置失败' + }, confirmTitle: '确认重置周限', confirmMessage: '将消耗 1 次重置次数立即恢复当前窗口，剩余 {count} 次。此操作不可撤销，确定继续吗？' }, @@ -885,6 +893,14 @@ export default { autoPause5hDisabled: '禁用 5h 自动暂停', autoPause7dDisabled: '禁用 7d 自动暂停', autoPauseDisabledHint: '开启后该账号永不进入自动暂停（即使全局默认阈值已配置）。', + autoResetCredit: { + title: '自动使用重置卡', + hint: '仅在实际用量达到阈值时使用最早到期的可用卡；默认关闭。无卡或失败时账号保持暂停。', + threshold5h: '5h 自动用卡阈值(%)', + threshold7d: '7d 自动用卡阈值(%)', + thresholdHint: '两个窗口独立判断，任一达到自身阈值即触发。可填写 0.1–100，默认均为 100。', + thresholdInvalid: '自动使用重置卡阈值必须在 0.1% 到 100% 之间。' + }, // Quota control (Anthropic OAuth/SetupToken only) quotaControl: { title: '配额控制', diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index df6075c5c9..10a72a0a17 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -1134,6 +1134,17 @@ export interface Account { available_count?: number credits?: { expires_at?: string }[] } + auto_reset_credit_enabled?: boolean + auto_reset_credit_5h_threshold?: number + auto_reset_credit_7d_threshold?: number + codex_auto_reset_credit_state?: { + status?: 'checking' | 'available' | 'resetting' | 'success' | 'no_credit' | 'failed' + trigger_window?: string + available_count?: number + checked_at?: string + last_result_at?: string + error_code?: string + } } & Record) proxy_id: number | null proxy_fallback_origin_id?: number | null