diff --git a/app/RELEASE_NOTES.md b/app/RELEASE_NOTES.md index 680d966..32bc599 100644 --- a/app/RELEASE_NOTES.md +++ b/app/RELEASE_NOTES.md @@ -1,6 +1,7 @@ # Release Notes ## Unreleased +- Added lazy, exact hourly request-statistics caching for faster administrator reports, with pricing invalidation and manual UTC-range rebuilds. - Added separate chat-token and rerank-search reporting and dashboard charts. - Added request-type-aware rerank pricing management and filters. - Added administrator-only Azure Retail Pricing preview, rule mapping, auditing, and transactional import. diff --git a/app/src/app/reporting/create/reporting-create-client.tsx b/app/src/app/reporting/create/reporting-create-client.tsx index ab83fff..a117a55 100644 --- a/app/src/app/reporting/create/reporting-create-client.tsx +++ b/app/src/app/reporting/create/reporting-create-client.tsx @@ -16,7 +16,9 @@ import { YAxis, ZAxis, } from "recharts"; +import { toast } from "sonner"; +import { Button } from "~/components/ui/button"; import { type ChartConfig, ChartContainer, @@ -76,7 +78,11 @@ const timeframeOptions = [ const pad = (value: number) => String(value).padStart(2, "0"); +const utcDateTimeLocalValue = (date: Date) => + `${date.getUTCFullYear()}-${pad(date.getUTCMonth() + 1)}-${pad(date.getUTCDate())}T${pad(date.getUTCHours())}:${pad(date.getUTCMinutes())}`; + export function ReportingCreateClient({ isAdmin }: { isAdmin: boolean }) { + const utils = api.useUtils(); const { data: groups = [], isLoading: groupsLoading } = api.reporting.listGroups.useQuery(); @@ -96,6 +102,26 @@ export function ReportingCreateClient({ isAdmin }: { isAdmin: boolean }) { const [quarterYear, setQuarterYear] = useState(currentYear); const [quarterValue, setQuarterValue] = useState(currentQuarter); const [yearValue, setYearValue] = useState(currentYear); + const [cacheFrom, setCacheFrom] = useState(() => { + const date = new Date(); + date.setUTCDate(date.getUTCDate() - 7); + return utcDateTimeLocalValue(date); + }); + const [cacheTo, setCacheTo] = useState(() => + utcDateTimeLocalValue(new Date()), + ); + const [cacheRebuiltAt, setCacheRebuiltAt] = useState(null); + const rebuildCache = api.reporting.rebuildCache.useMutation({ + onSuccess: async () => { + setCacheRebuiltAt(new Date().toISOString()); + await utils.reporting.getReport.invalidate(); + toast.success("Reporting-Cache erfolgreich neu aufgebaut."); + }, + onError: () => { + setCacheRebuiltAt(null); + toast.error("Reporting-Cache konnte nicht neu aufgebaut werden."); + }, + }); const availableGroups = useMemo(() => { const list = groups.map((group) => ({ @@ -178,6 +204,20 @@ export function ReportingCreateClient({ isAdmin }: { isAdmin: boolean }) { }, ); + const rebuildCacheForRange = () => { + if (!cacheFrom || !cacheTo) { + toast.error("Bitte Start und Ende des Zeitraums angeben."); + return; + } + if (cacheFrom >= cacheTo) { + toast.error("Das Ende muss nach dem Start liegen."); + return; + } + + setCacheRebuiltAt(null); + rebuildCache.mutate({ from: cacheFrom, to: cacheTo }); + }; + const numberFormatter = useMemo(() => new Intl.NumberFormat("de-DE"), []); const costFormatter = useMemo( () => @@ -506,6 +546,72 @@ export function ReportingCreateClient({ isAdmin }: { isAdmin: boolean }) { + {isAdmin ? ( +
+
+ Reporting-Cache neu aufbauen +
+

+ Zeitraum in UTC auswählen und die aggregierten Reporting-Daten + manuell neu berechnen. +

+
+
+ + setCacheFrom(event.target.value)} + type="datetime-local" + value={cacheFrom} + /> +
+
+ + setCacheTo(event.target.value)} + type="datetime-local" + value={cacheTo} + /> +
+ +
+ {rebuildCache.isError ? ( +
+ {rebuildCache.error?.message || + "Der Cache konnte nicht neu aufgebaut werden."} +
+ ) : null} + {cacheRebuiltAt ? ( + + Cache erfolgreich neu aufgebaut. + + ) : null} +
+ ) : null} +
@@ -1000,7 +1106,7 @@ export function ReportingCreateClient({ isAdmin }: { isAdmin: boolean }) { {costsUsed.map((cost) => ( {cost.model} diff --git a/app/src/server/api/routers/reporting.ts b/app/src/server/api/routers/reporting.ts index fb4fd5e..957da70 100644 --- a/app/src/server/api/routers/reporting.ts +++ b/app/src/server/api/routers/reporting.ts @@ -1,12 +1,10 @@ import { TRPCError } from "@trpc/server"; -import { and, eq, inArray, sql } from "drizzle-orm"; +import { and, eq, gte, inArray, isNull, lt, sql } from "drizzle-orm"; import { z } from "zod"; import { createTRPCRouter, protectedProcedure } from "~/server/api/trpc"; import { addCurrencyTotal, - addResolvedBreakdownCost, - addResolvedTotalCost, createBreakdownCostAggregate, createCurrencyTotals, createTotalCostAggregate, @@ -15,22 +13,21 @@ import { presentCurrencyTotals, scaledCostToNumber, } from "~/server/costAggregation"; -import { - buildCostStageIndex, - type CostStageRow, - resolveRequestCostStage, - resolveRerankCost, - tokenTypeLabelWithStage, -} from "~/server/costStageResolver"; import { apikeys, - costs, reportingGroupMembers, reportingGroups, reportingGroupViewers, - requests, + requestStatisticsCache, + requestStatisticsCacheBuckets, users, } from "~/server/db/schema"; +import { + ensureRequestStatisticsCache, + mergeCachedBreakdownCost, + mergeCachedTotalCost, + rebuildRequestStatisticsCache, +} from "~/server/requestStatisticsCache"; const adminProcedure = protectedProcedure.use(({ ctx, next }) => { if (!ctx.session.user.isAdmin) { @@ -385,7 +382,6 @@ export const reportingRouter = createTRPCRouter({ } const { start, end } = getDateRange(input.range); - const timeFilter = sql`${requests.requestTime} >= ${start.toISOString()} AND ${requests.requestTime} < ${end.toISOString()}`; const dayCount = Math.max( 1, Math.ceil((end.getTime() - start.getTime()) / 86_400_000), @@ -436,18 +432,7 @@ export const reportingRouter = createTRPCRouter({ } } - const scopedConditions = [timeFilter]; - if (scopedUserIds) { - scopedConditions.push(inArray(apikeys.owner, scopedUserIds)); - } - const scopedFilter = and(...scopedConditions); - - const costBucket = - input.range.type === "daily" - ? sql`date_trunc('hour', ${requests.requestTime})` - : sql`date_trunc('day', ${requests.requestTime})`; - const hourBucket = sql`extract(hour from ${requests.requestTime})`; - const dayBucket = sql`extract(dow from ${requests.requestTime})`; + await ensureRequestStatisticsCache(ctx.db, start, end); type UserModel = { model: string; @@ -490,7 +475,9 @@ export const reportingRouter = createTRPCRouter({ userId: string; name: string; model: string | null; - requestType: (typeof requests.requestType.enumValues)[number] | null; + requestType: + | (typeof requestStatisticsCache.requestType.enumValues)[number] + | null; inputTokens: number; cachedInputTokens: number; cacheWriteTokens: number; @@ -524,40 +511,6 @@ export const reportingRouter = createTRPCRouter({ const hourlyTotals = new Map(); const dayHourTotals = new Map(); - // Fetch and index all pricing rows once; resolve cost per request afterwards. - const costRows = await ctx.db - .select({ - model: costs.model, - price: costs.price, - validFrom: costs.validFrom, - requestType: costs.requestType, - billingUnit: costs.billingUnit, - tokenType: costs.tokenType, - unitOfMessure: costs.unitOfMessure, - currency: costs.currency, - stageType: costs.stageType, - stageMinTokens: costs.stageMinTokens, - stageMaxTokens: costs.stageMaxTokens, - }) - .from(costs); - - const costStageRows: CostStageRow[] = costRows.map((row) => ({ - model: row.model ?? "", - requestType: row.requestType, - billingUnit: row.billingUnit, - tokenType: row.tokenType ?? "", - price: Number(row.price ?? 0), - validFrom: row.validFrom ?? new Date(0), - unitOfMessure: (row.unitOfMessure ?? - null) as CostStageRow["unitOfMessure"], - currency: row.currency ? row.currency.trim().toUpperCase() : null, - stageType: row.stageType ?? "context_length", - stageMinTokens: Number(row.stageMinTokens ?? 0), - stageMaxTokens: row.stageMaxTokens ?? null, - })); - - const costStageIndex = buildCostStageIndex(costStageRows); - const usedCosts = new Map< string, { @@ -571,55 +524,6 @@ export const reportingRouter = createTRPCRouter({ } >(); - const registerUsedCost = ( - tokenType: "input" | "cached" | "cache_write" | "output", - usedCost: CostStageRow, - ) => { - const stageMin = usedCost.stageMinTokens; - const stageMax = usedCost.stageMaxTokens; - const tokenTypeLabel = tokenTypeLabelWithStage( - tokenType, - stageMin, - stageMax, - ); - const validFrom = usedCost.validFrom - ? new Date(usedCost.validFrom).toISOString().slice(0, 10) - : null; - const key = `${normalize(usedCost.model)}::${tokenTypeLabel}::${validFrom}`; - if (usedCosts.has(key)) return; - - usedCosts.set(key, { - model: usedCost.model, - tokenType: tokenTypeLabel, - billingUnit: "TOKENS", - price: usedCost.price, - unit: usedCost.unitOfMessure as "1M" | "1K" | null, - currency: usedCost.currency - ? usedCost.currency.trim().toUpperCase() - : null, - validFrom, - }); - }; - - const registerUsedSearchCost = (usedCost: CostStageRow) => { - const validFrom = usedCost.validFrom - ? new Date(usedCost.validFrom).toISOString().slice(0, 10) - : null; - const key = `${normalize(usedCost.model)}::SEARCHES::${validFrom}`; - if (usedCosts.has(key)) return; - usedCosts.set(key, { - model: usedCost.model, - tokenType: "searches", - billingUnit: "SEARCHES", - price: usedCost.price, - unit: usedCost.unitOfMessure as "1M" | "1K" | null, - currency: usedCost.currency - ? usedCost.currency.trim().toUpperCase() - : null, - validFrom, - }); - }; - const costBuckets = new Map< string, ReturnType & { date: Date } @@ -631,25 +535,53 @@ export const reportingRouter = createTRPCRouter({ >(); const reportCurrencyTotals = createCurrencyTotals(); + const cacheConditions = [ + gte(requestStatisticsCache.bucketStart, start.toISOString()), + lt(requestStatisticsCache.bucketStart, end.toISOString()), + isNull(requestStatisticsCacheBuckets.invalidatedAt), + ]; + if (scopedUserIds) { + cacheConditions.push(inArray(apikeys.owner, scopedUserIds)); + } const requestRowsForCost = await ctx.db .select({ + bucketStart: requestStatisticsCache.bucketStart, + apiKeyId: requestStatisticsCache.apiKeyId, userId: users.id, - model: requests.model, - requestTime: requests.requestTime, - inputTokenCount: requests.inputTokenCount, - cachedInputTokenCount: requests.cachedInputTokenCount, - cacheWriteTokenCount: requests.cacheWriteTokenCount, - outputTokenCount: requests.outputTokenCount, - requestType: requests.requestType, - searchUnits: requests.searchUnits, - bucket: costBucket.as("bucket"), - hour: hourBucket.as("hour"), - day: dayBucket.as("day"), + model: sql< + string | null + >`NULLIF(${requestStatisticsCache.model}, '')`.as("model"), + requestTime: requestStatisticsCache.bucketStart, + inputTokenCount: requestStatisticsCache.inputTokenCount, + cachedInputTokenCount: requestStatisticsCache.cachedInputTokenCount, + cacheWriteTokenCount: requestStatisticsCache.cacheWriteTokenCount, + outputTokenCount: requestStatisticsCache.outputTokenCount, + requestType: requestStatisticsCache.requestType, + searchUnits: requestStatisticsCache.searchUnits, + requestCount: requestStatisticsCache.requestCount, + totalCostScaled: requestStatisticsCache.totalCostScaled, + inputCostScaled: requestStatisticsCache.inputCostScaled, + cachedCostScaled: requestStatisticsCache.cachedCostScaled, + cacheWriteCostScaled: requestStatisticsCache.cacheWriteCostScaled, + outputCostScaled: requestStatisticsCache.outputCostScaled, + searchCostScaled: requestStatisticsCache.searchCostScaled, + currency: requestStatisticsCache.currency, + currencyTotals: requestStatisticsCache.currencyTotals, + missingCost: requestStatisticsCache.missingCost, + currencyIssue: requestStatisticsCache.currencyIssue, + usedCosts: requestStatisticsCache.usedCosts, }) - .from(requests) - .innerJoin(apikeys, eq(apikeys.uuid, requests.apiKeyId)) + .from(requestStatisticsCache) + .innerJoin( + requestStatisticsCacheBuckets, + eq( + requestStatisticsCacheBuckets.bucketStart, + requestStatisticsCache.bucketStart, + ), + ) + .innerJoin(apikeys, eq(apikeys.uuid, requestStatisticsCache.apiKeyId)) .innerJoin(users, eq(apikeys.owner, users.id)) - .where(scopedFilter); + .where(and(...cacheConditions)); for (const row of requestRowsForCost) { if (!row.requestTime) continue; @@ -680,7 +612,8 @@ export const reportingRouter = createTRPCRouter({ usage.searchUnits += searchUnits; usageMap.set(usageKey, usage); - const hour = Number(row.hour ?? 0); + const requestDate = new Date(row.requestTime); + const hour = requestDate.getUTCHours(); hourlyTotals.set( hour, (hourlyTotals.get(hour) ?? 0) + @@ -688,7 +621,7 @@ export const reportingRouter = createTRPCRouter({ cachedInputTokenCount + outputTokenCount, ); - const rawDay = Number(row.day ?? 0); + const rawDay = requestDate.getUTCDay(); const dayIndex = rawDay === 0 ? 6 : rawDay - 1; const dayHourKey = `${dayIndex}-${hour}`; dayHourTotals.set( @@ -698,6 +631,8 @@ export const reportingRouter = createTRPCRouter({ } const model = row.model ?? null; + // Keep parity with the uncached report: requests without a model + // contribute to user/token totals, but not to cost/model buckets. if (!model) continue; const inputTokenCount = Number(row.inputTokenCount ?? 0); @@ -705,8 +640,6 @@ export const reportingRouter = createTRPCRouter({ const cacheWriteTokens = Number(row.cacheWriteTokenCount ?? 0); const outputTokenCount = Number(row.outputTokenCount ?? 0); const searchUnits = Number(row.searchUnits ?? 0); - - // Keep output stable: ignore requests that contain no billed tokens. if ( inputTokenCount + cachedInputTokenCount + @@ -717,78 +650,63 @@ export const reportingRouter = createTRPCRouter({ ) continue; - const bucketValue = row.bucket; - if (!bucketValue) continue; - const bucketDate = new Date(bucketValue); + const hourDate = new Date(row.requestTime); + const bucketDate = + input.range.type === "daily" + ? hourDate + : new Date( + Date.UTC( + hourDate.getUTCFullYear(), + hourDate.getUTCMonth(), + hourDate.getUTCDate(), + ), + ); const bucketKey = bucketDate.toISOString(); - - const resolved = - row.requestType === "RERANK" - ? (() => { - const rerank = resolveRerankCost(costStageRows, { - model, - requestTime: new Date(row.requestTime), - searchUnits, - }); - return { - totalCost: rerank.cost, - currency: rerank.currency, - missing: rerank.missing, - inputCost: { cost: 0, usedCost: null }, - cachedCost: { cost: 0, usedCost: null }, - cacheWriteCost: { cost: 0, usedCost: null }, - outputCost: { cost: 0, usedCost: null }, - searchCost: { cost: rerank.cost, usedCost: rerank.usedCost }, - }; - })() - : (() => { - const chat = resolveRequestCostStage(costStageIndex, { - model, - requestTime: new Date(row.requestTime), - inputTokenCount, - cachedInputTokenCount, - cacheWriteTokenCount: cacheWriteTokens, - outputTokenCount, - }); - return { - ...chat, - searchCost: { cost: 0, usedCost: null }, - }; - })(); - const bucket = costBuckets.get(bucketKey) ?? { date: bucketDate, ...createTotalCostAggregate(), }; + mergeCachedTotalCost(bucket, row); + costBuckets.set(bucketKey, bucket); - addResolvedTotalCost(bucket, resolved); - if (!resolved.missing) { - if (resolved.inputCost.usedCost) { - registerUsedCost("input", resolved.inputCost.usedCost); - } - if (resolved.cachedCost.usedCost) { - registerUsedCost("cached", resolved.cachedCost.usedCost); - } - if (resolved.cacheWriteCost.usedCost) { - registerUsedCost("cache_write", resolved.cacheWriteCost.usedCost); - } - if (resolved.outputCost.usedCost) { - registerUsedCost("output", resolved.outputCost.usedCost); - } - if (resolved.searchCost?.usedCost) { - registerUsedSearchCost(resolved.searchCost.usedCost); + if (!row.missingCost) { + for (const usedCost of row.usedCosts ?? []) { + if (!usedCost || typeof usedCost !== "object") continue; + const detail = usedCost as { + model?: string; + tokenType?: string; + billingUnit?: "TOKENS" | "SEARCHES"; + price?: number; + unit?: "1M" | "1K" | null; + currency?: string | null; + validFrom?: string | null; + }; + if ( + typeof detail.model !== "string" || + typeof detail.tokenType !== "string" || + typeof detail.billingUnit !== "string" || + typeof detail.price !== "number" + ) + continue; + const key = `${normalize(detail.model)}::${detail.tokenType}::${detail.validFrom ?? null}`; + usedCosts.set(key, { + model: detail.model, + tokenType: detail.tokenType, + billingUnit: detail.billingUnit, + price: detail.price, + unit: detail.unit ?? null, + currency: detail.currency ?? null, + validFrom: detail.validFrom ?? null, + }); } } - costBuckets.set(bucketKey, bucket); - const userId = row.userId; const userModelKey = `${userId}::${model}::${row.requestType}`; const agg = costAggByUserModel.get(userModelKey) ?? createBreakdownCostAggregate(); - addResolvedBreakdownCost(agg, resolved); - + mergeCachedBreakdownCost(agg, row); costAggByUserModel.set(userModelKey, agg); } @@ -1100,4 +1018,42 @@ export const reportingRouter = createTRPCRouter({ costsUsed, }; }), + rebuildCache: adminProcedure + .input( + z.object({ + from: z + .string() + .trim() + .regex(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}$/), + to: z + .string() + .trim() + .regex(/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}$/), + }), + ) + .mutation(async ({ ctx, input }) => { + const start = new Date(`${input.from}:00Z`); + const end = new Date(`${input.to}:00Z`); + if ( + !Number.isFinite(start.getTime()) || + !Number.isFinite(end.getTime()) + ) { + throw new TRPCError({ + code: "BAD_REQUEST", + message: "Invalid UTC range", + }); + } + if (end <= start) { + throw new TRPCError({ + code: "BAD_REQUEST", + message: "The end of the range must be after its start", + }); + } + + await rebuildRequestStatisticsCache(ctx.db, start, end); + return { + from: start.toISOString(), + to: end.toISOString(), + }; + }), }); diff --git a/app/src/server/costAggregation.js b/app/src/server/costAggregation.js index 21332f5..d3c671d 100644 --- a/app/src/server/costAggregation.js +++ b/app/src/server/costAggregation.js @@ -174,6 +174,33 @@ export function addResolvedTotalCost(aggregate, resolved) { return aggregate; } +/** + * Merge an already-resolved total-cost aggregate into another aggregate. + * This is used by report caches, where individual requests have already + * been resolved with the correct price stage. + * + * @param {TotalCostAggregate} aggregate + * @param {TotalCostAggregate} source + */ +export function addTotalCostAggregate(aggregate, source) { + if (source.missing) { + aggregate.missing = true; + return aggregate; + } + + aggregate.totalCostScaled += source.totalCostScaled; + for (const [currency, total] of source.currencyTotals) { + aggregate.currencyTotals.set( + currency, + (aggregate.currencyTotals.get(currency) ?? 0n) + total, + ); + } + const currencyState = mergeCurrencyState(aggregate.currency, source.currency); + aggregate.currency = currencyState.currency; + aggregate.currencyIssue ||= source.currencyIssue || currencyState.issue; + return aggregate; +} + /** * @param {TotalCostAggregate | undefined} aggregate */ @@ -252,6 +279,35 @@ export function addResolvedBreakdownCost(aggregate, resolved) { return aggregate; } +/** + * Merge an already-resolved breakdown-cost aggregate into another aggregate. + * + * @param {BreakdownCostAggregate} aggregate + * @param {BreakdownCostAggregate} source + */ +export function addBreakdownCostAggregate(aggregate, source) { + if (source.missing) { + aggregate.missing = true; + return aggregate; + } + + aggregate.inputCostScaled += source.inputCostScaled; + aggregate.cachedCostScaled += source.cachedCostScaled; + aggregate.cacheWriteCostScaled += source.cacheWriteCostScaled; + aggregate.outputCostScaled += source.outputCostScaled; + aggregate.searchCostScaled += source.searchCostScaled; + for (const [currency, total] of source.currencyTotals) { + aggregate.currencyTotals.set( + currency, + (aggregate.currencyTotals.get(currency) ?? 0n) + total, + ); + } + const currencyState = mergeCurrencyState(aggregate.currency, source.currency); + aggregate.currency = currencyState.currency; + aggregate.currencyIssue ||= source.currencyIssue || currencyState.issue; + return aggregate; +} + /** * @param {BreakdownCostAggregate | undefined} aggregate */ diff --git a/app/src/server/db/schema.ts b/app/src/server/db/schema.ts index 2e7a264..c3de608 100644 --- a/app/src/server/db/schema.ts +++ b/app/src/server/db/schema.ts @@ -95,6 +95,116 @@ export const requests = pgTable( ], ); +export const requestStatisticsCacheBuckets = pgTable( + "request_statistics_cache_buckets", + { + bucketStart: timestamp("bucket_start", { + withTimezone: true, + mode: "string", + }).primaryKey(), + builtAt: timestamp("built_at", { + withTimezone: true, + mode: "string", + }) + .defaultNow() + .notNull(), + invalidatedAt: timestamp("invalidated_at", { + withTimezone: true, + mode: "string", + }), + }, +); + +export const requestStatisticsCache = pgTable( + "request_statistics_cache", + { + bucketStart: timestamp("bucket_start", { + withTimezone: true, + mode: "string", + }).notNull(), + apiKeyId: varchar("api_key_id", { length: 255 }).notNull(), + // An empty string represents a request whose model is NULL. Keeping the + // key non-null lets the aggregate use a regular composite primary key. + model: varchar({ length: 255 }).notNull().default(""), + requestType: requestType("request_type").notNull(), + requestCount: bigint("request_count", { mode: "bigint" }) + .notNull() + .default(0n), + inputTokenCount: bigint("input_token_count", { mode: "bigint" }) + .notNull() + .default(0n), + cachedInputTokenCount: bigint("cached_input_token_count", { + mode: "bigint", + }) + .notNull() + .default(0n), + cacheWriteTokenCount: bigint("cache_write_token_count", { mode: "bigint" }) + .notNull() + .default(0n), + outputTokenCount: bigint("output_token_count", { mode: "bigint" }) + .notNull() + .default(0n), + searchUnits: bigint("search_units", { mode: "bigint" }) + .notNull() + .default(0n), + totalCostScaled: bigint("total_cost_scaled", { mode: "bigint" }) + .notNull() + .default(0n), + inputCostScaled: bigint("input_cost_scaled", { mode: "bigint" }) + .notNull() + .default(0n), + cachedCostScaled: bigint("cached_cost_scaled", { mode: "bigint" }) + .notNull() + .default(0n), + cacheWriteCostScaled: bigint("cache_write_cost_scaled", { mode: "bigint" }) + .notNull() + .default(0n), + outputCostScaled: bigint("output_cost_scaled", { mode: "bigint" }) + .notNull() + .default(0n), + searchCostScaled: bigint("search_cost_scaled", { mode: "bigint" }) + .notNull() + .default(0n), + currency: char({ length: 3 }), + currencyTotals: jsonb("currency_totals") + .$type>() + .notNull() + .default(sql`'{}'::jsonb`), + missingCost: boolean("missing_cost").notNull().default(false), + currencyIssue: boolean("currency_issue").notNull().default(false), + usedCosts: jsonb("used_costs") + .$type() + .notNull() + .default(sql`'[]'::jsonb`), + }, + (table) => [ + primaryKey({ + columns: [ + table.bucketStart, + table.apiKeyId, + table.model, + table.requestType, + ], + name: "request_statistics_cache_pkey", + }), + foreignKey({ + columns: [table.bucketStart], + foreignColumns: [requestStatisticsCacheBuckets.bucketStart], + name: "request_statistics_cache_bucket_fkey", + }), + foreignKey({ + columns: [table.apiKeyId], + foreignColumns: [apikeys.uuid], + name: "request_statistics_cache_api_key_fkey", + }), + index("request_statistics_cache_bucket_idx").on(table.bucketStart), + index("request_statistics_cache_api_key_bucket_idx").on( + table.apiKeyId, + table.bucketStart, + ), + ], +); + export const apikeys = pgTable( "apikeys", { diff --git a/app/src/server/requestStatisticsCache.ts b/app/src/server/requestStatisticsCache.ts new file mode 100644 index 0000000..69a70f1 --- /dev/null +++ b/app/src/server/requestStatisticsCache.ts @@ -0,0 +1,603 @@ +import { and, gte, inArray, lt, sql } from "drizzle-orm"; + +import { + addBreakdownCostAggregate, + addResolvedBreakdownCost, + addResolvedTotalCost, + addTotalCostAggregate, + createBreakdownCostAggregate, + createTotalCostAggregate, +} from "~/server/costAggregation"; +import { + buildCostStageIndex, + type CostStageRow, + resolveRequestCostStage, + resolveRerankCost, +} from "~/server/costStageResolver"; +import type { db } from "~/server/db"; +import { + costs, + requestStatisticsCache, + requestStatisticsCacheBuckets, + requests, +} from "~/server/db/schema"; + +const HOUR_MS = 60 * 60 * 1000; + +type AppDatabase = typeof db; +type RequestType = (typeof requests.requestType.enumValues)[number]; + +export type CachedCostDetail = { + model: string; + tokenType: string; + billingUnit: "TOKENS" | "SEARCHES"; + price: number; + unit: "1M" | "1K" | null; + currency: string | null; + validFrom: string | null; +}; + +export type RequestStatisticsCacheRow = { + bucketStart: string; + apiKeyId: string; + model: string | null; + requestType: RequestType; + requestCount: bigint; + inputTokenCount: bigint; + cachedInputTokenCount: bigint; + cacheWriteTokenCount: bigint; + outputTokenCount: bigint; + searchUnits: bigint; + totalCostScaled: bigint; + inputCostScaled: bigint; + cachedCostScaled: bigint; + cacheWriteCostScaled: bigint; + outputCostScaled: bigint; + searchCostScaled: bigint; + currency: string | null; + currencyTotals: Record; + missingCost: boolean; + currencyIssue: boolean; + usedCosts: unknown[]; +}; + +type CostAggregate = ReturnType; +type BreakdownAggregate = ReturnType; + +type BucketAggregate = { + apiKeyId: string; + model: string; + requestType: RequestType; + requestCount: bigint; + inputTokenCount: bigint; + cachedInputTokenCount: bigint; + cacheWriteTokenCount: bigint; + outputTokenCount: bigint; + searchUnits: bigint; + total: CostAggregate; + breakdown: BreakdownAggregate; + usedCosts: Map; +}; + +function floorUtcHour(value: Date) { + return new Date( + Date.UTC( + value.getUTCFullYear(), + value.getUTCMonth(), + value.getUTCDate(), + value.getUTCHours(), + 0, + 0, + 0, + ), + ); +} + +function listBuckets(start: Date, end: Date) { + const buckets: Date[] = []; + for ( + let cursor = floorUtcHour(start); + cursor < end; + cursor = new Date(cursor.getTime() + HOUR_MS) + ) { + buckets.push(new Date(cursor)); + } + return buckets; +} + +function mergeRanges(buckets: Date[]) { + const ranges: Array<{ start: Date; end: Date }> = []; + for (const bucket of buckets) { + const previous = ranges[ranges.length - 1]; + if (previous && previous.end.getTime() === bucket.getTime()) { + previous.end = new Date(bucket.getTime() + HOUR_MS); + } else { + ranges.push({ + start: new Date(bucket), + end: new Date(bucket.getTime() + HOUR_MS), + }); + } + } + return ranges; +} + +function cacheKey( + apiKeyId: string, + model: string | null, + requestType: RequestType, +) { + return `${apiKeyId}\u0000${model ?? ""}\u0000${requestType}`; +} + +function emptyAggregate( + apiKeyId: string, + model: string | null, + requestType: RequestType, +): BucketAggregate { + return { + apiKeyId, + model: model ?? "", + requestType, + requestCount: 0n, + inputTokenCount: 0n, + cachedInputTokenCount: 0n, + cacheWriteTokenCount: 0n, + outputTokenCount: 0n, + searchUnits: 0n, + total: createTotalCostAggregate(), + breakdown: createBreakdownCostAggregate(), + usedCosts: new Map(), + }; +} + +function costDetail( + usedCost: CostStageRow, + tokenType: string, + stageLabel?: string, +): CachedCostDetail { + const validFrom = usedCost.validFrom + ? new Date(usedCost.validFrom).toISOString().slice(0, 10) + : null; + return { + model: usedCost.model, + tokenType: stageLabel ?? tokenType, + billingUnit: usedCost.billingUnit === "SEARCHES" ? "SEARCHES" : "TOKENS", + price: usedCost.price, + unit: (usedCost.unitOfMessure ?? null) as "1M" | "1K" | null, + currency: usedCost.currency ? usedCost.currency.trim().toUpperCase() : null, + validFrom, + }; +} + +function addUsedCost( + usedCosts: Map, + detail: CachedCostDetail, +) { + const key = [ + detail.model.toLowerCase(), + detail.tokenType, + detail.billingUnit, + detail.price, + detail.unit ?? "", + detail.currency ?? "", + detail.validFrom ?? "", + ].join("\u0000"); + usedCosts.set(key, detail); +} + +function addResolvedCosts( + aggregate: BucketAggregate, + resolved: + | ReturnType + | ReturnType, +) { + if ("inputCost" in resolved) { + addResolvedTotalCost(aggregate.total, resolved); + if (resolved.missing) return; + addResolvedBreakdownCost(aggregate.breakdown, resolved); + if (resolved.inputCost.usedCost) { + addUsedCost( + aggregate.usedCosts, + costDetail( + resolved.inputCost.usedCost, + "input", + `input (${resolved.inputCost.usedCost.stageMinTokens}..${resolved.inputCost.usedCost.stageMaxTokens === null ? "∞" : resolved.inputCost.usedCost.stageMaxTokens})`, + ), + ); + } + if (resolved.cachedCost.usedCost) { + addUsedCost( + aggregate.usedCosts, + costDetail( + resolved.cachedCost.usedCost, + "cached", + `cached (${resolved.cachedCost.usedCost.stageMinTokens}..${resolved.cachedCost.usedCost.stageMaxTokens === null ? "∞" : resolved.cachedCost.usedCost.stageMaxTokens})`, + ), + ); + } + if (resolved.cacheWriteCost.usedCost) { + addUsedCost( + aggregate.usedCosts, + costDetail( + resolved.cacheWriteCost.usedCost, + "cache_write", + `cache_write (${resolved.cacheWriteCost.usedCost.stageMinTokens}..${resolved.cacheWriteCost.usedCost.stageMaxTokens === null ? "∞" : resolved.cacheWriteCost.usedCost.stageMaxTokens})`, + ), + ); + } + if (resolved.outputCost.usedCost) { + addUsedCost( + aggregate.usedCosts, + costDetail( + resolved.outputCost.usedCost, + "output", + `output (${resolved.outputCost.usedCost.stageMinTokens}..${resolved.outputCost.usedCost.stageMaxTokens === null ? "∞" : resolved.outputCost.usedCost.stageMaxTokens})`, + ), + ); + } + return; + } + + addResolvedTotalCost(aggregate.total, { + totalCost: resolved.cost, + currency: resolved.currency, + missing: resolved.missing, + }); + if (resolved.missing) return; + + if (resolved.usedCost) { + addResolvedBreakdownCost(aggregate.breakdown, { + missing: false, + currency: resolved.currency, + inputCost: { cost: 0 }, + cachedCost: { cost: 0 }, + cacheWriteCost: { cost: 0 }, + outputCost: { cost: 0 }, + searchCost: { cost: resolved.cost }, + }); + addUsedCost(aggregate.usedCosts, costDetail(resolved.usedCost, "searches")); + } +} + +function serializeCurrencyTotals(currencyTotals: Map) { + return Object.fromEntries( + Array.from(currencyTotals.entries()).map(([currency, value]) => [ + currency, + value.toString(), + ]), + ); +} + +function deserializeCurrencyTotals(value: unknown) { + const result = new Map(); + if (!value || typeof value !== "object" || Array.isArray(value)) + return result; + for (const [currency, amount] of Object.entries(value)) { + try { + result.set(currency, BigInt(String(amount))); + } catch { + // Ignore malformed cache data; the bucket will be rebuilt if needed. + } + } + return result; +} + +function toCostAggregate(row: RequestStatisticsCacheRow): CostAggregate { + return { + missing: row.missingCost, + currency: row.currency, + totalCostScaled: row.totalCostScaled, + currencyTotals: deserializeCurrencyTotals(row.currencyTotals), + currencyIssue: row.currencyIssue, + }; +} + +function toBreakdownAggregate( + row: RequestStatisticsCacheRow, +): BreakdownAggregate { + return { + missing: row.missingCost, + currency: row.currency, + inputCostScaled: row.inputCostScaled, + cachedCostScaled: row.cachedCostScaled, + cacheWriteCostScaled: row.cacheWriteCostScaled, + outputCostScaled: row.outputCostScaled, + searchCostScaled: row.searchCostScaled, + currencyTotals: deserializeCurrencyTotals(row.currencyTotals), + currencyIssue: row.currencyIssue, + }; +} + +async function loadCostRows(database: AppDatabase) { + const rows = await database + .select({ + model: costs.model, + price: costs.price, + validFrom: costs.validFrom, + requestType: costs.requestType, + billingUnit: costs.billingUnit, + tokenType: costs.tokenType, + unitOfMessure: costs.unitOfMessure, + currency: costs.currency, + stageType: costs.stageType, + stageMinTokens: costs.stageMinTokens, + stageMaxTokens: costs.stageMaxTokens, + }) + .from(costs); + + return rows.map((row) => ({ + model: row.model, + price: Number(row.price ?? 0), + validFrom: row.validFrom ?? new Date(0), + requestType: row.requestType, + billingUnit: row.billingUnit, + tokenType: row.tokenType ?? "", + unitOfMessure: (row.unitOfMessure ?? null) as CostStageRow["unitOfMessure"], + currency: row.currency ? row.currency.trim().toUpperCase() : null, + stageType: row.stageType ?? "context_length", + stageMinTokens: Number(row.stageMinTokens ?? 0), + stageMaxTokens: row.stageMaxTokens ?? null, + })) satisfies CostStageRow[]; +} + +function buildBucketValues( + bucketStart: Date, + rows: Array<{ + apiKeyId: string; + model: string | null; + requestType: RequestType; + requestTime: string | null; + inputTokenCount: number | null; + cachedInputTokenCount: number | null; + cacheWriteTokenCount: number | null; + outputTokenCount: number | null; + searchUnits: number | null; + }>, + costStageRows: CostStageRow[], + costStageIndex: Map, +) { + const aggregates = new Map(); + + for (const row of rows) { + const aggregateKey = cacheKey(row.apiKeyId, row.model, row.requestType); + const aggregate = + aggregates.get(aggregateKey) ?? + emptyAggregate(row.apiKeyId, row.model, row.requestType); + aggregate.requestCount += 1n; + aggregate.inputTokenCount += BigInt(row.inputTokenCount ?? 0); + aggregate.cachedInputTokenCount += BigInt(row.cachedInputTokenCount ?? 0); + aggregate.cacheWriteTokenCount += BigInt(row.cacheWriteTokenCount ?? 0); + aggregate.outputTokenCount += BigInt(row.outputTokenCount ?? 0); + aggregate.searchUnits += BigInt(row.searchUnits ?? 0); + + const inputTokenCount = Number(row.inputTokenCount ?? 0); + const cachedInputTokenCount = Number(row.cachedInputTokenCount ?? 0); + const cacheWriteTokenCount = Number(row.cacheWriteTokenCount ?? 0); + const outputTokenCount = Number(row.outputTokenCount ?? 0); + const searchUnits = Number(row.searchUnits ?? 0); + if ( + row.model && + inputTokenCount + + cachedInputTokenCount + + cacheWriteTokenCount + + outputTokenCount + + searchUnits > + 0 + ) { + const requestTime = new Date(row.requestTime ?? bucketStart); + const resolved = + row.requestType === "RERANK" + ? resolveRerankCost(costStageRows, { + model: row.model, + requestTime, + searchUnits, + }) + : resolveRequestCostStage(costStageIndex, { + model: row.model, + requestTime, + inputTokenCount, + cachedInputTokenCount, + cacheWriteTokenCount, + outputTokenCount, + }); + addResolvedCosts(aggregate, resolved); + } + aggregates.set(aggregateKey, aggregate); + } + + const values = Array.from(aggregates.values()).map((aggregate) => ({ + bucketStart: bucketStart.toISOString(), + apiKeyId: aggregate.apiKeyId, + model: aggregate.model, + requestType: aggregate.requestType, + requestCount: aggregate.requestCount, + inputTokenCount: aggregate.inputTokenCount, + cachedInputTokenCount: aggregate.cachedInputTokenCount, + cacheWriteTokenCount: aggregate.cacheWriteTokenCount, + outputTokenCount: aggregate.outputTokenCount, + searchUnits: aggregate.searchUnits, + totalCostScaled: aggregate.total.totalCostScaled, + inputCostScaled: aggregate.breakdown.inputCostScaled, + cachedCostScaled: aggregate.breakdown.cachedCostScaled, + cacheWriteCostScaled: aggregate.breakdown.cacheWriteCostScaled, + outputCostScaled: aggregate.breakdown.outputCostScaled, + searchCostScaled: aggregate.breakdown.searchCostScaled, + currency: aggregate.total.currency, + currencyTotals: serializeCurrencyTotals(aggregate.total.currencyTotals), + missingCost: aggregate.total.missing, + currencyIssue: aggregate.total.currencyIssue, + usedCosts: Array.from(aggregate.usedCosts.values()), + })); + return { bucketStart: bucketStart.toISOString(), values }; +} + +type PreparedBucket = ReturnType; + +function chunks(values: T[], size: number) { + const result: T[][] = []; + for (let index = 0; index < values.length; index += size) { + result.push(values.slice(index, index + size)); + } + return result; +} + +async function persistBuckets( + database: AppDatabase, + buckets: PreparedBucket[], +) { + if (!buckets.length) return; + await database.transaction(async (tx) => { + const bucketStarts = buckets.map((bucket) => bucket.bucketStart); + // The aggregate rows reference the bucket marker, so create the markers + // before inserting the rows. The whole operation remains atomic. + for (const markerChunk of chunks( + bucketStarts.map((bucketStart) => ({ + bucketStart, + invalidatedAt: null, + })), + 250, + )) { + await tx + .insert(requestStatisticsCacheBuckets) + .values(markerChunk) + .onConflictDoUpdate({ + target: requestStatisticsCacheBuckets.bucketStart, + set: { builtAt: sql`now()`, invalidatedAt: null }, + }); + } + + for (const bucketStartChunk of chunks(bucketStarts, 250)) { + await tx + .delete(requestStatisticsCache) + .where(inArray(requestStatisticsCache.bucketStart, bucketStartChunk)); + } + + const values = buckets.flatMap((bucket) => bucket.values); + for (const valueChunk of chunks(values, 250)) { + await tx.insert(requestStatisticsCache).values(valueChunk); + } + }); +} + +async function populate( + database: AppDatabase, + start: Date, + end: Date, + force: boolean, +) { + const buckets = listBuckets(start, end); + if (!buckets.length) return; + + const markers = await database + .select({ + bucketStart: requestStatisticsCacheBuckets.bucketStart, + invalidatedAt: requestStatisticsCacheBuckets.invalidatedAt, + }) + .from(requestStatisticsCacheBuckets) + .where( + and( + gte( + requestStatisticsCacheBuckets.bucketStart, + buckets[0]?.toISOString() ?? start.toISOString(), + ), + lt( + requestStatisticsCacheBuckets.bucketStart, + new Date( + (buckets[buckets.length - 1]?.getTime() ?? end.getTime()) + HOUR_MS, + ).toISOString(), + ), + ), + ); + const validBuckets = new Set( + markers + .filter((marker) => marker.invalidatedAt === null) + .map((marker) => new Date(marker.bucketStart).getTime()), + ); + const missingBuckets = force + ? buckets + : buckets.filter((bucket) => !validBuckets.has(bucket.getTime())); + if (!missingBuckets.length) return; + + const costStageRows = await loadCostRows(database); + const costStageIndex = buildCostStageIndex(costStageRows); + const preparedBuckets: PreparedBucket[] = []; + for (const range of mergeRanges(missingBuckets)) { + const rows = await database + .select({ + apiKeyId: requests.apiKeyId, + model: requests.model, + requestType: requests.requestType, + requestTime: requests.requestTime, + inputTokenCount: requests.inputTokenCount, + cachedInputTokenCount: requests.cachedInputTokenCount, + cacheWriteTokenCount: requests.cacheWriteTokenCount, + outputTokenCount: requests.outputTokenCount, + searchUnits: requests.searchUnits, + }) + .from(requests) + .where( + and( + gte(requests.requestTime, range.start.toISOString()), + lt(requests.requestTime, range.end.toISOString()), + ), + ); + const rowsByBucket = new Map(); + for (const row of rows) { + if (!row.requestTime) continue; + const key = floorUtcHour(new Date(row.requestTime)).toISOString(); + const bucketRows = rowsByBucket.get(key) ?? []; + bucketRows.push(row); + rowsByBucket.set(key, bucketRows); + } + for (const bucket of missingBuckets.filter( + (candidate) => candidate >= range.start && candidate < range.end, + )) { + preparedBuckets.push( + buildBucketValues( + bucket, + rowsByBucket.get(bucket.toISOString()) ?? [], + costStageRows, + costStageIndex, + ), + ); + } + } + await persistBuckets(database, preparedBuckets); +} + +export async function ensureRequestStatisticsCache( + database: AppDatabase, + start: Date, + end: Date, +) { + await populate(database, start, end, false); +} + +export async function rebuildRequestStatisticsCache( + database: AppDatabase, + start: Date, + end: Date, +) { + await populate(database, start, end, true); +} + +export function cachedRowToAggregates(row: RequestStatisticsCacheRow) { + return { + total: toCostAggregate(row), + breakdown: toBreakdownAggregate(row), + }; +} + +export function mergeCachedTotalCost( + target: CostAggregate, + row: RequestStatisticsCacheRow, +) { + addTotalCostAggregate(target, toCostAggregate(row)); +} + +export function mergeCachedBreakdownCost( + target: BreakdownAggregate, + row: RequestStatisticsCacheRow, +) { + addBreakdownCostAggregate(target, toBreakdownAggregate(row)); +} diff --git a/db/migrations/20260826130000_request_statistics_cache.sql b/db/migrations/20260826130000_request_statistics_cache.sql new file mode 100644 index 0000000..192b429 --- /dev/null +++ b/db/migrations/20260826130000_request_statistics_cache.sql @@ -0,0 +1,106 @@ +CREATE TABLE IF NOT EXISTS request_statistics_cache_buckets ( + bucket_start timestamptz PRIMARY KEY, + built_at timestamptz NOT NULL DEFAULT now(), + invalidated_at timestamptz +); + +CREATE TABLE IF NOT EXISTS request_statistics_cache ( + bucket_start timestamptz NOT NULL REFERENCES request_statistics_cache_buckets(bucket_start) ON DELETE CASCADE, + api_key_id varchar(255) NOT NULL REFERENCES apiKeys(UUID) ON DELETE CASCADE, + model varchar(255) NOT NULL DEFAULT '', + request_type request_type NOT NULL, + request_count bigint NOT NULL DEFAULT 0, + input_token_count bigint NOT NULL DEFAULT 0, + cached_input_token_count bigint NOT NULL DEFAULT 0, + cache_write_token_count bigint NOT NULL DEFAULT 0, + output_token_count bigint NOT NULL DEFAULT 0, + search_units bigint NOT NULL DEFAULT 0, + total_cost_scaled bigint NOT NULL DEFAULT 0, + input_cost_scaled bigint NOT NULL DEFAULT 0, + cached_cost_scaled bigint NOT NULL DEFAULT 0, + cache_write_cost_scaled bigint NOT NULL DEFAULT 0, + output_cost_scaled bigint NOT NULL DEFAULT 0, + search_cost_scaled bigint NOT NULL DEFAULT 0, + currency char(3), + currency_totals jsonb NOT NULL DEFAULT '{}'::jsonb, + missing_cost boolean NOT NULL DEFAULT false, + currency_issue boolean NOT NULL DEFAULT false, + used_costs jsonb NOT NULL DEFAULT '[]'::jsonb, + CONSTRAINT request_statistics_cache_pkey PRIMARY KEY (bucket_start, api_key_id, model, request_type) +); + +CREATE INDEX IF NOT EXISTS request_statistics_cache_bucket_idx + ON request_statistics_cache (bucket_start); + +CREATE INDEX IF NOT EXISTS request_statistics_cache_api_key_bucket_idx + ON request_statistics_cache (api_key_id, bucket_start); + +CREATE OR REPLACE FUNCTION invalidate_request_statistics_cache_for_request() +RETURNS trigger +LANGUAGE plpgsql +AS $$ +BEGIN + IF TG_OP <> 'INSERT' AND OLD.request_time IS NOT NULL THEN + UPDATE request_statistics_cache_buckets + SET invalidated_at = now() + WHERE bucket_start = date_trunc('hour', OLD.request_time, 'UTC'); + END IF; + + IF TG_OP <> 'DELETE' AND NEW.request_time IS NOT NULL THEN + UPDATE request_statistics_cache_buckets + SET invalidated_at = now() + WHERE bucket_start = date_trunc('hour', NEW.request_time, 'UTC'); + END IF; + + RETURN COALESCE(NEW, OLD); +END; +$$; + +DROP TRIGGER IF EXISTS requests_invalidate_statistics_cache + ON requests; +CREATE TRIGGER requests_invalidate_statistics_cache +AFTER INSERT OR UPDATE OR DELETE ON requests +FOR EACH ROW +EXECUTE FUNCTION invalidate_request_statistics_cache_for_request(); + +CREATE OR REPLACE FUNCTION invalidate_request_statistics_cache_for_cost() +RETURNS trigger +LANGUAGE plpgsql +AS $$ +BEGIN + IF TG_OP <> 'INSERT' THEN + UPDATE request_statistics_cache_buckets + SET invalidated_at = now() + WHERE bucket_start >= (OLD.valid_from::timestamp AT TIME ZONE 'UTC') + AND EXISTS ( + SELECT 1 + FROM request_statistics_cache cached + WHERE cached.bucket_start = request_statistics_cache_buckets.bucket_start + AND lower(cached.model) = lower(OLD.model) + AND cached.request_type = OLD.request_type + ); + END IF; + + IF TG_OP <> 'DELETE' THEN + UPDATE request_statistics_cache_buckets + SET invalidated_at = now() + WHERE bucket_start >= (NEW.valid_from::timestamp AT TIME ZONE 'UTC') + AND EXISTS ( + SELECT 1 + FROM request_statistics_cache cached + WHERE cached.bucket_start = request_statistics_cache_buckets.bucket_start + AND lower(cached.model) = lower(NEW.model) + AND cached.request_type = NEW.request_type + ); + END IF; + + RETURN COALESCE(NEW, OLD); +END; +$$; + +DROP TRIGGER IF EXISTS costs_invalidate_statistics_cache + ON costs; +CREATE TRIGGER costs_invalidate_statistics_cache +AFTER INSERT OR UPDATE OR DELETE ON costs +FOR EACH ROW +EXECUTE FUNCTION invalidate_request_statistics_cache_for_cost(); diff --git a/db/migrations/atlas.sum b/db/migrations/atlas.sum index 681938a..6c09ab1 100644 --- a/db/migrations/atlas.sum +++ b/db/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:AZtd/W6Q1UHgZmpWt3qfPAu1GerPYyTC8uvKPKkug38= +h1:M+IJgytlEV67h1svjx6prA48NmxiXduDienWfaST2fI= 20240924151008_init.sql h1:rTjsfTqruGXjIUITaR6Pqt0n4lm+4EcdvESuYXgmyP4= 20240924154252_request_monitoring.sql h1:VWH6kcc2DWS7L9tBQ8lgS858/9f5HP7PMYZo0fZlCX8= 20241001124048_id_datatype.sql h1:Nb1OpAzcVshPzgEqQ3w/sCFjorO4h/GTOBeuISanu4k= @@ -27,3 +27,4 @@ h1:AZtd/W6Q1UHgZmpWt3qfPAu1GerPYyTC8uvKPKkug38= 20260818110000_add_gpt56_cache_write_accounting.sql h1:OpcV68YEKID7tO0vFeZU8loBRT3ZaqUHSJqgqiEoF/o= 20260825120000_reporting_request_indexes.sql h1:fHCPTleO4cX5OAcwoJd4n5UJD8vWty/r4sNlItuXyVw= 20260826100000_azure_retail_pricing.sql h1:Xd8OtUZg4VZlsYlSjLmYyxOwJVNHPXgK/Mdxft5lURc= +20260826130000_request_statistics_cache.sql h1:GKodt4zvUbq+KMFdKM6pPPI6v5Q37d9nFjP9j1CLcmc= diff --git a/db/schema.sql b/db/schema.sql index fde20c5..ad7fda1 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -121,3 +121,100 @@ CREATE TABLE IF NOT EXISTS requests ( ) ) ); + +CREATE TABLE IF NOT EXISTS request_statistics_cache_buckets ( + bucket_start timestamptz PRIMARY KEY, + built_at timestamptz NOT NULL DEFAULT now(), + invalidated_at timestamptz +); + +CREATE TABLE IF NOT EXISTS request_statistics_cache ( + bucket_start timestamptz NOT NULL REFERENCES request_statistics_cache_buckets(bucket_start) ON DELETE CASCADE, + api_key_id varchar(255) NOT NULL REFERENCES apiKeys(UUID) ON DELETE CASCADE, + model varchar(255) NOT NULL DEFAULT '', + request_type request_type NOT NULL, + request_count bigint NOT NULL DEFAULT 0, + input_token_count bigint NOT NULL DEFAULT 0, + cached_input_token_count bigint NOT NULL DEFAULT 0, + cache_write_token_count bigint NOT NULL DEFAULT 0, + output_token_count bigint NOT NULL DEFAULT 0, + search_units bigint NOT NULL DEFAULT 0, + total_cost_scaled bigint NOT NULL DEFAULT 0, + input_cost_scaled bigint NOT NULL DEFAULT 0, + cached_cost_scaled bigint NOT NULL DEFAULT 0, + cache_write_cost_scaled bigint NOT NULL DEFAULT 0, + output_cost_scaled bigint NOT NULL DEFAULT 0, + search_cost_scaled bigint NOT NULL DEFAULT 0, + currency char(3), + currency_totals jsonb NOT NULL DEFAULT '{}'::jsonb, + missing_cost boolean NOT NULL DEFAULT false, + currency_issue boolean NOT NULL DEFAULT false, + used_costs jsonb NOT NULL DEFAULT '[]'::jsonb, + CONSTRAINT request_statistics_cache_pkey PRIMARY KEY (bucket_start, api_key_id, model, request_type) +); + +CREATE INDEX IF NOT EXISTS request_statistics_cache_bucket_idx + ON request_statistics_cache (bucket_start); + +CREATE INDEX IF NOT EXISTS request_statistics_cache_api_key_bucket_idx + ON request_statistics_cache (api_key_id, bucket_start); + +CREATE OR REPLACE FUNCTION invalidate_request_statistics_cache_for_request() +RETURNS trigger +LANGUAGE plpgsql +AS $$ +BEGIN + IF TG_OP <> 'INSERT' AND OLD.request_time IS NOT NULL THEN + UPDATE request_statistics_cache_buckets + SET invalidated_at = now() + WHERE bucket_start = date_trunc('hour', OLD.request_time, 'UTC'); + END IF; + IF TG_OP <> 'DELETE' AND NEW.request_time IS NOT NULL THEN + UPDATE request_statistics_cache_buckets + SET invalidated_at = now() + WHERE bucket_start = date_trunc('hour', NEW.request_time, 'UTC'); + END IF; + RETURN COALESCE(NEW, OLD); +END; +$$; + +DROP TRIGGER IF EXISTS requests_invalidate_statistics_cache ON requests; +CREATE TRIGGER requests_invalidate_statistics_cache +AFTER INSERT OR UPDATE OR DELETE ON requests +FOR EACH ROW EXECUTE FUNCTION invalidate_request_statistics_cache_for_request(); + +CREATE OR REPLACE FUNCTION invalidate_request_statistics_cache_for_cost() +RETURNS trigger +LANGUAGE plpgsql +AS $$ +BEGIN + IF TG_OP <> 'INSERT' THEN + UPDATE request_statistics_cache_buckets + SET invalidated_at = now() + WHERE bucket_start >= (OLD.valid_from::timestamp AT TIME ZONE 'UTC') + AND EXISTS ( + SELECT 1 FROM request_statistics_cache cached + WHERE cached.bucket_start = request_statistics_cache_buckets.bucket_start + AND lower(cached.model) = lower(OLD.model) + AND cached.request_type = OLD.request_type + ); + END IF; + IF TG_OP <> 'DELETE' THEN + UPDATE request_statistics_cache_buckets + SET invalidated_at = now() + WHERE bucket_start >= (NEW.valid_from::timestamp AT TIME ZONE 'UTC') + AND EXISTS ( + SELECT 1 FROM request_statistics_cache cached + WHERE cached.bucket_start = request_statistics_cache_buckets.bucket_start + AND lower(cached.model) = lower(NEW.model) + AND cached.request_type = NEW.request_type + ); + END IF; + RETURN COALESCE(NEW, OLD); +END; +$$; + +DROP TRIGGER IF EXISTS costs_invalidate_statistics_cache ON costs; +CREATE TRIGGER costs_invalidate_statistics_cache +AFTER INSERT OR UPDATE OR DELETE ON costs +FOR EACH ROW EXECUTE FUNCTION invalidate_request_statistics_cache_for_cost();