diff --git a/PossibleRefactoring.md b/PossibleRefactoring.md new file mode 100644 index 00000000..3114b628 --- /dev/null +++ b/PossibleRefactoring.md @@ -0,0 +1,95 @@ +# Possible Refactoring + +Ordered by lowest effort and lowest risk first. + +## 1. Remove dead `CacheTypeBoth` + +Effort: very low +Risk: very low + +Why: +- Defined in `internal/responsecache/semantic.go`. +- No call sites found in the repo. + +How verified: +- Symbol searched: `CacheTypeBoth` +- Command: `rg -n "CacheTypeBoth"` + +Suggested action: +- Delete the constant and let tests confirm nothing depended on it. + +## 2. Deduplicate the dashboard's empty `cacheOverview` object + +Effort: low +Risk: very low + +Why: +- The same shape is repeated in: + - `internal/admin/dashboard/static/js/dashboard.js` + - `internal/admin/dashboard/static/js/modules/usage.js` + - `internal/admin/dashboard/static/js/modules/execution-plans.js` + +Suggested action: +- Keep a single `emptyCacheOverview()` factory and reuse it everywhere. + +## 3. Pick one owner for "cache overview is cached-only" + +Effort: low +Risk: low + +Why: +- The handler sets `CacheModeCached` in `internal/admin/handler.go`. +- Each reader sets it again in: + - `internal/usage/reader_sqlite.go` + - `internal/usage/reader_postgresql.go` + - `internal/usage/reader_mongodb.go` +- `GetCacheOverview()` already implies cached-only behavior. + +Suggested action: +- Keep the override in one place only. +- Prefer reader ownership so the behavior stays correct regardless of caller. + +## 4. Remove the legacy `ResponseCacheMiddleware.Middleware()` path + +Effort: medium +Risk: medium + +Why: +- Production flow now uses `HandleRequest()` from `internal/server/translated_inference_service.go`. +- `.Middleware()` in `internal/responsecache/responsecache.go` is only referenced by tests. + +How verified: +- Symbols searched: `Middleware()` and `HandleRequest(` +- Commands: + - `rg -n "\\.Middleware\\(\\)" | sort` + - `rg -n "HandleRequest\\(" | sort` + +Suggested action: +- Before deleting the compatibility wrapper, keep equivalent cache-hit and cache-miss coverage around `HandleRequest()`. +- Existing tests in `internal/responsecache/handle_request_test.go` already cover core hit/miss flows and should be expanded first if wrapper-specific assertions are still needed. +- Delete the compatibility wrapper. +- Only remove `internal/responsecache/middleware_test.go` after `HandleRequest()`-level coverage fully preserves the hit/miss, response header/status, and cache population assertions currently carried by the middleware wrapper tests. + +## 5. Centralize cache-type vocabulary across packages + +Effort: medium to high +Risk: medium + +Why: +- Overlapping cache constants and normalization logic exist in: + - `internal/usage/cache_type.go` + - `internal/auditlog/auditlog.go` + - `internal/responsecache/semantic.go` +- This increases the chance of drift when new cache types or modes are added. + +Suggested action: +- Introduce a small shared internal package for cache semantics. +- Do it only if it can be done without creating import cycles. + +## Recommended order + +1. Remove `CacheTypeBoth`. +2. Deduplicate the dashboard empty-state object. +3. Keep cached-only policy in one layer. +4. Remove the legacy middleware path. +5. Centralize cache semantics in a shared package. diff --git a/internal/admin/dashboard/static/js/dashboard.js b/internal/admin/dashboard/static/js/dashboard.js index 018467db..43fa8714 100644 --- a/internal/admin/dashboard/static/js/dashboard.js +++ b/internal/admin/dashboard/static/js/dashboard.js @@ -46,6 +46,18 @@ function dashboard() { // Data summary: { total_requests: 0, total_input_tokens: 0, total_output_tokens: 0, total_tokens: 0, total_input_cost: null, total_output_cost: null, total_cost: null }, daily: [], + cacheOverview: { + summary: { + total_hits: 0, + exact_hits: 0, + semantic_hits: 0, + total_input_tokens: 0, + total_output_tokens: 0, + total_tokens: 0, + total_saved_cost: null + }, + daily: [] + }, models: [], categories: [], activeCategory: 'all', diff --git a/internal/admin/dashboard/static/js/modules/charts.js b/internal/admin/dashboard/static/js/modules/charts.js index b9145fc3..8ef2bee5 100644 --- a/internal/admin/dashboard/static/js/modules/charts.js +++ b/internal/admin/dashboard/static/js/modules/charts.js @@ -1,33 +1,61 @@ (function(global) { function dashboardChartsModule() { return { - _overviewChartConfig(colors, labels, inputData, outputData) { + _overviewChartConfig(colors, labels, inputData, outputData, cacheInputData, cacheOutputData) { + const cacheEnabled = typeof this.cacheAnalyticsEnabled === 'function' && this.cacheAnalyticsEnabled(); + const datasets = [ + { + label: 'Input Tokens', + data: inputData, + borderColor: '#c2845a', + backgroundColor: 'rgba(194, 132, 90, 0.1)', + fill: true, + tension: 0.3, + pointRadius: 3, + pointHoverRadius: 5 + }, + { + label: 'Output Tokens', + data: outputData, + borderColor: '#7a9e7e', + backgroundColor: 'rgba(122, 158, 126, 0.1)', + fill: true, + tension: 0.3, + pointRadius: 3, + pointHoverRadius: 5 + } + ]; + if (cacheEnabled) { + datasets.push( + { + label: 'Local Cache Input Tokens', + data: cacheInputData, + borderColor: '#5f7dcf', + backgroundColor: 'rgba(95, 125, 207, 0.08)', + fill: false, + tension: 0.3, + pointRadius: 2, + pointHoverRadius: 4, + borderDash: [6, 4] + }, + { + label: 'Local Cache Output Tokens', + data: cacheOutputData, + borderColor: '#b0677b', + backgroundColor: 'rgba(176, 103, 123, 0.08)', + fill: false, + tension: 0.3, + pointRadius: 2, + pointHoverRadius: 4, + borderDash: [6, 4] + } + ); + } return { type: 'line', data: { labels: labels, - datasets: [ - { - label: 'Input Tokens', - data: inputData, - borderColor: '#c2845a', - backgroundColor: 'rgba(194, 132, 90, 0.1)', - fill: true, - tension: 0.3, - pointRadius: 3, - pointHoverRadius: 5 - }, - { - label: 'Output Tokens', - data: outputData, - borderColor: '#7a9e7e', - backgroundColor: 'rgba(122, 158, 126, 0.1)', - fill: true, - tension: 0.3, - pointRadius: 3, - pointHoverRadius: 5 - } - ] + datasets: datasets }, options: { responsive: true, @@ -182,7 +210,12 @@ const labels = filled.map((d) => d.date); const inputData = filled.map((d) => d.input_tokens); const outputData = filled.map((d) => d.output_tokens); - const config = this._overviewChartConfig(colors, labels, inputData, outputData); + const cacheByDate = {}; + const cacheDaily = this.fillMissingDays(this.cacheOverview && Array.isArray(this.cacheOverview.daily) ? this.cacheOverview.daily : []); + cacheDaily.forEach((d) => { cacheByDate[d.date] = d; }); + const cacheInputData = labels.map((label) => (cacheByDate[label] && cacheByDate[label].input_tokens) || 0); + const cacheOutputData = labels.map((label) => (cacheByDate[label] && cacheByDate[label].output_tokens) || 0); + const config = this._overviewChartConfig(colors, labels, inputData, outputData, cacheInputData, cacheOutputData); if (this.chart) { this.chart.destroy(); diff --git a/internal/admin/dashboard/static/js/modules/execution-plans.js b/internal/admin/dashboard/static/js/modules/execution-plans.js index 18510fd8..24865de1 100644 --- a/internal/admin/dashboard/static/js/modules/execution-plans.js +++ b/internal/admin/dashboard/static/js/modules/execution-plans.js @@ -61,6 +61,7 @@ 'LOGGING_ENABLED', 'USAGE_ENABLED', 'GUARDRAILS_ENABLED', + 'CACHE_ENABLED', 'REDIS_URL', 'SEMANTIC_CACHE_ENABLED' ]; @@ -80,6 +81,10 @@ }, executionPlanCacheVisible() { + const explicit = this.executionPlanRuntimeFlag('CACHE_ENABLED'); + if (explicit !== '') { + return this.executionPlanRuntimeBooleanFlag('CACHE_ENABLED', false); + } const redis = this.executionPlanRuntimeFlag('REDIS_URL'); const semantic = this.executionPlanRuntimeFlag('SEMANTIC_CACHE_ENABLED'); if (redis === '' && semantic === '') { @@ -671,11 +676,32 @@ if (payload && typeof payload === 'object' && !Array.isArray(payload) && payload[key] !== undefined && payload[key] !== null) { next[key] = String(payload[key]).trim(); } - } - this.executionPlanRuntimeConfig = next; - } catch (e) { - console.error('Failed to fetch dashboard config:', e); - this.executionPlanRuntimeConfig = {}; + } + this.executionPlanRuntimeConfig = next; + if (typeof this.fetchCacheOverview === 'function') { + if (this.executionPlanCacheVisible()) { + this.fetchCacheOverview(); + } else { + this.cacheOverview = { + summary: { + total_hits: 0, + exact_hits: 0, + semantic_hits: 0, + total_input_tokens: 0, + total_output_tokens: 0, + total_tokens: 0, + total_saved_cost: null + }, + daily: [] + }; + if (typeof this.renderChart === 'function') { + this.renderChart(); + } + } + } + } catch (e) { + console.error('Failed to fetch dashboard config:', e); + this.executionPlanRuntimeConfig = {}; } finally { if (timeoutID !== null && typeof clearTimeout === 'function') { clearTimeout(timeoutID); diff --git a/internal/admin/dashboard/static/js/modules/usage.js b/internal/admin/dashboard/static/js/modules/usage.js index 89b6b7b0..84d9cad7 100644 --- a/internal/admin/dashboard/static/js/modules/usage.js +++ b/internal/admin/dashboard/static/js/modules/usage.js @@ -1,6 +1,27 @@ (function(global) { function dashboardUsageModule() { return { + emptyCacheOverview() { + return { + summary: { + total_hits: 0, + exact_hits: 0, + semantic_hits: 0, + total_input_tokens: 0, + total_output_tokens: 0, + total_tokens: 0, + total_saved_cost: null + }, + daily: [] + }; + }, + + cacheAnalyticsEnabled() { + return typeof this.executionPlanRuntimeBooleanFlag === 'function' + ? this.executionPlanRuntimeBooleanFlag('CACHE_ENABLED', false) + : false; + }, + _usageQueryStr() { if (this.customStartDate && this.customEndDate) { return 'start_date=' + this._formatDate(this.customStartDate) + @@ -9,6 +30,61 @@ return 'days=' + this.days; }, + async fetchCacheOverview() { + if (!this.cacheAnalyticsEnabled()) { + this.cacheOverview = this.emptyCacheOverview(); + if (this.page === 'overview') this.renderChart(); + return; + } + + const controller = typeof this._startAbortableRequest === 'function' + ? this._startAbortableRequest('_cacheOverviewFetchController') + : null; + const options = { headers: this.headers() }; + if (controller) { + options.signal = controller.signal; + } + + try { + let queryStr; + if (this.customStartDate && this.customEndDate) { + queryStr = 'start_date=' + this._formatDate(this.customStartDate) + + '&end_date=' + this._formatDate(this.customEndDate); + } else { + queryStr = 'days=' + this.days; + } + queryStr += '&interval=' + this.interval; + + const res = await fetch('/admin/api/v1/cache/overview?' + queryStr, options); + if (!this.handleFetchResponse(res, 'cache overview')) { + this.cacheOverview = this.emptyCacheOverview(); + return; + } + const payload = await res.json(); + if (controller && controller.signal.aborted) { + return; + } + this.cacheOverview = payload && typeof payload === 'object' ? payload : this.emptyCacheOverview(); + if (!this.cacheOverview.summary) { + this.cacheOverview.summary = this.emptyCacheOverview().summary; + } + if (!Array.isArray(this.cacheOverview.daily)) { + this.cacheOverview.daily = []; + } + if (this.page === 'overview') this.renderChart(); + } catch (e) { + if (typeof this._isAbortError === 'function' && this._isAbortError(e)) { + return; + } + console.error('Failed to fetch cache overview:', e); + this.cacheOverview = this.emptyCacheOverview(); + } finally { + if (typeof this._clearAbortableRequest === 'function') { + this._clearAbortableRequest('_cacheOverviewFetchController', controller); + } + } + }, + async fetchUsage() { const controller = typeof this._startAbortableRequest === 'function' ? this._startAbortableRequest('_usageFetchController') @@ -48,6 +124,11 @@ this.summary = summary; this.daily = daily; this.renderChart(); + if (this.cacheAnalyticsEnabled()) { + this.fetchCacheOverview(); + } else { + this.cacheOverview = this.emptyCacheOverview(); + } if (this.page === 'usage') this.fetchUsagePage(); if (this.page === 'audit-logs') this.fetchAuditLog(true); } catch (e) { @@ -63,7 +144,11 @@ }, async fetchUsagePage() { - await Promise.all([this.fetchModelUsage(), this.fetchUsageLog(true)]); + const requests = [this.fetchModelUsage(), this.fetchUsageLog(true)]; + if (this.cacheAnalyticsEnabled()) { + requests.push(this.fetchCacheOverview()); + } + await Promise.all(requests); this.renderBarChart(); }, diff --git a/internal/admin/dashboard/templates/index.html b/internal/admin/dashboard/templates/index.html index f69002e5..1e29f240 100644 --- a/internal/admin/dashboard/templates/index.html +++ b/internal/admin/dashboard/templates/index.html @@ -39,6 +39,18 @@

Usage Overview

Estimated Cost
-
+
+
Cache Hits
+
-
+
+
+
Local Cache Input Tokens
+
-
+
+
+
Local Cache Output Tokens
+
-
+
@@ -127,6 +139,17 @@

Usage Analytics

{{template "auth-banner" .}} +
+
+
Cache Saved
+
-
+
+
+
Cache Hits
+
-
+
+
+

diff --git a/internal/admin/handler.go b/internal/admin/handler.go index 9b4f8e7d..fa2d3430 100644 --- a/internal/admin/handler.go +++ b/internal/admin/handler.go @@ -44,6 +44,7 @@ const ( DashboardConfigLoggingEnabled = "LOGGING_ENABLED" DashboardConfigUsageEnabled = "USAGE_ENABLED" DashboardConfigGuardrailsEnabled = "GUARDRAILS_ENABLED" + DashboardConfigCacheEnabled = "CACHE_ENABLED" DashboardConfigRedisURL = "REDIS_URL" DashboardConfigSemanticCacheEnabled = "SEMANTIC_CACHE_ENABLED" ) @@ -54,6 +55,7 @@ type DashboardConfigResponse struct { LoggingEnabled string `json:"LOGGING_ENABLED,omitempty"` UsageEnabled string `json:"USAGE_ENABLED,omitempty"` GuardrailsEnabled string `json:"GUARDRAILS_ENABLED,omitempty"` + CacheEnabled string `json:"CACHE_ENABLED,omitempty"` RedisURL string `json:"REDIS_URL,omitempty"` SemanticCacheEnabled string `json:"SEMANTIC_CACHE_ENABLED,omitempty"` } @@ -124,6 +126,7 @@ func normalizeDashboardRuntimeConfig(values DashboardConfigResponse) DashboardCo LoggingEnabled: strings.TrimSpace(values.LoggingEnabled), UsageEnabled: strings.TrimSpace(values.UsageEnabled), GuardrailsEnabled: strings.TrimSpace(values.GuardrailsEnabled), + CacheEnabled: strings.TrimSpace(values.CacheEnabled), RedisURL: strings.TrimSpace(values.RedisURL), SemanticCacheEnabled: strings.TrimSpace(values.SemanticCacheEnabled), } @@ -160,6 +163,7 @@ func parseUsageParams(c *echo.Context) (usage.UsageQueryParams, error) { if !validIntervals[params.Interval] { params.Interval = "daily" } + params.CacheMode = c.QueryParam("cache_mode") userPath, err := normalizeUserPathQueryParam("user_path", c.QueryParam("user_path")) if err != nil { @@ -273,6 +277,8 @@ func handleError(c *echo.Context, err error) error { // @Param days query int false "Number of days (default 30)" // @Param start_date query string false "Start date (YYYY-MM-DD)" // @Param end_date query string false "End date (YYYY-MM-DD)" +// @Param user_path query string false "Filter by tracked user path subtree" +// @Param cache_mode query string false "Cache mode filter: uncached, cached, all (default uncached)" // @Success 200 {object} usage.UsageSummary // @Failure 400 {object} core.GatewayError // @Failure 401 {object} core.GatewayError @@ -329,6 +335,8 @@ func usageSliceResponse[T any]( // @Param start_date query string false "Start date (YYYY-MM-DD)" // @Param end_date query string false "End date (YYYY-MM-DD)" // @Param interval query string false "Grouping interval: daily, weekly, monthly, yearly (default daily)" +// @Param user_path query string false "Filter by tracked user path subtree" +// @Param cache_mode query string false "Cache mode filter: uncached, cached, all (default uncached)" // @Success 200 {array} usage.DailyUsage // @Failure 400 {object} core.GatewayError // @Failure 401 {object} core.GatewayError @@ -348,6 +356,8 @@ func (h *Handler) DailyUsage(c *echo.Context) error { // @Param days query int false "Number of days (default 30)" // @Param start_date query string false "Start date (YYYY-MM-DD)" // @Param end_date query string false "End date (YYYY-MM-DD)" +// @Param user_path query string false "Filter by tracked user path subtree" +// @Param cache_mode query string false "Cache mode filter: uncached, cached, all (default uncached)" // @Success 200 {array} usage.ModelUsage // @Failure 400 {object} core.GatewayError // @Failure 401 {object} core.GatewayError @@ -370,6 +380,7 @@ func (h *Handler) UsageByModel(c *echo.Context) error { // @Param model query string false "Filter by model name" // @Param provider query string false "Filter by provider" // @Param user_path query string false "Filter by tracked user path subtree" +// @Param cache_mode query string false "Cache mode filter: uncached, cached, all (default uncached)" // @Param search query string false "Search across model, provider, request_id, provider_id" // @Param limit query int false "Page size (default 50, max 200)" // @Param offset query int false "Offset for pagination" @@ -419,6 +430,53 @@ func (h *Handler) UsageLog(c *echo.Context) error { return c.JSON(http.StatusOK, result) } +// CacheOverview handles GET /admin/api/v1/cache/overview +// +// @Summary Get cached-only usage overview +// @Tags admin +// @Produce json +// @Security BearerAuth +// @Param days query int false "Number of days (default 30)" +// @Param start_date query string false "Start date (YYYY-MM-DD)" +// @Param end_date query string false "End date (YYYY-MM-DD)" +// @Param interval query string false "Grouping interval: daily, weekly, monthly, yearly (default daily)" +// @Param user_path query string false "Filter by tracked user path subtree" +// @Param cache_mode query string false "Cache mode filter: uncached, cached, all (cache overview always uses cached mode)" +// @Success 200 {object} usage.CacheOverview +// @Failure 400 {object} core.GatewayError +// @Failure 401 {object} core.GatewayError +// @Failure 503 {object} core.GatewayError +// @Router /admin/api/v1/cache/overview [get] +func (h *Handler) CacheOverview(c *echo.Context) error { + if strings.TrimSpace(h.runtimeConfig.CacheEnabled) != "on" { + return handleError(c, featureUnavailableError("cache analytics is unavailable")) + } + if h.usageReader == nil { + return c.JSON(http.StatusOK, usage.CacheOverview{ + Daily: []usage.CacheOverviewDaily{}, + }) + } + + params, err := parseUsageParams(c) + if err != nil { + return handleError(c, err) + } + params.CacheMode = usage.CacheModeCached + + overview, err := h.usageReader.GetCacheOverview(c.Request().Context(), params) + if err != nil { + return handleError(c, err) + } + if overview == nil { + overview = &usage.CacheOverview{} + } + if overview.Daily == nil { + overview.Daily = []usage.CacheOverviewDaily{} + } + + return c.JSON(http.StatusOK, overview) +} + // AuditLog handles GET /admin/api/v1/audit/log // // @Summary Get paginated audit log entries diff --git a/internal/admin/handler_test.go b/internal/admin/handler_test.go index 617ceb2b..63909ac9 100644 --- a/internal/admin/handler_test.go +++ b/internal/admin/handler_test.go @@ -21,15 +21,18 @@ import ( // mockUsageReader implements usage.UsageReader for testing. type mockUsageReader struct { - summary *usage.UsageSummary - daily []usage.DailyUsage - modelUsage []usage.ModelUsage - usageLog *usage.UsageLogResult - lastUsageLog usage.UsageLogParams - summaryErr error - dailyErr error - modelUsageErr error - usageLogErr error + summary *usage.UsageSummary + daily []usage.DailyUsage + modelUsage []usage.ModelUsage + usageLog *usage.UsageLogResult + cacheOverview *usage.CacheOverview + lastUsageLog usage.UsageLogParams + lastCacheOverview usage.UsageQueryParams + summaryErr error + dailyErr error + modelUsageErr error + usageLogErr error + cacheErr error } type mockAuditReader struct { @@ -73,6 +76,14 @@ func (m *mockUsageReader) GetUsageLog(_ context.Context, params usage.UsageLogPa return m.usageLog, nil } +func (m *mockUsageReader) GetCacheOverview(_ context.Context, params usage.UsageQueryParams) (*usage.CacheOverview, error) { + m.lastCacheOverview = params + if m.cacheErr != nil { + return nil, m.cacheErr + } + return m.cacheOverview, nil +} + func (m *mockAuditReader) GetLogs(_ context.Context, params auditlog.LogQueryParams) (*auditlog.LogListResult, error) { m.lastQuery = params if m.logErr != nil { @@ -1155,6 +1166,7 @@ func TestDashboardConfig_ReturnsAllowlistedRuntimeFlags(t *testing.T) { LoggingEnabled: "on", UsageEnabled: "off", GuardrailsEnabled: "on", + CacheEnabled: "on", RedisURL: "on", SemanticCacheEnabled: "off", })) @@ -1183,6 +1195,9 @@ func TestDashboardConfig_ReturnsAllowlistedRuntimeFlags(t *testing.T) { if got := body.GuardrailsEnabled; got != "on" { t.Fatalf("GUARDRAILS_ENABLED = %q, want on", got) } + if got := body.CacheEnabled; got != "on" { + t.Fatalf("CACHE_ENABLED = %q, want on", got) + } if got := body.RedisURL; got != "on" { t.Fatalf("REDIS_URL = %q, want on", got) } @@ -1194,6 +1209,107 @@ func TestDashboardConfig_ReturnsAllowlistedRuntimeFlags(t *testing.T) { } } +func TestCacheOverview_FeatureUnavailableWhenCacheDisabled(t *testing.T) { + h := NewHandler(&mockUsageReader{}, nil, WithDashboardRuntimeConfig(DashboardConfigResponse{ + CacheEnabled: "off", + })) + c, rec := newHandlerContext("/admin/api/v1/cache/overview") + + if err := h.CacheOverview(c); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("expected 503, got %d", rec.Code) + } +} + +func TestCacheOverview_ReturnsPayloadWhenEnabled(t *testing.T) { + reader := &mockUsageReader{ + cacheOverview: &usage.CacheOverview{ + Summary: usage.CacheOverviewSummary{ + TotalHits: 4, + ExactHits: 3, + SemanticHits: 1, + TotalInput: 120, + TotalOutput: 60, + TotalTokens: 180, + }, + Daily: []usage.CacheOverviewDaily{ + {Date: "2026-03-31", Hits: 4, ExactHits: 3, SemanticHits: 1, InputTokens: 120, OutputTokens: 60, TotalTokens: 180}, + }, + }, + } + h := NewHandler(reader, nil, WithDashboardRuntimeConfig(DashboardConfigResponse{ + CacheEnabled: "on", + })) + c, rec := newHandlerContext("/admin/api/v1/cache/overview?days=30&user_path=/team") + + if err := h.CacheOverview(c); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + + var body usage.CacheOverview + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("failed to unmarshal: %v", err) + } + if body.Summary.TotalHits != 4 { + t.Fatalf("total_hits = %d, want 4", body.Summary.TotalHits) + } + if len(body.Daily) != 1 || body.Daily[0].ExactHits != 3 { + t.Fatalf("unexpected daily payload: %+v", body.Daily) + } + if reader.lastCacheOverview.CacheMode != usage.CacheModeCached { + t.Fatalf("CacheMode = %q, want %q", reader.lastCacheOverview.CacheMode, usage.CacheModeCached) + } + if reader.lastCacheOverview.UserPath != "/team" { + t.Fatalf("UserPath = %q, want %q", reader.lastCacheOverview.UserPath, "/team") + } +} + +func TestCacheOverview_ReturnsErrorWhenReaderFails(t *testing.T) { + reader := &mockUsageReader{cacheErr: errors.New("boom")} + h := NewHandler(reader, nil, WithDashboardRuntimeConfig(DashboardConfigResponse{ + CacheEnabled: "on", + })) + c, rec := newHandlerContext("/admin/api/v1/cache/overview?days=30") + + if err := h.CacheOverview(c); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if rec.Code != http.StatusInternalServerError { + t.Fatalf("expected 500, got %d", rec.Code) + } + + var body map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("failed to unmarshal: %v", err) + } + errorBody, ok := body["error"].(map[string]any) + if !ok { + t.Fatalf("expected error payload, got %v", body) + } + if got, ok := errorBody["type"].(string); !ok || got != "internal_error" { + t.Fatalf("error.type = %#v, want %q", errorBody["type"], "internal_error") + } + if got, ok := errorBody["message"].(string); !ok || got != "an unexpected error occurred" { + t.Fatalf("error.message = %#v, want %q", errorBody["message"], "an unexpected error occurred") + } else if strings.Contains(got, "boom") { + t.Fatalf("error.message leaked reader error: %q", got) + } + if _, ok := errorBody["param"]; !ok { + t.Fatalf("error.param missing from payload: %v", errorBody) + } + if _, ok := errorBody["code"]; !ok { + t.Fatalf("error.code missing from payload: %v", errorBody) + } + if reader.lastCacheOverview.CacheMode != usage.CacheModeCached { + t.Fatalf("CacheMode = %q, want %q", reader.lastCacheOverview.CacheMode, usage.CacheModeCached) + } +} + // --- handleError tests --- func TestHandleError_GatewayErrors(t *testing.T) { diff --git a/internal/app/app.go b/internal/app/app.go index 9357ee86..cb226484 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -115,6 +115,17 @@ func New(ctx context.Context, cfg Config) (*App, error) { } return nil, fmt.Errorf("failed to initialize usage tracking: %w", err) } + if usageResult == nil || usageResult.Logger == nil { + var usageCloseErr error + if usageResult != nil { + usageCloseErr = usageResult.Close() + } + closeErr := errors.Join(usageCloseErr, app.audit.Close(), app.providers.Close()) + if closeErr != nil { + return nil, fmt.Errorf("usage tracking initialization returned nil result (also: close error: %v)", closeErr) + } + return nil, fmt.Errorf("usage tracking initialization returned nil result") + } app.usage = usageResult // Initialize batch lifecycle storage. @@ -298,6 +309,7 @@ func New(ctx context.Context, cfg Config) (*App, error) { slog.Warn("ADMIN_UI_ENABLED=true requires ADMIN_ENDPOINTS_ENABLED=true — forcing UI to disabled") adminCfg.UIEnabled = false } + usageEnabledForDashboard := usageResult.Logger.Config().Enabled if adminCfg.EndpointsEnabled { adminHandler, dashHandler, adminErr := initAdmin( auditResult.Storage, @@ -307,7 +319,7 @@ func New(ctx context.Context, cfg Config) (*App, error) { app.aliases.Service, executionPlanResult.Service, guardrailRegistry, - dashboardRuntimeConfig(appCfg), + dashboardRuntimeConfig(appCfg, usageEnabledForDashboard), adminCfg.UIEnabled, ) if adminErr != nil { @@ -338,7 +350,7 @@ func New(ctx context.Context, cfg Config) (*App, error) { slog.Info("provider passthrough disabled") } - rcm, err := responsecache.NewResponseCacheMiddleware(appCfg.Cache.Response, cfg.AppConfig.RawProviders) + rcm, err := responsecache.NewResponseCacheMiddleware(appCfg.Cache.Response, cfg.AppConfig.RawProviders, usageResult.Logger, providerResult.Registry) if err != nil { var ( executionPlansCloseErr error @@ -768,17 +780,22 @@ func defaultExecutionPlanInput(cfg *config.Config) executionplans.CreateInput { } } -func dashboardRuntimeConfig(cfg *config.Config) admin.DashboardConfigResponse { +func dashboardRuntimeConfig(cfg *config.Config, usageEnabled bool) admin.DashboardConfigResponse { return admin.DashboardConfigResponse{ FeatureFallbackMode: dashboardFallbackModeValue(cfg), LoggingEnabled: dashboardEnabledValue(cfg != nil && cfg.Logging.Enabled), UsageEnabled: dashboardEnabledValue(cfg != nil && cfg.Usage.Enabled), GuardrailsEnabled: dashboardEnabledValue(cfg != nil && cfg.Guardrails.Enabled), + CacheEnabled: dashboardEnabledValue(cacheAnalyticsConfigured(cfg, usageEnabled)), RedisURL: dashboardEnabledValue(simpleResponseCacheConfigured(cfg)), SemanticCacheEnabled: dashboardEnabledValue(semanticResponseCacheConfigured(cfg)), } } +func cacheAnalyticsConfigured(cfg *config.Config, usageEnabled bool) bool { + return cfg != nil && usageEnabled && responseCacheConfigured(cfg.Cache.Response) +} + func dashboardEnabledValue(enabled bool) string { if enabled { return "on" diff --git a/internal/app/app_test.go b/internal/app/app_test.go index 6fba2d32..323d5828 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -46,7 +46,7 @@ func TestDashboardRuntimeConfig_ExposesFallbackMode(t *testing.T) { }, } - values := dashboardRuntimeConfig(cfg) + values := dashboardRuntimeConfig(cfg, false) if got := values.FeatureFallbackMode; got != string(config.FallbackModeManual) { t.Fatalf("dashboardRuntimeConfig()[%q] = %q, want %q", admin.DashboardConfigFeatureFallbackMode, got, config.FallbackModeManual) } @@ -59,7 +59,7 @@ func TestDashboardRuntimeConfig_InvalidFallbackModeDefaultsOff(t *testing.T) { }, } - values := dashboardRuntimeConfig(cfg) + values := dashboardRuntimeConfig(cfg, false) if got := values.FeatureFallbackMode; got != string(config.FallbackModeOff) { t.Fatalf("dashboardRuntimeConfig()[%q] = %q, want %q", admin.DashboardConfigFeatureFallbackMode, got, config.FallbackModeOff) } @@ -75,7 +75,7 @@ func TestDashboardRuntimeConfig_FallbackOverrideEnablesVisibilityWhenDefaultMode }, } - values := dashboardRuntimeConfig(cfg) + values := dashboardRuntimeConfig(cfg, false) if got := values.FeatureFallbackMode; got != string(config.FallbackModeManual) { t.Fatalf("dashboardRuntimeConfig()[%q] = %q, want %q", admin.DashboardConfigFeatureFallbackMode, got, config.FallbackModeManual) } @@ -87,7 +87,7 @@ func TestDashboardRuntimeConfig_ExposesFeatureAvailabilityFlags(t *testing.T) { Enabled: true, }, Usage: config.UsageConfig{ - Enabled: false, + Enabled: true, }, Guardrails: config.GuardrailsConfig{ Enabled: true, @@ -106,16 +106,19 @@ func TestDashboardRuntimeConfig_ExposesFeatureAvailabilityFlags(t *testing.T) { }, } - values := dashboardRuntimeConfig(cfg) + values := dashboardRuntimeConfig(cfg, true) if got := values.LoggingEnabled; got != "on" { t.Fatalf("dashboardRuntimeConfig()[%q] = %q, want on", admin.DashboardConfigLoggingEnabled, got) } - if got := values.UsageEnabled; got != "off" { - t.Fatalf("dashboardRuntimeConfig()[%q] = %q, want off", admin.DashboardConfigUsageEnabled, got) + if got := values.UsageEnabled; got != "on" { + t.Fatalf("dashboardRuntimeConfig()[%q] = %q, want on", admin.DashboardConfigUsageEnabled, got) } if got := values.GuardrailsEnabled; got != "on" { t.Fatalf("dashboardRuntimeConfig()[%q] = %q, want on", admin.DashboardConfigGuardrailsEnabled, got) } + if got := values.CacheEnabled; got != "on" { + t.Fatalf("dashboardRuntimeConfig()[%q] = %q, want on", admin.DashboardConfigCacheEnabled, got) + } if got := values.RedisURL; got != "on" { t.Fatalf("dashboardRuntimeConfig()[%q] = %q, want on", admin.DashboardConfigRedisURL, got) } @@ -123,3 +126,31 @@ func TestDashboardRuntimeConfig_ExposesFeatureAvailabilityFlags(t *testing.T) { t.Fatalf("dashboardRuntimeConfig()[%q] = %q, want off", admin.DashboardConfigSemanticCacheEnabled, got) } } + +func TestDashboardRuntimeConfig_HidesCacheAnalyticsWhenUsageDisabled(t *testing.T) { + cfg := &config.Config{ + Usage: config.UsageConfig{ + Enabled: false, + }, + Cache: config.CacheConfig{ + Response: config.ResponseCacheConfig{ + Simple: config.SimpleCacheConfig{ + Redis: &config.RedisResponseConfig{ + URL: "redis://localhost:6379", + }, + }, + }, + }, + } + + values := dashboardRuntimeConfig(cfg, false) + if got := values.UsageEnabled; got != "off" { + t.Fatalf("dashboardRuntimeConfig()[%q] = %q, want off", admin.DashboardConfigUsageEnabled, got) + } + if got := values.CacheEnabled; got != "off" { + t.Fatalf("dashboardRuntimeConfig()[%q] = %q, want off", admin.DashboardConfigCacheEnabled, got) + } + if got := values.RedisURL; got != "on" { + t.Fatalf("dashboardRuntimeConfig()[%q] = %q, want on", admin.DashboardConfigRedisURL, got) + } +} diff --git a/internal/responsecache/handle_request_test.go b/internal/responsecache/handle_request_test.go index 34f9f874..d509c547 100644 --- a/internal/responsecache/handle_request_test.go +++ b/internal/responsecache/handle_request_test.go @@ -13,8 +13,27 @@ import ( "gomodel/internal/auditlog" "gomodel/internal/cache" "gomodel/internal/core" + "gomodel/internal/usage" ) +type recordingUsageLogger struct { + entries []*usage.UsageEntry +} + +func (l *recordingUsageLogger) Write(entry *usage.UsageEntry) { + if entry != nil { + l.entries = append(l.entries, entry) + } +} + +func (l *recordingUsageLogger) Config() usage.Config { + return usage.Config{Enabled: true} +} + +func (l *recordingUsageLogger) Close() error { + return nil +} + func TestHandleRequest_SemanticMissPopulatesExactCache(t *testing.T) { store := cache.NewMapStore() defer store.Close() @@ -29,8 +48,8 @@ func TestHandleRequest_SemanticMissPopulatesExactCache(t *testing.T) { } m := &ResponseCacheMiddleware{ - simple: newSimpleCacheMiddleware(store, time.Hour), - semantic: newSemanticCacheMiddleware(emb, vecStore, semCfg), + simple: newSimpleCacheMiddleware(store, time.Hour, nil), + semantic: newSemanticCacheMiddleware(emb, vecStore, semCfg, nil), } body := []byte(`{"model":"gpt-4","messages":[{"role":"user","content":"handle-request-exact-backfill"}]}`) @@ -86,8 +105,8 @@ func TestHandleRequest_FallbackUsedSkipsCacheWrites(t *testing.T) { } m := &ResponseCacheMiddleware{ - simple: newSimpleCacheMiddleware(store, time.Hour), - semantic: newSemanticCacheMiddleware(emb, vecStore, semCfg), + simple: newSimpleCacheMiddleware(store, time.Hour, nil), + semantic: newSemanticCacheMiddleware(emb, vecStore, semCfg, nil), } body := []byte(`{"model":"gpt-4","messages":[{"role":"user","content":"fallback-skip-cache"}]}`) @@ -137,7 +156,7 @@ func TestHandleRequest_ExactHitMarksAuditEntryCacheType(t *testing.T) { defer store.Close() m := &ResponseCacheMiddleware{ - simple: newSimpleCacheMiddleware(store, time.Hour), + simple: newSimpleCacheMiddleware(store, time.Hour, nil), } body := []byte(`{"model":"gpt-4","messages":[{"role":"user","content":"mark-exact-cache-type"}]}`) @@ -177,3 +196,68 @@ func TestHandleRequest_ExactHitMarksAuditEntryCacheType(t *testing.T) { t.Fatalf("second request CacheType = %q, want %q", entry2.CacheType, auditlog.CacheTypeExact) } } + +func TestHandleRequest_ExactHitWritesSyntheticUsageEntry(t *testing.T) { + store := cache.NewMapStore() + defer store.Close() + + logger := &recordingUsageLogger{} + m := &ResponseCacheMiddleware{ + simple: newSimpleCacheMiddleware(store, time.Hour, newUsageHitRecorder(logger, nil)), + } + + body := []byte(`{"model":"gpt-4","messages":[{"role":"user","content":"cache-usage-hit"}]}`) + e := echo.New() + + run := func() *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + plan := &core.ExecutionPlan{ + Mode: core.ExecutionModeTranslated, + ProviderType: "openai", + Resolution: &core.RequestModelResolution{ + ResolvedSelector: core.ModelSelector{Provider: "openai", Model: "gpt-4"}, + }, + } + c.SetRequest(req.WithContext(core.WithExecutionPlan(req.Context(), plan))) + if err := m.HandleRequest(c, body, func() error { + return c.JSON(http.StatusOK, &core.ChatResponse{ + ID: "chatcmpl-cache-hit", + Model: "gpt-4", + Usage: core.Usage{ + PromptTokens: 11, + CompletionTokens: 5, + TotalTokens: 16, + }, + }) + }); err != nil { + t.Fatalf("HandleRequest: %v", err) + } + return rec + } + + rec1 := run() + if rec1.Header().Get("X-Cache") != "" { + t.Fatalf("first request should miss exact cache, got X-Cache=%q", rec1.Header().Get("X-Cache")) + } + + m.simple.wg.Wait() + + rec2 := run() + if rec2.Header().Get("X-Cache") != "HIT (exact)" { + t.Fatalf("second request should be exact hit, got X-Cache=%q", rec2.Header().Get("X-Cache")) + } + if len(logger.entries) != 1 { + t.Fatalf("expected 1 synthetic usage entry, got %d", len(logger.entries)) + } + entry := logger.entries[0] + if entry.CacheType != usage.CacheTypeExact { + t.Fatalf("CacheType = %q, want %q", entry.CacheType, usage.CacheTypeExact) + } + if entry.InputTokens != 11 || entry.OutputTokens != 5 || entry.TotalTokens != 16 { + t.Fatalf("unexpected tokens: %+v", entry) + } +} diff --git a/internal/responsecache/responsecache.go b/internal/responsecache/responsecache.go index d7c31450..9b66120a 100644 --- a/internal/responsecache/responsecache.go +++ b/internal/responsecache/responsecache.go @@ -10,6 +10,7 @@ import ( "gomodel/config" "gomodel/internal/cache" "gomodel/internal/embedding" + "gomodel/internal/usage" ) const responseCachePrefix = "gomodel:response:" @@ -23,8 +24,14 @@ type ResponseCacheMiddleware struct { // NewResponseCacheMiddleware creates middleware from config. // If neither simple nor semantic cache is configured, returns a no-op middleware. // rawProviders is threaded through to NewEmbedder for API-key credential resolution. -func NewResponseCacheMiddleware(cfg config.ResponseCacheConfig, rawProviders map[string]config.RawProviderConfig) (*ResponseCacheMiddleware, error) { +func NewResponseCacheMiddleware( + cfg config.ResponseCacheConfig, + rawProviders map[string]config.RawProviderConfig, + usageLogger usage.LoggerInterface, + pricingResolver usage.PricingResolver, +) (*ResponseCacheMiddleware, error) { m := &ResponseCacheMiddleware{} + hitRecorder := newUsageHitRecorder(usageLogger, pricingResolver) if cfg.Simple.Redis != nil && cfg.Simple.Redis.URL != "" { ttl := time.Duration(cfg.Simple.Redis.TTL) * time.Second @@ -43,7 +50,7 @@ func NewResponseCacheMiddleware(cfg config.ResponseCacheConfig, rawProviders map if err != nil { return nil, err } - m.simple = newSimpleCacheMiddleware(store, ttl) + m.simple = newSimpleCacheMiddleware(store, ttl, hitRecorder) slog.Info("response cache (simple/exact) enabled", "ttl_seconds", cfg.Simple.Redis.TTL, "prefix", prefix) } else { slog.Warn("response cache (simple/exact) is disabled; set cache.response.simple.redis.url to enable it") @@ -60,7 +67,7 @@ func NewResponseCacheMiddleware(cfg config.ResponseCacheConfig, rawProviders map _ = emb.Close() return nil, err } - m.semantic = newSemanticCacheMiddleware(emb, vs, sem) + m.semantic = newSemanticCacheMiddleware(emb, vs, sem, hitRecorder) slog.Info("response cache (semantic) enabled", "threshold", sem.SimilarityThreshold, "ttl_seconds", sem.TTL, @@ -141,6 +148,6 @@ func (m *ResponseCacheMiddleware) Close() error { // NewResponseCacheMiddlewareWithStore creates middleware with a custom store (for testing). func NewResponseCacheMiddlewareWithStore(store cache.Store, ttl time.Duration) *ResponseCacheMiddleware { return &ResponseCacheMiddleware{ - simple: newSimpleCacheMiddleware(store, ttl), + simple: newSimpleCacheMiddleware(store, ttl, nil), } } diff --git a/internal/responsecache/semantic.go b/internal/responsecache/semantic.go index 8083237f..8dae23e3 100644 --- a/internal/responsecache/semantic.go +++ b/internal/responsecache/semantic.go @@ -34,15 +34,17 @@ type semanticCacheMiddleware struct { cfg config.SemanticCacheConfig embedderIdentity string wg sync.WaitGroup + hitRecorder func(*echo.Context, []byte, string) } -func newSemanticCacheMiddleware(emb embedding.Embedder, store VecStore, cfg config.SemanticCacheConfig) *semanticCacheMiddleware { +func newSemanticCacheMiddleware(emb embedding.Embedder, store VecStore, cfg config.SemanticCacheConfig, hitRecorder func(*echo.Context, []byte, string)) *semanticCacheMiddleware { e := cfg.Embedder return &semanticCacheMiddleware{ embedder: emb, store: store, cfg: cfg, embedderIdentity: e.Provider + "\x00" + e.Model + "\x00" + e.ModelPath, + hitRecorder: hitRecorder, } } @@ -108,6 +110,9 @@ func (m *semanticCacheMiddleware) Handle(c *echo.Context, body []byte, next func c.Response().Header().Set("X-Cache", "HIT (semantic)") c.Response().WriteHeader(http.StatusOK) _, _ = c.Response().Write(results[0].Response) + if m.hitRecorder != nil { + m.hitRecorder(c, results[0].Response, CacheTypeSemantic) + } slog.Info("semantic cache hit", "path", path, "score", results[0].Score, diff --git a/internal/responsecache/semantic_test.go b/internal/responsecache/semantic_test.go index af9f1b33..081460b7 100644 --- a/internal/responsecache/semantic_test.go +++ b/internal/responsecache/semantic_test.go @@ -40,7 +40,7 @@ func newTestSemanticMiddleware(threshold float64, maxConvMessages int, excludeSy MaxConversationMessages: maxConvMessages, ExcludeSystemPrompt: excludeSystem, } - m := newSemanticCacheMiddleware(emb, store, cfg) + m := newSemanticCacheMiddleware(emb, store, cfg, nil) return m, store, emb } @@ -163,7 +163,7 @@ func TestSemanticCacheMiddleware_CacheMissOnLowScore(t *testing.T) { SimilarityThreshold: 0.99, TTL: 3600, MaxConversationMessages: 10, - }) + }, nil) body := []byte(`{"model":"gpt-4","messages":[{"role":"user","content":"hello"}]}`) @@ -334,7 +334,7 @@ func TestSemanticCacheMiddleware_HeaderThresholdOverride(t *testing.T) { SimilarityThreshold: 0.99, TTL: 3600, MaxConversationMessages: 10, - }) + }, nil) body := []byte(`{"model":"gpt-4","messages":[{"role":"user","content":"hello"}]}`) @@ -370,7 +370,7 @@ func TestSemanticCacheMiddleware_TTLExpiry(t *testing.T) { SimilarityThreshold: 0.90, TTL: 1, MaxConversationMessages: 10, - }) + }, nil) body := []byte(`{"model":"gpt-4","messages":[{"role":"user","content":"expiry test"}]}`) serveSemanticRequest(t, m, body, "") diff --git a/internal/responsecache/simple.go b/internal/responsecache/simple.go index ea91b2f7..569cfad4 100644 --- a/internal/responsecache/simple.go +++ b/internal/responsecache/simple.go @@ -42,16 +42,19 @@ type simpleCacheMiddleware struct { wg sync.WaitGroup jobs chan cacheWriteJob + hitRecorder func(*echo.Context, []byte, string) + workers sync.WaitGroup mu sync.RWMutex closed bool } -func newSimpleCacheMiddleware(store cache.Store, ttl time.Duration) *simpleCacheMiddleware { +func newSimpleCacheMiddleware(store cache.Store, ttl time.Duration, hitRecorder func(*echo.Context, []byte, string)) *simpleCacheMiddleware { m := &simpleCacheMiddleware{ - store: store, - ttl: ttl, - jobs: make(chan cacheWriteJob, cacheWriteQueueSize), + store: store, + ttl: ttl, + jobs: make(chan cacheWriteJob, cacheWriteQueueSize), + hitRecorder: hitRecorder, } m.startWorkers() return m @@ -112,6 +115,9 @@ func (m *simpleCacheMiddleware) TryHit(c *echo.Context, body []byte) (bool, erro c.Response().Header().Set("X-Cache", "HIT (exact)") c.Response().WriteHeader(http.StatusOK) _, _ = c.Response().Write(cached) + if m.hitRecorder != nil { + m.hitRecorder(c, cached, CacheTypeExact) + } slog.Info("response cache hit (exact)", "path", path, "request_id", c.Request().Header.Get("X-Request-ID"), diff --git a/internal/responsecache/usage_hit.go b/internal/responsecache/usage_hit.go new file mode 100644 index 00000000..4d22c629 --- /dev/null +++ b/internal/responsecache/usage_hit.go @@ -0,0 +1,60 @@ +package responsecache + +import ( + "log/slog" + "strings" + + "github.com/labstack/echo/v5" + + "gomodel/internal/core" + "gomodel/internal/usage" +) + +func newUsageHitRecorder(logger usage.LoggerInterface, pricingResolver usage.PricingResolver) func(*echo.Context, []byte, string) { + if logger == nil || !logger.Config().Enabled { + return nil + } + + return func(c *echo.Context, body []byte, cacheType string) { + if c == nil { + return + } + + ctx := c.Request().Context() + plan := core.GetExecutionPlan(ctx) + if plan != nil && !plan.UsageEnabled() { + return + } + + model := "" + provider := "" + if plan != nil { + provider = strings.TrimSpace(plan.ProviderType) + if plan.Resolution != nil { + model = strings.TrimSpace(plan.Resolution.ResolvedSelector.Model) + } + } + if provider == "" { + slog.Debug("cache hit usage skipped: missing provider type") + return + } + + endpoint := c.Request().URL.Path + requestID := core.GetRequestID(ctx) + if requestID == "" { + requestID = c.Request().Header.Get("X-Request-ID") + } + + var pricing *core.ModelPricing + if pricingResolver != nil { + pricing = pricingResolver.ResolvePricing(model, provider) + } + + entry := usage.ExtractFromCachedResponseBody(body, requestID, model, provider, endpoint, cacheType, pricing) + if entry == nil { + return + } + entry.UserPath = core.UserPathFromContext(ctx) + logger.Write(entry) + } +} diff --git a/internal/server/http.go b/internal/server/http.go index 004976a2..fb86fee8 100644 --- a/internal/server/http.go +++ b/internal/server/http.go @@ -277,6 +277,7 @@ func New(provider core.RoutableProvider, cfg *Config) *Server { if cfg != nil && cfg.AdminEndpointsEnabled && cfg.AdminHandler != nil { adminAPI := e.Group("/admin/api/v1") adminAPI.GET("/dashboard/config", cfg.AdminHandler.DashboardConfig) + adminAPI.GET("/cache/overview", cfg.AdminHandler.CacheOverview) adminAPI.GET("/usage/summary", cfg.AdminHandler.UsageSummary) adminAPI.GET("/usage/daily", cfg.AdminHandler.DailyUsage) adminAPI.GET("/usage/models", cfg.AdminHandler.UsageByModel) diff --git a/internal/usage/cache_type.go b/internal/usage/cache_type.go new file mode 100644 index 00000000..1e36ba69 --- /dev/null +++ b/internal/usage/cache_type.go @@ -0,0 +1,56 @@ +package usage + +import "strings" + +const ( + CacheTypeExact = "exact" + CacheTypeSemantic = "semantic" + + CacheModeUncached = "uncached" + CacheModeCached = "cached" + CacheModeAll = "all" +) + +func normalizeCacheType(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case CacheTypeExact: + return CacheTypeExact + case CacheTypeSemantic: + return CacheTypeSemantic + default: + return "" + } +} + +func normalizeCacheMode(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case CacheModeCached: + return CacheModeCached + case CacheModeAll: + return CacheModeAll + default: + return CacheModeUncached + } +} + +func cacheTypeValue(value string) any { + if normalized := normalizeCacheType(value); normalized != "" { + return normalized + } + return nil +} + +func normalizedUsageEntryForStorage(entry *UsageEntry) *UsageEntry { + if entry == nil { + return nil + } + + normalized := normalizeCacheType(entry.CacheType) + if normalized == entry.CacheType { + return entry + } + + cloned := *entry + cloned.CacheType = normalized + return &cloned +} diff --git a/internal/usage/cache_type_test.go b/internal/usage/cache_type_test.go new file mode 100644 index 00000000..c292701a --- /dev/null +++ b/internal/usage/cache_type_test.go @@ -0,0 +1,22 @@ +package usage + +import "testing" + +func TestNormalizedUsageEntryForStorageClearsInvalidCacheTypeWithoutMutatingInput(t *testing.T) { + entry := &UsageEntry{ + ID: "usage-1", + RequestID: "req-1", + CacheType: "invalid-cache-type", + } + + got := normalizedUsageEntryForStorage(entry) + if got == entry { + t.Fatal("expected invalid cache type to clone entry for normalization") + } + if got.CacheType != "" { + t.Fatalf("normalized CacheType = %q, want empty", got.CacheType) + } + if entry.CacheType != "invalid-cache-type" { + t.Fatalf("input CacheType mutated to %q", entry.CacheType) + } +} diff --git a/internal/usage/extractor.go b/internal/usage/extractor.go index ee30f8ab..610a3031 100644 --- a/internal/usage/extractor.go +++ b/internal/usage/extractor.go @@ -1,7 +1,9 @@ package usage import ( + "encoding/json" "maps" + "path" "strings" "time" @@ -232,6 +234,85 @@ func ExtractFromSSEUsage( return entry } +// ExtractFromCachedResponseBody converts a cached OpenAI-compatible response body into +// a synthetic usage entry for a cache hit. If the response body cannot be parsed, it +// still returns a minimal zero-token entry so cache-hit counts remain observable. +func ExtractFromCachedResponseBody( + body []byte, + requestID, model, provider, endpoint, cacheType string, + pricing ...*core.ModelPricing, +) *UsageEntry { + cacheType = normalizeCacheType(cacheType) + if cacheType == "" { + cacheType = CacheTypeExact + } + endpoint = normalizeCachedResponseEndpoint(endpoint) + + var entry *UsageEntry + switch endpoint { + case "/v1/chat/completions": + var resp core.ChatResponse + if err := json.Unmarshal(body, &resp); err == nil { + entry = ExtractFromChatResponse(&resp, requestID, provider, endpoint, pricing...) + } + case "/v1/responses": + var resp core.ResponsesResponse + if err := json.Unmarshal(body, &resp); err == nil { + entry = ExtractFromResponsesResponse(&resp, requestID, provider, endpoint, pricing...) + } + case "/v1/embeddings": + var resp core.EmbeddingResponse + if err := json.Unmarshal(body, &resp); err == nil { + entry = ExtractFromEmbeddingResponse(&resp, requestID, provider, endpoint, pricing...) + } + } + + if entry == nil { + entry = &UsageEntry{ + ID: uuid.New().String(), + RequestID: strings.TrimSpace(requestID), + Timestamp: time.Now().UTC(), + Model: strings.TrimSpace(model), + Provider: strings.TrimSpace(provider), + Endpoint: endpoint, + CacheType: cacheType, + ProviderID: "", + } + return entry + } + + entry.CacheType = cacheType + if normalized := strings.TrimSpace(requestID); normalized != "" { + entry.RequestID = normalized + } + if normalized := strings.TrimSpace(model); normalized != "" { + entry.Model = normalized + } + if normalized := strings.TrimSpace(provider); normalized != "" { + entry.Provider = normalized + } + if endpoint != "" { + entry.Endpoint = endpoint + } + return entry +} + +func normalizeCachedResponseEndpoint(endpoint string) string { + normalized := strings.TrimSpace(endpoint) + if normalized == "" { + return "" + } + + cleaned := path.Clean(normalized) + if cleaned == "." { + return "" + } + if strings.HasPrefix(normalized, "/") && !strings.HasPrefix(cleaned, "/") { + return "/" + cleaned + } + return cleaned +} + func pricingForEndpoint(pricing *core.ModelPricing, endpoint string) *core.ModelPricing { if pricing == nil { return nil diff --git a/internal/usage/extractor_test.go b/internal/usage/extractor_test.go index 471bed11..91baa802 100644 --- a/internal/usage/extractor_test.go +++ b/internal/usage/extractor_test.go @@ -1,6 +1,7 @@ package usage import ( + "encoding/json" "math" "testing" @@ -480,6 +481,120 @@ func TestExtractFromSSEUsageEmptyRawData(t *testing.T) { } } +func TestExtractFromCachedResponseBody(t *testing.T) { + t.Run("parses and overrides metadata", func(t *testing.T) { + resp := &core.ChatResponse{ + ID: "chatcmpl-cache", + Model: "gpt-4o-body", + Usage: core.Usage{ + PromptTokens: 42, + CompletionTokens: 18, + TotalTokens: 60, + }, + } + body, err := json.Marshal(resp) + if err != nil { + t.Fatalf("Marshal: %v", err) + } + + entry := ExtractFromCachedResponseBody(body, "req-cache", "gpt-4o", "openai", "/v1/chat/completions", CacheTypeExact) + if entry == nil { + t.Fatal("expected non-nil entry") + } + if entry.CacheType != CacheTypeExact { + t.Fatalf("CacheType = %q, want %q", entry.CacheType, CacheTypeExact) + } + if entry.RequestID != "req-cache" { + t.Fatalf("RequestID = %q, want %q", entry.RequestID, "req-cache") + } + if entry.Provider != "openai" { + t.Fatalf("Provider = %q, want %q", entry.Provider, "openai") + } + if entry.Endpoint != "/v1/chat/completions" { + t.Fatalf("Endpoint = %q, want %q", entry.Endpoint, "/v1/chat/completions") + } + if entry.Model != "gpt-4o" { + t.Fatalf("Model = %q, want %q", entry.Model, "gpt-4o") + } + if entry.InputTokens != 42 || entry.OutputTokens != 18 || entry.TotalTokens != 60 { + t.Fatalf("unexpected token counts: %+v", entry) + } + }) + + t.Run("normalizes equivalent endpoint paths", func(t *testing.T) { + resp := &core.ChatResponse{ + ID: "chatcmpl-cache", + Model: "gpt-4o-body", + Usage: core.Usage{ + PromptTokens: 7, + CompletionTokens: 3, + TotalTokens: 10, + }, + } + body, err := json.Marshal(resp) + if err != nil { + t.Fatalf("Marshal: %v", err) + } + + entry := ExtractFromCachedResponseBody(body, "req-cache", "gpt-4o", "openai", "/v1/chat/completions/", CacheTypeExact) + if entry == nil { + t.Fatal("expected non-nil entry") + } + if entry.Endpoint != "/v1/chat/completions" { + t.Fatalf("Endpoint = %q, want %q", entry.Endpoint, "/v1/chat/completions") + } + if entry.TotalTokens != 10 { + t.Fatalf("TotalTokens = %d, want 10", entry.TotalTokens) + } + }) + + t.Run("falls back to synthetic entry when body cannot be parsed", func(t *testing.T) { + entry := ExtractFromCachedResponseBody([]byte("{"), "req-cache-fallback", "gpt-4o", "openai", "/v1/chat/completions", CacheTypeExact) + if entry == nil { + t.Fatal("expected non-nil entry") + } + if entry.RequestID != "req-cache-fallback" { + t.Fatalf("RequestID = %q, want %q", entry.RequestID, "req-cache-fallback") + } + if entry.Provider != "openai" { + t.Fatalf("Provider = %q, want %q", entry.Provider, "openai") + } + if entry.Endpoint != "/v1/chat/completions" { + t.Fatalf("Endpoint = %q, want %q", entry.Endpoint, "/v1/chat/completions") + } + if entry.Model != "gpt-4o" { + t.Fatalf("Model = %q, want %q", entry.Model, "gpt-4o") + } + if entry.InputTokens != 0 || entry.OutputTokens != 0 || entry.TotalTokens != 0 { + t.Fatalf("expected zero-token synthetic entry, got %+v", entry) + } + }) + + t.Run("defaults unknown cache type to exact", func(t *testing.T) { + resp := &core.ChatResponse{ + ID: "chatcmpl-cache", + Model: "gpt-4o-body", + Usage: core.Usage{ + PromptTokens: 2, + CompletionTokens: 1, + TotalTokens: 3, + }, + } + body, err := json.Marshal(resp) + if err != nil { + t.Fatalf("Marshal: %v", err) + } + + entry := ExtractFromCachedResponseBody(body, "req-cache", "gpt-4o", "openai", "/v1/chat/completions", "unknown") + if entry == nil { + t.Fatal("expected non-nil entry") + } + if entry.CacheType != CacheTypeExact { + t.Fatalf("CacheType = %q, want %q", entry.CacheType, CacheTypeExact) + } + }) +} + func TestExtractFromChatResponse_WithBatchPricingEndpoint(t *testing.T) { pricing := &core.ModelPricing{ InputPerMtok: new(4.0), diff --git a/internal/usage/reader.go b/internal/usage/reader.go index 3ba814e0..0d42a824 100644 --- a/internal/usage/reader.go +++ b/internal/usage/reader.go @@ -12,6 +12,7 @@ type UsageQueryParams struct { Interval string // "daily", "weekly", "monthly", "yearly" TimeZone string // IANA timezone used for day-boundary interpretation and grouping UserPath string // subtree filter on tracked user path + CacheMode string // "uncached" (default), "cached", or "all" } // UsageSummary holds aggregated usage statistics over a time period. @@ -70,6 +71,7 @@ type UsageLogEntry struct { Provider string `json:"provider"` Endpoint string `json:"endpoint"` UserPath string `json:"user_path,omitempty"` + CacheType string `json:"cache_type,omitempty"` InputTokens int `json:"input_tokens"` OutputTokens int `json:"output_tokens"` TotalTokens int `json:"total_tokens"` @@ -88,6 +90,35 @@ type UsageLogResult struct { Offset int `json:"offset"` } +// CacheOverviewSummary holds cached-only aggregate statistics over a time period. +type CacheOverviewSummary struct { + TotalHits int `json:"total_hits"` + ExactHits int `json:"exact_hits"` + SemanticHits int `json:"semantic_hits"` + TotalInput int64 `json:"total_input_tokens"` + TotalOutput int64 `json:"total_output_tokens"` + TotalTokens int64 `json:"total_tokens"` + TotalSavedCost *float64 `json:"total_saved_cost"` +} + +// CacheOverviewDaily holds cached-only statistics for a single period. +type CacheOverviewDaily struct { + Date string `json:"date"` + Hits int `json:"hits"` + ExactHits int `json:"exact_hits"` + SemanticHits int `json:"semantic_hits"` + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` + TotalTokens int64 `json:"total_tokens"` + SavedCost *float64 `json:"saved_cost"` +} + +// CacheOverview aggregates cached-only summary and daily series for the dashboard. +type CacheOverview struct { + Summary CacheOverviewSummary `json:"summary"` + Daily []CacheOverviewDaily `json:"daily"` +} + // UsageReader provides read access to usage data for the admin API. type UsageReader interface { // GetSummary returns aggregated usage statistics for the given date range. @@ -103,4 +134,7 @@ type UsageReader interface { // GetUsageLog returns a paginated list of individual usage entries with optional filtering. GetUsageLog(ctx context.Context, params UsageLogParams) (*UsageLogResult, error) + + // GetCacheOverview returns cached-only aggregates for the admin dashboard. + GetCacheOverview(ctx context.Context, params UsageQueryParams) (*CacheOverview, error) } diff --git a/internal/usage/reader_mongodb.go b/internal/usage/reader_mongodb.go index 02318f35..f2a235de 100644 --- a/internal/usage/reader_mongodb.go +++ b/internal/usage/reader_mongodb.go @@ -3,6 +3,7 @@ package usage import ( "context" "fmt" + "regexp" "time" "go.mongodb.org/mongo-driver/v2/bson" @@ -25,20 +26,9 @@ func NewMongoDBReader(database *mongo.Database) (*MongoDBReader, error) { // GetSummary returns aggregated usage statistics for the given query parameters. func (r *MongoDBReader) GetSummary(ctx context.Context, params UsageQueryParams) (*UsageSummary, error) { pipeline := bson.A{} - matchFilters := bson.D{} - - if tsFilter := mongoDateRangeFilter(params); tsFilter != nil { - matchFilters = append(matchFilters, bson.E{Key: "timestamp", Value: tsFilter}) - } - if userPath, err := normalizeUsageUserPathFilter(params.UserPath); err != nil { + matchFilters, err := mongoUsageMatchFilters(params) + if err != nil { return nil, err - } else if userPath != "" { - matchFilters = append(matchFilters, bson.E{ - Key: "user_path", - Value: bson.D{ - {Key: "$regex", Value: usageUserPathSubtreeRegex(userPath)}, - }, - }) } if len(matchFilters) > 0 { pipeline = append(pipeline, bson.D{{Key: "$match", Value: matchFilters}}) @@ -98,20 +88,9 @@ func (r *MongoDBReader) GetSummary(ctx context.Context, params UsageQueryParams) // GetUsageByModel returns token and cost totals grouped by model and provider. func (r *MongoDBReader) GetUsageByModel(ctx context.Context, params UsageQueryParams) ([]ModelUsage, error) { pipeline := bson.A{} - matchFilters := bson.D{} - - if tsFilter := mongoDateRangeFilter(params); tsFilter != nil { - matchFilters = append(matchFilters, bson.E{Key: "timestamp", Value: tsFilter}) - } - if userPath, err := normalizeUsageUserPathFilter(params.UserPath); err != nil { + matchFilters, err := mongoUsageMatchFilters(params) + if err != nil { return nil, err - } else if userPath != "" { - matchFilters = append(matchFilters, bson.E{ - Key: "user_path", - Value: bson.D{ - {Key: "$regex", Value: usageUserPathSubtreeRegex(userPath)}, - }, - }) } if len(matchFilters) > 0 { pipeline = append(pipeline, bson.D{{Key: "$match", Value: matchFilters}}) @@ -178,36 +157,9 @@ func (r *MongoDBReader) GetUsageByModel(ctx context.Context, params UsageQueryPa func (r *MongoDBReader) GetUsageLog(ctx context.Context, params UsageLogParams) (*UsageLogResult, error) { limit, offset := clampLimitOffset(params.Limit, params.Offset) - matchFilters := bson.D{} - - if tsFilter := mongoDateRangeFilter(params.UsageQueryParams); tsFilter != nil { - matchFilters = append(matchFilters, bson.E{Key: "timestamp", Value: tsFilter}) - } - - if params.Model != "" { - matchFilters = append(matchFilters, bson.E{Key: "model", Value: params.Model}) - } - if params.Provider != "" { - matchFilters = append(matchFilters, bson.E{Key: "provider", Value: params.Provider}) - } - if userPath, err := normalizeUsageUserPathFilter(params.UserPath); err != nil { + matchFilters, err := mongoUsageLogMatchFilters(params) + if err != nil { return nil, err - } else if userPath != "" { - matchFilters = append(matchFilters, bson.E{ - Key: "user_path", - Value: bson.D{ - {Key: "$regex", Value: usageUserPathSubtreeRegex(userPath)}, - }, - }) - } - if params.Search != "" { - regex := bson.D{{Key: "$regex", Value: params.Search}, {Key: "$options", Value: "i"}} - matchFilters = append(matchFilters, bson.E{Key: "$or", Value: bson.A{ - bson.D{{Key: "model", Value: regex}}, - bson.D{{Key: "provider", Value: regex}}, - bson.D{{Key: "request_id", Value: regex}}, - bson.D{{Key: "provider_id", Value: regex}}, - }}) } pipeline := bson.A{} @@ -242,6 +194,7 @@ func (r *MongoDBReader) GetUsageLog(ctx context.Context, params UsageLogParams) Provider string `bson:"provider"` Endpoint string `bson:"endpoint"` UserPath string `bson:"user_path"` + CacheType string `bson:"cache_type"` InputTokens int `bson:"input_tokens"` OutputTokens int `bson:"output_tokens"` TotalTokens int `bson:"total_tokens"` @@ -282,6 +235,7 @@ func (r *MongoDBReader) GetUsageLog(ctx context.Context, params UsageLogParams) Provider: row.Provider, Endpoint: row.Endpoint, UserPath: row.UserPath, + CacheType: normalizeCacheType(row.CacheType), InputTokens: row.InputTokens, OutputTokens: row.OutputTokens, TotalTokens: row.TotalTokens, @@ -340,20 +294,9 @@ func (r *MongoDBReader) GetDailyUsage(ctx context.Context, params UsageQueryPara } pipeline := bson.A{} - matchFilters := bson.D{} - - if tsFilter := mongoDateRangeFilter(params); tsFilter != nil { - matchFilters = append(matchFilters, bson.E{Key: "timestamp", Value: tsFilter}) - } - if userPath, err := normalizeUsageUserPathFilter(params.UserPath); err != nil { + matchFilters, err := mongoUsageMatchFilters(params) + if err != nil { return nil, err - } else if userPath != "" { - matchFilters = append(matchFilters, bson.E{ - Key: "user_path", - Value: bson.D{ - {Key: "$regex", Value: usageUserPathSubtreeRegex(userPath)}, - }, - }) } if len(matchFilters) > 0 { pipeline = append(pipeline, bson.D{{Key: "$match", Value: matchFilters}}) @@ -423,3 +366,215 @@ func (r *MongoDBReader) GetDailyUsage(ctx context.Context, params UsageQueryPara return result, nil } + +// GetCacheOverview returns cached-only aggregates for the admin dashboard. +func (r *MongoDBReader) GetCacheOverview(ctx context.Context, params UsageQueryParams) (*CacheOverview, error) { + params.CacheMode = CacheModeCached + + matchFilters, err := mongoUsageMatchFilters(params) + if err != nil { + return nil, err + } + + interval := params.Interval + if interval == "" { + interval = "daily" + } + + pipeline := bson.A{} + if len(matchFilters) > 0 { + pipeline = append(pipeline, bson.D{{Key: "$match", Value: matchFilters}}) + } + pipeline = append(pipeline, bson.D{{Key: "$facet", Value: bson.D{ + {Key: "summary", Value: bson.A{ + bson.D{{Key: "$group", Value: bson.D{ + {Key: "_id", Value: nil}, + {Key: "total_hits", Value: bson.D{{Key: "$sum", Value: 1}}}, + {Key: "exact_hits", Value: bson.D{{Key: "$sum", Value: bson.D{{Key: "$cond", Value: bson.A{bson.D{{Key: "$eq", Value: bson.A{"$cache_type", CacheTypeExact}}}, 1, 0}}}}}}, + {Key: "semantic_hits", Value: bson.D{{Key: "$sum", Value: bson.D{{Key: "$cond", Value: bson.A{bson.D{{Key: "$eq", Value: bson.A{"$cache_type", CacheTypeSemantic}}}, 1, 0}}}}}}, + {Key: "total_input_tokens", Value: bson.D{{Key: "$sum", Value: "$input_tokens"}}}, + {Key: "total_output_tokens", Value: bson.D{{Key: "$sum", Value: "$output_tokens"}}}, + {Key: "total_tokens", Value: bson.D{{Key: "$sum", Value: "$total_tokens"}}}, + {Key: "total_saved_cost", Value: bson.D{{Key: "$sum", Value: bson.D{{Key: "$ifNull", Value: bson.A{"$total_cost", 0}}}}}}, + {Key: "has_saved_cost", Value: bson.D{{Key: "$sum", Value: bson.D{{Key: "$cond", Value: bson.A{bson.D{{Key: "$gt", Value: bson.A{"$total_cost", nil}}}, 1, 0}}}}}}, + }}}, + }}, + {Key: "daily", Value: bson.A{ + bson.D{{Key: "$group", Value: bson.D{ + {Key: "_id", Value: bson.D{{Key: "$dateToString", Value: bson.D{ + {Key: "format", Value: mongoDateFormat(interval)}, + {Key: "date", Value: "$timestamp"}, + {Key: "timezone", Value: usageTimeZone(params)}, + }}}}, + {Key: "hits", Value: bson.D{{Key: "$sum", Value: 1}}}, + {Key: "exact_hits", Value: bson.D{{Key: "$sum", Value: bson.D{{Key: "$cond", Value: bson.A{bson.D{{Key: "$eq", Value: bson.A{"$cache_type", CacheTypeExact}}}, 1, 0}}}}}}, + {Key: "semantic_hits", Value: bson.D{{Key: "$sum", Value: bson.D{{Key: "$cond", Value: bson.A{bson.D{{Key: "$eq", Value: bson.A{"$cache_type", CacheTypeSemantic}}}, 1, 0}}}}}}, + {Key: "input_tokens", Value: bson.D{{Key: "$sum", Value: "$input_tokens"}}}, + {Key: "output_tokens", Value: bson.D{{Key: "$sum", Value: "$output_tokens"}}}, + {Key: "total_tokens", Value: bson.D{{Key: "$sum", Value: "$total_tokens"}}}, + {Key: "saved_cost", Value: bson.D{{Key: "$sum", Value: bson.D{{Key: "$ifNull", Value: bson.A{"$total_cost", 0}}}}}}, + {Key: "has_saved_cost", Value: bson.D{{Key: "$sum", Value: bson.D{{Key: "$cond", Value: bson.A{bson.D{{Key: "$gt", Value: bson.A{"$total_cost", nil}}}, 1, 0}}}}}}, + }}}, + bson.D{{Key: "$sort", Value: bson.D{{Key: "_id", Value: 1}}}}, + }}, + }}}) + + overview := &CacheOverview{Daily: []CacheOverviewDaily{}} + cursor, err := r.collection.Aggregate(ctx, pipeline) + if err != nil { + return nil, fmt.Errorf("failed to aggregate cache overview: %w", err) + } + defer cursor.Close(ctx) + + var facetResult struct { + Summary []struct { + TotalHits int `bson:"total_hits"` + ExactHits int `bson:"exact_hits"` + SemanticHits int `bson:"semantic_hits"` + TotalInput int64 `bson:"total_input_tokens"` + TotalOutput int64 `bson:"total_output_tokens"` + TotalTokens int64 `bson:"total_tokens"` + TotalSavedCost float64 `bson:"total_saved_cost"` + HasSavedCost int `bson:"has_saved_cost"` + } `bson:"summary"` + Daily []struct { + Date string `bson:"_id"` + Hits int `bson:"hits"` + ExactHits int `bson:"exact_hits"` + SemanticHits int `bson:"semantic_hits"` + InputTokens int64 `bson:"input_tokens"` + OutputTokens int64 `bson:"output_tokens"` + TotalTokens int64 `bson:"total_tokens"` + SavedCost float64 `bson:"saved_cost"` + HasSavedCost int `bson:"has_saved_cost"` + } `bson:"daily"` + } + + if cursor.Next(ctx) { + if err := cursor.Decode(&facetResult); err != nil { + return nil, fmt.Errorf("failed to decode cache overview facet result: %w", err) + } + } + if err := cursor.Err(); err != nil { + return nil, fmt.Errorf("error iterating cache overview cursor: %w", err) + } + + if len(facetResult.Summary) > 0 { + row := facetResult.Summary[0] + overview.Summary = CacheOverviewSummary{ + TotalHits: row.TotalHits, + ExactHits: row.ExactHits, + SemanticHits: row.SemanticHits, + TotalInput: row.TotalInput, + TotalOutput: row.TotalOutput, + TotalTokens: row.TotalTokens, + } + if row.HasSavedCost > 0 { + overview.Summary.TotalSavedCost = &row.TotalSavedCost + } + } + + for _, row := range facetResult.Daily { + entry := CacheOverviewDaily{ + Date: row.Date, + Hits: row.Hits, + ExactHits: row.ExactHits, + SemanticHits: row.SemanticHits, + InputTokens: row.InputTokens, + OutputTokens: row.OutputTokens, + TotalTokens: row.TotalTokens, + } + if row.HasSavedCost > 0 { + entry.SavedCost = &row.SavedCost + } + overview.Daily = append(overview.Daily, entry) + } + + return overview, nil +} + +func mongoUsageMatchFilters(params UsageQueryParams) (bson.D, error) { + matchFilters := bson.D{} + if tsFilter := mongoDateRangeFilter(params); tsFilter != nil { + matchFilters = append(matchFilters, bson.E{Key: "timestamp", Value: tsFilter}) + } + userPath, err := normalizeUsageUserPathFilter(params.UserPath) + if err != nil { + return nil, err + } + if userPath != "" { + matchFilters = append(matchFilters, bson.E{ + Key: "user_path", + Value: bson.D{ + {Key: "$regex", Value: usageUserPathSubtreeRegex(userPath)}, + }, + }) + } + if filter := mongoCacheModeFilter(params.CacheMode); len(filter) > 0 { + matchFilters = append(matchFilters, filter...) + } + return matchFilters, nil +} + +func mongoUsageLogMatchFilters(params UsageLogParams) (bson.D, error) { + matchFilters, err := mongoUsageMatchFilters(params.UsageQueryParams) + if err != nil { + return nil, err + } + + if params.Model != "" { + matchFilters = append(matchFilters, bson.E{Key: "model", Value: params.Model}) + } + if params.Provider != "" { + matchFilters = append(matchFilters, bson.E{Key: "provider", Value: params.Provider}) + } + if params.Search != "" { + regex := bson.D{{Key: "$regex", Value: regexp.QuoteMeta(params.Search)}, {Key: "$options", Value: "i"}} + searchFilter := bson.D{{Key: "$or", Value: bson.A{ + bson.D{{Key: "model", Value: regex}}, + bson.D{{Key: "provider", Value: regex}}, + bson.D{{Key: "request_id", Value: regex}}, + bson.D{{Key: "provider_id", Value: regex}}, + }}} + matchFilters = mongoAndFilters(matchFilters, searchFilter) + } + + return matchFilters, nil +} + +func mongoAndFilters(filters ...bson.D) bson.D { + nonEmpty := make([]bson.D, 0, len(filters)) + for _, filter := range filters { + if len(filter) > 0 { + nonEmpty = append(nonEmpty, filter) + } + } + + switch len(nonEmpty) { + case 0: + return nil + case 1: + return nonEmpty[0] + default: + clauses := make(bson.A, 0, len(nonEmpty)) + for _, filter := range nonEmpty { + clauses = append(clauses, filter) + } + return bson.D{{Key: "$and", Value: clauses}} + } +} + +func mongoCacheModeFilter(mode string) bson.D { + switch normalizeCacheMode(mode) { + case CacheModeCached: + return bson.D{{Key: "cache_type", Value: bson.D{{Key: "$in", Value: bson.A{CacheTypeExact, CacheTypeSemantic}}}}} + case CacheModeAll: + return nil + default: + return bson.D{{Key: "$or", Value: bson.A{ + bson.D{{Key: "cache_type", Value: bson.D{{Key: "$exists", Value: false}}}}, + bson.D{{Key: "cache_type", Value: nil}}, + bson.D{{Key: "cache_type", Value: ""}}, + }}} + } +} diff --git a/internal/usage/reader_mongodb_test.go b/internal/usage/reader_mongodb_test.go new file mode 100644 index 00000000..ca0125f5 --- /dev/null +++ b/internal/usage/reader_mongodb_test.go @@ -0,0 +1,63 @@ +package usage + +import ( + "reflect" + "testing" + + "go.mongodb.org/mongo-driver/v2/bson" +) + +func TestMongoUsageLogMatchFiltersAndSearchWithCacheMode(t *testing.T) { + got, err := mongoUsageLogMatchFilters(UsageLogParams{ + UsageQueryParams: UsageQueryParams{ + CacheMode: CacheModeUncached, + }, + Search: "gpt", + }) + if err != nil { + t.Fatalf("mongoUsageLogMatchFilters() error = %v", err) + } + + regex := bson.D{{Key: "$regex", Value: "gpt"}, {Key: "$options", Value: "i"}} + want := bson.D{{Key: "$and", Value: bson.A{ + bson.D{{Key: "$or", Value: bson.A{ + bson.D{{Key: "cache_type", Value: bson.D{{Key: "$exists", Value: false}}}}, + bson.D{{Key: "cache_type", Value: nil}}, + bson.D{{Key: "cache_type", Value: ""}}, + }}}, + bson.D{{Key: "$or", Value: bson.A{ + bson.D{{Key: "model", Value: regex}}, + bson.D{{Key: "provider", Value: regex}}, + bson.D{{Key: "request_id", Value: regex}}, + bson.D{{Key: "provider_id", Value: regex}}, + }}}, + }}} + + if !reflect.DeepEqual(got, want) { + t.Fatalf("mongoUsageLogMatchFilters() = %#v, want %#v", got, want) + } +} + +func TestMongoUsageLogMatchFiltersEscapesSearchRegex(t *testing.T) { + got, err := mongoUsageLogMatchFilters(UsageLogParams{ + UsageQueryParams: UsageQueryParams{ + CacheMode: CacheModeAll, + }, + Search: "gpt.4+", + }) + if err != nil { + t.Fatalf("mongoUsageLogMatchFilters() error = %v", err) + } + + regex := bson.D{{Key: "$regex", Value: `gpt\.4\+`}, {Key: "$options", Value: "i"}} + want := bson.D{{Key: "$or", Value: bson.A{ + bson.D{{Key: "model", Value: regex}}, + bson.D{{Key: "provider", Value: regex}}, + bson.D{{Key: "request_id", Value: regex}}, + bson.D{{Key: "provider_id", Value: regex}}, + }}} + + if !reflect.DeepEqual(got, want) { + t.Fatalf("mongoUsageLogMatchFilters() = %#v, want %#v", got, want) + } +} diff --git a/internal/usage/reader_postgresql.go b/internal/usage/reader_postgresql.go index ccf8bb2b..2e530c2a 100644 --- a/internal/usage/reader_postgresql.go +++ b/internal/usage/reader_postgresql.go @@ -25,15 +25,10 @@ func NewPostgreSQLReader(pool *pgxpool.Pool) (*PostgreSQLReader, error) { // GetSummary returns aggregated usage statistics for the given query parameters. func (r *PostgreSQLReader) GetSummary(ctx context.Context, params UsageQueryParams) (*UsageSummary, error) { - conditions, args, nextIdx := pgDateRangeConditions(params, 1) - userPath, err := normalizeUsageUserPathFilter(params.UserPath) + conditions, args, _, err := pgUsageConditions(params, 1) if err != nil { return nil, err } - if userPath != "" { - conditions = append(conditions, fmt.Sprintf("(user_path = $%d OR user_path LIKE $%d ESCAPE '\\')", nextIdx, nextIdx+1)) - args = append(args, userPath, usageUserPathSubtreePattern(userPath)) - } where := buildWhereClause(conditions) costCols := `, SUM(input_cost), SUM(output_cost), SUM(total_cost)` @@ -54,15 +49,10 @@ func (r *PostgreSQLReader) GetSummary(ctx context.Context, params UsageQueryPara // GetUsageByModel returns token and cost totals grouped by model and provider. func (r *PostgreSQLReader) GetUsageByModel(ctx context.Context, params UsageQueryParams) ([]ModelUsage, error) { - conditions, args, nextIdx := pgDateRangeConditions(params, 1) - userPath, err := normalizeUsageUserPathFilter(params.UserPath) + conditions, args, _, err := pgUsageConditions(params, 1) if err != nil { return nil, err } - if userPath != "" { - conditions = append(conditions, fmt.Sprintf("(user_path = $%d OR user_path LIKE $%d ESCAPE '\\')", nextIdx, nextIdx+1)) - args = append(args, userPath, usageUserPathSubtreePattern(userPath)) - } where := buildWhereClause(conditions) costCols := `, SUM(input_cost), SUM(output_cost), SUM(total_cost)` @@ -95,8 +85,7 @@ func (r *PostgreSQLReader) GetUsageByModel(ctx context.Context, params UsageQuer func (r *PostgreSQLReader) GetUsageLog(ctx context.Context, params UsageLogParams) (*UsageLogResult, error) { limit, offset := clampLimitOffset(params.Limit, params.Offset) - conditions, args, argIdx := pgDateRangeConditions(params.UsageQueryParams, 1) - userPath, err := normalizeUsageUserPathFilter(params.UserPath) + conditions, args, argIdx, err := pgUsageConditions(params.UsageQueryParams, 1) if err != nil { return nil, err } @@ -111,11 +100,6 @@ func (r *PostgreSQLReader) GetUsageLog(ctx context.Context, params UsageLogParam args = append(args, params.Provider) argIdx++ } - if userPath != "" { - conditions = append(conditions, fmt.Sprintf("(user_path = $%d OR user_path LIKE $%d ESCAPE '\\')", argIdx, argIdx+1)) - args = append(args, userPath, usageUserPathSubtreePattern(userPath)) - argIdx += 2 - } if params.Search != "" { s := "%" + escapeLikeWildcards(params.Search) + "%" conditions = append(conditions, fmt.Sprintf("(model ILIKE $%d ESCAPE '\\' OR provider ILIKE $%d ESCAPE '\\' OR request_id ILIKE $%d ESCAPE '\\' OR provider_id ILIKE $%d ESCAPE '\\')", argIdx, argIdx, argIdx, argIdx)) @@ -133,7 +117,7 @@ func (r *PostgreSQLReader) GetUsageLog(ctx context.Context, params UsageLogParam } // Fetch page - dataQuery := fmt.Sprintf(`SELECT id, request_id, provider_id, timestamp, model, provider, endpoint, user_path, + dataQuery := fmt.Sprintf(`SELECT id, request_id, provider_id, timestamp, model, provider, endpoint, user_path, cache_type, input_tokens, output_tokens, total_tokens, COALESCE(input_cost, 0), COALESCE(output_cost, 0), COALESCE(total_cost, 0), raw_data, COALESCE(costs_calculation_caveat, '') FROM "usage"%s ORDER BY timestamp DESC LIMIT $%d OFFSET $%d`, where, argIdx, argIdx+1) dataArgs := append(append([]any(nil), args...), limit, offset) @@ -149,7 +133,8 @@ func (r *PostgreSQLReader) GetUsageLog(ctx context.Context, params UsageLogParam var e UsageLogEntry var rawDataJSON *string var userPath *string - if err := rows.Scan(&e.ID, &e.RequestID, &e.ProviderID, &e.Timestamp, &e.Model, &e.Provider, &e.Endpoint, &userPath, + var cacheType *string + if err := rows.Scan(&e.ID, &e.RequestID, &e.ProviderID, &e.Timestamp, &e.Model, &e.Provider, &e.Endpoint, &userPath, &cacheType, &e.InputTokens, &e.OutputTokens, &e.TotalTokens, &e.InputCost, &e.OutputCost, &e.TotalCost, &rawDataJSON, &e.CostsCalculationCaveat); err != nil { return nil, fmt.Errorf("failed to scan usage log row: %w", err) } @@ -161,6 +146,9 @@ func (r *PostgreSQLReader) GetUsageLog(ctx context.Context, params UsageLogParam if userPath != nil { e.UserPath = *userPath } + if cacheType != nil { + e.CacheType = normalizeCacheType(*cacheType) + } entries = append(entries, e) } @@ -216,15 +204,10 @@ func (r *PostgreSQLReader) GetDailyUsage(ctx context.Context, params UsageQueryP } groupExpr := pgGroupExpr(interval, usageTimeZone(params)) - conditions, args, nextIdx := pgDateRangeConditions(params, 1) - userPath, err := normalizeUsageUserPathFilter(params.UserPath) + conditions, args, _, err := pgUsageConditions(params, 1) if err != nil { return nil, err } - if userPath != "" { - conditions = append(conditions, fmt.Sprintf("(user_path = $%d OR user_path LIKE $%d ESCAPE '\\')", nextIdx, nextIdx+1)) - args = append(args, userPath, usageUserPathSubtreePattern(userPath)) - } where := buildWhereClause(conditions) costCols := `, SUM(input_cost), SUM(output_cost), SUM(total_cost)` @@ -253,6 +236,103 @@ func (r *PostgreSQLReader) GetDailyUsage(ctx context.Context, params UsageQueryP return result, nil } +// GetCacheOverview returns cached-only aggregates for the admin dashboard. +func (r *PostgreSQLReader) GetCacheOverview(ctx context.Context, params UsageQueryParams) (*CacheOverview, error) { + params.CacheMode = CacheModeCached + + conditions, args, _, err := pgUsageConditions(params, 3) + if err != nil { + return nil, err + } + where := buildWhereClause(conditions) + queryArgs := append([]any{CacheTypeExact, CacheTypeSemantic}, args...) + + summaryQuery := `SELECT COUNT(*), + COALESCE(SUM(CASE WHEN cache_type = $1 THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN cache_type = $2 THEN 1 ELSE 0 END), 0), + COALESCE(SUM(input_tokens), 0), + COALESCE(SUM(output_tokens), 0), + COALESCE(SUM(total_tokens), 0), + SUM(total_cost) + FROM "usage"` + where + + overview := &CacheOverview{} + if err := r.pool.QueryRow(ctx, summaryQuery, queryArgs...).Scan( + &overview.Summary.TotalHits, + &overview.Summary.ExactHits, + &overview.Summary.SemanticHits, + &overview.Summary.TotalInput, + &overview.Summary.TotalOutput, + &overview.Summary.TotalTokens, + &overview.Summary.TotalSavedCost, + ); err != nil { + return nil, fmt.Errorf("failed to query cache overview summary: %w", err) + } + + interval := params.Interval + if interval == "" { + interval = "daily" + } + groupExpr := pgGroupExpr(interval, usageTimeZone(params)) + dailyQuery := fmt.Sprintf(`SELECT %s as period, + COUNT(*), + COALESCE(SUM(CASE WHEN cache_type = $1 THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN cache_type = $2 THEN 1 ELSE 0 END), 0), + COALESCE(SUM(input_tokens), 0), + COALESCE(SUM(output_tokens), 0), + COALESCE(SUM(total_tokens), 0), + SUM(total_cost) + FROM "usage"%s GROUP BY %s ORDER BY period`, groupExpr, where, groupExpr) + + rows, err := r.pool.Query(ctx, dailyQuery, queryArgs...) + if err != nil { + return nil, fmt.Errorf("failed to query cache overview daily: %w", err) + } + defer rows.Close() + + overview.Daily = make([]CacheOverviewDaily, 0) + for rows.Next() { + var d CacheOverviewDaily + if err := rows.Scan(&d.Date, &d.Hits, &d.ExactHits, &d.SemanticHits, &d.InputTokens, &d.OutputTokens, &d.TotalTokens, &d.SavedCost); err != nil { + return nil, fmt.Errorf("failed to scan cache overview daily row: %w", err) + } + overview.Daily = append(overview.Daily, d) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("error iterating cache overview daily rows: %w", err) + } + + return overview, nil +} + func pgQuoteLiteral(value string) string { return "'" + strings.ReplaceAll(value, "'", "''") + "'" } + +func pgUsageConditions(params UsageQueryParams, argIdx int) (conditions []string, args []any, nextIdx int, err error) { + conditions, args, nextIdx = pgDateRangeConditions(params, argIdx) + userPath, err := normalizeUsageUserPathFilter(params.UserPath) + if err != nil { + return nil, nil, 0, err + } + if userPath != "" { + conditions = append(conditions, fmt.Sprintf("(user_path = $%d OR user_path LIKE $%d ESCAPE '\\')", nextIdx, nextIdx+1)) + args = append(args, userPath, usageUserPathSubtreePattern(userPath)) + nextIdx += 2 + } + if condition := pgCacheModeCondition(params.CacheMode); condition != "" { + conditions = append(conditions, condition) + } + return conditions, args, nextIdx, nil +} + +func pgCacheModeCondition(mode string) string { + switch normalizeCacheMode(mode) { + case CacheModeCached: + return "(cache_type = '" + CacheTypeExact + "' OR cache_type = '" + CacheTypeSemantic + "')" + case CacheModeAll: + return "" + default: + return "(cache_type IS NULL OR cache_type = '')" + } +} diff --git a/internal/usage/reader_sqlite.go b/internal/usage/reader_sqlite.go index 3c74ebbc..593e7c9d 100644 --- a/internal/usage/reader_sqlite.go +++ b/internal/usage/reader_sqlite.go @@ -25,15 +25,10 @@ func NewSQLiteReader(db *sql.DB) (*SQLiteReader, error) { // GetSummary returns aggregated usage statistics for the given query parameters. func (r *SQLiteReader) GetSummary(ctx context.Context, params UsageQueryParams) (*UsageSummary, error) { - conditions, args := sqliteDateRangeConditions(params) - userPath, err := normalizeUsageUserPathFilter(params.UserPath) + conditions, args, err := sqliteUsageConditions(params) if err != nil { return nil, err } - if userPath != "" { - conditions = append(conditions, "(user_path = ? OR user_path LIKE ? ESCAPE '\\')") - args = append(args, userPath, usageUserPathSubtreePattern(userPath)) - } where := buildWhereClause(conditions) costCols := `, SUM(input_cost), SUM(output_cost), SUM(total_cost)` @@ -54,15 +49,10 @@ func (r *SQLiteReader) GetSummary(ctx context.Context, params UsageQueryParams) // GetUsageByModel returns token and cost totals grouped by model and provider. func (r *SQLiteReader) GetUsageByModel(ctx context.Context, params UsageQueryParams) ([]ModelUsage, error) { - conditions, args := sqliteDateRangeConditions(params) - userPath, err := normalizeUsageUserPathFilter(params.UserPath) + conditions, args, err := sqliteUsageConditions(params) if err != nil { return nil, err } - if userPath != "" { - conditions = append(conditions, "(user_path = ? OR user_path LIKE ? ESCAPE '\\')") - args = append(args, userPath, usageUserPathSubtreePattern(userPath)) - } where := buildWhereClause(conditions) costCols := `, SUM(input_cost), SUM(output_cost), SUM(total_cost)` @@ -95,8 +85,7 @@ func (r *SQLiteReader) GetUsageByModel(ctx context.Context, params UsageQueryPar func (r *SQLiteReader) GetUsageLog(ctx context.Context, params UsageLogParams) (*UsageLogResult, error) { limit, offset := clampLimitOffset(params.Limit, params.Offset) - conditions, args := sqliteDateRangeConditions(params.UsageQueryParams) - userPath, err := normalizeUsageUserPathFilter(params.UserPath) + conditions, args, err := sqliteUsageConditions(params.UsageQueryParams) if err != nil { return nil, err } @@ -109,10 +98,6 @@ func (r *SQLiteReader) GetUsageLog(ctx context.Context, params UsageLogParams) ( conditions = append(conditions, "provider = ?") args = append(args, params.Provider) } - if userPath != "" { - conditions = append(conditions, "(user_path = ? OR user_path LIKE ? ESCAPE '\\')") - args = append(args, userPath, usageUserPathSubtreePattern(userPath)) - } if params.Search != "" { conditions = append(conditions, "(model LIKE ? ESCAPE '\\' OR provider LIKE ? ESCAPE '\\' OR request_id LIKE ? ESCAPE '\\' OR provider_id LIKE ? ESCAPE '\\')") s := "%" + escapeLikeWildcards(params.Search) + "%" @@ -129,7 +114,7 @@ func (r *SQLiteReader) GetUsageLog(ctx context.Context, params UsageLogParams) ( } // Fetch page - dataQuery := `SELECT id, request_id, provider_id, timestamp, model, provider, endpoint, user_path, + dataQuery := `SELECT id, request_id, provider_id, timestamp, model, provider, endpoint, user_path, cache_type, input_tokens, output_tokens, total_tokens, COALESCE(input_cost, 0), COALESCE(output_cost, 0), COALESCE(total_cost, 0), raw_data, COALESCE(costs_calculation_caveat, '') FROM usage` + where + ` ORDER BY ` + sqliteTimestampEpochExpr() + ` DESC, id DESC LIMIT ? OFFSET ?` dataArgs := append(append([]any(nil), args...), limit, offset) @@ -147,7 +132,8 @@ func (r *SQLiteReader) GetUsageLog(ctx context.Context, params UsageLogParams) ( var caveat *string var rawDataJSON *string var userPath sql.NullString - if err := rows.Scan(&e.ID, &e.RequestID, &e.ProviderID, &ts, &e.Model, &e.Provider, &e.Endpoint, &userPath, + var cacheType sql.NullString + if err := rows.Scan(&e.ID, &e.RequestID, &e.ProviderID, &ts, &e.Model, &e.Provider, &e.Endpoint, &userPath, &cacheType, &e.InputTokens, &e.OutputTokens, &e.TotalTokens, &e.InputCost, &e.OutputCost, &e.TotalCost, &rawDataJSON, &caveat); err != nil { return nil, fmt.Errorf("failed to scan usage log row: %w", err) } @@ -168,6 +154,9 @@ func (r *SQLiteReader) GetUsageLog(ctx context.Context, params UsageLogParams) ( if userPath.Valid { e.UserPath = userPath.String } + if cacheType.Valid { + e.CacheType = normalizeCacheType(cacheType.String) + } if caveat != nil { e.CostsCalculationCaveat = *caveat } @@ -248,15 +237,10 @@ func (r *SQLiteReader) GetDailyUsage(ctx context.Context, params UsageQueryParam return nil, err } - conditions, args := sqliteDateRangeConditions(params) - userPath, err := normalizeUsageUserPathFilter(params.UserPath) + conditions, args, err := sqliteUsageConditions(params) if err != nil { return nil, err } - if userPath != "" { - conditions = append(conditions, "(user_path = ? OR user_path LIKE ? ESCAPE '\\')") - args = append(args, userPath, usageUserPathSubtreePattern(userPath)) - } where := buildWhereClause(conditions) costCols := `, SUM(input_cost), SUM(output_cost), SUM(total_cost)` @@ -292,6 +276,79 @@ func (r *SQLiteReader) GetDailyUsage(ctx context.Context, params UsageQueryParam return result, nil } +// GetCacheOverview returns cached-only aggregates for the admin dashboard. +func (r *SQLiteReader) GetCacheOverview(ctx context.Context, params UsageQueryParams) (*CacheOverview, error) { + params.CacheMode = CacheModeCached + + conditions, args, err := sqliteUsageConditions(params) + if err != nil { + return nil, err + } + where := buildWhereClause(conditions) + + summaryQuery := `SELECT COUNT(*), + COALESCE(SUM(CASE WHEN cache_type = '` + CacheTypeExact + `' THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN cache_type = '` + CacheTypeSemantic + `' THEN 1 ELSE 0 END), 0), + COALESCE(SUM(input_tokens), 0), + COALESCE(SUM(output_tokens), 0), + COALESCE(SUM(total_tokens), 0), + SUM(total_cost) + FROM usage` + where + + overview := &CacheOverview{} + if err := r.db.QueryRowContext(ctx, summaryQuery, args...).Scan( + &overview.Summary.TotalHits, + &overview.Summary.ExactHits, + &overview.Summary.SemanticHits, + &overview.Summary.TotalInput, + &overview.Summary.TotalOutput, + &overview.Summary.TotalTokens, + &overview.Summary.TotalSavedCost, + ); err != nil { + return nil, fmt.Errorf("failed to query cache overview summary: %w", err) + } + + groupExpr, groupArgs, err := r.sqliteGroupExpr(ctx, params) + if err != nil { + return nil, err + } + dailyQuery := `WITH usage_periods AS ( + SELECT ` + groupExpr + ` AS period, + cache_type, input_tokens, output_tokens, total_tokens, total_cost + FROM usage` + where + ` + ) + SELECT period, + COUNT(*), + COALESCE(SUM(CASE WHEN cache_type = '` + CacheTypeExact + `' THEN 1 ELSE 0 END), 0), + COALESCE(SUM(CASE WHEN cache_type = '` + CacheTypeSemantic + `' THEN 1 ELSE 0 END), 0), + COALESCE(SUM(input_tokens), 0), + COALESCE(SUM(output_tokens), 0), + COALESCE(SUM(total_tokens), 0), + SUM(total_cost) + FROM usage_periods GROUP BY period ORDER BY period` + queryArgs := append(groupArgs, args...) + + rows, err := r.db.QueryContext(ctx, dailyQuery, queryArgs...) + if err != nil { + return nil, fmt.Errorf("failed to query cache overview daily: %w", err) + } + defer rows.Close() + + overview.Daily = make([]CacheOverviewDaily, 0) + for rows.Next() { + var d CacheOverviewDaily + if err := rows.Scan(&d.Date, &d.Hits, &d.ExactHits, &d.SemanticHits, &d.InputTokens, &d.OutputTokens, &d.TotalTokens, &d.SavedCost); err != nil { + return nil, fmt.Errorf("failed to scan cache overview daily row: %w", err) + } + overview.Daily = append(overview.Daily, d) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("error iterating cache overview daily rows: %w", err) + } + + return overview, nil +} + func sqliteOffsetModifier(offsetMinutes int) string { if offsetMinutes == 0 { return "" @@ -299,6 +356,33 @@ func sqliteOffsetModifier(offsetMinutes int) string { return fmt.Sprintf("%+d minutes", offsetMinutes) } +func sqliteUsageConditions(params UsageQueryParams) ([]string, []any, error) { + conditions, args := sqliteDateRangeConditions(params) + userPath, err := normalizeUsageUserPathFilter(params.UserPath) + if err != nil { + return nil, nil, err + } + if userPath != "" { + conditions = append(conditions, "(user_path = ? OR user_path LIKE ? ESCAPE '\\')") + args = append(args, userPath, usageUserPathSubtreePattern(userPath)) + } + if condition := sqliteCacheModeCondition(params.CacheMode); condition != "" { + conditions = append(conditions, condition) + } + return conditions, args, nil +} + +func sqliteCacheModeCondition(mode string) string { + switch normalizeCacheMode(mode) { + case CacheModeCached: + return "(cache_type = '" + CacheTypeExact + "' OR cache_type = '" + CacheTypeSemantic + "')" + case CacheModeAll: + return "" + default: + return "(cache_type IS NULL OR cache_type = '')" + } +} + type sqliteTimeZoneSegment struct { Until time.Time OffsetMinutes int diff --git a/internal/usage/store_mongodb.go b/internal/usage/store_mongodb.go index b59a7db8..1f626c35 100644 --- a/internal/usage/store_mongodb.go +++ b/internal/usage/store_mongodb.go @@ -82,6 +82,9 @@ func NewMongoDBStore(database *mongo.Database, retentionDays int) (*MongoDBStore { Keys: bson.D{{Key: "user_path", Value: 1}}, }, + { + Keys: bson.D{{Key: "cache_type", Value: 1}, {Key: "timestamp", Value: 1}}, + }, } // Add timestamp index - use TTL index if retention is configured, @@ -120,7 +123,7 @@ func (s *MongoDBStore) WriteBatch(ctx context.Context, entries []*UsageEntry) er // Convert entries to BSON documents docs := make([]any, len(entries)) for i, e := range entries { - docs[i] = e + docs[i] = normalizedUsageEntryForStorage(e) } // Use unordered insert for better performance (continues on errors) diff --git a/internal/usage/store_postgresql.go b/internal/usage/store_postgresql.go index af33ea77..91205c90 100644 --- a/internal/usage/store_postgresql.go +++ b/internal/usage/store_postgresql.go @@ -14,14 +14,14 @@ import ( ) const ( - usageInsertColumnCount = 16 + usageInsertColumnCount = 17 postgresMaxBindParameters = 65535 usageInsertMaxRowsPerQuery = postgresMaxBindParameters / usageInsertColumnCount ) const usageInsertPrefix = ` INSERT INTO usage (id, request_id, provider_id, timestamp, model, provider, - endpoint, user_path, input_tokens, output_tokens, total_tokens, raw_data, + endpoint, user_path, cache_type, input_tokens, output_tokens, total_tokens, raw_data, input_cost, output_cost, total_cost, costs_calculation_caveat) VALUES ` @@ -62,6 +62,7 @@ func NewPostgreSQLStore(pool *pgxpool.Pool, retentionDays int) (*PostgreSQLStore provider TEXT NOT NULL, endpoint TEXT NOT NULL, user_path TEXT, + cache_type TEXT, input_tokens INTEGER NOT NULL DEFAULT 0, output_tokens INTEGER NOT NULL DEFAULT 0, total_tokens INTEGER NOT NULL DEFAULT 0, @@ -79,6 +80,7 @@ func NewPostgreSQLStore(pool *pgxpool.Pool, retentionDays int) (*PostgreSQLStore "ALTER TABLE usage ADD COLUMN IF NOT EXISTS total_cost DOUBLE PRECISION", "ALTER TABLE usage ADD COLUMN IF NOT EXISTS costs_calculation_caveat TEXT DEFAULT ''", "ALTER TABLE usage ADD COLUMN IF NOT EXISTS user_path TEXT", + "ALTER TABLE usage ADD COLUMN IF NOT EXISTS cache_type TEXT", } for _, migration := range costMigrations { if _, err := pool.Exec(ctx, migration); err != nil { @@ -94,6 +96,7 @@ func NewPostgreSQLStore(pool *pgxpool.Pool, retentionDays int) (*PostgreSQLStore "CREATE INDEX IF NOT EXISTS idx_usage_model ON usage(model)", "CREATE INDEX IF NOT EXISTS idx_usage_provider ON usage(provider)", "CREATE INDEX IF NOT EXISTS idx_usage_user_path ON usage(user_path)", + "CREATE INDEX IF NOT EXISTS idx_usage_cache_type ON usage(cache_type)", "CREATE INDEX IF NOT EXISTS idx_usage_raw_data_gin ON usage USING GIN (raw_data)", } for _, idx := range indexes { @@ -180,6 +183,7 @@ func buildUsageInsert(entries []*UsageEntry) (string, []any) { placeholder := 1 for i, entry := range entries { + entry = normalizedUsageEntryForStorage(entry) if i > 0 { builder.WriteString(", ") } @@ -204,6 +208,7 @@ func buildUsageInsert(entries []*UsageEntry) (string, []any) { entry.Provider, entry.Endpoint, entry.UserPath, + cacheTypeValue(entry.CacheType), entry.InputTokens, entry.OutputTokens, entry.TotalTokens, diff --git a/internal/usage/store_postgresql_test.go b/internal/usage/store_postgresql_test.go index bc36ecbd..ca86d966 100644 --- a/internal/usage/store_postgresql_test.go +++ b/internal/usage/store_postgresql_test.go @@ -21,6 +21,7 @@ func TestBuildUsageInsert(t *testing.T) { Model: "gpt-4o-mini", Provider: "openai", Endpoint: "/v1/chat/completions", + CacheType: CacheTypeExact, InputTokens: 10, OutputTokens: 5, TotalTokens: 15, @@ -38,6 +39,7 @@ func TestBuildUsageInsert(t *testing.T) { Model: "gpt-4.1", Provider: "openai", Endpoint: "/v1/responses", + CacheType: "unexpected-cache-type", InputTokens: 20, OutputTokens: 8, TotalTokens: 28, @@ -50,29 +52,35 @@ func TestBuildUsageInsert(t *testing.T) { }) normalized := strings.Join(strings.Fields(query), " ") - wantQuery := "INSERT INTO usage (id, request_id, provider_id, timestamp, model, provider, endpoint, user_path, input_tokens, output_tokens, total_tokens, raw_data, input_cost, output_cost, total_cost, costs_calculation_caveat) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16), ($17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32) ON CONFLICT (id) DO NOTHING" + wantQuery := "INSERT INTO usage (id, request_id, provider_id, timestamp, model, provider, endpoint, user_path, cache_type, input_tokens, output_tokens, total_tokens, raw_data, input_cost, output_cost, total_cost, costs_calculation_caveat) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17), ($18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34) ON CONFLICT (id) DO NOTHING" if normalized != wantQuery { t.Fatalf("query = %q, want %q", normalized, wantQuery) } - if got, want := len(args), 32; got != want { + if got, want := len(args), 34; got != want { t.Fatalf("len(args) = %d, want %d", got, want) } if got := args[0]; got != "usage-1" { t.Fatalf("args[0] = %v, want usage-1", got) } - if got := args[16]; got != "usage-2" { - t.Fatalf("args[16] = %v, want usage-2", got) + if got := args[17]; got != "usage-2" { + t.Fatalf("args[17] = %v, want usage-2", got) } - if got := string(args[11].([]byte)); got != `{"cached_tokens":3}` { - t.Fatalf("args[11] = %q, want %q", got, `{"cached_tokens":3}`) + if got := args[8]; got != CacheTypeExact { + t.Fatalf("args[8] = %v, want %q", got, CacheTypeExact) } - rawData, ok := args[27].([]byte) + if got := string(args[12].([]byte)); got != `{"cached_tokens":3}` { + t.Fatalf("args[12] = %q, want %q", got, `{"cached_tokens":3}`) + } + if got := args[25]; got != nil { + t.Fatalf("args[25] = %v, want nil cache_type", got) + } + rawData, ok := args[29].([]byte) if !ok { - t.Fatalf("args[27] has type %T, want []byte", args[27]) + t.Fatalf("args[29] has type %T, want []byte", args[29]) } if rawData != nil { - t.Fatalf("args[27] = %v, want nil raw_data", rawData) + t.Fatalf("args[29] = %v, want nil raw_data", rawData) } } diff --git a/internal/usage/store_sqlite.go b/internal/usage/store_sqlite.go index feebfe83..9db886b3 100644 --- a/internal/usage/store_sqlite.go +++ b/internal/usage/store_sqlite.go @@ -12,11 +12,11 @@ import ( ) // SQLite has a default limit of 999 bindable parameters per query (SQLITE_MAX_VARIABLE_NUMBER). -// With 16 columns per usage entry, we can safely insert up to 62 entries per batch (62 * 16 = 992). +// maxEntriesPerBatch derives from maxSQLiteParams / columnsPerUsageEntry. const ( maxSQLiteParams = 999 - columnsPerUsageEntry = 16 - maxEntriesPerBatch = maxSQLiteParams / columnsPerUsageEntry // 62 entries + columnsPerUsageEntry = 17 + maxEntriesPerBatch = maxSQLiteParams / columnsPerUsageEntry // 58 entries ) // SQLiteStore implements UsageStore for SQLite databases. @@ -46,6 +46,7 @@ func NewSQLiteStore(db *sql.DB, retentionDays int) (*SQLiteStore, error) { provider TEXT NOT NULL, endpoint TEXT NOT NULL, user_path TEXT, + cache_type TEXT, input_tokens INTEGER NOT NULL DEFAULT 0, output_tokens INTEGER NOT NULL DEFAULT 0, total_tokens INTEGER NOT NULL DEFAULT 0, @@ -63,6 +64,7 @@ func NewSQLiteStore(db *sql.DB, retentionDays int) (*SQLiteStore, error) { "ALTER TABLE usage ADD COLUMN total_cost REAL", "ALTER TABLE usage ADD COLUMN costs_calculation_caveat TEXT DEFAULT ''", "ALTER TABLE usage ADD COLUMN user_path TEXT", + "ALTER TABLE usage ADD COLUMN cache_type TEXT", } for _, migration := range costMigrations { if _, err := db.Exec(migration); err != nil { @@ -82,6 +84,7 @@ func NewSQLiteStore(db *sql.DB, retentionDays int) (*SQLiteStore, error) { "CREATE INDEX IF NOT EXISTS idx_usage_model ON usage(model)", "CREATE INDEX IF NOT EXISTS idx_usage_provider ON usage(provider)", "CREATE INDEX IF NOT EXISTS idx_usage_user_path ON usage(user_path)", + "CREATE INDEX IF NOT EXISTS idx_usage_cache_type ON usage(cache_type)", } for _, idx := range indexes { if _, err := db.Exec(idx); err != nil { @@ -120,7 +123,8 @@ func (s *SQLiteStore) WriteBatch(ctx context.Context, entries []*UsageEntry) err values := make([]any, 0, len(chunk)*columnsPerUsageEntry) for j, e := range chunk { - placeholders[j] = "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" + e = normalizedUsageEntryForStorage(e) + placeholders[j] = "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" rawDataJSON := marshalRawData(e.RawData, e.ID) @@ -139,6 +143,7 @@ func (s *SQLiteStore) WriteBatch(ctx context.Context, entries []*UsageEntry) err e.Provider, e.Endpoint, e.UserPath, + cacheTypeValue(e.CacheType), e.InputTokens, e.OutputTokens, e.TotalTokens, @@ -151,7 +156,7 @@ func (s *SQLiteStore) WriteBatch(ctx context.Context, entries []*UsageEntry) err } query := `INSERT OR IGNORE INTO usage (id, request_id, provider_id, timestamp, model, provider, - endpoint, user_path, input_tokens, output_tokens, total_tokens, raw_data, + endpoint, user_path, cache_type, input_tokens, output_tokens, total_tokens, raw_data, input_cost, output_cost, total_cost, costs_calculation_caveat) VALUES ` + strings.Join(placeholders, ",") diff --git a/internal/usage/usage.go b/internal/usage/usage.go index ae62d22a..a0dea7bc 100644 --- a/internal/usage/usage.go +++ b/internal/usage/usage.go @@ -37,10 +37,11 @@ type UsageEntry struct { Timestamp time.Time `json:"timestamp" bson:"timestamp"` // Request context - Model string `json:"model" bson:"model"` - Provider string `json:"provider" bson:"provider"` - Endpoint string `json:"endpoint" bson:"endpoint"` - UserPath string `json:"user_path,omitempty" bson:"user_path,omitempty"` + Model string `json:"model" bson:"model"` + Provider string `json:"provider" bson:"provider"` + Endpoint string `json:"endpoint" bson:"endpoint"` + UserPath string `json:"user_path,omitempty" bson:"user_path,omitempty"` + CacheType string `json:"cache_type,omitempty" bson:"cache_type,omitempty"` // Standard token counts (normalized across providers) InputTokens int `json:"input_tokens" bson:"input_tokens"`