fix(cms): aggregate report costs in Postgres (#93)

Co-Authored-By: Codex <noreply@openai.com>

Author: snowopsdev

cms/src/components/ops/ReportsView.tsx

@@ -56,7 +56,7 @@ export async function ReportsView(props: AdminViewServerProps) {
       user: req.user,
       overrideAccess: false,
     }),
-    loadReportCosts(req, costWhere),
+    loadReportCosts(req, { createdAtFrom: periodStart ?? undefined }),
     req.payload.find({
       collection: 'pipeline-runs',
       select: { runId: true, status: true, errorSummary: true, completedAt: true },

cms/src/lib/reportQueries.ts

@@ -1,68 +1,75 @@
-import type { PayloadRequest, Where } from 'payload'
+import { sql } from '@payloadcms/db-postgres'
+import { executeAccess, Forbidden, type PayloadRequest } from 'payload'
 import type { CostReport, SpendRow } from '../components/ops/ReportsPanel'
 import type { StageKpiRow } from './opsKpis'
 
-/** Traverse append-only logs by ID, keeping just aggregates between batches. */
+/** The report's supported filters, bound as values rather than SQL fragments. */
+export type ReportCostFilter = {
+  createdAtFrom?: string
+  pipelineRunId?: string
+}
+
+/** Aggregate in one database snapshot without hydrating individual cost logs. */
 export async function loadReportCosts(
   req: PayloadRequest,
-  where: Where,
+  filter: ReportCostFilter,
 ): Promise<{
   aggregate: Pick<CostReport, 'totalUsd' | 'byStage' | 'byModel' | 'rowCount'>
   stages: StageKpiRow[]
 }> {
-  let cursor = 0
+  const access = await executeAccess({ req }, req.payload.collections['cost-log'].config.access.read)
+  // CostLog grants all rows to authenticated users. Fail closed if that policy
+  // ever becomes row-scoped, rather than silently bypassing a new constraint.
+  if (access !== true) throw new Forbidden(req.t)
+
+  const conditions = [sql`true`]
+  if (filter.createdAtFrom !== undefined) {
+    conditions.push(sql`created_at >= ${filter.createdAtFrom}::timestamptz`)
+  }
+  if (filter.pipelineRunId !== undefined) {
+    conditions.push(sql`pipeline_run_id = ${filter.pipelineRunId}`)
+  }
+  const grouped = await req.payload.db.drizzle.execute<{
+    stage: string | null
+    model: string | null
+    calls: string
+    cost_usd: string
+    input_tokens: string
+    output_tokens: string
+  }>(sql`
+    SELECT stage, model, count(*) AS calls,
+      coalesce(sum(cost_usd), 0) AS cost_usd,
+      coalesce(sum(input_tokens), 0) AS input_tokens,
+      coalesce(sum(output_tokens), 0) AS output_tokens
+    FROM ${req.payload.db.tables.cost_log}
+    WHERE ${sql.join(conditions, sql` AND `)}
+    GROUP BY stage, model
+  `)
+
   let rowCount = 0
   let totalUsd = 0
   const byModel = new Map<string, number>()
   const stages = new Map<string, StageKpiRow>()
-  // Pin the upper ID so concurrent appends cannot extend a report indefinitely.
-  const newest = await req.payload.find({
-    collection: 'cost-log',
-    where,
-    sort: '-id',
-    limit: 1,
-    depth: 0,
-    select: { costUsd: true },
-    user: req.user,
-    overrideAccess: false,
-  })
-  const upper = newest.docs[0]?.id ?? 0
-  while (cursor < upper) {
-    const result = await req.payload.find({
-      collection: 'cost-log',
-      where: { and: [where, { id: { greater_than: cursor, less_than_equal: upper } }] },
-      sort: 'id',
-      limit: 1000,
-      // Payload 3.88's Postgres adapter still applies LIMIT with pagination off.
-      // The ID cursor does not need a COUNT of the remaining rows on each batch.
-      pagination: false,
-      depth: 0,
-      select: { costUsd: true, stage: true, model: true, inputTokens: true, outputTokens: true },
-      user: req.user,
-      overrideAccess: false,
-    })
-    if (result.docs.length === 0) break
-    for (const row of result.docs) {
-      rowCount += 1
-      const usd = row.costUsd ?? 0
-      totalUsd += usd
-      const stage = row.stage ?? '(unknown)'
-      const entry = stages.get(stage) ?? {
-        stage,
-        calls: 0,
-        costUsd: 0,
-        inputTokens: 0,
-        outputTokens: 0,
-      }
-      entry.calls += 1
-      entry.costUsd += usd
-      entry.inputTokens += row.inputTokens ?? 0
-      entry.outputTokens += row.outputTokens ?? 0
-      stages.set(stage, entry)
-      const model = row.model ?? '(unknown)'
-      byModel.set(model, (byModel.get(model) ?? 0) + usd)
+  for (const row of grouped.rows) {
+    const calls = Number(row.calls)
+    const usd = Number(row.cost_usd)
+    rowCount += calls
+    totalUsd += usd
+    const stage = row.stage ?? '(unknown)'
+    const entry = stages.get(stage) ?? {
+      stage,
+      calls: 0,
+      costUsd: 0,
+      inputTokens: 0,
+      outputTokens: 0,
     }
-    cursor = result.docs[result.docs.length - 1].id
+    entry.calls += calls
+    entry.costUsd += usd
+    entry.inputTokens += Number(row.input_tokens)
+    entry.outputTokens += Number(row.output_tokens)
+    stages.set(stage, entry)
+    const model = row.model ?? '(unknown)'
+    byModel.set(model, (byModel.get(model) ?? 0) + usd)
   }
   const stageRows = [...stages.values()].sort((a, b) => b.costUsd - a.costUsd)
   const modelRows: SpendRow[] = [...byModel]

cms/tests/int/adminPerformance.int.spec.ts

@@ -134,7 +134,7 @@ describe('admin queries against Postgres', () => {
     ).toBe(false)
   })
 
-  it('accumulates all cost rows beyond 5,000 without reading request bodies', async () => {
+  it('aggregates all cost rows beyond 5,000 without hydrating individual logs', async () => {
     // Real inserts exercise Payload's stored numeric types and pagination.
     for (let offset = 0; offset < 5025; offset += 25) {
       await Promise.all(
@@ -155,11 +155,11 @@ describe('admin queries against Postgres', () => {
         ),
       )
     }
-    // Observe real query results, not mocked pagination. Correct totals alone
-    // would also pass if an adapter accidentally returned all rows at once.
+    // Totals must remain correct without transferring thousands of Payload
+    // documents into the report request, regardless of machine speed.
     const find = vi.spyOn(payload, 'find')
     try {
-      const result = await loadReportCosts(req, { pipelineRunId: { equals: prefix } })
+      const result = await loadReportCosts(req, { pipelineRunId: prefix })
       expect(result.aggregate.rowCount).toBe(5026)
       expect(result.aggregate.totalUsd).toBe(5026 * 0.25)
       expect(result.stages[0]).toMatchObject({
@@ -168,21 +168,49 @@ describe('admin queries against Postgres', () => {
         outputTokens: 5026,
       })
       expect(JSON.stringify(result)).not.toContain('UNUSED_COST_REQUEST')
-      const batchSizes: number[] = []
-      for (const [index, [options]] of find.mock.calls.entries()) {
-        if (options.collection === 'cost-log' && options.limit === 1000) {
-          const batch = await find.mock.results[index].value
-          batchSizes.push(batch.docs.length)
-          expect(JSON.stringify(batch.docs)).not.toContain('UNUSED_COST_REQUEST')
-        }
-      }
-      expect(batchSizes).toEqual([1000, 1000, 1000, 1000, 1000, 26])
+      expect(find.mock.calls.filter(([options]) => options.collection === 'cost-log')).toHaveLength(0)
     } finally {
       find.mockRestore()
     }
-    const empty = await loadReportCosts(req, { pipelineRunId: { equals: `${prefix}-empty` } })
+    const empty = await loadReportCosts(req, { pipelineRunId: `${prefix}-empty` })
     expect(empty.aggregate.rowCount).toBe(0)
   }, 120000)
+
+  it('preserves stage/model totals, null buckets, and inclusive period filters', async () => {
+    const pipelineRunId = `${prefix}-quoted'run`
+    const rows = [
+      { stage: 'generate' as const, model: 'alpha', costUsd: 0.5, inputTokens: 5, createdAt: '2026-01-01T00:00:00.000Z' },
+      { stage: 'generate' as const, model: 'alpha', costUsd: 0.25, outputTokens: 2 },
+      { stage: 'factCheck' as const, model: 'beta', costUsd: 1, inputTokens: 8, outputTokens: 3 },
+      {},
+      { model: '(unknown)', costUsd: 0.125 },
+    ]
+    for (const row of rows) {
+      await payload.create({
+        collection: 'cost-log',
+        overrideAccess: true,
+        data: { pipelineRunId, createdAt: '2026-02-01T00:00:00.000Z', ...row },
+      })
+    }
+    const all = await loadReportCosts(req, { pipelineRunId })
+    expect(all.aggregate).toEqual({
+      rowCount: 5,
+      totalUsd: 1.875,
+      byStage: [{ label: 'factCheck', usd: 1 }, { label: 'generate', usd: 0.75 }, { label: '(unknown)', usd: 0.125 }],
+      byModel: [{ label: 'beta', usd: 1 }, { label: 'alpha', usd: 0.75 }, { label: '(unknown)', usd: 0.125 }],
+    })
+    expect(all.stages).toEqual([
+      { stage: 'factCheck', calls: 1, costUsd: 1, inputTokens: 8, outputTokens: 3 },
+      { stage: 'generate', calls: 2, costUsd: 0.75, inputTokens: 5, outputTokens: 2 },
+      { stage: '(unknown)', calls: 2, costUsd: 0.125, inputTokens: 0, outputTokens: 0 },
+    ])
+    const recent = await loadReportCosts(req, { pipelineRunId, createdAtFrom: '2026-02-01T00:00:00.000Z' })
+    expect(recent.aggregate.rowCount).toBe(4)
+    expect(recent.aggregate.totalUsd).toBe(1.375)
+    const empty = await loadReportCosts(req, { pipelineRunId, createdAtFrom: '2026-03-01T00:00:00.000Z' })
+    expect(empty.aggregate).toEqual({ rowCount: 0, totalUsd: 0, byStage: [], byModel: [] })
+    expect(empty.stages).toEqual([])
+  })
 })
 
 it('report summaries preserve missing decisions, QA denominators, and spend', () => {