EnterpriseAI/ClosedCode
1
0
Fork 0
Code Issues Pull Requests Actions 27 Packages Projects Releases Wiki Activity

chore: generate

Browse Source
This commit is contained in:
opencode-agent[bot]
2026-07-10 11:38:07 +00:00
parent 08a2a623dc
commit 7dbfd0b934
1 changed files with 50 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/stats/core/src/stat-sync.ts
+50 -56
View File
@@ -20,65 +20,62 @@ const INCREMENTAL_LOOKBACK_MS = 2 * 3_600_000
export type SyncStatsResult = { ok: true; rows: number; startedAt: string; periodStart: string; periodEnd: string } export type SyncStatsResult = { ok: true; rows: number; startedAt: string; periodStart: string; periodEnd: string }
export type SyncStatsError = AthenaQueryError | AthenaQueryTimeoutError | DatabaseError export type SyncStatsError = AthenaQueryError | AthenaQueryTimeoutError | DatabaseError
export const syncStats: (options?: { full?: boolean }) => Effect.Effect< export const syncStats: (options?: {
SyncStatsResult, full?: boolean
SyncStatsError, }) => Effect.Effect<SyncStatsResult, SyncStatsError, Athena | ModelStatRepo | ProviderStatRepo | GeoStatRepo> =
Athena | ModelStatRepo | ProviderStatRepo | GeoStatRepo Effect.fn("StatSync.sync")(function* (options?: { full?: boolean }) {
> = Effect.fn("StatSync.sync")(function* (options?: { full?: boolean }) { const startedAt = yield* DateTime.nowAsDate
const startedAt = yield* DateTime.nowAsDate const periodEnd = new Date(Math.floor((startedAt.getTime() - DATALAKE_INGESTION_LAG_MS) / 60_000) * 60_000)
const periodEnd = new Date(Math.floor((startedAt.getTime() - DATALAKE_INGESTION_LAG_MS) / 60_000) * 60_000) const periodStart = options?.full ? fullPeriodStart(periodEnd) : incrementalPeriodStart(periodEnd)
const periodStart = options?.full ? fullPeriodStart(periodEnd) : incrementalPeriodStart(periodEnd) const athena = yield* Athena
const athena = yield* Athena const modelStats = yield* ModelStatRepo
const modelStats = yield* ModelStatRepo const providerStats = yield* ProviderStatRepo
const providerStats = yield* ProviderStatRepo const geoStats = yield* GeoStatRepo
const geoStats = yield* GeoStatRepo
yield* logRuntimeCheck() yield* logRuntimeCheck()
const rows = yield* athena.query(buildStatsQuery(periodStart, periodEnd)) const rows = yield* athena.query(buildStatsQuery(periodStart, periodEnd))
const modelRows = modelRowsFromAggregates( const modelRows = modelRowsFromAggregates(rows.filter((row) => row.dimension === "model").flatMap(toModelAggregate))
rows.filter((row) => row.dimension === "model").flatMap(toModelAggregate), const providerRows = providerRowsFromAggregates(
) rows.filter((row) => row.dimension === "provider").flatMap(toProviderAggregate),
const providerRows = providerRowsFromAggregates( )
rows.filter((row) => row.dimension === "provider").flatMap(toProviderAggregate), const geoRows = geoRowsFromAggregates(
) rows.filter((row) => row.dimension === "geo" || row.dimension === "geo_model").flatMap(toGeoAggregate),
const geoRows = geoRowsFromAggregates( )
rows.filter((row) => row.dimension === "geo" || row.dimension === "geo_model").flatMap(toGeoAggregate),
)
yield* Effect.all([modelStats.upsert(modelRows), providerStats.upsert(providerRows), geoStats.upsert(geoRows)], { yield* Effect.all([modelStats.upsert(modelRows), providerStats.upsert(providerRows), geoStats.upsert(geoRows)], {
concurrency: "unbounded", concurrency: "unbounded",
discard: true, discard: true,
}) })
yield* Effect.all( yield* Effect.all(
[ [
modelStats.deleteRetiredDimensions(modelRows), modelStats.deleteRetiredDimensions(modelRows),
providerStats.deleteRetiredDimensions(providerRows), providerStats.deleteRetiredDimensions(providerRows),
geoStats.deleteRetiredDimensions(geoRows), geoStats.deleteRetiredDimensions(geoRows),
], ],
{ concurrency: "unbounded", discard: true }, { concurrency: "unbounded", discard: true },
) )
yield* Effect.logInfo( yield* Effect.logInfo(
`stats sync complete ${JSON.stringify({ `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(), startedAt: startedAt.toISOString(),
periodStart: periodStart.toISOString(), periodStart: periodStart.toISOString(),
periodEnd: periodEnd.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(),
}
})
// May 27 was partial, so keep Athena stats anchored at the first complete day. // May 27 was partial, so keep Athena stats anchored at the first complete day.
function fullPeriodStart(periodEnd: Date) { function fullPeriodStart(periodEnd: Date) {
@@ -96,10 +93,7 @@ function fullPeriodStart(periodEnd: Date) {
// dimension cleanup). // dimension cleanup).
function incrementalPeriodStart(periodEnd: Date) { function incrementalPeriodStart(periodEnd: Date) {
return new Date( return new Date(
Math.max( Math.max(startOfIsoWeek(new Date(periodEnd.getTime() - INCREMENTAL_LOOKBACK_MS)).getTime(), STATS_DATA_START_MS),
startOfIsoWeek(new Date(periodEnd.getTime() - INCREMENTAL_LOOKBACK_MS)).getTime(),
STATS_DATA_START_MS,
),
) )
} }