dashboard_query_cache.go 8.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231
  1. package admin
  2. import (
  3. "context"
  4. "encoding/json"
  5. "fmt"
  6. "time"
  7. "github.com/Wei-Shaw/sub2api/internal/pkg/usagestats"
  8. )
  9. var (
  10. dashboardTrendCache = newSnapshotCache(30 * time.Second)
  11. dashboardModelStatsCache = newSnapshotCache(30 * time.Second)
  12. dashboardGroupStatsCache = newSnapshotCache(30 * time.Second)
  13. dashboardUsersTrendCache = newSnapshotCache(30 * time.Second)
  14. dashboardAPIKeysTrendCache = newSnapshotCache(30 * time.Second)
  15. )
  16. type dashboardTrendCacheKey struct {
  17. StartTime string `json:"start_time"`
  18. EndTime string `json:"end_time"`
  19. Granularity string `json:"granularity"`
  20. UserID int64 `json:"user_id"`
  21. APIKeyID int64 `json:"api_key_id"`
  22. AccountID int64 `json:"account_id"`
  23. GroupID int64 `json:"group_id"`
  24. Model string `json:"model"`
  25. RequestType *int16 `json:"request_type"`
  26. Stream *bool `json:"stream"`
  27. NativeCompactionV2 *bool `json:"native_compaction_v2"`
  28. BillingType *int8 `json:"billing_type"`
  29. UpstreamModelMismatch *bool `json:"upstream_model_mismatch"`
  30. }
  31. type dashboardModelGroupCacheKey struct {
  32. StartTime string `json:"start_time"`
  33. EndTime string `json:"end_time"`
  34. UserID int64 `json:"user_id"`
  35. APIKeyID int64 `json:"api_key_id"`
  36. AccountID int64 `json:"account_id"`
  37. GroupID int64 `json:"group_id"`
  38. ModelSource string `json:"model_source,omitempty"`
  39. RequestType *int16 `json:"request_type"`
  40. Stream *bool `json:"stream"`
  41. NativeCompactionV2 *bool `json:"native_compaction_v2"`
  42. BillingType *int8 `json:"billing_type"`
  43. UpstreamModelMismatch *bool `json:"upstream_model_mismatch"`
  44. }
  45. type dashboardEntityTrendCacheKey struct {
  46. StartTime string `json:"start_time"`
  47. EndTime string `json:"end_time"`
  48. Granularity string `json:"granularity"`
  49. Limit int `json:"limit"`
  50. }
  51. func cacheStatusValue(hit bool) string {
  52. if hit {
  53. return "hit"
  54. }
  55. return "miss"
  56. }
  57. func mustMarshalDashboardCacheKey(value any) string {
  58. raw, err := json.Marshal(value)
  59. if err != nil {
  60. return ""
  61. }
  62. return string(raw)
  63. }
  64. func snapshotPayloadAs[T any](payload any) (T, error) {
  65. typed, ok := payload.(T)
  66. if !ok {
  67. var zero T
  68. return zero, fmt.Errorf("unexpected cache payload type %T", payload)
  69. }
  70. return typed, nil
  71. }
  72. func (h *DashboardHandler) getUsageTrendCached(
  73. ctx context.Context,
  74. startTime, endTime time.Time,
  75. granularity string,
  76. userID, apiKeyID, accountID, groupID int64,
  77. model string,
  78. requestType *int16,
  79. stream *bool,
  80. nativeCompactionV2 *bool,
  81. billingType *int8,
  82. upstreamModelMismatch *bool,
  83. ) ([]usagestats.TrendDataPoint, bool, error) {
  84. key := mustMarshalDashboardCacheKey(dashboardTrendCacheKey{
  85. StartTime: startTime.UTC().Format(time.RFC3339),
  86. EndTime: endTime.UTC().Format(time.RFC3339),
  87. Granularity: granularity,
  88. UserID: userID,
  89. APIKeyID: apiKeyID,
  90. AccountID: accountID,
  91. GroupID: groupID,
  92. Model: model,
  93. RequestType: requestType,
  94. Stream: stream,
  95. NativeCompactionV2: nativeCompactionV2,
  96. BillingType: billingType,
  97. UpstreamModelMismatch: upstreamModelMismatch,
  98. })
  99. entry, hit, err := dashboardTrendCache.GetOrLoad(key, func() (any, error) {
  100. return h.dashboardService.GetUsageTrendWithUsageFilters(ctx, startTime, endTime, granularity, usagestats.UsageLogFilters{
  101. UserID: userID, APIKeyID: apiKeyID, AccountID: accountID, GroupID: groupID,
  102. Model: model, RequestType: requestType, Stream: stream, NativeCompactionV2: nativeCompactionV2, BillingType: billingType,
  103. UpstreamModelMismatch: upstreamModelMismatch,
  104. })
  105. })
  106. if err != nil {
  107. return nil, hit, err
  108. }
  109. trend, err := snapshotPayloadAs[[]usagestats.TrendDataPoint](entry.Payload)
  110. return trend, hit, err
  111. }
  112. func (h *DashboardHandler) getModelStatsCached(
  113. ctx context.Context,
  114. startTime, endTime time.Time,
  115. userID, apiKeyID, accountID, groupID int64,
  116. modelSource string,
  117. requestType *int16,
  118. stream *bool,
  119. nativeCompactionV2 *bool,
  120. billingType *int8,
  121. upstreamModelMismatch *bool,
  122. ) ([]usagestats.ModelStat, bool, error) {
  123. key := mustMarshalDashboardCacheKey(dashboardModelGroupCacheKey{
  124. StartTime: startTime.UTC().Format(time.RFC3339),
  125. EndTime: endTime.UTC().Format(time.RFC3339),
  126. UserID: userID,
  127. APIKeyID: apiKeyID,
  128. AccountID: accountID,
  129. GroupID: groupID,
  130. ModelSource: usagestats.NormalizeModelSource(modelSource),
  131. RequestType: requestType,
  132. Stream: stream,
  133. NativeCompactionV2: nativeCompactionV2,
  134. BillingType: billingType,
  135. UpstreamModelMismatch: upstreamModelMismatch,
  136. })
  137. entry, hit, err := dashboardModelStatsCache.GetOrLoad(key, func() (any, error) {
  138. return h.dashboardService.GetModelStatsWithUsageFiltersBySource(ctx, startTime, endTime, usagestats.UsageLogFilters{
  139. UserID: userID, APIKeyID: apiKeyID, AccountID: accountID, GroupID: groupID,
  140. RequestType: requestType, Stream: stream, NativeCompactionV2: nativeCompactionV2, BillingType: billingType,
  141. UpstreamModelMismatch: upstreamModelMismatch,
  142. }, modelSource)
  143. })
  144. if err != nil {
  145. return nil, hit, err
  146. }
  147. stats, err := snapshotPayloadAs[[]usagestats.ModelStat](entry.Payload)
  148. return stats, hit, err
  149. }
  150. func (h *DashboardHandler) getGroupStatsCached(
  151. ctx context.Context,
  152. startTime, endTime time.Time,
  153. userID, apiKeyID, accountID, groupID int64,
  154. requestType *int16,
  155. stream *bool,
  156. nativeCompactionV2 *bool,
  157. billingType *int8,
  158. upstreamModelMismatch *bool,
  159. ) ([]usagestats.GroupStat, bool, error) {
  160. key := mustMarshalDashboardCacheKey(dashboardModelGroupCacheKey{
  161. StartTime: startTime.UTC().Format(time.RFC3339),
  162. EndTime: endTime.UTC().Format(time.RFC3339),
  163. UserID: userID,
  164. APIKeyID: apiKeyID,
  165. AccountID: accountID,
  166. GroupID: groupID,
  167. RequestType: requestType,
  168. Stream: stream,
  169. NativeCompactionV2: nativeCompactionV2,
  170. BillingType: billingType,
  171. UpstreamModelMismatch: upstreamModelMismatch,
  172. })
  173. entry, hit, err := dashboardGroupStatsCache.GetOrLoad(key, func() (any, error) {
  174. return h.dashboardService.GetGroupStatsWithUsageFilters(ctx, startTime, endTime, usagestats.UsageLogFilters{
  175. UserID: userID, APIKeyID: apiKeyID, AccountID: accountID, GroupID: groupID,
  176. RequestType: requestType, Stream: stream, NativeCompactionV2: nativeCompactionV2, BillingType: billingType,
  177. UpstreamModelMismatch: upstreamModelMismatch,
  178. })
  179. })
  180. if err != nil {
  181. return nil, hit, err
  182. }
  183. stats, err := snapshotPayloadAs[[]usagestats.GroupStat](entry.Payload)
  184. return stats, hit, err
  185. }
  186. func (h *DashboardHandler) getAPIKeyUsageTrendCached(ctx context.Context, startTime, endTime time.Time, granularity string, limit int) ([]usagestats.APIKeyUsageTrendPoint, bool, error) {
  187. key := mustMarshalDashboardCacheKey(dashboardEntityTrendCacheKey{
  188. StartTime: startTime.UTC().Format(time.RFC3339),
  189. EndTime: endTime.UTC().Format(time.RFC3339),
  190. Granularity: granularity,
  191. Limit: limit,
  192. })
  193. entry, hit, err := dashboardAPIKeysTrendCache.GetOrLoad(key, func() (any, error) {
  194. return h.dashboardService.GetAPIKeyUsageTrend(ctx, startTime, endTime, granularity, limit)
  195. })
  196. if err != nil {
  197. return nil, hit, err
  198. }
  199. trend, err := snapshotPayloadAs[[]usagestats.APIKeyUsageTrendPoint](entry.Payload)
  200. return trend, hit, err
  201. }
  202. func (h *DashboardHandler) getUserUsageTrendCached(ctx context.Context, startTime, endTime time.Time, granularity string, limit int) ([]usagestats.UserUsageTrendPoint, bool, error) {
  203. key := mustMarshalDashboardCacheKey(dashboardEntityTrendCacheKey{
  204. StartTime: startTime.UTC().Format(time.RFC3339),
  205. EndTime: endTime.UTC().Format(time.RFC3339),
  206. Granularity: granularity,
  207. Limit: limit,
  208. })
  209. entry, hit, err := dashboardUsersTrendCache.GetOrLoad(key, func() (any, error) {
  210. return h.dashboardService.GetUserUsageTrend(ctx, startTime, endTime, granularity, limit)
  211. })
  212. if err != nil {
  213. return nil, hit, err
  214. }
  215. trend, err := snapshotPayloadAs[[]usagestats.UserUsageTrendPoint](entry.Payload)
  216. return trend, hit, err
  217. }