account_usage_service.go 66 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798
  1. package service
  2. import (
  3. "bytes"
  4. "context"
  5. "encoding/json"
  6. "fmt"
  7. "log"
  8. "log/slog"
  9. "math/rand/v2"
  10. "net/http"
  11. "strings"
  12. "sync"
  13. "time"
  14. httppool "github.com/Wei-Shaw/sub2api/internal/pkg/httpclient"
  15. openaipkg "github.com/Wei-Shaw/sub2api/internal/pkg/openai"
  16. "github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
  17. "github.com/Wei-Shaw/sub2api/internal/pkg/timezone"
  18. "github.com/Wei-Shaw/sub2api/internal/pkg/tlsfingerprint"
  19. "github.com/Wei-Shaw/sub2api/internal/pkg/usagestats"
  20. "github.com/Wei-Shaw/sub2api/internal/pkg/xai"
  21. "golang.org/x/sync/errgroup"
  22. "golang.org/x/sync/singleflight"
  23. )
  24. type UsageLogRepository interface {
  25. // Create creates a usage log and returns whether it was actually inserted.
  26. // inserted is false when the insert was skipped due to conflict (idempotent retries).
  27. Create(ctx context.Context, log *UsageLog) (inserted bool, err error)
  28. GetByID(ctx context.Context, id int64) (*UsageLog, error)
  29. Delete(ctx context.Context, id int64) error
  30. ListByUser(ctx context.Context, userID int64, params pagination.PaginationParams) ([]UsageLog, *pagination.PaginationResult, error)
  31. ListByAPIKey(ctx context.Context, apiKeyID int64, params pagination.PaginationParams) ([]UsageLog, *pagination.PaginationResult, error)
  32. ListByAccount(ctx context.Context, accountID int64, params pagination.PaginationParams) ([]UsageLog, *pagination.PaginationResult, error)
  33. ListByUserAndTimeRange(ctx context.Context, userID int64, startTime, endTime time.Time) ([]UsageLog, *pagination.PaginationResult, error)
  34. ListByAPIKeyAndTimeRange(ctx context.Context, apiKeyID int64, startTime, endTime time.Time) ([]UsageLog, *pagination.PaginationResult, error)
  35. ListByAccountAndTimeRange(ctx context.Context, accountID int64, startTime, endTime time.Time) ([]UsageLog, *pagination.PaginationResult, error)
  36. ListByModelAndTimeRange(ctx context.Context, modelName string, startTime, endTime time.Time) ([]UsageLog, *pagination.PaginationResult, error)
  37. GetAccountWindowStats(ctx context.Context, accountID int64, startTime time.Time) (*usagestats.AccountStats, error)
  38. GetAccountTodayStats(ctx context.Context, accountID int64) (*usagestats.AccountStats, error)
  39. // Admin dashboard stats
  40. GetDashboardStats(ctx context.Context) (*usagestats.DashboardStats, error)
  41. GetUsageTrendWithFilters(ctx context.Context, startTime, endTime time.Time, granularity string, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8) ([]usagestats.TrendDataPoint, error)
  42. GetModelStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, requestType *int16, stream *bool, billingType *int8) ([]usagestats.ModelStat, error)
  43. GetEndpointStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8) ([]usagestats.EndpointStat, error)
  44. GetUpstreamEndpointStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, model string, requestType *int16, stream *bool, billingType *int8) ([]usagestats.EndpointStat, error)
  45. GetGroupStatsWithFilters(ctx context.Context, startTime, endTime time.Time, userID, apiKeyID, accountID, groupID int64, requestType *int16, stream *bool, billingType *int8) ([]usagestats.GroupStat, error)
  46. GetUserBreakdownStats(ctx context.Context, startTime, endTime time.Time, dim usagestats.UserBreakdownDimension, limit int) ([]usagestats.UserBreakdownItem, error)
  47. GetAllGroupUsageSummary(ctx context.Context, todayStart time.Time) ([]usagestats.GroupUsageSummary, error)
  48. GetAPIKeyUsageTrend(ctx context.Context, startTime, endTime time.Time, granularity string, limit int) ([]usagestats.APIKeyUsageTrendPoint, error)
  49. GetUserUsageTrend(ctx context.Context, startTime, endTime time.Time, granularity string, limit int) ([]usagestats.UserUsageTrendPoint, error)
  50. GetUserSpendingRanking(ctx context.Context, startTime, endTime time.Time, limit int) (*usagestats.UserSpendingRankingResponse, error)
  51. GetBatchUserUsageStats(ctx context.Context, userIDs []int64, startTime, endTime time.Time) (map[int64]*usagestats.BatchUserUsageStats, error)
  52. GetBatchAPIKeyUsageStats(ctx context.Context, apiKeyIDs []int64, startTime, endTime time.Time) (map[int64]*usagestats.BatchAPIKeyUsageStats, error)
  53. // User dashboard stats
  54. GetUserDashboardStats(ctx context.Context, userID int64) (*usagestats.UserDashboardStats, error)
  55. GetAPIKeyDashboardStats(ctx context.Context, apiKeyID int64) (*usagestats.UserDashboardStats, error)
  56. GetUserUsageTrendByUserID(ctx context.Context, userID int64, startTime, endTime time.Time, granularity string) ([]usagestats.TrendDataPoint, error)
  57. GetUserModelStats(ctx context.Context, userID int64, startTime, endTime time.Time) ([]usagestats.ModelStat, error)
  58. // Admin usage listing/stats
  59. ListWithFilters(ctx context.Context, params pagination.PaginationParams, filters usagestats.UsageLogFilters) ([]UsageLog, *pagination.PaginationResult, error)
  60. GetGlobalStats(ctx context.Context, startTime, endTime time.Time) (*usagestats.UsageStats, error)
  61. GetStatsWithFilters(ctx context.Context, filters usagestats.UsageLogFilters) (*usagestats.UsageStats, error)
  62. // Account stats
  63. GetAccountUsageStats(ctx context.Context, accountID int64, startTime, endTime time.Time) (*usagestats.AccountUsageStatsResponse, error)
  64. // Aggregated stats (optimized)
  65. GetUserStatsAggregated(ctx context.Context, userID int64, startTime, endTime time.Time) (*usagestats.UsageStats, error)
  66. GetAPIKeyStatsAggregated(ctx context.Context, apiKeyID int64, startTime, endTime time.Time) (*usagestats.UsageStats, error)
  67. GetAccountStatsAggregated(ctx context.Context, accountID int64, startTime, endTime time.Time) (*usagestats.UsageStats, error)
  68. GetModelStatsAggregated(ctx context.Context, modelName string, startTime, endTime time.Time) (*usagestats.UsageStats, error)
  69. GetDailyStatsAggregated(ctx context.Context, userID int64, startTime, endTime time.Time) ([]map[string]any, error)
  70. }
  71. type accountWindowStatsBatchReader interface {
  72. GetAccountWindowStatsBatch(ctx context.Context, accountIDs []int64, startTime time.Time) (map[int64]*usagestats.AccountStats, error)
  73. }
  74. // apiUsageCache 缓存从 Anthropic API 获取的使用率数据(utilization, resets_at)
  75. // 同时支持缓存错误响应(负缓存),防止 429 等错误导致的重试风暴
  76. type apiUsageCache struct {
  77. response *ClaudeUsageResponse
  78. err error // 非 nil 表示缓存的错误(负缓存)
  79. timestamp time.Time
  80. }
  81. // windowStatsCache 缓存从本地数据库查询的窗口统计(requests, tokens, cost)
  82. type windowStatsCache struct {
  83. stats *WindowStats
  84. timestamp time.Time
  85. }
  86. // antigravityUsageCache 缓存 Antigravity 额度数据
  87. type antigravityUsageCache struct {
  88. usageInfo *UsageInfo
  89. timestamp time.Time
  90. }
  91. const (
  92. apiCacheTTL = 3 * time.Minute
  93. apiErrorCacheTTL = 1 * time.Minute // 负缓存 TTL:429 等错误缓存 1 分钟
  94. antigravityErrorTTL = 1 * time.Minute // Antigravity 错误缓存 TTL(可恢复错误)
  95. apiQueryMaxJitter = 800 * time.Millisecond // 用量查询最大随机延迟
  96. windowStatsCacheTTL = 1 * time.Minute
  97. openAIProbeCacheTTL = 10 * time.Minute
  98. grokProbeRetryTTL = 1 * time.Minute
  99. grokFreeQuotaWindow = 24 * time.Hour
  100. )
  101. // UsageCache 封装账户使用量相关的缓存
  102. type UsageCache struct {
  103. apiCache sync.Map // accountID -> *apiUsageCache
  104. windowStatsCache sync.Map // accountID -> *windowStatsCache
  105. antigravityCache sync.Map // accountID -> *antigravityUsageCache
  106. apiFlight singleflight.Group // 防止同一账号的并发请求击穿缓存(Anthropic)
  107. antigravityFlight singleflight.Group // 防止同一 Antigravity 账号的并发请求击穿缓存
  108. openAIProbeCache sync.Map // accountID -> time.Time
  109. grokProbeCache sync.Map // accountID -> last billing probe attempt
  110. }
  111. // NewUsageCache 创建 UsageCache 实例
  112. func NewUsageCache() *UsageCache {
  113. return &UsageCache{}
  114. }
  115. // WindowStats 窗口期统计
  116. //
  117. // cost: 账号口径费用(total_cost * account_rate_multiplier)
  118. // standard_cost: 标准费用(total_cost,不含倍率)
  119. // user_cost: 用户/API Key 口径费用(actual_cost,受分组倍率影响)
  120. type WindowStats struct {
  121. Requests int64 `json:"requests"`
  122. Tokens int64 `json:"tokens"`
  123. Cost float64 `json:"cost"`
  124. StandardCost float64 `json:"standard_cost"`
  125. UserCost float64 `json:"user_cost"`
  126. }
  127. // UsageProgress 使用量进度
  128. type UsageProgress struct {
  129. Utilization float64 `json:"utilization"` // 使用率百分比 (0-100+,100表示100%)
  130. ResetsAt *time.Time `json:"resets_at"` // 重置时间
  131. RemainingSeconds int `json:"remaining_seconds"` // 距重置剩余秒数
  132. WindowStats *WindowStats `json:"window_stats,omitempty"` // 窗口期统计(从窗口开始到当前的使用量)
  133. UsedRequests int64 `json:"used_requests,omitempty"`
  134. LimitRequests int64 `json:"limit_requests,omitempty"`
  135. }
  136. // AntigravityModelQuota Antigravity 单个模型的配额信息
  137. type AntigravityModelQuota struct {
  138. Utilization int `json:"utilization"` // 使用率 0-100
  139. ResetTime string `json:"reset_time"` // 重置时间 ISO8601
  140. }
  141. // AntigravityModelDetail Antigravity 单个模型的详细能力信息
  142. type AntigravityModelDetail struct {
  143. DisplayName string `json:"display_name,omitempty"`
  144. SupportsImages *bool `json:"supports_images,omitempty"`
  145. SupportsThinking *bool `json:"supports_thinking,omitempty"`
  146. ThinkingBudget *int `json:"thinking_budget,omitempty"`
  147. Recommended *bool `json:"recommended,omitempty"`
  148. MaxTokens *int `json:"max_tokens,omitempty"`
  149. MaxOutputTokens *int `json:"max_output_tokens,omitempty"`
  150. SupportedMimeTypes map[string]bool `json:"supported_mime_types,omitempty"`
  151. }
  152. // AICredit 表示 Antigravity 账号的 AI Credits 余额信息。
  153. type AICredit struct {
  154. CreditType string `json:"credit_type,omitempty"`
  155. Amount float64 `json:"amount,omitempty"`
  156. MinimumBalance float64 `json:"minimum_balance,omitempty"`
  157. }
  158. // UsageInfo 账号使用量信息
  159. type UsageInfo struct {
  160. Source string `json:"source,omitempty"` // "passive" or "active"
  161. UpdatedAt *time.Time `json:"updated_at,omitempty"` // 更新时间
  162. FiveHour *UsageProgress `json:"five_hour"` // 5小时窗口
  163. SevenDay *UsageProgress `json:"seven_day,omitempty"` // 7天窗口
  164. SevenDaySonnet *UsageProgress `json:"seven_day_sonnet,omitempty"` // 7天Sonnet窗口
  165. SevenDayFable *UsageProgress `json:"seven_day_fable,omitempty"` // 7天Fable窗口(响应头 7d_oi)
  166. GeminiSharedDaily *UsageProgress `json:"gemini_shared_daily,omitempty"` // Gemini shared pool RPD (Google One / Code Assist)
  167. GeminiProDaily *UsageProgress `json:"gemini_pro_daily,omitempty"` // Gemini Pro 日配额
  168. GeminiFlashDaily *UsageProgress `json:"gemini_flash_daily,omitempty"` // Gemini Flash 日配额
  169. GeminiSharedMinute *UsageProgress `json:"gemini_shared_minute,omitempty"` // Gemini shared pool RPM (Google One / Code Assist)
  170. GeminiProMinute *UsageProgress `json:"gemini_pro_minute,omitempty"` // Gemini Pro RPM
  171. GeminiFlashMinute *UsageProgress `json:"gemini_flash_minute,omitempty"` // Gemini Flash RPM
  172. // Antigravity 多模型配额
  173. AntigravityQuota map[string]*AntigravityModelQuota `json:"antigravity_quota,omitempty"`
  174. // Grok / xAI 被动额度快照
  175. GrokRequestQuota *xai.QuotaWindow `json:"grok_request_quota,omitempty"`
  176. GrokTokenQuota *xai.QuotaWindow `json:"grok_token_quota,omitempty"`
  177. GrokRetryAfterSeconds *int `json:"grok_retry_after_seconds,omitempty"`
  178. GrokEntitlementStatus string `json:"grok_entitlement_status,omitempty"`
  179. GrokQuotaSnapshotState string `json:"grok_quota_snapshot_state,omitempty"`
  180. GrokLastQuotaProbeAt string `json:"grok_last_quota_probe_at,omitempty"`
  181. GrokLastHeadersSeenAt string `json:"grok_last_headers_seen_at,omitempty"`
  182. GrokLastStatusCode int `json:"grok_last_status_code,omitempty"`
  183. GrokFreeTokenLimit int64 `json:"grok_free_token_limit,omitempty"`
  184. GrokLocalUsage *WindowStats `json:"grok_local_usage,omitempty"`
  185. GrokLocalUsage24h *WindowStats `json:"grok_local_usage_24h,omitempty"`
  186. GrokLocalUsage7d *WindowStats `json:"grok_local_usage_7d,omitempty"`
  187. GrokLocalUsageMonthly *WindowStats `json:"grok_local_usage_monthly,omitempty"`
  188. // ThirtyDay is the official monthly billing window (used/monthlyLimit %).
  189. ThirtyDay *UsageProgress `json:"thirty_day,omitempty"`
  190. GrokBilling *xai.BillingSummary `json:"grok_billing,omitempty"`
  191. // Antigravity 账号级信息
  192. SubscriptionTier string `json:"subscription_tier,omitempty"` // 归一化订阅等级: FREE/PRO/ULTRA/UNKNOWN
  193. SubscriptionTierRaw string `json:"subscription_tier_raw,omitempty"` // 上游原始订阅等级名称
  194. // Antigravity 模型详细能力信息(与 antigravity_quota 同 key)
  195. AntigravityQuotaDetails map[string]*AntigravityModelDetail `json:"antigravity_quota_details,omitempty"`
  196. // Antigravity AI Credits 余额
  197. AICredits []AICredit `json:"ai_credits,omitempty"`
  198. // Antigravity 废弃模型转发规则 (old_model_id -> new_model_id)
  199. ModelForwardingRules map[string]string `json:"model_forwarding_rules,omitempty"`
  200. // Antigravity 账号是否被上游禁止 (HTTP 403)
  201. IsForbidden bool `json:"is_forbidden,omitempty"`
  202. ForbiddenReason string `json:"forbidden_reason,omitempty"`
  203. ForbiddenType string `json:"forbidden_type,omitempty"` // "validation" / "violation" / "forbidden"
  204. ValidationURL string `json:"validation_url,omitempty"` // 验证/申诉链接
  205. // 状态标记(从 ForbiddenType / HTTP 错误码推导)
  206. NeedsVerify bool `json:"needs_verify,omitempty"` // 需要人工验证(forbidden_type=validation)
  207. IsBanned bool `json:"is_banned,omitempty"` // 账号被封(forbidden_type=violation)
  208. NeedsReauth bool `json:"needs_reauth,omitempty"` // token 失效需重新授权(401)
  209. // 错误码(机器可读):forbidden / unauthenticated / rate_limited / network_error
  210. ErrorCode string `json:"error_code,omitempty"`
  211. // 获取 usage 时的错误信息(降级返回,而非 500)
  212. Error string `json:"error,omitempty"`
  213. }
  214. // ClaudeUsageWindow Anthropic /api/oauth/usage 返回的单个用量窗口
  215. type ClaudeUsageWindow struct {
  216. Utilization float64 `json:"utilization"`
  217. ResetsAt string `json:"resets_at"`
  218. }
  219. // ClaudeUsageResponse Anthropic API返回的usage结构
  220. type ClaudeUsageResponse struct {
  221. FiveHour struct {
  222. Utilization float64 `json:"utilization"`
  223. ResetsAt string `json:"resets_at"`
  224. } `json:"five_hour"`
  225. SevenDay struct {
  226. Utilization float64 `json:"utilization"`
  227. ResetsAt string `json:"resets_at"`
  228. } `json:"seven_day"`
  229. SevenDaySonnet struct {
  230. Utilization float64 `json:"utilization"`
  231. ResetsAt string `json:"resets_at"`
  232. } `json:"seven_day_sonnet"`
  233. // Fable 专属 7d 窗口(对应响应头 7d_oi,claim 名为 seven_day_overage_included,
  234. // 见 anthropic-ratelimit-unified-representative-claim 头)。上游 usage API
  235. // 若不下发该字段,GetUsage 会用被动采样数据回填。
  236. SevenDayOverageIncluded ClaudeUsageWindow `json:"seven_day_overage_included"`
  237. }
  238. // ClaudeUsageFetchOptions 包含获取 Claude 用量数据所需的所有选项
  239. type ClaudeUsageFetchOptions struct {
  240. AccessToken string // OAuth access token
  241. ProxyURL string // 代理 URL(可选)
  242. AccountID int64 // 账号 ID(用于连接池隔离)
  243. TLSProfile *tlsfingerprint.Profile // TLS 指纹 Profile(nil 表示不启用)
  244. Fingerprint *Fingerprint // 缓存的指纹信息(User-Agent 等)
  245. }
  246. // ClaudeUsageFetcher fetches usage data from Anthropic OAuth API
  247. type ClaudeUsageFetcher interface {
  248. FetchUsage(ctx context.Context, accessToken, proxyURL string) (*ClaudeUsageResponse, error)
  249. // FetchUsageWithOptions 使用完整选项获取用量数据,支持 TLS 指纹和自定义 User-Agent
  250. FetchUsageWithOptions(ctx context.Context, opts *ClaudeUsageFetchOptions) (*ClaudeUsageResponse, error)
  251. }
  252. // AccountUsageService 账号使用量查询服务
  253. type AccountUsageService struct {
  254. accountRepo AccountRepository
  255. usageLogRepo UsageLogRepository
  256. usageFetcher ClaudeUsageFetcher
  257. geminiQuotaService *GeminiQuotaService
  258. antigravityQuotaFetcher *AntigravityQuotaFetcher
  259. grokQuotaFetcher *GrokQuotaFetcher
  260. grokQuotaService *GrokQuotaService
  261. openAIQuotaService *OpenAIQuotaService
  262. cache *UsageCache
  263. identityCache IdentityCache
  264. tlsFPProfileService *TLSFingerprintProfileService
  265. agentIdentityTaskMu sync.Mutex
  266. agentIdentityWS agentIdentityWSConnectionInvalidator
  267. }
  268. // NewAccountUsageService 创建AccountUsageService实例
  269. func NewAccountUsageService(
  270. accountRepo AccountRepository,
  271. usageLogRepo UsageLogRepository,
  272. usageFetcher ClaudeUsageFetcher,
  273. geminiQuotaService *GeminiQuotaService,
  274. antigravityQuotaFetcher *AntigravityQuotaFetcher,
  275. grokQuotaFetcher *GrokQuotaFetcher,
  276. grokQuotaService *GrokQuotaService,
  277. openAIQuotaService *OpenAIQuotaService,
  278. cache *UsageCache,
  279. identityCache IdentityCache,
  280. tlsFPProfileService *TLSFingerprintProfileService,
  281. ) *AccountUsageService {
  282. return &AccountUsageService{
  283. accountRepo: accountRepo,
  284. usageLogRepo: usageLogRepo,
  285. usageFetcher: usageFetcher,
  286. geminiQuotaService: geminiQuotaService,
  287. antigravityQuotaFetcher: antigravityQuotaFetcher,
  288. grokQuotaFetcher: grokQuotaFetcher,
  289. grokQuotaService: grokQuotaService,
  290. openAIQuotaService: openAIQuotaService,
  291. cache: cache,
  292. identityCache: identityCache,
  293. tlsFPProfileService: tlsFPProfileService,
  294. }
  295. }
  296. func supportsAnthropicPassiveUsage(account *Account) bool {
  297. return account != nil && account.IsAnthropicOAuthOrSetupToken()
  298. }
  299. func batchUsageErrorMessage(err error) string {
  300. if err == nil {
  301. return ""
  302. }
  303. return err.Error()
  304. }
  305. func (s *AccountUsageService) getUsageForAccount(ctx context.Context, account *Account, forceProbe bool) (*UsageInfo, error) {
  306. if account == nil {
  307. return nil, fmt.Errorf("account is required")
  308. }
  309. accountID := account.ID
  310. // Dedicated UI load-test accounts must remain fully interactive without ever
  311. // contacting Anthropic with synthetic credentials. Reuse the same persisted
  312. // passive snapshot that the account table loads on mount.
  313. if account.IsSyntheticUITest() && account.IsAnthropicOAuthOrSetupToken() {
  314. return s.getPassiveUsageForAccount(ctx, account)
  315. }
  316. if account.Platform == PlatformOpenAI && account.Type == AccountTypeOAuth {
  317. usage, err := s.getOpenAIUsage(ctx, account, forceProbe)
  318. if err == nil {
  319. s.tryClearRecoverableAccountError(ctx, account)
  320. }
  321. return usage, err
  322. }
  323. if account.Platform == PlatformGemini {
  324. usage, err := s.getGeminiUsage(ctx, account)
  325. if err == nil {
  326. s.tryClearRecoverableAccountError(ctx, account)
  327. }
  328. return usage, err
  329. }
  330. // Antigravity 平台:使用 AntigravityQuotaFetcher 获取额度
  331. if account.Platform == PlatformAntigravity {
  332. usage, err := s.getAntigravityUsage(ctx, account)
  333. if err == nil {
  334. s.tryClearRecoverableAccountError(ctx, account)
  335. }
  336. return usage, err
  337. }
  338. if account.Platform == PlatformGrok {
  339. usage, err := s.getGrokUsage(ctx, account, forceProbe)
  340. if err == nil && usage != nil && usage.Error == "" {
  341. s.tryClearRecoverableAccountError(ctx, account)
  342. }
  343. return usage, err
  344. }
  345. // 只有oauth类型账号可以通过API获取usage(有profile scope)
  346. if account.CanGetUsage() {
  347. var apiResp *ClaudeUsageResponse
  348. // 1. 检查缓存(成功响应 3 分钟 / 错误响应 1 分钟)
  349. if cached, ok := s.cache.apiCache.Load(accountID); ok {
  350. if cache, ok := cached.(*apiUsageCache); ok {
  351. age := time.Since(cache.timestamp)
  352. if cache.err != nil && age < apiErrorCacheTTL {
  353. // 负缓存命中:返回缓存的错误,避免重试风暴
  354. return nil, cache.err
  355. }
  356. if cache.response != nil && age < apiCacheTTL {
  357. apiResp = cache.response
  358. }
  359. }
  360. }
  361. // 2. 如果没有有效缓存,通过 singleflight 从 API 获取(防止并发击穿)
  362. if apiResp == nil {
  363. // 随机延迟:打散多账号并发请求,避免同一时刻大量相同 TLS 指纹请求
  364. // 触发上游反滥用检测。延迟范围 0~800ms,仅在缓存未命中时生效。
  365. jitter := time.Duration(rand.Int64N(int64(apiQueryMaxJitter)))
  366. select {
  367. case <-time.After(jitter):
  368. case <-ctx.Done():
  369. return nil, ctx.Err()
  370. }
  371. flightKey := fmt.Sprintf("usage:%d", accountID)
  372. result, flightErr, _ := s.cache.apiFlight.Do(flightKey, func() (any, error) {
  373. // 再次检查缓存(可能在等待 singleflight 期间被其他请求填充)
  374. if cached, ok := s.cache.apiCache.Load(accountID); ok {
  375. if cache, ok := cached.(*apiUsageCache); ok {
  376. age := time.Since(cache.timestamp)
  377. if cache.err != nil && age < apiErrorCacheTTL {
  378. return nil, cache.err
  379. }
  380. if cache.response != nil && age < apiCacheTTL {
  381. return cache.response, nil
  382. }
  383. }
  384. }
  385. resp, fetchErr := s.fetchOAuthUsageRaw(ctx, account)
  386. if fetchErr != nil {
  387. // 负缓存:缓存错误响应,防止后续请求重复触发 429
  388. s.cache.apiCache.Store(accountID, &apiUsageCache{
  389. err: fetchErr,
  390. timestamp: time.Now(),
  391. })
  392. return nil, fetchErr
  393. }
  394. // 缓存成功响应
  395. s.cache.apiCache.Store(accountID, &apiUsageCache{
  396. response: resp,
  397. timestamp: time.Now(),
  398. })
  399. return resp, nil
  400. })
  401. if flightErr != nil {
  402. return nil, flightErr
  403. }
  404. apiResp, _ = result.(*ClaudeUsageResponse)
  405. }
  406. // 3. 构建 UsageInfo(每次都重新计算 RemainingSeconds)
  407. now := time.Now()
  408. usage := s.buildUsageInfo(apiResp, &now)
  409. // 4. 添加窗口统计(有独立缓存,1 分钟)
  410. s.addWindowStats(ctx, account, usage)
  411. // 5. 将主动查询结果同步到被动缓存,下次 passive 加载即为最新值
  412. s.syncActiveToPassive(ctx, account.ID, usage)
  413. // 6. 上游 usage API 目前不一定下发 Fable 7d 窗口;缺失时回填被动采样
  414. // (7d_oi 响应头)的数据,避免主动查询后 7d F 进度条丢失。
  415. if usage.SevenDayFable == nil {
  416. usage.SevenDayFable = buildPassiveUsageWindow(account.Extra, "passive_usage_7d_oi_utilization", "passive_usage_7d_oi_reset")
  417. }
  418. s.tryClearRecoverableAccountError(ctx, account)
  419. return usage, nil
  420. }
  421. // Setup Token账号:根据session_window推算(没有profile scope,无法调用usage API)
  422. if account.Type == AccountTypeSetupToken {
  423. usage := s.estimateSetupTokenUsage(account)
  424. // 添加窗口统计
  425. s.addWindowStats(ctx, account, usage)
  426. return usage, nil
  427. }
  428. // API Key账号不支持usage查询
  429. return nil, fmt.Errorf("account type %s does not support usage query", account.Type)
  430. }
  431. // GetUsage 获取账号使用量
  432. // OAuth账号: 调用Anthropic API获取真实数据(需要profile scope),API响应缓存10分钟,窗口统计缓存1分钟
  433. // Setup Token账号: 根据session_window推算5h窗口,7d数据不可用(没有profile scope)
  434. // API Key账号: 不支持usage查询
  435. func (s *AccountUsageService) GetUsage(ctx context.Context, accountID int64, force ...bool) (*UsageInfo, error) {
  436. forceProbe := len(force) > 0 && force[0]
  437. account, err := s.accountRepo.GetByID(ctx, accountID)
  438. if err != nil {
  439. return nil, fmt.Errorf("get account failed: %w", err)
  440. }
  441. return s.getUsageForAccount(ctx, account, forceProbe)
  442. }
  443. // GetUsageForAccount 已加载账号的使用量直通入口(配额监控 fetcher 复用,
  444. // 避免缓存未命中时账号被加载两次——每次 GetByID 含 proxies/groups 联查)。
  445. func (s *AccountUsageService) GetUsageForAccount(ctx context.Context, account *Account, force ...bool) (*UsageInfo, error) {
  446. forceProbe := len(force) > 0 && force[0]
  447. return s.getUsageForAccount(ctx, account, forceProbe)
  448. }
  449. // GetUsageBatch 批量获取账号使用量。
  450. // Anthropic OAuth/SetupToken 统一走 passive 链路,其他账号复用现有主动查询逻辑。
  451. // 单个账号失败不会中断整批请求,错误会按账号返回。
  452. func (s *AccountUsageService) GetUsageBatch(ctx context.Context, accountIDs []int64, force bool) (map[int64]*UsageInfo, map[int64]string, error) {
  453. uniqueIDs := make([]int64, 0, len(accountIDs))
  454. seen := make(map[int64]struct{}, len(accountIDs))
  455. for _, accountID := range accountIDs {
  456. if accountID <= 0 {
  457. continue
  458. }
  459. if _, ok := seen[accountID]; ok {
  460. continue
  461. }
  462. seen[accountID] = struct{}{}
  463. uniqueIDs = append(uniqueIDs, accountID)
  464. }
  465. usageByAccount := make(map[int64]*UsageInfo, len(uniqueIDs))
  466. errorsByAccount := make(map[int64]string)
  467. if len(uniqueIDs) == 0 {
  468. return usageByAccount, errorsByAccount, nil
  469. }
  470. accounts, err := s.accountRepo.GetByIDs(ctx, uniqueIDs)
  471. if err != nil {
  472. return nil, nil, fmt.Errorf("get accounts failed: %w", err)
  473. }
  474. accountsByID := make(map[int64]*Account, len(accounts))
  475. for _, account := range accounts {
  476. if account == nil {
  477. continue
  478. }
  479. accountsByID[account.ID] = account
  480. }
  481. var mu sync.Mutex
  482. g, gctx := errgroup.WithContext(ctx)
  483. g.SetLimit(6)
  484. for _, accountID := range uniqueIDs {
  485. id := accountID
  486. account := accountsByID[id]
  487. if account == nil {
  488. errorsByAccount[id] = ErrAccountNotFound.Error()
  489. continue
  490. }
  491. g.Go(func() error {
  492. var usage *UsageInfo
  493. var usageErr error
  494. if supportsAnthropicPassiveUsage(account) {
  495. usage, usageErr = s.getPassiveUsageForAccount(gctx, account)
  496. } else {
  497. usage, usageErr = s.getUsageForAccount(gctx, account, force)
  498. }
  499. mu.Lock()
  500. defer mu.Unlock()
  501. if usageErr != nil {
  502. errorsByAccount[id] = batchUsageErrorMessage(usageErr)
  503. return nil
  504. }
  505. usageByAccount[id] = usage
  506. return nil
  507. })
  508. }
  509. if err := g.Wait(); err != nil {
  510. return nil, nil, err
  511. }
  512. return usageByAccount, errorsByAccount, nil
  513. }
  514. // GetPassiveUsage 从 Account.Extra 中的被动采样数据构建 UsageInfo,不调用外部 API。
  515. // 仅适用于 Anthropic OAuth / SetupToken 账号。
  516. func (s *AccountUsageService) GetPassiveUsage(ctx context.Context, accountID int64) (*UsageInfo, error) {
  517. account, err := s.accountRepo.GetByID(ctx, accountID)
  518. if err != nil {
  519. return nil, fmt.Errorf("get account failed: %w", err)
  520. }
  521. return s.getPassiveUsageForAccount(ctx, account)
  522. }
  523. func (s *AccountUsageService) getPassiveUsageForAccount(ctx context.Context, account *Account) (*UsageInfo, error) {
  524. if !supportsAnthropicPassiveUsage(account) {
  525. return nil, fmt.Errorf("passive usage only supported for Anthropic OAuth/SetupToken accounts")
  526. }
  527. // 复用 estimateSetupTokenUsage 构建 5h 窗口(OAuth 和 SetupToken 逻辑一致)
  528. info := s.estimateSetupTokenUsage(account)
  529. info.Source = "passive"
  530. // 设置采样时间
  531. if raw, ok := account.Extra["passive_usage_sampled_at"]; ok {
  532. if str, ok := raw.(string); ok {
  533. if t, err := time.Parse(time.RFC3339, str); err == nil {
  534. info.UpdatedAt = &t
  535. }
  536. }
  537. }
  538. // 构建 7d 窗口(从被动采样数据)
  539. info.SevenDay = buildPassiveUsageWindow(account.Extra, "passive_usage_7d_utilization", "passive_usage_7d_reset")
  540. // 构建 7d Fable 窗口(从被动采样的 7d_oi 响应头数据)
  541. info.SevenDayFable = buildPassiveUsageWindow(account.Extra, "passive_usage_7d_oi_utilization", "passive_usage_7d_oi_reset")
  542. // 添加窗口统计
  543. s.addWindowStats(ctx, account, info)
  544. if account.IsSyntheticUITest() {
  545. applySyntheticWindowStats(info, account.Extra)
  546. }
  547. return info, nil
  548. }
  549. func applySyntheticWindowStats(info *UsageInfo, extra map[string]any) {
  550. if info == nil || info.FiveHour == nil || len(extra) == 0 {
  551. return
  552. }
  553. raw, ok := extra["synthetic_window_stats"].(map[string]any)
  554. if !ok {
  555. return
  556. }
  557. info.FiveHour.WindowStats = &WindowStats{
  558. Requests: int64(parseExtraInt(raw["requests"])),
  559. Tokens: int64(parseExtraInt(raw["tokens"])),
  560. Cost: parseExtraFloat64(raw["cost"]),
  561. StandardCost: parseExtraFloat64(raw["standard_cost"]),
  562. UserCost: parseExtraFloat64(raw["user_cost"]),
  563. }
  564. }
  565. // buildPassiveUsageWindow 从 Extra 中的被动采样数据(utilization 为 0-1 小数、reset 为 Unix 秒)
  566. // 构建用量窗口,无数据时返回 nil。
  567. func buildPassiveUsageWindow(extra map[string]any, utilKey, resetKey string) *UsageProgress {
  568. util := parseExtraFloat64(extra[utilKey])
  569. resetRaw := parseExtraFloat64(extra[resetKey])
  570. if util <= 0 && resetRaw <= 0 {
  571. return nil
  572. }
  573. var resetAt *time.Time
  574. var remaining int
  575. if resetRaw > 0 {
  576. t := time.Unix(int64(resetRaw), 0)
  577. resetAt = &t
  578. remaining = int(time.Until(t).Seconds())
  579. if remaining < 0 {
  580. remaining = 0
  581. }
  582. }
  583. return &UsageProgress{
  584. Utilization: util * 100,
  585. ResetsAt: resetAt,
  586. RemainingSeconds: remaining,
  587. }
  588. }
  589. // syncActiveToPassive 将主动查询的最新数据回写到 Extra 被动缓存,
  590. // 这样下次被动加载时能看到最新值。
  591. func (s *AccountUsageService) syncActiveToPassive(ctx context.Context, accountID int64, usage *UsageInfo) {
  592. extraUpdates := make(map[string]any, 4)
  593. if usage.FiveHour != nil {
  594. extraUpdates["session_window_utilization"] = usage.FiveHour.Utilization / 100
  595. }
  596. if usage.SevenDay != nil {
  597. extraUpdates["passive_usage_7d_utilization"] = usage.SevenDay.Utilization / 100
  598. if usage.SevenDay.ResetsAt != nil {
  599. extraUpdates["passive_usage_7d_reset"] = usage.SevenDay.ResetsAt.Unix()
  600. }
  601. }
  602. if usage.SevenDayFable != nil {
  603. extraUpdates["passive_usage_7d_oi_utilization"] = usage.SevenDayFable.Utilization / 100
  604. if usage.SevenDayFable.ResetsAt != nil {
  605. extraUpdates["passive_usage_7d_oi_reset"] = usage.SevenDayFable.ResetsAt.Unix()
  606. }
  607. }
  608. if len(extraUpdates) > 0 {
  609. extraUpdates["passive_usage_sampled_at"] = time.Now().UTC().Format(time.RFC3339)
  610. if err := s.accountRepo.UpdateExtra(ctx, accountID, extraUpdates); err != nil {
  611. slog.Warn("sync_active_to_passive_failed", "account_id", accountID, "error", err)
  612. }
  613. }
  614. // 5h ResetsAt 必须回写到 SessionWindowEnd column,estimateSetupTokenUsage
  615. // 读这个字段作为窗口结束时间;只塞 Extra 会让 UI 一直拿到上个窗口的过期时间。
  616. if usage.FiveHour != nil && usage.FiveHour.ResetsAt != nil {
  617. if err := s.accountRepo.UpdateSessionWindowEnd(ctx, accountID, *usage.FiveHour.ResetsAt); err != nil {
  618. slog.Warn("sync_active_to_passive_session_window_end_failed", "account_id", accountID, "error", err)
  619. }
  620. }
  621. }
  622. func (s *AccountUsageService) getOpenAIUsage(ctx context.Context, account *Account, force bool) (*UsageInfo, error) {
  623. now := time.Now()
  624. usage := &UsageInfo{UpdatedAt: &now}
  625. if account == nil {
  626. return usage, nil
  627. }
  628. applyExtraToUsage(usage, account.Extra, now)
  629. if (force || shouldRefreshOpenAICodexSnapshot(account, usage, now)) && s.shouldProbeOpenAICodexSnapshot(account.ID, now, force) {
  630. if account.IsShadow() {
  631. // Spark shadow accounts fetch usage from /wham/usage (bengalfox channel)
  632. // via the shared OpenAIQuotaService, which resolves credentials from the
  633. // parent account. The result is written to the shadow row's own codex_*
  634. // Extra keys and immediately reflected in the returned UsageInfo.
  635. if s.openAIQuotaService != nil {
  636. if quotaUsage, err := s.openAIQuotaService.QueryUsage(ctx, account.ID); err == nil {
  637. if updates := buildCodexSparkWindowExtraUpdates(quotaUsage, now); len(updates) > 0 {
  638. mergeAccountExtra(account, updates)
  639. s.persistOpenAICodexProbeSnapshot(account.ID, updates)
  640. if account.ParentAccountID != nil {
  641. notifyOpenAIAutoReset(*account.ParentAccountID)
  642. }
  643. if usage.UpdatedAt == nil {
  644. usage.UpdatedAt = &now
  645. }
  646. applyExtraToUsage(usage, account.Extra, now)
  647. }
  648. }
  649. }
  650. } else {
  651. if updates, err := s.probeOpenAICodexSnapshot(ctx, account); err == nil && len(updates) > 0 {
  652. mergeAccountExtra(account, updates)
  653. if usage.UpdatedAt == nil {
  654. usage.UpdatedAt = &now
  655. }
  656. applyExtraToUsage(usage, account.Extra, now)
  657. }
  658. }
  659. }
  660. if s.usageLogRepo == nil {
  661. return usage, nil
  662. }
  663. if stats, err := s.usageLogRepo.GetAccountWindowStats(ctx, account.ID, codexWindowStatsStart(usage.FiveHour, 5*time.Hour, now)); err == nil {
  664. if usage.FiveHour == nil {
  665. usage.FiveHour = &UsageProgress{Utilization: 0}
  666. }
  667. usage.FiveHour.WindowStats = windowStatsFromAccountStats(stats)
  668. }
  669. if stats, err := s.usageLogRepo.GetAccountWindowStats(ctx, account.ID, codexWindowStatsStart(usage.SevenDay, 7*24*time.Hour, now)); err == nil {
  670. if usage.SevenDay == nil {
  671. usage.SevenDay = &UsageProgress{Utilization: 0}
  672. }
  673. usage.SevenDay.WindowStats = windowStatsFromAccountStats(stats)
  674. }
  675. return usage, nil
  676. }
  677. func shouldRefreshOpenAICodexSnapshot(account *Account, usage *UsageInfo, now time.Time) bool {
  678. if account == nil {
  679. return false
  680. }
  681. if usage == nil {
  682. return true
  683. }
  684. if usage.FiveHour == nil || usage.SevenDay == nil {
  685. return true
  686. }
  687. if account.IsRateLimited() {
  688. return true
  689. }
  690. return isOpenAICodexSnapshotStale(account, now)
  691. }
  692. func isOpenAICodexSnapshotStale(account *Account, now time.Time) bool {
  693. if account == nil || !account.IsOpenAIOAuth() {
  694. return false
  695. }
  696. // 普通账号的 codex 刷新走 probe(/responses 头),要求 WSv2;但 spark 影子走 QueryUsage
  697. // (/wham/usage body 的 codex_bengalfox),与 WSv2 无关——不能用 WSv2 门控其 staleness,否则首刷后
  698. // codex_5h/7d 已存在→staleness 恒 false→spark 窗口永久冻结(外审第9轮 P1)。影子改按
  699. // codex_usage_updated_at TTL 判定;实际查询频率仍由 shouldProbeOpenAICodexSnapshot 的缓存 TTL 节流。
  700. if !account.IsShadow() && !account.IsOpenAIResponsesWebSocketV2Enabled() {
  701. return false
  702. }
  703. if account.Extra == nil {
  704. return true
  705. }
  706. raw, ok := account.Extra["codex_usage_updated_at"]
  707. if !ok {
  708. return true
  709. }
  710. ts, err := parseTime(fmt.Sprint(raw))
  711. if err != nil {
  712. return true
  713. }
  714. return now.Sub(ts) >= openAIProbeCacheTTL
  715. }
  716. func (s *AccountUsageService) shouldProbeOpenAICodexSnapshot(accountID int64, now time.Time, force ...bool) bool {
  717. if s == nil || s.cache == nil || accountID <= 0 {
  718. return true
  719. }
  720. forceProbe := len(force) > 0 && force[0]
  721. if !forceProbe {
  722. if cached, ok := s.cache.openAIProbeCache.Load(accountID); ok {
  723. if ts, ok := cached.(time.Time); ok && now.Sub(ts) < openAIProbeCacheTTL {
  724. return false
  725. }
  726. }
  727. }
  728. s.cache.openAIProbeCache.Store(accountID, now)
  729. return true
  730. }
  731. func (s *AccountUsageService) probeOpenAICodexSnapshot(ctx context.Context, account *Account) (map[string]any, error) {
  732. if account == nil || !account.IsOAuth() {
  733. return nil, nil
  734. }
  735. accessToken := ""
  736. if !account.IsOpenAIAgentIdentity() {
  737. accessToken = account.GetOpenAIAccessToken()
  738. }
  739. if accessToken == "" && !account.IsOpenAIAgentIdentity() {
  740. return nil, fmt.Errorf("no access token available")
  741. }
  742. modelID := openaipkg.CodexUsageProbeModel
  743. payload := createOpenAITestPayload(modelID, true)
  744. payloadBytes, err := json.Marshal(payload)
  745. if err != nil {
  746. return nil, fmt.Errorf("marshal openai probe payload: %w", err)
  747. }
  748. reqCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
  749. defer cancel()
  750. req, err := http.NewRequestWithContext(reqCtx, http.MethodPost, chatgptCodexURL, bytes.NewReader(payloadBytes))
  751. if err != nil {
  752. return nil, fmt.Errorf("create openai probe request: %w", err)
  753. }
  754. req.Host = "chatgpt.com"
  755. req.Header.Set("Content-Type", "application/json")
  756. if account.IsOpenAIAgentIdentity() {
  757. authHeaders, authErr := buildAgentIdentityAuthenticationHeaders(ctx, s.accountRepo, s.agentIdentityWS, &s.agentIdentityTaskMu, account)
  758. if authErr != nil {
  759. return nil, fmt.Errorf("build Agent Identity authentication: %w", authErr)
  760. }
  761. for key, values := range authHeaders {
  762. for _, value := range values {
  763. req.Header.Add(key, value)
  764. }
  765. }
  766. } else {
  767. req.Header.Set("Authorization", "Bearer "+accessToken)
  768. }
  769. req.Header.Set("Accept", "text/event-stream")
  770. req.Header.Set("OpenAI-Beta", "responses=experimental")
  771. canonical := resolveCodexOutboundIdentity("")
  772. req.Header.Set("Originator", canonical.originator)
  773. req.Header.Set("Version", canonical.version)
  774. req.Header.Set("User-Agent", canonical.userAgent)
  775. if s.identityCache != nil {
  776. if fp, fpErr := s.identityCache.GetFingerprint(reqCtx, account.ID); fpErr == nil && fp != nil && strings.TrimSpace(fp.UserAgent) != "" {
  777. req.Header.Set("User-Agent", strings.TrimSpace(fp.UserAgent))
  778. }
  779. }
  780. // 与真实转发一致:账号级自定义 UA 同样作为管理员显式配置传入。
  781. // 上面写进 header 的指纹缓存 UA 只在强制统一被关闭时才参与配对(保持回滚后的历史语义);
  782. // 强制统一开启时客户端身份不参与构造,探针与真实转发用同一套规范身份出站。
  783. enforceCodexIdentityHeadersWithUA(req.Header, account.GetOpenAIUserAgent())
  784. setOpenAIChatGPTAccountHeaders(req.Header, account)
  785. proxyURL := ""
  786. if account.ProxyID != nil && account.Proxy != nil {
  787. proxyURL = account.Proxy.URL()
  788. }
  789. client, err := httppool.GetClient(httppool.Options{
  790. ProxyURL: proxyURL,
  791. Timeout: 15 * time.Second,
  792. ResponseHeaderTimeout: 10 * time.Second,
  793. })
  794. if err != nil {
  795. return nil, fmt.Errorf("build openai probe client: %w", err)
  796. }
  797. resp, err := client.Do(req)
  798. if err != nil {
  799. return nil, fmt.Errorf("openai codex probe request failed: %w", err)
  800. }
  801. defer func() { _ = resp.Body.Close() }()
  802. updates, err := extractOpenAICodexProbeUpdates(resp)
  803. if err != nil {
  804. return nil, err
  805. }
  806. if len(updates) > 0 {
  807. s.persistOpenAICodexProbeSnapshot(account.ID, updates)
  808. return updates, nil
  809. }
  810. return nil, nil
  811. }
  812. func (s *AccountUsageService) persistOpenAICodexProbeSnapshot(accountID int64, updates map[string]any) {
  813. if s == nil || s.accountRepo == nil || accountID <= 0 {
  814. return
  815. }
  816. if len(updates) == 0 {
  817. return
  818. }
  819. go func() {
  820. updateCtx, updateCancel := context.WithTimeout(context.Background(), 5*time.Second)
  821. defer updateCancel()
  822. if err := s.accountRepo.UpdateExtra(updateCtx, accountID, updates); err == nil {
  823. notifyOpenAIAutoReset(accountID)
  824. }
  825. }()
  826. }
  827. func extractOpenAICodexProbeUpdates(resp *http.Response) (map[string]any, error) {
  828. if resp == nil {
  829. return nil, nil
  830. }
  831. if snapshot := ParseCodexRateLimitHeaders(resp.Header); snapshot != nil {
  832. return buildCodexUsageExtraUpdates(snapshot, time.Now()), nil
  833. }
  834. if resp.StatusCode < 200 || resp.StatusCode >= 300 {
  835. return nil, fmt.Errorf("openai codex probe returned status %d", resp.StatusCode)
  836. }
  837. return nil, nil
  838. }
  839. func mergeAccountExtra(account *Account, updates map[string]any) {
  840. if account == nil || len(updates) == 0 {
  841. return
  842. }
  843. if account.Extra == nil {
  844. account.Extra = make(map[string]any, len(updates))
  845. }
  846. for k, v := range updates {
  847. account.Extra[k] = v
  848. }
  849. }
  850. // applyExtraToUsage rebuilds the codex 5h/7d windows in usage from the
  851. // account's Extra map. Called after mergeAccountExtra to make the in-memory
  852. // UsageInfo consistent with the just-persisted Extra values.
  853. func applyExtraToUsage(usage *UsageInfo, extra map[string]any, now time.Time) {
  854. if usage == nil {
  855. return
  856. }
  857. if progress := buildCodexUsageProgressFromExtra(extra, "5h", now); progress != nil {
  858. usage.FiveHour = progress
  859. }
  860. if progress := buildCodexUsageProgressFromExtra(extra, "7d", now); progress != nil {
  861. usage.SevenDay = progress
  862. }
  863. }
  864. func (s *AccountUsageService) getGeminiUsage(ctx context.Context, account *Account) (*UsageInfo, error) {
  865. now := time.Now()
  866. usage := &UsageInfo{
  867. UpdatedAt: &now,
  868. }
  869. if s.geminiQuotaService == nil || s.usageLogRepo == nil {
  870. return usage, nil
  871. }
  872. quota, ok := s.geminiQuotaService.QuotaForAccount(ctx, account)
  873. if !ok {
  874. return usage, nil
  875. }
  876. dayStart := geminiDailyWindowStart(now)
  877. stats, err := s.usageLogRepo.GetModelStatsWithFilters(ctx, dayStart, now, 0, 0, account.ID, 0, nil, nil, nil)
  878. if err != nil {
  879. return nil, fmt.Errorf("get gemini usage stats failed: %w", err)
  880. }
  881. dayTotals := geminiAggregateUsage(stats)
  882. dailyResetAt := geminiDailyResetTime(now)
  883. // Daily window (RPD)
  884. if quota.SharedRPD > 0 {
  885. totalReq := dayTotals.ProRequests + dayTotals.FlashRequests
  886. totalTokens := dayTotals.ProTokens + dayTotals.FlashTokens
  887. totalCost := dayTotals.ProCost + dayTotals.FlashCost
  888. usage.GeminiSharedDaily = buildGeminiUsageProgress(totalReq, quota.SharedRPD, dailyResetAt, totalTokens, totalCost, now)
  889. } else {
  890. usage.GeminiProDaily = buildGeminiUsageProgress(dayTotals.ProRequests, quota.ProRPD, dailyResetAt, dayTotals.ProTokens, dayTotals.ProCost, now)
  891. usage.GeminiFlashDaily = buildGeminiUsageProgress(dayTotals.FlashRequests, quota.FlashRPD, dailyResetAt, dayTotals.FlashTokens, dayTotals.FlashCost, now)
  892. }
  893. // Minute window (RPM) - fixed-window approximation: current minute [truncate(now), truncate(now)+1m)
  894. minuteStart := now.Truncate(time.Minute)
  895. minuteResetAt := minuteStart.Add(time.Minute)
  896. minuteStats, err := s.usageLogRepo.GetModelStatsWithFilters(ctx, minuteStart, now, 0, 0, account.ID, 0, nil, nil, nil)
  897. if err != nil {
  898. return nil, fmt.Errorf("get gemini minute usage stats failed: %w", err)
  899. }
  900. minuteTotals := geminiAggregateUsage(minuteStats)
  901. if quota.SharedRPM > 0 {
  902. totalReq := minuteTotals.ProRequests + minuteTotals.FlashRequests
  903. totalTokens := minuteTotals.ProTokens + minuteTotals.FlashTokens
  904. totalCost := minuteTotals.ProCost + minuteTotals.FlashCost
  905. usage.GeminiSharedMinute = buildGeminiUsageProgress(totalReq, quota.SharedRPM, minuteResetAt, totalTokens, totalCost, now)
  906. } else {
  907. usage.GeminiProMinute = buildGeminiUsageProgress(minuteTotals.ProRequests, quota.ProRPM, minuteResetAt, minuteTotals.ProTokens, minuteTotals.ProCost, now)
  908. usage.GeminiFlashMinute = buildGeminiUsageProgress(minuteTotals.FlashRequests, quota.FlashRPM, minuteResetAt, minuteTotals.FlashTokens, minuteTotals.FlashCost, now)
  909. }
  910. return usage, nil
  911. }
  912. // getAntigravityUsage 获取 Antigravity 账户额度
  913. func (s *AccountUsageService) getAntigravityUsage(ctx context.Context, account *Account) (*UsageInfo, error) {
  914. if s.antigravityQuotaFetcher == nil || !s.antigravityQuotaFetcher.CanFetch(account) {
  915. now := time.Now()
  916. return &UsageInfo{UpdatedAt: &now}, nil
  917. }
  918. // 1. 检查缓存
  919. if cached, ok := s.cache.antigravityCache.Load(account.ID); ok {
  920. if cache, ok := cached.(*antigravityUsageCache); ok {
  921. ttl := antigravityCacheTTL(cache.usageInfo)
  922. if time.Since(cache.timestamp) < ttl {
  923. usage := cache.usageInfo
  924. if usage.FiveHour != nil && usage.FiveHour.ResetsAt != nil {
  925. usage.FiveHour.RemainingSeconds = int(time.Until(*usage.FiveHour.ResetsAt).Seconds())
  926. }
  927. return usage, nil
  928. }
  929. }
  930. }
  931. // 2. singleflight 防止并发击穿
  932. flightKey := fmt.Sprintf("ag-usage:%d", account.ID)
  933. result, flightErr, _ := s.cache.antigravityFlight.Do(flightKey, func() (any, error) {
  934. // 再次检查缓存(等待期间可能已被填充)
  935. if cached, ok := s.cache.antigravityCache.Load(account.ID); ok {
  936. if cache, ok := cached.(*antigravityUsageCache); ok {
  937. ttl := antigravityCacheTTL(cache.usageInfo)
  938. if time.Since(cache.timestamp) < ttl {
  939. usage := cache.usageInfo
  940. // 重新计算 RemainingSeconds,避免返回过时的剩余秒数
  941. recalcAntigravityRemainingSeconds(usage)
  942. return usage, nil
  943. }
  944. }
  945. }
  946. // 使用独立 context,避免调用方 cancel 导致所有共享 flight 的请求失败
  947. fetchCtx, fetchCancel := context.WithTimeout(context.Background(), 30*time.Second)
  948. defer fetchCancel()
  949. proxyURL := s.antigravityQuotaFetcher.GetProxyURL(fetchCtx, account)
  950. fetchResult, err := s.antigravityQuotaFetcher.FetchQuota(fetchCtx, account, proxyURL)
  951. if err != nil {
  952. degraded := buildAntigravityDegradedUsage(err)
  953. enrichUsageWithAccountError(degraded, account)
  954. s.cache.antigravityCache.Store(account.ID, &antigravityUsageCache{
  955. usageInfo: degraded,
  956. timestamp: time.Now(),
  957. })
  958. return degraded, nil
  959. }
  960. enrichUsageWithAccountError(fetchResult.UsageInfo, account)
  961. s.cache.antigravityCache.Store(account.ID, &antigravityUsageCache{
  962. usageInfo: fetchResult.UsageInfo,
  963. timestamp: time.Now(),
  964. })
  965. return fetchResult.UsageInfo, nil
  966. })
  967. if flightErr != nil {
  968. return nil, flightErr
  969. }
  970. usage, ok := result.(*UsageInfo)
  971. if !ok || usage == nil {
  972. now := time.Now()
  973. return &UsageInfo{UpdatedAt: &now}, nil
  974. }
  975. return usage, nil
  976. }
  977. func (s *AccountUsageService) getGrokUsage(ctx context.Context, account *Account, force bool) (*UsageInfo, error) {
  978. if s.grokQuotaFetcher == nil {
  979. now := time.Now()
  980. return &UsageInfo{UpdatedAt: &now}, nil
  981. }
  982. var billingProbeResult *GrokQuotaProbeResult
  983. if account != nil && account.IsGrokOAuth() && s.grokQuotaService != nil && (force || grokBillingSnapshotNeedsRefresh(account, time.Now())) && s.shouldProbeGrokBilling(account.ID, time.Now(), force) {
  984. result, err := s.grokQuotaService.ProbeBilling(ctx, account.ID)
  985. if err == nil && result != nil && result.Billing != nil {
  986. billingProbeResult = result
  987. mergeAccountExtra(account, map[string]any{grokBillingExtraKey: result.Billing})
  988. } else if err != nil && force {
  989. return nil, err
  990. }
  991. }
  992. usage := s.grokQuotaFetcher.BuildUsageInfo(account)
  993. if usage.GrokQuotaSnapshotState == "" {
  994. if usage.ErrorCode == "quota_unknown" {
  995. usage.GrokQuotaSnapshotState = "unknown_until_first_response"
  996. } else {
  997. usage.GrokQuotaSnapshotState = "observed"
  998. }
  999. }
  1000. if account != nil {
  1001. if s.usageLogRepo != nil {
  1002. if stats, err := s.usageLogRepo.GetAccountTodayStats(ctx, account.ID); err == nil && stats != nil {
  1003. usage.GrokLocalUsage = windowStatsFromAccountStats(stats)
  1004. }
  1005. }
  1006. if billingProbeResult != nil {
  1007. usage.GrokLocalUsage24h = billingProbeResult.LocalUsage24h
  1008. usage.GrokLocalUsage7d = billingProbeResult.LocalUsage7d
  1009. usage.GrokLocalUsageMonthly = billingProbeResult.LocalUsageMonthly
  1010. } else if s.usageLogRepo != nil {
  1011. usage.GrokLocalUsage24h, usage.GrokLocalUsage7d, usage.GrokLocalUsageMonthly = grokLocalUsageForQuota(
  1012. ctx, s.usageLogRepo, account.ID, usage.GrokBilling, time.Now().UTC(),
  1013. )
  1014. }
  1015. // Attach local window stats to official 7d/30d progress bars.
  1016. if usage.SevenDay != nil && usage.GrokLocalUsage7d != nil {
  1017. usage.SevenDay.WindowStats = usage.GrokLocalUsage7d
  1018. }
  1019. if usage.ThirtyDay != nil && usage.GrokLocalUsageMonthly != nil {
  1020. usage.ThirtyDay.WindowStats = usage.GrokLocalUsageMonthly
  1021. }
  1022. }
  1023. enrichUsageWithAccountError(usage, account)
  1024. return usage, nil
  1025. }
  1026. func grokLocalUsageForQuota(
  1027. ctx context.Context,
  1028. repo UsageLogRepository,
  1029. accountID int64,
  1030. billing *xai.BillingSummary,
  1031. now time.Time,
  1032. ) (*WindowStats, *WindowStats, *WindowStats) {
  1033. if grokBillingHasAuthoritativeQuota(billing) {
  1034. weekly, monthly := grokLocalUsageForBilling(ctx, repo, accountID, billing, now)
  1035. return nil, weekly, monthly
  1036. }
  1037. return grokLocalUsage24h(ctx, repo, accountID, now), nil, nil
  1038. }
  1039. func grokLocalUsage24h(ctx context.Context, repo UsageLogRepository, accountID int64, now time.Time) *WindowStats {
  1040. if repo == nil || accountID <= 0 {
  1041. return nil
  1042. }
  1043. start := now.UTC().Add(-grokFreeQuotaWindow)
  1044. stats, err := repo.GetAccountWindowStats(ctx, accountID, start)
  1045. if err != nil {
  1046. slog.Warn("grok_rolling_24h_usage_query_failed", "account_id", accountID, "window_start", start, "error", err)
  1047. return nil
  1048. }
  1049. return windowStatsFromAccountStats(stats)
  1050. }
  1051. func grokLocalUsageForBilling(
  1052. ctx context.Context,
  1053. repo UsageLogRepository,
  1054. accountID int64,
  1055. billing *xai.BillingSummary,
  1056. now time.Time,
  1057. ) (*WindowStats, *WindowStats) {
  1058. var weekly *WindowStats
  1059. var monthly *WindowStats
  1060. if repo == nil || accountID <= 0 {
  1061. return weekly, monthly
  1062. }
  1063. if start, ok := currentGrokBillingWindow(billing, true, now); ok {
  1064. if stats, err := repo.GetAccountWindowStats(ctx, accountID, start); err == nil {
  1065. weekly = windowStatsFromAccountStats(stats)
  1066. } else {
  1067. slog.Warn("grok_window_usage_query_failed", "account_id", accountID, "window_start", start, "error", err)
  1068. }
  1069. }
  1070. if start, ok := currentGrokBillingWindow(billing, false, now); ok {
  1071. if stats, err := repo.GetAccountWindowStats(ctx, accountID, start); err == nil {
  1072. monthly = windowStatsFromAccountStats(stats)
  1073. } else {
  1074. slog.Warn("grok_monthly_usage_query_failed", "account_id", accountID, "window_start", start, "error", err)
  1075. }
  1076. }
  1077. return weekly, monthly
  1078. }
  1079. func currentGrokBillingWindow(billing *xai.BillingSummary, weekly bool, now time.Time) (time.Time, bool) {
  1080. if billing == nil {
  1081. return time.Time{}, false
  1082. }
  1083. startRaw, endRaw := billing.BillingPeriodStart, billing.BillingPeriodEnd
  1084. if weekly {
  1085. if billing.PeriodType != "weekly" {
  1086. return time.Time{}, false
  1087. }
  1088. startRaw, endRaw = billing.PeriodStart, billing.PeriodEnd
  1089. }
  1090. start, startErr := parseTime(strings.TrimSpace(startRaw))
  1091. end, endErr := parseTime(strings.TrimSpace(endRaw))
  1092. if startErr != nil || endErr != nil || now.Before(start) || !now.Before(end) {
  1093. return time.Time{}, false
  1094. }
  1095. return start, true
  1096. }
  1097. func grokBillingSnapshotNeedsRefresh(account *Account, now time.Time) bool {
  1098. if account == nil {
  1099. return false
  1100. }
  1101. billing, err := grokBillingSnapshotFromExtra(account.Extra)
  1102. if err != nil || billing == nil || billing.Partial || len(billing.FailedWindows) > 0 {
  1103. return true
  1104. }
  1105. stamp := strings.TrimSpace(billing.UpdatedAt)
  1106. if stamp == "" {
  1107. stamp = strings.TrimSpace(billing.FetchedAt)
  1108. }
  1109. updatedAt, err := parseTime(stamp)
  1110. return err != nil || now.Sub(updatedAt) >= openAIProbeCacheTTL
  1111. }
  1112. func (s *AccountUsageService) shouldProbeGrokBilling(accountID int64, now time.Time, force bool) bool {
  1113. if force || s == nil || s.cache == nil || accountID <= 0 {
  1114. return true
  1115. }
  1116. if cached, ok := s.cache.grokProbeCache.Load(accountID); ok {
  1117. if ts, ok := cached.(time.Time); ok && now.Sub(ts) < grokProbeRetryTTL {
  1118. return false
  1119. }
  1120. }
  1121. s.cache.grokProbeCache.Store(accountID, now)
  1122. return true
  1123. }
  1124. // recalcAntigravityRemainingSeconds 重新计算 Antigravity UsageInfo 中各窗口的 RemainingSeconds
  1125. // 用于从缓存取出时更新倒计时,避免返回过时的剩余秒数
  1126. func recalcAntigravityRemainingSeconds(info *UsageInfo) {
  1127. if info == nil {
  1128. return
  1129. }
  1130. if info.FiveHour != nil && info.FiveHour.ResetsAt != nil {
  1131. remaining := int(time.Until(*info.FiveHour.ResetsAt).Seconds())
  1132. if remaining < 0 {
  1133. remaining = 0
  1134. }
  1135. info.FiveHour.RemainingSeconds = remaining
  1136. }
  1137. }
  1138. // antigravityCacheTTL 根据 UsageInfo 内容决定缓存 TTL
  1139. // 403 forbidden 状态稳定,缓存与成功相同(3 分钟);
  1140. // 其他错误(401/网络)可能快速恢复,缓存 1 分钟。
  1141. func antigravityCacheTTL(info *UsageInfo) time.Duration {
  1142. if info == nil {
  1143. return antigravityErrorTTL
  1144. }
  1145. if info.IsForbidden {
  1146. return apiCacheTTL // 封号/验证状态不会很快变
  1147. }
  1148. if info.ErrorCode != "" || info.Error != "" {
  1149. return antigravityErrorTTL
  1150. }
  1151. return apiCacheTTL
  1152. }
  1153. // buildAntigravityDegradedUsage 从 FetchQuota 错误构建降级 UsageInfo
  1154. func buildAntigravityDegradedUsage(err error) *UsageInfo {
  1155. now := time.Now()
  1156. errMsg := fmt.Sprintf("usage API error: %v", err)
  1157. slog.Warn("antigravity usage fetch failed, returning degraded response", "error", err)
  1158. info := &UsageInfo{
  1159. UpdatedAt: &now,
  1160. Error: errMsg,
  1161. }
  1162. // 从错误信息推断 error_code 和状态标记
  1163. // 错误格式来自 antigravity/client.go: "fetchAvailableModels 失败 (HTTP %d): ..."
  1164. errStr := err.Error()
  1165. switch {
  1166. case strings.Contains(errStr, "HTTP 401") ||
  1167. strings.Contains(errStr, "UNAUTHENTICATED") ||
  1168. strings.Contains(errStr, "invalid_grant"):
  1169. info.ErrorCode = errorCodeUnauthenticated
  1170. info.NeedsReauth = true
  1171. case strings.Contains(errStr, "HTTP 429"):
  1172. info.ErrorCode = errorCodeRateLimited
  1173. default:
  1174. info.ErrorCode = errorCodeNetworkError
  1175. }
  1176. return info
  1177. }
  1178. // enrichUsageWithAccountError 结合账号错误状态修正 UsageInfo
  1179. // 场景 1(成功路径):FetchAvailableModels 正常返回,但账号已因 403 被标记为 error,
  1180. //
  1181. // 需要在正常 usage 数据上附加 forbidden/validation 信息。
  1182. //
  1183. // 场景 2(降级路径):被封号的账号 OAuth token 失效,FetchAvailableModels 返回 401,
  1184. //
  1185. // 降级逻辑设置了 needs_reauth,但账号实际是 403 封号/需验证,需覆盖为正确状态。
  1186. func enrichUsageWithAccountError(info *UsageInfo, account *Account) {
  1187. if info == nil || account == nil || account.Status != StatusError {
  1188. return
  1189. }
  1190. msg := strings.ToLower(account.ErrorMessage)
  1191. if !strings.Contains(msg, "403") && !strings.Contains(msg, "forbidden") &&
  1192. !strings.Contains(msg, "violation") && !strings.Contains(msg, "validation") {
  1193. return
  1194. }
  1195. fbType := classifyForbiddenType(account.ErrorMessage)
  1196. info.IsForbidden = true
  1197. info.ForbiddenType = fbType
  1198. info.ForbiddenReason = account.ErrorMessage
  1199. info.NeedsVerify = fbType == forbiddenTypeValidation
  1200. info.IsBanned = fbType == forbiddenTypeViolation
  1201. info.ValidationURL = extractValidationURL(account.ErrorMessage)
  1202. info.ErrorCode = errorCodeForbidden
  1203. info.NeedsReauth = false
  1204. }
  1205. // addWindowStats 为 usage 数据添加窗口期统计
  1206. // 使用独立缓存(1 分钟),与 API 缓存分离
  1207. func (s *AccountUsageService) addWindowStats(ctx context.Context, account *Account, usage *UsageInfo) {
  1208. // 修复:即使 FiveHour 为 nil,也要尝试获取统计数据
  1209. // 因为 SevenDay/SevenDaySonnet/SevenDayFable 可能需要
  1210. if usage.FiveHour == nil && usage.SevenDay == nil && usage.SevenDaySonnet == nil && usage.SevenDayFable == nil {
  1211. return
  1212. }
  1213. // 检查窗口统计缓存(1 分钟)
  1214. var windowStats *WindowStats
  1215. if cached, ok := s.cache.windowStatsCache.Load(account.ID); ok {
  1216. if cache, ok := cached.(*windowStatsCache); ok && time.Since(cache.timestamp) < windowStatsCacheTTL {
  1217. windowStats = cache.stats
  1218. }
  1219. }
  1220. // 如果没有缓存,从数据库查询
  1221. if windowStats == nil {
  1222. // 使用统一的窗口开始时间计算逻辑(考虑窗口过期情况)
  1223. startTime := account.GetCurrentWindowStartTime()
  1224. stats, err := s.usageLogRepo.GetAccountWindowStats(ctx, account.ID, startTime)
  1225. if err != nil {
  1226. log.Printf("Failed to get window stats for account %d: %v", account.ID, err)
  1227. return
  1228. }
  1229. windowStats = &WindowStats{
  1230. Requests: stats.Requests,
  1231. Tokens: stats.Tokens,
  1232. Cost: stats.Cost,
  1233. StandardCost: stats.StandardCost,
  1234. UserCost: stats.UserCost,
  1235. }
  1236. // 缓存窗口统计(1 分钟)
  1237. s.cache.windowStatsCache.Store(account.ID, &windowStatsCache{
  1238. stats: windowStats,
  1239. timestamp: time.Now(),
  1240. })
  1241. }
  1242. // 为 FiveHour 添加 WindowStats(5h 窗口统计)
  1243. if usage.FiveHour != nil {
  1244. usage.FiveHour.WindowStats = windowStats
  1245. }
  1246. }
  1247. // GetTodayStats 获取账号今日统计
  1248. func (s *AccountUsageService) GetTodayStats(ctx context.Context, accountID int64) (*WindowStats, error) {
  1249. stats, err := s.usageLogRepo.GetAccountTodayStats(ctx, accountID)
  1250. if err != nil {
  1251. return nil, fmt.Errorf("get today stats failed: %w", err)
  1252. }
  1253. return &WindowStats{
  1254. Requests: stats.Requests,
  1255. Tokens: stats.Tokens,
  1256. Cost: stats.Cost,
  1257. StandardCost: stats.StandardCost,
  1258. UserCost: stats.UserCost,
  1259. }, nil
  1260. }
  1261. // GetTodayStatsBatch 批量获取账号今日统计,优先走批量 SQL,失败时回退单账号查询。
  1262. func (s *AccountUsageService) GetTodayStatsBatch(ctx context.Context, accountIDs []int64) (map[int64]*WindowStats, error) {
  1263. uniqueIDs := make([]int64, 0, len(accountIDs))
  1264. seen := make(map[int64]struct{}, len(accountIDs))
  1265. for _, accountID := range accountIDs {
  1266. if accountID <= 0 {
  1267. continue
  1268. }
  1269. if _, exists := seen[accountID]; exists {
  1270. continue
  1271. }
  1272. seen[accountID] = struct{}{}
  1273. uniqueIDs = append(uniqueIDs, accountID)
  1274. }
  1275. result := make(map[int64]*WindowStats, len(uniqueIDs))
  1276. if len(uniqueIDs) == 0 {
  1277. return result, nil
  1278. }
  1279. startTime := timezone.Today()
  1280. if batchReader, ok := s.usageLogRepo.(accountWindowStatsBatchReader); ok {
  1281. statsByAccount, err := batchReader.GetAccountWindowStatsBatch(ctx, uniqueIDs, startTime)
  1282. if err == nil {
  1283. for _, accountID := range uniqueIDs {
  1284. result[accountID] = windowStatsFromAccountStats(statsByAccount[accountID])
  1285. }
  1286. return result, nil
  1287. }
  1288. }
  1289. var mu sync.Mutex
  1290. g, gctx := errgroup.WithContext(ctx)
  1291. g.SetLimit(8)
  1292. for _, accountID := range uniqueIDs {
  1293. id := accountID
  1294. g.Go(func() error {
  1295. stats, err := s.usageLogRepo.GetAccountWindowStats(gctx, id, startTime)
  1296. if err != nil {
  1297. return nil
  1298. }
  1299. mu.Lock()
  1300. result[id] = windowStatsFromAccountStats(stats)
  1301. mu.Unlock()
  1302. return nil
  1303. })
  1304. }
  1305. _ = g.Wait()
  1306. for _, accountID := range uniqueIDs {
  1307. if _, ok := result[accountID]; !ok {
  1308. result[accountID] = &WindowStats{}
  1309. }
  1310. }
  1311. return result, nil
  1312. }
  1313. func windowStatsFromAccountStats(stats *usagestats.AccountStats) *WindowStats {
  1314. if stats == nil {
  1315. return &WindowStats{}
  1316. }
  1317. return &WindowStats{
  1318. Requests: stats.Requests,
  1319. Tokens: stats.Tokens,
  1320. Cost: stats.Cost,
  1321. StandardCost: stats.StandardCost,
  1322. UserCost: stats.UserCost,
  1323. }
  1324. }
  1325. func buildCodexUsageProgressFromExtra(extra map[string]any, window string, now time.Time) *UsageProgress {
  1326. if len(extra) == 0 {
  1327. return nil
  1328. }
  1329. var (
  1330. usedPercentKey string
  1331. resetAfterKey string
  1332. resetAtKey string
  1333. )
  1334. switch window {
  1335. case "5h":
  1336. usedPercentKey = "codex_5h_used_percent"
  1337. resetAfterKey = "codex_5h_reset_after_seconds"
  1338. resetAtKey = "codex_5h_reset_at"
  1339. case "7d":
  1340. usedPercentKey = "codex_7d_used_percent"
  1341. resetAfterKey = "codex_7d_reset_after_seconds"
  1342. resetAtKey = "codex_7d_reset_at"
  1343. default:
  1344. return nil
  1345. }
  1346. usedRaw, ok := extra[usedPercentKey]
  1347. if !ok {
  1348. return nil
  1349. }
  1350. progress := &UsageProgress{Utilization: parseExtraFloat64(usedRaw)}
  1351. if resetAtRaw, ok := extra[resetAtKey]; ok {
  1352. if resetAt, err := parseTime(fmt.Sprint(resetAtRaw)); err == nil {
  1353. progress.ResetsAt = &resetAt
  1354. progress.RemainingSeconds = int(time.Until(resetAt).Seconds())
  1355. if progress.RemainingSeconds < 0 {
  1356. progress.RemainingSeconds = 0
  1357. }
  1358. }
  1359. }
  1360. if progress.ResetsAt == nil {
  1361. if resetAfterSeconds := parseExtraInt(extra[resetAfterKey]); resetAfterSeconds > 0 {
  1362. base := now
  1363. if updatedAtRaw, ok := extra["codex_usage_updated_at"]; ok {
  1364. if updatedAt, err := parseTime(fmt.Sprint(updatedAtRaw)); err == nil {
  1365. base = updatedAt
  1366. }
  1367. }
  1368. resetAt := base.Add(time.Duration(resetAfterSeconds) * time.Second)
  1369. progress.ResetsAt = &resetAt
  1370. progress.RemainingSeconds = int(time.Until(resetAt).Seconds())
  1371. if progress.RemainingSeconds < 0 {
  1372. progress.RemainingSeconds = 0
  1373. }
  1374. }
  1375. }
  1376. // 窗口已过期(resetAt 在 now 之前)→ 额度已重置,归零
  1377. if progress.ResetsAt != nil && !now.Before(*progress.ResetsAt) {
  1378. progress.Utilization = 0
  1379. }
  1380. return progress
  1381. }
  1382. func codexWindowStatsStart(progress *UsageProgress, fallbackWindow time.Duration, now time.Time) time.Time {
  1383. if progress != nil && progress.ResetsAt != nil && now.Before(*progress.ResetsAt) {
  1384. return progress.ResetsAt.Add(-fallbackWindow)
  1385. }
  1386. return now.Add(-fallbackWindow)
  1387. }
  1388. func (s *AccountUsageService) GetAccountUsageStats(ctx context.Context, accountID int64, startTime, endTime time.Time) (*usagestats.AccountUsageStatsResponse, error) {
  1389. stats, err := s.usageLogRepo.GetAccountUsageStats(ctx, accountID, startTime, endTime)
  1390. if err != nil {
  1391. return nil, fmt.Errorf("get account usage stats failed: %w", err)
  1392. }
  1393. return stats, nil
  1394. }
  1395. // fetchOAuthUsageRaw 从 Anthropic API 获取原始响应(不构建 UsageInfo)
  1396. // 如果账号开启了 TLS 指纹,则使用 TLS 指纹伪装
  1397. // 如果有缓存的 Fingerprint,则使用缓存的 User-Agent 等信息
  1398. func (s *AccountUsageService) fetchOAuthUsageRaw(ctx context.Context, account *Account) (*ClaudeUsageResponse, error) {
  1399. accessToken := account.GetCredential("access_token")
  1400. if accessToken == "" {
  1401. return nil, fmt.Errorf("no access token available")
  1402. }
  1403. var proxyURL string
  1404. if account.ProxyID != nil && account.Proxy != nil {
  1405. proxyURL = account.Proxy.URL()
  1406. }
  1407. // 构建完整的选项
  1408. opts := &ClaudeUsageFetchOptions{
  1409. AccessToken: accessToken,
  1410. ProxyURL: proxyURL,
  1411. AccountID: account.ID,
  1412. TLSProfile: s.tlsFPProfileService.ResolveTLSProfile(account),
  1413. }
  1414. // 尝试获取缓存的 Fingerprint(包含 User-Agent 等信息)
  1415. if s.identityCache != nil {
  1416. if fp, err := s.identityCache.GetFingerprint(ctx, account.ID); err == nil && fp != nil {
  1417. opts.Fingerprint = fp
  1418. }
  1419. }
  1420. return s.usageFetcher.FetchUsageWithOptions(ctx, opts)
  1421. }
  1422. // parseTime 尝试多种格式解析时间
  1423. func parseTime(s string) (time.Time, error) {
  1424. formats := []string{
  1425. time.RFC3339,
  1426. time.RFC3339Nano,
  1427. "2006-01-02T15:04:05Z",
  1428. "2006-01-02T15:04:05.000Z",
  1429. }
  1430. for _, format := range formats {
  1431. if t, err := time.Parse(format, s); err == nil {
  1432. return t, nil
  1433. }
  1434. }
  1435. return time.Time{}, fmt.Errorf("unable to parse time: %s", s)
  1436. }
  1437. func (s *AccountUsageService) tryClearRecoverableAccountError(ctx context.Context, account *Account) {
  1438. if account == nil || account.Status != StatusError {
  1439. return
  1440. }
  1441. msg := strings.ToLower(strings.TrimSpace(account.ErrorMessage))
  1442. if msg == "" {
  1443. return
  1444. }
  1445. if !strings.Contains(msg, "token refresh failed") &&
  1446. !strings.Contains(msg, "invalid_client") &&
  1447. !strings.Contains(msg, "missing_project_id") &&
  1448. !strings.Contains(msg, "unauthenticated") {
  1449. return
  1450. }
  1451. if err := s.accountRepo.ClearError(ctx, account.ID); err != nil {
  1452. log.Printf("[usage] failed to clear recoverable account error for account %d: %v", account.ID, err)
  1453. return
  1454. }
  1455. account.Status = StatusActive
  1456. account.ErrorMessage = ""
  1457. }
  1458. // buildUsageInfo 构建UsageInfo
  1459. func (s *AccountUsageService) buildUsageInfo(resp *ClaudeUsageResponse, updatedAt *time.Time) *UsageInfo {
  1460. info := &UsageInfo{
  1461. UpdatedAt: updatedAt,
  1462. }
  1463. // 5小时窗口 - 始终创建对象(即使 ResetsAt 为空)
  1464. info.FiveHour = &UsageProgress{
  1465. Utilization: resp.FiveHour.Utilization,
  1466. }
  1467. if resp.FiveHour.ResetsAt != "" {
  1468. if fiveHourReset, err := parseTime(resp.FiveHour.ResetsAt); err == nil {
  1469. info.FiveHour.ResetsAt = &fiveHourReset
  1470. info.FiveHour.RemainingSeconds = int(time.Until(fiveHourReset).Seconds())
  1471. } else {
  1472. log.Printf("Failed to parse FiveHour.ResetsAt: %s, error: %v", resp.FiveHour.ResetsAt, err)
  1473. }
  1474. }
  1475. // 7天窗口
  1476. if resp.SevenDay.ResetsAt != "" {
  1477. if sevenDayReset, err := parseTime(resp.SevenDay.ResetsAt); err == nil {
  1478. info.SevenDay = &UsageProgress{
  1479. Utilization: resp.SevenDay.Utilization,
  1480. ResetsAt: &sevenDayReset,
  1481. RemainingSeconds: int(time.Until(sevenDayReset).Seconds()),
  1482. }
  1483. } else {
  1484. log.Printf("Failed to parse SevenDay.ResetsAt: %s, error: %v", resp.SevenDay.ResetsAt, err)
  1485. info.SevenDay = &UsageProgress{
  1486. Utilization: resp.SevenDay.Utilization,
  1487. }
  1488. }
  1489. }
  1490. // 7天Sonnet窗口
  1491. if resp.SevenDaySonnet.ResetsAt != "" {
  1492. if sonnetReset, err := parseTime(resp.SevenDaySonnet.ResetsAt); err == nil {
  1493. info.SevenDaySonnet = &UsageProgress{
  1494. Utilization: resp.SevenDaySonnet.Utilization,
  1495. ResetsAt: &sonnetReset,
  1496. RemainingSeconds: int(time.Until(sonnetReset).Seconds()),
  1497. }
  1498. } else {
  1499. log.Printf("Failed to parse SevenDaySonnet.ResetsAt: %s, error: %v", resp.SevenDaySonnet.ResetsAt, err)
  1500. info.SevenDaySonnet = &UsageProgress{
  1501. Utilization: resp.SevenDaySonnet.Utilization,
  1502. }
  1503. }
  1504. }
  1505. // 7天Fable窗口(响应头 7d_oi 对应的窗口)
  1506. if fable := resp.SevenDayOverageIncluded; fable.ResetsAt != "" {
  1507. if fableReset, err := parseTime(fable.ResetsAt); err == nil {
  1508. info.SevenDayFable = &UsageProgress{
  1509. Utilization: fable.Utilization,
  1510. ResetsAt: &fableReset,
  1511. RemainingSeconds: int(time.Until(fableReset).Seconds()),
  1512. }
  1513. } else {
  1514. log.Printf("Failed to parse SevenDayFable.ResetsAt: %s, error: %v", fable.ResetsAt, err)
  1515. info.SevenDayFable = &UsageProgress{
  1516. Utilization: fable.Utilization,
  1517. }
  1518. }
  1519. }
  1520. return info
  1521. }
  1522. // estimateSetupTokenUsage 根据session_window推算Setup Token账号的使用量
  1523. func (s *AccountUsageService) estimateSetupTokenUsage(account *Account) *UsageInfo {
  1524. info := &UsageInfo{}
  1525. // 如果有session_window信息
  1526. if account.SessionWindowEnd != nil {
  1527. remaining := int(time.Until(*account.SessionWindowEnd).Seconds())
  1528. if remaining < 0 {
  1529. remaining = 0
  1530. }
  1531. // 优先使用响应头中存储的真实 utilization 值(0-1 小数,转为 0-100 百分比)
  1532. var utilization float64
  1533. var found bool
  1534. if stored, ok := account.Extra["session_window_utilization"]; ok {
  1535. switch v := stored.(type) {
  1536. case float64:
  1537. utilization = v * 100
  1538. found = true
  1539. case json.Number:
  1540. if f, err := v.Float64(); err == nil {
  1541. utilization = f * 100
  1542. found = true
  1543. }
  1544. }
  1545. }
  1546. // 如果没有存储的 utilization,回退到状态估算
  1547. if !found {
  1548. switch account.SessionWindowStatus {
  1549. case "rejected":
  1550. utilization = 100.0
  1551. case "allowed_warning":
  1552. utilization = 80.0
  1553. }
  1554. }
  1555. info.FiveHour = &UsageProgress{
  1556. Utilization: utilization,
  1557. ResetsAt: account.SessionWindowEnd,
  1558. RemainingSeconds: remaining,
  1559. }
  1560. // 窗口已过期(resetAt 在 now 之前)→ 额度已重置,归零;
  1561. // 与 Codex 分支 buildCodexUsageProgressFromExtra 保持一致,避免
  1562. // UI 在 active poll 没回写 SessionWindowEnd 时渲染矛盾状态。
  1563. if info.FiveHour.ResetsAt != nil && !time.Now().Before(*info.FiveHour.ResetsAt) {
  1564. info.FiveHour.Utilization = 0
  1565. info.FiveHour.ResetsAt = nil
  1566. info.FiveHour.RemainingSeconds = 0
  1567. }
  1568. } else {
  1569. // 没有窗口信息,返回空数据
  1570. info.FiveHour = &UsageProgress{
  1571. Utilization: 0,
  1572. RemainingSeconds: 0,
  1573. }
  1574. }
  1575. // Setup Token无法获取7d数据
  1576. return info
  1577. }
  1578. func buildGeminiUsageProgress(used, limit int64, resetAt time.Time, tokens int64, cost float64, now time.Time) *UsageProgress {
  1579. // limit <= 0 means "no local quota window" (unknown or unlimited).
  1580. if limit <= 0 {
  1581. return nil
  1582. }
  1583. utilization := (float64(used) / float64(limit)) * 100
  1584. remainingSeconds := int(resetAt.Sub(now).Seconds())
  1585. if remainingSeconds < 0 {
  1586. remainingSeconds = 0
  1587. }
  1588. resetCopy := resetAt
  1589. return &UsageProgress{
  1590. Utilization: utilization,
  1591. ResetsAt: &resetCopy,
  1592. RemainingSeconds: remainingSeconds,
  1593. UsedRequests: used,
  1594. LimitRequests: limit,
  1595. WindowStats: &WindowStats{
  1596. Requests: used,
  1597. Tokens: tokens,
  1598. Cost: cost,
  1599. },
  1600. }
  1601. }
  1602. // GetAccountWindowStats 获取账号在指定时间窗口内的使用统计
  1603. // 用于账号列表页面显示当前窗口费用
  1604. func (s *AccountUsageService) GetAccountWindowStats(ctx context.Context, accountID int64, startTime time.Time) (*usagestats.AccountStats, error) {
  1605. return s.usageLogRepo.GetAccountWindowStats(ctx, accountID, startTime)
  1606. }