usage_log_repo_query.go 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786
  1. package repository
  2. import (
  3. "context"
  4. "database/sql"
  5. "encoding/json"
  6. "fmt"
  7. "strings"
  8. "time"
  9. dbaccount "github.com/Wei-Shaw/sub2api/ent/account"
  10. dbapikey "github.com/Wei-Shaw/sub2api/ent/apikey"
  11. dbgroup "github.com/Wei-Shaw/sub2api/ent/group"
  12. "github.com/Wei-Shaw/sub2api/ent/schema/mixins"
  13. dbuser "github.com/Wei-Shaw/sub2api/ent/user"
  14. dbusersub "github.com/Wei-Shaw/sub2api/ent/usersubscription"
  15. "github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
  16. "github.com/Wei-Shaw/sub2api/internal/pkg/usagestats"
  17. "github.com/Wei-Shaw/sub2api/internal/service"
  18. )
  19. const usageLogSelectColumns = "id, user_id, api_key_id, account_id, request_id, model, requested_model, upstream_model, upstream_response_model, upstream_model_mismatch, group_id, subscription_id, input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens, cache_creation_5m_tokens, cache_creation_1h_tokens, image_output_tokens, image_output_cost, image_input_tokens, image_input_cost, input_cost, output_cost, cache_creation_cost, cache_read_cost, total_cost, actual_cost, rate_multiplier, account_rate_multiplier, billing_type, request_type, stream, openai_ws_mode, duration_ms, first_token_ms, user_agent, ip_address, image_count, image_size, image_input_size, image_output_size, image_size_source, image_size_breakdown, video_count, video_resolution, video_duration_seconds, service_tier, reasoning_effort, requested_reasoning_effort, inbound_endpoint, upstream_endpoint, cache_ttl_overridden, long_context_billing_applied, channel_id, model_mapping_chain, billing_tier, billing_mode, account_stats_cost, upstream_request_id, session_id, native_compaction_v2, created_at"
  20. func (r *usageLogRepository) GetByID(ctx context.Context, id int64) (log *service.UsageLog, err error) {
  21. query := "SELECT " + usageLogSelectColumns + " FROM usage_logs WHERE id = $1"
  22. rows, err := r.sql.QueryContext(ctx, query, id)
  23. if err != nil {
  24. return nil, err
  25. }
  26. defer func() {
  27. // 保持主错误优先;仅在无错误时回传 Close 失败。
  28. // 同时清空返回值,避免误用不完整结果。
  29. if closeErr := rows.Close(); closeErr != nil && err == nil {
  30. err = closeErr
  31. log = nil
  32. }
  33. }()
  34. if !rows.Next() {
  35. if err = rows.Err(); err != nil {
  36. return nil, err
  37. }
  38. return nil, service.ErrUsageLogNotFound
  39. }
  40. log, err = scanUsageLog(rows)
  41. if err != nil {
  42. return nil, err
  43. }
  44. if err = rows.Err(); err != nil {
  45. return nil, err
  46. }
  47. return log, nil
  48. }
  49. func (r *usageLogRepository) ListByUser(ctx context.Context, userID int64, params pagination.PaginationParams) ([]service.UsageLog, *pagination.PaginationResult, error) {
  50. return r.listUsageLogsWithPagination(ctx, "WHERE user_id = $1", []any{userID}, params)
  51. }
  52. func (r *usageLogRepository) ListByAPIKey(ctx context.Context, apiKeyID int64, params pagination.PaginationParams) ([]service.UsageLog, *pagination.PaginationResult, error) {
  53. return r.listUsageLogsWithPagination(ctx, "WHERE api_key_id = $1", []any{apiKeyID}, params)
  54. }
  55. func (r *usageLogRepository) ListByAccount(ctx context.Context, accountID int64, params pagination.PaginationParams) ([]service.UsageLog, *pagination.PaginationResult, error) {
  56. return r.listUsageLogsWithPagination(ctx, "WHERE account_id = $1", []any{accountID}, params)
  57. }
  58. func (r *usageLogRepository) ListByUserAndTimeRange(ctx context.Context, userID int64, startTime, endTime time.Time) ([]service.UsageLog, *pagination.PaginationResult, error) {
  59. query := "SELECT " + usageLogSelectColumns + " FROM usage_logs WHERE user_id = $1 AND created_at >= $2 AND created_at < $3 ORDER BY id DESC LIMIT 10000"
  60. logs, err := r.queryUsageLogs(ctx, query, userID, startTime, endTime)
  61. return logs, nil, err
  62. }
  63. func (r *usageLogRepository) ListByAPIKeyAndTimeRange(ctx context.Context, apiKeyID int64, startTime, endTime time.Time) ([]service.UsageLog, *pagination.PaginationResult, error) {
  64. query := "SELECT " + usageLogSelectColumns + " FROM usage_logs WHERE api_key_id = $1 AND created_at >= $2 AND created_at < $3 ORDER BY id DESC LIMIT 10000"
  65. logs, err := r.queryUsageLogs(ctx, query, apiKeyID, startTime, endTime)
  66. return logs, nil, err
  67. }
  68. func (r *usageLogRepository) ListByAccountAndTimeRange(ctx context.Context, accountID int64, startTime, endTime time.Time) ([]service.UsageLog, *pagination.PaginationResult, error) {
  69. query := "SELECT " + usageLogSelectColumns + " FROM usage_logs WHERE account_id = $1 AND created_at >= $2 AND created_at < $3 ORDER BY id DESC LIMIT 10000"
  70. logs, err := r.queryUsageLogs(ctx, query, accountID, startTime, endTime)
  71. return logs, nil, err
  72. }
  73. func (r *usageLogRepository) ListByModelAndTimeRange(ctx context.Context, modelName string, startTime, endTime time.Time) ([]service.UsageLog, *pagination.PaginationResult, error) {
  74. query := fmt.Sprintf("SELECT %s FROM usage_logs WHERE %s = $1 AND created_at >= $2 AND created_at < $3 ORDER BY id DESC LIMIT 10000", usageLogSelectColumns, rawUsageLogModelColumn)
  75. logs, err := r.queryUsageLogs(ctx, query, modelName, startTime, endTime)
  76. return logs, nil, err
  77. }
  78. func (r *usageLogRepository) Delete(ctx context.Context, id int64) error {
  79. _, err := r.sql.ExecContext(ctx, "DELETE FROM usage_logs WHERE id = $1", id)
  80. return err
  81. }
  82. // UsageLogFilters represents filters for usage log queries
  83. type UsageLogFilters = usagestats.UsageLogFilters
  84. // ListWithFilters lists usage logs with optional filters (for admin)
  85. func (r *usageLogRepository) ListWithFilters(ctx context.Context, params pagination.PaginationParams, filters UsageLogFilters) ([]service.UsageLog, *pagination.PaginationResult, error) {
  86. conditions := make([]string, 0, 9)
  87. args := make([]any, 0, 9)
  88. if filters.UserID > 0 {
  89. conditions = append(conditions, fmt.Sprintf("user_id = $%d", len(args)+1))
  90. args = append(args, filters.UserID)
  91. }
  92. if filters.APIKeyID > 0 {
  93. conditions = append(conditions, fmt.Sprintf("api_key_id = $%d", len(args)+1))
  94. args = append(args, filters.APIKeyID)
  95. }
  96. if filters.AccountID > 0 {
  97. conditions = append(conditions, fmt.Sprintf("account_id = $%d", len(args)+1))
  98. args = append(args, filters.AccountID)
  99. }
  100. if filters.GroupID > 0 {
  101. conditions = append(conditions, fmt.Sprintf("group_id = $%d", len(args)+1))
  102. args = append(args, filters.GroupID)
  103. }
  104. if requestID := strings.TrimSpace(filters.RequestID); requestID != "" {
  105. conditions = append(conditions, fmt.Sprintf("request_id = $%d", len(args)+1))
  106. args = append(args, requestID)
  107. }
  108. conditions, args = appendUsageLogModelWhereCondition(conditions, args, filters.Model, filters.ModelFilterSource)
  109. conditions, args = appendRequestTypeOrStreamWhereCondition(conditions, args, filters.RequestType, filters.Stream)
  110. conditions, args = appendNativeCompactionV2WhereCondition(conditions, args, filters.NativeCompactionV2, "")
  111. if filters.BillingType != nil {
  112. conditions = append(conditions, fmt.Sprintf("billing_type = $%d", len(args)+1))
  113. args = append(args, int16(*filters.BillingType))
  114. }
  115. conditions, args = appendUsageLogBillingModeWhereCondition(conditions, args, filters.BillingMode)
  116. if filters.UpstreamModelMismatch != nil {
  117. conditions = append(conditions, upstreamModelMismatchCondition("upstream_model_mismatch", *filters.UpstreamModelMismatch))
  118. }
  119. if filters.StartTime != nil {
  120. conditions = append(conditions, fmt.Sprintf("created_at >= $%d", len(args)+1))
  121. args = append(args, *filters.StartTime)
  122. }
  123. if filters.EndTime != nil {
  124. conditions = append(conditions, fmt.Sprintf("created_at < $%d", len(args)+1))
  125. args = append(args, *filters.EndTime)
  126. }
  127. whereClause := buildWhere(conditions)
  128. var (
  129. logs []service.UsageLog
  130. page *pagination.PaginationResult
  131. err error
  132. )
  133. if shouldUseFastUsageLogTotal(filters) {
  134. logs, page, err = r.listUsageLogsWithFastPagination(ctx, whereClause, args, params)
  135. } else {
  136. logs, page, err = r.listUsageLogsWithPagination(ctx, whereClause, args, params)
  137. }
  138. if err != nil {
  139. return nil, nil, err
  140. }
  141. if err := r.hydrateUsageLogAssociations(ctx, logs); err != nil {
  142. return nil, nil, err
  143. }
  144. return logs, page, nil
  145. }
  146. func upstreamModelMismatchCondition(column string, mismatch bool) string {
  147. if mismatch {
  148. return column + " IS TRUE"
  149. }
  150. return column + " IS FALSE"
  151. }
  152. func shouldUseFastUsageLogTotal(filters UsageLogFilters) bool {
  153. if filters.ExactTotal {
  154. return false
  155. }
  156. // 强选择过滤下记录集通常较小,保留精确总数。
  157. return filters.UserID == 0 && filters.APIKeyID == 0 && filters.AccountID == 0
  158. }
  159. func (r *usageLogRepository) listUsageLogsWithPagination(ctx context.Context, whereClause string, args []any, params pagination.PaginationParams) ([]service.UsageLog, *pagination.PaginationResult, error) {
  160. countQuery := "SELECT COUNT(*) FROM usage_logs " + whereClause
  161. var total int64
  162. if err := scanSingleRow(ctx, r.sql, countQuery, args, &total); err != nil {
  163. return nil, nil, err
  164. }
  165. limitPos := len(args) + 1
  166. offsetPos := len(args) + 2
  167. listArgs := append(append([]any{}, args...), params.Limit(), params.Offset())
  168. query := fmt.Sprintf("SELECT %s FROM usage_logs %s ORDER BY %s LIMIT $%d OFFSET $%d", usageLogSelectColumns, whereClause, usageLogOrderBy(params), limitPos, offsetPos)
  169. logs, err := r.queryUsageLogs(ctx, query, listArgs...)
  170. if err != nil {
  171. return nil, nil, err
  172. }
  173. return logs, paginationResultFromTotal(total, params), nil
  174. }
  175. func (r *usageLogRepository) listUsageLogsWithFastPagination(ctx context.Context, whereClause string, args []any, params pagination.PaginationParams) ([]service.UsageLog, *pagination.PaginationResult, error) {
  176. limit := params.Limit()
  177. offset := params.Offset()
  178. limitPos := len(args) + 1
  179. offsetPos := len(args) + 2
  180. listArgs := append(append([]any{}, args...), limit+1, offset)
  181. query := fmt.Sprintf("SELECT %s FROM usage_logs %s ORDER BY %s LIMIT $%d OFFSET $%d", usageLogSelectColumns, whereClause, usageLogOrderBy(params), limitPos, offsetPos)
  182. logs, err := r.queryUsageLogs(ctx, query, listArgs...)
  183. if err != nil {
  184. return nil, nil, err
  185. }
  186. hasMore := false
  187. if len(logs) > limit {
  188. hasMore = true
  189. logs = logs[:limit]
  190. }
  191. total := int64(offset) + int64(len(logs))
  192. if hasMore {
  193. // 只保证“还有下一页”,避免对超大表做全量 COUNT(*)。
  194. total = int64(offset) + int64(limit) + 1
  195. }
  196. return logs, paginationResultFromTotal(total, params), nil
  197. }
  198. func usageLogOrderBy(params pagination.PaginationParams) string {
  199. sortBy := strings.ToLower(strings.TrimSpace(params.SortBy))
  200. sortOrder := strings.ToUpper(params.NormalizedSortOrder(pagination.SortOrderDesc))
  201. var column string
  202. switch sortBy {
  203. case "model":
  204. column = "COALESCE(NULLIF(TRIM(requested_model), ''), model)"
  205. case "created_at":
  206. column = "created_at"
  207. default:
  208. column = "id"
  209. }
  210. if column == "id" {
  211. return fmt.Sprintf("id %s", sortOrder)
  212. }
  213. return fmt.Sprintf("%s %s, id %s", column, sortOrder, sortOrder)
  214. }
  215. func (r *usageLogRepository) queryUsageLogs(ctx context.Context, query string, args ...any) (logs []service.UsageLog, err error) {
  216. rows, err := r.sql.QueryContext(ctx, query, args...)
  217. if err != nil {
  218. return nil, err
  219. }
  220. defer func() {
  221. // 保持主错误优先;仅在无错误时回传 Close 失败。
  222. // 同时清空返回值,避免误用不完整结果。
  223. if closeErr := rows.Close(); closeErr != nil && err == nil {
  224. err = closeErr
  225. logs = nil
  226. }
  227. }()
  228. logs = make([]service.UsageLog, 0)
  229. for rows.Next() {
  230. var log *service.UsageLog
  231. log, err = scanUsageLog(rows)
  232. if err != nil {
  233. return nil, err
  234. }
  235. logs = append(logs, *log)
  236. }
  237. if err = rows.Err(); err != nil {
  238. return nil, err
  239. }
  240. return logs, nil
  241. }
  242. func (r *usageLogRepository) hydrateUsageLogAssociations(ctx context.Context, logs []service.UsageLog) error {
  243. // 关联数据使用 Ent 批量加载,避免把复杂 SQL 继续膨胀。
  244. if len(logs) == 0 {
  245. return nil
  246. }
  247. ids := collectUsageLogIDs(logs)
  248. users, err := r.loadUsers(ctx, ids.userIDs)
  249. if err != nil {
  250. return err
  251. }
  252. apiKeys, err := r.loadAPIKeys(ctx, ids.apiKeyIDs)
  253. if err != nil {
  254. return err
  255. }
  256. accounts, err := r.loadAccounts(ctx, ids.accountIDs)
  257. if err != nil {
  258. return err
  259. }
  260. groups, err := r.loadGroups(ctx, ids.groupIDs)
  261. if err != nil {
  262. return err
  263. }
  264. subs, err := r.loadSubscriptions(ctx, ids.subscriptionIDs)
  265. if err != nil {
  266. return err
  267. }
  268. for i := range logs {
  269. if user, ok := users[logs[i].UserID]; ok {
  270. logs[i].User = user
  271. }
  272. if key, ok := apiKeys[logs[i].APIKeyID]; ok {
  273. logs[i].APIKey = key
  274. }
  275. if acc, ok := accounts[logs[i].AccountID]; ok {
  276. logs[i].Account = acc
  277. }
  278. if logs[i].GroupID != nil {
  279. if group, ok := groups[*logs[i].GroupID]; ok {
  280. logs[i].Group = group
  281. }
  282. }
  283. if logs[i].SubscriptionID != nil {
  284. if sub, ok := subs[*logs[i].SubscriptionID]; ok {
  285. logs[i].Subscription = sub
  286. }
  287. }
  288. }
  289. return nil
  290. }
  291. type usageLogIDs struct {
  292. userIDs []int64
  293. apiKeyIDs []int64
  294. accountIDs []int64
  295. groupIDs []int64
  296. subscriptionIDs []int64
  297. }
  298. func collectUsageLogIDs(logs []service.UsageLog) usageLogIDs {
  299. idSet := func() map[int64]struct{} { return make(map[int64]struct{}) }
  300. userIDs := idSet()
  301. apiKeyIDs := idSet()
  302. accountIDs := idSet()
  303. groupIDs := idSet()
  304. subscriptionIDs := idSet()
  305. for i := range logs {
  306. userIDs[logs[i].UserID] = struct{}{}
  307. apiKeyIDs[logs[i].APIKeyID] = struct{}{}
  308. accountIDs[logs[i].AccountID] = struct{}{}
  309. if logs[i].GroupID != nil {
  310. groupIDs[*logs[i].GroupID] = struct{}{}
  311. }
  312. if logs[i].SubscriptionID != nil {
  313. subscriptionIDs[*logs[i].SubscriptionID] = struct{}{}
  314. }
  315. }
  316. return usageLogIDs{
  317. userIDs: setToSlice(userIDs),
  318. apiKeyIDs: setToSlice(apiKeyIDs),
  319. accountIDs: setToSlice(accountIDs),
  320. groupIDs: setToSlice(groupIDs),
  321. subscriptionIDs: setToSlice(subscriptionIDs),
  322. }
  323. }
  324. func (r *usageLogRepository) loadUsers(ctx context.Context, ids []int64) (map[int64]*service.User, error) {
  325. out := make(map[int64]*service.User)
  326. if len(ids) == 0 {
  327. return out, nil
  328. }
  329. // 无条件穿透软删除:ids 来自调用方已按 user_id 筛选的日志行;普通用户路径强制 UserID=本人(本人必为活跃用户),不会借此解析他人已删身份;仅 admin 路径可借此显示已删用户。
  330. models, err := r.client.User.Query().Where(dbuser.IDIn(ids...)).All(mixins.SkipSoftDelete(ctx))
  331. if err != nil {
  332. return nil, err
  333. }
  334. for _, m := range models {
  335. out[m.ID] = userEntityToService(m)
  336. }
  337. return out, nil
  338. }
  339. func (r *usageLogRepository) loadAPIKeys(ctx context.Context, ids []int64) (map[int64]*service.APIKey, error) {
  340. out := make(map[int64]*service.APIKey)
  341. if len(ids) == 0 {
  342. return out, nil
  343. }
  344. models, err := r.client.APIKey.Query().Where(dbapikey.IDIn(ids...)).All(ctx)
  345. if err != nil {
  346. return nil, err
  347. }
  348. for _, m := range models {
  349. out[m.ID] = apiKeyEntityToService(m)
  350. }
  351. return out, nil
  352. }
  353. func (r *usageLogRepository) loadAccounts(ctx context.Context, ids []int64) (map[int64]*service.Account, error) {
  354. out := make(map[int64]*service.Account)
  355. if len(ids) == 0 {
  356. return out, nil
  357. }
  358. models, err := r.client.Account.Query().Where(dbaccount.IDIn(ids...)).All(ctx)
  359. if err != nil {
  360. return nil, err
  361. }
  362. for _, m := range models {
  363. out[m.ID] = accountEntityToService(m)
  364. }
  365. return out, nil
  366. }
  367. func (r *usageLogRepository) loadGroups(ctx context.Context, ids []int64) (map[int64]*service.Group, error) {
  368. out := make(map[int64]*service.Group)
  369. if len(ids) == 0 {
  370. return out, nil
  371. }
  372. models, err := r.client.Group.Query().Where(dbgroup.IDIn(ids...)).All(ctx)
  373. if err != nil {
  374. return nil, err
  375. }
  376. for _, m := range models {
  377. out[m.ID] = groupEntityToService(m)
  378. }
  379. return out, nil
  380. }
  381. func (r *usageLogRepository) loadSubscriptions(ctx context.Context, ids []int64) (map[int64]*service.UserSubscription, error) {
  382. out := make(map[int64]*service.UserSubscription)
  383. if len(ids) == 0 {
  384. return out, nil
  385. }
  386. models, err := r.client.UserSubscription.Query().Where(dbusersub.IDIn(ids...)).All(ctx)
  387. if err != nil {
  388. return nil, err
  389. }
  390. for _, m := range models {
  391. out[m.ID] = userSubscriptionEntityToService(m)
  392. }
  393. return out, nil
  394. }
  395. func scanUsageLog(scanner interface{ Scan(...any) error }) (*service.UsageLog, error) {
  396. var (
  397. id int64
  398. userID int64
  399. apiKeyID int64
  400. accountID int64
  401. requestID sql.NullString
  402. model string
  403. requestedModel sql.NullString
  404. upstreamModel sql.NullString
  405. upstreamResponseModel sql.NullString
  406. upstreamModelMismatch sql.NullBool
  407. groupID sql.NullInt64
  408. subscriptionID sql.NullInt64
  409. inputTokens int
  410. outputTokens int
  411. cacheCreationTokens int
  412. cacheReadTokens int
  413. cacheCreation5m int
  414. cacheCreation1h int
  415. imageOutputTokens int
  416. imageOutputCost float64
  417. imageInputTokens int
  418. imageInputCost float64
  419. inputCost float64
  420. outputCost float64
  421. cacheCreationCost float64
  422. cacheReadCost float64
  423. totalCost float64
  424. actualCost float64
  425. rateMultiplier float64
  426. accountRateMultiplier sql.NullFloat64
  427. billingType int16
  428. requestTypeRaw int16
  429. stream bool
  430. openaiWSMode bool
  431. durationMs sql.NullInt64
  432. firstTokenMs sql.NullInt64
  433. userAgent sql.NullString
  434. ipAddress sql.NullString
  435. imageCount int
  436. imageSize sql.NullString
  437. imageInputSize sql.NullString
  438. imageOutputSize sql.NullString
  439. imageSizeSource sql.NullString
  440. imageSizeBreakdown sql.NullString
  441. videoCount int
  442. videoResolution sql.NullString
  443. videoDurationSeconds sql.NullInt64
  444. serviceTier sql.NullString
  445. reasoningEffort sql.NullString
  446. requestedReasoningEffort sql.NullString
  447. inboundEndpoint sql.NullString
  448. upstreamEndpoint sql.NullString
  449. cacheTTLOverridden bool
  450. longContextBillingApplied bool
  451. channelID sql.NullInt64
  452. modelMappingChain sql.NullString
  453. billingTier sql.NullString
  454. billingMode sql.NullString
  455. accountStatsCost sql.NullFloat64
  456. upstreamRequestID sql.NullString
  457. sessionID sql.NullString
  458. nativeCompactionV2 bool
  459. createdAt time.Time
  460. )
  461. if err := scanner.Scan(
  462. &id,
  463. &userID,
  464. &apiKeyID,
  465. &accountID,
  466. &requestID,
  467. &model,
  468. &requestedModel,
  469. &upstreamModel,
  470. &upstreamResponseModel,
  471. &upstreamModelMismatch,
  472. &groupID,
  473. &subscriptionID,
  474. &inputTokens,
  475. &outputTokens,
  476. &cacheCreationTokens,
  477. &cacheReadTokens,
  478. &cacheCreation5m,
  479. &cacheCreation1h,
  480. &imageOutputTokens,
  481. &imageOutputCost,
  482. &imageInputTokens,
  483. &imageInputCost,
  484. &inputCost,
  485. &outputCost,
  486. &cacheCreationCost,
  487. &cacheReadCost,
  488. &totalCost,
  489. &actualCost,
  490. &rateMultiplier,
  491. &accountRateMultiplier,
  492. &billingType,
  493. &requestTypeRaw,
  494. &stream,
  495. &openaiWSMode,
  496. &durationMs,
  497. &firstTokenMs,
  498. &userAgent,
  499. &ipAddress,
  500. &imageCount,
  501. &imageSize,
  502. &imageInputSize,
  503. &imageOutputSize,
  504. &imageSizeSource,
  505. &imageSizeBreakdown,
  506. &videoCount,
  507. &videoResolution,
  508. &videoDurationSeconds,
  509. &serviceTier,
  510. &reasoningEffort,
  511. &requestedReasoningEffort,
  512. &inboundEndpoint,
  513. &upstreamEndpoint,
  514. &cacheTTLOverridden,
  515. &longContextBillingApplied,
  516. &channelID,
  517. &modelMappingChain,
  518. &billingTier,
  519. &billingMode,
  520. &accountStatsCost,
  521. &upstreamRequestID,
  522. &sessionID,
  523. &nativeCompactionV2,
  524. &createdAt,
  525. ); err != nil {
  526. return nil, err
  527. }
  528. log := &service.UsageLog{
  529. ID: id,
  530. UserID: userID,
  531. APIKeyID: apiKeyID,
  532. AccountID: accountID,
  533. Model: model,
  534. RequestedModel: coalesceTrimmedString(requestedModel, model),
  535. InputTokens: inputTokens,
  536. OutputTokens: outputTokens,
  537. CacheCreationTokens: cacheCreationTokens,
  538. CacheReadTokens: cacheReadTokens,
  539. CacheCreation5mTokens: cacheCreation5m,
  540. CacheCreation1hTokens: cacheCreation1h,
  541. ImageOutputTokens: imageOutputTokens,
  542. ImageOutputCost: imageOutputCost,
  543. ImageInputTokens: imageInputTokens,
  544. ImageInputCost: imageInputCost,
  545. InputCost: inputCost,
  546. OutputCost: outputCost,
  547. CacheCreationCost: cacheCreationCost,
  548. CacheReadCost: cacheReadCost,
  549. TotalCost: totalCost,
  550. ActualCost: actualCost,
  551. RateMultiplier: rateMultiplier,
  552. AccountRateMultiplier: nullFloat64Ptr(accountRateMultiplier),
  553. BillingType: int8(billingType),
  554. RequestType: service.RequestTypeFromInt16(requestTypeRaw),
  555. NativeCompactionV2: nativeCompactionV2,
  556. ImageCount: imageCount,
  557. VideoCount: videoCount,
  558. CacheTTLOverridden: cacheTTLOverridden,
  559. LongContextBillingApplied: longContextBillingApplied,
  560. CreatedAt: createdAt,
  561. }
  562. // 先回填 legacy 字段,再基于 legacy + request_type 计算最终请求类型,保证历史数据兼容。
  563. log.Stream = stream
  564. log.OpenAIWSMode = openaiWSMode
  565. log.RequestType = log.EffectiveRequestType()
  566. log.Stream, log.OpenAIWSMode = service.ApplyLegacyRequestFields(log.RequestType, stream, openaiWSMode)
  567. if requestID.Valid {
  568. log.RequestID = requestID.String
  569. }
  570. if groupID.Valid {
  571. value := groupID.Int64
  572. log.GroupID = &value
  573. }
  574. if subscriptionID.Valid {
  575. value := subscriptionID.Int64
  576. log.SubscriptionID = &value
  577. }
  578. if durationMs.Valid {
  579. value := int(durationMs.Int64)
  580. log.DurationMs = &value
  581. }
  582. if firstTokenMs.Valid {
  583. value := int(firstTokenMs.Int64)
  584. log.FirstTokenMs = &value
  585. }
  586. if userAgent.Valid {
  587. log.UserAgent = &userAgent.String
  588. }
  589. if ipAddress.Valid {
  590. log.IPAddress = &ipAddress.String
  591. }
  592. if imageSize.Valid {
  593. log.ImageSize = &imageSize.String
  594. }
  595. if imageInputSize.Valid {
  596. log.ImageInputSize = &imageInputSize.String
  597. }
  598. if imageOutputSize.Valid {
  599. log.ImageOutputSize = &imageOutputSize.String
  600. }
  601. if imageSizeSource.Valid {
  602. log.ImageSizeSource = &imageSizeSource.String
  603. }
  604. log.ImageSizeBreakdown = stringIntMapFromNullJSON(imageSizeBreakdown)
  605. if videoResolution.Valid {
  606. log.VideoResolution = &videoResolution.String
  607. }
  608. if videoDurationSeconds.Valid {
  609. value := int(videoDurationSeconds.Int64)
  610. log.VideoDurationSeconds = &value
  611. }
  612. if serviceTier.Valid {
  613. log.ServiceTier = &serviceTier.String
  614. }
  615. if reasoningEffort.Valid {
  616. log.ReasoningEffort = &reasoningEffort.String
  617. }
  618. if requestedReasoningEffort.Valid {
  619. log.RequestedReasoningEffort = &requestedReasoningEffort.String
  620. }
  621. if inboundEndpoint.Valid {
  622. log.InboundEndpoint = &inboundEndpoint.String
  623. }
  624. if upstreamEndpoint.Valid {
  625. log.UpstreamEndpoint = &upstreamEndpoint.String
  626. }
  627. if upstreamModel.Valid {
  628. log.UpstreamModel = &upstreamModel.String
  629. }
  630. if upstreamResponseModel.Valid {
  631. log.UpstreamResponseModel = &upstreamResponseModel.String
  632. }
  633. if upstreamModelMismatch.Valid {
  634. value := upstreamModelMismatch.Bool
  635. log.UpstreamModelMismatch = &value
  636. }
  637. if channelID.Valid {
  638. value := channelID.Int64
  639. log.ChannelID = &value
  640. }
  641. if modelMappingChain.Valid {
  642. log.ModelMappingChain = &modelMappingChain.String
  643. }
  644. if billingTier.Valid {
  645. log.BillingTier = &billingTier.String
  646. }
  647. if billingMode.Valid {
  648. log.BillingMode = &billingMode.String
  649. }
  650. if accountStatsCost.Valid {
  651. log.AccountStatsCost = &accountStatsCost.Float64
  652. }
  653. if sessionID.Valid {
  654. log.SessionID = &sessionID.String
  655. }
  656. if upstreamRequestID.Valid {
  657. log.UpstreamRequestID = &upstreamRequestID.String
  658. }
  659. return log, nil
  660. }
  661. func nullInt64(v *int64) sql.NullInt64 {
  662. if v == nil {
  663. return sql.NullInt64{}
  664. }
  665. return sql.NullInt64{Int64: *v, Valid: true}
  666. }
  667. func nullInt(v *int) sql.NullInt64 {
  668. if v == nil {
  669. return sql.NullInt64{}
  670. }
  671. return sql.NullInt64{Int64: int64(*v), Valid: true}
  672. }
  673. func nullFloat64Ptr(v sql.NullFloat64) *float64 {
  674. if !v.Valid {
  675. return nil
  676. }
  677. out := v.Float64
  678. return &out
  679. }
  680. func nullString(v *string) sql.NullString {
  681. if v == nil || *v == "" {
  682. return sql.NullString{}
  683. }
  684. return sql.NullString{String: *v, Valid: true}
  685. }
  686. func nullBool(v *bool) sql.NullBool {
  687. if v == nil {
  688. return sql.NullBool{}
  689. }
  690. return sql.NullBool{Bool: *v, Valid: true}
  691. }
  692. func nullStringIntMapJSON(v map[string]int) any {
  693. if len(v) == 0 {
  694. return nil
  695. }
  696. payload, err := json.Marshal(v)
  697. if err != nil {
  698. return nil
  699. }
  700. return string(payload)
  701. }
  702. func stringIntMapFromNullJSON(v sql.NullString) map[string]int {
  703. if !v.Valid || strings.TrimSpace(v.String) == "" {
  704. return nil
  705. }
  706. var out map[string]int
  707. if err := json.Unmarshal([]byte(v.String), &out); err != nil {
  708. return nil
  709. }
  710. if len(out) == 0 {
  711. return nil
  712. }
  713. return out
  714. }
  715. func coalesceTrimmedString(v sql.NullString, fallback string) string {
  716. if v.Valid && strings.TrimSpace(v.String) != "" {
  717. return v.String
  718. }
  719. return fallback
  720. }
  721. func setToSlice(set map[int64]struct{}) []int64 {
  722. out := make([]int64, 0, len(set))
  723. for id := range set {
  724. out = append(out, id)
  725. }
  726. return out
  727. }