mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-16 15:43:13 +08:00
将 codex_cli_only 客户端识别从「单一 strict 开关 + 固定 OR 头集合」重构为 可逐项管理的引擎指纹信号列表,加固整条判定链,并补齐账号级 app-server 控制、 对齐前端设置文案。 判定链(每步可短路): - 账号未开 codex_cli_only → 不限制;gateway.force_codex_cli → 旁路放行 - 全局黑名单命中(OR 宽 deny)→ 立即拒 - 身份候选:官方 UA(strict,仅前缀)/ 官方 originator(OR)/ 全局白名单(双因子 AND) / 全局 app-server 开关 OR 账号 app-server 开关;均不命中 → 拒 - 版本门(仅官方候选):UA 须可解析引擎版本,再校验 [min,max] 区间 - 引擎指纹 AND 硬门:按信号列表逐条勾选 AND、每条行内变体 OR;无 Required 信号 → 放行 引擎指纹信号列表(唯一真源) - 新增 openai.EngineFingerprintSignal 类型 + EvaluateEngineFingerprint 求值器 (勾选 AND / 行内变体 OR / 无勾选 → 放行) - CodexRestrictionPolicy 增 EngineFingerprintSignals;信号列表单一决定是否启用指纹门, 不再保留独立「要求引擎指纹」总开关(与「信号全不选」语义重复) - 新设置键 codex_cli_only_engine_fingerprint_signals(默认只勾 x-codex- 前缀); 旧 body 指纹开关幂等迁移并入信号列表;wire 接线 - 黑/白名单自由条目、命名预设、版本区间 全局设置管线 - gateway 缺 settingService(仅测试/误配可达)时指纹门回退默认种子信号、失败关闭, 不再因零值 policy(nil 信号)失败开放 账号级 Codex app-server(替换已失效的 ClaudeCode 放行机制) - account.IsCodexCLIOnlyAppServerAllowed() 读 extra.codex_cli_only_allow_app_server, 仅在 codex_cli_only 开启时生效;候选身份门「全局 OR 账号」,与旧系统双层控制对齐 - 移除已无入口的 claude_code 预设机制(allowedClientRegistry / MatchAllowedClients / 账号 GetCodexCLIOnlyAllowedClients / reason);白名单 AllowedClientEntry / IsAllowedClientMatch 保留 门加固(反伪 + 写入校验) - 官方 UA 访问门改 strict:IsCodexOfficialClientRequestStrict 仅前缀匹配,收窄「浏览器前缀 + 中段 codex token」伪造面(strict 仍保留 Codex 家族前缀与 UA 尾部兜底,故对「任意前缀 + 官方尾部 (name;ver)」仍放行——与 UA 可伪造、真正反伪靠引擎指纹门的设计一致) - 官方客户端识别扩展:新增 codex-tui/、codex_vscode_copilot/ 前缀 + UA 尾部 (name;ver) 兜底 (恢复 CODEX_INTERNAL_ORIGINATOR_OVERRIDE 的真实 client,如 cccc→codex-tui),originator 改 精确集。该识别经 IsCodexOfficialClientByHeaders 被 passthrough 复用,故透传的官方判定一并 修正(codex-tui 等不再被误改写 UA)——非「行为不变」,属有意修正 - 白名单写入校验 ValidateCodexWhitelistEntriesJSON + AllowedClientEntry.IsWhitelistable: 双因子 AND 条目须可命中(非空 originator + 非空 ua_contains),拒绝写入会静默失效的死规则; 黑名单(OR 宽 deny,允许 originator-only)不受约束 管理端 / 前端 - handler / DTO / settings_view / 契约测试;gateway 接入判定链 - 信号列表编辑器(替换 body 开关)、api 类型、SettingsView;无勾选给常驻警告 - Create/Edit/Bulk 三弹窗「Codex Only」下新增 app-server 开关(OR 合并全局) - 文案:UA/Originator → User-Agent/Originator;黑/白名单重命名为 User-Agent/Originator 黑/白名单; 「允许 App Server 第三方客户端」→「Codex app-server」+ 简介示例;i18n zh/en 同步 - 移除死代码 HasCodex*Fingerprint helper 测试:引擎指纹求值器 / 账号 app-server(OR 语义)/ detector(含 N1 strict、失败关闭)/ 白名单写入校验 / BulkEdit spec 等;后端 build + service/openai/admin 单测全绿,前端 vue-tsc + vitest 全绿。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
702 lines
26 KiB
Go
702 lines
26 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"time"
|
||
|
||
dbent "github.com/Wei-Shaw/sub2api/ent"
|
||
"github.com/Wei-Shaw/sub2api/internal/config"
|
||
"github.com/Wei-Shaw/sub2api/internal/payment"
|
||
"github.com/Wei-Shaw/sub2api/internal/pkg/antigravity"
|
||
"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
|
||
"github.com/google/wire"
|
||
"github.com/redis/go-redis/v9"
|
||
)
|
||
|
||
// BuildInfo contains build information
|
||
type BuildInfo struct {
|
||
Version string
|
||
BuildType string
|
||
}
|
||
|
||
// ProvidePricingService creates and initializes PricingService
|
||
func ProvidePricingService(cfg *config.Config, remoteClient PricingRemoteClient) (*PricingService, error) {
|
||
svc := NewPricingService(cfg, remoteClient)
|
||
if err := svc.Initialize(); err != nil {
|
||
// Pricing service initialization failure should not block startup, use fallback prices
|
||
println("[Service] Warning: Pricing service initialization failed:", err.Error())
|
||
}
|
||
return svc, nil
|
||
}
|
||
|
||
// ProvideUpdateService creates UpdateService with BuildInfo
|
||
func ProvideUpdateService(cache UpdateCache, githubClient GitHubReleaseClient, buildInfo BuildInfo) *UpdateService {
|
||
return NewUpdateService(cache, githubClient, buildInfo.Version, buildInfo.BuildType)
|
||
}
|
||
|
||
// ProvideEmailQueueService creates EmailQueueService with default worker count
|
||
func ProvideEmailQueueService(emailService *EmailService) *EmailQueueService {
|
||
return NewEmailQueueService(emailService, 3)
|
||
}
|
||
|
||
// ProvideOAuthRefreshAPI creates OAuthRefreshAPI with the default lock TTL.
|
||
func ProvideOAuthRefreshAPI(accountRepo AccountRepository, tokenCache GeminiTokenCache) *OAuthRefreshAPI {
|
||
return NewOAuthRefreshAPI(accountRepo, tokenCache)
|
||
}
|
||
|
||
// ProvideOpenAIOAuthService creates OpenAIOAuthService with privacy/account enrichment support.
|
||
func ProvideOpenAIOAuthService(
|
||
proxyRepo ProxyRepository,
|
||
oauthClient OpenAIOAuthClient,
|
||
privacyClientFactory PrivacyClientFactory,
|
||
) *OpenAIOAuthService {
|
||
svc := NewOpenAIOAuthService(proxyRepo, oauthClient)
|
||
svc.SetPrivacyClientFactory(privacyClientFactory)
|
||
return svc
|
||
}
|
||
|
||
// ProvideTokenRefreshService creates and starts TokenRefreshService
|
||
func ProvideTokenRefreshService(
|
||
accountRepo AccountRepository,
|
||
oauthService *OAuthService,
|
||
openaiOAuthService *OpenAIOAuthService,
|
||
geminiOAuthService *GeminiOAuthService,
|
||
antigravityOAuthService *AntigravityOAuthService,
|
||
grokOAuthService *GrokOAuthService,
|
||
cacheInvalidator TokenCacheInvalidator,
|
||
schedulerCache SchedulerCache,
|
||
cfg *config.Config,
|
||
tempUnschedCache TempUnschedCache,
|
||
privacyClientFactory PrivacyClientFactory,
|
||
proxyRepo ProxyRepository,
|
||
refreshAPI *OAuthRefreshAPI,
|
||
runtimeBlocker AccountRuntimeBlocker,
|
||
) *TokenRefreshService {
|
||
svc := NewTokenRefreshService(accountRepo, oauthService, openaiOAuthService, geminiOAuthService, antigravityOAuthService, cacheInvalidator, schedulerCache, cfg, tempUnschedCache, grokOAuthService)
|
||
// 注入 OpenAI privacy opt-out 依赖
|
||
svc.SetPrivacyDeps(privacyClientFactory, proxyRepo)
|
||
// 注入统一 OAuth 刷新 API(消除 TokenRefreshService 与 TokenProvider 之间的竞争条件)
|
||
svc.SetRefreshAPI(refreshAPI)
|
||
// 调用侧显式注入后台刷新策略,避免策略漂移
|
||
svc.SetRefreshPolicy(DefaultBackgroundRefreshPolicy())
|
||
svc.SetAccountRuntimeBlocker(runtimeBlocker)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideClaudeTokenProvider creates ClaudeTokenProvider with OAuthRefreshAPI injection
|
||
func ProvideClaudeTokenProvider(
|
||
accountRepo AccountRepository,
|
||
tokenCache GeminiTokenCache,
|
||
oauthService *OAuthService,
|
||
refreshAPI *OAuthRefreshAPI,
|
||
) *ClaudeTokenProvider {
|
||
p := NewClaudeTokenProvider(accountRepo, tokenCache, oauthService)
|
||
executor := NewClaudeTokenRefresher(oauthService)
|
||
p.SetRefreshAPI(refreshAPI, executor)
|
||
p.SetRefreshPolicy(ClaudeProviderRefreshPolicy())
|
||
return p
|
||
}
|
||
|
||
// ProvideOpenAITokenProvider creates OpenAITokenProvider with OAuthRefreshAPI injection
|
||
func ProvideOpenAITokenProvider(
|
||
accountRepo AccountRepository,
|
||
tokenCache GeminiTokenCache,
|
||
openaiOAuthService *OpenAIOAuthService,
|
||
refreshAPI *OAuthRefreshAPI,
|
||
) *OpenAITokenProvider {
|
||
p := NewOpenAITokenProvider(accountRepo, tokenCache, openaiOAuthService)
|
||
executor := NewOpenAITokenRefresher(openaiOAuthService, accountRepo)
|
||
p.SetRefreshAPI(refreshAPI, executor)
|
||
p.SetRefreshPolicy(OpenAIProviderRefreshPolicy())
|
||
return p
|
||
}
|
||
|
||
// ProvideOpenAIQuotaService wires the OpenAI quota query/reset service.
|
||
// It depends on the OpenAI token provider for refreshed access tokens and the
|
||
// privacy client factory for the impersonated upstream HTTP client.
|
||
func ProvideOpenAIQuotaService(
|
||
accountRepo AccountRepository,
|
||
proxyRepo ProxyRepository,
|
||
tokenProvider *OpenAITokenProvider,
|
||
privacyClientFactory PrivacyClientFactory,
|
||
) *OpenAIQuotaService {
|
||
return NewOpenAIQuotaService(accountRepo, proxyRepo, tokenProvider, privacyClientFactory)
|
||
}
|
||
|
||
func ProvideGrokQuotaService(
|
||
accountRepo AccountRepository,
|
||
proxyRepo ProxyRepository,
|
||
tokenProvider *GrokTokenProvider,
|
||
httpUpstream HTTPUpstream,
|
||
) *GrokQuotaService {
|
||
return NewGrokQuotaService(accountRepo, proxyRepo, tokenProvider, httpUpstream)
|
||
}
|
||
|
||
// ProvideGeminiTokenProvider creates GeminiTokenProvider with OAuthRefreshAPI injection
|
||
func ProvideGeminiTokenProvider(
|
||
accountRepo AccountRepository,
|
||
tokenCache GeminiTokenCache,
|
||
geminiOAuthService *GeminiOAuthService,
|
||
refreshAPI *OAuthRefreshAPI,
|
||
) *GeminiTokenProvider {
|
||
p := NewGeminiTokenProvider(accountRepo, tokenCache, geminiOAuthService)
|
||
executor := NewGeminiTokenRefresher(geminiOAuthService)
|
||
p.SetRefreshAPI(refreshAPI, executor)
|
||
p.SetRefreshPolicy(GeminiProviderRefreshPolicy())
|
||
return p
|
||
}
|
||
|
||
// ProvideAntigravityTokenProvider creates AntigravityTokenProvider with OAuthRefreshAPI injection
|
||
func ProvideAntigravityTokenProvider(
|
||
accountRepo AccountRepository,
|
||
tokenCache GeminiTokenCache,
|
||
antigravityOAuthService *AntigravityOAuthService,
|
||
refreshAPI *OAuthRefreshAPI,
|
||
tempUnschedCache TempUnschedCache,
|
||
) *AntigravityTokenProvider {
|
||
p := NewAntigravityTokenProvider(accountRepo, tokenCache, antigravityOAuthService)
|
||
executor := NewAntigravityTokenRefresher(antigravityOAuthService)
|
||
p.SetRefreshAPI(refreshAPI, executor)
|
||
p.SetRefreshPolicy(AntigravityProviderRefreshPolicy())
|
||
p.SetTempUnschedCache(tempUnschedCache)
|
||
return p
|
||
}
|
||
|
||
// ProvideGrokTokenProvider creates GrokTokenProvider with OAuthRefreshAPI injection.
|
||
func ProvideGrokTokenProvider(
|
||
accountRepo AccountRepository,
|
||
tokenCache GeminiTokenCache,
|
||
grokOAuthService *GrokOAuthService,
|
||
refreshAPI *OAuthRefreshAPI,
|
||
tempUnschedCache TempUnschedCache,
|
||
) *GrokTokenProvider {
|
||
p := NewGrokTokenProvider(accountRepo, tokenCache)
|
||
executor := NewGrokTokenRefresher(grokOAuthService)
|
||
p.SetRefreshAPI(refreshAPI, executor)
|
||
p.SetRefreshPolicy(AntigravityProviderRefreshPolicy())
|
||
p.SetTempUnschedCache(tempUnschedCache)
|
||
return p
|
||
}
|
||
|
||
// ProvideDashboardAggregationService 创建并启动仪表盘聚合服务
|
||
func ProvideDashboardAggregationService(repo DashboardAggregationRepository, timingWheel *TimingWheelService, lockCache LeaderLockCache, db *sql.DB, cfg *config.Config) *DashboardAggregationService {
|
||
svc := NewDashboardAggregationService(repo, timingWheel, cfg)
|
||
svc.SetLeaderLock(lockCache, db)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideUsageCleanupService 创建并启动使用记录清理任务服务
|
||
func ProvideUsageCleanupService(repo UsageCleanupRepository, timingWheel *TimingWheelService, dashboardAgg *DashboardAggregationService, cfg *config.Config) *UsageCleanupService {
|
||
svc := NewUsageCleanupService(repo, timingWheel, dashboardAgg, cfg)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideAccountExpiryService creates and starts AccountExpiryService.
|
||
func ProvideAccountExpiryService(accountRepo AccountRepository) *AccountExpiryService {
|
||
svc := NewAccountExpiryService(accountRepo, time.Minute)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideProxyExpiryService creates and starts ProxyExpiryService.
|
||
func ProvideProxyExpiryService(proxyRepo ProxyRepository) *ProxyExpiryService {
|
||
svc := NewProxyExpiryService(proxyRepo, time.Minute)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideSubscriptionExpiryService creates and starts SubscriptionExpiryService.
|
||
func ProvideSubscriptionExpiryService(userSubRepo UserSubscriptionRepository, settingRepo SettingRepository, notificationEmailService *NotificationEmailService, lockCache LeaderLockCache, db *sql.DB) *SubscriptionExpiryService {
|
||
svc := NewSubscriptionExpiryService(userSubRepo, time.Minute)
|
||
svc.SetSettingRepository(settingRepo)
|
||
svc.SetNotificationEmailService(notificationEmailService)
|
||
svc.SetLeaderLock(lockCache, db)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideTimingWheelService creates and starts TimingWheelService
|
||
func ProvideTimingWheelService() (*TimingWheelService, error) {
|
||
svc, err := NewTimingWheelService()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
svc.Start()
|
||
return svc, nil
|
||
}
|
||
|
||
// ProvideDeferredService creates and starts DeferredService
|
||
func ProvideDeferredService(accountRepo AccountRepository, timingWheel *TimingWheelService) *DeferredService {
|
||
svc := NewDeferredService(accountRepo, timingWheel, 10*time.Second)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideConcurrencyService creates ConcurrencyService and starts slot cleanup worker.
|
||
func ProvideConcurrencyService(cache ConcurrencyCache, accountRepo AccountRepository, cfg *config.Config) *ConcurrencyService {
|
||
svc := NewConcurrencyService(cache)
|
||
if err := svc.CleanupStaleProcessSlots(context.Background()); err != nil {
|
||
logger.LegacyPrintf("service.concurrency", "Warning: startup cleanup stale process slots failed: %v", err)
|
||
}
|
||
if cfg != nil {
|
||
svc.SetAccountLoadBatchCacheTTL(time.Duration(cfg.Gateway.Scheduling.LoadBatchCacheTTLMS) * time.Millisecond)
|
||
svc.StartSlotCleanupWorker(accountRepo, cfg.Gateway.Scheduling.SlotCleanupInterval)
|
||
}
|
||
return svc
|
||
}
|
||
|
||
// ProvideUserMessageQueueService 创建用户消息串行队列服务并启动清理 worker
|
||
func ProvideUserMessageQueueService(cache UserMsgQueueCache, rpmCache RPMCache, cfg *config.Config) *UserMessageQueueService {
|
||
svc := NewUserMessageQueueService(cache, rpmCache, &cfg.Gateway.UserMessageQueue)
|
||
if cfg.Gateway.UserMessageQueue.CleanupIntervalSeconds > 0 {
|
||
svc.StartCleanupWorker(time.Duration(cfg.Gateway.UserMessageQueue.CleanupIntervalSeconds) * time.Second)
|
||
}
|
||
return svc
|
||
}
|
||
|
||
// ProvideSchedulerSnapshotService creates and starts SchedulerSnapshotService.
|
||
func ProvideSchedulerSnapshotService(
|
||
cache SchedulerCache,
|
||
outboxRepo SchedulerOutboxRepository,
|
||
accountRepo AccountRepository,
|
||
groupRepo GroupRepository,
|
||
cfg *config.Config,
|
||
) *SchedulerSnapshotService {
|
||
svc := NewSchedulerSnapshotService(cache, outboxRepo, accountRepo, groupRepo, cfg)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideRateLimitService creates RateLimitService with optional dependencies.
|
||
func ProvideRateLimitService(
|
||
accountRepo AccountRepository,
|
||
usageRepo UsageLogRepository,
|
||
cfg *config.Config,
|
||
geminiQuotaService *GeminiQuotaService,
|
||
tempUnschedCache TempUnschedCache,
|
||
timeoutCounterCache TimeoutCounterCache,
|
||
openAI403CounterCache OpenAI403CounterCache,
|
||
settingService *SettingService,
|
||
tokenCacheInvalidator TokenCacheInvalidator,
|
||
) *RateLimitService {
|
||
svc := NewRateLimitService(accountRepo, usageRepo, cfg, geminiQuotaService, tempUnschedCache)
|
||
svc.SetTimeoutCounterCache(timeoutCounterCache)
|
||
svc.SetOpenAI403CounterCache(openAI403CounterCache)
|
||
svc.SetSettingService(settingService)
|
||
svc.SetTokenCacheInvalidator(tokenCacheInvalidator)
|
||
return svc
|
||
}
|
||
|
||
// ProvideOpsMetricsCollector creates and starts OpsMetricsCollector.
|
||
func ProvideOpsMetricsCollector(
|
||
opsRepo OpsRepository,
|
||
settingRepo SettingRepository,
|
||
accountRepo AccountRepository,
|
||
concurrencyService *ConcurrencyService,
|
||
db *sql.DB,
|
||
redisClient *redis.Client,
|
||
cfg *config.Config,
|
||
) *OpsMetricsCollector {
|
||
collector := NewOpsMetricsCollector(opsRepo, settingRepo, accountRepo, concurrencyService, db, redisClient, cfg)
|
||
collector.Start()
|
||
return collector
|
||
}
|
||
|
||
// ProvideOpsAggregationService creates and starts OpsAggregationService (hourly/daily pre-aggregation).
|
||
func ProvideOpsAggregationService(
|
||
opsRepo OpsRepository,
|
||
settingRepo SettingRepository,
|
||
db *sql.DB,
|
||
redisClient *redis.Client,
|
||
cfg *config.Config,
|
||
) *OpsAggregationService {
|
||
svc := NewOpsAggregationService(opsRepo, settingRepo, db, redisClient, cfg)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideOpsAlertEvaluatorService creates and starts OpsAlertEvaluatorService.
|
||
func ProvideOpsAlertEvaluatorService(
|
||
opsService *OpsService,
|
||
opsRepo OpsRepository,
|
||
emailService *EmailService,
|
||
redisClient *redis.Client,
|
||
cfg *config.Config,
|
||
proxyRepo ProxyRepository,
|
||
) *OpsAlertEvaluatorService {
|
||
svc := NewOpsAlertEvaluatorService(opsService, opsRepo, emailService, redisClient, cfg, proxyRepo)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideOpsCleanupService creates and starts OpsCleanupService (cron scheduled).
|
||
// channelMonitorSvc 让维护任务(聚合 + 历史/聚合软删)跟随 ops 清理 cron 一起跑,
|
||
// 共享 leader lock + heartbeat。
|
||
// settingRepo 让 cleanup service 自己读 ops_advanced_settings.data_retention 覆盖 cfg;
|
||
// opsService 用来反向注入 cleanup hook,以便 UI 改清理设置时能 Reload cron。
|
||
func ProvideOpsCleanupService(
|
||
opsRepo OpsRepository,
|
||
db *sql.DB,
|
||
redisClient *redis.Client,
|
||
cfg *config.Config,
|
||
channelMonitorSvc *ChannelMonitorService,
|
||
settingRepo SettingRepository,
|
||
opsService *OpsService,
|
||
) *OpsCleanupService {
|
||
svc := NewOpsCleanupService(opsRepo, db, redisClient, cfg, channelMonitorSvc, settingRepo)
|
||
svc.Start()
|
||
if opsService != nil {
|
||
opsService.SetCleanupReloader(svc)
|
||
}
|
||
return svc
|
||
}
|
||
|
||
func ProvideOpsSystemLogSink(opsRepo OpsRepository) *OpsSystemLogSink {
|
||
sink := NewOpsSystemLogSink(opsRepo)
|
||
sink.Start()
|
||
logger.SetSink(sink)
|
||
return sink
|
||
}
|
||
|
||
func buildIdempotencyConfig(cfg *config.Config) IdempotencyConfig {
|
||
idempotencyCfg := DefaultIdempotencyConfig()
|
||
if cfg != nil {
|
||
if cfg.Idempotency.DefaultTTLSeconds > 0 {
|
||
idempotencyCfg.DefaultTTL = time.Duration(cfg.Idempotency.DefaultTTLSeconds) * time.Second
|
||
}
|
||
if cfg.Idempotency.SystemOperationTTLSeconds > 0 {
|
||
idempotencyCfg.SystemOperationTTL = time.Duration(cfg.Idempotency.SystemOperationTTLSeconds) * time.Second
|
||
}
|
||
if cfg.Idempotency.ProcessingTimeoutSeconds > 0 {
|
||
idempotencyCfg.ProcessingTimeout = time.Duration(cfg.Idempotency.ProcessingTimeoutSeconds) * time.Second
|
||
}
|
||
if cfg.Idempotency.FailedRetryBackoffSeconds > 0 {
|
||
idempotencyCfg.FailedRetryBackoff = time.Duration(cfg.Idempotency.FailedRetryBackoffSeconds) * time.Second
|
||
}
|
||
if cfg.Idempotency.MaxStoredResponseLen > 0 {
|
||
idempotencyCfg.MaxStoredResponseLen = cfg.Idempotency.MaxStoredResponseLen
|
||
}
|
||
idempotencyCfg.ObserveOnly = cfg.Idempotency.ObserveOnly
|
||
}
|
||
return idempotencyCfg
|
||
}
|
||
|
||
func ProvideIdempotencyCoordinator(repo IdempotencyRepository, cfg *config.Config) *IdempotencyCoordinator {
|
||
coordinator := NewIdempotencyCoordinator(repo, buildIdempotencyConfig(cfg))
|
||
SetDefaultIdempotencyCoordinator(coordinator)
|
||
return coordinator
|
||
}
|
||
|
||
func ProvideSystemOperationLockService(repo IdempotencyRepository, cfg *config.Config) *SystemOperationLockService {
|
||
return NewSystemOperationLockService(repo, buildIdempotencyConfig(cfg))
|
||
}
|
||
|
||
func ProvideIdempotencyCleanupService(repo IdempotencyRepository, cfg *config.Config) *IdempotencyCleanupService {
|
||
svc := NewIdempotencyCleanupService(repo, cfg)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideScheduledTestService creates ScheduledTestService.
|
||
func ProvideScheduledTestService(
|
||
planRepo ScheduledTestPlanRepository,
|
||
resultRepo ScheduledTestResultRepository,
|
||
) *ScheduledTestService {
|
||
return NewScheduledTestService(planRepo, resultRepo)
|
||
}
|
||
|
||
// ProvideScheduledTestRunnerService creates and starts ScheduledTestRunnerService.
|
||
func ProvideScheduledTestRunnerService(
|
||
planRepo ScheduledTestPlanRepository,
|
||
scheduledSvc *ScheduledTestService,
|
||
accountTestSvc *AccountTestService,
|
||
rateLimitSvc *RateLimitService,
|
||
cfg *config.Config,
|
||
) *ScheduledTestRunnerService {
|
||
svc := NewScheduledTestRunnerService(planRepo, scheduledSvc, accountTestSvc, rateLimitSvc, cfg)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideOpsScheduledReportService creates and starts OpsScheduledReportService.
|
||
func ProvideOpsScheduledReportService(
|
||
opsService *OpsService,
|
||
userService *UserService,
|
||
emailService *EmailService,
|
||
redisClient *redis.Client,
|
||
cfg *config.Config,
|
||
) *OpsScheduledReportService {
|
||
svc := NewOpsScheduledReportService(opsService, userService, emailService, redisClient, cfg)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideAPIKeyAuthCacheInvalidator 提供 API Key 认证缓存失效能力
|
||
func ProvideAPIKeyAuthCacheInvalidator(apiKeyService *APIKeyService) APIKeyAuthCacheInvalidator {
|
||
// Start Pub/Sub subscriber for L1 cache invalidation across instances
|
||
apiKeyService.StartAuthCacheInvalidationSubscriber(context.Background())
|
||
return apiKeyService
|
||
}
|
||
|
||
// ProvideBackupService creates and starts BackupService
|
||
func ProvideBackupService(
|
||
settingRepo SettingRepository,
|
||
cfg *config.Config,
|
||
encryptor SecretEncryptor,
|
||
storeFactory BackupObjectStoreFactory,
|
||
dumper DBDumper,
|
||
) *BackupService {
|
||
svc := NewBackupService(settingRepo, cfg, encryptor, storeFactory, dumper)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideOpsService constructs OpsService and wires the SettingService-backed quota
|
||
// auto-pause cache sink. Mirrors the SetCleanupReloader pattern: OpsService doesn't
|
||
// hold a *SettingService reference, but wire injects a tiny callback so writes to
|
||
// ops_advanced_settings immediately propagate into the scheduler hot-path cache.
|
||
func ProvideOpsService(
|
||
opsRepo OpsRepository,
|
||
settingRepo SettingRepository,
|
||
cfg *config.Config,
|
||
accountRepo AccountRepository,
|
||
userRepo UserRepository,
|
||
concurrencyService *ConcurrencyService,
|
||
gatewayService *GatewayService,
|
||
openAIGatewayService *OpenAIGatewayService,
|
||
geminiCompatService *GeminiMessagesCompatService,
|
||
antigravityGatewayService *AntigravityGatewayService,
|
||
systemLogSink *OpsSystemLogSink,
|
||
settingService *SettingService,
|
||
) *OpsService {
|
||
svc := NewOpsService(
|
||
opsRepo,
|
||
settingRepo,
|
||
cfg,
|
||
accountRepo,
|
||
userRepo,
|
||
concurrencyService,
|
||
gatewayService,
|
||
openAIGatewayService,
|
||
geminiCompatService,
|
||
antigravityGatewayService,
|
||
systemLogSink,
|
||
)
|
||
if settingService != nil {
|
||
svc.SetOpenAIQuotaAutoPauseSettingsSink(settingService.SetOpenAIQuotaAutoPauseSettings)
|
||
// Optional warm-up so the first scheduled request after process start observes
|
||
// a populated cache rather than zero defaults. Best-effort, sync-bounded.
|
||
settingService.WarmOpenAIQuotaAutoPauseSettings(context.Background())
|
||
}
|
||
return svc
|
||
}
|
||
|
||
// ProvideSettingService wires SettingService with group reader and proxy repo.
|
||
func ProvideSettingService(settingRepo SettingRepository, groupRepo GroupRepository, proxyRepo ProxyRepository, cfg *config.Config) *SettingService {
|
||
svc := NewSettingService(settingRepo, cfg)
|
||
svc.SetDefaultSubscriptionGroupReader(groupRepo)
|
||
svc.SetProxyRepository(proxyRepo)
|
||
if err := svc.LoadAPIKeyACLTrustForwardedIPSetting(context.Background()); err != nil {
|
||
logger.LegacyPrintf("service.setting", "Warning: load api key acl forwarded ip setting failed: %v", err)
|
||
}
|
||
if err := svc.MigrateOpenAIAllowClaudeCodeCodexPluginSetting(context.Background()); err != nil {
|
||
logger.LegacyPrintf("service.setting", "Warning: migrate openai allow Claude Code Codex plugin setting failed: %v", err)
|
||
}
|
||
if err := svc.MigrateCodexBodyFingerprintToSignals(context.Background()); err != nil {
|
||
logger.LegacyPrintf("service.setting", "Warning: migrate codex body fingerprint to signals failed: %v", err)
|
||
}
|
||
antigravity.SetUserAgentVersionResolver(svc.GetAntigravityUserAgentVersion)
|
||
return svc
|
||
}
|
||
|
||
// ProvideBillingCacheService wires BillingCacheService with its RPM dependencies.
|
||
func ProvideBillingCacheService(
|
||
cache BillingCache,
|
||
userRepo UserRepository,
|
||
subRepo UserSubscriptionRepository,
|
||
apiKeyRepo APIKeyRepository,
|
||
rpmCache UserRPMCache,
|
||
rateRepo UserGroupRateRepository,
|
||
cfg *config.Config,
|
||
userPlatformQuotaRepo UserPlatformQuotaRepository,
|
||
) *BillingCacheService {
|
||
return NewBillingCacheService(cache, userRepo, subRepo, apiKeyRepo, rpmCache, rateRepo, cfg, userPlatformQuotaRepo)
|
||
}
|
||
|
||
// ProvideAPIKeyService wires APIKeyService and connects rate-limit cache invalidation.
|
||
func ProvideAPIKeyService(
|
||
apiKeyRepo APIKeyRepository,
|
||
userRepo UserRepository,
|
||
groupRepo GroupRepository,
|
||
userSubRepo UserSubscriptionRepository,
|
||
userGroupRateRepo UserGroupRateRepository,
|
||
cache APIKeyCache,
|
||
cfg *config.Config,
|
||
billingCacheService *BillingCacheService,
|
||
) *APIKeyService {
|
||
svc := NewAPIKeyService(apiKeyRepo, userRepo, groupRepo, userSubRepo, userGroupRateRepo, cache, cfg)
|
||
svc.SetRateLimitCacheInvalidator(billingCacheService)
|
||
return svc
|
||
}
|
||
|
||
// ProviderSet is the Wire provider set for all services
|
||
var ProviderSet = wire.NewSet(
|
||
// Core services
|
||
NewAuthService,
|
||
NewUserService,
|
||
ProvideAPIKeyService,
|
||
ProvideAPIKeyAuthCacheInvalidator,
|
||
NewGroupService,
|
||
NewAccountService,
|
||
NewProxyService,
|
||
NewRedeemService,
|
||
NewPromoService,
|
||
NewUsageService,
|
||
NewDashboardService,
|
||
ProvidePricingService,
|
||
NewBillingService,
|
||
ProvideBillingCacheService,
|
||
NewAnnouncementService,
|
||
NewAdminService,
|
||
NewGatewayService,
|
||
NewOpenAIGatewayService,
|
||
wire.Bind(new(AccountRuntimeBlocker), new(*OpenAIGatewayService)),
|
||
NewOAuthService,
|
||
ProvideOpenAIOAuthService,
|
||
NewGrokOAuthService,
|
||
NewGeminiOAuthService,
|
||
NewGeminiQuotaService,
|
||
NewCompositeTokenCacheInvalidator,
|
||
wire.Bind(new(TokenCacheInvalidator), new(*CompositeTokenCacheInvalidator)),
|
||
NewAntigravityOAuthService,
|
||
ProvideOAuthRefreshAPI,
|
||
ProvideGeminiTokenProvider,
|
||
NewGeminiMessagesCompatService,
|
||
ProvideAntigravityTokenProvider,
|
||
ProvideGrokTokenProvider,
|
||
ProvideOpenAITokenProvider,
|
||
ProvideOpenAIQuotaService,
|
||
ProvideGrokQuotaService,
|
||
ProvideClaudeTokenProvider,
|
||
NewAntigravityGatewayService,
|
||
ProvideRateLimitService,
|
||
NewAccountUsageService,
|
||
NewAccountTestService,
|
||
ProvideSettingService,
|
||
NewDataManagementService,
|
||
ProvideBackupService,
|
||
ProvideOpsSystemLogSink,
|
||
ProvideOpsService,
|
||
ProvideOpsMetricsCollector,
|
||
ProvideOpsAggregationService,
|
||
ProvideOpsAlertEvaluatorService,
|
||
ProvideOpsCleanupService,
|
||
ProvideOpsScheduledReportService,
|
||
NewEmailService,
|
||
NewNotificationEmailService,
|
||
ProvideEmailQueueService,
|
||
NewTurnstileService,
|
||
NewSubscriptionService,
|
||
wire.Bind(new(DefaultSubscriptionAssigner), new(*SubscriptionService)),
|
||
ProvideConcurrencyService,
|
||
ProvideUserMessageQueueService,
|
||
NewUsageRecordWorkerPool,
|
||
ProvideSchedulerSnapshotService,
|
||
NewIdentityService,
|
||
NewCRSSyncService,
|
||
ProvideUpdateService,
|
||
ProvideTokenRefreshService,
|
||
ProvideAccountExpiryService,
|
||
ProvideProxyExpiryService,
|
||
ProvideSubscriptionExpiryService,
|
||
ProvideTimingWheelService,
|
||
ProvideDashboardAggregationService,
|
||
ProvideUsageCleanupService,
|
||
ProvideDeferredService,
|
||
NewAntigravityQuotaFetcher,
|
||
NewGrokQuotaFetcher,
|
||
NewUserAttributeService,
|
||
NewUsageCache,
|
||
NewTotpService,
|
||
NewErrorPassthroughService,
|
||
NewTLSFingerprintProfileService,
|
||
NewDigestSessionStore,
|
||
ProvideIdempotencyCoordinator,
|
||
ProvideSystemOperationLockService,
|
||
ProvideIdempotencyCleanupService,
|
||
ProvideScheduledTestService,
|
||
ProvideScheduledTestRunnerService,
|
||
NewGroupCapacityService,
|
||
NewChannelService,
|
||
NewModelPricingResolver,
|
||
NewContentModerationService,
|
||
NewAffiliateService,
|
||
ProvidePaymentConfigService,
|
||
ProvidePaymentService,
|
||
ProvidePaymentOrderExpiryService,
|
||
ProvideBalanceNotifyService,
|
||
ProvideChannelMonitorService,
|
||
ProvideChannelMonitorRunner,
|
||
NewChannelMonitorRequestTemplateService,
|
||
ProvideUserPlatformQuotaUsageFlusher,
|
||
)
|
||
|
||
// ProvideUserPlatformQuotaUsageFlusher 创建并启动 UserPlatformQuotaUsageFlusher。
|
||
func ProvideUserPlatformQuotaUsageFlusher(cfg *config.Config, cache BillingCache, quotaRepo UserPlatformQuotaRepository, tw *TimingWheelService) *UserPlatformQuotaUsageFlusher {
|
||
svc := NewUserPlatformQuotaUsageFlusher(cfg, cache, quotaRepo, tw)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvidePaymentConfigService wraps NewPaymentConfigService to accept the named
|
||
// payment.EncryptionKey type instead of raw []byte, avoiding Wire ambiguity.
|
||
func ProvidePaymentConfigService(entClient *dbent.Client, settingRepo SettingRepository, key payment.EncryptionKey) *PaymentConfigService {
|
||
return NewPaymentConfigService(entClient, settingRepo, []byte(key))
|
||
}
|
||
|
||
// ProvideBalanceNotifyService creates BalanceNotifyService
|
||
func ProvideBalanceNotifyService(emailService *EmailService, settingRepo SettingRepository, accountRepo AccountRepository, notificationEmailService *NotificationEmailService) *BalanceNotifyService {
|
||
svc := NewBalanceNotifyService(emailService, settingRepo, accountRepo)
|
||
svc.SetNotificationEmailService(notificationEmailService)
|
||
return svc
|
||
}
|
||
|
||
// ProvidePaymentService creates PaymentService and attaches notification email delivery.
|
||
func ProvidePaymentService(entClient *dbent.Client, registry *payment.Registry, loadBalancer payment.LoadBalancer, redeemService *RedeemService, subscriptionSvc *SubscriptionService, configService *PaymentConfigService, userRepo UserRepository, groupRepo GroupRepository, affiliateService *AffiliateService, notificationEmailService *NotificationEmailService) *PaymentService {
|
||
svc := NewPaymentService(entClient, registry, loadBalancer, redeemService, subscriptionSvc, configService, userRepo, groupRepo, affiliateService)
|
||
svc.SetNotificationEmailService(notificationEmailService)
|
||
return svc
|
||
}
|
||
|
||
// ProvidePaymentOrderExpiryService creates and starts PaymentOrderExpiryService.
|
||
func ProvidePaymentOrderExpiryService(paymentSvc *PaymentService, lockCache LeaderLockCache, db *sql.DB) *PaymentOrderExpiryService {
|
||
svc := NewPaymentOrderExpiryService(paymentSvc, 60*time.Second)
|
||
svc.SetLeaderLock(lockCache, db)
|
||
svc.Start()
|
||
return svc
|
||
}
|
||
|
||
// ProvideChannelMonitorService 创建渠道监控服务(CRUD + RunCheck + 用户视图聚合)。
|
||
// 加密器复用 wire 中已注入的 SecretEncryptor(AES-256-GCM)。
|
||
func ProvideChannelMonitorService(
|
||
repo ChannelMonitorRepository,
|
||
encryptor SecretEncryptor,
|
||
) *ChannelMonitorService {
|
||
return NewChannelMonitorService(repo, encryptor)
|
||
}
|
||
|
||
// ProvideChannelMonitorRunner 创建并启动渠道监控调度器。
|
||
// 通过 SetScheduler 注入回 service 后再 Start,确保启动时加载所有 enabled monitor,
|
||
// 后续 CRUD 也能即时同步任务表。Runner.Stop 由 cleanup function 调用。
|
||
// settingService 用于 runner 每次 fire 读取功能开关。
|
||
func ProvideChannelMonitorRunner(svc *ChannelMonitorService, settingService *SettingService) *ChannelMonitorRunner {
|
||
r := NewChannelMonitorRunner(svc, settingService)
|
||
svc.SetScheduler(r)
|
||
r.Start()
|
||
return r
|
||
}
|