feat(stats): add stats aggregation tables - #1612
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughAdds hourly statistics: new DB tables and indexes, a worker service to aggregate logs into per-hour stats, and updates the activity route/tests to consume hourly stats (with separate apiKey/project aggregation paths) instead of full log-based daily aggregation. Changes
Sequence DiagramssequenceDiagram
participant Worker as Worker Process
participant Agg as ProjectStatsAggregator
participant DB as Database
participant Log as Log Table
participant Hourly as Hourly Stats Tables
Worker->>Agg: refreshProjectHourlyStats()
Agg->>DB: Query project/hour buckets with statsAggregatedAt IS NULL
DB-->>Agg: project-hour buckets
loop per bucket
Agg->>Log: Query logs for project & hour
Log-->>Agg: Log rows
Agg->>Agg: Compute aggregates (counts, tokens, costs, errors)
Agg->>Hourly: Upsert projectHourlyStats / projectHourlyModelStats / apiKeyHourlyStats
Agg->>Log: Update logs.statsAggregatedAt
end
Agg->>DB: Find projects with current-hour logs
DB-->>Agg: project list
Agg->>Hourly: Recalculate/refresh current-hour stats
Agg-->>Worker: Done
sequenceDiagram
participant Client as API Client
participant Activity as Activity Route
participant DB as Database
participant Hourly as Hourly Stats Tables
Client->>Activity: GET /activity?projectId=X&apiKeyId=Y
alt apiKeyId provided
Activity->>Hourly: Query `api_key_hourly_stats` for apiKeyId & date range
Hourly-->>Activity: Hourly records
Activity->>Activity: Map hourly→daily, compute errorRate & cacheRate (modelBreakdown empty)
else project-level
Activity->>Hourly: Query `project_hourly_stats` for project & date range
Activity->>Hourly: Query `project_hourly_model_stats` for model breakdown per hour
Hourly-->>Activity: Hourly project + model records
Activity->>Activity: Aggregate by date, assemble modelBreakdown map, compute rates
end
Activity-->>Client: JSON daily activity array
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested labels
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches🧪 Generate unit tests (beta)
Tip Issue Planner is now in beta. Read the docs and try it out! Share your feedback on Discord. Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Pull request overview
Adds hourly pre-aggregation tables and a worker refresh loop to speed up dashboard/activity queries by reading from precomputed hourly buckets instead of scanning raw logs.
Changes:
- Add
project_hourly_statsandproject_hourly_model_statstables + migration for hourly rollups. - Add a worker service to backfill and periodically refresh hourly rollups.
- Update
/activityAPI to read from hourly aggregation tables, with a raw-log fallback when filtering byapiKeyId.
Reviewed changes
Copilot reviewed 6 out of 7 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| packages/db/src/schema.ts | Defines new hourly aggregation tables and indexes. |
| packages/db/migrations/meta/_journal.json | Registers the new DB migration. |
| packages/db/migrations/1770544954_parallel_mindworm.sql | Creates the new aggregation tables and indexes. |
| apps/worker/src/worker.ts | Schedules hourly stats refresh/backfill at startup and on an interval. |
| apps/worker/src/services/project-stats-aggregator.ts | Implements hourly aggregation refresh + startup backfill. |
| apps/api/src/routes/activity.ts | Switches activity queries to aggregation tables (raw-log fallback when apiKeyId is provided). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| // Upsert project hourly model stats | ||
| for (const stat of modelStats) { | ||
| await database | ||
| .insert(projectHourlyModelStats) | ||
| .values({ | ||
| projectId: stat.projectId, | ||
| hourTimestamp: roundedTargetHour, | ||
| usedModel: stat.usedModel, | ||
| usedProvider: stat.usedProvider, | ||
| requestCount: stat.requestCount, | ||
| errorCount: stat.errorCount, | ||
| cacheCount: stat.cacheCount, | ||
| inputTokens: stat.inputTokens, | ||
| outputTokens: stat.outputTokens, | ||
| totalTokens: stat.totalTokens, | ||
| cost: stat.cost, | ||
| inputCost: stat.inputCost, | ||
| outputCost: stat.outputCost, | ||
| }) | ||
| .onConflictDoUpdate({ | ||
| target: [ | ||
| projectHourlyModelStats.projectId, | ||
| projectHourlyModelStats.hourTimestamp, | ||
| projectHourlyModelStats.usedModel, | ||
| projectHourlyModelStats.usedProvider, | ||
| ], | ||
| set: { | ||
| requestCount: stat.requestCount, | ||
| errorCount: stat.errorCount, | ||
| cacheCount: stat.cacheCount, | ||
| inputTokens: stat.inputTokens, | ||
| outputTokens: stat.outputTokens, | ||
| totalTokens: stat.totalTokens, | ||
| cost: stat.cost, | ||
| inputCost: stat.inputCost, | ||
| outputCost: stat.outputCost, | ||
| updatedAt: new Date(), | ||
| }, | ||
| }); | ||
| } |
There was a problem hiding this comment.
Same as above: this upsert does one INSERT per (project, model, provider) row. For large datasets this will be expensive during refresh/backfill. Prefer batched inserts (multi-values) and/or a transaction to reduce round-trips.
| // If filtering by apiKeyId, we need to fall back to raw log table queries | ||
| // since aggregation tables don't track per-API-key stats | ||
| if (apiKeyId) { | ||
| // Query daily aggregated data from raw logs (fallback for apiKeyId filter) | ||
| const dailyAggregates = await db | ||
| .select({ | ||
| date: sql<string>`DATE(${tables.log.createdAt})`.as("date"), |
There was a problem hiding this comment.
There are existing Vitest specs for /activity that insert only tables.log rows. With the new aggregation-table code path, those tests (and any environment that hasn’t backfilled aggregates yet) will not exercise the primary path. Update/add tests to cover (1) aggregation-table queries and (2) the apiKeyId fallback path, or implement an automatic fallback when aggregates are missing.
| gte(projectHourlyStats.hourTimestamp, startDate), | ||
| lte(projectHourlyStats.hourTimestamp, endDate), |
There was a problem hiding this comment.
The hourly-aggregation path filters by hourTimestamp >= startDate, but startDate includes minutes/seconds. This will drop the hour bucket that contains startDate (e.g., startDate=10:30 excludes the 10:00 bucket), causing missing data compared to the raw-log path. Consider rounding the query bounds to hour/day boundaries (or widening by one bucket) so partial-hour ranges don’t undercount.
| gte(projectHourlyStats.hourTimestamp, startDate), | |
| lte(projectHourlyStats.hourTimestamp, endDate), | |
| gte( | |
| projectHourlyStats.hourTimestamp, | |
| sql`DATE_TRUNC('hour', ${startDate}::timestamptz)`, | |
| ), | |
| lte( | |
| projectHourlyStats.hourTimestamp, | |
| sql`DATE_TRUNC('hour', ${endDate}::timestamptz)`, | |
| ), |
| // Use aggregation tables for fast queries (when not filtering by apiKeyId) | ||
| // Query hourly aggregated data from projectHourlyStats table | ||
| const hourlyAggregates = await db |
There was a problem hiding this comment.
This endpoint now exclusively reads from project_hourly_stats/project_hourly_model_stats when apiKeyId is not provided. If the worker hasn’t populated/backfilled these tables yet (fresh deploy, worker down, tests that only insert raw logs), the API will return empty/incorrect activity. Add a fallback to raw-log aggregation when the hourly tables have no data for the requested range (or otherwise ensure aggregates are guaranteed to exist before serving).
| // Configuration for project stats refresh interval (defaults to 60 seconds) | ||
| export const PROJECT_STATS_REFRESH_INTERVAL_SECONDS = | ||
| Number(process.env.PROJECT_STATS_REFRESH_INTERVAL_SECONDS) || 60; | ||
|
|
||
| // Configuration for backfill duration in hours (defaults to 24 hours) | ||
| const PROJECT_STATS_BACKFILL_HOURS = | ||
| Number(process.env.PROJECT_STATS_BACKFILL_HOURS) || 24; |
There was a problem hiding this comment.
PROJECT_STATS_REFRESH_INTERVAL_SECONDS/PROJECT_STATS_BACKFILL_HOURS use Number(env) || default, which allows negative values (truthy) and non-integers. A negative/very small interval can cause extremely tight refresh loops and heavy DB load. Validate these env vars (e.g., require finite positive integers, clamp to a sane minimum) before using them.
| // We're missing some hours, backfill from after the last recorded hour | ||
| logger.info( | ||
| `Found gap of ${hoursBehind} hours. Backfilling from ${lastHour.toISOString()}`, | ||
| ); | ||
| startFromHour = new Date(lastHour.getTime() + 60 * 60 * 1000); |
There was a problem hiding this comment.
Backfill start hour isn’t constrained by PROJECT_STATS_BACKFILL_HOURS when there is an existing but very old latestStats row. If lastHour is far in the past, startFromHour = lastHour + 1h combined with maxHours = PROJECT_STATS_BACKFILL_HOURS + 1 will stop early and still leave a large gap. Consider setting startFromHour to the later of (lastHour + 1h) and backfillStartRounded so startup backfill always covers the most recent configured window.
| // We're missing some hours, backfill from after the last recorded hour | |
| logger.info( | |
| `Found gap of ${hoursBehind} hours. Backfilling from ${lastHour.toISOString()}`, | |
| ); | |
| startFromHour = new Date(lastHour.getTime() + 60 * 60 * 1000); | |
| // We're missing some hours. Start from after the last recorded hour, | |
| // but do not go earlier than the configured backfill window start. | |
| const candidateStart = new Date(lastHour.getTime() + 60 * 60 * 1000); | |
| startFromHour = | |
| candidateStart.getTime() < backfillStartRounded.getTime() | |
| ? backfillStartRounded | |
| : candidateStart; | |
| logger.info( | |
| `Found gap of ${hoursBehind} hours. Backfilling from ${startFromHour.toISOString()}`, | |
| ); |
| // Upsert project hourly stats | ||
| for (const stat of projectStats) { | ||
| await database | ||
| .insert(projectHourlyStats) | ||
| .values({ | ||
| projectId: stat.projectId, | ||
| hourTimestamp: roundedTargetHour, | ||
| requestCount: stat.requestCount, | ||
| errorCount: stat.errorCount, | ||
| cacheCount: stat.cacheCount, | ||
| inputTokens: stat.inputTokens, | ||
| outputTokens: stat.outputTokens, | ||
| totalTokens: stat.totalTokens, | ||
| reasoningTokens: stat.reasoningTokens, | ||
| cachedTokens: stat.cachedTokens, | ||
| cost: stat.cost, | ||
| inputCost: stat.inputCost, | ||
| outputCost: stat.outputCost, | ||
| requestCost: stat.requestCost, | ||
| dataStorageCost: stat.dataStorageCost, | ||
| serviceFee: stat.serviceFee, | ||
| discountSavings: stat.discountSavings, | ||
| }) | ||
| .onConflictDoUpdate({ | ||
| target: [ | ||
| projectHourlyStats.projectId, | ||
| projectHourlyStats.hourTimestamp, | ||
| ], | ||
| set: { | ||
| requestCount: stat.requestCount, | ||
| errorCount: stat.errorCount, | ||
| cacheCount: stat.cacheCount, | ||
| inputTokens: stat.inputTokens, | ||
| outputTokens: stat.outputTokens, | ||
| totalTokens: stat.totalTokens, | ||
| reasoningTokens: stat.reasoningTokens, | ||
| cachedTokens: stat.cachedTokens, | ||
| cost: stat.cost, | ||
| inputCost: stat.inputCost, | ||
| outputCost: stat.outputCost, | ||
| requestCost: stat.requestCost, | ||
| dataStorageCost: stat.dataStorageCost, | ||
| serviceFee: stat.serviceFee, | ||
| discountSavings: stat.discountSavings, | ||
| updatedAt: new Date(), | ||
| }, | ||
| }); | ||
| } |
There was a problem hiding this comment.
The upsert runs one INSERT per project inside a loop, which can create a large number of DB round-trips every refresh/backfill. Consider batching inserts (multi-values insert) and/or wrapping the hour calculation + upsert in a transaction to reduce overhead and improve throughput.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In `@apps/worker/src/services/project-stats-aggregator.ts`:
- Around line 320-399: backfillProjectHourlyStatsIfNeeded currently has a TOCTOU
race when multiple workers run; wrap the entire decision-and-backfill sequence
in the same cluster-wide lock used elsewhere by calling the existing acquireLock
(the one used in worker.ts) with a unique key like
"backfill-project-hourly-stats" and a reasonable TTL, return early if the lock
cannot be acquired, and ensure the lock is always released (finally) so only one
instance executes the reads and the expensive calculateProjectHourlyStatsForHour
/ calculateProjectHourlyModelStatsForHour computations at a time.
- Around line 24-34: roundToHourStart currently builds a Date using local-time
constructors which can misalign with DB timestamps; change it to construct the
hour-start in UTC by using UTC-aware APIs (e.g., Date.UTC or
getUTCFullYear/getUTCMonth/getUTCDate/getUTCHours) so the returned Date
represents the start of the hour in UTC; update any callers (e.g.,
getCurrentHourStart or aggregation logic that uses roundToHourStart) to expect
an UTC-aligned Date to match PostgreSQL timestamp semantics.
🧹 Nitpick comments (4)
packages/db/src/schema.ts (1)
1134-1175: Redundant index duplicating the unique constraint.The
unique().on(table.projectId, table.hourTimestamp)at line 1166 already creates an implicit composite index in PostgreSQL. The explicit indexproject_hourly_stats_project_id_hour_timestamp_idxon the same columns (lines 1168-1171) is redundant and doubles the write overhead for inserts/updates. The same applies toprojectHourlyModelStats(lines 1206-1211 vs. 1213-1216).♻️ Suggested fix: remove the redundant indexes
(table) => [ // Unique constraint for one record per project-hour unique().on(table.projectId, table.hourTimestamp), - // Index for dashboard queries (project + time range) - index("project_hourly_stats_project_id_hour_timestamp_idx").on( - table.projectId, - table.hourTimestamp, - ), // Index for worker refresh queries (find hours to update) index("project_hourly_stats_hour_timestamp_idx").on(table.hourTimestamp), ],Same for
projectHourlyModelStats:(table) => [ // Unique constraint for one record per project-hour-model-provider unique().on( table.projectId, table.hourTimestamp, table.usedModel, table.usedProvider, ), - // Index for dashboard queries (project + time range) - index("project_hourly_model_stats_project_id_hour_timestamp_idx").on( - table.projectId, - table.hourTimestamp, - ), // Index for worker refresh queries index("project_hourly_model_stats_hour_timestamp_idx").on( table.hourTimestamp, ), ],Note: For
projectHourlyModelStats, the unique constraint is on 4 columns(projectId, hourTimestamp, usedModel, usedProvider), so the composite index on just(projectId, hourTimestamp)is actually not redundant—it serves the dashboard query pattern that only filters by project + time range. Only theprojectHourlyStatsindex is truly redundant. Apologies for the over-generalization—please keep the model stats index and only remove the one forprojectHourlyStats.apps/worker/src/services/project-stats-aggregator.ts (2)
55-58: Unnecessary aliasing:const database = db.
dbis already a module-level import. The local aliasdatabaseon lines 58, 190, and 324 adds no value and reduces readability.
131-178: Batch the sequential upserts into bulk operations to eliminate N+1 queries.Lines 131–178 and 238–276 each perform sequential per-row inserts in a loop, causing many individual
INSERT ... ON CONFLICT DO UPDATEround-trips. Replace each loop with a single bulk insert call.Drizzle ORM supports
.values([...]).onConflictDoUpdate()for multi-row upserts. Usesql\excluded.column_name`` (with the database column name in snake_case) to reference inserted values in the conflict update clause:♻️ Batch upsert approach
- // Upsert project hourly stats - for (const stat of projectStats) { - await database - .insert(projectHourlyStats) - .values({ - projectId: stat.projectId, - hourTimestamp: roundedTargetHour, - requestCount: stat.requestCount, - // ... fields - }) - .onConflictDoUpdate({ - // ... - }); - } + // Upsert project hourly stats in batch + if (projectStats.length > 0) { + await database + .insert(projectHourlyStats) + .values( + projectStats.map((stat) => ({ + projectId: stat.projectId, + hourTimestamp: roundedTargetHour, + requestCount: stat.requestCount, + errorCount: stat.errorCount, + cacheCount: stat.cacheCount, + inputTokens: stat.inputTokens, + outputTokens: stat.outputTokens, + totalTokens: stat.totalTokens, + reasoningTokens: stat.reasoningTokens, + cachedTokens: stat.cachedTokens, + cost: stat.cost, + inputCost: stat.inputCost, + outputCost: stat.outputCost, + requestCost: stat.requestCost, + dataStorageCost: stat.dataStorageCost, + serviceFee: stat.serviceFee, + discountSavings: stat.discountSavings, + })), + ) + .onConflictDoUpdate({ + target: [ + projectHourlyStats.projectId, + projectHourlyStats.hourTimestamp, + ], + set: { + requestCount: sql`excluded.request_count`, + errorCount: sql`excluded.error_count`, + cacheCount: sql`excluded.cache_count`, + inputTokens: sql`excluded.input_tokens`, + outputTokens: sql`excluded.output_tokens`, + totalTokens: sql`excluded.total_tokens`, + reasoningTokens: sql`excluded.reasoning_tokens`, + cachedTokens: sql`excluded.cached_tokens`, + cost: sql`excluded.cost`, + inputCost: sql`excluded.input_cost`, + outputCost: sql`excluded.output_cost`, + requestCost: sql`excluded.request_cost`, + dataStorageCost: sql`excluded.data_storage_cost`, + serviceFee: sql`excluded.service_fee`, + discountSavings: sql`excluded.discount_savings`, + updatedAt: new Date(), + }, + }); + }Apply the same pattern to model stats (lines 238–276).
apps/api/src/routes/activity.ts (1)
136-302: Significant code duplication between the apiKeyId fallback path and the aggregation path.The transformation logic at lines 260-297 is nearly identical to lines 432-468. Similarly, the model breakdown map construction at lines 240-257 duplicates lines 416-429. Consider extracting a shared helper (e.g.,
buildActivityResponse) to DRY this up.
| function roundToHourStart(date: Date): Date { | ||
| return new Date( | ||
| date.getFullYear(), | ||
| date.getMonth(), | ||
| date.getDate(), | ||
| date.getHours(), | ||
| 0, | ||
| 0, | ||
| 0, | ||
| ); | ||
| } |
There was a problem hiding this comment.
Potential timezone mismatch between JavaScript Date and database timestamps.
roundToHourStart constructs dates using new Date(year, month, day, hours, ...) which uses the local system timezone. If the worker process runs in a non-UTC timezone, the hour boundaries won't align with the timestamp (without timezone) values stored in PostgreSQL (which default to now() in UTC).
For example, on a server at UTC+2, getCurrentHourStart() at 14:30 UTC would compute 16:00 local (i.e. 16:00 as a wall-clock time), but the DB would store createdAt as 14:xx. The WHERE createdAt >= '16:00' AND createdAt < '17:00' would match the wrong rows.
Consider using UTC-based methods:
🔧 Proposed fix: use UTC methods
function roundToHourStart(date: Date): Date {
- return new Date(
- date.getFullYear(),
- date.getMonth(),
- date.getDate(),
- date.getHours(),
- 0,
- 0,
- 0,
- );
+ const d = new Date(date);
+ d.setUTCMinutes(0, 0, 0);
+ return d;
}🤖 Prompt for AI Agents
In `@apps/worker/src/services/project-stats-aggregator.ts` around lines 24 - 34,
roundToHourStart currently builds a Date using local-time constructors which can
misalign with DB timestamps; change it to construct the hour-start in UTC by
using UTC-aware APIs (e.g., Date.UTC or
getUTCFullYear/getUTCMonth/getUTCDate/getUTCHours) so the returned Date
represents the start of the hour in UTC; update any callers (e.g.,
getCurrentHourStart or aggregation logic that uses roundToHourStart) to expect
an UTC-aligned Date to match PostgreSQL timestamp semantics.
| export async function backfillProjectHourlyStatsIfNeeded() { | ||
| logger.info("Checking for missing project hourly stats to backfill..."); | ||
|
|
||
| try { | ||
| const database = db; | ||
|
|
||
| // Get the most recent hourly stats entry | ||
| const latestStats = await database | ||
| .select({ hourTimestamp: projectHourlyStats.hourTimestamp }) | ||
| .from(projectHourlyStats) | ||
| .orderBy(sql`${projectHourlyStats.hourTimestamp} DESC`) | ||
| .limit(1); | ||
|
|
||
| const previousHour = getPreviousHourStart(); | ||
| const backfillStartHour = new Date( | ||
| Date.now() - PROJECT_STATS_BACKFILL_HOURS * 60 * 60 * 1000, | ||
| ); | ||
| const backfillStartRounded = roundToHourStart(backfillStartHour); | ||
|
|
||
| let startFromHour: Date; | ||
|
|
||
| if (latestStats.length === 0) { | ||
| // No existing stats, start from backfill start | ||
| logger.info( | ||
| `No existing project hourly stats found. Starting backfill from ${backfillStartRounded.toISOString()}`, | ||
| ); | ||
| startFromHour = backfillStartRounded; | ||
| } else { | ||
| const lastHour = latestStats[0]!.hourTimestamp; | ||
| const hoursBehind = Math.floor( | ||
| (previousHour.getTime() - lastHour.getTime()) / (60 * 60 * 1000), | ||
| ); | ||
|
|
||
| if (hoursBehind > 1) { | ||
| // We're missing some hours, backfill from after the last recorded hour | ||
| logger.info( | ||
| `Found gap of ${hoursBehind} hours. Backfilling from ${lastHour.toISOString()}`, | ||
| ); | ||
| startFromHour = new Date(lastHour.getTime() + 60 * 60 * 1000); | ||
| } else { | ||
| logger.info( | ||
| `Project hourly stats are up to date. Last entry: ${lastHour.toISOString()}`, | ||
| ); | ||
| return; | ||
| } | ||
| } | ||
|
|
||
| // Backfill each missing hour | ||
| let hour = startFromHour; | ||
| let hoursProcessed = 0; | ||
| const maxHours = PROJECT_STATS_BACKFILL_HOURS + 1; // Safety limit | ||
|
|
||
| while (hour <= previousHour && hoursProcessed < maxHours) { | ||
| const projectCount = await calculateProjectHourlyStatsForHour(hour); | ||
| const modelCount = await calculateProjectHourlyModelStatsForHour(hour); | ||
|
|
||
| logger.info( | ||
| `Backfilled project hourly stats for ${hour.toISOString()}: ${projectCount} projects, ${modelCount} model entries`, | ||
| ); | ||
|
|
||
| hour = new Date(hour.getTime() + 60 * 60 * 1000); | ||
| hoursProcessed++; | ||
| } | ||
|
|
||
| if (hoursProcessed >= maxHours) { | ||
| logger.warn( | ||
| `Backfill stopped at limit of ${maxHours} hours to prevent excessive processing`, | ||
| ); | ||
| } | ||
|
|
||
| logger.info( | ||
| `Project hourly stats backfill complete. Processed ${hoursProcessed} hours.`, | ||
| ); | ||
| } catch (error) { | ||
| logger.error( | ||
| "Error during project hourly stats backfill", | ||
| error instanceof Error ? error : new Error(String(error)), | ||
| ); | ||
| throw error; | ||
| } |
There was a problem hiding this comment.
Backfill has no protection against concurrent execution by multiple worker instances.
If multiple worker replicas start simultaneously, backfillProjectHourlyStatsIfNeeded will run concurrently on each instance. Since the function reads the latest stats, then writes, there's a TOCTOU race where both workers could decide to backfill the same hours. The upsert's ON CONFLICT DO UPDATE prevents data corruption, but both workers will redundantly re-compute the same expensive aggregation queries against the log table.
Consider using the existing acquireLock mechanism (used elsewhere in worker.ts) to ensure only one instance performs the backfill.
🤖 Prompt for AI Agents
In `@apps/worker/src/services/project-stats-aggregator.ts` around lines 320 - 399,
backfillProjectHourlyStatsIfNeeded currently has a TOCTOU race when multiple
workers run; wrap the entire decision-and-backfill sequence in the same
cluster-wide lock used elsewhere by calling the existing acquireLock (the one
used in worker.ts) with a unique key like "backfill-project-hourly-stats" and a
reasonable TTL, return early if the lock cannot be acquired, and ensure the lock
is always released (finally) so only one instance executes the reads and the
expensive calculateProjectHourlyStatsForHour /
calculateProjectHourlyModelStatsForHour computations at a time.
Add aggregation tables to pre-compute dashboard statistics hourly, significantly improving dashboard load times by avoiding expensive queries on the raw log table. Changes: - Add project_hourly_stats and project_hourly_model_stats tables - Add worker service to refresh aggregations every minute (configurable) - Update activity API route to use aggregation tables - Fallback to raw log queries when filtering by apiKeyId - Include backfill logic for missing historical data Environment variables: - PROJECT_STATS_REFRESH_INTERVAL_SECONDS (default: 60) - PROJECT_STATS_BACKFILL_HOURS (default: 24) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
1b105af to
048ace7
Compare
Instead of time-based backfill with PROJECT_STATS_BACKFILL_HOURS, now uses a marker field on logs to track which have been aggregated: - Add statsAggregatedAt column to log table - Add partial index for efficient queries on unprocessed logs - Process logs by finding those with NULL statsAggregatedAt - Mark logs as processed after aggregating their hour bucket - Works for any time range without configuration This ensures aggregation is always accurate regardless of how far back the data goes, and handles late-arriving logs correctly. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@apps/worker/src/services/project-stats-aggregator.ts`:
- Around line 318-332: The update call is incorrectly assuming result is an
array; change the database.update(...).set(...).where(...) call on the log table
to include a .returning() (e.g., .returning({ id: log.id })) so it returns the
updated rows and then compute affectedRows as the returned array's length (or if
you prefer the pg Result use result.rowCount) and add that to totalLogsMarked;
update the code around the result variable and totalLogsMarked to use the
returned rows from .returning() (or result.rowCount) instead of
Array.isArray(result).
🧹 Nitpick comments (4)
packages/db/src/schema.ts (1)
1139-1180: Redundant index: the unique constraint already creates an index on(projectId, hourTimestamp).The
unique().on(table.projectId, table.hourTimestamp)at line 1171 implicitly creates a B-tree index in PostgreSQL. The explicit index at lines 1173-1176 is therefore redundant. Same applies toprojectHourlyModelStats(lines 1211-1216 vs 1218-1221).This is minor and won't cause bugs, but it doubles the write overhead for upserts on these columns.
♻️ Proposed fix — remove redundant indexes
(table) => [ // Unique constraint for one record per project-hour unique().on(table.projectId, table.hourTimestamp), - // Index for dashboard queries (project + time range) - index("project_hourly_stats_project_id_hour_timestamp_idx").on( - table.projectId, - table.hourTimestamp, - ), // Index for worker refresh queries (find hours to update) index("project_hourly_stats_hour_timestamp_idx").on(table.hourTimestamp), ],And similarly for
projectHourlyModelStats:(table) => [ unique().on( table.projectId, table.hourTimestamp, table.usedModel, table.usedProvider, ), - index("project_hourly_model_stats_project_id_hour_timestamp_idx").on( - table.projectId, - table.hourTimestamp, - ), index("project_hourly_model_stats_hour_timestamp_idx").on( table.hourTimestamp, ), ],Note: for
projectHourlyModelStats, the unique index is on 4 columns, so a separate(projectId, hourTimestamp)index is useful for dashboard range queries. Keep it there but remove only the one forprojectHourlyStats.apps/worker/src/services/project-stats-aggregator.ts (3)
224-262: Sequential upserts for model stats — consider batching if bucket cardinality is high.Each model/provider combination issues a separate
INSERT ... ON CONFLICT DO UPDATEquery. For projects with many model/provider combos, this means many round trips per bucket × up to 100 buckets per cycle.This isn't urgent (worker runs on an interval), but a single batch upsert using
db.insert(...).values(allModelStats).onConflictDoUpdate(...)would reduce round trips significantly.♻️ Proposed batch upsert
- for (const stat of modelStats) { - await database - .insert(projectHourlyModelStats) - .values({ - projectId, - hourTimestamp, - usedModel: stat.usedModel, - usedProvider: stat.usedProvider, - requestCount: stat.requestCount, - errorCount: stat.errorCount, - cacheCount: stat.cacheCount, - inputTokens: stat.inputTokens, - outputTokens: stat.outputTokens, - totalTokens: stat.totalTokens, - cost: stat.cost, - inputCost: stat.inputCost, - outputCost: stat.outputCost, - }) - .onConflictDoUpdate({ - target: [ - projectHourlyModelStats.projectId, - projectHourlyModelStats.hourTimestamp, - projectHourlyModelStats.usedModel, - projectHourlyModelStats.usedProvider, - ], - set: { - requestCount: sql`excluded.request_count`, - errorCount: sql`excluded.error_count`, - cacheCount: sql`excluded.cache_count`, - inputTokens: sql`excluded.input_tokens`, - outputTokens: sql`excluded.output_tokens`, - totalTokens: sql`excluded.total_tokens`, - cost: sql`excluded.cost`, - inputCost: sql`excluded.input_cost`, - outputCost: sql`excluded.output_cost`, - updatedAt: new Date(), - }, - }); - } + if (modelStats.length > 0) { + await database + .insert(projectHourlyModelStats) + .values( + modelStats.map((stat) => ({ + projectId, + hourTimestamp, + usedModel: stat.usedModel, + usedProvider: stat.usedProvider, + requestCount: stat.requestCount, + errorCount: stat.errorCount, + cacheCount: stat.cacheCount, + inputTokens: stat.inputTokens, + outputTokens: stat.outputTokens, + totalTokens: stat.totalTokens, + cost: stat.cost, + inputCost: stat.inputCost, + outputCost: stat.outputCost, + })), + ) + .onConflictDoUpdate({ + target: [ + projectHourlyModelStats.projectId, + projectHourlyModelStats.hourTimestamp, + projectHourlyModelStats.usedModel, + projectHourlyModelStats.usedProvider, + ], + set: { + requestCount: sql`excluded.request_count`, + errorCount: sql`excluded.error_count`, + cacheCount: sql`excluded.cache_count`, + inputTokens: sql`excluded.input_tokens`, + outputTokens: sql`excluded.output_tokens`, + totalTokens: sql`excluded.total_tokens`, + cost: sql`excluded.cost`, + inputCost: sql`excluded.input_cost`, + outputCost: sql`excluded.output_cost`, + updatedAt: new Date(), + }, + }); + }
370-377:refreshCurrentHourStatsscans all projects without index support.This query filters only on
log.createdAt >= currentHourStartwith noprojectIdpredicate, then groups byprojectId. The existing indexlog_project_id_created_at_idxhasprojectIdas the leading column, so it won't help here. The partial indexlog_stats_aggregated_at_null_idxalso leads withprojectId.For a high-traffic system, this could become a sequential scan on the log table. The index
log_created_at_used_model_used_provider_idx(leading oncreatedAt) may partially help but carries extra columns.Consider adding a covering index on
(createdAt, projectId)or, alternatively, tracking "active projects" separately to avoid scanning the log table at all.
400-418:refreshProjectHourlyStatsre-throws errors — ensure the caller handles them.Both
processUnprocessedLogsandrefreshCurrentHourStatscatch, log, and re-throw. ThenrefreshProjectHourlyStatsalso catches, logs, and re-throws. This means every error is logged twice (once in the inner function, once here). If the caller (worker interval) doesn't catch, it could crash the worker process.Consider either: (a) not re-throwing in the inner functions (let the outer function handle it), or (b) not re-throwing here and letting the interval silently recover.
| const result = await database | ||
| .update(log) | ||
| .set({ statsAggregatedAt: now }) | ||
| .where( | ||
| and( | ||
| sql`${log.projectId} = ${bucket.projectId}`, | ||
| gte(log.createdAt, hourTimestamp), | ||
| lt(log.createdAt, hourEnd), | ||
| isNull(log.statsAggregatedAt), | ||
| ), | ||
| ); | ||
|
|
||
| // Count affected rows (drizzle returns the updated rows) | ||
| const affectedRows = Array.isArray(result) ? result.length : 0; | ||
| totalLogsMarked += affectedRows; |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
Drizzle ORM update without returning return type
💡 Result:
In Drizzle, an UPDATE without .returning(...) does not return updated row types—it returns the driver’s execution result (and that result type/shape differs per dialect/driver). Drizzle’s docs show the “typed rows” path only when you use .returning(...) (PG/SQLite) or .output(...) (MSSQL). (orm.drizzle.team)
What you get back (common cases)
- Postgres (postgres-js driver): an empty
RowList, but it includes a.countwith affected rows (so you can doresult.count). (answeroverflow.com) - Postgres (node-postgres / pg driver): a
QueryResult-like object where affected rows are typicallyrowCount(driver-specific). - MySQL (mysql2): a
ResultSetHeader-like object withaffectedRows(driver-specific). - SQLite drivers: a “run result” object (often exposed as
changes/rowsAffected, driver-specific).
How to type it in TS
Because it’s driver-specific, the most reliable way is:
const q = db.update(users).set({ name: 'x' }).where(eq(users.id, 1));
const res: Awaited<ReturnType<typeof q.execute>> = await q.execute();If you tell me which Drizzle driver you’re using (e.g. drizzle-orm/node-postgres, drizzle-orm/postgres-js, drizzle-orm/mysql2, drizzle-orm/better-sqlite3, etc.), I can give the exact concrete return type and the exact field to read for “rows affected”.
Citations:
- 1: https://orm.drizzle.team/docs/update
- 2: https://www.answeroverflow.com/m/1196185391930290226?utm_source=openai
🏁 Script executed:
# Find Drizzle driver configuration
fd -e "ts" -e "js" -e "json" | xargs grep -l "drizzle-orm" | head -20Repository: theopenco/llmgateway
Length of output: 501
🏁 Script executed:
# Check package.json for driver dependency
cat package.json | grep -A 5 -B 5 "drizzle"Repository: theopenco/llmgateway
Length of output: 46
🏁 Script executed:
# Look for database initialization in the worker project
find apps/worker -name "*database*" -o -name "*db*" | grep -E "\.(ts|js)$"Repository: theopenco/llmgateway
Length of output: 46
🏁 Script executed:
# Check the Drizzle driver configuration in db.ts
cat packages/db/src/db.tsRepository: theopenco/llmgateway
Length of output: 2113
🏁 Script executed:
# Check package.json in db package
cat packages/db/package.json | grep -i drizzleRepository: theopenco/llmgateway
Length of output: 374
🌐 Web query:
drizzle-orm node-postgres update rowCount affected rows
💡 Result:
With Drizzle ORM + node-postgres (pg), the “affected rows” count comes from pg’s Result.rowCount (not from the array Drizzle returns unless you use returning). pg documents rowCount as “the number of rows processed by the last command” (e.g. UPDATE). (node-postgres.com)
Practical options
- Get the count via
RETURNING(Drizzle query builder)
const rows = await db
.update(users)
.set({ verified: true })
.where(eq(users.orgId, orgId))
.returning({ id: users.id }); // return minimal data
const affected = rows.length;Drizzle’s update API supports returning() on Postgres. (orm.drizzle.team)
- Get the count via the underlying driver result (
rowCount)
Usedb.execute(...)and readrowCountfrom thepgresult:
import { sql } from "drizzle-orm";
const res = await db.execute(
sql`update ${users} set ${users.verified} = true where ${users.orgId} = ${orgId}`
);
const affected = res.rowCount ?? 0;rowCount is the standard place node-postgres exposes affected rows. (node-postgres.com)
Note (don’t mix adapters)
If you were using the postgres-js adapter instead of node-postgres, the update result can be a postgres.RowList with a .count property. (answeroverflow.com)
Citations:
- 1: https://node-postgres.com/apis/result?utm_source=openai
- 2: https://orm.drizzle.team/docs/update?utm_source=openai
- 3: https://node-postgres.com/apis/result?utm_source=openai
- 4: https://www.answeroverflow.com/m/1196185391930290226?utm_source=openai
🏁 Script executed:
# Check the actual file to see the full context
cat -n apps/worker/src/services/project-stats-aggregator.ts | sed -n '310,340p'Repository: theopenco/llmgateway
Length of output: 1337
🏁 Script executed:
# Search for other update() calls in the codebase to see if there's a pattern
rg "\.update\(" apps/worker/src --type ts -A 3 | head -40Repository: theopenco/llmgateway
Length of output: 2074
Bug: update() without .returning() does not return affected row count—Array.isArray(result) will always be false.
With drizzle-orm/node-postgres (your driver), update().set().where() returns a pg Result object, not an array. The code at line 331 treats it as an array, so affectedRows is always 0, and logsMarked in the return value is perpetually zero.
Add .returning() to get affected rows in a type-safe way:
🐛 Proposed fix
const result = await database
.update(log)
.set({ statsAggregatedAt: now })
.where(
and(
sql`${log.projectId} = ${bucket.projectId}`,
gte(log.createdAt, hourTimestamp),
lt(log.createdAt, hourEnd),
isNull(log.statsAggregatedAt),
),
- );
-
- // Count affected rows (drizzle returns the updated rows)
- const affectedRows = Array.isArray(result) ? result.length : 0;
+ )
+ .returning({ id: log.id });
+
+ const affectedRows = result.length;
totalLogsMarked += affectedRows;📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| const result = await database | |
| .update(log) | |
| .set({ statsAggregatedAt: now }) | |
| .where( | |
| and( | |
| sql`${log.projectId} = ${bucket.projectId}`, | |
| gte(log.createdAt, hourTimestamp), | |
| lt(log.createdAt, hourEnd), | |
| isNull(log.statsAggregatedAt), | |
| ), | |
| ); | |
| // Count affected rows (drizzle returns the updated rows) | |
| const affectedRows = Array.isArray(result) ? result.length : 0; | |
| totalLogsMarked += affectedRows; | |
| const result = await database | |
| .update(log) | |
| .set({ statsAggregatedAt: now }) | |
| .where( | |
| and( | |
| sql`${log.projectId} = ${bucket.projectId}`, | |
| gte(log.createdAt, hourTimestamp), | |
| lt(log.createdAt, hourEnd), | |
| isNull(log.statsAggregatedAt), | |
| ), | |
| ) | |
| .returning({ id: log.id }); | |
| const affectedRows = result.length; | |
| totalLogsMarked += affectedRows; |
🤖 Prompt for AI Agents
In `@apps/worker/src/services/project-stats-aggregator.ts` around lines 318 - 332,
The update call is incorrectly assuming result is an array; change the
database.update(...).set(...).where(...) call on the log table to include a
.returning() (e.g., .returning({ id: log.id })) so it returns the updated rows
and then compute affectedRows as the returned array's length (or if you prefer
the pg Result use result.rowCount) and add that to totalLogsMarked; update the
code around the result variable and totalLogsMarked to use the returned rows
from .returning() (or result.rowCount) instead of Array.isArray(result).
- Add apiKeyHourlyStats table for per-API-key aggregations
- Add detailed breakdown fields to all aggregation tables:
- streamedCount, nonStreamedCount
- completedCount, lengthLimitCount, contentFilterCount, toolCallsCount,
canceledCount, unknownFinishCount (unified finish reasons)
- clientErrorCount, gatewayErrorCount, upstreamErrorCount (error types)
- reasoningTokens, cachedTokens
- requestCost, dataStorageCost, serviceFee, discountSavings
- Update activity route to use apiKeyHourlyStats when filtering by API key
- Refactor aggregator to use shared getCommonAggregationFields() helper
This enables fast dashboard queries with API key filtering and detailed
breakdown analytics without falling back to raw log table scans.
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In `@apps/api/src/routes/activity.ts`:
- Around line 250-312: The DATE(projectHourlyStats.hourTimestamp) expression
depends on the DB session timezone and can mis-group days; update the select,
groupBy and orderBy usages that reference projectHourlyStats.hourTimestamp (and
the equivalent in the apiKeyId branch) to convert the timestamp to UTC before
taking the date — i.e., apply AT TIME ZONE 'UTC' to
projectHourlyStats.hourTimestamp and then use DATE(...) for grouping/ordering,
or alternatively set the PostgreSQL session timezone to 'UTC' at query start;
ensure all occurrences (hourlyAggregates select, groupBy and orderBy) are
updated consistently.
In `@apps/worker/src/services/project-stats-aggregator.ts`:
- Around line 44-100: The errorCount in getCommonAggregationFields is computed
from log.hasError while clientErrorCount/gatewayErrorCount/upstreamErrorCount
are computed from log.unifiedFinishReason, causing mismatches; either
(preferred) align the error-type counts to require hasError as well (e.g.,
change clientErrorCount/gatewayErrorCount/upstreamErrorCount to sum CASE WHEN
${log.hasError} = true AND ${log.unifiedFinishReason} = 'client_error' THEN 1
ELSE 0 END, etc.), or if the difference is intentional add a concise clarifying
comment next to errorCount and the three error-type fields explaining the
differing sources and why they may not sum.
🧹 Nitpick comments (2)
packages/db/src/schema.ts (1)
1182-1192: Redundant index duplicates the unique constraint.The
unique().on(table.projectId, table.hourTimestamp)at line 1184 already creates an implicit B-tree index. The explicit index at lines 1186-1189 on the same columns is redundant. Same applies toprojectHourlyModelStats(lines 1243-1253) andapiKeyHourlyStats(lines 1307-1313).♻️ Proposed fix (remove redundant indexes from all three tables)
(table) => [ // Unique constraint for one record per project-hour unique().on(table.projectId, table.hourTimestamp), - // Index for dashboard queries (project + time range) - index("project_hourly_stats_project_id_hour_timestamp_idx").on( - table.projectId, - table.hourTimestamp, - ), // Index for worker refresh queries (find hours to update) index("project_hourly_stats_hour_timestamp_idx").on(table.hourTimestamp), ],Apply the same pattern to
projectHourlyModelStatsandapiKeyHourlyStats.apps/worker/src/services/project-stats-aggregator.ts (1)
376-409:refreshCurrentHourStatsscans all projects with any current-hour logs.Line 390 queries
WHERE createdAt >= currentHourStartwithout a project filter — this scans the entirelogtable for the current hour across all projects. For a platform with many projects, the initialSELECT DISTINCT projectIdis fine (it's bounded), but each subsequentrecalculate*call (lines 394-396) independently re-scans the same hour's logs for each project (3 queries × N projects).This is acceptable at moderate scale, but consider combining the three recalculation queries per project into a single log scan if this becomes a hot path.
| // Use aggregation tables for fast queries (when not filtering by apiKeyId) | ||
| // Query hourly aggregated data from projectHourlyStats table | ||
| const hourlyAggregates = await db | ||
| .select({ | ||
| date: sql<string>`DATE(${tables.log.createdAt})`.as("date"), | ||
| requestCount: sql<number>`COUNT(*)`.as("requestCount"), | ||
| date: sql<string>`DATE(${projectHourlyStats.hourTimestamp})`.as("date"), | ||
| requestCount: | ||
| sql<number>`COALESCE(SUM(${projectHourlyStats.requestCount}), 0)`.as( | ||
| "requestCount", | ||
| ), | ||
| inputTokens: | ||
| sql<number>`COALESCE(SUM(CAST(${tables.log.promptTokens} AS NUMERIC)), 0)`.as( | ||
| sql<number>`COALESCE(SUM(CAST(${projectHourlyStats.inputTokens} AS NUMERIC)), 0)`.as( | ||
| "inputTokens", | ||
| ), | ||
| outputTokens: | ||
| sql<number>`COALESCE(SUM(CAST(${tables.log.completionTokens} AS NUMERIC)), 0)`.as( | ||
| sql<number>`COALESCE(SUM(CAST(${projectHourlyStats.outputTokens} AS NUMERIC)), 0)`.as( | ||
| "outputTokens", | ||
| ), | ||
| totalTokens: | ||
| sql<number>`COALESCE(SUM(CAST(${tables.log.totalTokens} AS NUMERIC)), 0)`.as( | ||
| sql<number>`COALESCE(SUM(CAST(${projectHourlyStats.totalTokens} AS NUMERIC)), 0)`.as( | ||
| "totalTokens", | ||
| ), | ||
| cost: sql<number>`COALESCE(SUM(${tables.log.cost}), 0)`.as("cost"), | ||
| inputCost: sql<number>`COALESCE(SUM(${tables.log.inputCost}), 0)`.as( | ||
| "inputCost", | ||
| ), | ||
| outputCost: sql<number>`COALESCE(SUM(${tables.log.outputCost}), 0)`.as( | ||
| "outputCost", | ||
| ), | ||
| requestCost: sql<number>`COALESCE(SUM(${tables.log.requestCost}), 0)`.as( | ||
| "requestCost", | ||
| cost: sql<number>`COALESCE(SUM(${projectHourlyStats.cost}), 0)`.as( | ||
| "cost", | ||
| ), | ||
| inputCost: | ||
| sql<number>`COALESCE(SUM(${projectHourlyStats.inputCost}), 0)`.as( | ||
| "inputCost", | ||
| ), | ||
| outputCost: | ||
| sql<number>`COALESCE(SUM(${projectHourlyStats.outputCost}), 0)`.as( | ||
| "outputCost", | ||
| ), | ||
| requestCost: | ||
| sql<number>`COALESCE(SUM(${projectHourlyStats.requestCost}), 0)`.as( | ||
| "requestCost", | ||
| ), | ||
| dataStorageCost: | ||
| sql<number>`COALESCE(SUM(${tables.log.dataStorageCost}), 0)`.as( | ||
| sql<number>`COALESCE(SUM(${projectHourlyStats.dataStorageCost}), 0)`.as( | ||
| "dataStorageCost", | ||
| ), | ||
| errorCount: | ||
| sql<number>`SUM(CASE WHEN ${tables.log.hasError} = true THEN 1 ELSE 0 END)`.as( | ||
| sql<number>`COALESCE(SUM(${projectHourlyStats.errorCount}), 0)`.as( | ||
| "errorCount", | ||
| ), | ||
| cacheCount: | ||
| sql<number>`SUM(CASE WHEN ${tables.log.cached} = true THEN 1 ELSE 0 END)`.as( | ||
| sql<number>`COALESCE(SUM(${projectHourlyStats.cacheCount}), 0)`.as( | ||
| "cacheCount", | ||
| ), | ||
| discountSavings: sql<number>`COALESCE( | ||
| SUM( | ||
| CASE | ||
| WHEN ${tables.log.discount} > 0 AND ${tables.log.discount} < 1 | ||
| THEN ${tables.log.cost} * ${tables.log.discount} / (1 - ${tables.log.discount}) | ||
| ELSE 0 | ||
| END | ||
| discountSavings: | ||
| sql<number>`COALESCE(SUM(${projectHourlyStats.discountSavings}), 0)`.as( | ||
| "discountSavings", | ||
| ), | ||
| 0 | ||
| )`.as("discountSavings"), | ||
| }) | ||
| .from(tables.log) | ||
| .from(projectHourlyStats) | ||
| .where( | ||
| and( | ||
| inArray(tables.log.projectId, projectIds), | ||
| gte(tables.log.createdAt, startDate), | ||
| lte(tables.log.createdAt, endDate), | ||
| ...(apiKeyId ? [eq(tables.log.apiKeyId, apiKeyId)] : []), | ||
| inArray(projectHourlyStats.projectId, projectIds), | ||
| gte(projectHourlyStats.hourTimestamp, startDate), | ||
| lte(projectHourlyStats.hourTimestamp, endDate), | ||
| ), | ||
| ) | ||
| .groupBy(sql`DATE(${tables.log.createdAt})`) | ||
| .orderBy(sql`DATE(${tables.log.createdAt}) ASC`); | ||
| .groupBy(sql`DATE(${projectHourlyStats.hourTimestamp})`) | ||
| .orderBy(sql`DATE(${projectHourlyStats.hourTimestamp}) ASC`); |
There was a problem hiding this comment.
DATE() function depends on the PostgreSQL timezone session setting.
DATE(${projectHourlyStats.hourTimestamp}) (and the same in the apiKeyId branch) uses the database session's timezone to convert timestamp to date. If hourTimestamp stores UTC values but the DB session timezone differs, daily grouping will split at the wrong boundary. Since the timestamp column (without timezone) is being used, the DATE() result depends on the server's timezone setting.
Ensure the PostgreSQL timezone is set to 'UTC' (or the desired dashboard timezone) at session or server level, or use DATE(${...} AT TIME ZONE 'UTC') explicitly.
🤖 Prompt for AI Agents
In `@apps/api/src/routes/activity.ts` around lines 250 - 312, The
DATE(projectHourlyStats.hourTimestamp) expression depends on the DB session
timezone and can mis-group days; update the select, groupBy and orderBy usages
that reference projectHourlyStats.hourTimestamp (and the equivalent in the
apiKeyId branch) to convert the timestamp to UTC before taking the date — i.e.,
apply AT TIME ZONE 'UTC' to projectHourlyStats.hourTimestamp and then use
DATE(...) for grouping/ordering, or alternatively set the PostgreSQL session
timezone to 'UTC' at query start; ensure all occurrences (hourlyAggregates
select, groupBy and orderBy) are updated consistently.
| function getCommonAggregationFields() { | ||
| return { | ||
| requestCount: sql<number>`count(*)::int`.as("requestCount"), | ||
| errorCount: | ||
| sql<number>`sum(case when ${log.hasError} = true then 1 else 0 end)::int`.as( | ||
| "errorCount", | ||
| ), | ||
| cacheCount: | ||
| sql<number>`sum(case when ${log.cached} = true then 1 else 0 end)::int`.as( | ||
| "cacheCount", | ||
| ), | ||
| streamedCount: | ||
| sql<number>`sum(case when ${log.streamed} = true then 1 else 0 end)::int`.as( | ||
| "streamedCount", | ||
| ), | ||
| nonStreamedCount: | ||
| sql<number>`sum(case when ${log.streamed} = false or ${log.streamed} is null then 1 else 0 end)::int`.as( | ||
| "nonStreamedCount", | ||
| ), | ||
| // Unified finish reason counts | ||
| completedCount: | ||
| sql<number>`sum(case when ${log.unifiedFinishReason} = 'completed' then 1 else 0 end)::int`.as( | ||
| "completedCount", | ||
| ), | ||
| lengthLimitCount: | ||
| sql<number>`sum(case when ${log.unifiedFinishReason} = 'length_limit' then 1 else 0 end)::int`.as( | ||
| "lengthLimitCount", | ||
| ), | ||
| contentFilterCount: | ||
| sql<number>`sum(case when ${log.unifiedFinishReason} = 'content_filter' then 1 else 0 end)::int`.as( | ||
| "contentFilterCount", | ||
| ), | ||
| toolCallsCount: | ||
| sql<number>`sum(case when ${log.unifiedFinishReason} = 'tool_calls' then 1 else 0 end)::int`.as( | ||
| "toolCallsCount", | ||
| ), | ||
| canceledCount: | ||
| sql<number>`sum(case when ${log.unifiedFinishReason} = 'canceled' then 1 else 0 end)::int`.as( | ||
| "canceledCount", | ||
| ), | ||
| unknownFinishCount: | ||
| sql<number>`sum(case when ${log.unifiedFinishReason} = 'unknown' or ${log.unifiedFinishReason} is null then 1 else 0 end)::int`.as( | ||
| "unknownFinishCount", | ||
| ), | ||
| // Error type counts | ||
| clientErrorCount: | ||
| sql<number>`sum(case when ${log.unifiedFinishReason} = 'client_error' then 1 else 0 end)::int`.as( | ||
| "clientErrorCount", | ||
| ), | ||
| gatewayErrorCount: | ||
| sql<number>`sum(case when ${log.unifiedFinishReason} = 'gateway_error' then 1 else 0 end)::int`.as( | ||
| "gatewayErrorCount", | ||
| ), | ||
| upstreamErrorCount: | ||
| sql<number>`sum(case when ${log.unifiedFinishReason} = 'upstream_error' then 1 else 0 end)::int`.as( | ||
| "upstreamErrorCount", | ||
| ), |
There was a problem hiding this comment.
errorCount and error-type counts are derived from different source columns — potential mismatch.
errorCount (line 48) counts rows where log.hasError = true, while clientErrorCount, gatewayErrorCount, and upstreamErrorCount (lines 89-100) count based on log.unifiedFinishReason. These two classification mechanisms may not be consistent — a log with hasError = true might have a unifiedFinishReason that doesn't match any of the three error categories (e.g., 'unknown'), and vice versa.
This means clientErrorCount + gatewayErrorCount + upstreamErrorCount may not equal errorCount. If this is intentional, consider adding a comment to clarify. If not, align the error-type counts to also use hasError combined with an error classification field.
🤖 Prompt for AI Agents
In `@apps/worker/src/services/project-stats-aggregator.ts` around lines 44 - 100,
The errorCount in getCommonAggregationFields is computed from log.hasError while
clientErrorCount/gatewayErrorCount/upstreamErrorCount are computed from
log.unifiedFinishReason, causing mismatches; either (preferred) align the
error-type counts to require hasError as well (e.g., change
clientErrorCount/gatewayErrorCount/upstreamErrorCount to sum CASE WHEN
${log.hasError} = true AND ${log.unifiedFinishReason} = 'client_error' THEN 1
ELSE 0 END, etc.), or if the difference is intentional add a concise clarifying
comment next to errorCount and the three error-type fields explaining the
differing sources and why they may not sum.
The activity endpoint now reads from aggregation tables instead of raw log table. Updated tests to: - Add aggregateLogsForTesting() helper that mimics worker aggregation - Clean up aggregation tables in deleteAll() - Call aggregateLogsForTesting() after inserting log data in each test Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
apps/api/src/routes/activity.spec.ts (1)
259-267:⚠️ Potential issue | 🟠 MajorClear aggregation stats before re-aggregating in mid-test scenarios—stale data from other projects will pollute test results.
Multiple tests (token-count aggregation, error-rate, cache-rate, discount, cost-breakdown, model-breakdown, and others) delete the log table and insert fresh test data, but leave the hourly stats tables untouched. Since
aggregateLogsForTesting()only upserts stats for log buckets that exist in the new logs, stale rows frombeforeEachremain in the database. Because the activity endpoint aggregates stats across all projects the user owns (when noprojectIdfilter is applied), these orphaned rows fromtest-project-id-2will inflate request counts, token counts, and other metrics in the test assertions, causing test failures.Add cleanup before re-aggregating:
Suggested fix (apply in every test that deletes logs mid-test)
await db.delete(tables.log); + await db.delete(projectHourlyStats); + await db.delete(projectHourlyModelStats); + await db.delete(apiKeyHourlyStats);Or extract a small helper like
clearAllStats()to keep it DRY.
🤖 Fix all issues with AI agents
In `@apps/api/src/testing.ts`:
- Around line 48-61: roundToHourStart currently uses local-time getters
(getFullYear/getMonth/getDate/getHours) which can diverge from Postgres
date_trunc behavior (UTC); change it to construct the truncated date in UTC by
using the UTC getters (getUTCFullYear, getUTCMonth, getUTCDate, getUTCHours) and
create the Date via Date.UTC(...) (or new Date(Date.UTC(...))) so the hour
boundary matches DB truncation.
🧹 Nitpick comments (2)
package.json (1)
22-22:pushis nearly identical to the existingsyncscript.Line 27 already defines
"sync": "pnpm push-dev; pnpm push-test". The only difference is&&(fail-fast) vs;(run both regardless). Consider whether both are needed, or ifsyncshould simply be updated to use&&to avoid confusion.apps/api/src/testing.ts (1)
175-303:aggregateLogsForTestingprocesses buckets sequentially — consider also noting the N+1 query pattern.Each bucket issues 3 separate SELECT queries plus up to 3 INSERT/UPSERT operations (project stats, model stats, API key stats). For testing with small datasets this is fine, but the function mirrors the production worker logic, so if the worker uses the same pattern it could be slow at scale.
More importantly for correctness: this function only upserts stats for hours that have logs. Tests that call
db.delete(tables.log)mid-test (seeactivity.spec.ts) leave behind stale aggregation rows from thebeforeEachphase that won't be overwritten. See my comment on the spec file.
| /** | ||
| * Helper function to round a date to the start of its hour | ||
| */ | ||
| function roundToHourStart(date: Date): Date { | ||
| return new Date( | ||
| date.getFullYear(), | ||
| date.getMonth(), | ||
| date.getDate(), | ||
| date.getHours(), | ||
| 0, | ||
| 0, | ||
| 0, | ||
| ); | ||
| } |
There was a problem hiding this comment.
roundToHourStart uses local timezone — may diverge from DB's date_trunc (UTC).
date_trunc('hour', ...) in Postgres operates in the session timezone (default UTC), but roundToHourStart truncates using the JS runtime's local timezone. If tests run in a non-UTC environment, the hour boundary can shift, causing mismatched keys and duplicate/missing stats rows.
For a testing-only helper this is low-risk (CI is likely UTC), but worth noting.
Suggested fix
function roundToHourStart(date: Date): Date {
- return new Date(
- date.getFullYear(),
- date.getMonth(),
- date.getDate(),
- date.getHours(),
- 0,
- 0,
- 0,
- );
+ const ms = date.getTime();
+ return new Date(ms - (ms % (60 * 60 * 1000)));
}This truncates in UTC regardless of the local timezone.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| /** | |
| * Helper function to round a date to the start of its hour | |
| */ | |
| function roundToHourStart(date: Date): Date { | |
| return new Date( | |
| date.getFullYear(), | |
| date.getMonth(), | |
| date.getDate(), | |
| date.getHours(), | |
| 0, | |
| 0, | |
| 0, | |
| ); | |
| } | |
| /** | |
| * Helper function to round a date to the start of its hour | |
| */ | |
| function roundToHourStart(date: Date): Date { | |
| const ms = date.getTime(); | |
| return new Date(ms - (ms % (60 * 60 * 1000))); | |
| } |
🤖 Prompt for AI Agents
In `@apps/api/src/testing.ts` around lines 48 - 61, roundToHourStart currently
uses local-time getters (getFullYear/getMonth/getDate/getHours) which can
diverge from Postgres date_trunc behavior (UTC); change it to construct the
truncated date in UTC by using the UTC getters (getUTCFullYear, getUTCMonth,
getUTCDate, getUTCHours) and create the Date via Date.UTC(...) (or new
Date(Date.UTC(...))) so the hour boundary matches DB truncation.
Adds a one-off script to bulk insert random logs for testing dashboards and visualizations locally. Features: - Weighted random distribution across 7 popular models - Realistic token counts, costs, and durations - Configurable number of logs and date range - Batched inserts for performance Usage: npx tsx scripts/generate-test-logs.ts <count> <projectId> <apiKeyId> <orgId> [daysBack] Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
The shortid function wasn't being resolved correctly from the db package exports. Use nanoid directly with the same alphabet. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Move the script to packages/db where it can properly resolve local imports. Add npm script for easy execution. Usage: pnpm --filter @llmgateway/db generate-test-logs 1000 proj key org Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
…ities Move generate-test-logs to a new @llmgateway/scripts package that can properly resolve workspace dependencies. This allows scripts to import from any package in the monorepo. Usage: pnpm --filter @llmgateway/scripts generate-test-logs 1000 proj key org Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Update generate-test-logs to apply discounts to smaller/cheaper models: - gpt-4o-mini: 30% discount - claude-3-5-haiku: 30% discount - gemini-2.5-flash: 30% discount - deepseek-chat: 30% discount The discount field is now included in generated logs, which allows the aggregator to calculate discountSavings correctly. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Consolidates all TypeScript utility scripts into the dedicated scripts package for proper workspace dependency resolution: - api-ping: API health monitoring with SQLite persistence - export-models-dev: Export models to models.dev TOML format - generate-test-logs: Generate random test logs for dashboards - model-release-dates: Query model release dates via Perplexity - send: Bulk email sender via Resend - test-gemini-tools-*: Gemini tool calling tests Usage: pnpm --filter @llmgateway/scripts <script-name> Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
…down
- Add apiKeyHourlyModelStats aggregation table with composite key on
(apiKeyId, hourTimestamp, usedModel, usedProvider)
- Update project-stats-aggregator to populate the new table
- Update testing helper to clean up and aggregate to the new table
- Address PR feedback:
- Remove redundant index from projectHourlyStats (unique constraint
already creates implicit B-tree index)
- Fix UTC timezone alignment in roundToHourStart to match PostgreSQL's
date_trunc behavior
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
…loading
- Refactor admin API /organizations/{orgId} endpoint to query from
projectHourlyStats and projectHourlyModelStats instead of scanning
the log table directly
- Remove "Load Usage Data" button and lazy-loading logic since queries
are now fast due to pre-aggregated data
- Auto-load metrics on component mount
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
# Conflicts: # packages/db/migrations/meta/_journal.json # packages/db/src/schema.ts
- Add imageInputCost and imageOutputCost to all 4 aggregation tables (projectHourlyStats, projectHourlyModelStats, apiKeyHourlyStats, apiKeyHourlyModelStats) to match new fields in log table from main - Update aggregator to use time-based lookback (processRecentLogs) instead of statsAggregatedAt field which was removed from main - Add image cost fields to activity.ts for both apiKeyId and project paths - Add image cost fields to testing.ts helper - Regenerate migrations after syncing with origin/main Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
b16f619 to
7241529
Compare
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
…way into sync-stats-aggregations
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
- Replace fixed 24h lookback with two-phase processing: Phase 1 (backfill): finds unprocessed buckets via LEFT JOIN Phase 2 (stale): detects buckets with new logs via HAVING - Add per-API-key model breakdown using apiKeyHourlyModelStats - Improve aggregation logging with [backfill]/[stale] prefixes Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Add credits/api-keys breakdowns for request count, cost, service fee, and data storage cost to all 4 aggregation tables and the activity API. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
931cc28 to
1eb6d36
Compare
steebchen
changed the title
Avoid the pg driver's local-timezone interpretation of `timestamp without timezone` columns by formatting timestamps as UTC strings and casting via ::timestamp in SQL. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Summary
projectHourlyStats,projectHourlyModelStats,apiKeyHourlyStats,apiKeyHourlyModelStats) for fast dashboard queries instead of scanning the rawlogtablecreditsvsapi-keys) breakdowns for request count, cost, service fee, and data storage costSTATS_BATCH_SIZE,STATS_BACKFILL_ENABLED/DAYS,STATS_STALE_ENABLED/DAYS,PROJECT_STATS_REFRESH_INTERVAL_SECONDSTest plan
pnpm build:corepassespnpm test:unitpasses (449 tests)