From 6f6fe01b3eb001c60a371091057438d1c014d669 Mon Sep 17 00:00:00 2001 From: RainbowBird Date: Wed, 3 Jun 2026 00:19:34 +0800 Subject: [PATCH] refactor(server): openai route gateway Co-authored-by: Neko --- apps/server/src/app.ts | 12 +- apps/server/src/routes/openai/v1/gateway.ts | 218 ++++++++++++++++++ apps/server/src/routes/openai/v1/guards.ts | 19 -- .../routes/openai/v1/{ => http}/response.ts | 0 apps/server/src/routes/openai/v1/index.ts | 127 +++++----- .../openai/v1/{ => middlewares}/billing.ts | 18 +- .../src/routes/openai/v1/middlewares/index.ts | 3 + .../openai/v1/{ => middlewares}/telemetry.ts | 10 +- .../openai/v1/middlewares/traffic-control.ts | 78 +++++++ .../chat-completions/index.ts} | 63 ++--- .../speech-catalog/index.ts} | 53 +++-- .../v1/operations/speech-generation/index.ts | 25 ++ .../server/src/routes/openai/v1/route.test.ts | 73 ++++-- apps/server/src/routes/openai/v1/speech.ts | 30 --- 14 files changed, 533 insertions(+), 196 deletions(-) create mode 100644 apps/server/src/routes/openai/v1/gateway.ts delete mode 100644 apps/server/src/routes/openai/v1/guards.ts rename apps/server/src/routes/openai/v1/{ => http}/response.ts (100%) rename apps/server/src/routes/openai/v1/{ => middlewares}/billing.ts (88%) create mode 100644 apps/server/src/routes/openai/v1/middlewares/index.ts rename apps/server/src/routes/openai/v1/{ => middlewares}/telemetry.ts (94%) create mode 100644 apps/server/src/routes/openai/v1/middlewares/traffic-control.ts rename apps/server/src/routes/openai/v1/{chat.ts => operations/chat-completions/index.ts} (91%) rename apps/server/src/routes/openai/v1/{catalog.ts => operations/speech-catalog/index.ts} (82%) create mode 100644 apps/server/src/routes/openai/v1/operations/speech-generation/index.ts delete mode 100644 apps/server/src/routes/openai/v1/speech.ts diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index 0589b1b8f..3607aa948 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -216,7 +216,17 @@ export async function buildApp(deps: AppDeps) { // Built once so the OpenAI-compat and audio routers share the same closure // (helpers like recordMetrics / recordRequestLog cross both surfaces) but // mount at different prefixes — see the `.route` calls below. - const v1Routes = createV1Routes(deps.fluxService, deps.billingService, deps.configKV, deps.requestLogService, deps.ttsMeter, deps.llmRouter, deps.otel?.genAi, deps.otel?.revenue, deps.otel?.rateLimit) + const v1Routes = createV1Routes({ + fluxService: deps.fluxService, + billingService: deps.billingService, + configKV: deps.configKV, + requestLogService: deps.requestLogService, + ttsMeter: deps.ttsMeter, + llmRouter: deps.llmRouter, + genAi: deps.otel?.genAi, + revenue: deps.otel?.revenue, + rateLimitMetrics: deps.otel?.rateLimit, + }) const builtApp = app .use('*', sessionMiddleware(deps.auth, deps.env)) diff --git a/apps/server/src/routes/openai/v1/gateway.ts b/apps/server/src/routes/openai/v1/gateway.ts new file mode 100644 index 000000000..8dd729e9a --- /dev/null +++ b/apps/server/src/routes/openai/v1/gateway.ts @@ -0,0 +1,218 @@ +import type { Context, Handler, MiddlewareHandler } from 'hono' + +import type { HonoEnv } from '../../../types/hono' +import type { ChatCompletionsOperationRequest } from './operations/chat-completions' +import type { SpeechGenerationOperationRequest } from './operations/speech-generation' +import type { V1RouteDeps } from './types' + +import { Hono } from 'hono' + +export type GatewayCallback = ( + context: V1GatewayContext, +) => Promise + +export type GatewayMiddleware = ( + context: V1GatewayContext, + next: () => Promise, +) => Promise + +export type V1HttpSurface = 'audio' | 'openai' + +export interface V1GatewayOperationInput { + 'chat.completions': ChatCompletionsOperationRequest + 'speech.generate': SpeechGenerationOperationRequest +} + +export type V1GatewayOperationName = keyof V1GatewayOperationInput + +export type V1GatewayPlugin = (gateway: V1GatewayRuntime) => void + +export interface V1GatewayContext { + deps: V1RouteDeps + hono: Context + input: V1GatewayOperationInput[Name] +} + +export interface V1GatewayRuntime { + deps: V1RouteDeps + handler: ( + name: Name, + parse: (hono: Context) => V1GatewayOperationInput[Name] | Promise, + callback: GatewayCallback, + ) => Handler + route: (surface: V1HttpSurface) => V1GatewayRoute + use: { + (plugin: V1GatewayPlugin): V1GatewayRuntime + (name: Name, middleware: GatewayMiddleware): V1GatewayRuntime + } + useHono: (surface: V1HttpSurface | '*', path: string, middleware: MiddlewareHandler) => V1GatewayRuntime +} + +export interface V1GatewayRoute { + deps: V1RouteDeps + get: (path: string, handler: Handler | V1GatewayRouteHandler) => V1GatewayRoute + handler: V1GatewayRuntime['handler'] + post: (path: string, handler: Handler | V1GatewayRouteHandler) => V1GatewayRoute + route: Hono + use: (name: Name, middleware: GatewayMiddleware) => V1GatewayRoute + useHono: (path: string, middleware: MiddlewareHandler) => V1GatewayRoute +} + +const routeHandlerMarker = Symbol('v1-gateway-route-handler') + +export interface V1GatewayRouteHandler { + (scope: Pick): Handler + [routeHandlerMarker]: true +} + +export function routeHandler(handler: (scope: Pick) => Handler): V1GatewayRouteHandler { + return Object.assign(handler, { [routeHandlerMarker]: true as const }) +} + +interface RegisteredHttpMiddleware { + middleware: MiddlewareHandler + path: string + surface: V1HttpSurface | '*' +} + +type OperationMiddlewares = { + [Name in V1GatewayOperationName]: GatewayMiddleware[] +} + +function cloneOperationMiddlewares(input: OperationMiddlewares): OperationMiddlewares { + return { + 'chat.completions': [...input['chat.completions']], + 'speech.generate': [...input['speech.generate']], + } +} + +/** + * Runs an OpenAI gateway callback through operation-scoped middleware. + * + * Use when: + * - The middleware needs parsed gateway input such as user id, model, body, + * session id, or abort signal. + * - The behavior is not a generic HTTP concern and should not receive Hono + * `Context`. + * + * Expects: + * - `callback` is the concrete gateway business callback. + * - `middlewares` are ordered from outermost to innermost. + * + * Returns: + * - A response produced by the gateway callback chain. + */ +export function runGatewayMiddlewares( + context: V1GatewayContext, + callback: GatewayCallback, + middlewares: GatewayMiddleware[], +): Promise { + const runnable = middlewares.reduceRight>( + (next, middleware) => ctx => middleware(ctx, () => next(ctx)), + callback, + ) + return runnable(context) +} + +export function createV1Gateway(deps: V1RouteDeps): V1GatewayRuntime { + const httpMiddlewares: RegisteredHttpMiddleware[] = [] + const operationMiddlewares: OperationMiddlewares = { + 'chat.completions': [], + 'speech.generate': [], + } + + let gateway: V1GatewayRuntime + + function use(plugin: V1GatewayPlugin): V1GatewayRuntime + function use(name: Name, middleware: GatewayMiddleware): V1GatewayRuntime + function use( + arg1: V1GatewayPlugin | Name, + arg2?: GatewayMiddleware, + ): V1GatewayRuntime { + if (typeof arg1 === 'function') { + arg1(gateway) + } + else if (arg2) { + operationMiddlewares[arg1].push(arg2) + } + return gateway + } + + function handlerWithMiddlewares( + middlewares: OperationMiddlewares, + name: Name, + parse: (hono: Context) => V1GatewayOperationInput[Name] | Promise, + callback: GatewayCallback, + ): Handler { + return async (hono) => { + const input = await parse(hono) + return runGatewayMiddlewares( + { deps, hono, input }, + callback, + middlewares[name], + ) + } + } + + function createRoute(surface: V1HttpSurface): V1GatewayRoute { + const route = new Hono() + + for (const registered of httpMiddlewares) { + if (registered.surface === '*' || registered.surface === surface) + route.use(registered.path, registered.middleware) + } + + function makeBuilder(scopedOperationMiddlewares: OperationMiddlewares): V1GatewayRoute { + let builder: V1GatewayRoute + + function register(method: 'get' | 'post', path: string, handler: Handler | V1GatewayRouteHandler): V1GatewayRoute { + route[method](path, resolveRouteHandler(builder, handler)) + return builder + } + + builder = { + deps, + get: (path, handler) => register('get', path, handler), + handler(name, parse, callback) { + return handlerWithMiddlewares(scopedOperationMiddlewares, name, parse, callback) + }, + post: (path, handler) => register('post', path, handler), + route, + use(name, middleware) { + const nextMiddlewares = cloneOperationMiddlewares(scopedOperationMiddlewares) + nextMiddlewares[name].push(middleware) + return makeBuilder(nextMiddlewares) + }, + useHono(path, middleware) { + route.use(path, middleware) + return builder + }, + } + + return builder + } + + return makeBuilder(cloneOperationMiddlewares(operationMiddlewares)) + } + + gateway = { + deps, + handler(name, parse, callback) { + return handlerWithMiddlewares(operationMiddlewares, name, parse, callback) + }, + route: createRoute, + use, + useHono(surface, path, middleware) { + httpMiddlewares.push({ surface, path, middleware }) + return gateway + }, + } + + return gateway +} + +function resolveRouteHandler(scope: V1GatewayRoute, handler: Handler | V1GatewayRouteHandler): Handler { + if (routeHandlerMarker in handler) + return (handler as V1GatewayRouteHandler)(scope) + return handler as Handler +} diff --git a/apps/server/src/routes/openai/v1/guards.ts b/apps/server/src/routes/openai/v1/guards.ts deleted file mode 100644 index 72c84794e..000000000 --- a/apps/server/src/routes/openai/v1/guards.ts +++ /dev/null @@ -1,19 +0,0 @@ -import type { V1RouteDeps } from './types' - -import { authGuard } from '../../../middlewares/auth' -import { configGuard } from '../../../middlewares/config-guard' -import { rateLimiter } from '../../../middlewares/rate-limit' - -export function createV1RouteGuards(deps: V1RouteDeps) { - return { - authGuard, - chatGuard: configGuard(deps.configKV, ['FLUX_PER_REQUEST'], 'Service is not available yet'), - completionsRateLimit: rateLimiter({ - max: 60, - windowSec: 60, - metrics: deps.rateLimitMetrics, - routeLabel: 'openai.completions', - }), - ttsGuard: configGuard(deps.configKV, ['FLUX_PER_1K_CHARS_TTS'], 'TTS service is not available yet'), - } -} diff --git a/apps/server/src/routes/openai/v1/response.ts b/apps/server/src/routes/openai/v1/http/response.ts similarity index 100% rename from apps/server/src/routes/openai/v1/response.ts rename to apps/server/src/routes/openai/v1/http/response.ts diff --git a/apps/server/src/routes/openai/v1/index.ts b/apps/server/src/routes/openai/v1/index.ts index b74228551..ddaeaadbc 100644 --- a/apps/server/src/routes/openai/v1/index.ts +++ b/apps/server/src/routes/openai/v1/index.ts @@ -1,58 +1,27 @@ -import type { PostHog } from 'posthog-node' +import type { Context } from 'hono' -import type { GenAiMetrics, RateLimitMetrics, RevenueMetrics } from '../../../otel' -import type { ConfigKVService } from '../../../services/adapters/config-kv' -import type { BillingService } from '../../../services/domain/billing/billing-service' -import type { FluxMeter } from '../../../services/domain/billing/flux-meter' -import type { FluxService } from '../../../services/domain/flux' -import type { LlmRouterService } from '../../../services/domain/llm-router' -import type { RequestLogService } from '../../../services/domain/request-log' import type { HonoEnv } from '../../../types/hono' -import type { LlmTracingDeps } from './types' +import type { LlmTracingDeps, V1RouteDeps } from './types' -import { Hono } from 'hono' - -import { createAudioCatalogHandlers } from './catalog' -import { createChatCompletionHandler } from './chat' -import { createV1RouteGuards } from './guards' -import { createSpeechHandler } from './speech' +import { authGuard } from '../../../middlewares/auth' +import { configGuard } from '../../../middlewares/config-guard' +import { createV1Gateway } from './gateway' +import { chatCompletionsRateLimit } from './middlewares' +import { chatCompletions } from './operations/chat-completions' +import { createSpeechCatalogOperation } from './operations/speech-catalog' +import { speechGeneration } from './operations/speech-generation' import { defaultLlmTracing } from './types' -export function createV1Routes( - fluxService: FluxService, - billingService: BillingService, - configKV: ConfigKVService, - requestLogService: RequestLogService, - ttsMeter: FluxMeter, - llmRouter: LlmRouterService, - genAi?: GenAiMetrics | null, - revenue?: RevenueMetrics | null, - rateLimitMetrics?: RateLimitMetrics | null, - posthog?: PostHog | null, - llmTracing: LlmTracingDeps = defaultLlmTracing, -) { - const deps = { - fluxService, - billingService, - configKV, - requestLogService, - ttsMeter, - llmRouter, - genAi, - revenue, - rateLimitMetrics, - posthog, - llmTracing, - } - const guards = createV1RouteGuards(deps) - const handleCompletion = createChatCompletionHandler(deps) - const handleTTS = createSpeechHandler(deps) - const { - handleListStreamingTTSModels, - handleListStreamingVoices, - handleListTTSModels, - handleListVoices, - } = createAudioCatalogHandlers(deps) +export interface CreateV1RoutesDeps extends Omit { + llmTracing?: LlmTracingDeps +} + +export function createV1Routes(input: CreateV1RoutesDeps) { + const deps: V1RouteDeps = { ...input, llmTracing: input.llmTracing ?? defaultLlmTracing } + const gateway = createV1Gateway(deps) + .useHono('*', '*', authGuard) + .useHono('openai', '/chat/*', configGuard(deps.configKV, ['FLUX_PER_REQUEST'], 'Service is not available yet')) + .useHono('audio', '/speech', configGuard(deps.configKV, ['FLUX_PER_1K_CHARS_TTS'], 'TTS service is not available yet')) // OpenAI-compatible surface (mounted at /api/v1/openai). Only routes that // mirror an actual OpenAI public endpoint belong here. Audio used to live @@ -60,22 +29,58 @@ export function createV1Routes( // real OpenAI route and the streaming TTS protocol has nothing to do with // OpenAI — keeping them here mislabelled the surface, so audio now mounts // at /api/v1/audio (see `audioRoutes` below). - const openaiRoutes = new Hono() - .use('*', guards.authGuard) - .post('/chat/completions', guards.completionsRateLimit, guards.chatGuard, handleCompletion) - .post('/chat/completion', guards.completionsRateLimit, guards.chatGuard, handleCompletion) + const openai = gateway.route('openai') + .use('chat.completions', chatCompletionsRateLimit({ metrics: deps.rateLimitMetrics })) + const openaiRoutes = openai + .post('/chat/completions', openai.handler( + 'chat.completions', + async (c) => { + const user = c.get('user')! + const body = await c.req.json() as Record + + return { + userId: user.id, + body, + sessionId: c.req.header('x-airi-session-id'), + abortSignal: c.req.raw.signal, + } + }, + chatCompletions(deps), + )) + .route + + const audio = gateway.route('audio') + const speechCatalog = createSpeechCatalogOperation(deps) // AIRI audio surface (mounted at /api/v1/audio). Lives outside /openai/ so // the `/voices`, `/voices/streaming`, and `/models` extensions aren't // misread as OpenAI-compatible. `/audio/speech/ws` is registered // separately in app.ts because it needs the WebSocket upgrade middleware. - const audioRoutes = new Hono() - .use('*', guards.authGuard) - .post('/speech', guards.ttsGuard, handleTTS) - .get('/voices', handleListVoices) - .get('/voices/streaming', handleListStreamingVoices) - .get('/models', handleListTTSModels) - .get('/models/streaming', handleListStreamingTTSModels) + const audioRoutes = audio + .post('/speech', audio.handler( + 'speech.generate', + async (c) => { + const user = c.get('user')! + const body = await c.req.json() as Record + + return { + userId: user.id, + body, + sessionId: c.req.header('x-airi-session-id'), + abortSignal: c.req.raw.signal, + } + }, + speechGeneration(deps), + )) + .get('/voices', (c: Context) => speechCatalog.listVoices({ + requestedModel: c.req.query('model'), + })) + .get('/voices/streaming', (c: Context) => speechCatalog.listStreamingVoices({ + model: c.req.query('model'), + })) + .get('/models', () => speechCatalog.listSpeechModels()) + .get('/models/streaming', () => speechCatalog.listStreamingSpeechModels()) + .route return { openaiRoutes, audioRoutes } } diff --git a/apps/server/src/routes/openai/v1/billing.ts b/apps/server/src/routes/openai/v1/middlewares/billing.ts similarity index 88% rename from apps/server/src/routes/openai/v1/billing.ts rename to apps/server/src/routes/openai/v1/middlewares/billing.ts index e6a4e6d8f..6bbc9d0a2 100644 --- a/apps/server/src/routes/openai/v1/billing.ts +++ b/apps/server/src/routes/openai/v1/middlewares/billing.ts @@ -1,13 +1,13 @@ -import type { RevenueMetrics } from '../../../otel' -import type { ConfigKVService } from '../../../services/adapters/config-kv' -import type { UsageInfo } from '../../../services/domain/billing/billing' -import type { BillingService } from '../../../services/domain/billing/billing-service' -import type { FluxMeter } from '../../../services/domain/billing/flux-meter' -import type { FluxService } from '../../../services/domain/flux' +import type { RevenueMetrics } from '../../../../otel' +import type { ConfigKVService } from '../../../../services/adapters/config-kv' +import type { UsageInfo } from '../../../../services/domain/billing/billing' +import type { BillingService } from '../../../../services/domain/billing/billing-service' +import type { FluxMeter } from '../../../../services/domain/billing/flux-meter' +import type { FluxService } from '../../../../services/domain/flux' -import { calculateFluxFromUsage } from '../../../services/domain/billing/billing' -import { createPaymentRequiredError } from '../../../utils/error' -import { GEN_AI_ATTR_REQUEST_MODEL } from '../../../utils/observability' +import { calculateFluxFromUsage } from '../../../../services/domain/billing/billing' +import { createPaymentRequiredError } from '../../../../utils/error' +import { GEN_AI_ATTR_REQUEST_MODEL } from '../../../../utils/observability' export interface ChatFluxDebitInput extends UsageInfo { billingService: BillingService diff --git a/apps/server/src/routes/openai/v1/middlewares/index.ts b/apps/server/src/routes/openai/v1/middlewares/index.ts new file mode 100644 index 000000000..8b3d4fffe --- /dev/null +++ b/apps/server/src/routes/openai/v1/middlewares/index.ts @@ -0,0 +1,3 @@ +export * from './billing' +export * from './telemetry' +export * from './traffic-control' diff --git a/apps/server/src/routes/openai/v1/telemetry.ts b/apps/server/src/routes/openai/v1/middlewares/telemetry.ts similarity index 94% rename from apps/server/src/routes/openai/v1/telemetry.ts rename to apps/server/src/routes/openai/v1/middlewares/telemetry.ts index 65cc074d7..558487006 100644 --- a/apps/server/src/routes/openai/v1/telemetry.ts +++ b/apps/server/src/routes/openai/v1/middlewares/telemetry.ts @@ -1,7 +1,7 @@ -import type { GenAiMetrics } from '../../../otel' -import type { UsageInfo } from '../../../services/domain/billing/billing' -import type { LlmRouteContext } from '../../../services/domain/llm-router' -import type { RequestLogService } from '../../../services/domain/request-log' +import type { GenAiMetrics } from '../../../../otel' +import type { UsageInfo } from '../../../../services/domain/billing/billing' +import type { LlmRouteContext } from '../../../../services/domain/llm-router' +import type { RequestLogService } from '../../../../services/domain/request-log' import { useLogger } from '@guiiai/logg' import { context, SpanStatusCode, trace } from '@opentelemetry/api' @@ -15,7 +15,7 @@ import { GEN_AI_ATTR_REQUEST_MODEL, GEN_AI_ATTR_USAGE_INPUT_TOKENS, GEN_AI_ATTR_USAGE_OUTPUT_TOKENS, -} from '../../../utils/observability' +} from '../../../../utils/observability' export const tracer = trace.getTracer('v1-completions') diff --git a/apps/server/src/routes/openai/v1/middlewares/traffic-control.ts b/apps/server/src/routes/openai/v1/middlewares/traffic-control.ts new file mode 100644 index 000000000..c1b142794 --- /dev/null +++ b/apps/server/src/routes/openai/v1/middlewares/traffic-control.ts @@ -0,0 +1,78 @@ +import type { RateLimitMetrics } from '../../../../otel' +import type { GatewayMiddleware, V1GatewayContext, V1GatewayOperationName } from '../gateway' + +type RateLimitKeyType = 'ip' | 'model' | 'user' + +interface GatewayRateLimitClassification { + key: string + keyType: RateLimitKeyType + model?: string +} + +interface GatewayRateLimitOptions { + classify: (context: V1GatewayContext) => GatewayRateLimitClassification + max: number + metrics?: RateLimitMetrics | null + routeLabel: string + windowSec: number +} + +interface RateLimitBucket { + count: number + resetAt: number +} + +export function chatCompletionsRateLimit(input: { + metrics?: RateLimitMetrics | null +}): GatewayMiddleware<'chat.completions'> { + return createGatewayRateLimiter({ + classify: context => ({ + key: context.input.userId, + keyType: 'user', + model: typeof context.input.body.model === 'string' ? context.input.body.model : 'auto', + }), + max: 60, + metrics: input.metrics, + routeLabel: 'openai.completions', + windowSec: 60, + }) +} + +function createGatewayRateLimiter(opts: GatewayRateLimitOptions): GatewayMiddleware { + const buckets = new Map() + + return async function limitGatewayOperation(context, next) { + const now = Date.now() + const classification = opts.classify(context) + const bucketKey = `${opts.routeLabel}:${classification.keyType}:${classification.key}` + const existing = buckets.get(bucketKey) + const bucket = existing && existing.resetAt > now + ? existing + : { count: 0, resetAt: now + opts.windowSec * 1000 } + + if (bucket.count >= opts.max) { + opts.metrics?.blocked.add(1, { + route: opts.routeLabel, + key_type: classification.keyType, + limit: String(opts.max), + }) + const retryAfterSec = Math.max(1, Math.ceil((bucket.resetAt - now) / 1000)) + return Response.json( + { error: 'TOO_MANY_REQUESTS', message: 'Too many requests' }, + { + status: 429, + headers: { + 'RateLimit-Limit': String(opts.max), + 'RateLimit-Remaining': '0', + 'RateLimit-Reset': String(retryAfterSec), + 'Retry-After': String(retryAfterSec), + }, + }, + ) + } + + bucket.count += 1 + buckets.set(bucketKey, bucket) + return next() + } +} diff --git a/apps/server/src/routes/openai/v1/chat.ts b/apps/server/src/routes/openai/v1/operations/chat-completions/index.ts similarity index 91% rename from apps/server/src/routes/openai/v1/chat.ts rename to apps/server/src/routes/openai/v1/operations/chat-completions/index.ts index ac350b671..71d273fc7 100644 --- a/apps/server/src/routes/openai/v1/chat.ts +++ b/apps/server/src/routes/openai/v1/operations/chat-completions/index.ts @@ -1,23 +1,28 @@ -import type { Context, Handler } from 'hono' - -import type { UsageInfo } from '../../../services/domain/billing/billing' -import type { HonoEnv } from '../../../types/hono' -import type { V1RouteDeps } from './types' +import type { UsageInfo } from '../../../../../services/domain/billing/billing' +import type { GatewayCallback } from '../../gateway' +import type { V1RouteDeps } from '../../types' import { useLogger } from '@guiiai/logg' -import { captureSafe } from '../../../services/adapters/posthog' -import { extractUsageFromBody } from '../../../services/domain/billing/billing' -import { nanoid } from '../../../utils/id' -import { createOpenAiRouteBilling } from './billing' -import { buildSafeResponseHeaders } from './response' -import { createRouteTelemetry, newRouteContext } from './telemetry' +import { captureSafe } from '../../../../../services/adapters/posthog' +import { extractUsageFromBody } from '../../../../../services/domain/billing/billing' +import { nanoid } from '../../../../../utils/id' +import { buildSafeResponseHeaders } from '../../http/response' +import { createOpenAiRouteBilling } from '../../middlewares/billing' +import { createRouteTelemetry, newRouteContext } from '../../middlewares/telemetry' type ChatBilling = ReturnType type ChatBillingPolicy = Awaited> type RouteTelemetry = ReturnType -export function createChatCompletionHandler(deps: V1RouteDeps): Handler { +export interface ChatCompletionsOperationRequest { + userId: string + body: Record + sessionId?: string + abortSignal?: AbortSignal +} + +export function chatCompletions(deps: V1RouteDeps): GatewayCallback<'chat.completions'> { const logger = useLogger('v1-completions').useGlobalConfig() const telemetry = createRouteTelemetry({ genAi: deps.genAi, @@ -25,18 +30,18 @@ export function createChatCompletionHandler(deps: V1RouteDeps): Handler }) const billing = createOpenAiRouteBilling(deps) - return async function handleCompletion(c: Context) { - const user = c.get('user')! + return async (context) => { + const input = context.input // Generated up-front so incoming, completion, partial-debit, debit-failure, // and request-log entries all carry the same correlation id. Re-used as // the billing requestId (both streaming and non-streaming branches) for // DB-level idempotency. const requestId = nanoid() - const billingPolicy = await billing.authorizeChat(user.id) + const billingPolicy = await billing.authorizeChat(input.userId) - const body = await c.req.json() - let requestModel = body.model || 'auto' + const body = input.body + let requestModel = typeof body.model === 'string' && body.model.length > 0 ? body.model : 'auto' if (requestModel === 'auto') { requestModel = await deps.configKV.getOrThrow('DEFAULT_CHAT_MODEL') @@ -45,7 +50,7 @@ export function createChatCompletionHandler(deps: V1RouteDeps): Handler const stream = !!body.stream logger.withFields({ requestId, - userId: user.id, + userId: input.userId, model: requestModel, stream, messageCount: Array.isArray(body.messages) ? body.messages.length : undefined, @@ -67,7 +72,7 @@ export function createChatCompletionHandler(deps: V1RouteDeps): Handler // caller hangs up. Without this the streaming-cancel path records // fluxConsumed: 0 while real cost was incurred — a silent revenue leak. // Source: codex review 2026-05-15 HIGH #1. - const clientAbort = c.req.raw.signal + const clientAbort = input.abortSignal const routeCtx = newRouteContext() let response: Response try { @@ -81,8 +86,8 @@ export function createChatCompletionHandler(deps: V1RouteDeps): Handler model: routeCtx.upstreamModel ?? requestModel, requestId, stream, - userId: user.id, - sessionId: c.req.header('x-airi-session-id'), + userId: input.userId, + sessionId: input.sessionId, }).fail('Router exhausted or unknown model') telemetry.recordMetrics({ model: requestModel, status: 502, type: 'chat', provider: routeCtx.provider, durationMs: Date.now() - startedAt, fluxConsumed: 0 }) throw err @@ -102,8 +107,8 @@ export function createChatCompletionHandler(deps: V1RouteDeps): Handler model: langfuseModel, requestId, stream, - userId: user.id, - sessionId: c.req.header('x-airi-session-id'), + userId: input.userId, + sessionId: input.sessionId, }) if (!response.ok) { @@ -113,7 +118,7 @@ export function createChatCompletionHandler(deps: V1RouteDeps): Handler // Emit server-side so funnels see real HTTP status — the client only // ever observes "stream closed" and cannot tell 401 / 429 / 5xx apart. void captureSafe(deps.posthog ?? null, { - distinctId: user.id, + distinctId: input.userId, event: 'llm_request_failed', properties: { model: requestModel, @@ -123,7 +128,7 @@ export function createChatCompletionHandler(deps: V1RouteDeps): Handler }, }) - logger.withFields({ requestId, userId: user.id, model: requestModel, status: response.status, durationMs }) + logger.withFields({ requestId, userId: input.userId, model: requestModel, status: response.status, durationMs }) .warn('chat completion delivered with upstream error status') return new Response(response.body, { @@ -141,7 +146,7 @@ export function createChatCompletionHandler(deps: V1RouteDeps): Handler startedAt, durationMs, requestId, - userId: user.id, + userId: input.userId, requestModel, routeCtxProvider: routeCtx.provider, billing, @@ -152,14 +157,13 @@ export function createChatCompletionHandler(deps: V1RouteDeps): Handler } return completeNonStreamingChat({ - c, deps, response, generationTrace, span, durationMs, requestId, - userId: user.id, + userId: input.userId, requestModel, routeCtxProvider: routeCtx.provider, billing, @@ -360,7 +364,6 @@ function streamChatCompletion(input: { } async function completeNonStreamingChat(input: { - c: Context deps: V1RouteDeps response: Response generationTrace: ReturnType @@ -452,5 +455,5 @@ async function completeNonStreamingChat(input: { stream: false, }).log('chat completion delivered') - return input.c.json(responseBody) + return Response.json(responseBody) } diff --git a/apps/server/src/routes/openai/v1/catalog.ts b/apps/server/src/routes/openai/v1/operations/speech-catalog/index.ts similarity index 82% rename from apps/server/src/routes/openai/v1/catalog.ts rename to apps/server/src/routes/openai/v1/operations/speech-catalog/index.ts index b61ba080a..6db7ccabb 100644 --- a/apps/server/src/routes/openai/v1/catalog.ts +++ b/apps/server/src/routes/openai/v1/operations/speech-catalog/index.ts @@ -1,24 +1,29 @@ -import type { Context, Handler } from 'hono' - -import type { HonoEnv } from '../../../types/hono' -import type { V1RouteDeps } from './types' +import type { V1RouteDeps } from '../../types' import { useLogger } from '@guiiai/logg' import { ofetch } from 'ofetch' -import { createBadGatewayError, createBadRequestError, createServiceUnavailableError } from '../../../utils/error' +import { createBadGatewayError, createBadRequestError, createServiceUnavailableError } from '../../../../../utils/error' -export interface AudioCatalogHandlers { - handleListStreamingTTSModels: Handler - handleListStreamingVoices: Handler - handleListTTSModels: Handler - handleListVoices: Handler +export interface SpeechCatalogOperation { + listSpeechModels: () => Promise + listStreamingSpeechModels: () => Promise + listStreamingVoices: (input: ListStreamingVoicesInput) => Promise + listVoices: (input: ListVoicesInput) => Promise } -export function createAudioCatalogHandlers(deps: V1RouteDeps): AudioCatalogHandlers { +export interface ListStreamingVoicesInput { + model?: string +} + +export interface ListVoicesInput { + requestedModel?: string +} + +export function createSpeechCatalogOperation(deps: V1RouteDeps): SpeechCatalogOperation { const logger = useLogger('v1-completions').useGlobalConfig() - async function handleListVoices(c: Context) { + async function listVoices(input: ListVoicesInput) { // Voice catalogs are per-model. Live providers (Azure) call upstream // via unspeech; static providers (cosyvoice, volcengine) return their // bundled JSON. The Redis cache + invalidation lives one layer down @@ -29,7 +34,7 @@ export function createAudioCatalogHandlers(deps: V1RouteDeps): AudioCatalogHandl // No implicit fallback: an empty `?model=` is a client bug (the UI is // expected to pass either an explicit model id or the `auto` alias) and // returns 400 instead of silently resolving to DEFAULT_TTS_MODEL. - const requested = c.req.query('model') + const requested = input.requestedModel if (requested === undefined || requested === '') throw createBadRequestError('audio voices: ?model= is required (use `auto` to defer to DEFAULT_TTS_MODEL)', 'MISSING_MODEL') @@ -49,11 +54,11 @@ export function createAudioCatalogHandlers(deps: V1RouteDeps): AudioCatalogHandl /** * Voice catalog for the streaming TTS provider (`/audio/speech/ws`). * - * Errors propagate verbatim: missing config → 503, malformed upstream - * URL → 502, unspeech network failure → 502, unspeech non-2xx → 502. - * No empty-array fallback — the UI surfaces a real failure state. + * Errors propagate verbatim: missing config -> 503, malformed upstream + * URL -> 502, unspeech network failure -> 502, unspeech non-2xx -> 502. + * No empty-array fallback: the UI surfaces a real failure state. */ - async function handleListStreamingVoices(c: Context) { + async function listStreamingVoices(input: ListStreamingVoicesInput) { const unspeech = await deps.configKV.getOptional('UNSPEECH_UPSTREAM') if (!unspeech?.streaming?.baseURL) throw createServiceUnavailableError('streaming tts upstream not configured', 'STREAMING_TTS_NOT_CONFIGURED') @@ -61,7 +66,7 @@ export function createAudioCatalogHandlers(deps: V1RouteDeps): AudioCatalogHandl // Pass through the api_resource_id (e.g. `seed-tts-2.0`). unspeech // filters the embedded Volcengine catalogue server-side; absent model // means "return everything streaming-safe". - const model = c.req.query('model') + const model = input.model let voicesURL: string try { @@ -117,7 +122,7 @@ export function createAudioCatalogHandlers(deps: V1RouteDeps): AudioCatalogHandl return Response.json({ voices: data.voices, recommended }) } - async function handleListTTSModels(_c: Context) { + async function listSpeechModels() { // Surface the concrete TTS models the operator has configured. The UI // should select an explicit model id so voice catalog requests stay // model-scoped instead of hiding behind DEFAULT_TTS_MODEL. @@ -132,7 +137,7 @@ export function createAudioCatalogHandlers(deps: V1RouteDeps): AudioCatalogHandl }) } - async function handleListStreamingTTSModels(_c: Context) { + async function listStreamingSpeechModels() { const unspeech = await deps.configKV.getOptional('UNSPEECH_UPSTREAM') const models = unspeech?.streaming?.models ?? [] // `available` is the operator-controlled visibility switch the client gates @@ -152,9 +157,9 @@ export function createAudioCatalogHandlers(deps: V1RouteDeps): AudioCatalogHandl } return { - handleListStreamingTTSModels, - handleListStreamingVoices, - handleListTTSModels, - handleListVoices, + listSpeechModels, + listStreamingSpeechModels, + listStreamingVoices, + listVoices, } } diff --git a/apps/server/src/routes/openai/v1/operations/speech-generation/index.ts b/apps/server/src/routes/openai/v1/operations/speech-generation/index.ts new file mode 100644 index 000000000..f93d8da5e --- /dev/null +++ b/apps/server/src/routes/openai/v1/operations/speech-generation/index.ts @@ -0,0 +1,25 @@ +import type { GatewayCallback } from '../../gateway' +import type { V1RouteDeps } from '../../types' + +import { createOpenAiSpeechService } from '../../../../../services/domain/openai-speech' + +export interface SpeechGenerationOperationRequest { + userId: string + body: Record + sessionId?: string + abortSignal?: AbortSignal +} + +export function speechGeneration(deps: V1RouteDeps): GatewayCallback<'speech.generate'> { + const speechService = createOpenAiSpeechService({ + configKV: deps.configKV, + fluxService: deps.fluxService, + genAi: deps.genAi, + llmRouter: deps.llmRouter, + llmTracing: deps.llmTracing, + requestLogService: deps.requestLogService, + ttsMeter: deps.ttsMeter, + }) + + return context => speechService.handleSpeechRequest(context.input) +} diff --git a/apps/server/src/routes/openai/v1/route.test.ts b/apps/server/src/routes/openai/v1/route.test.ts index 1f510ca85..7b919f547 100644 --- a/apps/server/src/routes/openai/v1/route.test.ts +++ b/apps/server/src/routes/openai/v1/route.test.ts @@ -137,19 +137,19 @@ function createTestApp( llmRouter?: LlmRouterService, llmTracing = createMockLlmTracing(), ) { - const { openaiRoutes, audioRoutes } = createV1Routes( + const { openaiRoutes, audioRoutes } = createV1Routes({ fluxService, - billingService ?? createMockBillingService(), + billingService: billingService ?? createMockBillingService(), configKV, - requestLogService ?? createMockRequestLogService(), - ttsMeter ?? createMockTtsMeter(), - llmRouter ?? createMockLlmRouter(), - null, - null, - null, - null, + requestLogService: requestLogService ?? createMockRequestLogService(), + ttsMeter: ttsMeter ?? createMockTtsMeter(), + llmRouter: llmRouter ?? createMockLlmRouter(), + genAi: null, + revenue: null, + rateLimitMetrics: null, + posthog: null, llmTracing, - ) + }) const app = new Hono() app.onError((err, c) => { @@ -263,6 +263,50 @@ describe('v1CompletionsRoutes', () => { expect(billingService.consumeFluxForLLM).not.toHaveBeenCalled() }) + it('rate-limits chat completions at the gateway operation boundary', async () => { + globalThis.fetch = vi.fn(async () => + Response.json({ + id: 'chatcmpl-test', + choices: [{ message: { role: 'assistant', content: 'ok' } }], + usage: { prompt_tokens: 1, completion_tokens: 1 }, + })) as any + const llmRouter = createMockLlmRouter() + const app = createTestApp( + createMockFluxService(1000), + createMockConfigKV(), + createMockBillingService(1000), + undefined, + undefined, + llmRouter, + ) + + for (let i = 0; i < 60; i += 1) { + const res = await app.fetch( + new Request('http://localhost/api/v1/openai/chat/completions', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ model: 'auto', messages: [{ role: 'user', content: `hi ${i}` }] }), + }), + { user: testUser } as any, + ) + expect(res.status).toBe(200) + } + + const limited = await app.fetch( + new Request('http://localhost/api/v1/openai/chat/completions', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ model: 'auto', messages: [{ role: 'user', content: 'blocked' }] }), + }), + { user: testUser } as any, + ) + const body = await limited.json() + + expect(limited.status).toBe(429) + expect(body).toEqual({ error: 'TOO_MANY_REQUESTS', message: 'Too many requests' }) + expect(llmRouter.route).toHaveBeenCalledTimes(60) + }) + // ROOT CAUSE: // // Before: when usage arrived and `fluxConsumed > balance`, debitFlux @@ -1213,12 +1257,7 @@ describe('v1CompletionsRoutes', () => { expect(res.status).toBe(404) }) - it('pOST /api/v1/openai/chat/completion (singular) should also work', async () => { - globalThis.fetch = vi.fn(async () => new Response('{}', { - status: 200, - headers: { 'Content-Type': 'application/json' }, - })) - + it('pOST /api/v1/openai/chat/completion (singular) should return 404', async () => { const app = createTestApp(createMockFluxService(), createMockConfigKV()) const res = await app.fetch( @@ -1229,7 +1268,7 @@ describe('v1CompletionsRoutes', () => { }), { user: testUser } as any, ) - expect(res.status).toBe(200) + expect(res.status).toBe(404) }) }) }) diff --git a/apps/server/src/routes/openai/v1/speech.ts b/apps/server/src/routes/openai/v1/speech.ts deleted file mode 100644 index bd72855d2..000000000 --- a/apps/server/src/routes/openai/v1/speech.ts +++ /dev/null @@ -1,30 +0,0 @@ -import type { Context, Handler } from 'hono' - -import type { HonoEnv } from '../../../types/hono' -import type { V1RouteDeps } from './types' - -import { createOpenAiSpeechService } from '../../../services/domain/openai-speech' - -export function createSpeechHandler(deps: V1RouteDeps): Handler { - const speechService = createOpenAiSpeechService({ - configKV: deps.configKV, - fluxService: deps.fluxService, - genAi: deps.genAi, - llmRouter: deps.llmRouter, - llmTracing: deps.llmTracing, - requestLogService: deps.requestLogService, - ttsMeter: deps.ttsMeter, - }) - - return async function handleTTS(c: Context) { - const user = c.get('user')! - const body = await c.req.json() as Record - - return speechService.handleSpeechRequest({ - userId: user.id, - body, - sessionId: c.req.header('x-airi-session-id'), - abortSignal: c.req.raw.signal, - }) - } -}