fix: logo 右半部分从 CODING 改为 CODE

去掉难以正确渲染的 N 和 G 字母，右半部分简化为 CODE（4 字母），
与左半部分 AIR 组合为 AIR CODE。

packages/stats/core/src/athena.ts

@@ -0,0 +1,139 @@
+import {
+  AthenaClient as AwsAthenaClient,
+  GetQueryExecutionCommand,
+  GetQueryResultsCommand,
+  StartQueryExecutionCommand,
+  type Row,
+} from "@aws-sdk/client-athena"
+import { Effect, Layer, Schema } from "effect"
+import * as Context from "effect/Context"
+import { Resource } from "sst/resource"
+
+const ATHENA_MAX_POLL_ATTEMPTS = 60
+const ATHENA_PAGE_SIZE = 1000
+
+export type AthenaData = Record<string, string>
+
+export class AthenaQueryError extends Schema.TaggedErrorClass<AthenaQueryError>()("AthenaQueryError", {
+  message: Schema.String,
+  queryExecutionId: Schema.optional(Schema.String),
+  cause: Schema.optional(Schema.Defect),
+}) {}
+
+export class AthenaQueryTimeoutError extends Schema.TaggedErrorClass<AthenaQueryTimeoutError>()(
+  "AthenaQueryTimeoutError",
+  {
+    message: Schema.String,
+    queryExecutionId: Schema.String,
+  },
+) {}
+
+export declare namespace Athena {
+  export interface Service {
+    readonly query: (query: string) => Effect.Effect<AthenaData[], AthenaQueryError | AthenaQueryTimeoutError>
+  }
+}
+
+export class Athena extends Context.Service<Athena, Athena.Service>()("@opencode/stats/Athena") {
+  static readonly layer: Layer.Layer<Athena> = Layer.effect(
+    Athena,
+    Effect.sync(() => {
+      const client = new AwsAthenaClient({ region: Resource.InferenceEvent.region })
+
+      const query = Effect.fn("Athena.query")(function* (query: string) {
+        const started = yield* Effect.tryPromise({
+          try: () =>
+            client.send(
+              new StartQueryExecutionCommand({
+                QueryString: query,
+                WorkGroup: Resource.InferenceEvent.workgroup,
+                QueryExecutionContext: {
+                  Catalog: Resource.InferenceEvent.catalog,
+                  Database: Resource.InferenceEvent.database,
+                },
+              }),
+            ),
+          catch: (cause) => new AthenaQueryError({ message: "Failed to start Athena stats query", cause }),
+        })
+        const queryExecutionId = started.QueryExecutionId
+        if (!queryExecutionId)
+          return yield* new AthenaQueryError({ message: "Athena did not return a query execution id" })
+
+        yield* poll(client, queryExecutionId)
+        return yield* results(client, queryExecutionId)
+      })
+
+      return Athena.of({ query })
+    }),
+  )
+}
+
+const poll: (
+  client: AwsAthenaClient,
+  queryExecutionId: string,
+  attempt?: number,
+) => Effect.Effect<void, AthenaQueryError | AthenaQueryTimeoutError> = Effect.fn("Athena.poll")(function* (
+  client: AwsAthenaClient,
+  queryExecutionId: string,
+  attempt = 0,
+) {
+  if (attempt > 0) yield* Effect.sleep("2 seconds")
+
+  const result = yield* Effect.tryPromise({
+    try: () => client.send(new GetQueryExecutionCommand({ QueryExecutionId: queryExecutionId })),
+    catch: (cause) => new AthenaQueryError({ message: "Failed to poll Athena stats query", queryExecutionId, cause }),
+  })
+  const status = result.QueryExecution?.Status
+
+  if (status?.State === "SUCCEEDED") return
+  if (status?.State === "FAILED" || status?.State === "CANCELLED")
+    return yield* new AthenaQueryError({
+      message: `Athena stats query ${status.State.toLowerCase()}: ${status.StateChangeReason ?? "unknown reason"}`,
+      queryExecutionId,
+    })
+
+  if (attempt >= ATHENA_MAX_POLL_ATTEMPTS - 1)
+    return yield* new AthenaQueryTimeoutError({
+      message: `Athena stats query ${queryExecutionId} did not complete`,
+      queryExecutionId,
+    })
+
+  return yield* poll(client, queryExecutionId, attempt + 1)
+})
+
+const results: (
+  client: AwsAthenaClient,
+  queryExecutionId: string,
+  nextToken?: string,
+) => Effect.Effect<AthenaData[], AthenaQueryError> = Effect.fn("Athena.results")(function* (
+  client: AwsAthenaClient,
+  queryExecutionId: string,
+  nextToken?: string,
+) {
+  const result = yield* Effect.tryPromise({
+    try: () =>
+      client.send(
+        new GetQueryResultsCommand({
+          QueryExecutionId: queryExecutionId,
+          NextToken: nextToken,
+          MaxResults: ATHENA_PAGE_SIZE,
+        }),
+      ),
+    catch: (cause) => new AthenaQueryError({ message: "Failed to read Athena stats results", queryExecutionId, cause }),
+  })
+  const columns = result.ResultSet?.ResultSetMetadata?.ColumnInfo?.map((item) => item.Name ?? "") ?? []
+  const rows = (result.ResultSet?.Rows ?? []).slice(nextToken ? 0 : 1).map((row) => rowData(columns, row))
+
+  if (!result.NextToken) return rows
+  return [...rows, ...(yield* results(client, queryExecutionId, result.NextToken))]
+})
+
+function rowData(columns: string[], row: Row): AthenaData {
+  return Object.fromEntries(
+    columns.flatMap((column, index) => {
+      const value = row.Data?.[index]?.VarCharValue
+      if (!column || value === undefined) return []
+      return [[column, value]]
+    }),
+  )
+}

packages/stats/core/src/config.ts

@@ -0,0 +1,23 @@
+import { Config, ConfigProvider, Effect, Layer, Schema } from "effect"
+import * as Context from "effect/Context"
+import { Resource } from "sst/resource"
+
+export class AppConfigValue extends Schema.Class<AppConfigValue>("AppConfigValue")({
+  stage: Schema.NonEmptyString,
+  publicUrl: Schema.NonEmptyString,
+}) {}
+
+const decodeAppConfigValue = Schema.decodeUnknownSync(AppConfigValue)
+
+const config = Config.all({
+  stage: Config.succeed(Resource.App.stage),
+  publicUrl: Config.string("PUBLIC_URL").pipe(Config.withDefault("http://localhost:3000")),
+}).pipe(Config.map(decodeAppConfigValue))
+
+export class AppConfig extends Context.Service<AppConfig, AppConfigValue>()("@opencode/stats/AppConfig") {
+  static readonly config = config
+  static readonly layer: Layer.Layer<AppConfig, never, never> = Layer.effect(
+    AppConfig,
+    config.parse(ConfigProvider.fromEnv()).pipe(Effect.orDie),
+  )
+}

packages/stats/core/src/database.ts

@@ -0,0 +1,79 @@
+import { Client } from "@planetscale/database"
+import { drizzle } from "drizzle-orm/planetscale-serverless"
+import { migrate as drizzleMigrate } from "drizzle-orm/planetscale-serverless/migrator"
+import { Config, ConfigProvider, Effect, Layer, Schema } from "effect"
+import * as Context from "effect/Context"
+import * as schema from "./database/schema"
+import { Resource } from "sst/resource"
+
+export const DatabaseUrl = Schema.NonEmptyString.pipe(Schema.brand("DatabaseUrl"))
+export type DatabaseUrl = typeof DatabaseUrl.Type
+
+export class DatabaseSettings extends Schema.Class<DatabaseSettings>("DatabaseSettings")({
+  url: DatabaseUrl,
+  migrationsDir: Schema.NonEmptyString,
+}) {}
+
+const decodeDatabaseSettings = Schema.decodeUnknownSync(DatabaseSettings)
+
+const config = Config.all({
+  url: Config.nonEmptyString("DATABASE_URL").pipe(Config.withDefault(Resource.StatsDatabase.url)),
+  migrationsDir: Config.nonEmptyString("DATABASE_MIGRATIONS_DIR").pipe(Config.withDefault("./migrations")),
+}).pipe(Config.map(decodeDatabaseSettings))
+
+export class DatabaseConfig extends Context.Service<DatabaseConfig, DatabaseSettings>()(
+  "@opencode/stats/DatabaseConfig",
+) {
+  static readonly config = config
+  static readonly layer: Layer.Layer<DatabaseConfig, never, never> = Layer.effect(
+    DatabaseConfig,
+    config.parse(ConfigProvider.fromEnv()).pipe(Effect.orDie),
+  )
+}
+
+function makeDrizzle(settings: DatabaseSettings) {
+  return drizzle({ client: new Client({ url: settings.url }), schema })
+}
+
+export type Drizzle = ReturnType<typeof makeDrizzle>
+
+export class DrizzleClient extends Context.Service<DrizzleClient, Drizzle>()("@opencode/stats/DrizzleClient") {
+  static readonly layer: Layer.Layer<DrizzleClient, never, DatabaseConfig> = Layer.effect(
+    DrizzleClient,
+    Effect.map(DatabaseConfig, makeDrizzle),
+  )
+}
+
+export class DatabaseError extends Schema.TaggedErrorClass<DatabaseError>()("DatabaseError", {
+  cause: Schema.Defect,
+}) {}
+
+export const catchDbError = Effect.mapError((cause) => DatabaseError.make({ cause }))
+
+export class MigrationError extends Schema.TaggedErrorClass<MigrationError>()("MigrationError", {
+  message: Schema.String,
+  cause: Schema.optional(Schema.Defect),
+}) {}
+
+export const migrate = Effect.fn("Database.migrate")(function* () {
+  const settings = yield* DatabaseConfig
+  yield* Effect.logInfo("applying database migrations").pipe(
+    Effect.annotateLogs({ migrationsDir: settings.migrationsDir }),
+  )
+  const result = yield* Effect.tryPromise({
+    try: () =>
+      drizzleMigrate(drizzle({ client: new Client({ url: settings.url }) }), {
+        migrationsFolder: settings.migrationsDir,
+      }),
+    catch: (cause) => new MigrationError({ message: "Failed to apply database migrations", cause }),
+  })
+  if (result)
+    return yield* new MigrationError({
+      message: `Failed to initialize database migrations: ${result.exitCode}`,
+    })
+  yield* Effect.logInfo("database migrations complete").pipe(
+    Effect.annotateLogs({ migrationsDir: settings.migrationsDir }),
+  )
+})
+
+export const layer = Layer.mergeAll(DatabaseConfig.layer, DrizzleClient.layer.pipe(Layer.provide(DatabaseConfig.layer)))

packages/stats/core/src/database/schema.ts

@@ -0,0 +1,160 @@
+import { bigint, char, datetime, decimal, index, int, mysqlTable, uniqueIndex, varchar } from "drizzle-orm/mysql-core"
+
+export const modelStat = mysqlTable(
+  "model_stat",
+  {
+    ...periodColumns(),
+    provider: varchar({ length: 128 }).notNull(),
+    model: varchar({ length: 256 }).notNull(),
+    provider_model: varchar({ length: 256 }).notNull().default(""),
+    ...metricColumns(),
+    rank_by_tokens: int(),
+    rank_by_requests: int(),
+    rank_by_cost: int(),
+    ...timestampColumns(),
+  },
+  (table) => [
+    uniqueIndex("uniq_model_period").on(
+      table.grain,
+      table.period_key,
+      table.dataset,
+      table.tier,
+      table.client,
+      table.source,
+      table.provider,
+      table.model,
+    ),
+    index("idx_leaderboard_tokens").on(table.grain, table.period_key, table.dataset, table.tier, table.total_tokens),
+    index("idx_model").on(table.model, table.grain, table.period_key),
+  ],
+)
+
+export const providerStat = mysqlTable(
+  "provider_stat",
+  {
+    ...periodColumns(),
+    provider: varchar({ length: 128 }).notNull(),
+    ...metricColumns(),
+    ...marketShareColumns(),
+    rank_by_tokens: int(),
+    rank_by_requests: int(),
+    rank_by_sessions: int(),
+    rank_by_cost: int(),
+    ...timestampColumns(),
+  },
+  (table) => [
+    uniqueIndex("uniq_provider_period").on(
+      table.grain,
+      table.period_key,
+      table.dataset,
+      table.tier,
+      table.client,
+      table.source,
+      table.provider,
+    ),
+    index("idx_provider_leaderboard_tokens").on(
+      table.grain,
+      table.period_key,
+      table.dataset,
+      table.tier,
+      table.total_tokens,
+    ),
+    index("idx_provider_market_share").on(
+      table.grain,
+      table.period_key,
+      table.dataset,
+      table.tier,
+      table.market_share_tokens,
+    ),
+    index("idx_provider_rank").on(table.grain, table.period_key, table.dataset, table.tier, table.rank_by_tokens),
+    index("idx_provider").on(table.provider, table.grain, table.period_key),
+  ],
+)
+
+export const geoStat = mysqlTable(
+  "geo_stat",
+  {
+    ...periodColumns(),
+    provider: varchar({ length: 128 }).notNull().default("all"),
+    model: varchar({ length: 256 }).notNull().default("all"),
+    country: char({ length: 2 }).notNull(),
+    continent: varchar({ length: 8 }).notNull().default(""),
+    ...metricColumns(),
+    ...marketShareColumns(),
+    rank_by_tokens: int(),
+    rank_by_requests: int(),
+    rank_by_sessions: int(),
+    rank_by_cost: int(),
+    ...timestampColumns(),
+  },
+  (table) => [
+    uniqueIndex("uniq_country_period").on(
+      table.grain,
+      table.period_key,
+      table.dataset,
+      table.tier,
+      table.client,
+      table.source,
+      table.provider,
+      table.model,
+      table.country,
+    ),
+    index("idx_country_map_tokens").on(table.grain, table.period_key, table.dataset, table.tier, table.total_tokens),
+    index("idx_country_rank").on(table.grain, table.period_key, table.dataset, table.tier, table.rank_by_tokens),
+    index("idx_country").on(table.country, table.grain, table.period_key),
+    index("idx_continent").on(table.continent, table.grain, table.period_key),
+    index("idx_country_model").on(table.model, table.country, table.grain, table.period_key),
+  ],
+)
+
+function periodColumns() {
+  return {
+    id: bigint({ mode: "number" }).autoincrement().primaryKey(),
+    grain: varchar({ length: 16 }).notNull(),
+    period_key: varchar({ length: 32 }).notNull(),
+    dataset: varchar({ length: 64 }).notNull().default("all"),
+    tier: varchar({ length: 64 }).notNull().default("all"),
+    client: varchar({ length: 64 }).notNull().default("all"),
+    source: varchar({ length: 64 }).notNull().default("all"),
+  }
+}
+
+function metricColumns() {
+  return {
+    sessions: bigint({ mode: "number" }).notNull().default(0),
+    requests: bigint({ mode: "number" }).notNull().default(0),
+    input_tokens: bigint({ mode: "number" }).notNull().default(0),
+    output_tokens: bigint({ mode: "number" }).notNull().default(0),
+    reasoning_tokens: bigint({ mode: "number" }).notNull().default(0),
+    cache_read_tokens: bigint({ mode: "number" }).notNull().default(0),
+    total_tokens: bigint({ mode: "number" }).notNull().default(0),
+    input_cost_microcents: bigint({ mode: "number" }).notNull().default(0),
+    output_cost_microcents: bigint({ mode: "number" }).notNull().default(0),
+    total_cost_microcents: bigint({ mode: "number" }).notNull().default(0),
+    avg_duration_ms: decimal({ precision: 12, scale: 2, mode: "number" }),
+    p50_duration_ms: int(),
+    p95_duration_ms: int(),
+    avg_ttfb_ms: decimal({ precision: 12, scale: 2, mode: "number" }),
+    p50_ttfb_ms: int(),
+    p95_ttfb_ms: int(),
+    avg_output_tps: decimal({ precision: 12, scale: 4, mode: "number" }),
+    success_count: bigint({ mode: "number" }).notNull().default(0),
+    error_count: bigint({ mode: "number" }).notNull().default(0),
+    sample_count: bigint({ mode: "number" }).notNull().default(0),
+  }
+}
+
+function marketShareColumns() {
+  return {
+    market_share_tokens: decimal({ precision: 10, scale: 6, mode: "number" }),
+    market_share_requests: decimal({ precision: 10, scale: 6, mode: "number" }),
+    market_share_sessions: decimal({ precision: 10, scale: 6, mode: "number" }),
+  }
+}
+
+function timestampColumns() {
+  return {
+    created_at: datetime({ mode: "date" }).notNull().defaultNow(),
+    updated_at: datetime({ mode: "date" }).notNull().defaultNow().onUpdateNow(),
+  }
+}

packages/stats/core/src/domain/geo.ts

@@ -0,0 +1,232 @@
+import { and, asc, eq, inArray, or } from "drizzle-orm"
+import { Effect, Layer } from "effect"
+import * as Context from "effect/Context"
+import { DatabaseError, DrizzleClient } from "../database"
+import { geoStat } from "../database/schema"
+import { RETIRED_STAT_MODELS, RETIRED_STAT_PROVIDERS } from "./model-normalization"
+import {
+  chunks,
+  collapseRows,
+  inserted,
+  rankRowsWithMarketShare,
+  statPeriodKey,
+  statRowScope,
+  synthesizeAllTierRows,
+  toStatBaseRow,
+  UPSERT_CHUNK_SIZE,
+  type StatBaseAggregate,
+} from "./stat"
+
+export type GeoStatRow = typeof geoStat.$inferInsert
+export type GeoStatAggregate = StatBaseAggregate & {
+  provider: string
+  model: string
+  country: string
+  continent: string
+}
+export type GeoStatMetric = {
+  periodKey: string
+  updatedAt: Date
+  tier: string
+  provider: string
+  model: string
+  country: string
+  continent: string
+  totalTokens: number
+}
+
+export declare namespace GeoStatRepo {
+  export interface Service {
+    readonly listDaily: (opts?: {
+      readonly provider?: string
+      readonly model?: string
+    }) => Effect.Effect<GeoStatMetric[], DatabaseError>
+    readonly listByPeriod: (opts: {
+      readonly grain: string
+      readonly periodKey: string
+      readonly dataset?: string
+      readonly tier?: string
+      readonly client?: string
+      readonly source?: string
+      readonly provider?: string
+      readonly model?: string
+    }) => Effect.Effect<GeoStatRow[], DatabaseError>
+    readonly upsert: (rows: GeoStatRow[]) => Effect.Effect<void, DatabaseError>
+    readonly deleteRetiredDimensions: (rows: GeoStatRow[]) => Effect.Effect<void, DatabaseError>
+  }
+}
+
+export class GeoStatRepo extends Context.Service<GeoStatRepo, GeoStatRepo.Service>()("@opencode/stats/GeoStatRepo") {
+  static readonly layer: Layer.Layer<GeoStatRepo, never, DrizzleClient> = Layer.effect(
+    GeoStatRepo,
+    Effect.gen(function* () {
+      const db = yield* DrizzleClient
+
+      const listDaily = Effect.fn("GeoStatRepo.listDaily")(function* (opts?: {
+        readonly provider?: string
+        readonly model?: string
+      }) {
+        const scope =
+          opts?.model && opts.provider
+            ? and(eq(geoStat.provider, opts.provider), eq(geoStat.model, opts.model))
+            : opts?.model
+              ? eq(geoStat.model, opts.model)
+              : and(eq(geoStat.provider, "all"), eq(geoStat.model, "all"))
+        return yield* Effect.tryPromise({
+          try: () =>
+            db
+              .select({
+                periodKey: geoStat.period_key,
+                updatedAt: geoStat.updated_at,
+                tier: geoStat.tier,
+                provider: geoStat.provider,
+                model: geoStat.model,
+                country: geoStat.country,
+                continent: geoStat.continent,
+                totalTokens: geoStat.total_tokens,
+              })
+              .from(geoStat)
+              .where(and(eq(geoStat.grain, "day"), eq(geoStat.client, "all"), eq(geoStat.source, "all"), scope))
+              .orderBy(asc(geoStat.period_key)),
+          catch: (cause) => DatabaseError.make({ cause }),
+        })
+      })
+
+      const listByPeriod = Effect.fn("GeoStatRepo.listByPeriod")(function* (opts: {
+        readonly grain: string
+        readonly periodKey: string
+        readonly dataset?: string
+        readonly tier?: string
+        readonly client?: string
+        readonly source?: string
+        readonly provider?: string
+        readonly model?: string
+      }) {
+        return yield* Effect.tryPromise({
+          try: () =>
+            db
+              .select()
+              .from(geoStat)
+              .where(
+                and(
+                  eq(geoStat.grain, opts.grain),
+                  eq(geoStat.period_key, opts.periodKey),
+                  eq(geoStat.dataset, opts.dataset ?? "zen"),
+                  eq(geoStat.tier, opts.tier ?? "all"),
+                  eq(geoStat.client, opts.client ?? "all"),
+                  eq(geoStat.source, opts.source ?? "all"),
+                  eq(geoStat.provider, opts.provider ?? "all"),
+                  eq(geoStat.model, opts.model ?? "all"),
+                ),
+              ),
+          catch: (cause) => DatabaseError.make({ cause }),
+        })
+      })
+
+      const upsert = Effect.fn("GeoStatRepo.upsert")(function* (rows: GeoStatRow[]) {
+        yield* Effect.forEach(
+          chunks(rows, UPSERT_CHUNK_SIZE),
+          (chunk) =>
+            Effect.tryPromise({
+              try: () =>
+                db
+                  .insert(geoStat)
+                  .values(chunk)
+                  .onDuplicateKeyUpdate({
+                    set: {
+                      continent: inserted("continent"),
+                      sessions: inserted("sessions"),
+                      requests: inserted("requests"),
+                      input_tokens: inserted("input_tokens"),
+                      output_tokens: inserted("output_tokens"),
+                      reasoning_tokens: inserted("reasoning_tokens"),
+                      cache_read_tokens: inserted("cache_read_tokens"),
+                      total_tokens: inserted("total_tokens"),
+                      input_cost_microcents: inserted("input_cost_microcents"),
+                      output_cost_microcents: inserted("output_cost_microcents"),
+                      total_cost_microcents: inserted("total_cost_microcents"),
+                      avg_duration_ms: inserted("avg_duration_ms"),
+                      p50_duration_ms: inserted("p50_duration_ms"),
+                      p95_duration_ms: inserted("p95_duration_ms"),
+                      avg_ttfb_ms: inserted("avg_ttfb_ms"),
+                      p50_ttfb_ms: inserted("p50_ttfb_ms"),
+                      p95_ttfb_ms: inserted("p95_ttfb_ms"),
+                      avg_output_tps: inserted("avg_output_tps"),
+                      success_count: inserted("success_count"),
+                      error_count: inserted("error_count"),
+                      sample_count: inserted("sample_count"),
+                      market_share_tokens: inserted("market_share_tokens"),
+                      market_share_requests: inserted("market_share_requests"),
+                      market_share_sessions: inserted("market_share_sessions"),
+                      rank_by_tokens: inserted("rank_by_tokens"),
+                      rank_by_requests: inserted("rank_by_requests"),
+                      rank_by_sessions: inserted("rank_by_sessions"),
+                      rank_by_cost: inserted("rank_by_cost"),
+                    },
+                  }),
+              catch: (cause) => DatabaseError.make({ cause }),
+            }),
+          { discard: true },
+        )
+      })
+
+      const deleteRetiredDimensions = Effect.fn("GeoStatRepo.deleteRetiredDimensions")(function* (rows: GeoStatRow[]) {
+        const scope = statRowScope(rows)
+        if (!scope) return
+
+        yield* Effect.tryPromise({
+          try: () =>
+            db
+              .delete(geoStat)
+              .where(
+                and(
+                  inArray(geoStat.grain, scope.grains),
+                  inArray(geoStat.period_key, scope.periodKeys),
+                  inArray(geoStat.dataset, scope.datasets),
+                  inArray(geoStat.client, scope.clients),
+                  inArray(geoStat.source, scope.sources),
+                  or(inArray(geoStat.provider, RETIRED_STAT_PROVIDERS), inArray(geoStat.model, RETIRED_STAT_MODELS)),
+                ),
+              ),
+          catch: (cause) => DatabaseError.make({ cause }),
+        })
+      })
+
+      return GeoStatRepo.of({ listDaily, listByPeriod, upsert, deleteRetiredDimensions })
+    }),
+  )
+}
+
+export function rowsFromAggregates(aggregates: GeoStatAggregate[]) {
+  return rankRowsWithMarketShare(
+    [
+      ...synthesizeAllTierRows(
+        collapseRows(aggregates.filter((item) => item.grain === "week").map(toRow), dimensionKey),
+        dimensionKey,
+      ),
+      ...synthesizeAllTierRows(
+        collapseRows(aggregates.filter((item) => item.grain === "day").map(toRow), dimensionKey),
+        dimensionKey,
+      ),
+    ],
+    marketShareKey,
+  )
+}
+
+function toRow(data: GeoStatAggregate): GeoStatRow {
+  return {
+    ...toStatBaseRow(data),
+    provider: data.provider,
+    model: data.model,
+    country: data.country,
+    continent: data.continent,
+  }
+}
+
+function dimensionKey(row: GeoStatRow) {
+  return [row.provider, row.model, row.country].join("\u0000")
+}
+
+function marketShareKey(row: GeoStatRow) {
+  return [statPeriodKey(row), row.provider, row.model].join("\u0000")
+}

packages/stats/core/src/domain/home.ts

@@ -0,0 +1,876 @@
+import { Effect } from "effect"
+import { DatabaseError } from "../database"
+import { GeoStatRepo, type GeoStatMetric } from "./geo"
+import { ModelStatRepo, type ModelStatMetric } from "./model"
+import { ProviderStatRepo, type ProviderStatMetric } from "./provider"
+
+export type UsageProduct = "All Users" | "Zen" | "Go" | "Enterprise"
+export type TokenProduct = "Zen" | "Go" | "Enterprise"
+export type UsageRange = "1D" | "1W" | "2W" | "1M" | "2M" | "3M" | "YTD" | "ALL"
+export type UsagePoint = { date: string; segments: { model: string; value: number }[] }
+export type MarketDay = { date: string; total: number; authors: { author: string; share: number; tokens: number }[] }
+export type LeaderboardEntry = {
+  model: string
+  provider: string
+  author: string
+  tokens: number
+  change: number | null
+  rank: number
+}
+export type TokenCostEntry = { model: string; total: number; input: number; output: number; cached: number }
+export type CacheRatioEntry = { model: string; ratio: number; cached: number; uncached: number; total: number }
+export type SessionCostEntry = { model: string; cost: number; tokens: number }
+export type CountryEntry = { country: string; continent: string; tokens: number; share: number; rank: number }
+export type ModelUsagePoint = { date: string; tokens: number; sessions: number; cost: number }
+export type ModelMixEntry = { label: string; tokens: number; share: number }
+export type ModelProductEntry = { product: string; tokens: number; sessions: number; share: number }
+export type ModelPeerEntry = {
+  model: string
+  provider: string
+  author: string
+  rank: number
+  tokens: number
+  share: number
+  slug: string
+}
+export type LabUsageModelEntry = {
+  model: string
+  provider: string
+  author: string
+  tokens: number
+  share: number
+  slug: string
+}
+export type StatsModelData = {
+  updatedAt: string | null
+  model: string
+  slug: string
+  provider: string
+  author: string
+  rank: number
+  previousRank: number | null
+  totalModels: number
+  tokenShare: number
+  tokenChange: number
+  totals: {
+    sessions: number
+    tokens: number
+    cost: number
+    tokensPerSession: number
+    costPerSession: number
+    costPerMillion: number
+    cacheRatio: number
+  }
+  usage: ModelUsagePoint[]
+  tokenMix: ModelMixEntry[]
+  productMix: ModelProductEntry[]
+  country: Record<UsageRange, CountryEntry[]>
+  peers: ModelPeerEntry[]
+}
+export type StatsLabData = {
+  updatedAt: string | null
+  provider: string
+  author: string
+  tokenShare: number
+  tokenChange: number
+  totals: {
+    sessions: number
+    tokens: number
+    models: number
+  }
+  usage: ModelUsagePoint[]
+  models: LabUsageModelEntry[]
+}
+export type StatsHomeData = {
+  updatedAt: string | null
+  usage: Record<UsageProduct, Record<UsageRange, UsagePoint[]>>
+  leaderboard: Record<UsageProduct, Record<UsageRange, LeaderboardEntry[]>>
+  market: Record<UsageRange, MarketDay[]>
+  tokenCost: Record<TokenProduct, TokenCostEntry[]>
+  cacheRatio: Record<TokenProduct, CacheRatioEntry[]>
+  sessionCost: Record<TokenProduct, SessionCostEntry[]>
+  country: Record<UsageRange, CountryEntry[]>
+}
+
+const DAY_MS = 86_400_000
+const TOKEN_SCALE = 1_000_000
+const DOLLARS_PER_MICROCENT = 1 / 100_000_000
+const METRIC_MODEL_LIMIT = 10
+const LEADERBOARD_CHANGE_MIN_MULTIPLE = 10
+const months = ["JAN", "FEB", "MAR", "APR", "MAY", "JUN", "JUL", "AUG", "SEP", "OCT", "NOV", "DEC"] as const
+
+type StatMetricRow = Omit<ModelStatMetric, "updatedAt"> & {
+  periodStart: number
+  updatedAt: number
+}
+type ProviderMetricRow = Omit<ProviderStatMetric, "updatedAt"> & {
+  periodStart: number
+  updatedAt: number
+}
+type GeoMetricRow = Omit<GeoStatMetric, "updatedAt"> & {
+  periodStart: number
+  updatedAt: number
+}
+
+type DateWindow = { start: number; end: number; previousStart: number; previousEnd: number }
+type Bucket = { start: number; end: number; label: string }
+type ModelAggregate = {
+  model: string
+  provider: string
+  sessions: number
+  inputTokens: number
+  outputTokens: number
+  reasoningTokens: number
+  cacheReadTokens: number
+  totalTokens: number
+  inputCostMicrocents: number
+  outputCostMicrocents: number
+  totalCostMicrocents: number
+}
+
+export const getStatsHomeData: () => Effect.Effect<
+  StatsHomeData,
+  DatabaseError,
+  ModelStatRepo | ProviderStatRepo | GeoStatRepo
+> = Effect.fn("StatsHome.getData")(function* () {
+  const modelStats = yield* ModelStatRepo
+  const providerStats = yield* ProviderStatRepo
+  const geoStats = yield* GeoStatRepo
+  const [modelRows, providerRows, geoRows] = yield* Effect.all(
+    [modelStats.listDaily(), providerStats.listDaily(), geoStats.listDaily()],
+    { concurrency: "unbounded" },
+  )
+  return buildStatsHomeData(modelRows, providerRows, geoRows)
+})
+
+export const getStatsModelData: (
+  model: string,
+  provider?: string,
+) => Effect.Effect<StatsModelData | null, DatabaseError, ModelStatRepo | GeoStatRepo> = Effect.fn("StatsModel.getData")(
+  function* (model, provider) {
+    const modelStats = yield* ModelStatRepo
+    const geoStats = yield* GeoStatRepo
+    const modelRows = yield* modelStats.listDaily()
+    const normalized = modelRows.flatMap(normalizeStatRow)
+    const resolvedModel = resolveModelName(model, normalized, provider)
+    if (!resolvedModel) return null
+    return buildStatsModelData(
+      resolvedModel,
+      modelRows,
+      yield* geoStats.listDaily({
+        model: resolvedModel,
+        provider: resolveModelProvider(resolvedModel, normalized, provider),
+      }),
+      provider,
+    )
+  },
+)
+
+export const getStatsLabData: (provider: string) => Effect.Effect<StatsLabData | null, DatabaseError, ModelStatRepo> =
+  Effect.fn("StatsLab.getData")(function* (provider) {
+    const modelStats = yield* ModelStatRepo
+    return buildStatsLabData(provider, yield* modelStats.listDaily())
+  })
+
+function buildStatsHomeData(
+  modelRows: ModelStatMetric[],
+  providerRows: ProviderStatMetric[],
+  geoRows: GeoStatMetric[],
+): StatsHomeData {
+  const normalized = modelRows.flatMap(normalizeStatRow)
+  const providers = providerRows.flatMap(normalizeProviderRow)
+  const geo = geoRows.flatMap(normalizeGeoRow)
+  const periods = [...normalized, ...providers, ...geo]
+  if (periods.length === 0) return emptyStatsHomeData()
+
+  const earliest = Math.min(...periods.map((row) => row.periodStart))
+  const latest = Math.max(...periods.map((row) => row.periodStart))
+  const latestUpdate = Math.max(...periods.map((row) => row.updatedAt))
+
+  return {
+    updatedAt: new Date(latestUpdate).toISOString(),
+    usage: createUsageProductRecord((product) =>
+      createRangeRecord((range) => buildUsagePoints(normalized, product, range, getWindow(range, earliest, latest))),
+    ),
+    leaderboard: createUsageProductRecord((product) =>
+      createRangeRecord((range) => buildLeaderboard(normalized, product, getWindow(range, earliest, latest))),
+    ),
+    market: createRangeRecord((range) => buildMarketShare(providers, "Go", range, getWindow(range, earliest, latest))),
+    tokenCost: createTokenProductRecord((product) =>
+      buildTokenCost(normalized, product, getWindow("1W", earliest, latest)),
+    ),
+    cacheRatio: createTokenProductRecord((product) =>
+      buildCacheRatio(normalized, product, getWindow("1W", earliest, latest)),
+    ),
+    sessionCost: createTokenProductRecord((product) =>
+      buildSessionCost(normalized, product, getWindow("1W", earliest, latest)),
+    ),
+    country: createRangeRecord((range) => buildCountryStats(geo, getWindow(range, earliest, latest))),
+  }
+}
+
+function buildStatsModelData(
+  modelParam: string,
+  modelRows: ModelStatMetric[],
+  geoRows: GeoStatMetric[],
+  providerParam?: string,
+): StatsModelData | null {
+  const normalized = modelRows.flatMap(normalizeStatRow)
+  const geo = geoRows.flatMap(normalizeGeoRow)
+  if (normalized.length === 0) return null
+
+  const model = resolveModelName(modelParam, normalized, providerParam)
+  if (!model) return null
+
+  const modelScopedRows = normalized.filter((row) => row.model === model)
+  const earliest = Math.min(...normalized.map((row) => row.periodStart))
+  const latest = Math.max(...normalized.map((row) => row.periodStart))
+  const latestUpdate = Math.max(...modelScopedRows.map((row) => row.updatedAt))
+  const window = getWindow("2M", earliest, latest)
+  const currentRows = rowsForProduct(modelScopedRows, "All Users", window.start, window.end)
+  const previousRows = rowsForProduct(modelScopedRows, "All Users", window.previousStart, window.previousEnd)
+  const current = combineRowsForModel(model, currentRows)
+  const previous = combineRowsForModel(model, previousRows)
+  const peers = aggregateByModelName(rowsForProduct(normalized, "All Users", window.start, window.end))
+    .filter((item) => item.totalTokens > 0)
+    .toSorted((a, b) => b.totalTokens - a.totalTokens || a.model.localeCompare(b.model))
+  const previousPeers = aggregateByModelName(
+    rowsForProduct(normalized, "All Users", window.previousStart, window.previousEnd),
+  )
+    .filter((item) => item.totalTokens > 0)
+    .toSorted((a, b) => b.totalTokens - a.totalTokens || a.model.localeCompare(b.model))
+  const rank = Math.max(1, peers.findIndex((item) => item.model === model) + 1)
+  const previousRankIndex = previousPeers.findIndex((item) => item.model === model)
+  const totalTokens = peers.reduce((sum, item) => sum + item.totalTokens, 0)
+
+  return {
+    updatedAt: Number.isFinite(latestUpdate) ? new Date(latestUpdate).toISOString() : null,
+    model,
+    slug: modelSlug(model),
+    provider: current.provider,
+    author: formatProvider(current.provider),
+    rank,
+    previousRank: previousRankIndex >= 0 ? previousRankIndex + 1 : null,
+    totalModels: peers.length,
+    tokenShare: totalTokens > 0 ? round((current.totalTokens / totalTokens) * 100, 2) : 0,
+    tokenChange: percentChange(current.totalTokens, previous.totalTokens),
+    totals: {
+      sessions: current.sessions,
+      tokens: current.totalTokens,
+      cost: round(microcentsToDollars(current.totalCostMicrocents), 2),
+      tokensPerSession: current.sessions > 0 ? Math.round(current.totalTokens / current.sessions) : 0,
+      costPerSession:
+        current.sessions > 0 ? round(microcentsToDollars(current.totalCostMicrocents) / current.sessions, 4) : 0,
+      costPerMillion: costPerMillion(current.totalCostMicrocents, current.totalTokens),
+      cacheRatio:
+        current.inputTokens + current.cacheReadTokens > 0
+          ? round((current.cacheReadTokens / (current.inputTokens + current.cacheReadTokens)) * 100, 1)
+          : 0,
+    },
+    usage: buildModelUsage(currentRows, window, "2M"),
+    tokenMix: buildModelTokenMix(current),
+    productMix: buildModelProductMix(modelScopedRows, window, current),
+    country: createRangeRecord((range) => buildCountryStats(geo, getWindow(range, earliest, latest))),
+    peers: buildModelPeers(peers, rank, totalTokens),
+  }
+}
+
+function buildStatsLabData(providerParam: string, modelRows: ModelStatMetric[]): StatsLabData | null {
+  const normalized = modelRows.flatMap(normalizeStatRow)
+  if (normalized.length === 0) return null
+
+  const provider = resolveProviderName(providerParam, normalized)
+  if (!provider) return null
+
+  const providerRows = normalized.filter((row) => providerMatches(row.provider, provider))
+  if (providerRows.length === 0) return null
+
+  const earliest = Math.min(...normalized.map((row) => row.periodStart))
+  const latest = Math.max(...normalized.map((row) => row.periodStart))
+  const latestUpdate = Math.max(...providerRows.map((row) => row.updatedAt))
+  const window = getWindow("2M", earliest, latest)
+  const currentRows = rowsForProduct(providerRows, "All Users", window.start, window.end)
+  const previousRows = rowsForProduct(providerRows, "All Users", window.previousStart, window.previousEnd)
+  const current = combineRowsForModel("", currentRows)
+  const previous = combineRowsForModel("", previousRows)
+  const allCurrent = aggregateByModel(rowsForProduct(normalized, "All Users", window.start, window.end))
+  const totalTokens = allCurrent.reduce((sum, item) => sum + item.totalTokens, 0)
+  const models = aggregateByModel(currentRows)
+    .filter((item) => item.totalTokens > 0)
+    .toSorted((a, b) => b.totalTokens - a.totalTokens || a.model.localeCompare(b.model))
+
+  return {
+    updatedAt: Number.isFinite(latestUpdate) ? new Date(latestUpdate).toISOString() : null,
+    provider,
+    author: formatProvider(provider),
+    tokenShare: totalTokens > 0 ? round((current.totalTokens / totalTokens) * 100, 2) : 0,
+    tokenChange: percentChange(current.totalTokens, previous.totalTokens),
+    totals: {
+      sessions: current.sessions,
+      tokens: current.totalTokens,
+      models: models.length,
+    },
+    usage: buildModelUsage(currentRows, window, "2M"),
+    models: models.map((item) => ({
+      model: item.model,
+      provider: item.provider,
+      author: formatProvider(item.provider),
+      tokens: item.totalTokens,
+      share: current.totalTokens > 0 ? round((item.totalTokens / current.totalTokens) * 100, 2) : 0,
+      slug: modelSlug(item.model),
+    })),
+  }
+}
+
+function emptyStatsHomeData(): StatsHomeData {
+  return {
+    updatedAt: null,
+    usage: createUsageProductRecord(() => createRangeRecord(() => [])),
+    leaderboard: createUsageProductRecord(() => createRangeRecord(() => [])),
+    market: createRangeRecord(() => []),
+    tokenCost: createTokenProductRecord(() => []),
+    cacheRatio: createTokenProductRecord(() => []),
+    sessionCost: createTokenProductRecord(() => []),
+    country: createRangeRecord(() => []),
+  }
+}
+
+function buildUsagePoints(rows: StatMetricRow[], product: UsageProduct, range: UsageRange, window: DateWindow) {
+  const windowRows = rowsForProduct(rows, product, window.start, window.end)
+  const modelOrder = aggregateByModel(windowRows)
+    .toSorted((a, b) => b.totalTokens - a.totalTokens)
+    .slice(0, 6)
+    .map((item) => ({ key: modelKey(item.provider, item.model), model: item.model }))
+
+  return createBuckets(window, range).map((bucket) => {
+    const bucketRows = aggregateByModel(rowsForProduct(rows, product, bucket.start, bucket.end))
+    const byModel = new Map(bucketRows.map((item) => [modelKey(item.provider, item.model), item.totalTokens]))
+    const segmentTokens = modelOrder.map((model) => ({ model: model.model, tokens: byModel.get(model.key) ?? 0 }))
+    const knownTokens = segmentTokens.reduce((sum, item) => sum + item.tokens, 0)
+    const totalTokens = bucketRows.reduce((sum, item) => sum + item.totalTokens, 0)
+    return {
+      date: bucket.label,
+      segments: [
+        ...segmentTokens.map((item) => ({ model: item.model, value: round(item.tokens / 1_000_000_000_000, 4) })),
+        { model: "Other", value: round(Math.max(totalTokens - knownTokens, 0) / 1_000_000_000_000, 4) },
+      ],
+    }
+  })
+}
+
+function buildLeaderboard(rows: StatMetricRow[], product: UsageProduct, window: DateWindow) {
+  const previous = new Map(
+    aggregateByModel(rowsForProduct(rows, product, window.previousStart, window.previousEnd)).map((item) => [
+      modelKey(item.provider, item.model),
+      item.totalTokens,
+    ]),
+  )
+
+  return aggregateByModel(rowsForProduct(rows, product, window.start, window.end))
+    .toSorted((a, b) => b.totalTokens - a.totalTokens)
+    .slice(0, 18)
+    .map((item, index) => ({
+      model: item.model,
+      provider: item.provider,
+      author: formatProvider(item.provider),
+      tokens: Math.round(item.totalTokens / 1_000_000_000),
+      change: leaderboardChange(item.totalTokens, previous.get(modelKey(item.provider, item.model)) ?? 0),
+      rank: index + 1,
+    }))
+}
+
+function buildMarketShare(rows: ProviderMetricRow[], product: UsageProduct, range: UsageRange, window: DateWindow) {
+  return createBuckets(window, range).flatMap((bucket) => {
+    const total = aggregateByProvider(rowsForProduct(rows, product, bucket.start, bucket.end)).toSorted(
+      (a, b) => b.tokens - a.tokens,
+    )
+    const totalTokens = total.reduce((sum, item) => sum + item.tokens, 0)
+    if (totalTokens === 0) return []
+
+    const authors = total.slice(0, 8)
+    const knownTokens = authors.reduce((sum, item) => sum + item.tokens, 0)
+    const withOther = [...authors, { provider: "Other", tokens: Math.max(totalTokens - knownTokens, 0) }].filter(
+      (item) => item.tokens > 0,
+    )
+
+    return [
+      {
+        date: bucket.label,
+        total: round(totalTokens / 1_000_000_000_000, 2),
+        authors: withOther.map((item) => ({
+          author: item.provider === "Other" ? "Other" : formatProvider(item.provider),
+          share: round((item.tokens / totalTokens) * 100, 1),
+          tokens: round(item.tokens / 1_000_000_000_000, 2),
+        })),
+      },
+    ]
+  })
+}
+
+function buildCountryStats(rows: GeoMetricRow[], window: DateWindow) {
+  const countries = aggregateByCountry(rowsForProduct(rows, "All Users", window.start, window.end))
+    .filter((item) => item.tokens > 0 && item.country !== "AQ")
+    .toSorted((a, b) => b.tokens - a.tokens)
+  const totalTokens = countries.reduce((sum, item) => sum + item.tokens, 0)
+  if (totalTokens === 0) return []
+
+  return countries.map((item, index) => ({
+    country: item.country,
+    continent: item.continent,
+    tokens: round(item.tokens / 1_000_000_000_000, 4),
+    share: round((item.tokens / totalTokens) * 100, 1),
+    rank: index + 1,
+  }))
+}
+
+function buildTokenCost(rows: StatMetricRow[], product: TokenProduct, window: DateWindow) {
+  return topModelsByUsage(rows, product, window)
+    .flatMap((item) => {
+      const total = costPerMillion(item.totalCostMicrocents, item.totalTokens)
+      if (total === 0) return []
+      return [
+        {
+          model: item.model,
+          total,
+          input: costPerMillion(item.inputCostMicrocents, item.inputTokens),
+          output: costPerMillion(item.outputCostMicrocents, item.outputTokens + item.reasoningTokens),
+          cached: costPerMillion(item.inputCostMicrocents, item.inputTokens + item.cacheReadTokens),
+        },
+      ]
+    })
+    .toSorted((a, b) => a.total - b.total)
+}
+
+function buildCacheRatio(rows: StatMetricRow[], product: TokenProduct, window: DateWindow) {
+  return topModelsByUsage(rows, product, window)
+    .flatMap((item) => {
+      const total = item.inputTokens + item.cacheReadTokens
+      if (total === 0) return []
+      return [
+        {
+          model: item.model,
+          ratio: round((item.cacheReadTokens / total) * 100, 1),
+          cached: round(item.cacheReadTokens / 1_000_000_000, 1),
+          uncached: round(item.inputTokens / 1_000_000_000, 1),
+          total: round(total / 1_000_000_000, 1),
+        },
+      ]
+    })
+    .toSorted((a, b) => b.ratio - a.ratio || b.cached - a.cached)
+}
+
+function buildSessionCost(rows: StatMetricRow[], product: TokenProduct, window: DateWindow) {
+  return topModelsByUsage(rows, product, window)
+    .flatMap((item) => {
+      if (item.sessions === 0) return []
+      const cost = round(microcentsToDollars(item.totalCostMicrocents) / item.sessions, 4)
+      if (cost === 0) return []
+      return [{ model: item.model, cost, tokens: Math.round(item.totalTokens / item.sessions) }]
+    })
+    .toSorted((a, b) => a.cost - b.cost)
+}
+
+function topModelsByUsage(rows: StatMetricRow[], product: TokenProduct, window: DateWindow) {
+  return aggregateByModel(rowsForProduct(rows, product, window.start, window.end))
+    .toSorted((a, b) => b.totalTokens - a.totalTokens)
+    .slice(0, METRIC_MODEL_LIMIT)
+}
+
+function buildModelUsage(rows: StatMetricRow[], window: DateWindow, range: UsageRange) {
+  return createBuckets(window, range).map((bucket) => {
+    const aggregate = combineRowsForModel(
+      "",
+      rows.filter((row) => row.periodStart >= bucket.start && row.periodStart < bucket.end),
+    )
+    return {
+      date: bucket.label,
+      tokens: aggregate.totalTokens,
+      sessions: aggregate.sessions,
+      cost: round(microcentsToDollars(aggregate.totalCostMicrocents), 2),
+    }
+  })
+}
+
+function buildModelTokenMix(aggregate: ModelAggregate): ModelMixEntry[] {
+  const items = [
+    { label: "Input", tokens: aggregate.inputTokens },
+    { label: "Output", tokens: aggregate.outputTokens },
+    { label: "Reasoning", tokens: aggregate.reasoningTokens },
+    { label: "Cached", tokens: aggregate.cacheReadTokens },
+  ].filter((item) => item.tokens > 0)
+  const total = items.reduce((sum, item) => sum + item.tokens, 0)
+  if (total === 0) return []
+  return items.map((item) => ({ ...item, share: round((item.tokens / total) * 100, 1) }))
+}
+
+function buildModelProductMix(
+  rows: StatMetricRow[],
+  window: DateWindow,
+  fallback: ModelAggregate,
+): ModelProductEntry[] {
+  const products = ["Go", "Zen", "Enterprise"] as const
+  const items = products.flatMap((product) => {
+    const aggregate = combineRowsForModel(
+      fallback.model,
+      rows.filter((row) => row.tier === product && row.periodStart >= window.start && row.periodStart < window.end),
+    )
+    if (aggregate.totalTokens === 0) return []
+    return [{ product, tokens: aggregate.totalTokens, sessions: aggregate.sessions }]
+  })
+  const total = items.reduce((sum, item) => sum + item.tokens, 0)
+  if (total > 0) return items.map((item) => ({ ...item, share: round((item.tokens / total) * 100, 1) }))
+  if (fallback.totalTokens === 0) return []
+  return [{ product: "All Users", tokens: fallback.totalTokens, sessions: fallback.sessions, share: 100 }]
+}
+
+function buildModelPeers(peers: ModelAggregate[], rank: number, totalTokens: number): ModelPeerEntry[] {
+  const start = Math.max(0, Math.min(rank - 4, Math.max(peers.length - 7, 0)))
+  return peers.slice(start, start + 7).map((item, index) => ({
+    model: item.model,
+    provider: item.provider,
+    author: formatProvider(item.provider),
+    rank: start + index + 1,
+    tokens: item.totalTokens,
+    share: totalTokens > 0 ? round((item.totalTokens / totalTokens) * 100, 2) : 0,
+    slug: modelSlug(item.model),
+  }))
+}
+
+function rowsForProduct<T extends { periodStart: number; tier: string }>(
+  rows: T[],
+  product: UsageProduct,
+  start: number,
+  end: number,
+) {
+  const windowRows = rows.filter((row) => row.periodStart >= start && row.periodStart < end)
+  if (product !== "All Users") return windowRows.filter((row) => row.tier === product)
+
+  const allRows = windowRows.filter((row) => row.tier === "all")
+  if (allRows.length > 0) return allRows
+  return windowRows.filter((row) => row.tier !== "all")
+}
+
+function aggregateByModel(rows: StatMetricRow[]) {
+  return Object.values(
+    rows.reduce<Record<string, ModelAggregate>>((result, row) => {
+      const key = modelKey(row.provider, row.model)
+      result[key] = combineModelAggregate(result[key], row)
+      return result
+    }, {}),
+  )
+}
+
+function aggregateByModelName(rows: StatMetricRow[]) {
+  return Object.values(
+    rows.reduce<Record<string, ModelAggregate>>((result, row) => {
+      result[row.model] = combineModelAggregate(result[row.model], row)
+      return result
+    }, {}),
+  )
+}
+
+function aggregateByProvider(rows: ProviderMetricRow[]) {
+  return Object.values(
+    rows.reduce<Record<string, { provider: string; tokens: number }>>((result, row) => {
+      result[row.provider] = {
+        provider: row.provider,
+        tokens: (result[row.provider]?.tokens ?? 0) + row.totalTokens,
+      }
+      return result
+    }, {}),
+  )
+}
+
+function aggregateByCountry(rows: GeoMetricRow[]) {
+  return Object.values(
+    rows.reduce<Record<string, { country: string; continent: string; tokens: number }>>((result, row) => {
+      result[row.country] = {
+        country: row.country,
+        continent: result[row.country]?.continent || row.continent,
+        tokens: (result[row.country]?.tokens ?? 0) + row.totalTokens,
+      }
+      return result
+    }, {}),
+  )
+}
+
+function combineRowsForModel(model: string, rows: StatMetricRow[]): ModelAggregate {
+  const aggregate = rows.reduce<ModelAggregate | undefined>(
+    (result, row) => combineModelAggregate(result, row),
+    undefined,
+  )
+  if (aggregate) return { ...aggregate, model: model || aggregate.model }
+  return {
+    model,
+    provider: "unknown",
+    sessions: 0,
+    inputTokens: 0,
+    outputTokens: 0,
+    reasoningTokens: 0,
+    cacheReadTokens: 0,
+    totalTokens: 0,
+    inputCostMicrocents: 0,
+    outputCostMicrocents: 0,
+    totalCostMicrocents: 0,
+  }
+}
+
+function combineModelAggregate(current: ModelAggregate | undefined, row: StatMetricRow): ModelAggregate {
+  return {
+    model: row.model,
+    provider: row.provider,
+    sessions: (current?.sessions ?? 0) + row.sessions,
+    inputTokens: (current?.inputTokens ?? 0) + row.inputTokens,
+    outputTokens: (current?.outputTokens ?? 0) + row.outputTokens,
+    reasoningTokens: (current?.reasoningTokens ?? 0) + row.reasoningTokens,
+    cacheReadTokens: (current?.cacheReadTokens ?? 0) + row.cacheReadTokens,
+    totalTokens: (current?.totalTokens ?? 0) + row.totalTokens,
+    inputCostMicrocents: (current?.inputCostMicrocents ?? 0) + row.inputCostMicrocents,
+    outputCostMicrocents: (current?.outputCostMicrocents ?? 0) + row.outputCostMicrocents,
+    totalCostMicrocents: (current?.totalCostMicrocents ?? 0) + row.totalCostMicrocents,
+  }
+}
+
+function getWindow(range: UsageRange, earliest: number, latest: number): DateWindow {
+  const end = latest + DAY_MS
+  const start = Math.max(
+    earliest,
+    range === "1D"
+      ? latest
+      : range === "1W"
+        ? latest - 6 * DAY_MS
+        : range === "2W"
+          ? latest - 13 * DAY_MS
+          : range === "1M"
+            ? latest - 27 * DAY_MS
+            : range === "2M"
+              ? latest - 55 * DAY_MS
+              : range === "3M"
+                ? latest - 89 * DAY_MS
+                : range === "YTD"
+                  ? Date.UTC(new Date(latest).getUTCFullYear(), 0, 1)
+                  : earliest,
+  )
+  const duration = end - start
+  return { start, end, previousStart: start - duration, previousEnd: start }
+}
+
+function createBuckets(window: DateWindow, range: UsageRange): Bucket[] {
+  const span = Math.max(window.end - window.start, DAY_MS)
+  const count =
+    range === "1D"
+      ? 1
+      : range === "1W" || range === "2W" || range === "1M" || range === "2M" || range === "3M"
+        ? Math.ceil(span / DAY_MS)
+        : Math.max(1, Math.min(7, Math.ceil(span / DAY_MS)))
+  const size = span / count
+  return Array.from({ length: count }, (_, index) => {
+    const start = window.start + index * size
+    const end = index === count - 1 ? window.end : window.start + (index + 1) * size
+    return { start, end, label: formatBucketLabel(start, end, range) }
+  })
+}
+
+function createUsageProductRecord<T>(value: (product: UsageProduct) => T): Record<UsageProduct, T> {
+  return {
+    "All Users": value("All Users"),
+    Zen: value("Zen"),
+    Go: value("Go"),
+    Enterprise: value("Enterprise"),
+  }
+}
+
+function createTokenProductRecord<T>(value: (product: TokenProduct) => T): Record<TokenProduct, T> {
+  return {
+    Zen: value("Zen"),
+    Go: value("Go"),
+    Enterprise: value("Enterprise"),
+  }
+}
+
+function createRangeRecord<T>(value: (range: UsageRange) => T): Record<UsageRange, T> {
+  return {
+    "1D": value("1D"),
+    "1W": value("1W"),
+    "2W": value("2W"),
+    "1M": value("1M"),
+    "2M": value("2M"),
+    "3M": value("3M"),
+    YTD: value("YTD"),
+    ALL: value("ALL"),
+  }
+}
+
+function normalizeStatRow(row: ModelStatMetric): StatMetricRow[] {
+  const periodStart = periodKeyTime(row.periodKey)
+  const updatedAt = dateTime(row.updatedAt)
+  if (!Number.isFinite(periodStart) || !Number.isFinite(updatedAt)) return []
+  return [
+    {
+      ...row,
+      periodStart,
+      updatedAt,
+      tier: normalizeTier(row.tier),
+      provider: row.provider || "unknown",
+      model: row.model || "unknown",
+    },
+  ]
+}
+
+function normalizeProviderRow(row: ProviderStatMetric): ProviderMetricRow[] {
+  const periodStart = periodKeyTime(row.periodKey)
+  const updatedAt = dateTime(row.updatedAt)
+  if (!Number.isFinite(periodStart) || !Number.isFinite(updatedAt)) return []
+  return [
+    {
+      ...row,
+      periodStart,
+      updatedAt,
+      tier: normalizeTier(row.tier),
+      provider: row.provider || "unknown",
+    },
+  ]
+}
+
+function normalizeGeoRow(row: GeoStatMetric): GeoMetricRow[] {
+  const periodStart = periodKeyTime(row.periodKey)
+  const updatedAt = dateTime(row.updatedAt)
+  if (!Number.isFinite(periodStart) || !Number.isFinite(updatedAt)) return []
+  return [
+    {
+      ...row,
+      periodStart,
+      updatedAt,
+      tier: normalizeTier(row.tier),
+      provider: row.provider || "all",
+      model: row.model || "all",
+      country: row.country || "ZZ",
+      continent: row.continent || "",
+    },
+  ]
+}
+
+function normalizeTier(value: string) {
+  const normalized = value.toLowerCase()
+  if (normalized === "paid" || normalized === "zen") return "Zen"
+  if (normalized === "go") return "Go"
+  if (normalized === "enterprise") return "Enterprise"
+  if (normalized === "all") return "all"
+  return value
+}
+
+function dateTime(value: Date | string) {
+  return (value instanceof Date ? value : new Date(value)).getTime()
+}
+
+function periodKeyTime(value: string) {
+  const match = /^(\d{4})-(\d{2})-(\d{2})$/.exec(value)
+  if (!match) return Number.NaN
+  return Date.UTC(Number(match[1]), Number(match[2]) - 1, Number(match[3]))
+}
+
+function formatBucketLabel(start: number, _end: number, range: UsageRange) {
+  const date = new Date(start)
+  if (range === "YTD") return months[date.getUTCMonth()]
+  if (range === "ALL")
+    return date.getUTCFullYear() === new Date().getUTCFullYear()
+      ? months[date.getUTCMonth()]
+      : String(date.getUTCFullYear())
+  return formatDay(start)
+}
+
+function formatDay(value: number) {
+  const date = new Date(value)
+  return `${months[date.getUTCMonth()]} ${date.getUTCDate()}`
+}
+
+function formatProvider(provider: string) {
+  const known: Record<string, string> = {
+    anthropic: "Anthropic",
+    deepseek: "DeepSeek",
+    google: "Google",
+    minimax: "MiniMax",
+    moonshot: "Moonshot",
+    moonshotai: "Moonshot",
+    nvidia: "NVIDIA",
+    opencode: "opencode",
+    openai: "OpenAI",
+    qwen: "Qwen",
+    tencent: "Tencent",
+    xai: "xAI",
+    xiaomi: "Xiaomi",
+    zhipu: "Zhipu",
+    zhipuai: "Zhipu",
+  }
+  const normalized = provider.toLowerCase().replace(/[^a-z0-9]/g, "")
+  return known[normalized] ?? provider.replace(/[-_]/g, " ").replace(/\b\w/g, (letter) => letter.toUpperCase())
+}
+
+function resolveModelName(modelParam: string, rows: StatMetricRow[], providerParam?: string) {
+  const input = modelParam.trim()
+  if (!input) return undefined
+  const normalizedInput = input.toLowerCase()
+  const inputSlug = modelSlug(input)
+  const candidates = providerParam
+    ? aggregateByModel(rows).filter((item) => providerMatches(item.provider, providerParam))
+    : aggregateByModelName(rows)
+  return candidates
+    .filter((item) => item.model.toLowerCase() === normalizedInput || modelSlug(item.model) === inputSlug)
+    .toSorted((a, b) => b.totalTokens - a.totalTokens || a.model.localeCompare(b.model))[0]?.model
+}
+
+function resolveModelProvider(model: string, rows: StatMetricRow[], providerParam?: string) {
+  return aggregateByModel(rows)
+    .filter((item) => item.model === model && (!providerParam || providerMatches(item.provider, providerParam)))
+    .toSorted((a, b) => b.totalTokens - a.totalTokens || a.provider.localeCompare(b.provider))[0]?.provider
+}
+
+function providerMatches(provider: string, providerParam: string) {
+  return modelSlug(provider) === modelSlug(providerParam)
+}
+
+function resolveProviderName(providerParam: string, rows: StatMetricRow[]) {
+  const input = providerParam.trim()
+  if (!input) return undefined
+  const inputSlug = modelSlug(input)
+  return aggregateByModel(rows)
+    .filter((item) => modelSlug(item.provider) === inputSlug)
+    .toSorted((a, b) => b.totalTokens - a.totalTokens || a.provider.localeCompare(b.provider))[0]?.provider
+}
+
+export function modelSlug(value: string) {
+  return value
+    .trim()
+    .toLowerCase()
+    .replace(/[^a-z0-9]+/g, "-")
+    .replace(/^-+|-+$/g, "")
+    .replace(/-{2,}/g, "-")
+}
+
+function modelKey(provider: string, model: string) {
+  return `${provider}\u0000${model}`
+}
+
+function costPerMillion(costMicrocents: number, tokens: number) {
+  if (tokens <= 0 || costMicrocents <= 0) return 0
+  return round((microcentsToDollars(costMicrocents) / tokens) * TOKEN_SCALE, 2)
+}
+
+function microcentsToDollars(value: number) {
+  return value * DOLLARS_PER_MICROCENT
+}
+
+function percentChange(current: number, previous: number) {
+  if (previous <= 0) return current > 0 ? 100 : 0
+  return Math.round(((current - previous) / previous) * 100)
+}
+
+function leaderboardChange(current: number, previous: number) {
+  if (current <= 0) return 0
+  if (previous <= 0 || current >= previous * LEADERBOARD_CHANGE_MIN_MULTIPLE) return null
+  return percentChange(current, previous)
+}
+
+function round(value: number, digits: number) {
+  return Number(value.toFixed(digits))
+}

packages/stats/core/src/domain/inference.test.ts

@@ -0,0 +1,99 @@
+import { describe, expect, test } from "bun:test"
+import { toGeoAggregate, toModelAggregate, toProviderAggregate } from "./inference"
+import { modelAuthor, normalizeInferenceModel, statModel, statProvider } from "./model-normalization"
+
+describe("inference stat normalization", () => {
+  test("normalizes model suffixes used by router/provider variants", () => {
+    expect(normalizeInferenceModel("deepseek-v4-flash-free")).toBe("deepseek-v4-flash")
+    expect(normalizeInferenceModel("deepseek-v4-flash:global")).toBe("deepseek-v4-flash")
+    expect(normalizeInferenceModel("mimo-v2.5-free")).toBe("mimo-v2.5")
+    expect(normalizeInferenceModel("nemotron-3-super-free")).toBe("nemotron-3-super")
+    expect(normalizeInferenceModel("mimo-v2.5-free:global")).toBe("mimo-v2.5")
+  })
+
+  test("maps normalized model ids to public authors", () => {
+    expect(modelAuthor("big-pickle")).toBe("unknown")
+    expect(modelAuthor("claude-sonnet-4-5")).toBe("anthropic")
+    expect(modelAuthor("deepseek-v4-pro")).toBe("deepseek")
+    expect(modelAuthor("gemini-3.5-flash")).toBe("google")
+    expect(modelAuthor("glm-5.1")).toBe("zhipu")
+    expect(modelAuthor("gpt-5.5-pro")).toBe("openai")
+    expect(modelAuthor("grok-build-0.1")).toBe("xai")
+    expect(modelAuthor("hy3-preview")).toBe("tencent")
+    expect(modelAuthor("kimi-k2.6")).toBe("moonshot")
+    expect(modelAuthor("mimo-v2-omni")).toBe("xiaomi")
+    expect(modelAuthor("minimax-m2.7")).toBe("minimax")
+    expect(modelAuthor("nemotron-3-super-free")).toBe("nvidia")
+    expect(modelAuthor("qwen3.7-max")).toBe("qwen")
+    expect(modelAuthor("alpha-gpt-next")).toBeUndefined()
+  })
+
+  test("uses provider.model to resolve opencode route providers", () => {
+    expect(statModel("big-pickle", "claude-sonnet-4-5")).toBe("claude-sonnet-4-5")
+    expect(statModel("big-pickle", "gpt-5-free")).toBe("gpt-5")
+    expect(statModel("big-pickle", "")).toBe("unknown")
+    expect(statProvider("big-pickle", "claude-sonnet-4-5", "opencode")).toBe("anthropic")
+    expect(statProvider("big-pickle", "gpt-5", "opencode")).toBe("openai")
+    expect(statProvider("big-pickle", "", "opencode")).toBe("unknown")
+    expect(statProvider("unknown", "", "custom-provider")).toBe("custom-provider")
+  })
+
+  test("model aggregates prefer provider.model and use normalized model", () => {
+    expect(toModelAggregate(aggregate("alpha-gpt-next", "openai"))).toEqual([])
+
+    expect(toModelAggregate(aggregate("deepseek-v4-flash-free", "not-public-provider"))).toMatchObject([
+      {
+        period_key: "2026-05-20",
+        provider: "deepseek",
+        model: "deepseek-v4-flash",
+      },
+    ])
+
+    expect(
+      toModelAggregate({ ...aggregate("big-pickle", "opencode"), provider_model: "claude-sonnet-4-5" }),
+    ).toMatchObject([
+      {
+        provider: "anthropic",
+        model: "claude-sonnet-4-5",
+        provider_model: "claude-sonnet-4-5",
+      },
+    ])
+  })
+
+  test("provider aggregates never keep opencode as the provider", () => {
+    expect(toProviderAggregate({ ...aggregate("big-pickle", "opencode"), provider_model: "gpt-5" })).toMatchObject([
+      { provider: "openai" },
+    ])
+    expect(toProviderAggregate(aggregate("big-pickle", "opencode"))).toMatchObject([{ provider: "unknown" }])
+  })
+
+  test("geo aggregates never keep opencode or big-pickle dimensions", () => {
+    expect(toGeoAggregate({ ...aggregate("big-pickle", "opencode"), country: "US" })).toMatchObject([
+      { provider: "unknown", model: "unknown", country: "US" },
+    ])
+  })
+
+  test("model aggregates use ISO week period keys", () => {
+    expect(
+      toModelAggregate({
+        ...aggregate("gpt-5.5-pro", "openai"),
+        grain: "week",
+        period_key: "2026-W20",
+      }),
+    ).toMatchObject([{ period_key: "2026-W20" }])
+  })
+})
+
+function aggregate(model: string, provider: string) {
+  return {
+    grain: "day",
+    period_key: "2026-05-20",
+    dataset: "zen",
+    tier: "Paid",
+    provider,
+    model,
+    sessions: "1",
+    requests: "1",
+    sample_count: "1",
+  }
+}

packages/stats/core/src/domain/inference.ts

@@ -0,0 +1,263 @@
+import { Resource } from "sst/resource"
+import type { AthenaData } from "../athena"
+import type { GeoStatAggregate } from "./geo"
+import type { ModelStatAggregate } from "./model"
+import {
+  EXCLUDED_MODELS,
+  MODEL_AUTHOR_RULES,
+  RETIRED_STAT_PROVIDERS,
+  statModel,
+  statProvider,
+} from "./model-normalization"
+import type { ProviderStatAggregate } from "./provider"
+import { normalizeCountry, normalizeTier, type StatBaseAggregate } from "./stat"
+
+export type StatDimension = "model" | "provider" | "geo" | "geo_model"
+
+export function buildStatsQuery(periodStart: Date, periodEnd: Date, dimension: StatDimension) {
+  const periodStartValue = sqlString(periodStart.toISOString())
+  const periodEndValue = sqlString(periodEnd.toISOString())
+  const sourceTable = [Resource.InferenceEvent.catalog, Resource.InferenceEvent.database, Resource.InferenceEvent.table]
+    .map(sqlIdentifier)
+    .join(".")
+  const dimensionSql = (() => {
+    if (dimension === "model")
+      return {
+        select: "provider, model, COALESCE(MAX(NULLIF(provider_model, '')), '') AS provider_model",
+        groupBy: "provider, model",
+      }
+    if (dimension === "provider") return { select: "provider", groupBy: "provider" }
+    if (dimension === "geo_model")
+      return {
+        select: "provider, model, country, COALESCE(MAX(NULLIF(continent, '')), '') AS continent",
+        groupBy: "provider, model, country",
+      }
+    return {
+      select: "'all' AS provider, 'all' AS model, country, COALESCE(MAX(NULLIF(continent, '')), '') AS continent",
+      groupBy: "country",
+    }
+  })()
+  const aggregateColumns = `
+    COUNT(DISTINCT session) AS sessions,
+    COUNT(*) AS requests,
+    COALESCE(SUM(tokens_input), 0) AS input_tokens,
+    COALESCE(SUM(tokens_output), 0) AS output_tokens,
+    COALESCE(SUM(tokens_reasoning), 0) AS reasoning_tokens,
+    COALESCE(SUM(tokens_cache_read), 0) AS cache_read_tokens,
+    COALESCE(SUM(tokens_total), 0) AS total_tokens,
+    COALESCE(SUM(cost_input_microcents), 0) AS input_cost_microcents,
+    COALESCE(SUM(cost_output_microcents), 0) AS output_cost_microcents,
+    COALESCE(SUM(cost_total_microcents), 0) AS total_cost_microcents,
+    AVG(duration_ms) AS avg_duration_ms,
+    approx_percentile(CAST(duration_ms AS double), 0.5) AS p50_duration_ms,
+    approx_percentile(CAST(duration_ms AS double), 0.95) AS p95_duration_ms,
+    AVG(ttfb_ms) AS avg_ttfb_ms,
+    approx_percentile(CAST(ttfb_ms AS double), 0.5) AS p50_ttfb_ms,
+    approx_percentile(CAST(ttfb_ms AS double), 0.95) AS p95_ttfb_ms,
+    AVG(output_tps) AS avg_output_tps,
+    SUM(CASE WHEN status >= 200 AND status < 400 THEN 1 ELSE 0 END) AS success_count,
+    SUM(CASE WHEN status >= 400 THEN 1 ELSE 0 END) AS error_count,
+    COUNT(*) AS sample_count`
+
+  return `
+WITH normalized AS (
+  SELECT
+    from_iso8601_timestamp(event_timestamp) AS event_time,
+    model AS raw_model,
+    ${statModelSql("model", "provider_model")} AS model,
+    COALESCE(NULLIF(provider_model, ''), '') AS provider_model,
+    COALESCE(NULLIF(provider, ''), '') AS raw_provider,
+    UPPER(COALESCE(NULLIF(cf_country, ''), 'ZZ')) AS country,
+    COALESCE(NULLIF(cf_continent, ''), '') AS continent,
+    session,
+    status,
+    duration AS duration_ms,
+    time_to_first_byte AS ttfb_ms,
+    timestamp_first_byte,
+    timestamp_last_byte,
+    tokens_input,
+    tokens_output,
+    tokens_reasoning,
+    tokens_cache_read,
+    tokens_cache_write_5m,
+    tokens_cache_write_1h,
+    cost_input_microcents,
+    cost_output_microcents,
+    cost_total_microcents,
+    cost_input,
+    cost_output,
+    cost_total,
+    source
+  FROM ${sourceTable}
+  WHERE event_type = 'completions'
+    AND model IS NOT NULL
+    AND model <> ''
+    AND event_timestamp >= ${periodStartValue}
+    AND event_timestamp < ${periodEndValue}
+), filtered AS (
+  SELECT
+    event_time,
+    CASE
+      WHEN source = 'lite' THEN 'Go'
+      WHEN raw_model IN ('gpt-5-nano', 'grok-code', 'big-pickle') OR regexp_like(raw_model, '-free(:global)?$') THEN 'Free'
+      ELSE 'Paid'
+    END AS tier,
+    ${statProviderSql("model", "provider_model", "raw_provider")} AS provider,
+    provider_model,
+    model,
+    country,
+    continent,
+    session,
+    status,
+    duration_ms,
+    ttfb_ms,
+    CASE
+      WHEN timestamp_last_byte - timestamp_first_byte < 100 THEN null
+      ELSE CAST(tokens_output AS double) / (timestamp_last_byte - timestamp_first_byte) * 1000
+    END AS output_tps,
+    tokens_input,
+    tokens_output,
+    tokens_reasoning,
+    tokens_cache_read,
+    COALESCE(tokens_cache_read, 0) + COALESCE(tokens_cache_write_5m, 0) + COALESCE(tokens_cache_write_1h, 0) + COALESCE(tokens_input, 0) + COALESCE(tokens_output, 0) AS tokens_total,
+    COALESCE(cost_input_microcents, cost_input * 1000000) AS cost_input_microcents,
+    COALESCE(cost_output_microcents, cost_output * 1000000) AS cost_output_microcents,
+    COALESCE(cost_total_microcents, cost_total * 1000000) AS cost_total_microcents
+  FROM normalized
+  WHERE lower(model) NOT IN (${[...EXCLUDED_MODELS].map(sqlString).join(", ")})
+), weekly AS (
+  SELECT
+    concat(CAST(year_of_week(event_time) AS varchar), '-W', lpad(CAST(week(event_time) AS varchar), 2, '0')) AS week_key,
+    *
+  FROM filtered
+), daily AS (
+  SELECT substr(to_iso8601(date_trunc('day', event_time)), 1, 10) AS day_key, *
+  FROM filtered
+)
+SELECT
+  'week' AS grain,
+  week_key AS period_key,
+  ${sqlString(Resource.StatsSyncConfig.dataset)} AS dataset,
+  tier,
+  ${dimensionSql.select},
+  ${aggregateColumns}
+FROM weekly
+GROUP BY week_key, tier, ${dimensionSql.groupBy}
+UNION ALL
+SELECT
+  'day' AS grain,
+  day_key AS period_key,
+  ${sqlString(Resource.StatsSyncConfig.dataset)} AS dataset,
+  tier,
+  ${dimensionSql.select},
+  ${aggregateColumns}
+FROM daily
+GROUP BY day_key, tier, ${dimensionSql.groupBy}
+ORDER BY grain, period_key, total_tokens DESC
+`
+}
+
+export function toModelAggregate(data: AthenaData): ModelStatAggregate[] {
+  const model = statModel(data.model, data.provider_model)
+  const provider = statProvider(model, data.provider_model, data.provider)
+  if (!provider) return []
+
+  return toStatBaseAggregate(data).flatMap((base) => [
+    { ...base, provider, model, provider_model: data.provider_model || "" },
+  ])
+}
+
+export function toProviderAggregate(data: AthenaData): ProviderStatAggregate[] {
+  return toStatBaseAggregate(data).flatMap((base) => [
+    { ...base, provider: statProvider(data.model, data.provider_model, data.provider) || "unknown" },
+  ])
+}
+
+export function toGeoAggregate(data: AthenaData): GeoStatAggregate[] {
+  return toStatBaseAggregate(data).flatMap((base) => [
+    {
+      ...base,
+      provider: statProvider(data.model, data.provider_model, data.provider) || "all",
+      model: statModel(data.model || "all", data.provider_model),
+      country: normalizeCountry(data.country),
+      continent: data.continent || "",
+    },
+  ])
+}
+
+function toStatBaseAggregate(data: AthenaData): StatBaseAggregate[] {
+  const grain = data.grain === "day" || data.grain === "week" ? data.grain : undefined
+  if (!grain || !data.period_key) return []
+
+  return [
+    {
+      grain,
+      period_key: data.period_key,
+      dataset: data.dataset || Resource.StatsSyncConfig.dataset,
+      tier: normalizeTier(data.tier || "unknown"),
+      sessions: integer(data, "sessions"),
+      requests: integer(data, "requests"),
+      input_tokens: integer(data, "input_tokens"),
+      output_tokens: integer(data, "output_tokens"),
+      reasoning_tokens: integer(data, "reasoning_tokens"),
+      cache_read_tokens: integer(data, "cache_read_tokens"),
+      total_tokens: integer(data, "total_tokens"),
+      input_cost_microcents: integer(data, "input_cost_microcents"),
+      output_cost_microcents: integer(data, "output_cost_microcents"),
+      total_cost_microcents: integer(data, "total_cost_microcents"),
+      avg_duration_ms: nullableNumber(data, "avg_duration_ms"),
+      p50_duration_ms: nullableInteger(data, "p50_duration_ms"),
+      p95_duration_ms: nullableInteger(data, "p95_duration_ms"),
+      avg_ttfb_ms: nullableNumber(data, "avg_ttfb_ms"),
+      p50_ttfb_ms: nullableInteger(data, "p50_ttfb_ms"),
+      p95_ttfb_ms: nullableInteger(data, "p95_ttfb_ms"),
+      avg_output_tps: nullableNumber(data, "avg_output_tps"),
+      success_count: integer(data, "success_count"),
+      error_count: integer(data, "error_count"),
+      sample_count: integer(data, "sample_count"),
+    },
+  ]
+}
+
+function integer(data: AthenaData, key: string) {
+  return Math.round(number(data, key))
+}
+
+function nullableNumber(data: AthenaData, key: string) {
+  if (data[key] === undefined || data[key] === "") return null
+  return Number(number(data, key).toFixed(2))
+}
+
+function nullableInteger(data: AthenaData, key: string) {
+  if (data[key] === undefined || data[key] === "") return null
+  return Math.round(number(data, key))
+}
+
+function number(data: AthenaData, key: string) {
+  const value = Number(data[key])
+  return Number.isFinite(value) ? value : 0
+}
+
+function sqlIdentifier(value: string) {
+  return `"${value.replace(/"/g, '""')}"`
+}
+
+function sqlString(value: string) {
+  return `'${value.replace(/'/g, "''")}'`
+}
+
+function statModelSql(model: string, providerModel: string) {
+  return `COALESCE(NULLIF(regexp_replace(CASE
+      WHEN lower(${model}) = 'big-pickle' THEN NULLIF(${providerModel}, '')
+      ELSE ${model}
+    END, '(-free|:global)+$', ''), ''), 'unknown')`
+}
+
+function statProviderSql(model: string, providerModel: string, provider: string) {
+  return `CASE
+${MODEL_AUTHOR_RULES.map((item) => `      WHEN strpos(lower(${providerModel}), ${sqlString(item.match)}) > 0 THEN ${sqlString(item.author)}`).join("\n")}
+${MODEL_AUTHOR_RULES.map((item) => `      WHEN strpos(lower(${model}), ${sqlString(item.match)}) > 0 THEN ${sqlString(item.author)}`).join("\n")}
+      WHEN ${provider} <> '' AND lower(${provider}) NOT IN (${RETIRED_STAT_PROVIDERS.map(sqlString).join(", ")}) THEN ${provider}
+      ELSE 'unknown'
+    END`
+}

packages/stats/core/src/domain/model-normalization.ts

@@ -0,0 +1,49 @@
+export const MODEL_AUTHOR_RULES = [
+  { match: "claude", author: "anthropic" },
+  { match: "gemini", author: "google" },
+  { match: "deepseek", author: "deepseek" },
+  { match: "glm", author: "zhipu" },
+  { match: "gpt", author: "openai" },
+  { match: "grok", author: "xai" },
+  { match: "hy3", author: "tencent" },
+  { match: "kimi", author: "moonshot" },
+  { match: "mimo", author: "xiaomi" },
+  { match: "minimax", author: "minimax" },
+  { match: "nemotron", author: "nvidia" },
+  { match: "qwen", author: "qwen" },
+] as const
+export const EXCLUDED_MODELS = new Set(["alpha-gpt-next"])
+export const RETIRED_STAT_MODELS = ["big-pickle"]
+export const RETIRED_STAT_PROVIDERS = ["opencode"]
+
+export function normalizeInferenceModel(value: string | undefined) {
+  return (value || "unknown").replace(/(-free|:global)+$/, "") || "unknown"
+}
+
+export function modelAuthor(value: string | undefined) {
+  const model = normalizeInferenceModel(value).toLowerCase()
+  if (EXCLUDED_MODELS.has(model)) return undefined
+
+  return MODEL_AUTHOR_RULES.find((item) => model.includes(item.match))?.author ?? "unknown"
+}
+
+export function statModel(model: string | undefined, providerModel: string | undefined) {
+  const normalized = normalizeInferenceModel(model)
+  if (RETIRED_STAT_MODELS.includes(normalized.toLowerCase())) return normalizeInferenceModel(providerModel)
+  return normalized
+}
+
+export function statProvider(
+  model: string | undefined,
+  providerModel: string | undefined,
+  provider: string | undefined,
+) {
+  const modelAuthorValue = modelAuthor(statModel(model, providerModel))
+  if (!modelAuthorValue) return undefined
+
+  const providerModelAuthor = modelAuthor(providerModel)
+  if (providerModelAuthor && providerModelAuthor !== "unknown") return providerModelAuthor
+  if (modelAuthorValue !== "unknown") return modelAuthorValue
+  if (provider && !RETIRED_STAT_PROVIDERS.includes(provider.toLowerCase())) return provider
+  return modelAuthorValue
+}

packages/stats/core/src/domain/model.ts

@@ -0,0 +1,202 @@
+import { and, asc, eq, inArray, or } from "drizzle-orm"
+import { Effect, Layer } from "effect"
+import * as Context from "effect/Context"
+import { DatabaseError, DrizzleClient } from "../database"
+import { modelStat } from "../database/schema"
+import { RETIRED_STAT_MODELS, RETIRED_STAT_PROVIDERS } from "./model-normalization"
+import {
+  chunks,
+  collapseRows,
+  inserted,
+  rankBy,
+  statPeriodKey,
+  statRowScope,
+  synthesizeAllTierRows,
+  toStatBaseRow,
+  UPSERT_CHUNK_SIZE,
+  type StatBaseAggregate,
+} from "./stat"
+
+export type ModelStatRow = typeof modelStat.$inferInsert
+export type ModelStatAggregate = StatBaseAggregate & { provider: string; model: string; provider_model: string }
+
+export type ModelStatMetric = {
+  periodKey: string
+  updatedAt: Date
+  tier: string
+  provider: string
+  model: string
+  sessions: number
+  inputTokens: number
+  outputTokens: number
+  reasoningTokens: number
+  cacheReadTokens: number
+  totalTokens: number
+  inputCostMicrocents: number
+  outputCostMicrocents: number
+  totalCostMicrocents: number
+}
+
+export declare namespace ModelStatRepo {
+  export interface Service {
+    readonly listDaily: () => Effect.Effect<ModelStatMetric[], DatabaseError>
+    readonly upsert: (rows: ModelStatRow[]) => Effect.Effect<void, DatabaseError>
+    readonly deleteRetiredDimensions: (rows: ModelStatRow[]) => Effect.Effect<void, DatabaseError>
+  }
+}
+
+export class ModelStatRepo extends Context.Service<ModelStatRepo, ModelStatRepo.Service>()(
+  "@opencode/stats/ModelStatRepo",
+) {
+  static readonly layer: Layer.Layer<ModelStatRepo, never, DrizzleClient> = Layer.effect(
+    ModelStatRepo,
+    Effect.gen(function* () {
+      const db = yield* DrizzleClient
+
+      const listDaily = Effect.fn("ModelStatRepo.listDaily")(function* () {
+        return yield* Effect.tryPromise({
+          try: () =>
+            db
+              .select({
+                periodKey: modelStat.period_key,
+                updatedAt: modelStat.updated_at,
+                tier: modelStat.tier,
+                provider: modelStat.provider,
+                model: modelStat.model,
+                sessions: modelStat.sessions,
+                inputTokens: modelStat.input_tokens,
+                outputTokens: modelStat.output_tokens,
+                reasoningTokens: modelStat.reasoning_tokens,
+                cacheReadTokens: modelStat.cache_read_tokens,
+                totalTokens: modelStat.total_tokens,
+                inputCostMicrocents: modelStat.input_cost_microcents,
+                outputCostMicrocents: modelStat.output_cost_microcents,
+                totalCostMicrocents: modelStat.total_cost_microcents,
+              })
+              .from(modelStat)
+              .where(and(eq(modelStat.grain, "day"), eq(modelStat.client, "all"), eq(modelStat.source, "all")))
+              .orderBy(asc(modelStat.period_key)),
+          catch: (cause) => DatabaseError.make({ cause }),
+        })
+      })
+
+      const upsert = Effect.fn("ModelStatRepo.upsert")(function* (rows: ModelStatRow[]) {
+        yield* Effect.forEach(
+          chunks(rows, UPSERT_CHUNK_SIZE),
+          (chunk) =>
+            Effect.tryPromise({
+              try: () =>
+                db
+                  .insert(modelStat)
+                  .values(chunk)
+                  .onDuplicateKeyUpdate({
+                    set: {
+                      provider_model: inserted("provider_model"),
+                      sessions: inserted("sessions"),
+                      requests: inserted("requests"),
+                      input_tokens: inserted("input_tokens"),
+                      output_tokens: inserted("output_tokens"),
+                      reasoning_tokens: inserted("reasoning_tokens"),
+                      cache_read_tokens: inserted("cache_read_tokens"),
+                      total_tokens: inserted("total_tokens"),
+                      input_cost_microcents: inserted("input_cost_microcents"),
+                      output_cost_microcents: inserted("output_cost_microcents"),
+                      total_cost_microcents: inserted("total_cost_microcents"),
+                      avg_duration_ms: inserted("avg_duration_ms"),
+                      p50_duration_ms: inserted("p50_duration_ms"),
+                      p95_duration_ms: inserted("p95_duration_ms"),
+                      avg_ttfb_ms: inserted("avg_ttfb_ms"),
+                      p50_ttfb_ms: inserted("p50_ttfb_ms"),
+                      p95_ttfb_ms: inserted("p95_ttfb_ms"),
+                      avg_output_tps: inserted("avg_output_tps"),
+                      success_count: inserted("success_count"),
+                      error_count: inserted("error_count"),
+                      sample_count: inserted("sample_count"),
+                      rank_by_tokens: inserted("rank_by_tokens"),
+                      rank_by_requests: inserted("rank_by_requests"),
+                      rank_by_cost: inserted("rank_by_cost"),
+                    },
+                  }),
+              catch: (cause) => DatabaseError.make({ cause }),
+            }),
+          { discard: true },
+        )
+      })
+
+      const deleteRetiredDimensions = Effect.fn("ModelStatRepo.deleteRetiredDimensions")(function* (
+        rows: ModelStatRow[],
+      ) {
+        const scope = statRowScope(rows)
+        if (!scope) return
+
+        yield* Effect.tryPromise({
+          try: () =>
+            db
+              .delete(modelStat)
+              .where(
+                and(
+                  inArray(modelStat.grain, scope.grains),
+                  inArray(modelStat.period_key, scope.periodKeys),
+                  inArray(modelStat.dataset, scope.datasets),
+                  inArray(modelStat.client, scope.clients),
+                  inArray(modelStat.source, scope.sources),
+                  or(
+                    inArray(modelStat.provider, RETIRED_STAT_PROVIDERS),
+                    inArray(modelStat.model, RETIRED_STAT_MODELS),
+                  ),
+                ),
+              ),
+          catch: (cause) => DatabaseError.make({ cause }),
+        })
+      })
+
+      return ModelStatRepo.of({ listDaily, upsert, deleteRetiredDimensions })
+    }),
+  )
+}
+
+export function rowsFromAggregates(aggregates: ModelStatAggregate[]) {
+  return rankRows([
+    ...synthesizeAllTierRows(
+      collapseRows(aggregates.filter((item) => item.grain === "week").map(toRow), dimensionKey),
+      dimensionKey,
+    ),
+    ...synthesizeAllTierRows(
+      collapseRows(aggregates.filter((item) => item.grain === "day").map(toRow), dimensionKey),
+      dimensionKey,
+    ),
+  ])
+}
+
+function toRow(data: ModelStatAggregate): ModelStatRow {
+  return {
+    ...toStatBaseRow(data),
+    provider: data.provider,
+    model: data.model,
+    provider_model: data.provider_model,
+  }
+}
+
+function rankRows(rows: ModelStatRow[]) {
+  return Object.values(
+    rows.reduce<Record<string, ModelStatRow[]>>((result, row) => {
+      const key = statPeriodKey(row)
+      result[key] = [...(result[key] ?? []), row]
+      return result
+    }, {}),
+  ).flatMap((group) => {
+    const tokenRanks = rankBy(group, (row) => row.total_tokens ?? 0)
+    const requestRanks = rankBy(group, (row) => row.requests ?? 0)
+    const costRanks = rankBy(group, (row) => row.total_cost_microcents ?? 0)
+    return group.map((row) => ({
+      ...row,
+      rank_by_tokens: tokenRanks.get(row) ?? null,
+      rank_by_requests: requestRanks.get(row) ?? null,
+      rank_by_cost: costRanks.get(row) ?? null,
+    }))
+  })
+}
+
+function dimensionKey(row: ModelStatRow) {
+  return [row.provider, row.model].join("\u0000")
+}

packages/stats/core/src/domain/provider.ts

@@ -0,0 +1,195 @@
+import { and, asc, eq, inArray } from "drizzle-orm"
+import { Effect, Layer } from "effect"
+import * as Context from "effect/Context"
+import { DatabaseError, DrizzleClient } from "../database"
+import { providerStat } from "../database/schema"
+import { RETIRED_STAT_PROVIDERS } from "./model-normalization"
+import {
+  chunks,
+  collapseRows,
+  inserted,
+  rankRowsWithMarketShare,
+  statRowScope,
+  synthesizeAllTierRows,
+  toStatBaseRow,
+  UPSERT_CHUNK_SIZE,
+  type StatBaseAggregate,
+} from "./stat"
+
+export type ProviderStatRow = typeof providerStat.$inferInsert
+export type ProviderStatAggregate = StatBaseAggregate & { provider: string }
+export type ProviderStatMetric = {
+  periodKey: string
+  updatedAt: Date
+  tier: string
+  provider: string
+  totalTokens: number
+}
+
+export declare namespace ProviderStatRepo {
+  export interface Service {
+    readonly listDaily: () => Effect.Effect<ProviderStatMetric[], DatabaseError>
+    readonly listByPeriod: (opts: {
+      readonly grain: string
+      readonly periodKey: string
+      readonly dataset?: string
+      readonly tier?: string
+      readonly client?: string
+      readonly source?: string
+    }) => Effect.Effect<ProviderStatRow[], DatabaseError>
+    readonly upsert: (rows: ProviderStatRow[]) => Effect.Effect<void, DatabaseError>
+    readonly deleteRetiredDimensions: (rows: ProviderStatRow[]) => Effect.Effect<void, DatabaseError>
+  }
+}
+
+export class ProviderStatRepo extends Context.Service<ProviderStatRepo, ProviderStatRepo.Service>()(
+  "@opencode/stats/ProviderStatRepo",
+) {
+  static readonly layer: Layer.Layer<ProviderStatRepo, never, DrizzleClient> = Layer.effect(
+    ProviderStatRepo,
+    Effect.gen(function* () {
+      const db = yield* DrizzleClient
+
+      const listDaily = Effect.fn("ProviderStatRepo.listDaily")(function* () {
+        return yield* Effect.tryPromise({
+          try: () =>
+            db
+              .select({
+                periodKey: providerStat.period_key,
+                updatedAt: providerStat.updated_at,
+                tier: providerStat.tier,
+                provider: providerStat.provider,
+                totalTokens: providerStat.total_tokens,
+              })
+              .from(providerStat)
+              .where(and(eq(providerStat.grain, "day"), eq(providerStat.client, "all"), eq(providerStat.source, "all")))
+              .orderBy(asc(providerStat.period_key)),
+          catch: (cause) => DatabaseError.make({ cause }),
+        })
+      })
+
+      const listByPeriod = Effect.fn("ProviderStatRepo.listByPeriod")(function* (opts: {
+        readonly grain: string
+        readonly periodKey: string
+        readonly dataset?: string
+        readonly tier?: string
+        readonly client?: string
+        readonly source?: string
+      }) {
+        return yield* Effect.tryPromise({
+          try: () =>
+            db
+              .select()
+              .from(providerStat)
+              .where(
+                and(
+                  eq(providerStat.grain, opts.grain),
+                  eq(providerStat.period_key, opts.periodKey),
+                  eq(providerStat.dataset, opts.dataset ?? "zen"),
+                  eq(providerStat.tier, opts.tier ?? "all"),
+                  eq(providerStat.client, opts.client ?? "all"),
+                  eq(providerStat.source, opts.source ?? "all"),
+                ),
+              ),
+          catch: (cause) => DatabaseError.make({ cause }),
+        })
+      })
+
+      const upsert = Effect.fn("ProviderStatRepo.upsert")(function* (rows: ProviderStatRow[]) {
+        yield* Effect.forEach(
+          chunks(rows, UPSERT_CHUNK_SIZE),
+          (chunk) =>
+            Effect.tryPromise({
+              try: () =>
+                db
+                  .insert(providerStat)
+                  .values(chunk)
+                  .onDuplicateKeyUpdate({
+                    set: {
+                      sessions: inserted("sessions"),
+                      requests: inserted("requests"),
+                      input_tokens: inserted("input_tokens"),
+                      output_tokens: inserted("output_tokens"),
+                      reasoning_tokens: inserted("reasoning_tokens"),
+                      cache_read_tokens: inserted("cache_read_tokens"),
+                      total_tokens: inserted("total_tokens"),
+                      input_cost_microcents: inserted("input_cost_microcents"),
+                      output_cost_microcents: inserted("output_cost_microcents"),
+                      total_cost_microcents: inserted("total_cost_microcents"),
+                      avg_duration_ms: inserted("avg_duration_ms"),
+                      p50_duration_ms: inserted("p50_duration_ms"),
+                      p95_duration_ms: inserted("p95_duration_ms"),
+                      avg_ttfb_ms: inserted("avg_ttfb_ms"),
+                      p50_ttfb_ms: inserted("p50_ttfb_ms"),
+                      p95_ttfb_ms: inserted("p95_ttfb_ms"),
+                      avg_output_tps: inserted("avg_output_tps"),
+                      success_count: inserted("success_count"),
+                      error_count: inserted("error_count"),
+                      sample_count: inserted("sample_count"),
+                      market_share_tokens: inserted("market_share_tokens"),
+                      market_share_requests: inserted("market_share_requests"),
+                      market_share_sessions: inserted("market_share_sessions"),
+                      rank_by_tokens: inserted("rank_by_tokens"),
+                      rank_by_requests: inserted("rank_by_requests"),
+                      rank_by_sessions: inserted("rank_by_sessions"),
+                      rank_by_cost: inserted("rank_by_cost"),
+                    },
+                  }),
+              catch: (cause) => DatabaseError.make({ cause }),
+            }),
+          { discard: true },
+        )
+      })
+
+      const deleteRetiredDimensions = Effect.fn("ProviderStatRepo.deleteRetiredDimensions")(function* (
+        rows: ProviderStatRow[],
+      ) {
+        const scope = statRowScope(rows)
+        if (!scope) return
+
+        yield* Effect.tryPromise({
+          try: () =>
+            db
+              .delete(providerStat)
+              .where(
+                and(
+                  inArray(providerStat.grain, scope.grains),
+                  inArray(providerStat.period_key, scope.periodKeys),
+                  inArray(providerStat.dataset, scope.datasets),
+                  inArray(providerStat.client, scope.clients),
+                  inArray(providerStat.source, scope.sources),
+                  inArray(providerStat.provider, RETIRED_STAT_PROVIDERS),
+                ),
+              ),
+          catch: (cause) => DatabaseError.make({ cause }),
+        })
+      })
+
+      return ProviderStatRepo.of({ listDaily, listByPeriod, upsert, deleteRetiredDimensions })
+    }),
+  )
+}
+
+export function rowsFromAggregates(aggregates: ProviderStatAggregate[]) {
+  return rankRowsWithMarketShare([
+    ...synthesizeAllTierRows(
+      collapseRows(aggregates.filter((item) => item.grain === "week").map(toRow), dimensionKey),
+      dimensionKey,
+    ),
+    ...synthesizeAllTierRows(
+      collapseRows(aggregates.filter((item) => item.grain === "day").map(toRow), dimensionKey),
+      dimensionKey,
+    ),
+  ])
+}
+
+function toRow(data: ProviderStatAggregate): ProviderStatRow {
+  return {
+    ...toStatBaseRow(data),
+    provider: data.provider,
+  }
+}
+
+function dimensionKey(row: ProviderStatRow) {
+  return row.provider
+}

packages/stats/core/src/domain/stat.ts

@@ -0,0 +1,261 @@
+import { sql } from "drizzle-orm"
+
+export const UPSERT_CHUNK_SIZE = 500
+const DAY_MS = 86_400_000
+
+export type StatGrain = "day" | "week"
+
+export type StatBaseAggregate = {
+  grain: StatGrain
+  period_key: string
+  dataset: string
+  tier: string
+  sessions: number
+  requests: number
+  input_tokens: number
+  output_tokens: number
+  reasoning_tokens: number
+  cache_read_tokens: number
+  total_tokens: number
+  input_cost_microcents: number
+  output_cost_microcents: number
+  total_cost_microcents: number
+  avg_duration_ms: number | null
+  p50_duration_ms: number | null
+  p95_duration_ms: number | null
+  avg_ttfb_ms: number | null
+  p50_ttfb_ms: number | null
+  p95_ttfb_ms: number | null
+  avg_output_tps: number | null
+  success_count: number
+  error_count: number
+  sample_count: number
+}
+
+export type StatBaseRow = {
+  grain: string
+  period_key: string
+  dataset?: string
+  tier?: string
+  client?: string
+  source?: string
+  sessions?: number
+  requests?: number
+  input_tokens?: number
+  output_tokens?: number
+  reasoning_tokens?: number
+  cache_read_tokens?: number
+  total_tokens?: number
+  input_cost_microcents?: number
+  output_cost_microcents?: number
+  total_cost_microcents?: number
+  avg_duration_ms?: number | null
+  p50_duration_ms?: number | null
+  p95_duration_ms?: number | null
+  avg_ttfb_ms?: number | null
+  p50_ttfb_ms?: number | null
+  p95_ttfb_ms?: number | null
+  avg_output_tps?: number | null
+  success_count?: number
+  error_count?: number
+  sample_count?: number
+}
+
+export function toStatBaseRow(data: StatBaseAggregate) {
+  return {
+    grain: data.grain,
+    period_key: data.period_key,
+    dataset: data.dataset,
+    tier: data.tier,
+    client: "all",
+    source: "all",
+    sessions: data.sessions,
+    requests: data.requests,
+    input_tokens: data.input_tokens,
+    output_tokens: data.output_tokens,
+    reasoning_tokens: data.reasoning_tokens,
+    cache_read_tokens: data.cache_read_tokens,
+    total_tokens: data.total_tokens,
+    input_cost_microcents: data.input_cost_microcents,
+    output_cost_microcents: data.output_cost_microcents,
+    total_cost_microcents: data.total_cost_microcents,
+    avg_duration_ms: data.avg_duration_ms,
+    p50_duration_ms: data.p50_duration_ms,
+    p95_duration_ms: data.p95_duration_ms,
+    avg_ttfb_ms: data.avg_ttfb_ms,
+    p50_ttfb_ms: data.p50_ttfb_ms,
+    p95_ttfb_ms: data.p95_ttfb_ms,
+    avg_output_tps: data.avg_output_tps,
+    success_count: data.success_count,
+    error_count: data.error_count,
+    sample_count: data.sample_count,
+  }
+}
+
+export function synthesizeAllTierRows<T extends StatBaseRow>(rows: T[], dimensionKey: (row: T) => string) {
+  return [
+    ...rows,
+    ...Object.values(
+      rows.reduce<Record<string, T>>((result, row) => {
+        const key = [row.grain, row.period_key, row.dataset, row.client, row.source, dimensionKey(row)].join("\u0000")
+        result[key] = result[key] ? combineRows(result[key], row) : { ...row, tier: "all" }
+        return result
+      }, {}),
+    ),
+  ]
+}
+
+export function collapseRows<T extends StatBaseRow>(rows: T[], dimensionKey: (row: T) => string) {
+  return Object.values(
+    rows.reduce<Record<string, T>>((result, row) => {
+      const key = [row.grain, row.period_key, row.dataset, row.tier, row.client, row.source, dimensionKey(row)].join(
+        "\u0000",
+      )
+      result[key] = result[key] ? combineRows(result[key], row) : row
+      return result
+    }, {}),
+  )
+}
+
+export function combineRows<T extends StatBaseRow>(left: T, right: T): T {
+  return {
+    ...left,
+    sessions: (left.sessions ?? 0) + (right.sessions ?? 0),
+    requests: (left.requests ?? 0) + (right.requests ?? 0),
+    input_tokens: (left.input_tokens ?? 0) + (right.input_tokens ?? 0),
+    output_tokens: (left.output_tokens ?? 0) + (right.output_tokens ?? 0),
+    reasoning_tokens: (left.reasoning_tokens ?? 0) + (right.reasoning_tokens ?? 0),
+    cache_read_tokens: (left.cache_read_tokens ?? 0) + (right.cache_read_tokens ?? 0),
+    total_tokens: (left.total_tokens ?? 0) + (right.total_tokens ?? 0),
+    input_cost_microcents: (left.input_cost_microcents ?? 0) + (right.input_cost_microcents ?? 0),
+    output_cost_microcents: (left.output_cost_microcents ?? 0) + (right.output_cost_microcents ?? 0),
+    total_cost_microcents: (left.total_cost_microcents ?? 0) + (right.total_cost_microcents ?? 0),
+    avg_duration_ms: weightedAverage(left.avg_duration_ms, left.requests, right.avg_duration_ms, right.requests),
+    p50_duration_ms: null,
+    p95_duration_ms: null,
+    avg_ttfb_ms: weightedAverage(left.avg_ttfb_ms, left.requests, right.avg_ttfb_ms, right.requests),
+    p50_ttfb_ms: null,
+    p95_ttfb_ms: null,
+    avg_output_tps: weightedAverage(left.avg_output_tps, left.requests, right.avg_output_tps, right.requests),
+    success_count: (left.success_count ?? 0) + (right.success_count ?? 0),
+    error_count: (left.error_count ?? 0) + (right.error_count ?? 0),
+    sample_count: (left.sample_count ?? 0) + (right.sample_count ?? 0),
+  }
+}
+
+export function statPeriodKey(row: StatBaseRow) {
+  return [row.grain, row.period_key, row.dataset, row.tier, row.client, row.source].join("\u0000")
+}
+
+export function statRowScope(rows: StatBaseRow[]) {
+  if (rows.length === 0) return
+  return {
+    grains: unique(rows.map((row) => row.grain)),
+    periodKeys: unique(rows.map((row) => row.period_key)),
+    datasets: unique(rows.map((row) => row.dataset ?? "all")),
+    clients: unique(rows.map((row) => row.client ?? "all")),
+    sources: unique(rows.map((row) => row.source ?? "all")),
+  }
+}
+
+export function periodKeyFor(grain: StatGrain, periodStart: Date) {
+  if (grain === "week") return isoWeekId(periodStart)
+  return utcDateId(periodStart)
+}
+
+export function startOfUtcDay(value: Date) {
+  return new Date(Date.UTC(value.getUTCFullYear(), value.getUTCMonth(), value.getUTCDate()))
+}
+
+export function startOfIsoWeek(value: Date) {
+  return new Date(
+    Date.UTC(value.getUTCFullYear(), value.getUTCMonth(), value.getUTCDate() - (value.getUTCDay() || 7) + 1),
+  )
+}
+
+export function isoWeekId(value: Date) {
+  const thursday = new Date(
+    Date.UTC(value.getUTCFullYear(), value.getUTCMonth(), value.getUTCDate() + 4 - (value.getUTCDay() || 7)),
+  )
+  return `${thursday.getUTCFullYear()}-W${String(Math.ceil(((thursday.getTime() - Date.UTC(thursday.getUTCFullYear(), 0, 1)) / DAY_MS + 1) / 7)).padStart(2, "0")}`
+}
+
+function utcDateId(value: Date) {
+  return `${value.getUTCFullYear()}-${String(value.getUTCMonth() + 1).padStart(2, "0")}-${String(value.getUTCDate()).padStart(2, "0")}`
+}
+
+export function rankBy<T extends StatBaseRow>(rows: T[], value: (row: T) => number) {
+  return new Map(rows.toSorted((a, b) => value(b) - value(a)).map((row, index) => [row, index + 1]))
+}
+
+export function rankRowsWithMarketShare<T extends StatBaseRow>(
+  rows: T[],
+  groupKey: (row: T) => string = statPeriodKey,
+) {
+  return Object.values(
+    rows.reduce<Record<string, T[]>>((result, row) => {
+      const key = groupKey(row)
+      result[key] = [...(result[key] ?? []), row]
+      return result
+    }, {}),
+  ).flatMap((group) => {
+    const tokens = group.reduce((sum, row) => sum + (row.total_tokens ?? 0), 0)
+    const requests = group.reduce((sum, row) => sum + (row.requests ?? 0), 0)
+    const sessions = group.reduce((sum, row) => sum + (row.sessions ?? 0), 0)
+    const tokenRanks = rankBy(group, (row) => row.total_tokens ?? 0)
+    const requestRanks = rankBy(group, (row) => row.requests ?? 0)
+    const sessionRanks = rankBy(group, (row) => row.sessions ?? 0)
+    const costRanks = rankBy(group, (row) => row.total_cost_microcents ?? 0)
+    return group.map((row) => ({
+      ...row,
+      market_share_tokens: share(row.total_tokens, tokens),
+      market_share_requests: share(row.requests, requests),
+      market_share_sessions: share(row.sessions, sessions),
+      rank_by_tokens: tokenRanks.get(row) ?? null,
+      rank_by_requests: requestRanks.get(row) ?? null,
+      rank_by_sessions: sessionRanks.get(row) ?? null,
+      rank_by_cost: costRanks.get(row) ?? null,
+    }))
+  })
+}
+
+export function share(value: number | null | undefined, total: number) {
+  if (total <= 0) return null
+  return Number(((value ?? 0) / total).toFixed(6))
+}
+
+export function chunks<T>(items: T[], size: number) {
+  return Array.from({ length: Math.ceil(items.length / size) }, (_, index) =>
+    items.slice(index * size, (index + 1) * size),
+  )
+}
+
+function unique(values: string[]) {
+  return [...new Set(values)]
+}
+
+export function inserted(column: string) {
+  return sql.raw(`values(\`${column}\`)`)
+}
+
+export function weightedAverage(
+  left: number | null | undefined,
+  leftWeight = 0,
+  right: number | null | undefined,
+  rightWeight = 0,
+) {
+  const totalWeight =
+    (left === null || left === undefined ? 0 : leftWeight) + (right === null || right === undefined ? 0 : rightWeight)
+  if (totalWeight === 0) return null
+  return Number((((left ?? 0) * leftWeight + (right ?? 0) * rightWeight) / totalWeight).toFixed(2))
+}
+
+export function normalizeTier(value: string) {
+  if (value === "Paid") return "Zen"
+  return value
+}
+
+export function normalizeCountry(value: string | undefined) {
+  if (!value || value.length !== 2) return "ZZ"
+  return value.toUpperCase()
+}

packages/stats/core/src/honeycomb-backfill.ts

@@ -0,0 +1,993 @@
+import { Client } from "@planetscale/database"
+import { readdir } from "node:fs/promises"
+import path from "node:path"
+import { drizzle } from "drizzle-orm/planetscale-serverless"
+import { geoStat, modelStat, providerStat } from "./database/schema"
+import { statModel, statProvider } from "./domain/model-normalization"
+import {
+  chunks,
+  collapseRows,
+  inserted,
+  isoWeekId,
+  normalizeCountry,
+  normalizeTier,
+  periodKeyFor,
+  rankBy,
+  rankRowsWithMarketShare,
+  startOfIsoWeek,
+  startOfUtcDay,
+  statPeriodKey,
+  synthesizeAllTierRows,
+  toStatBaseRow,
+  type StatBaseAggregate,
+} from "./domain/stat"
+
+const DAY_MS = 86_400_000
+const DEFAULT_UPSERT_CHUNK_SIZE = 100
+const DEFAULT_TIERS = ["Go", "Free", "Paid"]
+const FREE_MODELS = new Set(["gpt-5-nano", "grok-code", "big-pickle"])
+
+type Grain = "day" | "week"
+type MetricDimension = "model" | "provider" | "geo" | "geo-model"
+type LookupDimension = "model-provider-model" | "geo-continent"
+type ImportKey = `${MetricDimension | LookupDimension}-${Grain}`
+type QuerySpec = {
+  name: string
+  importKey: ImportKey
+  importFlag: `--${ImportKey}`
+  query: ReturnType<typeof metricQuery>
+}
+type RawRow = Record<string, string>
+type ImportOptions = {
+  dataset: string
+  databaseUrl: string | undefined
+  directories: string[]
+  dryRun: boolean
+  periodStart: Date | undefined
+  upsertChunkSize: number
+  files: Partial<Record<ImportKey, string[]>>
+}
+type ModelAggregate = StatBaseAggregate & { provider: string; model: string; provider_model: string }
+type ProviderAggregate = StatBaseAggregate & { provider: string }
+type GeoAggregate = StatBaseAggregate & { provider: string; model: string; country: string; continent: string }
+type ModelStatRow = typeof modelStat.$inferInsert
+type ProviderStatRow = typeof providerStat.$inferInsert
+type GeoStatRow = typeof geoStat.$inferInsert
+
+const inputKeys = [
+  "model-day",
+  "model-week",
+  "model-provider-model-day",
+  "model-provider-model-week",
+  "provider-day",
+  "provider-week",
+  "geo-day",
+  "geo-week",
+  "geo-model-day",
+  "geo-model-week",
+  "geo-continent-day",
+  "geo-continent-week",
+] as const satisfies ImportKey[]
+
+if (import.meta.main) await main()
+
+async function main() {
+  const command = process.argv[2]
+  if (command === "queries") return printQueries(process.argv.slice(3))
+  if (command === "import") return importFiles(process.argv.slice(3))
+  usage()
+}
+
+function printQueries(args: string[]) {
+  const flags = parseFlags(args)
+  const limit = parseIntegerFlag(flags, "limit") ?? 1000
+  const tiers = parseListFlag(flags, "tiers") ?? DEFAULT_TIERS
+  const queries = buildQueries(limit, tiers)
+  const only = flags.get("only")?.[0]
+
+  if (only) {
+    const item = queries.find((query) => query.name === only)
+    if (!item) fail(`Unknown --only ${only}. Expected one of: ${queries.map((query) => query.name).join(", ")}`)
+    console.log(JSON.stringify(item.query, null, 2))
+    return
+  }
+
+  console.log(
+    JSON.stringify(
+      {
+        tiers,
+        import_hint: "bun src/honeycomb-backfill.ts import --dir downloads",
+        queries,
+      },
+      null,
+      2,
+    ),
+  )
+}
+
+async function importFiles(args: string[]) {
+  const parsed = parseImportOptions(args)
+  const opts = { ...parsed, files: mergeFiles(parsed.files, await discoverFiles(parsed.directories)) }
+  if (!inputKeys.some((key) => opts.files[key]?.length)) fail("No CSV or JSON import files were provided or discovered")
+  const providerModelLookup = new Map([
+    ...(await lookupRows(opts.files["model-provider-model-day"], "day", opts, modelProviderModelLookup)),
+    ...(await lookupRows(opts.files["model-provider-model-week"], "week", opts, modelProviderModelLookup)),
+  ])
+  const continentLookup = new Map([
+    ...(await lookupRows(opts.files["geo-continent-day"], "day", opts, geoContinentLookup)),
+    ...(await lookupRows(opts.files["geo-continent-week"], "week", opts, geoContinentLookup)),
+  ])
+  const modelAggregates = [
+    ...(await metricRows(opts.files["model-day"], "day", opts, (row, base) =>
+      modelAggregate(row, base, providerModelLookup),
+    )),
+    ...(await metricRows(opts.files["model-week"], "week", opts, (row, base) =>
+      modelAggregate(row, base, providerModelLookup),
+    )),
+  ]
+  const modelRows = modelRowsFromAggregates(modelAggregates)
+  const providerRows = providerRowsFromAggregates([
+    ...(await metricRows(opts.files["provider-day"], "day", opts, (row, base) => ({
+      ...base,
+      provider: provider(row) ?? "unknown",
+    }))),
+    ...(await metricRows(opts.files["provider-week"], "week", opts, (row, base) => ({
+      ...base,
+      provider: provider(row) ?? "unknown",
+    }))),
+  ])
+  const geoRows = geoRowsFromAggregates([
+    ...(await metricRows(opts.files["geo-day"], "day", opts, (row, base) => ({
+      ...base,
+      provider: "all",
+      model: "all",
+      country: country(row),
+      continent: continentLookup.get(lookupKey(base, country(row))) ?? continent(row),
+    }))),
+    ...(await metricRows(opts.files["geo-week"], "week", opts, (row, base) => ({
+      ...base,
+      provider: "all",
+      model: "all",
+      country: country(row),
+      continent: continentLookup.get(lookupKey(base, country(row))) ?? continent(row),
+    }))),
+    ...(await metricRows(opts.files["geo-model-day"], "day", opts, (row, base) =>
+      geoModelAggregate(row, base, continentLookup),
+    )),
+    ...(await metricRows(opts.files["geo-model-week"], "week", opts, (row, base) =>
+      geoModelAggregate(row, base, continentLookup),
+    )),
+  ])
+
+  console.log(
+    JSON.stringify(
+      {
+        inputs: Object.fromEntries(
+          inputKeys.flatMap((key) => (opts.files[key]?.length ? [[key, opts.files[key].length]] : [])),
+        ),
+        modelRows: modelRows.length,
+        providerRows: providerRows.length,
+        geoRows: geoRows.length,
+        dryRun: opts.dryRun,
+        upsertChunkSize: opts.upsertChunkSize,
+      },
+      null,
+      2,
+    ),
+  )
+
+  if (opts.dryRun) return
+  if (!opts.databaseUrl) fail("DATABASE_URL is required unless --dry-run is set")
+
+  const db = drizzle({ client: new Client({ url: opts.databaseUrl }) })
+  await upsertModelRows(db, modelRows, opts.upsertChunkSize)
+  await upsertProviderRows(db, providerRows, opts.upsertChunkSize)
+  await upsertGeoRows(db, geoRows, opts.upsertChunkSize)
+}
+
+function buildQueries(limit: number, tiers: string[]): QuerySpec[] {
+  const daily = tiers.flatMap((tier) => [
+    querySpec(
+      "model-day",
+      tier,
+      metricQuery(["date", "tier", "stat_provider_2", "stat_model_2"], limit, tierFilters(tier)),
+    ),
+    querySpec("provider-day", tier, metricQuery(["date", "tier", "stat_provider_2"], limit, tierFilters(tier))),
+    querySpec("geo-day", tier, metricQuery(["date", "tier", "country", "continent"], limit, tierFilters(tier))),
+    querySpec(
+      "geo-model-day",
+      tier,
+      metricQuery(
+        ["date", "tier", "stat_provider_2", "stat_model_2", "country", "continent"],
+        limit,
+        tierFilters(tier),
+      ),
+    ),
+  ])
+  const weekly = tiers.flatMap((tier) => [
+    querySpec(
+      "model-week",
+      tier,
+      metricQuery(["week", "tier", "stat_provider_2", "stat_model_2"], limit, tierFilters(tier)),
+    ),
+    querySpec("provider-week", tier, metricQuery(["week", "tier", "stat_provider_2"], limit, tierFilters(tier))),
+    querySpec("geo-week", tier, metricQuery(["week", "tier", "country", "continent"], limit, tierFilters(tier))),
+    querySpec(
+      "geo-model-week",
+      tier,
+      metricQuery(
+        ["week", "tier", "stat_provider_2", "stat_model_2", "country", "continent"],
+        limit,
+        tierFilters(tier),
+      ),
+    ),
+  ])
+
+  return [...daily, ...weekly]
+}
+
+function querySpec(importKey: ImportKey, tier: string, query: ReturnType<typeof metricQuery>) {
+  return {
+    name: `${importKey}-${queryNameSegment(tier)}`,
+    importKey,
+    importFlag: `--${importKey}` as const,
+    query,
+  }
+}
+
+function metricQuery(breakdowns: string[], limit: number, filters: ReturnType<typeof commonFilters> = []) {
+  return {
+    granularity: 0,
+    breakdowns,
+    calculations: [
+      { op: "COUNT_DISTINCT", column: "session" },
+      { op: "COUNT" },
+      { op: "SUM", column: "tokens.input" },
+      { op: "SUM", column: "tokens.output" },
+      { op: "SUM", column: "tokens.reasoning" },
+      { op: "SUM", column: "tokens.cache_read" },
+      { op: "SUM", column: "tokens" },
+      { op: "SUM", column: "cost.input.microcents" },
+      { op: "SUM", column: "cost.output.microcents" },
+      { op: "SUM", column: "cost.total.microcents" },
+      { op: "AVG", column: "duration" },
+      { op: "P50", column: "duration" },
+      { op: "P95", column: "duration" },
+      { op: "AVG", column: "time_to_first_byte" },
+      { op: "P50", column: "time_to_first_byte" },
+      { op: "P95", column: "time_to_first_byte" },
+      { op: "AVG", column: "tps.output" },
+    ],
+    filters: [...commonFilters(), ...filters],
+    filter_combination: "AND",
+    orders: [{ column: "tokens", op: "SUM", order: "descending" }],
+    havings: [],
+    limit,
+    formulas: [],
+  }
+}
+
+function tierFilters(tier: string) {
+  if (tier === "all") return []
+  return [{ column: "tier", op: "=", value: tier }]
+}
+
+function queryNameSegment(value: string) {
+  return (
+    value
+      .toLowerCase()
+      .replace(/[^a-z0-9]+/g, "-")
+      .replace(/^-|-$/g, "") || "all"
+  )
+}
+
+function commonFilters() {
+  return [
+    { column: "event_type", op: "=", value: "completions" },
+    { column: "model", op: "exists" },
+    { column: "model", op: "!=", value: "" },
+    { column: "model", op: "!=", value: "alpha-gpt-next" },
+  ]
+}
+
+function metricRows<T extends StatBaseAggregate>(
+  files: string[] | undefined,
+  grain: Grain,
+  opts: ImportOptions,
+  map: (row: RawRow, base: StatBaseAggregate) => T | T[],
+) {
+  if (!files) return Promise.resolve([])
+  return readFiles(files).then((rows) => rows.flatMap((row) => map(row, baseAggregate(row, grain, opts))))
+}
+
+function lookupRows(
+  files: string[] | undefined,
+  grain: Grain,
+  opts: ImportOptions,
+  map: (row: RawRow, grain: Grain, opts: ImportOptions) => readonly (readonly [string, string])[],
+) {
+  if (!files) return Promise.resolve([])
+  return readFiles(files).then((rows) =>
+    Array.from(
+      rows
+        .flatMap((row) => map(row, grain, opts))
+        .reduce((result, [key, value]) => {
+          if (value && value > (result.get(key) ?? "")) result.set(key, value)
+          return result
+        }, new Map<string, string>()),
+    ),
+  )
+}
+
+async function readFiles(files: string[]) {
+  return (await Promise.all(files.map(readRows))).flat()
+}
+
+async function discoverFiles(directories: string[]) {
+  const classified = await Promise.all(
+    (await Promise.all(directories.map(filesInDirectory))).flat().map(async (file) => ({
+      file,
+      key: classifyRows(file, await readRows(file)),
+    })),
+  )
+  return classified.reduce<Partial<Record<ImportKey, string[]>>>((result, item) => {
+    return { ...result, [item.key]: [...(result[item.key] ?? []), item.file] }
+  }, {})
+}
+
+async function filesInDirectory(directory: string): Promise<string[]> {
+  return (
+    await Promise.all(
+      (await readdir(directory, { withFileTypes: true })).map((entry) => {
+        const file = path.join(directory, entry.name)
+        if (entry.isDirectory()) return filesInDirectory(file)
+        if (entry.isFile() && /\.(csv|json)$/i.test(entry.name)) return Promise.resolve([file])
+        return Promise.resolve([])
+      }),
+    )
+  ).flat()
+}
+
+function classifyRows(file: string, rows: RawRow[]): ImportKey {
+  if (rows.length === 0) fail(`Cannot classify empty export: ${file}`)
+  const headers = new Set(rows.flatMap((row) => Object.keys(row).map(normalizeHeader)))
+  const grain: Grain = headers.has("date") ? "day" : "week"
+  if (hasHeader(headers, ["country", "cf.country"])) {
+    if (hasHeader(headers, ["model", "stat_model", "stat_model_2"]) && hasMetricHeaders(headers))
+      return `geo-model-${grain}`
+    return hasMetricHeaders(headers) ? `geo-${grain}` : `geo-continent-${grain}`
+  }
+  if (hasHeader(headers, ["model", "stat_model", "stat_model_2"]))
+    return hasMetricHeaders(headers) ? `model-${grain}` : `model-provider-model-${grain}`
+  if (
+    hasHeader(headers, [
+      "provider",
+      "provider.normalized",
+      "stat_provider",
+      "stat_provider_2",
+      "provider.model",
+      "provider_model",
+    ])
+  )
+    return `provider-${grain}`
+  fail(`Cannot classify export from columns in ${file}`)
+}
+
+function hasMetricHeaders(headers: Set<string>) {
+  return ["sumtokens", "sumtokensinput", "inputtokens", "totaltokens", "avgduration", "countdistinctsession"].some(
+    (header) => headers.has(header),
+  )
+}
+
+function hasHeader(headers: Set<string>, names: string[]) {
+  return names.some((name) => headers.has(normalizeHeader(name)))
+}
+
+function mergeFiles(left: Partial<Record<ImportKey, string[]>>, right: Partial<Record<ImportKey, string[]>>) {
+  return inputKeys.reduce<Partial<Record<ImportKey, string[]>>>((result, key) => {
+    const files = [...(left[key] ?? []), ...(right[key] ?? [])]
+    if (files.length === 0) return result
+    return { ...result, [key]: files }
+  }, {})
+}
+
+function modelProviderModelLookup(row: RawRow, grain: Grain, opts: ImportOptions): [string, string][] {
+  const base = basePeriod(row, grain, opts)
+  const value = providerModel(row)
+  const author = provider(row)
+  if (!value || !author) return []
+  return [[lookupKey({ ...base, dataset: opts.dataset, tier: tier(row), grain }, author, model(row)), value]]
+}
+
+function modelAggregate(
+  row: RawRow,
+  base: StatBaseAggregate,
+  providerModelLookup: Map<string, string>,
+): ModelAggregate[] {
+  const author = provider(row)
+  if (!author) return []
+
+  return [
+    {
+      ...base,
+      provider: author,
+      model: model(row),
+      provider_model: providerModelLookup.get(lookupKey(base, author, model(row))) ?? providerModel(row),
+    },
+  ]
+}
+
+function geoContinentLookup(row: RawRow, grain: Grain, opts: ImportOptions): [string, string][] {
+  const base = basePeriod(row, grain, opts)
+  const value = continent(row)
+  if (!value) return []
+  return [[lookupKey({ ...base, dataset: opts.dataset, tier: tier(row), grain }, country(row)), value]]
+}
+
+function geoModelAggregate(row: RawRow, base: StatBaseAggregate, continentLookup: Map<string, string>): GeoAggregate[] {
+  const author = provider(row)
+  if (!author) return []
+
+  return [
+    {
+      ...base,
+      provider: author,
+      model: model(row),
+      country: country(row),
+      continent: continentLookup.get(lookupKey(base, country(row))) ?? continent(row),
+    },
+  ]
+}
+
+function baseAggregate(row: RawRow, grain: Grain, opts: ImportOptions): StatBaseAggregate {
+  return {
+    ...basePeriod(row, grain, opts),
+    grain,
+    dataset: opts.dataset,
+    tier: tier(row),
+    sessions: integer(row, "sessions", ["COUNT_DISTINCT(session)"]),
+    requests: integer(row, "requests", ["COUNT", "COUNT()"]),
+    input_tokens: integer(row, "input_tokens", ["SUM(tokens.input)", "SUM(tokens_input)"]),
+    output_tokens: integer(row, "output_tokens", ["SUM(tokens.output)", "SUM(tokens_output)"]),
+    reasoning_tokens: integer(row, "reasoning_tokens", ["SUM(tokens.reasoning)", "SUM(tokens_reasoning)"]),
+    cache_read_tokens: integer(row, "cache_read_tokens", ["SUM(tokens.cache_read)", "SUM(tokens_cache_read)"]),
+    total_tokens: integer(row, "total_tokens", ["SUM(stat_tokens_total)", "SUM(tokens)", "SUM(tokens_total)"]),
+    input_cost_microcents: integer(row, "input_cost_microcents", [
+      "SUM(cost.input.microcents)",
+      "SUM(stat_cost_input_microcents)",
+    ]),
+    output_cost_microcents: integer(row, "output_cost_microcents", [
+      "SUM(cost.output.microcents)",
+      "SUM(stat_cost_output_microcents)",
+    ]),
+    total_cost_microcents: integer(row, "total_cost_microcents", [
+      "SUM(cost.total.microcents)",
+      "SUM(stat_cost_total_microcents)",
+    ]),
+    avg_duration_ms: nullableNumber(row, "avg_duration_ms", ["AVG(duration)", "AVG(duration_ms)"]),
+    p50_duration_ms: nullableInteger(row, "p50_duration_ms", ["P50(duration)", "P50(duration_ms)"]),
+    p95_duration_ms: nullableInteger(row, "p95_duration_ms", ["P95(duration)", "P95(duration_ms)"]),
+    avg_ttfb_ms: nullableNumber(row, "avg_ttfb_ms", ["AVG(time_to_first_byte)", "AVG(ttfb_ms)"]),
+    p50_ttfb_ms: nullableInteger(row, "p50_ttfb_ms", ["P50(time_to_first_byte)", "P50(ttfb_ms)"]),
+    p95_ttfb_ms: nullableInteger(row, "p95_ttfb_ms", ["P95(time_to_first_byte)", "P95(ttfb_ms)"]),
+    avg_output_tps: nullableNumber(row, "avg_output_tps", ["AVG(tps.output)", "AVG(stat_output_tps)"]),
+    success_count: integer(row, "success_count", ["SUM(success)", "SUM(is_success)", "SUM(stat_success)"]),
+    error_count: integer(row, "error_count", ["SUM(error)", "SUM(is_error)", "SUM(stat_error)"]),
+    sample_count: integer(row, "sample_count", ["COUNT", "COUNT()"]),
+  }
+}
+
+function basePeriod(row: RawRow, grain: Grain, opts: ImportOptions) {
+  return { period_key: periodKey(row, grain, opts) }
+}
+
+function periodKey(row: RawRow, grain: Grain, opts: ImportOptions) {
+  if (grain === "week") {
+    const week = parseWeek(row)
+    if (week) return week
+    fail("weekly imports require a week or period_key column")
+  }
+
+  const time = parseTime(row)
+  const start = time ? startOfUtcDay(time) : opts.periodStart
+  if (!start) fail("daily imports require a time column or --period-start")
+  return periodKeyFor("day", start)
+}
+
+function modelRowsFromAggregates(aggregates: ModelAggregate[]) {
+  return rankModelRows([
+    ...synthesizeAllTierRows(
+      collapseRows(aggregates.filter((item) => item.grain === "week").map(toModelRow), modelDimensionKey),
+      modelDimensionKey,
+    ),
+    ...synthesizeAllTierRows(
+      collapseRows(aggregates.filter((item) => item.grain === "day").map(toModelRow), modelDimensionKey),
+      modelDimensionKey,
+    ),
+  ])
+}
+
+function providerRowsFromAggregates(aggregates: ProviderAggregate[]) {
+  return rankRowsWithMarketShare([
+    ...synthesizeAllTierRows(
+      collapseRows(aggregates.filter((item) => item.grain === "week").map(toProviderRow), providerDimensionKey),
+      providerDimensionKey,
+    ),
+    ...synthesizeAllTierRows(
+      collapseRows(aggregates.filter((item) => item.grain === "day").map(toProviderRow), providerDimensionKey),
+      providerDimensionKey,
+    ),
+  ])
+}
+
+function geoRowsFromAggregates(aggregates: GeoAggregate[]) {
+  return rankRowsWithMarketShare(
+    [
+      ...synthesizeAllTierRows(
+        collapseRows(aggregates.filter((item) => item.grain === "week").map(toGeoRow), geoDimensionKey),
+        geoDimensionKey,
+      ),
+      ...synthesizeAllTierRows(
+        collapseRows(aggregates.filter((item) => item.grain === "day").map(toGeoRow), geoDimensionKey),
+        geoDimensionKey,
+      ),
+    ],
+    geoMarketShareKey,
+  )
+}
+
+function toModelRow(data: ModelAggregate): ModelStatRow {
+  return { ...toStatBaseRow(data), provider: data.provider, model: data.model, provider_model: data.provider_model }
+}
+
+function toProviderRow(data: ProviderAggregate): ProviderStatRow {
+  return { ...toStatBaseRow(data), provider: data.provider }
+}
+
+function toGeoRow(data: GeoAggregate): GeoStatRow {
+  return {
+    ...toStatBaseRow(data),
+    provider: data.provider,
+    model: data.model,
+    country: data.country,
+    continent: data.continent,
+  }
+}
+
+function rankModelRows(rows: ModelStatRow[]) {
+  return Object.values(
+    rows.reduce<Record<string, ModelStatRow[]>>((result, row) => {
+      const key = statPeriodKey(row)
+      result[key] = [...(result[key] ?? []), row]
+      return result
+    }, {}),
+  ).flatMap((group) => {
+    const tokenRanks = rankBy(group, (row) => row.total_tokens ?? 0)
+    const requestRanks = rankBy(group, (row) => row.requests ?? 0)
+    const costRanks = rankBy(group, (row) => row.total_cost_microcents ?? 0)
+    return group.map((row) => ({
+      ...row,
+      rank_by_tokens: tokenRanks.get(row) ?? null,
+      rank_by_requests: requestRanks.get(row) ?? null,
+      rank_by_cost: costRanks.get(row) ?? null,
+    }))
+  })
+}
+
+function modelDimensionKey(row: ModelStatRow) {
+  return [row.provider, row.model].join("\u0000")
+}
+
+function providerDimensionKey(row: ProviderStatRow) {
+  return row.provider
+}
+
+function geoDimensionKey(row: GeoStatRow) {
+  return [row.provider, row.model, row.country].join("\u0000")
+}
+
+function geoMarketShareKey(row: GeoStatRow) {
+  return [statPeriodKey(row), row.provider, row.model].join("\u0000")
+}
+
+function lookupKey(base: { grain: string; period_key: string; dataset: string; tier: string }, ...dimension: string[]) {
+  return [base.grain, base.period_key, base.dataset, base.tier, ...dimension].join("\u0000")
+}
+
+function tier(row: RawRow) {
+  return normalizeTier(cell(row, ["stat_tier", "tier"]) || deriveTier(row))
+}
+
+function deriveTier(row: RawRow) {
+  const source = cell(row, ["source"])
+  const value = model(row)
+  if (source === "lite") return "Go"
+  if (FREE_MODELS.has(value) || /-free(:global)?$/.test(rawModel(row))) return "Free"
+  return "Zen"
+}
+
+function provider(row: RawRow) {
+  return statProvider(model(row), providerModel(row), cell(row, ["stat_provider_2", "stat_provider"]))
+}
+
+function model(row: RawRow) {
+  return statModel(cell(row, ["stat_model_2", "stat_model"]) || rawModel(row), providerModel(row))
+}
+
+function rawModel(row: RawRow) {
+  return cell(row, ["model"]) || "unknown"
+}
+
+function providerModel(row: RawRow) {
+  return cell(row, ["provider.model", "provider_model"]) || ""
+}
+
+function country(row: RawRow) {
+  return normalizeCountry(cell(row, ["stat_country", "cf.country", "cf_country", "country"]))
+}
+
+function continent(row: RawRow) {
+  return cell(row, ["cf.continent", "cf_continent", "continent"]) || ""
+}
+
+function integer(row: RawRow, name: string, aliases: string[] = []) {
+  return Math.round(number(row, name, aliases))
+}
+
+function nullableInteger(row: RawRow, name: string, aliases: string[] = []) {
+  if (!hasCell(row, [name, ...aliases])) return null
+  return Math.round(number(row, name, aliases))
+}
+
+function nullableNumber(row: RawRow, name: string, aliases: string[] = []) {
+  if (!hasCell(row, [name, ...aliases])) return null
+  return Number(number(row, name, aliases).toFixed(2))
+}
+
+function number(row: RawRow, name: string, aliases: string[] = []) {
+  const value = Number(cell(row, [name, ...aliases]).replace(/,/g, ""))
+  return Number.isFinite(value) ? value : 0
+}
+
+function hasCell(row: RawRow, names: string[]) {
+  return names.some((name) => row[name] !== undefined && row[name] !== "")
+}
+
+function cell(row: RawRow, names: string[]) {
+  const normalized = normalizedCells(row)
+  return (
+    names.flatMap((name) => [row[name], normalized.get(normalizeHeader(name))]).find((value) => value !== undefined) ??
+    ""
+  )
+}
+
+function normalizedCells(row: RawRow) {
+  return new Map(Object.entries(row).map(([key, value]) => [normalizeHeader(key), value]))
+}
+
+function normalizeHeader(value: string) {
+  return value.toLowerCase().replace(/[^a-z0-9]+/g, "")
+}
+
+function parseTime(row: RawRow) {
+  const value = cell(row, ["date", "time", "timestamp", "datetime", "bucket"])
+  if (!value) return undefined
+  const numeric = Number(value)
+  const date = Number.isFinite(numeric)
+    ? new Date(numeric > 10_000_000_000 ? numeric : numeric * 1000)
+    : new Date(value)
+  if (Number.isNaN(date.getTime())) fail(`Invalid time value: ${value}`)
+  return date
+}
+
+function parseWeek(row: RawRow) {
+  const value = cell(row, ["period_key", "week", "stat_week"])
+  if (!value) return undefined
+
+  const match = /^(\d{4})-W(\d{1,2})$/.exec(value)
+  if (!match) fail(`Invalid week value: ${value}`)
+
+  const year = Number(match[1])
+  const week = Number(match[2])
+  if (week < 1 || week > 53) fail(`Invalid week value: ${value}`)
+
+  const start = new Date(startOfIsoWeek(new Date(Date.UTC(year, 0, 4))).getTime() + (week - 1) * 7 * DAY_MS)
+  const id = `${year}-W${String(week).padStart(2, "0")}`
+  if (isoWeekId(start) !== id) fail(`Invalid week value: ${value}`)
+  return id
+}
+
+async function readRows(file: string) {
+  const text = await Bun.file(file).text()
+  if (file.toLowerCase().endsWith(".json")) {
+    const parsed: unknown = JSON.parse(text)
+    return rowsFromJson(parsed)
+  }
+  return rowsFromCsv(text)
+}
+
+function rowsFromJson(value: unknown): RawRow[] {
+  if (Array.isArray(value)) return value.flatMap(rowFromUnknown)
+  if (!isRecord(value)) fail("JSON imports must be an array of rows or an object with results/data/rows")
+
+  const rows = [value.results, value.data, value.rows].flatMap((candidate) =>
+    Array.isArray(candidate) ? candidate.flatMap(rowFromUnknown) : [],
+  )
+  if (rows.length === 0) fail("JSON import did not contain rows")
+  return rows
+}
+
+function rowFromUnknown(value: unknown): RawRow[] {
+  if (!isRecord(value)) return []
+  const nested = isRecord(value.data) ? value.data : {}
+  return [
+    Object.fromEntries(
+      Object.entries({ ...value, ...nested }).flatMap(([key, item]) => {
+        if (key === "data") return []
+        return [[key, cellValue(item)]]
+      }),
+    ),
+  ]
+}
+
+function rowsFromCsv(text: string): RawRow[] {
+  const [headers, ...rows] = csvRecords(text).filter((row) => row.some((value) => value.trim() !== ""))
+  if (!headers) return []
+  return rows.map((row) =>
+    Object.fromEntries(headers.map((header, index) => [header.trim(), row[index]?.trim() ?? ""])),
+  )
+}
+
+function csvRecords(text: string) {
+  const rows: string[][] = []
+  let row: string[] = []
+  let field = ""
+  let quoted = false
+
+  for (let index = 0; index < text.length; index++) {
+    const char = text[index]
+    const next = text[index + 1]
+    if (quoted) {
+      if (char === '"' && next === '"') {
+        field += '"'
+        index++
+        continue
+      }
+      if (char === '"') {
+        quoted = false
+        continue
+      }
+      field += char
+      continue
+    }
+    if (char === '"') {
+      quoted = true
+      continue
+    }
+    if (char === ",") {
+      row.push(field)
+      field = ""
+      continue
+    }
+    if (char === "\n") {
+      row.push(field)
+      rows.push(row)
+      row = []
+      field = ""
+      continue
+    }
+    if (char === "\r") continue
+    field += char
+  }
+
+  row.push(field)
+  rows.push(row)
+  return rows
+}
+
+function cellValue(value: unknown) {
+  if (value === null || value === undefined) return ""
+  if (typeof value === "string") return value
+  if (typeof value === "number" || typeof value === "boolean" || typeof value === "bigint") return String(value)
+  return JSON.stringify(value) ?? ""
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value)
+}
+
+async function upsertModelRows(db: ReturnType<typeof drizzle>, rows: ModelStatRow[], chunkSize: number) {
+  const batches = chunks(rows, chunkSize)
+  console.log(JSON.stringify({ table: "model_stat", batches: batches.length, chunkSize }))
+  for (const chunk of batches) {
+    await db
+      .insert(modelStat)
+      .values(chunk)
+      .onDuplicateKeyUpdate({
+        set: {
+          provider_model: inserted("provider_model"),
+          sessions: inserted("sessions"),
+          requests: inserted("requests"),
+          input_tokens: inserted("input_tokens"),
+          output_tokens: inserted("output_tokens"),
+          reasoning_tokens: inserted("reasoning_tokens"),
+          cache_read_tokens: inserted("cache_read_tokens"),
+          total_tokens: inserted("total_tokens"),
+          input_cost_microcents: inserted("input_cost_microcents"),
+          output_cost_microcents: inserted("output_cost_microcents"),
+          total_cost_microcents: inserted("total_cost_microcents"),
+          avg_duration_ms: inserted("avg_duration_ms"),
+          p50_duration_ms: inserted("p50_duration_ms"),
+          p95_duration_ms: inserted("p95_duration_ms"),
+          avg_ttfb_ms: inserted("avg_ttfb_ms"),
+          p50_ttfb_ms: inserted("p50_ttfb_ms"),
+          p95_ttfb_ms: inserted("p95_ttfb_ms"),
+          avg_output_tps: inserted("avg_output_tps"),
+          success_count: inserted("success_count"),
+          error_count: inserted("error_count"),
+          sample_count: inserted("sample_count"),
+          rank_by_tokens: inserted("rank_by_tokens"),
+          rank_by_requests: inserted("rank_by_requests"),
+          rank_by_cost: inserted("rank_by_cost"),
+        },
+      })
+  }
+}
+
+async function upsertProviderRows(db: ReturnType<typeof drizzle>, rows: ProviderStatRow[], chunkSize: number) {
+  const batches = chunks(rows, chunkSize)
+  console.log(JSON.stringify({ table: "provider_stat", batches: batches.length, chunkSize }))
+  for (const chunk of batches) {
+    await db
+      .insert(providerStat)
+      .values(chunk)
+      .onDuplicateKeyUpdate({
+        set: {
+          sessions: inserted("sessions"),
+          requests: inserted("requests"),
+          input_tokens: inserted("input_tokens"),
+          output_tokens: inserted("output_tokens"),
+          reasoning_tokens: inserted("reasoning_tokens"),
+          cache_read_tokens: inserted("cache_read_tokens"),
+          total_tokens: inserted("total_tokens"),
+          input_cost_microcents: inserted("input_cost_microcents"),
+          output_cost_microcents: inserted("output_cost_microcents"),
+          total_cost_microcents: inserted("total_cost_microcents"),
+          avg_duration_ms: inserted("avg_duration_ms"),
+          p50_duration_ms: inserted("p50_duration_ms"),
+          p95_duration_ms: inserted("p95_duration_ms"),
+          avg_ttfb_ms: inserted("avg_ttfb_ms"),
+          p50_ttfb_ms: inserted("p50_ttfb_ms"),
+          p95_ttfb_ms: inserted("p95_ttfb_ms"),
+          avg_output_tps: inserted("avg_output_tps"),
+          success_count: inserted("success_count"),
+          error_count: inserted("error_count"),
+          sample_count: inserted("sample_count"),
+          market_share_tokens: inserted("market_share_tokens"),
+          market_share_requests: inserted("market_share_requests"),
+          market_share_sessions: inserted("market_share_sessions"),
+          rank_by_tokens: inserted("rank_by_tokens"),
+          rank_by_requests: inserted("rank_by_requests"),
+          rank_by_sessions: inserted("rank_by_sessions"),
+          rank_by_cost: inserted("rank_by_cost"),
+        },
+      })
+  }
+}
+
+async function upsertGeoRows(db: ReturnType<typeof drizzle>, rows: GeoStatRow[], chunkSize: number) {
+  const batches = chunks(rows, chunkSize)
+  console.log(JSON.stringify({ table: "geo_stat", batches: batches.length, chunkSize }))
+  for (const chunk of batches) {
+    await db
+      .insert(geoStat)
+      .values(chunk)
+      .onDuplicateKeyUpdate({
+        set: {
+          continent: inserted("continent"),
+          sessions: inserted("sessions"),
+          requests: inserted("requests"),
+          input_tokens: inserted("input_tokens"),
+          output_tokens: inserted("output_tokens"),
+          reasoning_tokens: inserted("reasoning_tokens"),
+          cache_read_tokens: inserted("cache_read_tokens"),
+          total_tokens: inserted("total_tokens"),
+          input_cost_microcents: inserted("input_cost_microcents"),
+          output_cost_microcents: inserted("output_cost_microcents"),
+          total_cost_microcents: inserted("total_cost_microcents"),
+          avg_duration_ms: inserted("avg_duration_ms"),
+          p50_duration_ms: inserted("p50_duration_ms"),
+          p95_duration_ms: inserted("p95_duration_ms"),
+          avg_ttfb_ms: inserted("avg_ttfb_ms"),
+          p50_ttfb_ms: inserted("p50_ttfb_ms"),
+          p95_ttfb_ms: inserted("p95_ttfb_ms"),
+          avg_output_tps: inserted("avg_output_tps"),
+          success_count: inserted("success_count"),
+          error_count: inserted("error_count"),
+          sample_count: inserted("sample_count"),
+          market_share_tokens: inserted("market_share_tokens"),
+          market_share_requests: inserted("market_share_requests"),
+          market_share_sessions: inserted("market_share_sessions"),
+          rank_by_tokens: inserted("rank_by_tokens"),
+          rank_by_requests: inserted("rank_by_requests"),
+          rank_by_sessions: inserted("rank_by_sessions"),
+          rank_by_cost: inserted("rank_by_cost"),
+        },
+      })
+  }
+}
+
+function parseImportOptions(args: string[]): ImportOptions {
+  const flags = parseFlags(args)
+  const files = inputKeys.reduce<Partial<Record<ImportKey, string[]>>>((result, key) => {
+    const values = flags.get(key)
+    if (!values) return result
+    return { ...result, [key]: values }
+  }, {})
+  return {
+    dataset: flags.get("dataset")?.[0] ?? "zen",
+    databaseUrl: flags.get("database-url")?.[0] ?? process.env.DATABASE_URL,
+    directories: flags.get("dir") ?? flags.get("directory") ?? [],
+    dryRun: flags.has("dry-run"),
+    periodStart: parseDateFlag(flags, "period-start"),
+    upsertChunkSize: parseIntegerFlag(flags, "upsert-chunk-size") ?? DEFAULT_UPSERT_CHUNK_SIZE,
+    files,
+  }
+}
+
+function parseFlags(args: string[]) {
+  const result = new Map<string, string[]>()
+  for (let index = 0; index < args.length; index++) {
+    const arg = args[index]
+    if (!arg.startsWith("--")) fail(`Unexpected argument: ${arg}`)
+    const name = arg.slice(2)
+    if (name === "dry-run" || name === "include-weekly") {
+      result.set(name, ["true"])
+      continue
+    }
+    const nextFlag = args.findIndex((value, valueIndex) => valueIndex > index && value.startsWith("--"))
+    const values = args.slice(index + 1, nextFlag === -1 ? args.length : nextFlag)
+    if (values.length === 0) fail(`Missing value for --${name}`)
+    result.set(name, [...(result.get(name) ?? []), ...values])
+    index += values.length
+  }
+  return result
+}
+
+function parseDateFlag(flags: Map<string, string[]>, name: string) {
+  const value = flags.get(name)?.[0]
+  if (!value) return undefined
+  const date = new Date(value)
+  if (Number.isNaN(date.getTime())) fail(`Invalid --${name}: ${value}`)
+  return date
+}
+
+function parseIntegerFlag(flags: Map<string, string[]>, name: string) {
+  const value = flags.get(name)?.[0]
+  if (!value) return undefined
+  const parsed = Number(value)
+  if (!Number.isInteger(parsed) || parsed <= 0) fail(`Invalid --${name}: ${value}`)
+  return parsed
+}
+
+function parseListFlag(flags: Map<string, string[]>, name: string) {
+  const value = flags.get(name)?.[0]
+  if (!value) return undefined
+  if (value === "all") return ["all"]
+  return value
+    .split(",")
+    .map((item) => item.trim())
+    .filter(Boolean)
+}
+
+function usage(): never {
+  fail(`Usage:
+  bun src/honeycomb-backfill.ts queries [--tiers Go,Free,Paid] [--limit 1000]
+  bun src/honeycomb-backfill.ts import [--dry-run] [--upsert-chunk-size 100] [--database-url URL] --dir downloads
+  bun src/honeycomb-backfill.ts import [--dry-run] [--upsert-chunk-size 100] [--database-url URL] --model-day file.csv [--model-day more.csv] ...`)
+}
+
+function fail(message: string): never {
+  console.error(message)
+  process.exit(1)
+}

packages/stats/core/src/index.ts

@@ -0,0 +1,11 @@
+export * as Athena from "./athena"
+export * as AppConfig from "./config"
+export * as Database from "./database"
+export * as GeoStat from "./domain/geo"
+export * as StatsHome from "./domain/home"
+export * as Inference from "./domain/inference"
+export * as ModelStat from "./domain/model"
+export * as ProviderStat from "./domain/provider"
+export * as Stat from "./domain/stat"
+export * as Runtime from "./runtime"
+export * as StatSync from "./stat-sync"

packages/stats/core/src/migrate.ts

@@ -0,0 +1,4 @@
+import { Effect } from "effect"
+import { layer, migrate } from "./database"
+
+await Effect.runPromise(migrate().pipe(Effect.provide(layer)))

packages/stats/core/src/resource.d.ts

@@ -0,0 +1,28 @@
+import "sst/resource"
+
+declare module "sst/resource" {
+  export interface Resource {
+    InferenceEvent: {
+      catalog: string
+      database: string
+      region: string
+      table: string
+      tableBucket: string
+      type: "sst.sst.Linkable"
+      workgroup: string
+    }
+    StatsSyncConfig: {
+      dataset: string
+      type: "sst.sst.Linkable"
+    }
+    StatsDatabase: {
+      database: string
+      host: string
+      password: string
+      port: number
+      type: "sst.sst.Linkable"
+      url: string
+      username: string
+    }
+  }
+}

packages/stats/core/src/runtime.ts

@@ -0,0 +1,14 @@
+import { Layer, ManagedRuntime } from "effect"
+import { AppConfig } from "./config"
+import { layer as databaseLayer } from "./database"
+import { GeoStatRepo } from "./domain/geo"
+import { ModelStatRepo } from "./domain/model"
+import { ProviderStatRepo } from "./domain/provider"
+
+const repoLayer = Layer.mergeAll(ModelStatRepo.layer, ProviderStatRepo.layer, GeoStatRepo.layer).pipe(
+  Layer.provide(databaseLayer),
+)
+
+export const layer = Layer.mergeAll(AppConfig.layer, databaseLayer, repoLayer)
+export const runtime = ManagedRuntime.make(layer)
+export type RuntimeServices = ManagedRuntime.ManagedRuntime.Services<typeof runtime>

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

@@ -0,0 +1,101 @@
+import { DateTime, Effect } from "effect"
+import { Resource } from "sst/resource"
+import { Athena, AthenaQueryError, AthenaQueryTimeoutError } from "./athena"
+import { DatabaseError } from "./database"
+import { GeoStatRepo, rowsFromAggregates as geoRowsFromAggregates } from "./domain/geo"
+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 STATS_DATA_START_MS = new Date("2026-05-28T00:00:00.000Z").getTime()
+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
+
+export const syncStats: () => Effect.Effect<
+  SyncStatsResult,
+  SyncStatsError,
+  Athena | ModelStatRepo | ProviderStatRepo | GeoStatRepo
+> = 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)
+  // May 27 was partial, so keep Athena stats anchored at the first complete day.
+  const periodStart = new Date(Math.max(startOfIsoWeek(periodEnd).getTime() - WEEK_MS, STATS_DATA_START_MS))
+  const athena = yield* Athena
+  const modelStats = yield* ModelStatRepo
+  const providerStats = yield* ProviderStatRepo
+  const geoStats = yield* GeoStatRepo
+
+  yield* logRuntimeCheck()
+
+  const [modelAggregates, providerAggregates, geoAggregates, geoModelAggregates] = yield* Effect.all(
+    [
+      athena
+        .query(buildStatsQuery(periodStart, periodEnd, "model"))
+        .pipe(Effect.map((rows) => rows.flatMap(toModelAggregate))),
+      athena
+        .query(buildStatsQuery(periodStart, periodEnd, "provider"))
+        .pipe(Effect.map((rows) => rows.flatMap(toProviderAggregate))),
+      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, ...geoModelAggregates])
+
+  yield* Effect.all([modelStats.upsert(modelRows), providerStats.upsert(providerRows), geoStats.upsert(geoRows)], {
+    concurrency: "unbounded",
+    discard: true,
+  })
+  yield* Effect.all(
+    [
+      modelStats.deleteRetiredDimensions(modelRows),
+      providerStats.deleteRetiredDimensions(providerRows),
+      geoStats.deleteRetiredDimensions(geoRows),
+    ],
+    { concurrency: "unbounded", discard: true },
+  )
+
+  yield* Effect.logInfo(
+    `stats sync complete ${JSON.stringify({
+      startedAt: startedAt.toISOString(),
+      periodStart: periodStart.toISOString(),
+      periodEnd: periodEnd.toISOString(),
+      rows: modelRows.length,
+      providerRows: providerRows.length,
+      geoRows: geoRows.length,
+      stage: Resource.App.stage,
+    })}`,
+  )
+
+  return {
+    ok: true,
+    rows: modelRows.length,
+    startedAt: startedAt.toISOString(),
+    periodStart: periodStart.toISOString(),
+    periodEnd: periodEnd.toISOString(),
+  }
+})
+
+function logRuntimeCheck() {
+  return Effect.logInfo(
+    `athena stats runtime check ${JSON.stringify({
+      catalog: Resource.InferenceEvent.catalog,
+      database: Resource.InferenceEvent.database,
+      dataset: Resource.StatsSyncConfig.dataset,
+      table: Resource.InferenceEvent.table,
+      workgroup: Resource.InferenceEvent.workgroup,
+      region: Resource.InferenceEvent.region,
+      stage: Resource.App.stage,
+    })}`,
+  )
+}