fix(stats): better stat table periods, provider/model norm

packages/stats/core/src/stat-sync.ts

@@ -6,8 +6,10 @@ import { GeoStatRepo, rowsFromAggregates as geoRowsFromAggregates } from "./doma
 import { buildStatsQuery, toGeoAggregate, toModelAggregate, toProviderAggregate } from "./domain/inference"
 import { ModelStatRepo, rowsFromAggregates as modelRowsFromAggregates } from "./domain/model"
 import { ProviderStatRepo, rowsFromAggregates as providerRowsFromAggregates } from "./domain/provider"
+import { startOfIsoWeek } from "./domain/stat"
 
 const DATALAKE_INGESTION_LAG_MS = 5 * 60_000
+const WEEK_MS = 7 * 86_400_000
 
 export type SyncStatsResult = { ok: true; rows: number; startedAt: string; periodStart: string; periodEnd: string }
 export type SyncStatsError = AthenaQueryError | AthenaQueryTimeoutError | DatabaseError
@@ -19,9 +21,7 @@ export const syncStats: () => Effect.Effect<
 > = Effect.fn("StatSync.sync")(function* () {
   const startedAt = yield* DateTime.nowAsDate
   const periodEnd = new Date(Math.floor((startedAt.getTime() - DATALAKE_INGESTION_LAG_MS) / 60_000) * 60_000)
-  const periodStart = new Date(
-    Date.UTC(periodEnd.getUTCFullYear(), periodEnd.getUTCMonth(), periodEnd.getUTCDate() - 6),
-  )
+  const periodStart = new Date(startOfIsoWeek(periodEnd).getTime() - WEEK_MS)
   const athena = yield* Athena
   const modelStats = yield* ModelStatRepo
   const providerStats = yield* ProviderStatRepo
@@ -29,7 +29,7 @@ export const syncStats: () => Effect.Effect<
 
   yield* logRuntimeCheck()
 
-  const [modelAggregates, providerAggregates, geoAggregates] = yield* Effect.all(
+  const [modelAggregates, providerAggregates, geoAggregates, geoModelAggregates] = yield* Effect.all(
     [
       athena
         .query(buildStatsQuery(periodStart, periodEnd, "model"))
@@ -40,12 +40,15 @@ export const syncStats: () => Effect.Effect<
       athena
         .query(buildStatsQuery(periodStart, periodEnd, "geo"))
         .pipe(Effect.map((rows) => rows.flatMap(toGeoAggregate))),
+      athena
+        .query(buildStatsQuery(periodStart, periodEnd, "geo_model"))
+        .pipe(Effect.map((rows) => rows.flatMap(toGeoAggregate))),
     ],
     { concurrency: "unbounded" },
   )
   const modelRows = modelRowsFromAggregates(modelAggregates)
   const providerRows = providerRowsFromAggregates(providerAggregates)
-  const geoRows = geoRowsFromAggregates(geoAggregates)
+  const geoRows = geoRowsFromAggregates([...geoAggregates, ...geoModelAggregates])
 
   yield* Effect.all([modelStats.upsert(modelRows), providerStats.upsert(providerRows), geoStats.upsert(geoRows)], {
     concurrency: "unbounded",