feat: 支持 OpenAI 重置卡按用量阈值自动使用

This commit is contained in:
NellPoi
2026-08-24 13:28:33 +08:00
parent 7075ae0d82
commit 6f972145b7
31 changed files with 1906 additions and 66 deletions
+7
View File
@@ -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()
+10 -2
View File
@@ -318,7 +318,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)
@@ -345,7 +346,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,
@@ -426,6 +427,7 @@ func provideCleanup(
upstreamBillingProbe *service.UpstreamBillingProbeService,
ollamaCloudUsage *service.OllamaCloudUsageService,
auditLog *service.AuditLogService,
openAIAutoReset *service.OpenAIQuotaAutoResetService,
promptAudit *securityaudit.PromptService,
pluginManager *service.PluginManager,
) func() {
@@ -445,6 +447,12 @@ func provideCleanup(
}
return nil
}},
{"OpenAIQuotaAutoResetService", func() error {
if openAIAutoReset != nil {
openAIAutoReset.Stop()
}
return nil
}},
{"OpsIngressRejectAggregator", func() error {
if opsIngressReject != nil {
opsIngressReject.Stop()
+1
View File
@@ -94,6 +94,7 @@ func TestProvideCleanup_WithMinimalDependencies_NoPanic(t *testing.T) {
nil, // upstreamBillingProbe
nil, // ollamaCloudUsage
nil, // auditLog
nil, // openAIAutoReset
nil, // promptAudit
nil, // pluginManager
)
@@ -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
@@ -47,6 +41,14 @@ const (
// it — spending a second credit.
const openAIQuotaResetPostProcessTimeout = 8 * time.Second
// 保留 handler 包内的告警码名称,避免共享工作流下沉到 service 后破坏既有测试和包内调用。
// 实际值统一引用 service 常量,防止手动重置与自动重置的响应码发生漂移。
const (
openAIQuotaResetWarningCacheRefreshFailed = service.OpenAIQuotaResetWarningCacheRefreshFailed
openAIQuotaResetWarningAccountRecoveryFailed = service.OpenAIQuotaResetWarningAccountRecoveryFailed
openAIQuotaResetWarningAccountRefreshFailed = service.OpenAIQuotaResetWarningAccountRefreshFailed
)
type openAIQuotaResetResponse struct {
service.OpenAIQuotaResetResult
Quota *service.OpenAIQuotaUsage `json:"quota,omitempty"`
@@ -493,6 +495,7 @@ func (h *OpenAIOAuthHandler) QueryQuota(c *gin.Context) {
response.ErrorFrom(c, err)
return
}
service.NotifyOpenAIAutoResetCredit(accountID)
response.Success(c, usage)
}
@@ -523,6 +526,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 +604,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)
}
+1
View File
@@ -196,6 +196,7 @@ func ProvideHandlers(
batchImageHandler *BatchImageHandler,
_ *service.IdempotencyCoordinator,
_ *service.IdempotencyCleanupService,
_ *service.OpenAIQuotaAutoResetService,
) *Handlers {
return &Handlers{
Auth: authHandler,
@@ -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)
}
}()
}
+15
View File
@@ -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)
@@ -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
@@ -389,6 +389,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
@@ -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 {
@@ -508,6 +508,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
@@ -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 {
@@ -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
@@ -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)
}
}()
}
@@ -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)
}
@@ -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
}
@@ -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 &copy, 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 &copy, 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")
}
@@ -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
}
@@ -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) {
@@ -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
}
@@ -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)
@@ -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)
@@ -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 时间戳
+24
View File
@@ -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,
@@ -2266,6 +2266,66 @@
</div>
</div>
<div
v-if="account?.platform === 'openai' && account?.type === 'oauth' && !isSparkShadow"
class="space-y-4 border-t border-gray-200 pt-4 dark:border-dark-600"
data-testid="auto-reset-credit-settings"
>
<div class="flex items-center justify-between gap-4">
<div class="min-w-0">
<label class="input-label mb-0">{{ t('admin.accounts.autoResetCredit.title') }}</label>
<p class="mt-1 text-xs text-gray-500 dark:text-gray-400">
{{ t('admin.accounts.autoResetCredit.hint') }}
</p>
</div>
<button
type="button"
data-testid="auto-reset-credit-enabled"
@click="autoResetCreditEnabled = !autoResetCreditEnabled"
:class="[
'relative inline-flex h-6 w-11 flex-shrink-0 cursor-pointer rounded-full border-2 border-transparent transition-colors duration-200 ease-in-out focus:outline-none focus:ring-2 focus:ring-primary-500 focus:ring-offset-2',
autoResetCreditEnabled ? 'bg-primary-600' : 'bg-gray-200 dark:bg-dark-600'
]"
>
<span
:class="[
'pointer-events-none inline-block h-5 w-5 transform rounded-full bg-white shadow ring-0 transition duration-200 ease-in-out',
autoResetCreditEnabled ? 'translate-x-5' : 'translate-x-0'
]"
/>
</button>
</div>
<div class="grid gap-4 sm:grid-cols-2">
<div>
<label class="input-label">{{ t('admin.accounts.autoResetCredit.threshold5h') }}</label>
<input
v-model.number="autoResetCredit5hThreshold"
type="number"
min="0.1"
max="100"
step="0.1"
class="input"
:disabled="!autoResetCreditEnabled"
data-testid="auto-reset-credit-5h-threshold"
/>
</div>
<div>
<label class="input-label">{{ t('admin.accounts.autoResetCredit.threshold7d') }}</label>
<input
v-model.number="autoResetCredit7dThreshold"
type="number"
min="0.1"
max="100"
step="0.1"
class="input"
:disabled="!autoResetCreditEnabled"
data-testid="auto-reset-credit-7d-threshold"
/>
</div>
</div>
<p class="input-hint">{{ t('admin.accounts.autoResetCredit.thresholdHint') }}</p>
</div>
<!-- 配额控制 (Anthropic OAuth/SetupToken: 亲和 + 窗口费用 + 会话 + RPM 等) -->
<div
v-if="account?.platform === 'anthropic' && (account?.type === 'oauth' || account?.type === 'setup-token')"
@@ -3098,6 +3158,9 @@ const autoPause5hThreshold = ref<number | null>(null)
const autoPause7dThreshold = ref<number | null>(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<string, unknown> = { ...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) {
@@ -63,6 +63,32 @@
</button>
</div>
<div
v-if="autoResetState"
class="flex flex-wrap items-center gap-1 text-[10px]"
data-testid="auto-reset-credit-state"
>
<span
class="inline-flex items-center rounded px-1.5 py-0.5 font-medium"
:class="autoResetStateClass"
>
{{ autoResetStateLabel }}
<span v-if="autoResetState.trigger_window" class="ml-1 tabular-nums">
{{ autoResetState.trigger_window }}
</span>
</span>
<span v-if="autoResetState.checked_at" class="text-gray-500 dark:text-gray-400">
{{ formatResetCreditExpiry(autoResetState.checked_at, 'short') }}
</span>
<span
v-if="autoResetState.error_code"
class="max-w-full truncate text-red-600 dark:text-red-400"
:title="autoResetState.error_code"
>
{{ autoResetState.error_code }}
</span>
</div>
<div v-if="primaryResetCreditExpiry" class="space-y-1">
<div class="flex flex-wrap items-center gap-1">
<span
@@ -172,6 +198,42 @@ const resetWarning = ref<string | null>(null)
const showResetConfirm = ref(false)
const showResetCreditDetails = ref(false)
type AutoResetCreditState = NonNullable<NonNullable<Account['extra']>['codex_auto_reset_credit_state']>
const validAutoResetStatuses = new Set(['checking', 'available', 'resetting', 'success', 'no_credit', 'failed'])
const autoResetState = computed<AutoResetCreditState | null>(() => {
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<string, string> = {
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
@@ -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()
})
})
@@ -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()
})
})
@@ -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?'
},
@@ -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: '配额控制',
+11
View File
@@ -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<string, unknown>)
proxy_id: number | null
proxy_fallback_origin_id?: number | null