From 6d7292bacad1ed3fd756527ce135f9f311aa404e Mon Sep 17 00:00:00 2001 From: Dean Eckert Date: Tue, 25 Aug 2026 15:47:50 +0200 Subject: [PATCH] feat(reporting): added better indexing and made the report creation more efficient Signed-off-by: Dean Eckert --- app/src/server/api/routers/reporting.ts | 268 +++++++++--------- app/src/server/db/schema.ts | 7 + ...260825120000_reporting_request_indexes.sql | 8 + db/migrations/atlas.sum | 3 +- local-dev/docker-compose.yml | 12 +- 5 files changed, 155 insertions(+), 143 deletions(-) create mode 100644 db/migrations/20260825120000_reporting_request_indexes.sql diff --git a/app/src/server/api/routers/reporting.ts b/app/src/server/api/routers/reporting.ts index 5c0c84e..fb4fd5e 100644 --- a/app/src/server/api/routers/reporting.ts +++ b/app/src/server/api/routers/reporting.ts @@ -436,45 +436,6 @@ export const reportingRouter = createTRPCRouter({ } } - const usersBase = ctx.db - .select({ - userId: users.id, - name: users.name, - model: requests.model, - requestType: requests.requestType, - inputTokens: - sql`greatest(coalesce(sum(${requests.inputTokenCount} - ${requests.cachedInputTokenCount} - ${requests.cacheWriteTokenCount}), 0), 0)`.as( - "inputTokens", - ), - cachedInputTokens: - sql`coalesce(sum(${requests.cachedInputTokenCount}), 0)`.as( - "cachedInputTokens", - ), - cacheWriteTokens: - sql`coalesce(sum(${requests.cacheWriteTokenCount}), 0)`.as( - "cacheWriteTokens", - ), - outputTokens: - sql`coalesce(sum(${requests.outputTokenCount}), 0)`.as( - "outputTokens", - ), - searchUnits: - sql`coalesce(sum(${requests.searchUnits}), 0)`.as( - "searchUnits", - ), - }) - .from(users) - .leftJoin(apikeys, eq(apikeys.owner, users.id)) - .leftJoin( - requests, - and(eq(requests.apiKeyId, apikeys.uuid), timeFilter), - ); - - const usageRows = await (scopedUserIds - ? usersBase.where(inArray(users.id, scopedUserIds)) - : usersBase - ).groupBy(users.id, users.name, requests.model, requests.requestType); - const scopedConditions = [timeFilter]; if (scopedUserIds) { scopedConditions.push(inArray(apikeys.owner, scopedUserIds)); @@ -485,6 +446,83 @@ export const reportingRouter = createTRPCRouter({ 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})`; + + type UserModel = { + model: string; + requestType: string; + inputTokens: number; + cachedInputTokens: number; + cacheWriteTokens: number; + outputTokens: number; + searchUnits: number; + inputCost: number | null; + cachedCost: number | null; + cacheWriteCost: number | null; + outputCost: number | null; + searchCost: number | null; + totalCost: number | null; + currency: string | null; + currencyIssue: boolean; + currencyTotals: Array<{ currency: string; totalCost: number }>; + }; + type UserEntry = { + id: string; + name: string; + inputTokens: number; + cachedInputTokens: number; + cacheWriteTokens: number; + outputTokens: number; + searchUnits: number; + totalCostScaled: bigint; + totalCostHasKnownCost: boolean; + totalCostHasMissingCost: boolean; + currencyIssue: boolean; + currency: string | null; + currencyTotals: Map; + models: UserModel[]; + }; + const usersMap = new Map(); + const usageMap = new Map< + string, + { + userId: string; + name: string; + model: string | null; + requestType: (typeof requests.requestType.enumValues)[number] | null; + inputTokens: number; + cachedInputTokens: number; + cacheWriteTokens: number; + outputTokens: number; + searchUnits: number; + } + >(); + const emptyUser = (id: string, name: string): UserEntry => ({ + id, + name, + inputTokens: 0, + cachedInputTokens: 0, + cacheWriteTokens: 0, + outputTokens: 0, + searchUnits: 0, + totalCostScaled: 0n, + totalCostHasKnownCost: false, + totalCostHasMissingCost: false, + currencyIssue: false, + currency: null, + currencyTotals: createCurrencyTotals(), + models: [], + }); + const reportUsers = await ctx.db + .select({ id: users.id, name: users.name }) + .from(users) + .where(scopedUserIds ? inArray(users.id, scopedUserIds) : undefined); + for (const user of reportUsers) { + usersMap.set(user.id, emptyUser(user.id, user.name ?? user.id)); + } + 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 @@ -605,17 +643,62 @@ export const reportingRouter = createTRPCRouter({ requestType: requests.requestType, searchUnits: requests.searchUnits, bucket: costBucket.as("bucket"), + hour: hourBucket.as("hour"), + day: dayBucket.as("day"), }) .from(requests) .innerJoin(apikeys, eq(apikeys.uuid, requests.apiKeyId)) .innerJoin(users, eq(apikeys.owner, users.id)) - .where(scopedFilter) - .orderBy(requests.requestTime); + .where(scopedFilter); for (const row of requestRowsForCost) { + if (!row.requestTime) continue; + const user = usersMap.get(row.userId); + if (user) { + const inputTokenCount = Number(row.inputTokenCount ?? 0); + const cachedInputTokenCount = Number(row.cachedInputTokenCount ?? 0); + const cacheWriteTokens = Number(row.cacheWriteTokenCount ?? 0); + const outputTokenCount = Number(row.outputTokenCount ?? 0); + const searchUnits = Number(row.searchUnits ?? 0); + const usageKey = `${row.userId}::${row.model ?? ""}::${row.requestType ?? ""}`; + const usage = usageMap.get(usageKey) ?? { + userId: row.userId, + name: user.name, + model: row.model, + requestType: row.requestType, + inputTokens: 0, + cachedInputTokens: 0, + cacheWriteTokens: 0, + outputTokens: 0, + searchUnits: 0, + }; + usage.inputTokens += + inputTokenCount - cachedInputTokenCount - cacheWriteTokens; + usage.cachedInputTokens += cachedInputTokenCount; + usage.cacheWriteTokens += cacheWriteTokens; + usage.outputTokens += outputTokenCount; + usage.searchUnits += searchUnits; + usageMap.set(usageKey, usage); + + const hour = Number(row.hour ?? 0); + hourlyTotals.set( + hour, + (hourlyTotals.get(hour) ?? 0) + + inputTokenCount + + cachedInputTokenCount + + outputTokenCount, + ); + const rawDay = Number(row.day ?? 0); + const dayIndex = rawDay === 0 ? 6 : rawDay - 1; + const dayHourKey = `${dayIndex}-${hour}`; + dayHourTotals.set( + dayHourKey, + (dayHourTotals.get(dayHourKey) ?? 0) + outputTokenCount, + ); + } + const model = row.model ?? null; if (!model) continue; - if (!row.requestTime) continue; const inputTokenCount = Number(row.inputTokenCount ?? 0); const cachedInputTokenCount = Number(row.cachedInputTokenCount ?? 0); @@ -752,54 +835,12 @@ export const reportingRouter = createTRPCRouter({ }; }); - const hourBucket = sql`extract(hour from ${requests.requestTime})`; - const hourlyRows = await ctx.db - .select({ - hour: hourBucket.as("hour"), - totalTokens: - sql`coalesce(sum(${requests.inputTokenCount} + ${requests.cachedInputTokenCount} + ${requests.outputTokenCount}), 0)`.as( - "totalTokens", - ), - }) - .from(requests) - .leftJoin(apikeys, eq(apikeys.uuid, requests.apiKeyId)) - .where(scopedFilter) - .groupBy(hourBucket); - - const hourlyTotals = new Map(); - for (const row of hourlyRows) { - const hour = Number(row.hour ?? 0); - hourlyTotals.set(hour, Number(row.totalTokens ?? 0)); - } - const hourlyTokens = Array.from({ length: 24 }, (_, hour) => ({ hour, avgTokens: (hourlyTotals.get(hour) ?? 0) / dayCount, })); - const dayBucket = sql`extract(dow from ${requests.requestTime})`; - const hourlyDayRows = await ctx.db - .select({ - day: dayBucket.as("day"), - hour: hourBucket.as("hour"), - outputTokens: - sql`coalesce(sum(${requests.outputTokenCount}), 0)`.as( - "outputTokens", - ), - }) - .from(requests) - .leftJoin(apikeys, eq(apikeys.uuid, requests.apiKeyId)) - .where(scopedFilter) - .groupBy(dayBucket, hourBucket); - const dayLabels = ["Mo", "Di", "Mi", "Do", "Fr", "Sa", "So"]; - const dayHourTotals = new Map(); - for (const row of hourlyDayRows) { - const rawDay = Number(row.day ?? 0); - const dayIndex = rawDay === 0 ? 6 : rawDay - 1; - const hour = Number(row.hour ?? 0); - dayHourTotals.set(`${dayIndex}-${hour}`, Number(row.outputTokens ?? 0)); - } const hourlyOutputByDay = Array.from({ length: 7 }, (_, dayIndex) => { const dayLabel = dayLabels[dayIndex] ?? String(dayIndex); @@ -815,42 +856,6 @@ export const reportingRouter = createTRPCRouter({ })); }).flat(); - type UserModel = { - model: string; - requestType: string; - inputTokens: number; - cachedInputTokens: number; - cacheWriteTokens: number; - outputTokens: number; - searchUnits: number; - inputCost: number | null; - cachedCost: number | null; - cacheWriteCost: number | null; - outputCost: number | null; - searchCost: number | null; - totalCost: number | null; - currency: string | null; - currencyIssue: boolean; - currencyTotals: Array<{ currency: string; totalCost: number }>; - }; - type UserEntry = { - id: string; - name: string; - inputTokens: number; - cachedInputTokens: number; - cacheWriteTokens: number; - outputTokens: number; - searchUnits: number; - totalCostScaled: bigint; - totalCostHasKnownCost: boolean; - totalCostHasMissingCost: boolean; - currencyIssue: boolean; - currency: string | null; - currencyTotals: Map; - models: UserModel[]; - }; - const usersMap = new Map(); - const modelTotals = new Map< string, { @@ -871,24 +876,15 @@ export const reportingRouter = createTRPCRouter({ let totalCurrencyIssue = false; let totalCurrency: string | null = null; + const usageRows = Array.from(usageMap.values()).map((row) => ({ + ...row, + inputTokens: Math.max(0, row.inputTokens), + })); + for (const row of usageRows) { const id = row.userId; - const entry: UserEntry = usersMap.get(id) ?? { - id, - name: row.name ?? id, - inputTokens: 0, - cachedInputTokens: 0, - cacheWriteTokens: 0, - outputTokens: 0, - searchUnits: 0, - totalCostScaled: 0n, - totalCostHasKnownCost: false, - totalCostHasMissingCost: false, - currencyIssue: false, - currency: null, - currencyTotals: createCurrencyTotals(), - models: [], - }; + const entry = usersMap.get(id); + if (!entry) continue; const inputTokens = Number(row.inputTokens ?? 0); const cachedTokens = Number(row.cachedInputTokens ?? 0); diff --git a/app/src/server/db/schema.ts b/app/src/server/db/schema.ts index fbeeb14..6fd67af 100644 --- a/app/src/server/db/schema.ts +++ b/app/src/server/db/schema.ts @@ -5,6 +5,7 @@ import { char, date, foreignKey, + index, integer, jsonb, pgEnum, @@ -81,6 +82,11 @@ export const requests = pgTable( .notNull(), }, (table) => [ + index("requests_request_time_idx").on(table.requestTime), + index("requests_api_key_id_request_time_idx").on( + table.apiKeyId, + table.requestTime, + ), foreignKey({ columns: [table.apiKeyId], foreignColumns: [apikeys.uuid], @@ -99,6 +105,7 @@ export const apikeys = pgTable( deactivated: boolean("deactivated").default(false).notNull(), }, (table) => [ + index("apikeys_owner_idx").on(table.owner), foreignKey({ columns: [table.owner], foreignColumns: [users.id], diff --git a/db/migrations/20260825120000_reporting_request_indexes.sql b/db/migrations/20260825120000_reporting_request_indexes.sql new file mode 100644 index 0000000..004621e --- /dev/null +++ b/db/migrations/20260825120000_reporting_request_indexes.sql @@ -0,0 +1,8 @@ +CREATE INDEX IF NOT EXISTS requests_request_time_idx + ON requests (request_time); + +CREATE INDEX IF NOT EXISTS requests_api_key_id_request_time_idx + ON requests (api_key_id, request_time); + +CREATE INDEX IF NOT EXISTS apikeys_owner_idx + ON apikeys (owner); diff --git a/db/migrations/atlas.sum b/db/migrations/atlas.sum index b8ab93b..b1a2e96 100644 --- a/db/migrations/atlas.sum +++ b/db/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:sdM65UZKKLGZMBTXe/1jAzb4vYP94DKMkk25Jp/92BI= +h1:vsZUoaeKtCqy04645t/eYEG7t+6TZmrZPadGjkmA+50= 20240924151008_init.sql h1:rTjsfTqruGXjIUITaR6Pqt0n4lm+4EcdvESuYXgmyP4= 20240924154252_request_monitoring.sql h1:VWH6kcc2DWS7L9tBQ8lgS858/9f5HP7PMYZo0fZlCX8= 20241001124048_id_datatype.sql h1:Nb1OpAzcVshPzgEqQ3w/sCFjorO4h/GTOBeuISanu4k= @@ -25,3 +25,4 @@ h1:sdM65UZKKLGZMBTXe/1jAzb4vYP94DKMkk25Jp/92BI= 20260324120000_drop_backend_pricing_columns.sql h1:ucey3gRgAcRru2kVpNs22+aMbh76jPguUZoKQ7qJl0s= 20260718100000_add_request_types_and_search_units.sql h1:DB3/cpabrYd/E7x18iXCafhMqVMw5TSdvYiBJYXtV0o= 20260818110000_add_gpt56_cache_write_accounting.sql h1:OpcV68YEKID7tO0vFeZU8loBRT3ZaqUHSJqgqiEoF/o= +20260825120000_reporting_request_indexes.sql h1:fHCPTleO4cX5OAcwoJd4n5UJD8vWty/r4sNlItuXyVw= diff --git a/local-dev/docker-compose.yml b/local-dev/docker-compose.yml index 9e9b77e..d5ac84d 100644 --- a/local-dev/docker-compose.yml +++ b/local-dev/docker-compose.yml @@ -1,17 +1,17 @@ services: db: - image: postgres:15 + image: postgres:18 restart: unless-stopped ports: - "54329:5432" environment: - POSTGRES_USER: ${POSTGRES_USER:-openai} + POSTGRES_USER: ${POSTGRES_USER:-oai} POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-openai_pass} POSTGRES_DB: ${POSTGRES_DB:-openai_proxy} volumes: - - ./postgres-data:/var/lib/postgresql/data + - ./postgres-data:/var/lib/postgresql healthcheck: - test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:-openai} -d ${POSTGRES_DB:-openai_proxy}"] + test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:-oai} -d ${POSTGRES_DB:-openai_proxy}"] interval: 2s timeout: 3s retries: 20 @@ -32,11 +32,11 @@ services: environment: # Ensure the app connects to the compose db service DATABASE_HOST: db - DATABASE_USERNAME: ${POSTGRES_USER:-openai} + DATABASE_USERNAME: ${POSTGRES_USER:-oai} DATABASE_PASSWORD: ${POSTGRES_PASSWORD:-openai_pass} DATABASE_NAME: ${POSTGRES_DB:-openai_proxy} # Provide a full connection string for tools that prefer it. - DATABASE_PATH: "postgresql://${POSTGRES_USER:-openai}:${POSTGRES_PASSWORD:-openai_pass}@db/${POSTGRES_DB:-openai_proxy}?sslmode=disable" + DATABASE_PATH: "postgresql://${POSTGRES_USER:-oai}:${POSTGRES_PASSWORD:-openai_pass}@db/${POSTGRES_DB:-openai_proxy}?sslmode=disable" RUST_LOG: "DEBUG" volumes: # Keep DB migrations and editable DB schema available to the container