pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-07-26 09:52:11 +03:00
Code Issues Packages Projects Releases Wiki Activity

Fix analytics history and log token accounting (#2904)

Browse Source 
Integrated into release/v3.8.6
This commit is contained in:
Halil Tezcan KARABULUT
2026-05-29 19:18:14 +03:00
committed by GitHub
parent a15750d968
commit efcd062002
6 changed files with 254 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

181
src/app/api/usage/analytics/route.ts
View File

@@ -2,6 +2,7 @@ import { NextResponse } from "next/server";
import { requireManagementAuth } from "@/lib/api/requireManagementAuth";
import { getApiKeys } from "@/lib/db/apiKeys";
import { getDbInstance } from "@/lib/db/core";
import { getUserDatabaseSettings } from "@/lib/db/databaseSettings";
function getRangeStartIso(range: string): string | null {
const end = new Date();
@@ -327,6 +328,14 @@ export async function GET(request: Request) {
}
}
// Compute the raw-data cutoff: rows older than this may have been rolled up to
// daily_usage_summary and deleted from usage_history.
const dbSettings = getUserDatabaseSettings();
const rawRetentionDays = dbSettings.aggregation?.rawDataRetentionDays ?? 30;
const rawCutoff = new Date();
rawCutoff.setDate(rawCutoff.getDate() - rawRetentionDays);
const rawCutoffIso = rawCutoff.toISOString();
const conditions = [];
const params: Record<string, string> = {};
@@ -351,6 +360,93 @@ export async function GET(request: Request) {
const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : "";
// Build a UNION data source that merges recent raw rows with aggregated history.
// daily_usage_summary rows are included only when the query window extends before rawCutoffIso.
// The api_key filter is intentionally NOT applied to daily_usage_summary (api_key not stored there).
const needsAggregated = !sinceIso || sinceIso < rawCutoffIso;
const rawConditions: string[] = [];
if (sinceIso) rawConditions.push("timestamp >= @since");
if (untilIso) rawConditions.push("timestamp <= @until");
if (apiKeyWhere) rawConditions.push(apiKeyWhere);
const rawWhere = rawConditions.length > 0 ? `WHERE ${rawConditions.join(" AND ")}` : "";
// Aggregated rows only span dates within the requested window (no api_key filter).
const aggConditions: string[] = [];
if (sinceIso) {
// Use date comparison on the summary's date column (YYYY-MM-DD).
const sinceDate = sinceIso.split("T")[0];
aggConditions.push("date >= @sinceDate");
params.sinceDate = sinceDate;
}
if (untilIso) {
const untilDate = untilIso.split("T")[0];
aggConditions.push("date <= @untilDate");
params.untilDate = untilDate;
}
const aggWhere = aggConditions.length > 0 ? `WHERE ${aggConditions.join(" AND ")}` : "";
// Unified source CTE: columns aligned to usage_history shape needed by analytics queries.
// Fields not available in daily_usage_summary default to 0/NULL.
const unifiedSource = needsAggregated
? `(
SELECT
timestamp,
provider,
model,
tokens_input,
tokens_output,
tokens_cache_read,
tokens_cache_creation,
tokens_reasoning,
service_tier,
success,
latency_ms,
connection_id,
api_key_id,
api_key_name,
combo_name,
requested_model
FROM usage_history
${rawWhere}
UNION ALL
SELECT
date || 'T12:00:00.000Z' as timestamp,
provider,
model,
total_input_tokens as tokens_input,
total_output_tokens as tokens_output,
0 as tokens_cache_read,
0 as tokens_cache_creation,
0 as tokens_reasoning,
'standard' as service_tier,
1 as success,
0 as latency_ms,
NULL as connection_id,
NULL as api_key_id,
NULL as api_key_name,
NULL as combo_name,
NULL as requested_model
FROM daily_usage_summary
${aggWhere}
)`
: `(SELECT
timestamp, provider, model,
tokens_input, tokens_output,
tokens_cache_read, tokens_cache_creation, tokens_reasoning,
service_tier, success, latency_ms,
connection_id, api_key_id, api_key_name,
combo_name, requested_model
FROM usage_history
${whereClause}
)`;
// When using the unified source the WHERE filters are already embedded inside.
// For the original whereClause-based queries that still reference usage_history directly
// (e.g. fallbackRow, accountRows) we keep them as-is since they need joins or
// columns only present in usage_history.
const unifiedWhere = ""; // no additional WHERE needed — filters embedded in unifiedSource
// Fetch pricing data for cost calculation (no rows loaded)
const { getPricing } = await import("@/lib/db/settings");
const rawPricingByProvider = (await getPricing()) as PricingByProvider;
@@ -383,8 +479,8 @@ export async function GET(request: Request) {
COALESCE(AVG(latency_ms), 0) as avgLatencyMs,
COALESCE(MIN(timestamp), '') as firstRequest,
COALESCE(MAX(timestamp), '') as lastRequest
FROM usage_history
${whereClause}
FROM ${unifiedSource} AS _u
${unifiedWhere}
`
)
.get(params) as Record<string, unknown>;
@@ -398,8 +494,8 @@ export async function GET(request: Request) {
COALESCE(SUM(tokens_input), 0) as promptTokens,
COALESCE(SUM(tokens_output), 0) as completionTokens,
COALESCE(SUM(tokens_input + tokens_output), 0) as totalTokens
FROM usage_history
${whereClause}
FROM ${unifiedSource} AS _u
${unifiedWhere}
GROUP BY DATE(timestamp)
ORDER BY date ASC
`
@@ -419,8 +515,8 @@ export async function GET(request: Request) {
COALESCE(SUM(tokens_cache_read), 0) as cacheReadTokens,
COALESCE(SUM(tokens_cache_creation), 0) as cacheCreationTokens,
COALESCE(SUM(tokens_reasoning), 0) as reasoningTokens
FROM usage_history
${whereClause}
FROM ${unifiedSource} AS _u
${unifiedWhere}
GROUP BY DATE(timestamp), LOWER(provider), LOWER(model), serviceTier
ORDER BY date ASC
`
@@ -478,8 +574,8 @@ export async function GET(request: Request) {
COALESCE(AVG(latency_ms), 0) as avgLatencyMs,
COALESCE(SUM(CASE WHEN success = 1 THEN 1 ELSE 0 END), 0) as successfulRequests,
COALESCE(MAX(timestamp), '') as lastUsed
FROM usage_history
${whereClause}
FROM ${unifiedSource} AS _u
${unifiedWhere}
GROUP BY LOWER(model), LOWER(provider), serviceTier
ORDER BY requests DESC
`
@@ -498,8 +594,8 @@ export async function GET(request: Request) {
COALESCE(SUM(tokens_cache_read), 0) as cacheReadTokens,
COALESCE(SUM(tokens_cache_creation), 0) as cacheCreationTokens,
COALESCE(SUM(tokens_reasoning), 0) as reasoningTokens
FROM usage_history
${whereClause}
FROM ${unifiedSource} AS _u
${unifiedWhere}
GROUP BY LOWER(provider), LOWER(model), serviceTier
`
)
@@ -516,8 +612,8 @@ export async function GET(request: Request) {
COALESCE(SUM(tokens_input + tokens_output), 0) as totalTokens,
COALESCE(AVG(latency_ms), 0) as avgLatencyMs,
COALESCE(SUM(CASE WHEN success = 1 THEN 1 ELSE 0 END), 0) as successfulRequests
FROM usage_history
${whereClause}
FROM ${unifiedSource} AS _u
${unifiedWhere}
GROUP BY LOWER(provider)
ORDER BY requests DESC
`
@@ -608,9 +704,8 @@ export async function GET(request: Request) {
COALESCE(SUM(tokens_cache_creation), 0) as cacheCreationTokens,
COALESCE(SUM(tokens_reasoning), 0) as reasoningTokens,
COALESCE(SUM(tokens_input + tokens_output), 0) as totalTokens
FROM usage_history
${whereClause}
FROM ${unifiedSource} AS _u
${unifiedWhere}
GROUP BY serviceTier, LOWER(provider), LOWER(model)
`
)
@@ -661,8 +756,8 @@ export async function GET(request: Request) {
strftime('%w', timestamp) as dayOfWeek,
COUNT(*) as requests,
COALESCE(SUM(tokens_input + tokens_output), 0) as totalTokens
FROM usage_history
${whereClause}
FROM ${unifiedSource} AS _u
${unifiedWhere}
GROUP BY DATE(timestamp), strftime('%w', timestamp)
)
GROUP BY dayOfWeek
@@ -1116,19 +1211,58 @@ export async function GET(request: Request) {
}
const presetSinceIso = getRangeStartIso(presetRange);
const presetConditions = [];
const presetParams: Record<string, string> = {};
// Build unified source for preset cost queries (same UNION logic as main query).
const presetNeedsAggregated = !presetSinceIso || presetSinceIso < rawCutoffIso;
const presetRawConds: string[] = [];
if (presetSinceIso) {
presetConditions.push("timestamp >= @presetSince");
presetRawConds.push("timestamp >= @presetSince");
presetParams.presetSince = presetSinceIso;
}
if (apiKeyWhere) {
presetConditions.push(apiKeyWhere);
presetRawConds.push(apiKeyWhere);
Object.assign(presetParams, params);
}
const presetRawWhere =
presetRawConds.length > 0 ? `WHERE ${presetRawConds.join(" AND ")}` : "";
const presetWhere =
presetConditions.length > 0 ? `WHERE ${presetConditions.join(" AND ")}` : "";
const presetAggConds: string[] = [];
if (presetSinceIso) {
const presetSinceDate = presetSinceIso.split("T")[0];
presetAggConds.push("date >= @presetSinceDate");
presetParams.presetSinceDate = presetSinceDate;
}
const presetAggWhere =
presetAggConds.length > 0 ? `WHERE ${presetAggConds.join(" AND ")}` : "";
const presetUnifiedSource = presetNeedsAggregated
? `(
SELECT timestamp, provider, model, service_tier,
tokens_input, tokens_output,
tokens_cache_read, tokens_cache_creation, tokens_reasoning
FROM usage_history
${presetRawWhere}
UNION ALL
SELECT
date || 'T12:00:00.000Z' as timestamp,
provider, model,
'standard' as service_tier,
total_input_tokens as tokens_input,
total_output_tokens as tokens_output,
0 as tokens_cache_read,
0 as tokens_cache_creation,
0 as tokens_reasoning
FROM daily_usage_summary
${presetAggWhere}
)`
: `(SELECT timestamp, provider, model, service_tier,
tokens_input, tokens_output,
tokens_cache_read, tokens_cache_creation, tokens_reasoning
FROM usage_history
${presetRawWhere}
)`;
const presetModelRows = db
.prepare(
@@ -1142,8 +1276,7 @@ export async function GET(request: Request) {
COALESCE(SUM(tokens_cache_read), 0) as cacheReadTokens,
COALESCE(SUM(tokens_cache_creation), 0) as cacheCreationTokens,
COALESCE(SUM(tokens_reasoning), 0) as reasoningTokens
FROM usage_history
${presetWhere}
FROM ${presetUnifiedSource} AS _pu
GROUP BY LOWER(model), LOWER(provider), serviceTier
`
)

11
src/lib/db/cleanup.ts
View File

@@ -6,6 +6,7 @@
import { getDbInstance } from "./core";
import { getUserDatabaseSettings } from "./databaseSettings";
import { rollupUsageHistoryBeforeDate } from "@/lib/usage/aggregateHistory";
interface CleanupResult {
deleted: number;
@@ -85,9 +86,19 @@ export async function cleanupUsageHistory(): Promise<CleanupResult> {
const cutoffDate = new Date();
cutoffDate.setDate(cutoffDate.getDate() - retentionDays);
const cutoffISO = cutoffDate.toISOString();
const cutoffDateStr = cutoffISO.split("T")[0];
const result: CleanupResult = { deleted: 0, errors: 0 };
try {
// Roll up rows that are about to be deleted into daily_usage_summary so that
// the analytics route can still surface historical data via the UNION query.
await rollupUsageHistoryBeforeDate(cutoffDateStr);
} catch (err: unknown) {
// Non-fatal: log but continue with deletion so cleanup still runs.
console.error("[Cleanup] Error rolling up usage_history before deletion:", err);
}
try {
const stmt = db.prepare("DELETE FROM usage_history WHERE timestamp < ?");
const runResult = stmt.run(cutoffISO);

61
src/lib/usage/aggregateHistory.ts
View File

@@ -1,6 +1,6 @@
/**
* Aggregation utility functions for usage data summarization.
* Rolls up quota_snapshots into hourly and daily summary tables.
* Rolls up usage_history (and quota_snapshots) into daily summary tables.
*
* @module lib/usage/aggregateHistory
*/
@@ -130,6 +130,65 @@ export async function rollupHourlyQuota(
return result;
}
/**
* Roll up usage_history into daily_usage_summary before raw rows are deleted.
* This is the authoritative rollup — sourced from actual per-request token data,
* not from quota_snapshots. Should be called before cleanupUsageHistory() deletes rows.
*
* The ON CONFLICT clause uses SUM so re-running is additive-safe: if a date already
* has a partial rollup (e.g. from a previous partial cleanup), new rows accumulate.
*
* @param beforeDate - ISO date string (YYYY-MM-DD). Rows strictly before this date are rolled up.
* @returns Aggregation result with counts
*/
export async function rollupUsageHistoryBeforeDate(beforeDate: string): Promise<AggregationResult> {
const db = getDbInstance();
const result: AggregationResult = {
processed: 0,
inserted: 0,
errors: 0,
};
try {
const aggregateQuery = `
INSERT INTO daily_usage_summary (provider, model, date, total_requests, total_input_tokens, total_output_tokens, total_cost)
SELECT
LOWER(provider) as provider,
LOWER(model) as model,
DATE(timestamp) as date,
COUNT(*) as total_requests,
COALESCE(SUM(tokens_input), 0) as total_input_tokens,
COALESCE(SUM(tokens_output), 0) as total_output_tokens,
0.0 as total_cost
FROM usage_history
WHERE DATE(timestamp) < ?
AND provider IS NOT NULL AND provider != ''
AND model IS NOT NULL AND model != ''
GROUP BY LOWER(provider), LOWER(model), DATE(timestamp)
ON CONFLICT(provider, model, date) DO UPDATE SET
total_requests = daily_usage_summary.total_requests + excluded.total_requests,
total_input_tokens = daily_usage_summary.total_input_tokens + excluded.total_input_tokens,
total_output_tokens = daily_usage_summary.total_output_tokens + excluded.total_output_tokens
`;
const stmt = db.prepare(aggregateQuery);
const runResult = stmt.run(beforeDate);
result.processed = runResult.changes;
result.inserted = runResult.changes;
console.log(
`[Aggregation] usage_history rollup: ${result.inserted} rows for dates before ${beforeDate}`
);
} catch (err: any) {
console.error("[Aggregation] usage_history rollup error:", err);
result.errors++;
}
return result;
}
/**
* Get the cutoff date for raw data based on retention settings.
* Data older than this should be aggregated and cleaned up.

15
src/lib/usage/tokenAccounting.ts
View File

@@ -48,6 +48,13 @@ export function getLoggedInputTokens(tokens: unknown): number {
return toFiniteNumber(tokenRecord.input);
}
// Prefer prompt_tokens when present: translators normalize provider-specific
// usage into OpenAI shape there, and may keep input_tokens for compatibility.
// Treating that input_tokens as raw Claude usage would add cache tokens again.
if (tokenRecord.prompt_tokens !== undefined && tokenRecord.prompt_tokens !== null) {
return toFiniteNumber(tokenRecord.prompt_tokens);
}
if (tokenRecord.input_tokens !== undefined && tokenRecord.input_tokens !== null) {
// Anthropic / anthropic-compatible-cc streaming: input_tokens is only the
// non-cached portion. The cache counters sit as separate top-level fields
@@ -60,13 +67,7 @@ export function getLoggedInputTokens(tokens: unknown): number {
);
}
// prompt_tokens from translator/extractor already includes cache tokens:
// - OpenAI format: prompt_tokens inherently includes cached
// - Claude non-streaming: extractUsageFromResponse sums input + cache_read + cache_creation
// - Claude streaming: extractUsage (after fix) sums input + cache_read + cache_creation
// Do NOT add cache fields here — would double-count.
const promptTokens = toFiniteNumber(tokenRecord.prompt_tokens);
return promptTokens;
return 0;
}
export function getLoggedOutputTokens(tokens: unknown): number {

9
src/shared/validation/settingsSchemas.ts
View File

@@ -8,7 +8,7 @@
import { z } from "zod";
import { COMBO_CONFIG_MODES } from "@/shared/constants/comboConfigMode";
import { MAX_REQUEST_BODY_LIMIT_MB, MIN_REQUEST_BODY_LIMIT_MB } from "@/shared/constants/bodySize";
import { HIDEABLE_SIDEBAR_ITEM_IDS, SIDEBAR_SECTIONS } from "@/shared/constants/sidebarVisibility";
import { HIDEABLE_SIDEBAR_ITEM_IDS } from "@/shared/constants/sidebarVisibility";
import { ACCOUNT_FALLBACK_STRATEGY_VALUES } from "@/shared/constants/routingStrategies";
const signatureCacheModeValues = ["enabled", "bypass", "bypass-strict"] as const;
@@ -38,11 +38,6 @@ export const updateSettingsSchema = z.object({
autoRefreshProviderQuotaInterval: z.number().int().min(10).max(3600).optional(),
debugMode: z.boolean().optional(),
hiddenSidebarItems: z.array(z.enum(HIDEABLE_SIDEBAR_ITEM_IDS)).optional(),
sidebarSectionOrder: z
.array(z.enum(SIDEBAR_SECTIONS.map((s) => s.id) as [string, ...string[]]))
.optional(),
sidebarItemOrder: z.record(z.string(), z.array(z.string().max(100))).optional(),
sidebarActivePreset: z.enum(["all", "minimal", "developer", "admin"]).nullable().optional(),
comboConfigMode: z.enum(COMBO_CONFIG_MODES).optional(),
codexServiceTier: z
.object({
@@ -316,7 +311,7 @@ export const databaseSettingsSchema = z.object(
// Aggregation settings
aggregation: z.object({
enabled: z.boolean(),
rawDataRetentionDays: z.number().int().min(1).max(90),
rawDataRetentionDays: z.number().int().min(1).max(3650),
granularity: z.literal("hourly").or(z.literal("daily")).or(z.literal("weekly")),
}),

57
tests/unit/token-accounting-input-fix.test.ts
View File

@@ -1,51 +1,15 @@
/**
* Unit tests for getLoggedInputTokens fix — Anthropic / anthropic-compatible-cc
*
* The bug: Claude streaming sets prompt_tokens = input_tokens (non-cached only).
* Fix: extractUsage in usageTracking.ts now sums input + cache_read + cache_creation
* into prompt_tokens, consistent with the non-streaming extractor.
*
* getLoggedInputTokens itself also has a safety-net: when raw `input_tokens`
* is present (e.g. from a raw API response), it adds cache tokens too.
* getLoggedInputTokens has a safety-net: when raw `input_tokens` is present
* (e.g. from a raw API response), it adds cache tokens too. When both
* `prompt_tokens` and `input_tokens` are present, `prompt_tokens` wins because
* stream translators keep `input_tokens` as a compatibility alias.
*/
import { describe, it } from "node:test";
import assert from "node:assert/strict";
// ── Inline the fixed logic from tokenAccounting.ts ──────────────────────
function asRecord(value) {
return value && typeof value === "object" && !Array.isArray(value) ? value : {};
}
function toFiniteNumber(value) {
if (typeof value === "number" && Number.isFinite(value)) return value;
if (typeof value === "string" && value.trim().length > 0) {
const parsed = Number(value);
return Number.isFinite(parsed) ? parsed : 0;
}
return 0;
}
function getLoggedInputTokens(tokens) {
const tokenRecord = asRecord(tokens);
if (tokenRecord.input !== undefined && tokenRecord.input !== null) {
return toFiniteNumber(tokenRecord.input);
}
if (tokenRecord.input_tokens !== undefined && tokenRecord.input_tokens !== null) {
return (
toFiniteNumber(tokenRecord.input_tokens) +
toFiniteNumber(tokenRecord.cache_read_input_tokens) +
toFiniteNumber(tokenRecord.cache_creation_input_tokens)
);
}
// prompt_tokens from translator/extractor already includes cache tokens
const promptTokens = toFiniteNumber(tokenRecord.prompt_tokens);
return promptTokens;
}
import { getLoggedInputTokens } from "../../src/lib/usage/tokenAccounting.ts";
// ── Tests ────────────────────────────────────────────────────────────────
@@ -95,6 +59,17 @@ describe("getLoggedInputTokens — input fix for Anthropic streaming", () => {
assert.equal(getLoggedInputTokens(tokens), 113616);
});
it("translated Claude stream usage: prompt_tokens wins over compatibility input_tokens", () => {
const tokens = {
prompt_tokens: 600_000,
input_tokens: 600_000,
completion_tokens: 1_000,
output_tokens: 1_000,
cache_read_input_tokens: 600_000,
};
assert.equal(getLoggedInputTokens(tokens), 600_000);
});
it("OpenAI format: prompt_tokens=1000, no cache top-level fields → 1000", () => {
const tokens = {
prompt_tokens: 1000,