fix(usage): aggregate provider window costs in SQL (#8892)

Validated in local merge-train T4 (HouMinXi+Zartharas+Andrian+artickc)
2026-08-06 07:12:12 +03:00 · 2026-08-05 21:52:38 -05:00
parent bf37618a61
commit 63c484d062
2 changed files with 383 additions and 107 deletions
--- a/src/lib/usage/providerWindowCosts.ts
+++ b/src/lib/usage/providerWindowCosts.ts
@@ -1,3 +1,11 @@
+/**
+ * @file providerWindowCosts.ts
+ * @description Provider weekly USD cost breakdown for the dashboard modal.
+ *
+ * @changes
+ * - [2026-07-24] [Composer] - Aggregate usage_history in SQL instead of loading all rows into JS
+ */
+
 import { getCostSummary } from "@/domain/costRules";
 import { getApiKeys } from "@/lib/db/apiKeys";
 import { getDbInstance } from "@/lib/db/core";
@@ -10,6 +18,27 @@ const RECORDED_COST_MATCH_TOLERANCE_MS = 30_000;

 type JsonRecord = Record<string, unknown>;

+interface UsageHistoryFilter {
+  whereSql: string;
+  params: Record<string, unknown>;
+}
+
+interface AggregatedUsageCostRow {
+  apiKeyId: string | null;
+  apiKeyName: string | null;
+  provider: string;
+  model: string;
+  serviceTier: string;
+  requests: number;
+  promptTokens: number;
+  completionTokens: number;
+  cacheReadTokens: number;
+  cacheCreationTokens: number;
+  reasoningTokens: number;
+  totalTokens: number;
+  lastUsed: string | null;
+}
+
 interface UsageCostRow {
  id: number;
  apiKeyId: string | null;
@@ -26,6 +55,11 @@ interface UsageCostRow {
  timestamp: string | null;
 }

+interface RecordedCostSummary {
+  totalCost: number;
+  entryCount: number;
+}
+
 interface RecordedCostRow {
  rowId: number;
  apiKeyId: string;
@@ -264,7 +298,7 @@ async function getCurrentApiKeyNames(): Promise<Map<string, string>> {
  return names;
 }

-function uniqueApiKeyIds(rows: UsageCostRow[]): string[] {
+function uniqueApiKeyIds(rows: Array<{ apiKeyId: string | null }>): string[] {
  return Array.from(
    new Set(
      rows
@@ -274,6 +308,162 @@ function uniqueApiKeyIds(rows: UsageCostRow[]): string[] {
  );
 }

+function buildUsageHistoryFilter(
+  providerKey: string,
+  windowStartAt: string,
+  nowIso: string,
+  windowResetAt: string | null,
+  connectionId: string | null
+): UsageHistoryFilter {
+  const where = [
+    "LOWER(provider) = @provider",
+    "timestamp >= @since",
+    "timestamp <= @nowIso",
+    "COALESCE(success, 1) = 1",
+  ];
+  const params: Record<string, unknown> = {
+    provider: providerKey,
+    since: windowStartAt,
+    nowIso,
+  };
+  if (windowResetAt) {
+    where.push("timestamp < @resetAt");
+    params.resetAt = windowResetAt;
+  }
+  if (connectionId) {
+    where.push("connection_id = @connectionId");
+    params.connectionId = connectionId;
+  }
+  return { whereSql: where.join(" AND "), params };
+}
+
+function fetchAggregatedUsageRows(filter: UsageHistoryFilter): AggregatedUsageCostRow[] {
+  return getDbInstance()
+    .prepare<AggregatedUsageCostRow>(
+      `
+      SELECT
+        NULLIF(api_key_id, '') as apiKeyId,
+        NULLIF(api_key_name, '') as apiKeyName,
+        LOWER(provider) as provider,
+        LOWER(model) as model,
+        COALESCE(NULLIF(service_tier, ''), 'standard') as serviceTier,
+        COUNT(*) as requests,
+        COALESCE(SUM(tokens_input), 0) as promptTokens,
+        COALESCE(SUM(tokens_output), 0) as completionTokens,
+        COALESCE(SUM(tokens_cache_read), 0) as cacheReadTokens,
+        COALESCE(SUM(tokens_cache_creation), 0) as cacheCreationTokens,
+        COALESCE(SUM(tokens_reasoning), 0) as reasoningTokens,
+        COALESCE(SUM(tokens_input + tokens_output), 0) as totalTokens,
+        MAX(timestamp) as lastUsed
+      FROM usage_history
+      WHERE ${filter.whereSql}
+      GROUP BY
+        COALESCE(NULLIF(api_key_id, ''), ''),
+        COALESCE(NULLIF(api_key_name, ''), ''),
+        LOWER(provider),
+        LOWER(model),
+        COALESCE(NULLIF(service_tier, ''), 'standard')
+      ORDER BY totalTokens DESC
+      `
+    )
+    .all(filter.params);
+}
+
+function fetchUsageRequestCountByApiKey(filter: UsageHistoryFilter): Map<string, number> {
+  const rows = getDbInstance()
+    .prepare<{ apiKeyId: string; requestCount: number }>(
+      `
+      SELECT
+        COALESCE(NULLIF(api_key_id, ''), '') as apiKeyId,
+        COUNT(*) as requestCount
+      FROM usage_history
+      WHERE ${filter.whereSql}
+      GROUP BY COALESCE(NULLIF(api_key_id, ''), '')
+      `
+    )
+    .all(filter.params);
+
+  const counts = new Map<string, number>();
+  for (const row of rows) {
+    if (!row.apiKeyId) continue;
+    counts.set(row.apiKeyId, toNumber(row.requestCount));
+  }
+  return counts;
+}
+
+function fetchDetailedUsageRowsForApiKey(
+  filter: UsageHistoryFilter,
+  apiKeyId: string
+): UsageCostRow[] {
+  return getDbInstance()
+    .prepare<UsageCostRow>(
+      `
+      SELECT
+        id,
+        NULLIF(api_key_id, '') as apiKeyId,
+        NULLIF(api_key_name, '') as apiKeyName,
+        LOWER(provider) as provider,
+        LOWER(model) as model,
+        COALESCE(NULLIF(service_tier, ''), 'standard') as serviceTier,
+        COALESCE(tokens_input, 0) as promptTokens,
+        COALESCE(tokens_output, 0) as completionTokens,
+        COALESCE(tokens_cache_read, 0) as cacheReadTokens,
+        COALESCE(tokens_cache_creation, 0) as cacheCreationTokens,
+        COALESCE(tokens_reasoning, 0) as reasoningTokens,
+        COALESCE(tokens_input + tokens_output, 0) as totalTokens,
+        timestamp
+      FROM usage_history
+      WHERE ${filter.whereSql}
+        AND api_key_id = @apiKeyId
+      ORDER BY timestamp ASC, id ASC
+      `
+    )
+    .all({ ...filter.params, apiKeyId });
+}
+
+function getRecordedCostSummariesByApiKey(
+  apiKeyIds: string[],
+  sinceMs: number,
+  untilMs: number
+): Map<string, RecordedCostSummary> {
+  if (apiKeyIds.length === 0) return new Map();
+
+  try {
+    const params: Record<string, unknown> = {
+      sinceMs: Math.max(0, sinceMs - RECORDED_COST_MATCH_TOLERANCE_MS),
+      untilMs: untilMs + RECORDED_COST_MATCH_TOLERANCE_MS,
+    };
+    const placeholders = appendNamedPlaceholders(params, "apiKey", apiKeyIds);
+    const rows = getDbInstance()
+      .prepare<{ apiKeyId: string; totalCost: number; entryCount: number }>(
+        `
+        SELECT
+          api_key_id as apiKeyId,
+          COALESCE(SUM(cost), 0) as totalCost,
+          COUNT(*) as entryCount
+        FROM domain_cost_history
+        WHERE api_key_id IN (${placeholders})
+          AND timestamp >= @sinceMs
+          AND timestamp <= @untilMs
+        GROUP BY api_key_id
+      `
+      )
+      .all(params);
+
+    const summaries = new Map<string, RecordedCostSummary>();
+    for (const row of rows) {
+      if (!row.apiKeyId) continue;
+      summaries.set(row.apiKeyId, {
+        totalCost: Math.max(0, toNumber(row.totalCost)),
+        entryCount: toNumber(row.entryCount),
+      });
+    }
+    return summaries;
+  } catch {
+    return new Map();
+  }
+}
+
 function appendNamedPlaceholders(
  params: Record<string, unknown>,
  prefix: string,
@@ -387,6 +577,75 @@ async function getUsageRowCostUsd(
  );
 }

+async function getAggregatedGroupCostUsd(row: AggregatedUsageCostRow): Promise<number> {
+  return roundUsd(
+    await calculateCost(
+      row.provider,
+      row.model,
+      {
+        input: toNumber(row.promptTokens),
+        output: toNumber(row.completionTokens),
+        cacheRead: toNumber(row.cacheReadTokens),
+        cacheCreation: toNumber(row.cacheCreationTokens),
+        reasoning: toNumber(row.reasoningTokens),
+      },
+      { serviceTier: row.serviceTier }
+    )
+  );
+}
+
+function distributeRecordedCostAcrossGroups(
+  groups: AggregatedUsageCostRow[],
+  recordedTotalUsd: number,
+  groupCalculatedCosts: number[]
+): number[] {
+  const calculatedTotal = groupCalculatedCosts.reduce((sum, value) => sum + value, 0);
+  if (calculatedTotal <= 0) {
+    const evenShare = recordedTotalUsd / Math.max(groups.length, 1);
+    return groups.map(() => roundUsd(evenShare));
+  }
+  return groupCalculatedCosts.map((value) =>
+    roundUsd((recordedTotalUsd * value) / calculatedTotal)
+  );
+}
+
+async function buildApiKeyCostAllocations(args: {
+  groups: AggregatedUsageCostRow[];
+  filter: UsageHistoryFilter;
+  usageRequestCount: number;
+  recordedSummary: RecordedCostSummary | undefined;
+  recordedCostsByApiKey: Map<string, RecordedCostRow[]>;
+  usedRecordedRows: Set<number>;
+}): Promise<number[]> {
+  const calculatedCosts = await Promise.all(
+    args.groups.map((group) => getAggregatedGroupCostUsd(group))
+  );
+
+  if (!args.recordedSummary || args.recordedSummary.entryCount <= 0) {
+    return calculatedCosts;
+  }
+
+  if (args.recordedSummary.entryCount === args.usageRequestCount) {
+    return distributeRecordedCostAcrossGroups(
+      args.groups,
+      args.recordedSummary.totalCost,
+      calculatedCosts
+    );
+  }
+
+  const apiKeyId = args.groups[0]?.apiKeyId;
+  if (!apiKeyId) return calculatedCosts;
+
+  const detailedRows = fetchDetailedUsageRowsForApiKey(args.filter, apiKeyId);
+  const rowCosts = await Promise.all(
+    detailedRows.map((row) =>
+      getUsageRowCostUsd(row, args.recordedCostsByApiKey, args.usedRecordedRows)
+    )
+  );
+  const detailedTotal = roundUsd(rowCosts.reduce((sum, value) => sum + value, 0));
+  return distributeRecordedCostAcrossGroups(args.groups, detailedTotal, calculatedCosts);
+}
+
 export async function getProviderWindowCostBreakdown({
  provider,
  connectionId = null,
@@ -402,130 +661,112 @@ export async function getProviderWindowCostBreakdown({
  const windowStartAt = new Date(window.startMs).toISOString();
  const windowResetAt = window.resetMs ? new Date(window.resetMs).toISOString() : null;
  const nowIso = new Date(nowMs).toISOString();
-
-  const where = [
-    "LOWER(provider) = @provider",
-    "timestamp >= @since",
-    "timestamp <= @nowIso",
-    "COALESCE(success, 1) = 1",
-  ];
-  const params: Record<string, unknown> = {
-    provider: providerKey,
-    since: windowStartAt,
+  const filter = buildUsageHistoryFilter(
+    providerKey,
+    windowStartAt,
    nowIso,
-  };
-  if (windowResetAt) {
-    where.push("timestamp < @resetAt");
-    params.resetAt = windowResetAt;
-  }
-  if (connectionId) {
-    where.push("connection_id = @connectionId");
-    params.connectionId = connectionId;
-  }
-
-  const usageRows = getDbInstance()
-    .prepare<UsageCostRow>(
-      `
-      SELECT
-        id,
-        NULLIF(api_key_id, '') as apiKeyId,
-        NULLIF(api_key_name, '') as apiKeyName,
-        LOWER(provider) as provider,
-        LOWER(model) as model,
-        COALESCE(NULLIF(service_tier, ''), 'standard') as serviceTier,
-        COALESCE(tokens_input, 0) as promptTokens,
-        COALESCE(tokens_output, 0) as completionTokens,
-        COALESCE(tokens_cache_read, 0) as cacheReadTokens,
-        COALESCE(tokens_cache_creation, 0) as cacheCreationTokens,
-        COALESCE(tokens_reasoning, 0) as reasoningTokens,
-        COALESCE(tokens_input + tokens_output, 0) as totalTokens,
-        timestamp
-      FROM usage_history
-      WHERE ${where.join(" AND ")}
-      ORDER BY timestamp ASC, id ASC
-      `
-    )
-    .all(params);
-
-  const currentApiKeyNames = await getCurrentApiKeyNames();
-  const recordedCostsByApiKey = getRecordedCostsByApiKey(
-    uniqueApiKeyIds(usageRows),
-    window.startMs,
-    nowMs
+    windowResetAt,
+    connectionId
  );
+
+  const aggregatedRows = fetchAggregatedUsageRows(filter);
+  const usageRequestCounts = fetchUsageRequestCountByApiKey(filter);
+  const currentApiKeyNames = await getCurrentApiKeyNames();
+  const apiKeyIds = uniqueApiKeyIds(aggregatedRows);
+  const recordedSummaries = getRecordedCostSummariesByApiKey(apiKeyIds, window.startMs, nowMs);
+  const recordedCostsByApiKey = getRecordedCostsByApiKey(apiKeyIds, window.startMs, nowMs);
  const usedRecordedRows = new Set<number>();
  const byApiKey = new Map<string, ProviderWindowCostAggregateRow>();

-  for (const row of usageRows) {
-    const apiKeyId = row.apiKeyId || null;
-    const apiKeyName = row.apiKeyName || null;
-    const apiKeyKey = makeApiKeyKey(apiKeyId, apiKeyName);
+  const groupsByApiKey = new Map<string, AggregatedUsageCostRow[]>();
+  for (const row of aggregatedRows) {
+    const apiKeyKey = makeApiKeyKey(row.apiKeyId, row.apiKeyName);
+    const list = groupsByApiKey.get(apiKeyKey) ?? [];
+    list.push(row);
+    groupsByApiKey.set(apiKeyKey, list);
+  }
+
+  for (const [apiKeyKey, groups] of groupsByApiKey.entries()) {
+    const first = groups[0];
+    const apiKeyId = first.apiKeyId || null;
+    const apiKeyName = first.apiKeyName || null;
    const displayName =
      (apiKeyId ? currentApiKeyNames.get(apiKeyId) : null) ||
      apiKeyName ||
      apiKeyId ||
      "Unattributed";
-    const costUsd = roundUsd(
-      await getUsageRowCostUsd(row, recordedCostsByApiKey, usedRecordedRows)
-    );
+    const usageRequestCount = apiKeyId
+      ? (usageRequestCounts.get(apiKeyId) ?? 0)
+      : groups.reduce((sum, group) => sum + toNumber(group.requests), 0);
+    const groupCosts = await buildApiKeyCostAllocations({
+      groups,
+      filter,
+      usageRequestCount,
+      recordedSummary: apiKeyId ? recordedSummaries.get(apiKeyId) : undefined,
+      recordedCostsByApiKey,
+      usedRecordedRows,
+    });

-    let aggregate = byApiKey.get(apiKeyKey);
-    if (!aggregate) {
-      let limitUsd: number | null = null;
-      let limitPeriod: string | null = null;
-      let budgetResetAt: string | null = null;
-      if (apiKeyId) {
-        const summary = getCostSummary(apiKeyId);
-        if (summary.activeLimitUsd > 0) {
-          limitUsd = summary.activeLimitUsd;
-          limitPeriod = summary.resetInterval;
-          budgetResetAt =
-            typeof summary.nextResetAt === "number" && Number.isFinite(summary.nextResetAt)
-              ? new Date(summary.nextResetAt).toISOString()
-              : null;
-        }
+    let limitUsd: number | null = null;
+    let limitPeriod: string | null = null;
+    let budgetResetAt: string | null = null;
+    if (apiKeyId) {
+      const summary = getCostSummary(apiKeyId);
+      if (summary.activeLimitUsd > 0) {
+        limitUsd = summary.activeLimitUsd;
+        limitPeriod = summary.resetInterval;
+        budgetResetAt =
+          typeof summary.nextResetAt === "number" && Number.isFinite(summary.nextResetAt)
+            ? new Date(summary.nextResetAt).toISOString()
+            : null;
      }
-      aggregate = {
-        apiKeyKey,
-        apiKeyId,
-        apiKeyName: displayName,
-        requests: 0,
-        promptTokens: 0,
-        completionTokens: 0,
-        totalTokens: 0,
-        costUsd: 0,
-        limitUsd,
-        limitPeriod,
-        limitUsedPercent: null,
-        budgetResetAt,
-        lastUsed: null,
-        models: [],
-        modelMap: new Map(),
-      };
-      byApiKey.set(apiKeyKey, aggregate);
    }

-    aggregate.requests += 1;
-    aggregate.promptTokens += toNumber(row.promptTokens);
-    aggregate.completionTokens += toNumber(row.completionTokens);
-    aggregate.totalTokens += toNumber(row.totalTokens);
-    aggregate.costUsd = roundUsd(aggregate.costUsd + costUsd);
-    if (!aggregate.lastUsed || (row.timestamp && row.timestamp > aggregate.lastUsed)) {
-      aggregate.lastUsed = row.timestamp || aggregate.lastUsed;
-    }
-    const modelKey = `${row.provider}\0${row.model}\0${row.serviceTier}`;
-    const model = aggregate.modelMap.get(modelKey) ?? {
-      model: row.model,
-      provider: row.provider,
-      serviceTier: row.serviceTier,
+    const aggregate: ProviderWindowCostAggregateRow = {
+      apiKeyKey,
+      apiKeyId,
+      apiKeyName: displayName,
      requests: 0,
+      promptTokens: 0,
+      completionTokens: 0,
      totalTokens: 0,
      costUsd: 0,
+      limitUsd,
+      limitPeriod,
+      limitUsedPercent: null,
+      budgetResetAt,
+      lastUsed: null,
+      models: [],
+      modelMap: new Map(),
    };
-    model.requests += 1;
-    model.totalTokens += toNumber(row.totalTokens);
-    model.costUsd = roundUsd(model.costUsd + costUsd);
-    aggregate.modelMap.set(modelKey, model);
+
+    groups.forEach((row, index) => {
+      const costUsd = groupCosts[index] ?? 0;
+      aggregate.requests += toNumber(row.requests);
+      aggregate.promptTokens += toNumber(row.promptTokens);
+      aggregate.completionTokens += toNumber(row.completionTokens);
+      aggregate.totalTokens += toNumber(row.totalTokens);
+      aggregate.costUsd = roundUsd(aggregate.costUsd + costUsd);
+      if (!aggregate.lastUsed || (row.lastUsed && row.lastUsed > aggregate.lastUsed)) {
+        aggregate.lastUsed = row.lastUsed || aggregate.lastUsed;
+      }
+
+      const modelKey = `${row.provider}\0${row.model}\0${row.serviceTier}`;
+      const model = aggregate.modelMap.get(modelKey) ?? {
+        model: row.model,
+        provider: row.provider,
+        serviceTier: row.serviceTier,
+        requests: 0,
+        totalTokens: 0,
+        costUsd: 0,
+      };
+      model.requests += toNumber(row.requests);
+      model.totalTokens += toNumber(row.totalTokens);
+      model.costUsd = roundUsd(model.costUsd + costUsd);
+      aggregate.modelMap.set(modelKey, model);
+    });
+
+    byApiKey.set(apiKeyKey, aggregate);
  }

  const breakdownRows = Array.from(byApiKey.values())
--- a/tests/unit/provider-window-costs.test.ts
+++ b/tests/unit/provider-window-costs.test.ts
@@ -450,3 +450,38 @@ test("provider window costs prefer recorded USD history over repricing usage tok
  assert.equal(result.rows[0].apiKeyName, "Recorded USD Key");
  assert.equal(result.rows[0].costUsd, 17);
 });
+
+test("provider window costs aggregate many usage rows in SQL without per-row scans", async () => {
+  await localDb.updatePricing({
+    "featherless-ai": {
+      "glm-5.2": { input: 1, output: 1, cached: 0, cache_creation: 0, reasoning: 0 },
+    },
+  });
+
+  const key = await apiKeys.createApiKey("Featherless Key", "machine-featherless-agg");
+
+  for (let index = 0; index < 250; index += 1) {
+    await usageHistory.saveRequestUsage({
+      provider: "featherless-ai",
+      model: "glm-5.2",
+      connectionId: "featherless-conn",
+      apiKeyId: key.id,
+      apiKeyName: "Featherless Key",
+      tokens: { input: 1_000, output: 0 },
+      timestamp: new Date(Date.parse("2026-06-26T10:00:00.000Z") + index * 1_000).toISOString(),
+    });
+  }
+
+  const result = await getProviderWindowCostBreakdown({
+    provider: "featherless-ai",
+    connectionId: "featherless-conn",
+    now: Date.parse("2026-06-28T12:00:00.000Z"),
+  });
+
+  assert.equal(result.rows.length, 1);
+  assert.equal(result.rows[0].requests, 250);
+  assert.equal(result.rows[0].promptTokens, 250_000);
+  assert.equal(result.rows[0].costUsd, 0.25);
+  assert.equal(result.rows[0].models.length, 1);
+  assert.equal(result.rows[0].models[0].requests, 250);
+});