diff --git a/packages/stats/app/src/routes/[lab]/[model].tsx b/packages/stats/app/src/routes/[lab]/[model].tsx index e0560957ac..c9a3e6ef6d 100644 --- a/packages/stats/app/src/routes/[lab]/[model].tsx +++ b/packages/stats/app/src/routes/[lab]/[model].tsx @@ -470,6 +470,7 @@ function ModelMomentumSection(props: { data: StatsModelPageData | null }) { value={formatInteger(data().totals.sessions)} /> + span { + color: var(--stats-faint); + font-size: 13px; + font-weight: 400; +} + +[data-page="stats"] [data-component="retention-chart"] a > strong { + min-width: 0; + overflow-wrap: anywhere; + font-size: 13px; + font-weight: 600; + line-height: 18px; +} + +[data-page="stats"] [data-component="retention-chart"] a > b, +[data-page="stats"] [data-component="retention-chart"] a > em { + font-size: 13px; + font-style: normal; + font-weight: 500; + text-align: right; + white-space: nowrap; +} + +[data-page="stats"] [data-component="retention-chart"] a > em { + color: var(--stats-muted); +} + +[data-page="stats"] [data-component="retention-marker"] { + position: relative; + display: grid; + grid-template-columns: repeat(4, 1fr); + align-items: center; + height: 16px; + background: linear-gradient(var(--stats-line-strong), var(--stats-line-strong)) center / 100% 1px no-repeat; +} + +[data-page="stats"] [data-component="retention-marker"] > span { + justify-self: end; + width: 1px; + height: 8px; + background: var(--stats-line-strong); +} + +[data-page="stats"] [data-component="retention-marker"] > em { + position: absolute; + top: 50%; + left: var(--retention-position); + width: 7px; + height: 16px; + background: var(--stats-muted); + transform: translate(-50%, -50%); +} + +[data-page="stats"] [data-component="retention-marker"][data-active="true"] > em { + background: var(--stats-accent); +} + [data-page="stats"] [data-component="section-bridge"]:hover { color: var(--stats-text); text-decoration: none; @@ -3521,7 +3641,7 @@ body { [data-page="stats"] [data-slot="model-momentum-metrics"] { display: grid; - grid-template-columns: repeat(4, minmax(0, 1fr)); + grid-template-columns: repeat(5, minmax(0, 1fr)); gap: 12px; min-width: 0; } @@ -6671,6 +6791,57 @@ body { } } +@media (max-width: 74rem) { + [data-page="stats"] [data-slot="retention-heading"], + [data-page="stats"] [data-component="retention-chart"] a { + grid-template-columns: 40px minmax(140px, 220px) minmax(140px, 1fr) 60px 76px; + gap: 12px; + } +} + +@media (max-width: 47.999rem) { + [data-page="stats"] [data-component="retention-chart"] { + margin-top: 28px; + } + + [data-page="stats"] [data-slot="retention-heading"] { + display: none; + } + + [data-page="stats"] [data-component="retention-chart"] a { + grid-template-columns: 28px minmax(0, 1fr) 58px 62px; + grid-template-rows: auto 16px; + gap: 8px 10px; + min-height: 68px; + padding: 10px; + } + + [data-page="stats"] [data-component="retention-chart"] a > span { + grid-column: 1; + grid-row: 1; + } + + [data-page="stats"] [data-component="retention-chart"] a > strong { + grid-column: 2; + grid-row: 1; + } + + [data-page="stats"] [data-component="retention-marker"] { + grid-column: 2 / -1; + grid-row: 2; + } + + [data-page="stats"] [data-component="retention-chart"] a > b { + grid-column: 3; + grid-row: 1; + } + + [data-page="stats"] [data-component="retention-chart"] a > em { + grid-column: 4; + grid-row: 1; + } +} + [data-page="stats"] [data-component="compare-model-modal-scrim"] { position: fixed; inset: 0; @@ -7368,6 +7539,7 @@ body { [data-page="stats"] [data-section="top-models"], [data-page="stats"] [data-section="leaderboard"], [data-page="stats"] [data-section="unique-users"], + [data-page="stats"] [data-section="retention"], [data-page="stats"] [data-section="market-share"], [data-page="stats"] [data-section="geo-breakdown"], [data-page="stats"] [data-section="token-cost"], @@ -7529,6 +7701,10 @@ body { grid-template-columns: repeat(2, minmax(0, 1fr)); } + [data-page="stats"] [data-component="model-momentum-metric"]:last-child:nth-child(odd) { + grid-column: 1 / -1; + } + [data-page="stats"] [data-component="model-metric-grid"], [data-page="stats"] [data-component="model-metric-grid"][data-variant="dense"], [data-page="stats"] [data-component="model-efficiency-grid"], @@ -7668,6 +7844,7 @@ body { [data-page="stats"] [data-section="top-models"], [data-page="stats"] [data-section="leaderboard"], [data-page="stats"] [data-section="unique-users"], + [data-page="stats"] [data-section="retention"], [data-page="stats"] [data-section="market-share"], [data-page="stats"] [data-section="geo-breakdown"], [data-page="stats"] [data-section="token-cost"], diff --git a/packages/stats/app/src/routes/index.tsx b/packages/stats/app/src/routes/index.tsx index 30491c8983..0e9888d2a0 100644 --- a/packages/stats/app/src/routes/index.tsx +++ b/packages/stats/app/src/routes/index.tsx @@ -8,6 +8,7 @@ import { type CountryEntry, type LeaderboardEntry, type MarketDay, + type RetentionEntry, type SessionCostEntry, type TokenCostEntry, type UsagePoint, @@ -69,6 +70,7 @@ type StatsHomePageData = { tokenCost: TokenCostEntry[] cacheRatio: CacheRatioEntry[] sessionCost: SessionCostEntry[] + retention: RetentionEntry[] country: CountryEntry[] } @@ -88,6 +90,7 @@ const getData = query(async () => { tokenCost: priceTokenCostFromCatalog(stats.tokenCost.Go, catalog), cacheRatio: stats.cacheRatio.Go, sessionCost: stats.sessionCost.Go, + retention: stats.retention, country: stats.country["2M"], } satisfies StatsHomePageData }, "getStatsHomeData") @@ -146,6 +149,7 @@ export default function StatsHome() { + @@ -616,6 +620,86 @@ function UniqueUsersSection(props: { data: UsagePoint[] }) { ) } +function RetentionSection(props: { data: RetentionEntry[] }) { + const language = useLanguage() + const [activeIndex, setActiveIndex] = createSignal(0) + + return ( +
+ + 0} + fallback={ + + } + > + + +
+ ) +} + +function RetentionMarker(props: { rate: number; active: boolean }) { + const fill = createMemo(() => Math.min(100, Math.max(0, props.rate))) + return ( + + ) +} + +function formatRetentionRate(value: number) { + return `${value.toFixed(1)}%` +} + function isTopModelsBlankHover(bar: HTMLElement, clientY: number) { const stack = bar.querySelector('[data-slot="top-models-stack"]') if (!stack) return true diff --git a/packages/stats/core/migrations/20260826000000_model_retention/migration.sql b/packages/stats/core/migrations/20260826000000_model_retention/migration.sql new file mode 100644 index 0000000000..e69d7f5bf4 --- /dev/null +++ b/packages/stats/core/migrations/20260826000000_model_retention/migration.sql @@ -0,0 +1,18 @@ +CREATE TABLE `model_retention` ( + `id` bigint AUTO_INCREMENT NOT NULL, + `cohort_date` char(10) NOT NULL, + `dataset` varchar(64) NOT NULL DEFAULT 'all', + `tier` varchar(64) NOT NULL DEFAULT 'all', + `provider` varchar(128) NOT NULL, + `model` varchar(256) NOT NULL, + `eligible_users` bigint NOT NULL DEFAULT 0, + `retained_users` bigint NOT NULL DEFAULT 0, + `created_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + CONSTRAINT `model_retention_id` PRIMARY KEY(`id`), + CONSTRAINT `uniq_model_retention_cohort` UNIQUE(`cohort_date`,`dataset`,`tier`,`provider`,`model`) +); +--> statement-breakpoint +CREATE INDEX `idx_model_retention_recent` ON `model_retention` (`dataset`,`tier`,`cohort_date`); +--> statement-breakpoint +CREATE INDEX `idx_model_retention_model` ON `model_retention` (`model`,`cohort_date`); diff --git a/packages/stats/core/src/database/schema.ts b/packages/stats/core/src/database/schema.ts index d5bfa314bf..dcf8d52332 100644 --- a/packages/stats/core/src/database/schema.ts +++ b/packages/stats/core/src/database/schema.ts @@ -107,6 +107,32 @@ export const geoStat = mysqlTable( ], ) +export const modelRetention = mysqlTable( + "model_retention", + { + id: bigint({ mode: "number" }).autoincrement().primaryKey(), + cohort_date: char({ length: 10 }).notNull(), + dataset: varchar({ length: 64 }).notNull().default("all"), + tier: varchar({ length: 64 }).notNull().default("all"), + provider: varchar({ length: 128 }).notNull(), + model: varchar({ length: 256 }).notNull(), + eligible_users: bigint({ mode: "number" }).notNull().default(0), + retained_users: bigint({ mode: "number" }).notNull().default(0), + ...timestampColumns(), + }, + (table) => [ + uniqueIndex("uniq_model_retention_cohort").on( + table.cohort_date, + table.dataset, + table.tier, + table.provider, + table.model, + ), + index("idx_model_retention_recent").on(table.dataset, table.tier, table.cohort_date), + index("idx_model_retention_model").on(table.model, table.cohort_date), + ], +) + function periodColumns() { return { id: bigint({ mode: "number" }).autoincrement().primaryKey(), diff --git a/packages/stats/core/src/domain/home.test.ts b/packages/stats/core/src/domain/home.test.ts new file mode 100644 index 0000000000..c8b665048c --- /dev/null +++ b/packages/stats/core/src/domain/home.test.ts @@ -0,0 +1,48 @@ +import { describe, expect, test } from "bun:test" +import type { RetentionMetricRow } from "./home" + +process.env.SST_RESOURCE_App = JSON.stringify({ name: "opencode", stage: "test" }) +process.env.SST_RESOURCE_StatsDatabase = JSON.stringify({ url: "mysql://localhost/stats" }) + +const { buildRetentionEntries } = await import("./home") + +describe("retention aggregates", () => { + test("pools the latest seven cohorts and ranks models above the sample floor", () => { + const rows = [ + ...cohorts("model-a", "provider-a", 8, 20, 10), + ...cohorts("model-b", "provider-b", 8, 20, 12), + ...cohorts("small-model", "provider-c", 8, 10, 9), + ] + const entries = buildRetentionEntries(rows) + + expect(entries.find((item) => item.model === "model-a")).toMatchObject({ + eligibleUserDays: 140, + retainedUserDays: 70, + rate: 50, + rank: 2, + }) + expect(entries.find((item) => item.model === "model-b")).toMatchObject({ + eligibleUserDays: 140, + retainedUserDays: 84, + rate: 60, + rank: 1, + }) + expect(entries.find((item) => item.model === "small-model")).toMatchObject({ + eligibleUserDays: 70, + retainedUserDays: 63, + rate: 90, + rank: null, + }) + }) +}) + +function cohorts(model: string, provider: string, count: number, eligibleUsers: number, retainedUsers: number) { + return Array.from({ length: count }, (_, index) => ({ + cohortDate: `2026-08-${String(index + 1).padStart(2, "0")}`, + updatedAt: Date.UTC(2026, 7, index + 9), + provider, + model, + eligibleUsers, + retainedUsers, + })) satisfies RetentionMetricRow[] +} diff --git a/packages/stats/core/src/domain/home.ts b/packages/stats/core/src/domain/home.ts index e0fd2be37b..c1784ed18a 100644 --- a/packages/stats/core/src/domain/home.ts +++ b/packages/stats/core/src/domain/home.ts @@ -5,6 +5,7 @@ import { DatabaseError } from "../database" import type { GeoStatMetric } from "./geo" import { ModelStatRepo, type ModelStatMetric } from "./model" import { statProvider } from "./model-normalization" +import { isMissingRetentionTable } from "./retention" import { DATA_SITE_TIERS, normalizeTier } from "./stat" export type UsageProduct = "All Users" | "Zen" | "Go" | "Enterprise" @@ -23,6 +24,15 @@ export type LeaderboardEntry = { 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 RetentionEntry = { + model: string + provider: string + author: string + rate: number + eligibleUserDays: number + retainedUserDays: number + rank: number | null +} export type CountryEntry = { country: string; continent: string; tokens: number; share: number; rank: number } export type ModelUsagePoint = { date: string; tokens: number; users: number; sessions: number; cost: number } export type ModelMixEntry = { label: string; tokens: number; share: number } @@ -54,6 +64,7 @@ export type StatsModelData = { totalModels: number tokenShare: number tokenChange: number + retention7d: RetentionEntry | null totals: { sessions: number uniqueUsers: number @@ -114,6 +125,7 @@ export type StatsHomeData = { tokenCost: Record cacheRatio: Record sessionCost: Record + retention: RetentionEntry[] country: Record } @@ -129,6 +141,9 @@ 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 RETENTION_MODEL_LIMIT = 15 +const RETENTION_MIN_ELIGIBLE_USER_DAYS = 100 +const RETENTION_COHORT_DAYS = 7 const TOP_MODEL_SEGMENT_LIMIT = 9 // Preserve the response shape while the public site presents Go and Free as one cohort. const SITE_PRODUCT = "Go" @@ -144,6 +159,14 @@ type GeoMetricRow = Omit & { periodStart: number updatedAt: number } +export type RetentionMetricRow = { + cohortDate: string + updatedAt: number + provider: string + model: string + eligibleUsers: number + retainedUsers: number +} type DateWindow = { start: number; end: number; previousStart: number; previousEnd: number } type Bucket = { start: number; end: number; label: string } @@ -167,8 +190,12 @@ type RawRow = Record export function getStatsHomeData(): Effect.Effect { return Effect.tryPromise({ try: async () => { - const [modelRows, geoRows] = await Promise.all([listModelDaily(), listGeoDaily()]) - return buildStatsHomeData(modelRows, geoRows) + const [modelRows, geoRows, retentionRows] = await Promise.all([ + listModelDaily(), + listGeoDaily(), + listRetentionDaily(), + ]) + return buildStatsHomeData(modelRows, geoRows, retentionRows) }, catch: (cause) => new StatsDataError(cause), }) @@ -180,7 +207,7 @@ export function getStatsModelData( ): Effect.Effect { return Effect.tryPromise({ try: async () => { - const modelRows = await listModelDaily() + const [modelRows, retentionRows] = await Promise.all([listModelDaily(), listRetentionDaily()]) const normalized = modelRows.flatMap(normalizeStatRow) const resolvedModel = resolveModelName(model, normalized, provider) if (!resolvedModel) return null @@ -192,6 +219,7 @@ export function getStatsModelData( provider: resolveModelProvider(resolvedModel, normalized, provider), }), provider, + retentionRows, ) }, catch: (cause) => new StatsDataError(cause), @@ -260,6 +288,27 @@ async function listGeoDaily(opts?: { provider?: string; model?: string }): Promi })) } +async function listRetentionDaily(): Promise { + try { + return ( + await queryRows( + `select cohort_date, updated_at, provider, model, eligible_users, retained_users + from model_retention where dataset = 'zen' and tier = 'all' order by cohort_date`, + ) + ).map((row) => ({ + cohortDate: stringValue(row.cohort_date), + updatedAt: dateValue(row.updated_at).getTime(), + provider: stringValue(row.provider), + model: stringValue(row.model), + eligibleUsers: numberValue(row.eligible_users), + retainedUsers: numberValue(row.retained_users), + })) + } catch (cause) { + if (isMissingRetentionTable(cause)) return [] + throw cause + } +} + async function queryRows(query: string, params: string[] = []) { return (await new Client({ url: databaseUrl() }).execute(query, params)).rows as RawRow[] } @@ -309,7 +358,11 @@ export const getStatsModelComparisonData = ( { provider: secondProvider, model: secondModel }, ]) -function buildStatsHomeData(modelRows: ModelStatMetric[], geoRows: GeoStatMetric[]): StatsHomeData { +function buildStatsHomeData( + modelRows: ModelStatMetric[], + geoRows: GeoStatMetric[], + retentionRows: RetentionMetricRow[], +): StatsHomeData { const normalized = modelRows.flatMap(normalizeStatRow) const geo = geoRows.flatMap(normalizeGeoRow) const periods = [...normalized, ...geo] @@ -357,6 +410,9 @@ function buildStatsHomeData(modelRows: ModelStatMetric[], geoRows: GeoStatMetric sessionCost: createTokenProductRecord((product) => buildSessionCost(normalized, product, getWindow("1W", earliest, latest)), ), + retention: buildRetentionEntries(retentionRows) + .filter((item) => item.rank !== null) + .slice(0, RETENTION_MODEL_LIMIT), country: createRangeRecord((range) => buildCountryStats(geo, getWindow(range, earliest, latest))), } } @@ -366,6 +422,7 @@ function buildStatsModelData( modelRows: ModelStatMetric[], geoRows: GeoStatMetric[], providerParam?: string, + retentionRows: RetentionMetricRow[] = [], ): StatsModelData | null { const normalized = modelRows.flatMap(normalizeStatRow) const geo = geoRows.flatMap(normalizeGeoRow) @@ -401,6 +458,7 @@ function buildStatsModelData( const peerRank = rankIndex >= 0 ? rankIndex + 1 : 1 const totalTokens = windowPeers.reduce((sum, item) => sum + item.totalTokens, 0) const peerTokens = rankPeers.reduce((sum, item) => sum + item.totalTokens, 0) + const retention7d = buildRetentionEntries(retentionRows).find((item) => item.model === model) ?? null return { updatedAt: Number.isFinite(latestUpdate) ? new Date(latestUpdate).toISOString() : null, @@ -413,6 +471,7 @@ function buildStatsModelData( totalModels: windowPeers.length, tokenShare: totalTokens > 0 ? round((current.totalTokens / totalTokens) * 100, 2) : 0, tokenChange: percentChange(current.totalTokens, previous.totalTokens), + retention7d, totals: { sessions: current.sessions, uniqueUsers: current.uniqueUsers, @@ -509,10 +568,41 @@ function emptyStatsHomeData(): StatsHomeData { tokenCost: createTokenProductRecord(() => []), cacheRatio: createTokenProductRecord(() => []), sessionCost: createTokenProductRecord(() => []), + retention: [], country: createRangeRecord(() => []), } } +export function buildRetentionEntries(rows: RetentionMetricRow[]): RetentionEntry[] { + const cohortDates = [...new Set(rows.map((row) => row.cohortDate))].toSorted().slice(-RETENTION_COHORT_DAYS) + const aggregate = rows + .filter((row) => cohortDates.includes(row.cohortDate)) + .reduce>>((result, row) => { + const current = result.get(row.model) + result.set(row.model, { + model: row.model, + provider: current?.provider ?? row.provider, + eligibleUserDays: (current?.eligibleUserDays ?? 0) + row.eligibleUsers, + retainedUserDays: (current?.retainedUserDays ?? 0) + row.retainedUsers, + }) + return result + }, new Map()) + const entries = [...aggregate.values()].map((item) => ({ + ...item, + author: formatProvider(item.provider), + rate: item.eligibleUserDays > 0 ? round((item.retainedUserDays / item.eligibleUserDays) * 100, 1) : 0, + })) + const ranks = new Map( + entries + .filter((item) => item.eligibleUserDays >= RETENTION_MIN_ELIGIBLE_USER_DAYS) + .toSorted((a, b) => b.rate - a.rate || b.eligibleUserDays - a.eligibleUserDays || a.model.localeCompare(b.model)) + .map((item, index) => [item.model, index + 1]), + ) + return entries + .map((item) => ({ ...item, rank: ranks.get(item.model) ?? null })) + .toSorted((a, b) => (a.rank ?? Number.MAX_SAFE_INTEGER) - (b.rank ?? Number.MAX_SAFE_INTEGER)) +} + function buildUsagePoints( rows: StatMetricRow[], product: UsageProduct, diff --git a/packages/stats/core/src/domain/inference.test.ts b/packages/stats/core/src/domain/inference.test.ts index ad95dad18b..c534ac2b7a 100644 --- a/packages/stats/core/src/domain/inference.test.ts +++ b/packages/stats/core/src/domain/inference.test.ts @@ -1,5 +1,12 @@ import { describe, expect, test } from "bun:test" -import { buildStatsQueries, toGeoAggregate, toModelAggregate, toProviderAggregate } from "./inference" +import { + buildRetentionQueries, + buildStatsQueries, + toGeoAggregate, + toModelAggregate, + toProviderAggregate, + toRetentionAggregate, +} from "./inference" import { modelAuthor, normalizeInferenceModel, statModel, statProvider } from "./model-normalization" describe("inference stat normalization", () => { @@ -155,6 +162,52 @@ describe("inference stat normalization", () => { expect(query).toContain("(source = 'inference-legacy' AND started_at < '2026-08-11T10:57:48.186Z')") expect(query).toContain("(source = 'inference' AND started_at >= '2026-08-11T10:57:48.186Z')") }) + + test("builds complete seven-day cohort retention queries", () => { + const queries = buildRetentionQueries(new Date("2026-08-10T00:00:00.000Z"), new Date("2026-08-20T00:00:00.000Z"), { + namespace: "inference", + table: "generation", + dataset: "zen", + }) + + expect(queries).toHaveLength(1) + expect(queries[0]?.cohortDates).toEqual(["2026-08-10", "2026-08-11", "2026-08-12"]) + expect(queries[0]?.query).toContain("ROW_NUMBER() OVER") + expect(queries[0]?.query).toContain("PARTITION BY cohort_date, user_key") + expect(queries[0]?.query).toContain("ORDER BY total_tokens DESC, requests DESC, model ASC") + expect(queries[0]?.query).toContain("WHEN '2026-08-17' THEN '2026-08-10'") + expect(queries[0]?.query).toContain("WHEN '2026-08-19' THEN '2026-08-12'") + expect(queries[0]?.query).toContain("started_at >= '2026-08-10T00:00:00.000Z'") + expect(queries[0]?.query).toContain("started_at < '2026-08-20T00:00:00.000Z'") + expect(queries[0]?.query).toContain("LEFT JOIN returned ON primary_models.user_key = returned.user_key") + expect(queries[0]?.query).toContain("primary_models.cohort_date = returned.cohort_date") + expect(queries[0]?.query).toContain("COUNT(*) AS eligible_users") + expect(queries[0]?.query).toContain("LIMIT 10000") + }) + + test("maps retention query results", () => { + expect( + toRetentionAggregate({ + cohort_date: "2026-08-10", + dataset: "zen", + tier: "all", + provider: "deepseek", + model: "deepseek-v4-flash-free", + eligible_users: "125", + retained_users: "74", + }), + ).toEqual([ + { + cohortDate: "2026-08-10", + dataset: "zen", + tier: "all", + provider: "deepseek", + model: "deepseek-v4-flash", + eligibleUsers: 125, + retainedUsers: 74, + }, + ]) + }) }) function aggregate(model: string, provider: string) { diff --git a/packages/stats/core/src/domain/inference.ts b/packages/stats/core/src/domain/inference.ts index ee0468407f..cf475d2809 100644 --- a/packages/stats/core/src/domain/inference.ts +++ b/packages/stats/core/src/domain/inference.ts @@ -12,6 +12,7 @@ import { statProvider, } from "./model-normalization" import type { ProviderStatAggregate } from "./provider" +import type { RetentionStatAggregate } from "./retention" import { normalizeCountry, normalizeTier, @@ -23,6 +24,7 @@ import { export type StatDimension = "model" | "provider" | "geo" | "geo_model" export type StatsQuerySource = { namespace: string; table: string; dataset: string } +export type RetentionQuery = { cohortDates: string[]; query: string } type StatsQueryFamily = "usage" | "geo" const DAY_MS = 86_400_000 @@ -46,6 +48,141 @@ export function buildStatsQueries(periodStart: Date, periodEnd: Date, input?: St ) } +export function buildRetentionQueries(periodStart: Date, periodEnd: Date, input?: StatsQuerySource): RetentionQuery[] { + const source = input ?? { + namespace: Resource.R2Sql.namespace, + table: Resource.R2Sql.table, + dataset: Resource.StatsSyncConfig.dataset, + } + const periods = retentionPeriods(periodStart, periodEnd) + if (periods.length === 0) return [] + return [ + { + cohortDates: periods.map((period) => period.start.toISOString().slice(0, 10)), + query: buildRetentionQuery(periods, source), + }, + ] +} + +function buildRetentionQuery( + periods: { start: Date; end: Date; returnStart: Date; returnEnd: Date }[], + source: StatsQuerySource, +) { + const first = periods[0] + const last = periods.at(-1)! + const scanStartValue = sqlString(first.start.toISOString()) + const scanEndValue = sqlString(last.returnEnd.toISOString()) + const ingestEndValue = sqlString(new Date(last.returnEnd.getTime() + DAY_MS).toISOString()) + const sourceTable = [source.namespace, source.table].map(sqlIdentifier).join(".") + const activityDates = [ + ...new Map( + periods.flatMap((period) => [period.start, period.returnStart]).map((date) => [date.toISOString(), date]), + ).values(), + ].toSorted((a, b) => a.getTime() - b.getTime()) + const activityDateSql = `CASE +${activityDates + .map( + (date) => + ` WHEN started_at >= ${sqlString(date.toISOString())} AND started_at < ${sqlString(new Date(date.getTime() + DAY_MS).toISOString())} THEN ${sqlString(date.toISOString().slice(0, 10))}`, + ) + .join("\n")} + ELSE null + END` + const cohortDates = periods.map((period) => sqlString(period.start.toISOString().slice(0, 10))).join(", ") + const returnDates = periods.map((period) => sqlString(period.returnStart.toISOString().slice(0, 10))).join(", ") + const returnCohortSql = `CASE activity_date +${periods + .map( + (period) => + ` WHEN ${sqlString(period.returnStart.toISOString().slice(0, 10))} THEN ${sqlString(period.start.toISOString().slice(0, 10))}`, + ) + .join("\n")} + END` + + return ` +WITH normalized AS ( + SELECT + ${activityDateSql} AS activity_date, + ${statModelSql("model_requested", "route_model")} AS model, + COALESCE(NULLIF(route_model, ''), '') AS provider_model, + COALESCE(NULLIF(provider_id, ''), '') AS raw_provider, + COALESCE(NULLIF(user_id, ''), NULLIF(workspace_id, ''), NULLIF(service_api_key_id, '')) AS user_key, + COALESCE(tokens_cache_read, 0) + COALESCE(tokens_cache_write, 0) + COALESCE(tokens_input, 0) + COALESCE(tokens_output, 0) AS tokens_total + FROM ${sourceTable} + WHERE event_type = 'generation.completed' + AND source IN ('inference', 'inference-legacy') + AND ( + (source = 'inference-legacy' AND started_at < ${sqlString(LIVE_SOURCE_START)}) + OR (source = 'inference' AND started_at >= ${sqlString(LIVE_SOURCE_START)}) + ) + AND (product = 'go' OR (${freeTierSql("model_tier", "model_requested")})) + AND model_requested IS NOT NULL + AND model_requested <> '' + AND __ingest_ts >= ${scanStartValue} + AND __ingest_ts < ${ingestEndValue} + AND started_at >= ${scanStartValue} + AND started_at < ${scanEndValue} +), filtered AS ( + SELECT + activity_date, + ${statProviderSql("model", "provider_model", "raw_provider")} AS provider, + model, + user_key, + tokens_total + FROM normalized + WHERE activity_date IS NOT NULL + AND user_key <> '' + AND lower(model) NOT IN (${[...EXCLUDED_MODELS].map(sqlString).join(", ")}) +), model_usage AS ( + SELECT + activity_date AS cohort_date, + user_key, + provider, + model, + SUM(tokens_total) AS total_tokens, + COUNT(*) AS requests + FROM filtered + WHERE activity_date IN (${cohortDates}) + GROUP BY activity_date, user_key, provider, model +), ranked_models AS ( + SELECT + cohort_date, + user_key, + provider, + model, + ROW_NUMBER() OVER ( + PARTITION BY cohort_date, user_key + ORDER BY total_tokens DESC, requests DESC, model ASC + ) AS model_rank + FROM model_usage +), primary_models AS ( + SELECT cohort_date, user_key, provider, model + FROM ranked_models + WHERE model_rank = 1 +), returned AS ( + SELECT + ${returnCohortSql} AS cohort_date, + user_key + FROM filtered + WHERE activity_date IN (${returnDates}) + GROUP BY ${returnCohortSql}, user_key +) +SELECT + primary_models.cohort_date, + ${sqlString(source.dataset)} AS dataset, + 'all' AS tier, + primary_models.provider, + primary_models.model, + COUNT(*) AS eligible_users, + SUM(CASE WHEN returned.user_key IS NULL THEN 0 ELSE 1 END) AS retained_users +FROM primary_models +LEFT JOIN returned ON primary_models.user_key = returned.user_key + AND primary_models.cohort_date = returned.cohort_date +GROUP BY primary_models.cohort_date, primary_models.provider, primary_models.model +LIMIT 10000 +` +} + function buildStatsQuery( period: { grain: "day" | "week"; key: string; start: Date; end: Date }, source: StatsQuerySource, @@ -223,6 +360,21 @@ export function toGeoAggregate(data: R2SqlData): GeoStatAggregate[] { ]) } +export function toRetentionAggregate(data: R2SqlData): RetentionStatAggregate[] { + if (!data.cohort_date || !data.model) return [] + return [ + { + cohortDate: data.cohort_date, + dataset: data.dataset || Resource.StatsSyncConfig.dataset, + tier: data.tier || "all", + provider: statProvider(data.model, "", data.provider) || "unknown", + model: statModel(data.model, undefined), + eligibleUsers: integer(data, "eligible_users"), + retainedUsers: integer(data, "retained_users"), + }, + ] +} + function toStatBaseAggregate(data: R2SqlData): StatBaseAggregate[] { const grain = data.grain === "day" || data.grain === "week" ? data.grain : undefined if (!grain || !data.period_key) return [] @@ -300,6 +452,18 @@ function statPeriods(grain: "day" | "week", periodStart: Date, periodEnd: Date) }) } +function retentionPeriods(periodStart: Date, periodEnd: Date) { + const first = startOfUtcDay(periodStart) + const last = new Date(startOfUtcDay(periodEnd).getTime() - WEEK_MS) + const count = Math.max(0, Math.floor((last.getTime() - first.getTime()) / DAY_MS)) + return Array.from({ length: count }, (_, index) => { + const start = new Date(first.getTime() + index * DAY_MS) + const end = new Date(start.getTime() + DAY_MS) + const returnStart = new Date(start.getTime() + WEEK_MS) + return { start, end, returnStart, returnEnd: new Date(returnStart.getTime() + DAY_MS) } + }) +} + function statModelSql(model: string, providerModel: string) { return `COALESCE(NULLIF(regexp_replace(CASE WHEN lower(${model}) = 'big-pickle' THEN regexp_replace(NULLIF(${providerModel}, ''), '^.*/', '') diff --git a/packages/stats/core/src/domain/retention.ts b/packages/stats/core/src/domain/retention.ts new file mode 100644 index 0000000000..14b154aac7 --- /dev/null +++ b/packages/stats/core/src/domain/retention.ts @@ -0,0 +1,110 @@ +import { and, eq, inArray } from "drizzle-orm" +import { Context, Effect, Layer } from "effect" +import { DatabaseError, DrizzleClient } from "../database" +import { modelRetention } from "../database/schema" +import { chunks, UPSERT_CHUNK_SIZE } from "./stat" + +export type RetentionStatRow = typeof modelRetention.$inferInsert +export type RetentionStatAggregate = { + cohortDate: string + dataset: string + tier: string + provider: string + model: string + eligibleUsers: number + retainedUsers: number +} + +export declare namespace RetentionStatRepo { + export interface Service { + readonly available: () => Effect.Effect + readonly replace: ( + rows: RetentionStatRow[], + scope: { cohortDates: string[]; dataset: string; tier: string }, + ) => Effect.Effect + } +} + +export class RetentionStatRepo extends Context.Service()( + "@opencode/stats/RetentionStatRepo", +) { + static readonly layer: Layer.Layer = Layer.effect( + RetentionStatRepo, + Effect.gen(function* () { + const db = yield* DrizzleClient + + const available = Effect.fn("RetentionStatRepo.available")(function* () { + return yield* Effect.tryPromise({ + try: async () => { + try { + await db.select({ id: modelRetention.id }).from(modelRetention).limit(1) + return true + } catch (cause) { + if (isMissingRetentionTable(cause)) return false + throw cause + } + }, + catch: (cause) => DatabaseError.make({ cause }), + }) + }) + + const replace = Effect.fn("RetentionStatRepo.replace")(function* ( + rows: RetentionStatRow[], + scope: { cohortDates: string[]; dataset: string; tier: string }, + ) { + if (scope.cohortDates.length === 0) return + + yield* Effect.tryPromise({ + try: () => + db + .delete(modelRetention) + .where( + and( + inArray(modelRetention.cohort_date, scope.cohortDates), + eq(modelRetention.dataset, scope.dataset), + eq(modelRetention.tier, scope.tier), + ), + ), + catch: (cause) => DatabaseError.make({ cause }), + }) + yield* Effect.forEach( + chunks(rows, UPSERT_CHUNK_SIZE), + (chunk) => + Effect.tryPromise({ + try: () => db.insert(modelRetention).values(chunk), + catch: (cause) => DatabaseError.make({ cause }), + }), + { discard: true }, + ) + }) + + return RetentionStatRepo.of({ available, replace }) + }), + ) +} + +export function rowsFromAggregates(aggregates: RetentionStatAggregate[]): RetentionStatRow[] { + return aggregates.map((row) => ({ + cohort_date: row.cohortDate, + dataset: row.dataset, + tier: row.tier, + provider: row.provider, + model: row.model, + eligible_users: row.eligibleUsers, + retained_users: row.retainedUsers, + })) +} + +export function isMissingRetentionTable(cause: unknown): boolean { + const text = errorText(cause).toLowerCase() + return text.includes("model_retention") && text.includes("exist") +} + +function errorText(cause: unknown): string { + if (cause instanceof Error) return `${cause.message} ${errorText((cause as { cause?: unknown }).cause)}` + if (typeof cause === "object" && cause) + return Object.values(cause as Record) + .map(errorText) + .join(" ") + return String(cause) +} diff --git a/packages/stats/core/src/index.ts b/packages/stats/core/src/index.ts index 52ff565cb8..834625f601 100644 --- a/packages/stats/core/src/index.ts +++ b/packages/stats/core/src/index.ts @@ -6,6 +6,7 @@ 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 RetentionStat from "./domain/retention" export * as Stat from "./domain/stat" export * as Runtime from "./runtime" export * as StatSync from "./stat-sync" diff --git a/packages/stats/core/src/runtime.ts b/packages/stats/core/src/runtime.ts index cc1dccad24..1c0a7b8ac5 100644 --- a/packages/stats/core/src/runtime.ts +++ b/packages/stats/core/src/runtime.ts @@ -4,10 +4,14 @@ import { layer as databaseLayer } from "./database" import { GeoStatRepo } from "./domain/geo" import { ModelStatRepo } from "./domain/model" import { ProviderStatRepo } from "./domain/provider" +import { RetentionStatRepo } from "./domain/retention" -const repoLayer = Layer.mergeAll(ModelStatRepo.layer, ProviderStatRepo.layer, GeoStatRepo.layer).pipe( - Layer.provide(databaseLayer), -) +const repoLayer = Layer.mergeAll( + ModelStatRepo.layer, + ProviderStatRepo.layer, + GeoStatRepo.layer, + RetentionStatRepo.layer, +).pipe(Layer.provide(databaseLayer)) export const layer = Layer.mergeAll(AppConfig.layer, databaseLayer, repoLayer) export const runtime = ManagedRuntime.make(layer) diff --git a/packages/stats/core/src/stat-sync.ts b/packages/stats/core/src/stat-sync.ts index ceec6f7e6d..cd8cdf35b6 100644 --- a/packages/stats/core/src/stat-sync.ts +++ b/packages/stats/core/src/stat-sync.ts @@ -2,16 +2,25 @@ import { DateTime, Effect } from "effect" import { Resource } from "sst/resource" import { DatabaseError } from "./database" import { GeoStatRepo, rowsFromAggregates as geoRowsFromAggregates } from "./domain/geo" -import { buildStatsQueries, toGeoAggregate, toModelAggregate, toProviderAggregate } from "./domain/inference" +import { + buildRetentionQueries, + buildStatsQueries, + toGeoAggregate, + toModelAggregate, + toProviderAggregate, + toRetentionAggregate, +} from "./domain/inference" import { ModelStatRepo, rowsFromAggregates as modelRowsFromAggregates } from "./domain/model" import { ProviderStatRepo, rowsFromAggregates as providerRowsFromAggregates } from "./domain/provider" -import { startOfIsoWeek } from "./domain/stat" +import { RetentionStatRepo, rowsFromAggregates as retentionRowsFromAggregates } from "./domain/retention" +import { startOfIsoWeek, startOfUtcDay } from "./domain/stat" import { R2Sql, R2SqlQueryError } from "./r2-sql" 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 const DISPLAY_WINDOW_MS = 56 * 86_400_000 +const RETENTION_INCREMENTAL_LOOKBACK_MS = 9 * 86_400_000 // Anchor incremental passes to the ISO week containing this lookback, so the pass // after a week boundary still recomputes the previous week's final aggregates even // if the boundary pass itself failed. @@ -19,11 +28,12 @@ const INCREMENTAL_LOOKBACK_MS = 2 * 3_600_000 export type SyncStatsResult = { ok: true; rows: number; startedAt: string; periodStart: string; periodEnd: string } export type SyncStatsError = R2SqlQueryError | DatabaseError +type SyncStatsServices = R2Sql | ModelStatRepo | ProviderStatRepo | GeoStatRepo | RetentionStatRepo export const syncStats: (options?: { full?: boolean -}) => Effect.Effect = - Effect.fn("StatSync.sync")(function* (options?: { full?: boolean }) { +}) => Effect.Effect = Effect.fn("StatSync.sync")( + function* (options?: { full?: boolean }) { const startedAt = yield* DateTime.nowAsDate const periodEnd = new Date(Math.floor((startedAt.getTime() - DATALAKE_INGESTION_LAG_MS) / 60_000) * 60_000) const periodStart = options?.full ? fullPeriodStart(periodEnd) : incrementalPeriodStart(periodEnd) @@ -31,6 +41,7 @@ export const syncStats: (options?: { const modelStats = yield* ModelStatRepo const providerStats = yield* ProviderStatRepo const geoStats = yield* GeoStatRepo + const retentionStats = yield* RetentionStatRepo yield* logRuntimeCheck() @@ -44,11 +55,39 @@ export const syncStats: (options?: { const geoRows = geoRowsFromAggregates( rows.filter((row) => row.dimension === "geo" || row.dimension === "geo_model").flatMap(toGeoAggregate), ) + const retentionAvailable = yield* retentionStats.available() + const retentionQueries = retentionAvailable + ? buildRetentionQueries( + options?.full + ? periodStart + : new Date( + Math.max(startOfUtcDay(periodEnd).getTime() - RETENTION_INCREMENTAL_LOOKBACK_MS, STATS_DATA_START_MS), + ), + startOfUtcDay(periodEnd), + ) + : [] + const retentionRows = retentionRowsFromAggregates( + yield* Effect.forEach(retentionQueries, (item) => r2Sql.query(item.query), { concurrency: 4 }).pipe( + Effect.map((batches) => batches.flatMap((batch) => batch.flatMap(toRetentionAggregate))), + ), + ) - yield* Effect.all([modelStats.upsert(modelRows), providerStats.upsert(providerRows), geoStats.upsert(geoRows)], { - concurrency: "unbounded", - discard: true, - }) + yield* Effect.all( + [ + modelStats.upsert(modelRows), + providerStats.upsert(providerRows), + geoStats.upsert(geoRows), + retentionStats.replace(retentionRows, { + cohortDates: retentionQueries.flatMap((item) => item.cohortDates), + dataset: Resource.StatsSyncConfig.dataset, + tier: "all", + }), + ], + { + concurrency: "unbounded", + discard: true, + }, + ) yield* Effect.all( [ modelStats.deleteRetiredDimensions(modelRows), @@ -66,6 +105,8 @@ export const syncStats: (options?: { rows: modelRows.length, providerRows: providerRows.length, geoRows: geoRows.length, + retentionRows: retentionRows.length, + retentionAvailable, stage: Resource.App.stage, })}`, ) @@ -77,7 +118,8 @@ export const syncStats: (options?: { periodStart: periodStart.toISOString(), periodEnd: periodEnd.toISOString(), } - }) + }, +) // May 27 was partial, so keep stats anchored at the first complete day. function fullPeriodStart(periodEnd: Date) {