feat(stats): add retention metrics
This commit is contained in:
Adam
parent
1216c55094
commit
a0f36c9df7
13 changed files with 841 additions and 18 deletions
|
|
@ -470,6 +470,7 @@ function ModelMomentumSection(props: { data: StatsModelPageData | null }) {
|
|||
value={formatInteger(data().totals.sessions)}
|
||||
/>
|
||||
<MomentumMetric label={i18n.t("model.tokenShare")} value={formatPercent(data().tokenShare)} />
|
||||
<MomentumMetric label="7D Retention" value={formatModelRetention(data())} />
|
||||
<MomentumMetric
|
||||
label="Rank"
|
||||
value={formatRankLabel(data().rank)}
|
||||
|
|
@ -540,6 +541,11 @@ function MomentumMetric(props: { label: string; value: string; watermark?: strin
|
|||
)
|
||||
}
|
||||
|
||||
function formatModelRetention(data: StatsModelPageData) {
|
||||
if (!data.retention7d || data.retention7d.eligibleUserDays < 100) return "Pending"
|
||||
return formatPercent(data.retention7d.rate)
|
||||
}
|
||||
|
||||
function ModelUsageSection(props: { data: StatsModelPageData | null }) {
|
||||
const i18n = useI18n()
|
||||
return (
|
||||
|
|
|
|||
|
|
@ -1737,6 +1737,7 @@ body {
|
|||
:is(
|
||||
[data-section="leaderboard"],
|
||||
[data-section="unique-users"],
|
||||
[data-section="retention"],
|
||||
[data-section="market-share"],
|
||||
[data-section="geo-breakdown"],
|
||||
[data-section="token-cost"],
|
||||
|
|
@ -1774,6 +1775,125 @@ body {
|
|||
white-space: nowrap;
|
||||
}
|
||||
|
||||
[data-page="stats"] [data-component="retention-chart"] {
|
||||
display: grid;
|
||||
gap: 8px;
|
||||
min-width: 0;
|
||||
margin-top: 40px;
|
||||
}
|
||||
|
||||
[data-page="stats"] [data-slot="retention-heading"],
|
||||
[data-page="stats"] [data-component="retention-chart"] a {
|
||||
display: grid;
|
||||
grid-template-columns: 40px minmax(160px, 260px) minmax(180px, 1fr) 64px 84px;
|
||||
align-items: center;
|
||||
gap: 16px;
|
||||
min-width: 0;
|
||||
}
|
||||
|
||||
[data-page="stats"] [data-slot="retention-heading"] {
|
||||
min-height: 28px;
|
||||
color: var(--stats-faint);
|
||||
font-size: 11px;
|
||||
font-style: normal;
|
||||
font-weight: 500;
|
||||
line-height: 14px;
|
||||
text-transform: uppercase;
|
||||
}
|
||||
|
||||
[data-page="stats"] [data-slot="retention-heading"] :is(strong, b, em, i) {
|
||||
font: inherit;
|
||||
font-style: normal;
|
||||
}
|
||||
|
||||
[data-page="stats"] [data-slot="retention-heading"] :is(b, em) {
|
||||
text-align: right;
|
||||
}
|
||||
|
||||
[data-page="stats"] [data-component="retention-chart"] ol {
|
||||
display: grid;
|
||||
gap: 4px;
|
||||
min-width: 0;
|
||||
margin: 0;
|
||||
padding: 0;
|
||||
list-style: none;
|
||||
}
|
||||
|
||||
[data-page="stats"] [data-component="retention-chart"] a {
|
||||
box-sizing: border-box;
|
||||
min-height: 48px;
|
||||
padding: 8px 12px;
|
||||
border: 1px solid var(--stats-line);
|
||||
color: var(--stats-text);
|
||||
background: var(--stats-layer);
|
||||
text-decoration: none;
|
||||
}
|
||||
|
||||
[data-page="stats"] [data-component="retention-chart"] a:hover,
|
||||
[data-page="stats"] [data-component="retention-chart"] a:focus-visible,
|
||||
[data-page="stats"] [data-component="retention-chart"] a[data-active="true"] {
|
||||
border-color: var(--stats-line-strong);
|
||||
background: var(--stats-layer-2);
|
||||
outline: none;
|
||||
}
|
||||
|
||||
[data-page="stats"] [data-component="retention-chart"] a > 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"],
|
||||
|
|
|
|||
|
|
@ -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() {
|
|||
<Hero updatedAt={stats().updatedAt} />
|
||||
<TopModelsSection data={stats().usage} leaderboard={stats().leaderboard} />
|
||||
<UniqueUsersSection data={stats().users} />
|
||||
<RetentionSection data={stats().retention} />
|
||||
<SessionCostSection data={stats().sessionCost} />
|
||||
<TokenCostSection data={stats().tokenCost} />
|
||||
<CacheRatioSection data={stats().cacheRatio} />
|
||||
|
|
@ -616,6 +620,86 @@ function UniqueUsersSection(props: { data: UsagePoint[] }) {
|
|||
)
|
||||
}
|
||||
|
||||
function RetentionSection(props: { data: RetentionEntry[] }) {
|
||||
const language = useLanguage()
|
||||
const [activeIndex, setActiveIndex] = createSignal(0)
|
||||
|
||||
return (
|
||||
<section id="retention" data-section="retention">
|
||||
<SectionTitle
|
||||
id="retention"
|
||||
title="7-Day Retention"
|
||||
description="Share of users returning to any OpenCode model seven days later, grouped by their primary model. Latest seven complete cohorts; minimum 100 eligible user-days."
|
||||
/>
|
||||
<Show
|
||||
when={props.data.length > 0}
|
||||
fallback={
|
||||
<EmptyState
|
||||
title="No retention data"
|
||||
description="Retention appears after seven complete user cohorts are available."
|
||||
/>
|
||||
}
|
||||
>
|
||||
<div data-component="retention-chart">
|
||||
<div data-slot="retention-heading" aria-hidden="true">
|
||||
<span>Rank</span>
|
||||
<strong>Model</strong>
|
||||
<i>Retention</i>
|
||||
<b>Rate</b>
|
||||
<em>Eligible</em>
|
||||
</div>
|
||||
<ol>
|
||||
<For each={props.data}>
|
||||
{(item, index) => (
|
||||
<li>
|
||||
<a
|
||||
data-active={activeIndex() === index() ? "true" : undefined}
|
||||
href={language.route(
|
||||
`${import.meta.env.BASE_URL}${modelSlug(item.provider)}/${modelSlug(item.model)}`,
|
||||
)}
|
||||
onPointerEnter={() => setActiveIndex(index())}
|
||||
onFocus={() => setActiveIndex(index())}
|
||||
onClick={() => setActiveIndex(index())}
|
||||
aria-label={`${item.model}, ${formatRetentionRate(item.rate)} seven-day retention, ${formatUsers(item.eligibleUserDays)} eligible user-days`}
|
||||
>
|
||||
<span>{item.rank === null ? "–" : String(item.rank).padStart(2, "0")}</span>
|
||||
<strong>{item.model}</strong>
|
||||
<RetentionMarker rate={item.rate} active={activeIndex() === index()} />
|
||||
<b>{formatRetentionRate(item.rate)}</b>
|
||||
<em>{formatUsers(item.eligibleUserDays)}</em>
|
||||
</a>
|
||||
</li>
|
||||
)}
|
||||
</For>
|
||||
</ol>
|
||||
</div>
|
||||
</Show>
|
||||
</section>
|
||||
)
|
||||
}
|
||||
|
||||
function RetentionMarker(props: { rate: number; active: boolean }) {
|
||||
const fill = createMemo(() => Math.min(100, Math.max(0, props.rate)))
|
||||
return (
|
||||
<i
|
||||
data-component="retention-marker"
|
||||
data-active={props.active ? "true" : undefined}
|
||||
style={{ "--retention-position": `${fill()}%` } as JSX.CSSProperties}
|
||||
aria-hidden="true"
|
||||
>
|
||||
<span />
|
||||
<span />
|
||||
<span />
|
||||
<span />
|
||||
<em />
|
||||
</i>
|
||||
)
|
||||
}
|
||||
|
||||
function formatRetentionRate(value: number) {
|
||||
return `${value.toFixed(1)}%`
|
||||
}
|
||||
|
||||
function isTopModelsBlankHover(bar: HTMLElement, clientY: number) {
|
||||
const stack = bar.querySelector<HTMLElement>('[data-slot="top-models-stack"]')
|
||||
if (!stack) return true
|
||||
|
|
|
|||
|
|
@ -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`);
|
||||
|
|
@ -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(),
|
||||
|
|
|
|||
48
packages/stats/core/src/domain/home.test.ts
Normal file
48
packages/stats/core/src/domain/home.test.ts
Normal file
|
|
@ -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[]
|
||||
}
|
||||
|
|
@ -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<TokenProduct, TokenCostEntry[]>
|
||||
cacheRatio: Record<TokenProduct, CacheRatioEntry[]>
|
||||
sessionCost: Record<TokenProduct, SessionCostEntry[]>
|
||||
retention: RetentionEntry[]
|
||||
country: Record<UsageRange, CountryEntry[]>
|
||||
}
|
||||
|
||||
|
|
@ -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<GeoStatMetric, "updatedAt"> & {
|
|||
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<string, unknown>
|
|||
export function getStatsHomeData(): Effect.Effect<StatsHomeData, StatsDataError> {
|
||||
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<StatsModelData | null, StatsDataError> {
|
||||
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<RetentionMetricRow[]> {
|
||||
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<Map<string, Omit<RetentionEntry, "author" | "rate" | "rank">>>((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,
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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}, ''), '^.*/', '')
|
||||
|
|
|
|||
110
packages/stats/core/src/domain/retention.ts
Normal file
110
packages/stats/core/src/domain/retention.ts
Normal file
|
|
@ -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<boolean, DatabaseError>
|
||||
readonly replace: (
|
||||
rows: RetentionStatRow[],
|
||||
scope: { cohortDates: string[]; dataset: string; tier: string },
|
||||
) => Effect.Effect<void, DatabaseError>
|
||||
}
|
||||
}
|
||||
|
||||
export class RetentionStatRepo extends Context.Service<RetentionStatRepo, RetentionStatRepo.Service>()(
|
||||
"@opencode/stats/RetentionStatRepo",
|
||||
) {
|
||||
static readonly layer: Layer.Layer<RetentionStatRepo, never, DrizzleClient> = 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<string, unknown>)
|
||||
.map(errorText)
|
||||
.join(" ")
|
||||
return String(cause)
|
||||
}
|
||||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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<SyncStatsResult, SyncStatsError, R2Sql | ModelStatRepo | ProviderStatRepo | GeoStatRepo> =
|
||||
Effect.fn("StatSync.sync")(function* (options?: { full?: boolean }) {
|
||||
}) => Effect.Effect<SyncStatsResult, SyncStatsError, SyncStatsServices> = 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) {
|
||||
|
|
|
|||
Loading…
Reference in a new issue