From e87b6165af6b795f5c19b5e1024054b957dc33da Mon Sep 17 00:00:00 2001 From: Luca Steeb Date: Wed, 12 Aug 2026 01:16:46 +0200 Subject: [PATCH] feat(worker): rebuild global stats from logs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The mode/kind migration (1786136603_warm_omega_flight.sql) left a large slice of global_model_stats / global_source_stats as used_mode and org_kind 'unknown', so the admin Global Stats page cannot make Credits + BYOK add up to All and shows almost no organization-kind attribution. Rebuilding from the source data fixes it exactly, because data retention never deletes log rows — cleanupExpiredLogData only nulls the verbose payload columns (messages, content, raw/upstream request and response, tools, customHeaders, userAgent, responsesApiData). Every column the aggregator reads survives, so the rebuild is lossless and needs no proration or scaling. Each day goes through the aggregator's own recomputeDayFully(), which deletes that day and re-aggregates its 24 hours, so the script is idempotent and resumable. Newest day first, so the recent days people actually look at are correct within minutes rather than at the end of the run, and the table is never fully empty. Today and yesterday are skipped: the incremental walker owns today and the safety net owns yesterday, and racing either risks interleaving a delete with an insert. Dry run by default; --commit to apply. Co-Authored-By: Claude Opus 5 (1M context) --- apps/worker/package.json | 1 + .../src/scripts/rebuild-global-stats.ts | 180 ++++++++++++++++++ .../src/services/global-stats-aggregator.ts | 2 +- 3 files changed, 182 insertions(+), 1 deletion(-) create mode 100644 apps/worker/src/scripts/rebuild-global-stats.ts diff --git a/apps/worker/package.json b/apps/worker/package.json index 4c93a9192f..71e3783bae 100644 --- a/apps/worker/package.json +++ b/apps/worker/package.json @@ -20,6 +20,7 @@ "dev": "concurrently \"tsc-watch --onSuccess 'resolve-tspaths'\" \"nodemon --exec node --enable-source-maps --env-file=../../.env dist/index.js\"", "format": "eslint --fix . && prettier --write .", "lint": "eslint . && prettier --check .", + "rebuild-global-stats": "tsx --env-file=../../.env ./src/scripts/rebuild-global-stats.ts", "start": "node --enable-source-maps dist/index.js" }, "dependencies": { diff --git a/apps/worker/src/scripts/rebuild-global-stats.ts b/apps/worker/src/scripts/rebuild-global-stats.ts new file mode 100644 index 0000000000..3bc6d11356 --- /dev/null +++ b/apps/worker/src/scripts/rebuild-global-stats.ts @@ -0,0 +1,180 @@ +/* eslint-disable no-console */ +/** + * Rebuild global_model_stats / global_source_stats from `log`. + * + * Why this is possible at all: data retention never deletes log rows. The + * cleanup job (cleanupExpiredLogData) only nulls the verbose payload columns — + * messages, content, raw/upstream request and response, tools, customHeaders, + * userAgent, responsesApiData — and sets data_retention_cleaned_up. Every column + * the aggregator reads (used_model, used_provider, used_mode, source, + * organization_id, created_at, the token columns, the cost columns, has_error, + * cached, streamed, unified_finish_reason) survives untouched, so a rebuild is + * lossless. + * + * That makes this the correct fix for rows the mode/kind migration + * (1786136603_warm_omega_flight.sql) left as 'unknown': re-deriving from the + * source data gives an exact used_mode and org_kind for every row, with no + * proration and no scaling. + * + * Each day is handled by the aggregator's own recomputeDayFully(), which deletes + * that day's rows and re-aggregates its 24 hours. So the script is idempotent + * and resumable: re-running any day is safe, and a crashed run is continued by + * re-running with --from/--to for whatever is left. + * + * Ordering: newest day first by default, so the recent days people actually look + * at are correct within minutes instead of at the very end of the run. + * + * Today and yesterday are skipped. The incremental walker owns today, and the + * safety net wipes-and-recomputes yesterday; racing either of them risks + * interleaving a delete with an insert and double-counting a day. + * + * Usage: + * pnpm --filter worker rebuild-global-stats # dry run: report the plan + * pnpm --filter worker rebuild-global-stats --commit # rebuild everything + * pnpm --filter worker rebuild-global-stats --commit --limit=7 # rebuild the 7 newest days first + * pnpm --filter worker rebuild-global-stats --commit --from=2025-05-20 --to=2026-08-01 + * pnpm --filter worker rebuild-global-stats --commit --oldest-first + * + * Environment: + * DATABASE_URL - defaults to local postgres if unset + */ + +import { recomputeDayFully } from "@/services/global-stats-aggregator.js"; + +import { db, sql } from "@llmgateway/db"; + +const DAY_MS = 24 * 60 * 60 * 1000; + +function parseFlag(name: string): string | undefined { + const flag = `--${name}=`; + const arg = process.argv.find((a) => a.startsWith(flag)); + return arg ? arg.slice(flag.length) : undefined; +} + +function hasFlag(name: string): boolean { + return process.argv.includes(`--${name}`); +} + +function floorToUTCDay(d: Date): Date { + return new Date( + Date.UTC(d.getUTCFullYear(), d.getUTCMonth(), d.getUTCDate(), 0, 0, 0, 0), + ); +} + +function toDateString(d: Date): string { + return d.toISOString().slice(0, 10); +} + +/** + * Earliest day worth rebuilding: whichever of the first log and the first + * existing stats row is older, so a rebuild can also fill in days the walker + * never reached. + * + * `order by created_at limit 1` rather than `min(created_at)` — the log table is + * huge, and this form is guaranteed to ride the index on (created_at, ...) + * instead of risking a sequential scan. + */ +async function resolveStart(): Promise { + const result = await db.execute(sql` + select least( + (select created_at from log order by created_at limit 1), + (select min(day_timestamp) from global_model_stats), + (select min(day_timestamp) from global_source_stats) + ) as start + `); + const start = (result.rows[0] as { start: Date | string | null } | undefined) + ?.start; + return start ? floorToUTCDay(new Date(start)) : null; +} + +async function main(): Promise { + const commit = hasFlag("commit"); + const oldestFirst = hasFlag("oldest-first"); + const limit = parseFlag("limit") ? Number(parseFlag("limit")) : undefined; + + const fromFlag = parseFlag("from"); + const toFlag = parseFlag("to"); + + const today = floorToUTCDay(new Date()); + // Today belongs to the walker, yesterday to the safety net. + const defaultEnd = new Date(today.getTime() - 2 * DAY_MS); // eslint-disable-line no-mixed-operators + + const start = fromFlag + ? floorToUTCDay(new Date(`${fromFlag}T00:00:00Z`)) + : await resolveStart(); + if (!start) { + console.log("No logs and no global stats rows — nothing to rebuild."); + return; + } + + let end = toFlag + ? floorToUTCDay(new Date(`${toFlag}T00:00:00Z`)) + : defaultEnd; + if (end.getTime() > defaultEnd.getTime()) { + console.log( + `Clamping --to to ${toDateString(defaultEnd)}: today and yesterday are owned by the walker and the safety net.`, + ); + end = defaultEnd; + } + + if (end.getTime() < start.getTime()) { + console.log("Nothing to rebuild — the range is empty."); + return; + } + + const days: Date[] = []; + for (let d = new Date(start); d <= end; d = new Date(d.getTime() + DAY_MS)) { + days.push(new Date(d)); + } + if (!oldestFirst) { + days.reverse(); + } + const planned = limit ? days.slice(0, limit) : days; + + console.log( + `${commit ? "REBUILDING" : "DRY RUN"} — ${toDateString(start)} .. ${toDateString(end)} (${planned.length} of ${days.length} days, ${oldestFirst ? "oldest" : "newest"} first)`, + ); + + if (!commit) { + const before = await db.execute(sql` + select + (select count(*) from global_model_stats + where used_mode = 'unknown' or org_kind = 'unknown') as model_rows_unattributed, + (select count(*) from global_source_stats + where used_mode = 'unknown' or org_kind = 'unknown') as source_rows_unattributed + `); + console.log(before.rows[0]); + console.log("\nDry run — nothing was written. Re-run with --commit."); + return; + } + + const startedAt = Date.now(); + let done = 0; + for (const day of planned) { + const completed = await recomputeDayFully(day); + if (!completed) { + console.log(`Stop requested — halted after ${done} day(s).`); + break; + } + done++; + if (done % 10 === 0 || done === planned.length) { + const elapsed = (Date.now() - startedAt) / 1000; + const rate = done / elapsed; + const remaining = planned.length - done; + console.log( + `${done}/${planned.length} days (${toDateString(day)}) — ${elapsed.toFixed(0)}s elapsed, ~${rate > 0 ? Math.round(remaining / rate) : 0}s remaining`, + ); + } + } + + console.log( + `Rebuilt ${done} day(s) in ${((Date.now() - startedAt) / 1000).toFixed(0)}s.`, + ); +} + +main() + .then(() => process.exit(0)) + .catch((error) => { + console.error(error); + process.exit(1); + }); diff --git a/apps/worker/src/services/global-stats-aggregator.ts b/apps/worker/src/services/global-stats-aggregator.ts index 9262debc43..5bad1666d0 100644 --- a/apps/worker/src/services/global-stats-aggregator.ts +++ b/apps/worker/src/services/global-stats-aggregator.ts @@ -286,7 +286,7 @@ async function setLastSafetyNetDay(day: Date): Promise { // Returns true on full completion. Returns false if stop was requested // mid-walk; the caller must NOT mark the day as recomputed in that case so the // next worker start retries from the partially-recomputed state. -async function recomputeDayFully(day: Date): Promise { +export async function recomputeDayFully(day: Date): Promise { const dayStr = formatUTCTimestamp(day); await db.transaction(async (tx) => {