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) => {