perf(server): add query indexes and cache admin metrics (#2213)
Signed-off-by: RainbowBird <git@luoling.moe>
This commit is contained in:
@@ -0,0 +1,8 @@
|
|||||||
|
CREATE INDEX "oauth_refresh_token_token_idx" ON "oauth_refresh_token" USING btree ("token");--> statement-breakpoint
|
||||||
|
CREATE INDEX "oauth_refresh_token_user_id_idx" ON "oauth_refresh_token" USING btree ("user_id");--> statement-breakpoint
|
||||||
|
CREATE INDEX "oauth_refresh_token_session_id_idx" ON "oauth_refresh_token" USING btree ("session_id");--> statement-breakpoint
|
||||||
|
CREATE INDEX "oauth_refresh_token_client_id_idx" ON "oauth_refresh_token" USING btree ("client_id");--> statement-breakpoint
|
||||||
|
CREATE INDEX "session_expires_at_idx" ON "session" USING btree ("expires_at");--> statement-breakpoint
|
||||||
|
CREATE INDEX "chat_members_user_id_member_type_chat_id_idx" ON "chat_members" USING btree ("user_id","member_type","chat_id");--> statement-breakpoint
|
||||||
|
CREATE INDEX "chat_members_chat_id_member_type_user_id_idx" ON "chat_members" USING btree ("chat_id","member_type","user_id");--> statement-breakpoint
|
||||||
|
CREATE INDEX "messages_chat_id_seq_active_idx" ON "messages" USING btree ("chat_id","seq") WHERE "messages"."deleted_at" IS NULL;
|
||||||
File diff suppressed because it is too large
Load Diff
@@ -127,6 +127,13 @@
|
|||||||
"when": 1782912836523,
|
"when": 1782912836523,
|
||||||
"tag": "0017_nappy_dagger",
|
"tag": "0017_nappy_dagger",
|
||||||
"breakpoints": true
|
"breakpoints": true
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"idx": 18,
|
||||||
|
"version": "7",
|
||||||
|
"when": 1785833974268,
|
||||||
|
"tag": "0018_blushing_scarlet_witch",
|
||||||
|
"breakpoints": true
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
}
|
}
|
||||||
@@ -0,0 +1,73 @@
|
|||||||
|
import type { HonoEnv } from '../../types/hono'
|
||||||
|
|
||||||
|
import { Hono } from 'hono'
|
||||||
|
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||||
|
|
||||||
|
import { createAdminRoutes } from '.'
|
||||||
|
import { mockDB } from '../../libs/mock-db'
|
||||||
|
|
||||||
|
import * as schema from '../../schemas'
|
||||||
|
|
||||||
|
afterEach(() => {
|
||||||
|
vi.restoreAllMocks()
|
||||||
|
})
|
||||||
|
|
||||||
|
describe('admin metrics', () => {
|
||||||
|
it('reuses a snapshot for 60 seconds before reading fresh metrics', async () => {
|
||||||
|
let now = Date.parse('2026-08-04T00:00:00.000Z')
|
||||||
|
vi.spyOn(Date, 'now').mockImplementation(() => now)
|
||||||
|
|
||||||
|
const db = await mockDB(schema)
|
||||||
|
await db.insert(schema.user).values({
|
||||||
|
id: 'admin-1',
|
||||||
|
name: 'Admin',
|
||||||
|
email: 'admin@example.com',
|
||||||
|
emailVerified: true,
|
||||||
|
role: 'admin',
|
||||||
|
})
|
||||||
|
|
||||||
|
const app = new Hono<HonoEnv>()
|
||||||
|
.use('*', async (c, next) => {
|
||||||
|
c.set('user', {
|
||||||
|
id: 'admin-1',
|
||||||
|
name: 'Admin',
|
||||||
|
email: 'admin@example.com',
|
||||||
|
emailVerified: true,
|
||||||
|
image: null,
|
||||||
|
role: 'admin',
|
||||||
|
banned: false,
|
||||||
|
banReason: null,
|
||||||
|
banExpires: null,
|
||||||
|
createdAt: new Date(),
|
||||||
|
updatedAt: new Date(),
|
||||||
|
})
|
||||||
|
c.set('session', null)
|
||||||
|
await next()
|
||||||
|
})
|
||||||
|
.route('/api/admin', createAdminRoutes({
|
||||||
|
db,
|
||||||
|
billingService: {} as never,
|
||||||
|
configKV: {} as never,
|
||||||
|
}))
|
||||||
|
|
||||||
|
const firstResponse = await app.request('/api/admin/metrics')
|
||||||
|
expect(firstResponse.status).toBe(200)
|
||||||
|
expect(await firstResponse.json()).toMatchObject({ totalUsers: 1, verifiedUsers: 1, adminSeats: 1 })
|
||||||
|
|
||||||
|
await db.insert(schema.user).values({
|
||||||
|
id: 'user-2',
|
||||||
|
name: 'User',
|
||||||
|
email: 'user@example.com',
|
||||||
|
emailVerified: false,
|
||||||
|
})
|
||||||
|
|
||||||
|
const cachedResponse = await app.request('/api/admin/metrics')
|
||||||
|
expect(cachedResponse.status).toBe(200)
|
||||||
|
expect(await cachedResponse.json()).toMatchObject({ totalUsers: 1, verifiedUsers: 1, adminSeats: 1 })
|
||||||
|
|
||||||
|
now += 60_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 })
|
||||||
|
})
|
||||||
|
})
|
||||||
@@ -19,6 +19,7 @@ import { createBadRequestError, createNotFoundError } from '../../utils/error'
|
|||||||
import { createQueryIntegerSchema } from '../../utils/http-query'
|
import { createQueryIntegerSchema } from '../../utils/http-query'
|
||||||
|
|
||||||
const MAX_FLUX_ADJUSTMENT = 1_000_000_000
|
const MAX_FLUX_ADJUSTMENT = 1_000_000_000
|
||||||
|
const ADMIN_METRICS_CACHE_TTL_MS = 60_000
|
||||||
|
|
||||||
const ListUsersQuerySchema = object({
|
const ListUsersQuerySchema = object({
|
||||||
limit: createQueryIntegerSchema({
|
limit: createQueryIntegerSchema({
|
||||||
@@ -58,6 +59,86 @@ export interface AdminRoutesDeps {
|
|||||||
configKV: ConfigKVService
|
configKV: ConfigKVService
|
||||||
}
|
}
|
||||||
|
|
||||||
|
interface AdminMetricsSnapshot {
|
||||||
|
totalUsers: number
|
||||||
|
verifiedUsers: number
|
||||||
|
activeSessions: number
|
||||||
|
currentFlux: number
|
||||||
|
issuedFlux: number
|
||||||
|
llmRequests24h: number
|
||||||
|
llmFlux24h: number
|
||||||
|
adminSeats: number
|
||||||
|
grafanaEmbedUrl: null
|
||||||
|
}
|
||||||
|
|
||||||
|
function createAdminMetricsReader(db: Database) {
|
||||||
|
let cached: { value: AdminMetricsSnapshot, expiresAt: number } | undefined
|
||||||
|
let inFlight: Promise<AdminMetricsSnapshot> | undefined
|
||||||
|
|
||||||
|
return async function readAdminMetrics(): Promise<AdminMetricsSnapshot> {
|
||||||
|
const now = Date.now()
|
||||||
|
if (cached && cached.expiresAt > now)
|
||||||
|
return cached.value
|
||||||
|
|
||||||
|
// A polling burst can arrive immediately after expiry. Share that refresh
|
||||||
|
// within this API process so only one set of aggregate queries reaches DB.
|
||||||
|
if (inFlight)
|
||||||
|
return inFlight
|
||||||
|
|
||||||
|
inFlight = (async () => {
|
||||||
|
const yesterday = new Date(Date.now() - 24 * 60 * 60 * 1000)
|
||||||
|
|
||||||
|
const [
|
||||||
|
totalUsers,
|
||||||
|
verifiedUsers,
|
||||||
|
activeSessions,
|
||||||
|
currentFlux,
|
||||||
|
issuedFlux,
|
||||||
|
llmRequests24h,
|
||||||
|
llmFlux24h,
|
||||||
|
adminUsers,
|
||||||
|
] = 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({ total: sql<number>`coalesce(sum(${userFlux.flux}), 0)::int` }).from(userFlux).where(isNull(userFlux.deletedAt)),
|
||||||
|
db.select({ total: sql<number>`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<number>`coalesce(sum(${llmRequestLog.fluxConsumed}), 0)::int` }).from(llmRequestLog).where(gt(llmRequestLog.createdAt, yesterday)),
|
||||||
|
db
|
||||||
|
.select({ count: count() })
|
||||||
|
.from(userTable)
|
||||||
|
.where(sql<boolean>`'admin' = any(regexp_split_to_array(coalesce(${userTable.role}, ''), '\\s*,\\s*'))`),
|
||||||
|
])
|
||||||
|
|
||||||
|
const value: AdminMetricsSnapshot = {
|
||||||
|
totalUsers: Number(totalUsers[0]?.count ?? 0),
|
||||||
|
verifiedUsers: Number(verifiedUsers[0]?.count ?? 0),
|
||||||
|
activeSessions: Number(activeSessions[0]?.count ?? 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),
|
||||||
|
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
|
||||||
|
})()
|
||||||
|
|
||||||
|
try {
|
||||||
|
return await inFlight
|
||||||
|
}
|
||||||
|
finally {
|
||||||
|
// Failed refreshes are intentionally not cached, so the next poll retries.
|
||||||
|
inFlight = undefined
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
function serializeUser(row: {
|
function serializeUser(row: {
|
||||||
id: string
|
id: string
|
||||||
name: string
|
name: string
|
||||||
@@ -142,6 +223,8 @@ async function ensureUserExists(db: Database, userId: string) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
export function createAdminRoutes(deps: AdminRoutesDeps) {
|
export function createAdminRoutes(deps: AdminRoutesDeps) {
|
||||||
|
const readAdminMetrics = createAdminMetricsReader(deps.db)
|
||||||
|
|
||||||
return new Hono<HonoEnv>()
|
return new Hono<HonoEnv>()
|
||||||
.use('*', authGuard)
|
.use('*', authGuard)
|
||||||
.use('*', adminGuard)
|
.use('*', adminGuard)
|
||||||
@@ -161,42 +244,7 @@ export function createAdminRoutes(deps: AdminRoutesDeps) {
|
|||||||
})
|
})
|
||||||
|
|
||||||
.get('/metrics', async (c) => {
|
.get('/metrics', async (c) => {
|
||||||
const yesterday = new Date(Date.now() - 24 * 60 * 60 * 1000)
|
return c.json(await readAdminMetrics())
|
||||||
|
|
||||||
const [
|
|
||||||
totalUsers,
|
|
||||||
verifiedUsers,
|
|
||||||
activeSessions,
|
|
||||||
currentFlux,
|
|
||||||
issuedFlux,
|
|
||||||
llmRequests24h,
|
|
||||||
llmFlux24h,
|
|
||||||
adminUsers,
|
|
||||||
] = await Promise.all([
|
|
||||||
deps.db.select({ count: count() }).from(userTable),
|
|
||||||
deps.db.select({ count: count() }).from(userTable).where(eq(userTable.emailVerified, true)),
|
|
||||||
deps.db.select({ count: count() }).from(sessionTable).where(gt(sessionTable.expiresAt, new Date())),
|
|
||||||
deps.db.select({ total: sql<number>`coalesce(sum(${userFlux.flux}), 0)::int` }).from(userFlux).where(isNull(userFlux.deletedAt)),
|
|
||||||
deps.db.select({ total: sql<number>`coalesce(sum(${fluxTransaction.amount}) filter (where ${fluxTransaction.type} in ('credit', 'initial', 'promo')), 0)::int` }).from(fluxTransaction),
|
|
||||||
deps.db.select({ count: count() }).from(llmRequestLog).where(gt(llmRequestLog.createdAt, yesterday)),
|
|
||||||
deps.db.select({ total: sql<number>`coalesce(sum(${llmRequestLog.fluxConsumed}), 0)::int` }).from(llmRequestLog).where(gt(llmRequestLog.createdAt, yesterday)),
|
|
||||||
deps.db
|
|
||||||
.select({ count: count() })
|
|
||||||
.from(userTable)
|
|
||||||
.where(sql<boolean>`'admin' = any(regexp_split_to_array(coalesce(${userTable.role}, ''), '\\s*,\\s*'))`),
|
|
||||||
])
|
|
||||||
|
|
||||||
return c.json({
|
|
||||||
totalUsers: Number(totalUsers[0]?.count ?? 0),
|
|
||||||
verifiedUsers: Number(verifiedUsers[0]?.count ?? 0),
|
|
||||||
activeSessions: Number(activeSessions[0]?.count ?? 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),
|
|
||||||
grafanaEmbedUrl: null,
|
|
||||||
})
|
|
||||||
})
|
})
|
||||||
|
|
||||||
.get('/users', async (c) => {
|
.get('/users', async (c) => {
|
||||||
|
|||||||
@@ -53,7 +53,10 @@ export const session = pgTable(
|
|||||||
// disabledPaths, but the column stays so the schema matches the plugin.
|
// disabledPaths, but the column stays so the schema matches the plugin.
|
||||||
impersonatedBy: text('impersonated_by'),
|
impersonatedBy: text('impersonated_by'),
|
||||||
},
|
},
|
||||||
table => [index('session_userId_idx').on(table.userId)],
|
table => [
|
||||||
|
index('session_userId_idx').on(table.userId),
|
||||||
|
index('session_expires_at_idx').on(table.expiresAt),
|
||||||
|
],
|
||||||
)
|
)
|
||||||
|
|
||||||
export const account = pgTable(
|
export const account = pgTable(
|
||||||
@@ -137,25 +140,34 @@ export const oauthClient = pgTable('oauth_client', {
|
|||||||
metadata: jsonb('metadata'),
|
metadata: jsonb('metadata'),
|
||||||
})
|
})
|
||||||
|
|
||||||
export const oauthRefreshToken = pgTable('oauth_refresh_token', {
|
export const oauthRefreshToken = pgTable(
|
||||||
id: text('id').primaryKey(),
|
'oauth_refresh_token',
|
||||||
token: text('token').notNull(),
|
{
|
||||||
clientId: text('client_id')
|
id: text('id').primaryKey(),
|
||||||
.notNull()
|
token: text('token').notNull(),
|
||||||
.references(() => oauthClient.clientId, { onDelete: 'cascade' }),
|
clientId: text('client_id')
|
||||||
sessionId: text('session_id').references(() => session.id, {
|
.notNull()
|
||||||
onDelete: 'set null',
|
.references(() => oauthClient.clientId, { onDelete: 'cascade' }),
|
||||||
}),
|
sessionId: text('session_id').references(() => session.id, {
|
||||||
userId: text('user_id')
|
onDelete: 'set null',
|
||||||
.notNull()
|
}),
|
||||||
.references(() => user.id, { onDelete: 'cascade' }),
|
userId: text('user_id')
|
||||||
referenceId: text('reference_id'),
|
.notNull()
|
||||||
expiresAt: timestamp('expires_at'),
|
.references(() => user.id, { onDelete: 'cascade' }),
|
||||||
createdAt: timestamp('created_at'),
|
referenceId: text('reference_id'),
|
||||||
revoked: timestamp('revoked'),
|
expiresAt: timestamp('expires_at'),
|
||||||
authTime: timestamp('auth_time'),
|
createdAt: timestamp('created_at'),
|
||||||
scopes: text('scopes').array().notNull(),
|
revoked: timestamp('revoked'),
|
||||||
})
|
authTime: timestamp('auth_time'),
|
||||||
|
scopes: text('scopes').array().notNull(),
|
||||||
|
},
|
||||||
|
table => [
|
||||||
|
index('oauth_refresh_token_token_idx').on(table.token),
|
||||||
|
index('oauth_refresh_token_user_id_idx').on(table.userId),
|
||||||
|
index('oauth_refresh_token_session_id_idx').on(table.sessionId),
|
||||||
|
index('oauth_refresh_token_client_id_idx').on(table.clientId),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
|
||||||
export const oauthAccessToken = pgTable('oauth_access_token', {
|
export const oauthAccessToken = pgTable('oauth_access_token', {
|
||||||
id: text('id').primaryKey(),
|
id: text('id').primaryKey(),
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
import type { InferInsertModel, InferSelectModel } from 'drizzle-orm'
|
import type { InferInsertModel, InferSelectModel } from 'drizzle-orm'
|
||||||
|
|
||||||
import { integer, pgTable, text, timestamp } from 'drizzle-orm/pg-core'
|
import { sql } from 'drizzle-orm'
|
||||||
|
import { index, integer, pgTable, text, timestamp } from 'drizzle-orm/pg-core'
|
||||||
|
|
||||||
import { nanoid } from '../utils/id'
|
import { nanoid } from '../utils/id'
|
||||||
|
|
||||||
@@ -66,6 +67,10 @@ export const chatMembers = pgTable(
|
|||||||
userId: text('user_id'),
|
userId: text('user_id'),
|
||||||
characterId: text('character_id'),
|
characterId: text('character_id'),
|
||||||
},
|
},
|
||||||
|
table => [
|
||||||
|
index('chat_members_user_id_member_type_chat_id_idx').on(table.userId, table.memberType, table.chatId),
|
||||||
|
index('chat_members_chat_id_member_type_user_id_idx').on(table.chatId, table.memberType, table.userId),
|
||||||
|
],
|
||||||
)
|
)
|
||||||
|
|
||||||
export const messages = pgTable(
|
export const messages = pgTable(
|
||||||
@@ -89,6 +94,11 @@ export const messages = pgTable(
|
|||||||
updatedAt: timestamp('updated_at').defaultNow().notNull(),
|
updatedAt: timestamp('updated_at').defaultNow().notNull(),
|
||||||
deletedAt: timestamp('deleted_at'),
|
deletedAt: timestamp('deleted_at'),
|
||||||
},
|
},
|
||||||
|
table => [
|
||||||
|
index('messages_chat_id_seq_active_idx')
|
||||||
|
.on(table.chatId, table.seq)
|
||||||
|
.where(sql`${table.deletedAt} IS NULL`),
|
||||||
|
],
|
||||||
)
|
)
|
||||||
|
|
||||||
export type Message = InferSelectModel<typeof messages>
|
export type Message = InferSelectModel<typeof messages>
|
||||||
|
|||||||
Reference in New Issue
Block a user