diff --git a/server/apps/api/src/app.ts b/server/apps/api/src/app.ts index 15d17b367..d944e5a29 100644 --- a/server/apps/api/src/app.ts +++ b/server/apps/api/src/app.ts @@ -49,11 +49,8 @@ import { resolveRequestAuth } from './libs/request-auth' import { createUnauthorizedWsEvents } from './libs/ws-auth' import { sessionMiddleware } from './middlewares/auth' import { emitOtelLog, initOtel } from './otel' -import { registerActiveSessionsGauge } from './otel/gauges/active-sessions' -import { registerDistinctActiveUsersGauge } from './otel/gauges/distinct-active-users' -import { registerRollingActiveUsersGauge } from './otel/gauges/rolling-active-users' -import { registerTotalUsersGauge } from './otel/gauges/total-users' import { registerTtsPoolGauge } from './otel/gauges/tts-pool' +import { createDiscardingUserMetricsSnapshotRecorder, registerUserMetricsSnapshotGauges } from './otel/gauges/user-metrics-snapshot' import { registerWsOnlineUsersGauge } from './otel/gauges/ws-online-users' import { createAdminRoutes } from './routes/admin' import { createAdminUiRoutes } from './routes/admin-ui' @@ -128,6 +125,9 @@ interface AppDeps { export async function buildApp(deps: AppDeps) { const logger = useLogger('app').useGlobalConfig() + const userMetricsRecorder = deps.otel + ? registerUserMetricsSnapshotGauges(deps.otel.auth) + : createDiscardingUserMetricsSnapshotRecorder() const app = new Hono() .use('*', async (c, next) => { @@ -456,6 +456,7 @@ export async function buildApp(deps: AppDeps) { db: deps.db, billingService: deps.billingService, configKV: deps.configKV, + userMetricsRecorder, })) /** @@ -823,21 +824,7 @@ export async function createApp() { providerCatalogService, ttsConcurrencyLedger, }) - // Register the cluster-wide ObservableGauges for sessions / users. Each - // replica polls the same DB (cached inside each gauge, in-flight coalesced); - // dashboards aggregate with avg()/max(), not sum(). See - // observability-conventions.md. - // - // Both gauges share the same `session` table: `user.active_sessions` is - // `COUNT(*)` (row inflation prone), `user.distinct_active` is - // `COUNT(DISTINCT user_id)` (real active-user count). Comparing the two - // surfaces session-row leakage from missing GC + per-OIDC-token row - // creation. if (resolved.otel) { - registerTotalUsersGauge(resolved.otel.auth.totalUsers, resolved.db, resolved.otel.observability.metricReadErrors) - registerActiveSessionsGauge(resolved.otel.auth.activeSessions, resolved.db, resolved.otel.observability.metricReadErrors) - registerDistinctActiveUsersGauge(resolved.otel.auth.distinctActiveUsers, resolved.db, resolved.otel.observability.metricReadErrors) - registerRollingActiveUsersGauge(resolved.otel.auth.rollingActiveUsers, resolved.db, resolved.otel.observability.metricReadErrors) registerTtsPoolGauge(resolved.otel.gateway.poolInflight, resolved.ttsConcurrencyLedger, resolved.otel.observability.metricReadErrors) registerWsOnlineUsersGauge(resolved.otel.engagement.wsUsersOnline, resolved.redis, resolved.otel.observability.metricReadErrors) } diff --git a/server/apps/api/src/otel/gauges/active-sessions.ts b/server/apps/api/src/otel/gauges/active-sessions.ts deleted file mode 100644 index ee5540795..000000000 --- a/server/apps/api/src/otel/gauges/active-sessions.ts +++ /dev/null @@ -1,102 +0,0 @@ -import type { AuthMetrics, ObservabilityMetrics } from '..' -import type { Database } from '../../libs/db' - -import { useLogger } from '@guiiai/logg' -import { count, gt } from 'drizzle-orm' - -import { session as sessionTable } from '../../schemas/accounts' - -/** - * Wire the `user.active_sessions` ObservableGauge to a Postgres `COUNT(*)` - * over the Better Auth session table. - * - * Use when: - * - Assembling DI in `createApp()`, exactly once per process. - * - * Expects: - * - `gauge` is the ObservableGauge handle created in `initOtel`. - * - `db` is the migrated Drizzle handle. - * - `metricReadErrors` is the shared counter used to track failures inside - * metric callbacks — increments are labelled with the originating metric - * name so on-call can spot which gauge is degraded. - * - * Multi-replica note: - * - This is a cluster-wide gauge — every replica reads the same DB and - * reports the same value. Dashboards MUST aggregate with `avg()`, NOT - * `sum()`. See observability-conventions.md. - * - * Concurrency: - * - Multiple OTel collection cycles can race (forced flushes, multiple - * readers). The in-flight promise lock keeps at most one DB query in - * flight per process; all other concurrent callbacks await the same - * result instead of stampeding the DB. - * - * Failure mode: - * - On DB error we increment `airi.observability.read_errors{metric}` and - * intentionally DO NOT call `result.observe(...)`. Letting the gauge - * skip an export cycle lets Prometheus staleness handle "DB is broken" - * correctly — an absence-based alert will fire after ~5 minutes. The - * previous version silently observed the stale cached value forever, - * which masked permanent DB failures. - */ -export function registerActiveSessionsGauge( - gauge: AuthMetrics['activeSessions'], - db: Database, - metricReadErrors: ObservabilityMetrics['metricReadErrors'], -) { - const log = useLogger('active-sessions-gauge').useGlobalConfig() - const CACHE_TTL_MS = 10_000 - - let cachedAt = 0 - let cachedCount = 0 - // Single shared promise representing "a refresh is in progress". All - // callbacks that arrive during a refresh attach to this and observe the - // same outcome. Reset to null when the refresh resolves. - let refreshInFlight: Promise | null = null - - async function refresh(): Promise { - try { - // Use the app clock (`new Date()`) rather than DB clock (`NOW()`) so - // we agree with Better Auth's own session validity check, which uses - // `new Date()` in its session lookup (`better-auth/dist/session.mjs` - // and `dist/internal-adapter.mjs`). A DB/app clock skew would - // otherwise let this gauge disagree with auth-layer reality. - const rows = await db - .select({ count: count() }) - .from(sessionTable) - .where(gt(sessionTable.expiresAt, new Date())) - cachedCount = Number(rows[0]?.count ?? 0) - cachedAt = Date.now() - return true - } - catch (err) { - log.withError(err).warn('Failed to read active sessions for gauge') - metricReadErrors.add(1, { metric: 'user.active_sessions' }) - return false - } - } - - gauge.addCallback(async (result) => { - const now = Date.now() - - // Cache fresh — serve last good value without touching the DB. - if (cachedAt !== 0 && now - cachedAt < CACHE_TTL_MS) { - result.observe(cachedCount) - return - } - - // Coalesce concurrent refreshes onto one in-flight promise. - if (!refreshInFlight) { - refreshInFlight = refresh().finally(() => { - refreshInFlight = null - }) - } - const ok = await refreshInFlight - - if (ok) { - result.observe(cachedCount) - } - // else: deliberately do nothing — let Prometheus staleness expose the - // outage instead of masking it with a stale cached number. - }) -} diff --git a/server/apps/api/src/otel/gauges/distinct-active-users.ts b/server/apps/api/src/otel/gauges/distinct-active-users.ts deleted file mode 100644 index a6f4e7269..000000000 --- a/server/apps/api/src/otel/gauges/distinct-active-users.ts +++ /dev/null @@ -1,94 +0,0 @@ -import type { AuthMetrics, ObservabilityMetrics } from '..' -import type { Database } from '../../libs/db' - -import { useLogger } from '@guiiai/logg' -import { countDistinct, gt } from 'drizzle-orm' - -import { session as sessionTable } from '../../schemas/accounts' - -/** - * Wire the `user.distinct_active` ObservableGauge to a Postgres - * `COUNT(DISTINCT user_id)` over the Better Auth session table. - * - * Use when: - * - Assembling DI in `createApp()`, exactly once per process. - * - * Why this exists alongside `registerActiveSessionsGauge`: - * - `user.active_sessions` is `COUNT(*)` — counts session **rows**. Better - * Auth creates a new row per sign-in and per OIDC access-token issuance - * (the `oauth_access_token` table has a FK to `session.id`) and never GCs - * expired rows, so the row-count drifts up over time independently of - * the real user base. We've seen this metric show ~80K on a small - * deployment where the real distinct-user count is ~hundreds. - * - `user.distinct_active` is `COUNT(DISTINCT user_id)` — the actual - * active-user gauge. Pair with `user.active_sessions` to spot session - * inflation: if rows / users ratio climbs past ~5 it's probably time - * to add a session-GC cron or shorten Better Auth's `expiresIn`. - * - * Multi-replica note: - * - Cluster-wide gauge — every replica reads the same DB and reports the - * same value. Dashboards MUST aggregate with `avg()`, NOT `sum()`. See - * observability-conventions.md. - * - * Concurrency: - * - Same in-flight promise lock pattern as `registerActiveSessionsGauge`, - * so concurrent OTel collection cycles fold into one DB query. - * - * Failure mode: - * - DB error → increment `airi.observability.read_errors{metric}` and skip - * `result.observe(...)`. Prometheus staleness exposes the outage instead - * of pinning a stale cached value forever. - */ -export function registerDistinctActiveUsersGauge( - gauge: AuthMetrics['distinctActiveUsers'], - db: Database, - metricReadErrors: ObservabilityMetrics['metricReadErrors'], -) { - const log = useLogger('distinct-active-users-gauge').useGlobalConfig() - const CACHE_TTL_MS = 10_000 - - let cachedAt = 0 - let cachedCount = 0 - let refreshInFlight: Promise | null = null - - async function refresh(): Promise { - try { - // Use the app clock (`new Date()`) for the same reason as - // `registerActiveSessionsGauge`: agree with Better Auth's own - // session-validity check, which uses `new Date()` rather than - // `NOW()`. - const rows = await db - .select({ count: countDistinct(sessionTable.userId) }) - .from(sessionTable) - .where(gt(sessionTable.expiresAt, new Date())) - cachedCount = Number(rows[0]?.count ?? 0) - cachedAt = Date.now() - return true - } - catch (err) { - log.withError(err).warn('Failed to read distinct active users for gauge') - metricReadErrors.add(1, { metric: 'user.distinct_active' }) - return false - } - } - - gauge.addCallback(async (result) => { - const now = Date.now() - - if (cachedAt !== 0 && now - cachedAt < CACHE_TTL_MS) { - result.observe(cachedCount) - return - } - - if (!refreshInFlight) { - refreshInFlight = refresh().finally(() => { - refreshInFlight = null - }) - } - const ok = await refreshInFlight - - if (ok) { - result.observe(cachedCount) - } - }) -} diff --git a/server/apps/api/src/otel/gauges/rolling-active-users.test.ts b/server/apps/api/src/otel/gauges/rolling-active-users.test.ts deleted file mode 100644 index 1489db079..000000000 --- a/server/apps/api/src/otel/gauges/rolling-active-users.test.ts +++ /dev/null @@ -1,138 +0,0 @@ -import type { AuthMetrics, ObservabilityMetrics } from '..' -import type { Database } from '../../libs/db' - -import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' - -import { registerRollingActiveUsersGauge } from './rolling-active-users' - -/** - * Capture the callback registered via `gauge.addCallback` and a spyable - * `observe` so tests can drive collection cycles by hand. - * - * @example - * const { gauge, observe, run } = makeGauge() - * register...(gauge, db, errs) - * await run() - * expect(observe).toHaveBeenCalledWith(5, { window: '24h' }) - */ -function makeGauge() { - let cb: ((result: { observe: (v: number, attrs: Record) => void }) => void | Promise) | null = null - const observe = vi.fn() - const gauge = { - addCallback: vi.fn((fn: typeof cb) => { cb = fn }), - } as unknown as AuthMetrics['rollingActiveUsers'] - return { - gauge, - observe, - run: async () => { - if (!cb) - throw new Error('no callback registered') - await cb({ observe }) - }, - } -} - -/** - * Drizzle's `db.select({...}).from(table)` is awaited directly (thenable - * query builder). Mock it as `select -> { from: () => Promise }`. - * - * @example - * const db = makeDb([{ dau: '5', wau: '10', mau: '20' }]) - */ -function makeDb(rows: unknown[], opts: { reject?: boolean } = {}) { - const from = vi.fn(() => (opts.reject ? Promise.reject(new Error('db down')) : Promise.resolve(rows))) - const select = vi.fn(() => ({ from })) - return { db: { select } as unknown as Database, select, from } -} - -function makeReadErrors() { - const add = vi.fn() - return { metricReadErrors: { add } as unknown as ObservabilityMetrics['metricReadErrors'], add } -} - -describe('registerRollingActiveUsersGauge', () => { - beforeEach(() => { - vi.useFakeTimers() - }) - afterEach(() => { - vi.useRealTimers() - vi.restoreAllMocks() - }) - - it('observes one point per window with the window attribute', async () => { - // Postgres count(*) comes back as a numeric string — assert we coerce it. - const { db } = makeDb([{ dau: '5', wau: '12', mau: '40' }]) - const { metricReadErrors } = makeReadErrors() - const { gauge, observe, run } = makeGauge() - - registerRollingActiveUsersGauge(gauge, db, metricReadErrors) - await run() - - expect(observe).toHaveBeenCalledTimes(3) - expect(observe).toHaveBeenNthCalledWith(1, 5, { window: '24h' }) - expect(observe).toHaveBeenNthCalledWith(2, 12, { window: '7d' }) - expect(observe).toHaveBeenNthCalledWith(3, 40, { window: '30d' }) - }) - - it('serves cached values without re-querying inside the 60s TTL', async () => { - const { db, select } = makeDb([{ dau: '3', wau: '7', mau: '9' }]) - const { metricReadErrors } = makeReadErrors() - const { gauge, observe, run } = makeGauge() - - registerRollingActiveUsersGauge(gauge, db, metricReadErrors) - await run() - await vi.advanceTimersByTimeAsync(30_000) - await run() - - // Second collection within TTL hits cache: still one DB query, but six - // observes (3 per cycle). - expect(select).toHaveBeenCalledTimes(1) - expect(observe).toHaveBeenCalledTimes(6) - }) - - it('re-queries after the TTL expires', async () => { - const { db, select } = makeDb([{ dau: '1', wau: '1', mau: '1' }]) - const { metricReadErrors } = makeReadErrors() - const { gauge, run } = makeGauge() - - registerRollingActiveUsersGauge(gauge, db, metricReadErrors) - await run() - await vi.advanceTimersByTimeAsync(61_000) - await run() - - expect(select).toHaveBeenCalledTimes(2) - }) - - it('on DB error: increments read-errors and does not observe', async () => { - // ROOT CAUSE: - // - // A silent gauge that observes a stale/zero value on DB failure masks the - // outage forever — an absence-based alert can never fire. - // - // We fixed this by skipping result.observe(...) on error and bumping - // airi.observability.read_errors{metric}, so Prometheus staleness exposes - // the broken DB instead. - const { db } = makeDb([], { reject: true }) - const { metricReadErrors, add } = makeReadErrors() - const { gauge, observe, run } = makeGauge() - - registerRollingActiveUsersGauge(gauge, db, metricReadErrors) - await run() - - expect(observe).not.toHaveBeenCalled() - expect(add).toHaveBeenCalledWith(1, { metric: 'user.active_rolling' }) - }) - - it('coalesces concurrent collection cycles into one DB query', async () => { - const { db, select } = makeDb([{ dau: '2', wau: '4', mau: '6' }]) - const { metricReadErrors } = makeReadErrors() - const { gauge, run } = makeGauge() - - registerRollingActiveUsersGauge(gauge, db, metricReadErrors) - // Fire two callbacks before the first refresh resolves — the in-flight - // lock must fold them into a single query. - await Promise.all([run(), run()]) - - expect(select).toHaveBeenCalledTimes(1) - }) -}) diff --git a/server/apps/api/src/otel/gauges/rolling-active-users.ts b/server/apps/api/src/otel/gauges/rolling-active-users.ts deleted file mode 100644 index 5a980f242..000000000 --- a/server/apps/api/src/otel/gauges/rolling-active-users.ts +++ /dev/null @@ -1,132 +0,0 @@ -import type { AuthMetrics, ObservabilityMetrics } from '..' -import type { Database } from '../../libs/db' - -import { useLogger } from '@guiiai/logg' -import { sql } from 'drizzle-orm' - -import { user as userTable } from '../../schemas/accounts' - -/** - * Trailing windows reported by the `user.active_rolling` gauge. The `label` - * becomes the Prometheus `window` attribute (`user_active_rolling{window="24h"}`) - * and `ms` is the lookback applied to `user.last_seen_at`. - */ -const WINDOWS = [ - { label: '24h', ms: 24 * 60 * 60 * 1000 }, - { label: '7d', ms: 7 * 24 * 60 * 60 * 1000 }, - { label: '30d', ms: 30 * 24 * 60 * 60 * 1000 }, -] as const - -type WindowLabel = (typeof WINDOWS)[number]['label'] -type RollingCounts = Record - -/** - * Wire the `user.active_rolling` ObservableGauge to a Postgres - * `COUNT(*) FILTER (WHERE last_seen_at > cutoff)` over the `user` table, one - * filter per trailing window (DAU / WAU / MAU). - * - * Use when: - * - Assembling DI in `createApp()`, exactly once per process. - * - * Why this exists alongside `registerDistinctActiveUsersGauge`: - * - `user.distinct_active` counts users with a currently non-expired session - * ("signed in right now"). It cannot answer "how many users came back this - * week" because expired sessions drop out. - * - `user.active_rolling` reads `user.last_seen_at` — touched on sign-in and - * on every OIDC access-token refresh (~hourly) — so it measures activity - * over a trailing window regardless of session state. This is the standard - * DAU / WAU / MAU engagement funnel. - * - * Expects: - * - `gauge` is the ObservableGauge handle created in `initOtel`. - * - `db` is the migrated Drizzle handle. - * - `metricReadErrors` is the shared counter; failures inside the callback - * increment it labelled with the originating metric name. - * - * Multi-replica note: - * - Cluster-wide gauge — every replica reads the same DB and reports the same - * value per window. Dashboards MUST aggregate with `max()`/`avg()`, NOT - * `sum()`. See observability-conventions.md. - * - * Concurrency: - * - Same in-flight promise lock as the sibling gauges, so concurrent OTel - * collection cycles fold into one DB query. Cached for 60s — DAU/WAU/MAU - * move slowly and the query scans the whole `user` table, so a longer TTL - * than the per-row session gauges keeps DB load low. - * - * Failure mode: - * - DB error → increment `airi.observability.read_errors{metric}` and skip - * `result.observe(...)`. Prometheus staleness exposes the outage instead of - * pinning stale values forever. - */ -export function registerRollingActiveUsersGauge( - gauge: AuthMetrics['rollingActiveUsers'], - db: Database, - metricReadErrors: ObservabilityMetrics['metricReadErrors'], -) { - const log = useLogger('rolling-active-users-gauge').useGlobalConfig() - const CACHE_TTL_MS = 60_000 - - let cachedAt = 0 - let cached: RollingCounts = { '24h': 0, '7d': 0, '30d': 0 } - let refreshInFlight: Promise | null = null - - async function refresh(): Promise { - try { - // Compute cutoffs from the app clock so all three windows are anchored - // to the same instant within one query. `id` is the PK, so - // `count(*)` == `count(distinct id)` — no DISTINCT needed. - const now = Date.now() - const since24h = new Date(now - WINDOWS[0].ms) - const since7d = new Date(now - WINDOWS[1].ms) - const since30d = new Date(now - WINDOWS[2].ms) - const rows = await db - .select({ - dau: sql`count(*) filter (where ${userTable.lastSeenAt} > ${since24h})`, - wau: sql`count(*) filter (where ${userTable.lastSeenAt} > ${since7d})`, - mau: sql`count(*) filter (where ${userTable.lastSeenAt} > ${since30d})`, - }) - .from(userTable) - cached = { - '24h': Number(rows[0]?.dau ?? 0), - '7d': Number(rows[0]?.wau ?? 0), - '30d': Number(rows[0]?.mau ?? 0), - } - cachedAt = Date.now() - return true - } - catch (err) { - log.withError(err).warn('Failed to read rolling active users for gauge') - metricReadErrors.add(1, { metric: 'user.active_rolling' }) - return false - } - } - - function observeAll(result: Parameters[0]>[0]) { - for (const w of WINDOWS) - result.observe(cached[w.label], { window: w.label }) - } - - gauge.addCallback(async (result) => { - const now = Date.now() - - // Cache fresh — serve last good values without touching the DB. - if (cachedAt !== 0 && now - cachedAt < CACHE_TTL_MS) { - observeAll(result) - return - } - - // Coalesce concurrent refreshes onto one in-flight promise. - if (!refreshInFlight) { - refreshInFlight = refresh().finally(() => { - refreshInFlight = null - }) - } - const ok = await refreshInFlight - - if (ok) - observeAll(result) - // else: deliberately skip observe — let Prometheus staleness expose the - // DB outage instead of masking it with stale numbers. - }) -} diff --git a/server/apps/api/src/otel/gauges/total-users.test.ts b/server/apps/api/src/otel/gauges/total-users.test.ts deleted file mode 100644 index 4f71677ed..000000000 --- a/server/apps/api/src/otel/gauges/total-users.test.ts +++ /dev/null @@ -1,106 +0,0 @@ -import type { AuthMetrics, ObservabilityMetrics } from '..' -import type { Database } from '../../libs/db' - -import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' - -import { registerTotalUsersGauge } from './total-users' - -/** - * Capture the callback registered via `gauge.addCallback` and a spyable - * `observe` so tests can drive collection cycles by hand. - * - * @example - * const { gauge, observe, run } = makeGauge() - * registerTotalUsersGauge(gauge, db, errs) - * await run() - * expect(observe).toHaveBeenCalledWith(42) - */ -function makeGauge() { - let cb: ((result: { observe: (v: number) => void }) => void | Promise) | null = null - const observe = vi.fn() - const gauge = { - addCallback: vi.fn((fn: typeof cb) => { cb = fn }), - } as unknown as AuthMetrics['totalUsers'] - return { - gauge, - observe, - run: async () => { - if (!cb) - throw new Error('no callback registered') - await cb({ observe }) - }, - } -} - -/** - * Drizzle's `db.select({...}).from(table)` is awaited directly (thenable - * query builder). Mock it as `select -> { from: () => Promise }`. - * - * @example - * const db = makeDb([{ count: '12' }]) - */ -function makeDb(rows: unknown[], opts: { reject?: boolean } = {}) { - const from = vi.fn(() => (opts.reject ? Promise.reject(new Error('db down')) : Promise.resolve(rows))) - const select = vi.fn(() => ({ from })) - return { db: { select } as unknown as Database, select, from } -} - -function makeReadErrors() { - const add = vi.fn() - return { metricReadErrors: { add } as unknown as ObservabilityMetrics['metricReadErrors'], add } -} - -describe('registerTotalUsersGauge', () => { - beforeEach(() => { - vi.useFakeTimers() - }) - afterEach(() => { - vi.useRealTimers() - vi.restoreAllMocks() - }) - - it('observes the total user count from the user table', async () => { - const { db } = makeDb([{ count: '42' }]) - const { metricReadErrors } = makeReadErrors() - const { gauge, observe, run } = makeGauge() - - registerTotalUsersGauge(gauge, db, metricReadErrors) - await run() - - expect(observe).toHaveBeenCalledWith(42) - }) - - it('serves cached values without re-querying inside the 60s TTL', async () => { - const { db, select } = makeDb([{ count: '3' }]) - const { metricReadErrors } = makeReadErrors() - const { gauge, observe, run } = makeGauge() - - registerTotalUsersGauge(gauge, db, metricReadErrors) - await run() - await vi.advanceTimersByTimeAsync(30_000) - await run() - - expect(select).toHaveBeenCalledTimes(1) - expect(observe).toHaveBeenCalledTimes(2) - }) - - it('on DB error: increments read-errors and does not observe', async () => { - // ROOT CAUSE: - // - // A total-user gauge that observes a stale/zero value on DB failure masks - // the outage forever — an absence-based alert can never fire. - // - // We fixed this by skipping result.observe(...) on error and bumping - // airi.observability.read_errors{metric}, so Prometheus staleness exposes - // the broken DB instead. - const { db } = makeDb([], { reject: true }) - const { metricReadErrors, add } = makeReadErrors() - const { gauge, observe, run } = makeGauge() - - registerTotalUsersGauge(gauge, db, metricReadErrors) - await run() - - expect(observe).not.toHaveBeenCalled() - expect(add).toHaveBeenCalledWith(1, { metric: 'user.total' }) - }) -}) diff --git a/server/apps/api/src/otel/gauges/total-users.ts b/server/apps/api/src/otel/gauges/total-users.ts deleted file mode 100644 index df558473d..000000000 --- a/server/apps/api/src/otel/gauges/total-users.ts +++ /dev/null @@ -1,81 +0,0 @@ -import type { AuthMetrics, ObservabilityMetrics } from '..' -import type { Database } from '../../libs/db' - -import { useLogger } from '@guiiai/logg' -import { count } from 'drizzle-orm' - -import { user as userTable } from '../../schemas/accounts' - -/** - * Wire the `user.total` ObservableGauge to a Postgres `COUNT(*)` over the - * Better Auth user table. - * - * Use when: - * - Assembling DI in `createApp()`, exactly once per process. - * - * Expects: - * - `gauge` is the ObservableGauge handle created in `initOtel`. - * - `db` is the migrated Drizzle handle. - * - `metricReadErrors` is the shared counter used to track failures inside - * metric callbacks. - * - * Multi-replica note: - * - Cluster-wide gauge — every replica reads the same DB and reports the same - * value. Dashboards MUST aggregate with `max()`/`avg()`, NOT `sum()`. - * - * Concurrency: - * - Same in-flight promise lock as the sibling DB-backed gauges, so concurrent - * OTel collection cycles fold into one DB query. - * - * Failure mode: - * - DB error → increment `airi.observability.read_errors{metric}` and skip - * `result.observe(...)`. Prometheus staleness exposes the outage instead of - * pinning stale values forever. - */ -export function registerTotalUsersGauge( - gauge: AuthMetrics['totalUsers'], - db: Database, - metricReadErrors: ObservabilityMetrics['metricReadErrors'], -) { - const log = useLogger('total-users-gauge').useGlobalConfig() - const CACHE_TTL_MS = 60_000 - - let cachedAt = 0 - let cachedCount = 0 - let refreshInFlight: Promise | null = null - - async function refresh(): Promise { - try { - const rows = await db - .select({ count: count() }) - .from(userTable) - cachedCount = Number(rows[0]?.count ?? 0) - cachedAt = Date.now() - return true - } - catch (err) { - log.withError(err).warn('Failed to read total users for gauge') - metricReadErrors.add(1, { metric: 'user.total' }) - return false - } - } - - gauge.addCallback(async (result) => { - const now = Date.now() - - if (cachedAt !== 0 && now - cachedAt < CACHE_TTL_MS) { - result.observe(cachedCount) - return - } - - if (!refreshInFlight) { - refreshInFlight = refresh().finally(() => { - refreshInFlight = null - }) - } - const ok = await refreshInFlight - - if (ok) - result.observe(cachedCount) - }) -} diff --git a/server/apps/api/src/otel/gauges/user-metrics-snapshot.test.ts b/server/apps/api/src/otel/gauges/user-metrics-snapshot.test.ts new file mode 100644 index 000000000..48b39a982 --- /dev/null +++ b/server/apps/api/src/otel/gauges/user-metrics-snapshot.test.ts @@ -0,0 +1,127 @@ +import type { ObservableCallback, ObservableResult } from '@opentelemetry/api' + +import { afterEach, describe, expect, it, vi } from 'vitest' + +import { registerUserMetricsSnapshotGauges, USER_METRICS_SNAPSHOT_MAX_AGE_MS } from './user-metrics-snapshot' + +afterEach(() => { + vi.restoreAllMocks() +}) + +function createGaugeProbe() { + let callback: ObservableCallback | undefined + const observe = vi.fn() + + return { + gauge: { + addCallback(value: ObservableCallback) { + callback = value + }, + }, + observe, + async collect() { + if (!callback) + throw new Error('Gauge callback was not registered') + + const result: ObservableResult = { observe } + await callback(result) + }, + } +} + +describe('registerUserMetricsSnapshotGauges', () => { + it('keeps periodic collection passive and only observes an explicitly recorded snapshot', async () => { + const refreshedAt = Date.parse('2026-08-04T00:00:00.000Z') + vi.spyOn(Date, 'now').mockReturnValue(refreshedAt) + const totalUsers = createGaugeProbe() + const activeSessions = createGaugeProbe() + const distinctActiveUsers = createGaugeProbe() + const rollingActiveUsers = createGaugeProbe() + + const recorder = registerUserMetricsSnapshotGauges({ + totalUsers: totalUsers.gauge, + activeSessions: activeSessions.gauge, + distinctActiveUsers: distinctActiveUsers.gauge, + rollingActiveUsers: rollingActiveUsers.gauge, + }) + + await Promise.all([ + totalUsers.collect(), + activeSessions.collect(), + distinctActiveUsers.collect(), + rollingActiveUsers.collect(), + ]) + + expect(totalUsers.observe).not.toHaveBeenCalled() + expect(activeSessions.observe).not.toHaveBeenCalled() + expect(distinctActiveUsers.observe).not.toHaveBeenCalled() + expect(rollingActiveUsers.observe).not.toHaveBeenCalled() + + recorder.record( + { + totalUsers: 42, + activeSessions: 7, + distinctActiveUsers: 5, + rollingActiveUsers: { + '24h': 9, + '7d': 18, + '30d': 30, + }, + }, + refreshedAt, + ) + + await Promise.all([ + totalUsers.collect(), + activeSessions.collect(), + distinctActiveUsers.collect(), + rollingActiveUsers.collect(), + ]) + + expect(totalUsers.observe).toHaveBeenLastCalledWith(42) + expect(activeSessions.observe).toHaveBeenLastCalledWith(7) + expect(distinctActiveUsers.observe).toHaveBeenLastCalledWith(5) + expect(rollingActiveUsers.observe).toHaveBeenNthCalledWith(1, 9, { window: '24h' }) + expect(rollingActiveUsers.observe).toHaveBeenNthCalledWith(2, 18, { window: '7d' }) + expect(rollingActiveUsers.observe).toHaveBeenNthCalledWith(3, 30, { window: '30d' }) + }) + + it('stops observing a snapshot after its bounded freshness interval', async () => { + const refreshedAt = Date.parse('2026-08-04T00:00:00.000Z') + vi.spyOn(Date, 'now').mockReturnValue(refreshedAt) + const totalUsers = createGaugeProbe() + const activeSessions = createGaugeProbe() + const distinctActiveUsers = createGaugeProbe() + const rollingActiveUsers = createGaugeProbe() + + const recorder = registerUserMetricsSnapshotGauges({ + totalUsers: totalUsers.gauge, + activeSessions: activeSessions.gauge, + distinctActiveUsers: distinctActiveUsers.gauge, + rollingActiveUsers: rollingActiveUsers.gauge, + }) + + recorder.record( + { + totalUsers: 42, + activeSessions: 7, + distinctActiveUsers: 5, + rollingActiveUsers: { '24h': 9, '7d': 18, '30d': 30 }, + }, + refreshedAt, + ) + + vi.mocked(Date.now).mockReturnValue(refreshedAt + USER_METRICS_SNAPSHOT_MAX_AGE_MS) + await Promise.all([ + totalUsers.collect(), + activeSessions.collect(), + distinctActiveUsers.collect(), + rollingActiveUsers.collect(), + ]) + + expect(totalUsers.observe).not.toHaveBeenCalled() + expect(activeSessions.observe).not.toHaveBeenCalled() + expect(distinctActiveUsers.observe).not.toHaveBeenCalled() + expect(rollingActiveUsers.observe).not.toHaveBeenCalled() + }) +}) diff --git a/server/apps/api/src/otel/gauges/user-metrics-snapshot.ts b/server/apps/api/src/otel/gauges/user-metrics-snapshot.ts new file mode 100644 index 000000000..6e6b50fe8 --- /dev/null +++ b/server/apps/api/src/otel/gauges/user-metrics-snapshot.ts @@ -0,0 +1,91 @@ +import type { ObservableGauge } from '@opentelemetry/api' + +const ROLLING_WINDOWS = ['24h', '7d', '30d'] as const + +// The admin endpoint caches DB aggregates for 60 seconds. Two minutes keeps a +// continuously refreshed dashboard stable across cache rollover and several +// 15-second OTel collections, while still bounding stale replica dominance. +export const USER_METRICS_SNAPSHOT_MAX_AGE_MS = 2 * 60_000 + +type RollingWindow = (typeof ROLLING_WINDOWS)[number] + +export interface UserMetricsSnapshot { + totalUsers: number + activeSessions: number + distinctActiveUsers: number + rollingActiveUsers: Record +} + +export interface UserMetricsSnapshotRecorder { + record: (snapshot: UserMetricsSnapshot, refreshedAt: number) => void +} + +type ObservableGaugeRegistration = Pick + +export interface UserMetricsSnapshotGauges { + totalUsers: ObservableGaugeRegistration + activeSessions: ObservableGaugeRegistration + distinctActiveUsers: ObservableGaugeRegistration + rollingActiveUsers: ObservableGaugeRegistration +} + +/** + * Export the latest explicitly refreshed user-metrics snapshot without doing + * I/O from OTel's periodic collection callbacks. Until a request records the + * first snapshot, or after the last database refresh becomes stale, the + * gauges intentionally emit no points. + */ +export function registerUserMetricsSnapshotGauges( + gauges: UserMetricsSnapshotGauges, +): UserMetricsSnapshotRecorder { + let latest: { snapshot: UserMetricsSnapshot, refreshedAt: number } | undefined + + function readFreshSnapshot(): UserMetricsSnapshot | undefined { + if (!latest) + return undefined + + if (Date.now() - latest.refreshedAt >= USER_METRICS_SNAPSHOT_MAX_AGE_MS) { + latest = undefined + return undefined + } + + return latest.snapshot + } + + gauges.totalUsers.addCallback((result) => { + const snapshot = readFreshSnapshot() + if (snapshot) + result.observe(snapshot.totalUsers) + }) + + gauges.activeSessions.addCallback((result) => { + const snapshot = readFreshSnapshot() + if (snapshot) + result.observe(snapshot.activeSessions) + }) + + gauges.distinctActiveUsers.addCallback((result) => { + const snapshot = readFreshSnapshot() + if (snapshot) + result.observe(snapshot.distinctActiveUsers) + }) + + gauges.rollingActiveUsers.addCallback((result) => { + const snapshot = readFreshSnapshot() + if (!snapshot) + return + + for (const window of ROLLING_WINDOWS) + result.observe(snapshot.rollingActiveUsers[window], { window }) + }) + + return { + record(snapshot, refreshedAt) { + latest = { snapshot, refreshedAt } + }, + } +} + +export function createDiscardingUserMetricsSnapshotRecorder(): UserMetricsSnapshotRecorder { + return { record() {} } +} diff --git a/server/apps/api/src/otel/index.ts b/server/apps/api/src/otel/index.ts index 4a2586661..d342363bd 100644 --- a/server/apps/api/src/otel/index.ts +++ b/server/apps/api/src/otel/index.ts @@ -76,39 +76,42 @@ export interface AuthMetrics { userRegistered: Counter userLogin: Counter /** - * Pull-based gauge for total registered users. + * Gauge for the latest requested total registered-user snapshot. * * Use when: * - Reporting current account-base size. Pair with * {@link AuthMetrics.userRegistered} for signup deltas over a time window. * * Expects: - * - Backed by `SELECT COUNT(*) FROM "user"`. Same cluster-wide truth as the - * other DB-backed gauges; dashboards MUST aggregate with `max()`/`avg()`, - * not `sum()`. + * - Refreshed when an authorized caller requests `/api/admin/metrics`. + * OTel collection only reads the in-process snapshot and never queries + * Postgres. The gauge stops emitting after the bounded freshness window + * so an old replica can become stale. Dashboards MUST aggregate with + * `max()`/`avg()`, not `sum()`. */ totalUsers: ObservableGauge /** - * Cluster-wide active session count, sourced from Postgres (Better Auth - * `session` table where `expires_at > NOW()`). + * Latest requested active-session count from the Better Auth `session` + * table where `expires_at` is in the future. * * Why ObservableGauge instead of UpDownCounter: * - UpDownCounter drifts: TTL expiration never fires a -1, and multi- * replica deploys split +1 / -1 across instances (signin on A, signout * on B). The previous implementation went unboundedly positive. - * - Reading from the source-of-truth DB at scrape time makes the metric - * self-correcting. + * - The admin metrics request refreshes this source-of-truth snapshot; + * periodic OTel collection does no I/O, so telemetry cannot keep a + * serverless database awake. * * Multi-replica note: - * - Every replica reads the same DB and reports the same value, so the + * - Only replicas with a fresh admin metrics snapshot emit a point. The * dashboard MUST aggregate with `max()` (or `avg()`), NOT `sum()`. - * Using sum() would multiply the real count by the replica count. * - See `server/apps/api/docs/ai-context/observability-conventions.md`, * "Multi-Replica Considerations". */ activeSessions: ObservableGauge /** - * Pull-based gauge for distinct users with ≥1 non-expired session. + * Gauge for the latest requested count of distinct users with at least one + * non-expired session. * * Use when: * - Querying real "active users" — not session rows. Better Auth creates a @@ -117,13 +120,13 @@ export interface AuthMetrics { * over time even when the actual user base is small. * * Expects: - * - Backed by `SELECT COUNT(DISTINCT user_id) FROM session WHERE expires_at > now()`. - * Same cluster-wide truth as `activeSessions`; dashboards MUST aggregate - * with `avg()`, not `sum()` — see observability-conventions.md. + * - Refreshed by `/api/admin/metrics` using `COUNT(DISTINCT user_id)`. + * Periodic collection reads memory only; dashboards MUST aggregate with + * `avg()`/`max()`, not `sum()`. */ distinctActiveUsers: ObservableGauge /** - * Pull-based gauge for rolling-window distinct active users (DAU / WAU / + * Gauge for the latest requested rolling-window active users (DAU / WAU / * MAU). * * Use when: @@ -133,13 +136,13 @@ export interface AuthMetrics { * currently-live session. * * Expects: - * - Backed by `COUNT(*) FILTER (WHERE last_seen_at > now() - window)` over - * the `user` table. `last_seen_at` is touched on sign-in and on every + * - Refreshed by `/api/admin/metrics` using `COUNT(*) FILTER` over the + * `user` table. `last_seen_at` is touched on sign-in and on every * OIDC access-token refresh (~hourly), so it is a per-user last-activity * timestamp (see the `user.lastSeenAt` schema note). * - Observed once per window with a `window` attribute (`24h` / `7d` / - * `30d`). Same cluster-wide truth as the other DB-backed gauges; - * dashboards MUST aggregate with `max()`/`avg()`, not `sum()`. + * `30d`). Periodic collection reads memory only; dashboards MUST + * aggregate with `max()`/`avg()`, not `sum()`. */ rollingActiveUsers: ObservableGauge } @@ -325,10 +328,10 @@ export interface RateLimitMetrics { export interface ObservabilityMetrics { /** - * Counts failures inside metric-pipeline callbacks (e.g. a DB-backed - * ObservableGauge that couldn't read from Postgres). Use for self-monitoring - * — when this is rising, treat the affected gauge's reported value as - * potentially stale. + * Counts failures inside metric-pipeline callbacks (for example, a Redis- + * backed ObservableGauge that could not refresh). Use for self-monitoring — + * when this is rising, treat the affected gauge's value as potentially + * stale. * * Labels: `metric` (the failing gauge's logical name). */ @@ -402,16 +405,16 @@ export function initOtel(env: Env): OtelInstance | null { description: 'Number of user sign-ins', }), totalUsers: meter.createObservableGauge(METRIC_USER_TOTAL, { - description: 'Total registered users sourced from Postgres (cluster-wide; dashboard must use max(), not sum())', + description: 'Fresh admin-requested total registered-user snapshot (memory-only and expires when stale; dashboard must use max(), not sum())', }), activeSessions: meter.createObservableGauge(METRIC_USER_ACTIVE_SESSIONS, { - description: 'Active user sessions sourced from Postgres (cluster-wide; dashboard must use avg(), not sum())', + description: 'Fresh admin-requested active-session snapshot (memory-only and expires when stale; dashboard must use max(), not sum())', }), distinctActiveUsers: meter.createObservableGauge(METRIC_USER_DISTINCT_ACTIVE, { - description: 'Distinct users with ≥1 non-expired session — true active-user count, immune to per-row session inflation (cluster-wide; dashboard must use avg(), not sum())', + description: 'Fresh admin-requested distinct active-user snapshot, immune to per-row session inflation and omitted when stale (dashboard must use max(), not sum())', }), rollingActiveUsers: meter.createObservableGauge(METRIC_USER_ACTIVE_ROLLING, { - description: 'Rolling-window distinct active users (DAU/WAU/MAU) from user.last_seen_at, labelled by window=24h|7d|30d (cluster-wide; dashboard must use max(), not sum())', + description: 'Fresh admin-requested DAU/WAU/MAU snapshot from user.last_seen_at, omitted when stale and labelled by window=24h|7d|30d (dashboard must use max(), not sum())', }), } diff --git a/server/apps/api/src/routes/admin/index.test.ts b/server/apps/api/src/routes/admin/index.test.ts index 8ea48849e..9ef3a0878 100644 --- a/server/apps/api/src/routes/admin/index.test.ts +++ b/server/apps/api/src/routes/admin/index.test.ts @@ -18,6 +18,7 @@ describe('admin metrics', () => { vi.spyOn(Date, 'now').mockImplementation(() => now) const db = await mockDB(schema) + const recordUserMetrics = vi.fn() await db.insert(schema.user).values({ id: 'admin-1', name: 'Admin', @@ -48,11 +49,24 @@ describe('admin metrics', () => { db, billingService: {} as never, configKV: {} as never, + userMetricsRecorder: { record: recordUserMetrics }, })) const firstResponse = await app.request('/api/admin/metrics') expect(firstResponse.status).toBe(200) - expect(await firstResponse.json()).toMatchObject({ totalUsers: 1, verifiedUsers: 1, adminSeats: 1 }) + expect(await firstResponse.json()).toMatchObject({ + totalUsers: 1, + verifiedUsers: 1, + adminSeats: 1, + activeSessions: 0, + distinctActiveUsers: 0, + rollingActiveUsers: { '24h': 0, '7d': 0, '30d': 0 }, + }) + expect(recordUserMetrics).toHaveBeenLastCalledWith(expect.objectContaining({ + totalUsers: 1, + activeSessions: 0, + distinctActiveUsers: 0, + }), now) await db.insert(schema.user).values({ id: 'user-2', @@ -61,13 +75,18 @@ describe('admin metrics', () => { emailVerified: false, }) + now += 30_000 const cachedResponse = await app.request('/api/admin/metrics') expect(cachedResponse.status).toBe(200) expect(await cachedResponse.json()).toMatchObject({ totalUsers: 1, verifiedUsers: 1, adminSeats: 1 }) + expect(recordUserMetrics).toHaveBeenCalledTimes(2) + expect(recordUserMetrics).toHaveBeenLastCalledWith(expect.anything(), Date.parse('2026-08-04T00:00:00.000Z')) - now += 60_001 + now += 30_001 const refreshedResponse = await app.request('/api/admin/metrics') expect(refreshedResponse.status).toBe(200) expect(await refreshedResponse.json()).toMatchObject({ totalUsers: 2, verifiedUsers: 1, adminSeats: 1 }) + expect(recordUserMetrics).toHaveBeenCalledTimes(3) + expect(recordUserMetrics).toHaveBeenLastCalledWith(expect.anything(), now) }) }) diff --git a/server/apps/api/src/routes/admin/index.ts b/server/apps/api/src/routes/admin/index.ts index 4d0b2b39a..d60ddd5cc 100644 --- a/server/apps/api/src/routes/admin/index.ts +++ b/server/apps/api/src/routes/admin/index.ts @@ -1,11 +1,12 @@ import type { Context } from 'hono' import type { Database } from '../../libs/db' +import type { UserMetricsSnapshot, UserMetricsSnapshotRecorder } from '../../otel/gauges/user-metrics-snapshot' import type { ConfigKVService } from '../../services/adapters/config-kv' import type { BillingService } from '../../services/domain/billing/billing-service' import type { HonoEnv } from '../../types/hono' -import { and, asc, count, desc, eq, gt, ilike, isNull, or, sql } from 'drizzle-orm' +import { and, asc, count, countDistinct, desc, eq, gt, ilike, isNull, or, sql } from 'drizzle-orm' import { Hono } from 'hono' import { integer, maxLength, maxValue, minValue, nonEmpty, number, object, optional, pipe, safeParse, string } from 'valibot' @@ -57,12 +58,11 @@ export interface AdminRoutesDeps { db: Database billingService: BillingService configKV: ConfigKVService + userMetricsRecorder: UserMetricsSnapshotRecorder } -interface AdminMetricsSnapshot { - totalUsers: number +interface AdminMetricsSnapshot extends UserMetricsSnapshot { verifiedUsers: number - activeSessions: number currentFlux: number issuedFlux: number llmRequests24h: number @@ -71,14 +71,19 @@ interface AdminMetricsSnapshot { grafanaEmbedUrl: null } -function createAdminMetricsReader(db: Database) { - let cached: { value: AdminMetricsSnapshot, expiresAt: number } | undefined - let inFlight: Promise | undefined +interface AdminMetricsRead { + value: AdminMetricsSnapshot + refreshedAt: number +} - return async function readAdminMetrics(): Promise { +function createAdminMetricsReader(db: Database) { + let cached: (AdminMetricsRead & { expiresAt: number }) | undefined + let inFlight: Promise | undefined + + return async function readAdminMetrics(): Promise { const now = Date.now() if (cached && cached.expiresAt > now) - return cached.value + return cached // A polling burst can arrive immediately after expiry. Share that refresh // within this API process so only one set of aggregate queries reaches DB. @@ -86,47 +91,71 @@ function createAdminMetricsReader(db: Database) { return inFlight inFlight = (async () => { - const yesterday = new Date(Date.now() - 24 * 60 * 60 * 1000) + const currentTime = Date.now() + const yesterday = new Date(currentTime - 24 * 60 * 60 * 1000) + const weekAgo = new Date(currentTime - 7 * 24 * 60 * 60 * 1000) + const monthAgo = new Date(currentTime - 30 * 24 * 60 * 60 * 1000) const [ - totalUsers, - verifiedUsers, - activeSessions, + users, + sessions, currentFlux, issuedFlux, - llmRequests24h, - llmFlux24h, - adminUsers, + llmUsage24h, ] = await Promise.all([ - db.select({ count: count() }).from(userTable), - db.select({ count: count() }).from(userTable).where(eq(userTable.emailVerified, true)), - db.select({ count: count() }).from(sessionTable).where(gt(sessionTable.expiresAt, new Date())), + db.select({ + totalUsers: count(), + verifiedUsers: sql`count(*) filter (where ${userTable.emailVerified} = true)`, + adminSeats: sql`count(*) filter (where 'admin' = any(regexp_split_to_array(coalesce(${userTable.role}, ''), '\\s*,\\s*')))`, + active24h: sql`count(*) filter (where ${userTable.lastSeenAt} > ${yesterday})`, + active7d: sql`count(*) filter (where ${userTable.lastSeenAt} > ${weekAgo})`, + active30d: sql`count(*) filter (where ${userTable.lastSeenAt} > ${monthAgo})`, + }).from(userTable), + db + .select({ + activeSessions: count(), + distinctActiveUsers: countDistinct(sessionTable.userId), + }) + .from(sessionTable) + .where(gt(sessionTable.expiresAt, new Date(currentTime))), db.select({ total: sql`coalesce(sum(${userFlux.flux}), 0)::int` }).from(userFlux).where(isNull(userFlux.deletedAt)), db.select({ total: sql`coalesce(sum(${fluxTransaction.amount}) filter (where ${fluxTransaction.type} in ('credit', 'initial', 'promo')), 0)::int` }).from(fluxTransaction), - db.select({ count: count() }).from(llmRequestLog).where(gt(llmRequestLog.createdAt, yesterday)), - db.select({ total: sql`coalesce(sum(${llmRequestLog.fluxConsumed}), 0)::int` }).from(llmRequestLog).where(gt(llmRequestLog.createdAt, yesterday)), db - .select({ count: count() }) - .from(userTable) - .where(sql`'admin' = any(regexp_split_to_array(coalesce(${userTable.role}, ''), '\\s*,\\s*'))`), + .select({ + count: count(), + total: sql`coalesce(sum(${llmRequestLog.fluxConsumed}), 0)::int`, + }) + .from(llmRequestLog) + .where(gt(llmRequestLog.createdAt, yesterday)), ]) const value: AdminMetricsSnapshot = { - totalUsers: Number(totalUsers[0]?.count ?? 0), - verifiedUsers: Number(verifiedUsers[0]?.count ?? 0), - activeSessions: Number(activeSessions[0]?.count ?? 0), + totalUsers: Number(users[0]?.totalUsers ?? 0), + verifiedUsers: Number(users[0]?.verifiedUsers ?? 0), + activeSessions: Number(sessions[0]?.activeSessions ?? 0), + distinctActiveUsers: Number(sessions[0]?.distinctActiveUsers ?? 0), + rollingActiveUsers: { + '24h': Number(users[0]?.active24h ?? 0), + '7d': Number(users[0]?.active7d ?? 0), + '30d': Number(users[0]?.active30d ?? 0), + }, currentFlux: Number(currentFlux[0]?.total ?? 0), issuedFlux: Number(issuedFlux[0]?.total ?? 0), - llmRequests24h: Number(llmRequests24h[0]?.count ?? 0), - llmFlux24h: Number(llmFlux24h[0]?.total ?? 0), - adminSeats: Number(adminUsers[0]?.count ?? 0), + llmRequests24h: Number(llmUsage24h[0]?.count ?? 0), + llmFlux24h: Number(llmUsage24h[0]?.total ?? 0), + adminSeats: Number(users[0]?.adminSeats ?? 0), grafanaEmbedUrl: null, } // Expiry starts after the refresh finishes; slow aggregate queries should // not shorten the period during which the completed snapshot is reused. - cached = { value, expiresAt: Date.now() + ADMIN_METRICS_CACHE_TTL_MS } - return value + const refreshedAt = Date.now() + cached = { + value, + refreshedAt, + expiresAt: refreshedAt + ADMIN_METRICS_CACHE_TTL_MS, + } + return cached })() try { @@ -244,7 +273,9 @@ export function createAdminRoutes(deps: AdminRoutesDeps) { }) .get('/metrics', async (c) => { - return c.json(await readAdminMetrics()) + const snapshot = await readAdminMetrics() + deps.userMetricsRecorder.record(snapshot.value, snapshot.refreshedAt) + return c.json(snapshot.value) }) .get('/users', async (c) => {