diff --git a/package.json b/package.json index c14b82d..944b17a 100644 --- a/package.json +++ b/package.json @@ -11,6 +11,7 @@ "zod": "^3.25.76" }, "devDependencies": { + "typescript": "^5.9.3", "wrangler": "^4.61.1" }, "packageManager": "pnpm@10.10.0+sha512.d615db246fe70f25dcfea6d8d73dee782ce23e2245e3c4f6f888249fb568149318637dca73c2c5c8ef2a4ca0d5657fb9567188bfab47f566d1ee6ce987815c39" diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 15b87b4..7b36494 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -18,6 +18,9 @@ importers: specifier: ^3.25.76 version: 3.25.76 devDependencies: + typescript: + specifier: ^5.9.3 + version: 5.9.3 wrangler: specifier: ^4.61.1 version: 4.61.1 @@ -471,6 +474,11 @@ packages: tslib@2.8.1: resolution: {integrity: sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==} + typescript@5.9.3: + resolution: {integrity: sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==} + engines: {node: '>=14.17'} + hasBin: true + undici@7.18.2: resolution: {integrity: sha512-y+8YjDFzWdQlSE9N5nzKMT3g4a5UBX1HKowfdXh0uvAnTaqqwqB92Jt4UXBAeKekDs5IaDKyJFR4X1gYVCgXcw==} engines: {node: '>=20.18.1'} @@ -877,6 +885,8 @@ snapshots: tslib@2.8.1: optional: true + typescript@5.9.3: {} + undici@7.18.2: {} unenv@2.0.0-rc.24: diff --git a/src/providers/azure-openai-proxy.ts b/src/providers/azure-openai-proxy.ts index b6f0a7f..b5b4fc3 100644 --- a/src/providers/azure-openai-proxy.ts +++ b/src/providers/azure-openai-proxy.ts @@ -1,7 +1,6 @@ import { Context } from "hono" -import { streamText } from 'hono/streaming' +import { checkoutUsageData, handleStreamResponse } from "./shared/openai-stream-utils" -// 辅助函数 const buildProxyRequest = ( request: Request, reqJson: any, @@ -11,7 +10,6 @@ const buildProxyRequest = ( const url = new URL(request.url) const targetUrl = new URL(config.endpoint) - // 如果 endpoint 末尾不带 #,则设置标准的 Azure OpenAI pathname(保留原有 path 前缀) if (!config.endpoint.endsWith('#')) { const basePath = targetUrl.pathname.endsWith('/') ? targetUrl.pathname.slice(0, -1) @@ -25,6 +23,8 @@ const buildProxyRequest = ( const targetHeaders = new Headers(request.headers) targetHeaders.delete("Authorization") + targetHeaders.delete("Host") + targetHeaders.delete("Cookie") targetHeaders.set("api-key", config.api_key) return new Request(targetUrl, { @@ -34,80 +34,6 @@ const buildProxyRequest = ( }) } -const checkoutUsageData = async ( - saveUsage: (usage: Usage) => Promise, - response: Response | OpenAIResponse -): Promise => { - try { - const res = response instanceof Response - ? await response.clone().json() - : response - if (!res.usage) return; - await saveUsage(res.usage) - } catch (error) { - console.error("Error logging usage data:", error) - } -} - -const processStreamData = async ( - lines: string[], - usageSaved: { value: boolean }, - saveUsage: (usage: Usage) => Promise -): Promise => { - if (usageSaved.value) return - const processedLines = lines - .map(line => line.trim()) - .filter(line => line.length > 0 && line.startsWith('data:')) - .map(line => line.replace('data: ', '').trim()) - .filter(line => line !== '[DONE]') - - for (const jsonContent of processedLines) { - try { - const res = JSON.parse(jsonContent) as OpenAIResponse - if (!res.usage) continue - await saveUsage(res.usage) - usageSaved.value = true - break - } catch (e) { - console.error("Error parsing stream data:", e) - } - } -} - -const handleStreamResponse = async ( - c: Context, - streamForServer: ReadableStream | undefined, - saveUsage: (usage: Usage) => Promise -): Promise => { - const reader = streamForServer?.getReader() - if (!reader) { - throw new Error("No reader found in response body") - } - - const decoder = new TextDecoder('utf-8') - let buffer = "" - const usageSaved = { value: false } - while (true) { - const { done, value } = await reader.read() - if (done) break - - const chunk = decoder.decode(value, { stream: true }) - buffer += chunk - - if (!chunk.includes('\n')) continue - - const lines = buffer.split('\n') - buffer = lines.pop() || "" - - await processStreamData(lines, usageSaved, saveUsage) - } - - // 处理最后剩余的数据 - if (buffer.trim()) { - await processStreamData([buffer], usageSaved, saveUsage) - } -} - export default { async fetch( c: Context, @@ -115,10 +41,8 @@ export default { requestBody: any, saveUsage: (usage: Usage) => Promise, ): Promise { - // model 已在上层完成映射,直接作为 deploymentName 使用 const { model: deploymentName, stream } = requestBody; - // 强制包含使用数据 if (stream) { requestBody.stream_options = { ...(requestBody.stream_options || {}), @@ -126,13 +50,9 @@ export default { } } - // 构建目标请求 const proxyRequest = buildProxyRequest(c.req.raw, requestBody, config, deploymentName) - - // 发送请求并获取响应 const response = await fetch(proxyRequest) - // 处理流式响应 if (stream) { const [streamForClient, streamForServer] = response.body?.tee() || [] c.executionCtx.waitUntil(handleStreamResponse(c, streamForServer, saveUsage)) @@ -143,7 +63,6 @@ export default { }) } - // 处理常规响应 if (response.ok) { await checkoutUsageData(saveUsage, response) } diff --git a/src/providers/azure-openai-responses-proxy.ts b/src/providers/azure-openai-responses-proxy.ts index 68ee305..4049466 100644 --- a/src/providers/azure-openai-responses-proxy.ts +++ b/src/providers/azure-openai-responses-proxy.ts @@ -1,10 +1,10 @@ import { Context } from "hono" import { + handleStreamResponse, OpenAIResponsesResponse, - OpenAIResponsesStreamEvent, extractUsageFromResponse, estimateUsageFromBodies, -} from "./shared/usage-utils" +} from "./shared/responses-stream-utils" const buildProxyRequest = ( request: Request, @@ -15,7 +15,6 @@ const buildProxyRequest = ( const targetUrl = new URL(config.endpoint) if (!config.endpoint.endsWith('#')) { - // Azure OpenAI Responses uses /openai/v1/responses targetUrl.pathname = `/openai${url.pathname}` } @@ -25,6 +24,8 @@ const buildProxyRequest = ( const targetHeaders = new Headers(request.headers) targetHeaders.delete("Authorization") + targetHeaders.delete("Host") + targetHeaders.delete("Cookie") targetHeaders.set("api-key", config.api_key) return new Request(targetUrl, { @@ -34,95 +35,6 @@ const buildProxyRequest = ( }) } -const processStreamData = async ( - lines: string[], - usageSaved: { value: boolean }, - outputText: { value: string }, - outputLimit: number, - saveUsage: (usage: Usage) => Promise -): Promise => { - if (usageSaved.value) return - const processedLines = lines - .map(line => line.trim()) - .filter(line => line.length > 0 && line.startsWith('data:')) - .map(line => line.replace('data:', '').trim()) - .filter(line => line !== '[DONE]') - - for (const jsonContent of processedLines) { - try { - const event = JSON.parse(jsonContent) as OpenAIResponsesStreamEvent - if (event.type === "response.output_text.delta") { - const deltaText = typeof event.delta === "string" - ? event.delta - : (typeof event.text === "string" ? event.text : "") - if (deltaText) { - if (outputText.value.length < outputLimit) { - outputText.value += deltaText - } - } - continue - } - if (event.type !== "response.completed") { - continue - } - const usage = extractUsageFromResponse(event.response) - if (usage && !usageSaved.value) { - await saveUsage(usage) - usageSaved.value = true - } - } catch (e) { - console.error("Error parsing stream data:", e) - } - } -} - -const handleStreamResponse = async ( - c: Context, - streamForServer: ReadableStream | undefined, - requestBody: any, - saveUsage: (usage: Usage) => Promise -): Promise => { - const reader = streamForServer?.getReader() - if (!reader) { - throw new Error("No reader found in response body") - } - - const decoder = new TextDecoder('utf-8') - let buffer = "" - const usageSaved = { value: false } - const outputText = { value: "" } - const outputLimit = 200_000 - while (true) { - const { done, value } = await reader.read() - if (done) break - - const chunk = decoder.decode(value, { stream: true }) - buffer += chunk - - if (!chunk.includes('\n')) continue - - const lines = buffer.split('\n') - buffer = lines.pop() || "" - - await processStreamData(lines, usageSaved, outputText, outputLimit, saveUsage) - } - - if (buffer.trim()) { - await processStreamData([buffer], usageSaved, outputText, outputLimit, saveUsage) - } - - if (!usageSaved.value) { - const estimatedUsage = estimateUsageFromBodies( - requestBody, - undefined, - outputText.value - ) - if (estimatedUsage) { - await saveUsage(estimatedUsage) - } - } -} - export default { async fetch( c: Context, @@ -132,8 +44,6 @@ export default { ): Promise { const { stream } = requestBody - // model 已在上层完成映射 - const proxyRequest = buildProxyRequest(c.req.raw, requestBody, config) const response = await fetch(proxyRequest) @@ -154,7 +64,7 @@ export default { if (usage) { await saveUsage(usage) } else { - const estimatedUsage = estimateUsageFromBodies(reqJson, resJson) + const estimatedUsage = estimateUsageFromBodies(requestBody, resJson) if (estimatedUsage) { await saveUsage(estimatedUsage) } diff --git a/src/providers/claude-to-openai-proxy.ts b/src/providers/claude-to-openai-proxy.ts index d0a903b..030b682 100644 --- a/src/providers/claude-to-openai-proxy.ts +++ b/src/providers/claude-to-openai-proxy.ts @@ -582,7 +582,7 @@ export default { if (response.ok) { try { - const resJson = await response.clone().json(); + const resJson = await response.clone().json>(); const usage = normalizeUsage(resJson?.usage); if (usage) { await saveUsage(usage); diff --git a/src/providers/index.ts b/src/providers/index.ts index dd5593d..ca4689e 100644 --- a/src/providers/index.ts +++ b/src/providers/index.ts @@ -1,83 +1,15 @@ import { Context, Hono } from "hono" import { contentJson, fromHono, OpenAPIRoute } from 'chanfana'; - -import azureOpenaiProxy from "./azure-openai-proxy" -import openaiProxy from "./openai-proxy" -import claudeProxy from "./claude-proxy" -import claudeToOpenaiProxy from "./claude-to-openai-proxy" -import openaiResponsesProxy from "./openai-responses-proxy" -import azureOpenaiResponsesProxy from "./azure-openai-responses-proxy" -import utils, { findDeploymentMapping } from "../utils" -import { TokenUtils } from "../admin/token_utils" -import { CONSTANTS } from "../constants" import { z } from "zod"; +import { resolveRouteId } from "./shared/route-policy" +import { resolveChannel } from "./shared/channel-resolver" +import { getProvider } from "./shared/provider-registry" +import { ModelsEndpoint } from "./models" + export const api = fromHono(new Hono()) -const providerMap: Record< - string, - ( - c: Context, - config: ChannelConfig, - requestBody: any, - saveUsage: (usage: Usage) => Promise - ) => Promise -> = { - "azure-openai": azureOpenaiProxy.fetch, - "openai": openaiProxy.fetch, - "claude": claudeProxy.fetch, - "claude-to-openai": claudeToOpenaiProxy.fetch, - "openai-responses": openaiResponsesProxy.fetch, - "azure-openai-responses": azureOpenaiResponsesProxy.fetch, -}; - -const getApiKeyFromHeaders = (c: Context): string | null => { - const authHeader = c.req.raw.headers.get('Authorization'); - const xApiKey = c.req.raw.headers.get('x-api-key'); - - if (authHeader) { - return authHeader.replace("Bearer ", "").trim(); - } - if (xApiKey) { - return xApiKey.trim(); - } - return null; -} - -const fetchTokenData = async (c: Context, apiKey: string) => { - const tokenResult = await c.env.DB.prepare( - `SELECT * FROM api_token WHERE key = ?` - ).bind(apiKey).first(); - - if (!tokenResult || !tokenResult.value) { - return null; - } - - return { - tokenData: JSON.parse(tokenResult.value as string) as ApiTokenData, - usage: tokenResult.usage as number || 0, - }; -} - -const fetchChannelsForToken = async ( - c: Context, - tokenData: ApiTokenData -) => { - const channelKeys = tokenData.channel_keys; - - if (!channelKeys || channelKeys.length === 0) { - return await c.env.DB.prepare( - `SELECT key, value FROM channel_config` - ).all(); - } - - const channelQuery = channelKeys.map(() => '?').join(','); - return await c.env.DB.prepare( - `SELECT key, value FROM channel_config WHERE key IN (${channelQuery})` - ).bind(...channelKeys).all(); -} - -class ProxyEndpoint extends OpenAPIRoute { +class UnifiedProxyEndpoint extends OpenAPIRoute { schema = { tags: ['OpenAI Proxy'], request: { @@ -97,206 +29,26 @@ class ProxyEndpoint extends OpenAPIRoute { }; async handle(c: Context) { - // Support both OpenAI format (Authorization: Bearer xxx) and Claude format (x-api-key: xxx) - const apiKey = getApiKeyFromHeaders(c); - if (!apiKey) { - return c.text("Authorization header or x-api-key not found", 401); + const routeId = resolveRouteId(c.req.path) + if (!routeId) { + return c.text("Unknown route", 404) } - const tokenInfo = await fetchTokenData(c, apiKey); - if (!tokenInfo) { - return c.text("Invalid API key", 401); + const result = await resolveChannel(c, routeId) + if (result instanceof Response) return result + + const { channel, requestBody, saveUsage } = result + + const provider = getProvider(channel.config.type || "") + if (!provider) { + return c.text("Channel type not supported", 400) } - const { tokenData, usage } = tokenInfo; - - // Check if token has sufficient quota - if (usage >= tokenData.total_quota) { - return c.text("Quota exceeded", 402); - } // Get available channel configs based on token permissions - const channelsResult = await fetchChannelsForToken(c, tokenData); - - if (!channelsResult.results || channelsResult.results.length === 0) { - return c.text("No available channels for this token", 401); - } - - let requestBody: any; - try { - requestBody = await c.req.json(); - } catch (error) { - return c.text("Invalid JSON body", 400); - } - const model = requestBody.model; - if (!model) { - return c.text("Model is required", 400); - } - - // Filter channels that support the requested model - const availableChannels: Array<{ key: string, config: ChannelConfig, mapping: { pattern: string, deployment: string } }> = []; - - for (const row of channelsResult.results) { - const config = JSON.parse(row.value) as ChannelConfig; - - // Check if channel supports the model (has deployment mapping for it) - const mapping = findDeploymentMapping(config.deployment_mapper, model); - if (mapping) { - availableChannels.push({ - key: row.key, - config: config, - mapping: mapping - }); - } - } - - if (availableChannels.length === 0) { - if (!channelsResult.results || channelsResult.results.length === 0) { - return c.text("No available channels for this token", 404); - } - return c.text(`Model not mapped: ${model}. Please configure deployment_mapper.`, 400); - } - - // Randomly select one available channel - const randomIndex = Math.floor(Math.random() * availableChannels.length); - const selectedChannel = availableChannels[randomIndex]; - const targetChannelKey = selectedChannel.key; - const targetChannelConfig = selectedChannel.config; - - // 统一在上层完成模型名映射 - requestBody.model = selectedChannel.mapping.deployment; - - if (!targetChannelConfig.type) { - return c.text("Channel type invalid", 400); - } - - const proxyFetch = providerMap[targetChannelConfig.type]; - if (!proxyFetch) { - return c.text("Channel type not supported", 400); - } - - return proxyFetch(c, targetChannelConfig, requestBody, - async (usage: Usage) => { - try { - // 使用转换后的模型名计费 - await TokenUtils.processUsage(c, apiKey, requestBody.model, targetChannelKey, targetChannelConfig, usage); - } catch (error) { - console.error('Error processing usage:', error); - } - } - ); + return provider(c, channel.config, requestBody, saveUsage) } } -api.post("/v1/chat/completions", ProxyEndpoint) -api.post("/v1/messages", ProxyEndpoint) // Claude API endpoint - -class ResponsesProxyEndpoint extends OpenAPIRoute { - schema = { - tags: ['OpenAI Responses Proxy'], - request: { - headers: z.object({ - 'Authorization': z.string().optional().describe("Token for authentication (OpenAI format)"), - 'x-api-key': z.string().optional().describe("API key for authentication (Claude format)"), - }), - body: contentJson({ - schema: z.any(), - }), - }, - responses: { - 200: { - description: 'Successful response', - }, - }, - }; - - async handle(c: Context) { - const apiKey = getApiKeyFromHeaders(c); - if (!apiKey) { - return c.text("Authorization header or x-api-key not found", 401); - } - - const tokenInfo = await fetchTokenData(c, apiKey); - if (!tokenInfo) { - return c.text("Invalid API key", 401); - } - - const { tokenData, usage } = tokenInfo; - - if (usage >= tokenData.total_quota) { - return c.text("Quota exceeded", 402); - } - - const channelsResult = await fetchChannelsForToken(c, tokenData); - - if (!channelsResult.results || channelsResult.results.length === 0) { - return c.text("No available channels for this token", 401); - } - - let requestBody: any; - try { - requestBody = await c.req.json(); - } catch (error) { - return c.text("Invalid JSON body", 400); - } - const model = requestBody.model; - if (!model) { - return c.text("Model is required", 400); - } - - const allowedTypes: ChannelType[] = ["openai-responses", "azure-openai-responses"]; - const availableChannels: Array<{ key: string, config: ChannelConfig, mapping: { pattern: string, deployment: string } }> = []; - - for (const row of channelsResult.results) { - const config = JSON.parse(row.value) as ChannelConfig; - - if (!config.type || !allowedTypes.includes(config.type)) { - continue; - } - - const mapping = findDeploymentMapping(config.deployment_mapper, model); - if (mapping) { - availableChannels.push({ - key: row.key, - config: config, - mapping: mapping - }); - } - } - - if (availableChannels.length === 0) { - if (!channelsResult.results || channelsResult.results.length === 0) { - return c.text("No available channels for this token", 404); - } - return c.text(`Model not mapped: ${model}. Please configure deployment_mapper.`, 400); - } - - const randomIndex = Math.floor(Math.random() * availableChannels.length); - const selectedChannel = availableChannels[randomIndex]; - const targetChannelKey = selectedChannel.key; - const targetChannelConfig = selectedChannel.config; - - // 统一在上层完成模型名映射 - requestBody.model = selectedChannel.mapping.deployment; - - const proxyFetch = providerMap[targetChannelConfig.type || ""]; - if (!proxyFetch) { - return c.text("Channel type not supported", 400); - } - - return proxyFetch(c, targetChannelConfig, requestBody, - async (usage: Usage) => { - try { - // 使用转换后的模型名计费 - await TokenUtils.processUsage(c, apiKey, requestBody.model, targetChannelKey, targetChannelConfig, usage); - } catch (error) { - console.error('Error processing usage:', error); - } - } - ); - } -} - -api.post("/v1/responses", ResponsesProxyEndpoint) - -// Models endpoint -import { ModelsEndpoint } from "./models" +api.post("/v1/chat/completions", UnifiedProxyEndpoint) +api.post("/v1/messages", UnifiedProxyEndpoint) +api.post("/v1/responses", UnifiedProxyEndpoint) api.get("/v1/models", ModelsEndpoint) diff --git a/src/providers/models.ts b/src/providers/models.ts index d0369e7..55d3828 100644 --- a/src/providers/models.ts +++ b/src/providers/models.ts @@ -1,81 +1,7 @@ import { Context } from "hono"; import { OpenAPIRoute } from "chanfana"; import { z } from "zod"; - -const getApiKeyFromHeaders = (c: Context): string | null => { - const authHeader = c.req.raw.headers.get('Authorization'); - const xApiKey = c.req.raw.headers.get('x-api-key'); - - if (authHeader) { - return authHeader.replace("Bearer ", "").trim(); - } - if (xApiKey) { - return xApiKey.trim(); - } - return null; -} - -const fetchTokenData = async (c: Context, apiKey: string) => { - const tokenResult = await c.env.DB.prepare( - `SELECT * FROM api_token WHERE key = ?` - ).bind(apiKey).first(); - - if (!tokenResult || !tokenResult.value) { - return null; - } - - return { - tokenData: JSON.parse(tokenResult.value as string) as ApiTokenData, - usage: tokenResult.usage as number || 0, - }; -} - -const fetchChannelsForToken = async ( - c: Context, - tokenData: ApiTokenData -) => { - const channelKeys = tokenData.channel_keys; - - if (!channelKeys || channelKeys.length === 0) { - return await c.env.DB.prepare( - `SELECT key, value FROM channel_config` - ).all(); - } - - const channelQuery = channelKeys.map(() => '?').join(','); - return await c.env.DB.prepare( - `SELECT key, value FROM channel_config WHERE key IN (${channelQuery})` - ).bind(...channelKeys).all(); -} - -const fetchAllChannels = async (c: Context) => { - return await c.env.DB.prepare( - `SELECT key, value FROM channel_config` - ).all(); -} - -const fetchChannelsForRequest = async ( - c: Context, - apiKey: string | null -) => { - if (!apiKey) { - return { - channelsResult: await fetchAllChannels(c), - }; - } - - const tokenInfo = await fetchTokenData(c, apiKey); - if (!tokenInfo) { - return { - errorResponse: c.text("Invalid API key", 401), - }; - } - - const { tokenData } = tokenInfo; - return { - channelsResult: await fetchChannelsForToken(c, tokenData), - }; -} +import { getApiKeyFromHeaders, fetchTokenData, fetchChannelsForToken } from "./shared/auth"; export class ModelsEndpoint extends OpenAPIRoute { schema = { @@ -109,12 +35,17 @@ export class ModelsEndpoint extends OpenAPIRoute { async handle(c: Context) { const apiKey = getApiKeyFromHeaders(c); - const { channelsResult, errorResponse } = await fetchChannelsForRequest(c, apiKey); - - if (errorResponse) { - return errorResponse; + if (!apiKey) { + return c.json({ object: "list", data: [] }); } + const tokenInfo = await fetchTokenData(c, apiKey); + if (!tokenInfo) { + return c.text("Invalid API key", 401); + } + + const channelsResult = await fetchChannelsForToken(c, tokenInfo.tokenData); + if (!channelsResult || !channelsResult.results || channelsResult.results.length === 0) { return c.json({ object: "list", @@ -122,7 +53,6 @@ export class ModelsEndpoint extends OpenAPIRoute { }); } - // Collect all unique models from all available channels const modelsSet = new Set(); for (const row of channelsResult.results) { @@ -134,7 +64,6 @@ export class ModelsEndpoint extends OpenAPIRoute { } } - // Convert to OpenAI models format const models = Array.from(modelsSet).sort().map((modelId) => ({ id: modelId, object: "model" as const, diff --git a/src/providers/openai-proxy.ts b/src/providers/openai-proxy.ts index 78da4c2..1def896 100644 --- a/src/providers/openai-proxy.ts +++ b/src/providers/openai-proxy.ts @@ -1,7 +1,6 @@ import { Context } from "hono" -import { streamText } from 'hono/streaming' +import { checkoutUsageData, handleStreamResponse } from "./shared/openai-stream-utils" -// 辅助函数 const buildProxyRequest = ( request: Request, reqJson: any, @@ -13,6 +12,8 @@ const buildProxyRequest = ( targetUrl.pathname = url.pathname const targetHeaders = new Headers(request.headers) + targetHeaders.delete("Host") + targetHeaders.delete("Cookie") targetHeaders.set("Authorization", `Bearer ${config.api_key}`) return new Request(targetUrl, { @@ -22,80 +23,6 @@ const buildProxyRequest = ( }) } -const checkoutUsageData = async ( - saveUsage: (usage: Usage) => Promise, - response: Response | OpenAIResponse -): Promise => { - try { - const res = response instanceof Response - ? await response.clone().json() - : response - if (!res.usage) return; - await saveUsage(res.usage) - } catch (error) { - console.error("Error logging usage data:", error) - } -} - -const processStreamData = async ( - lines: string[], - usageSaved: { value: boolean }, - saveUsage: (usage: Usage) => Promise -): Promise => { - if (usageSaved.value) return - const processedLines = lines - .map(line => line.trim()) - .filter(line => line.length > 0 && line.startsWith('data:')) - .map(line => line.replace('data: ', '').trim()) - .filter(line => line !== '[DONE]') - - for (const jsonContent of processedLines) { - try { - const res = JSON.parse(jsonContent) as OpenAIResponse - if (!res.usage) continue - await saveUsage(res.usage) - usageSaved.value = true - break - } catch (e) { - console.error("Error parsing stream data:", e) - } - } -} - -const handleStreamResponse = async ( - c: Context, - streamForServer: ReadableStream | undefined, - saveUsage: (usage: Usage) => Promise -): Promise => { - const reader = streamForServer?.getReader() - if (!reader) { - throw new Error("No reader found in response body") - } - - const decoder = new TextDecoder('utf-8') - let buffer = "" - const usageSaved = { value: false } - while (true) { - const { done, value } = await reader.read() - if (done) break - - const chunk = decoder.decode(value, { stream: true }) - buffer += chunk - - if (!chunk.includes('\n')) continue - - const lines = buffer.split('\n') - buffer = lines.pop() || "" - - await processStreamData(lines, usageSaved, saveUsage) - } - - // 处理最后剩余的数据 - if (buffer.trim()) { - await processStreamData([buffer], usageSaved, saveUsage) - } -} - export default { async fetch( c: Context, @@ -105,7 +32,6 @@ export default { ): Promise { const { stream } = requestBody; - // 强制包含使用数据 if (stream) { requestBody.stream_options = { ...(requestBody.stream_options || {}), @@ -113,15 +39,9 @@ export default { } } - // model 已在上层完成映射 - - // 构建目标请求 const proxyRequest = buildProxyRequest(c.req.raw, requestBody, config) - - // 发送请求并获取响应 const response = await fetch(proxyRequest) - // 处理流式响应 if (stream) { const [streamForClient, streamForServer] = response.body?.tee() || [] c.executionCtx.waitUntil(handleStreamResponse(c, streamForServer, saveUsage)) @@ -132,7 +52,6 @@ export default { }) } - // 处理常规响应 if (response.ok) { await checkoutUsageData(saveUsage, response) } diff --git a/src/providers/openai-responses-proxy.ts b/src/providers/openai-responses-proxy.ts index fa5c1f0..248f1b5 100644 --- a/src/providers/openai-responses-proxy.ts +++ b/src/providers/openai-responses-proxy.ts @@ -1,10 +1,10 @@ import { Context } from "hono" import { + handleStreamResponse, OpenAIResponsesResponse, - OpenAIResponsesStreamEvent, extractUsageFromResponse, estimateUsageFromBodies, -} from "./shared/usage-utils" +} from "./shared/responses-stream-utils" const buildProxyRequest = ( request: Request, @@ -17,6 +17,8 @@ const buildProxyRequest = ( targetUrl.pathname = url.pathname const targetHeaders = new Headers(request.headers) + targetHeaders.delete("Host") + targetHeaders.delete("Cookie") targetHeaders.set("Authorization", `Bearer ${config.api_key}`) return new Request(targetUrl, { @@ -26,95 +28,6 @@ const buildProxyRequest = ( }) } -const processStreamData = async ( - lines: string[], - usageSaved: { value: boolean }, - outputText: { value: string }, - outputLimit: number, - saveUsage: (usage: Usage) => Promise -): Promise => { - if (usageSaved.value) return - const processedLines = lines - .map(line => line.trim()) - .filter(line => line.length > 0 && line.startsWith('data:')) - .map(line => line.replace('data:', '').trim()) - .filter(line => line !== '[DONE]') - - for (const jsonContent of processedLines) { - try { - const event = JSON.parse(jsonContent) as OpenAIResponsesStreamEvent - if (event.type === "response.output_text.delta") { - const deltaText = typeof event.delta === "string" - ? event.delta - : (typeof event.text === "string" ? event.text : "") - if (deltaText) { - if (outputText.value.length < outputLimit) { - outputText.value += deltaText - } - } - continue - } - if (event.type !== "response.completed") { - continue - } - const usage = extractUsageFromResponse(event.response) - if (usage && !usageSaved.value) { - await saveUsage(usage) - usageSaved.value = true - } - } catch (e) { - console.error("Error parsing stream data:", e) - } - } -} - -const handleStreamResponse = async ( - c: Context, - streamForServer: ReadableStream | undefined, - requestBody: any, - saveUsage: (usage: Usage) => Promise -): Promise => { - const reader = streamForServer?.getReader() - if (!reader) { - throw new Error("No reader found in response body") - } - - const decoder = new TextDecoder('utf-8') - let buffer = "" - const usageSaved = { value: false } - const outputText = { value: "" } - const outputLimit = 200_000 - while (true) { - const { done, value } = await reader.read() - if (done) break - - const chunk = decoder.decode(value, { stream: true }) - buffer += chunk - - if (!chunk.includes('\n')) continue - - const lines = buffer.split('\n') - buffer = lines.pop() || "" - - await processStreamData(lines, usageSaved, outputText, outputLimit, saveUsage) - } - - if (buffer.trim()) { - await processStreamData([buffer], usageSaved, outputText, outputLimit, saveUsage) - } - - if (!usageSaved.value) { - const estimatedUsage = estimateUsageFromBodies( - requestBody, - undefined, - outputText.value - ) - if (estimatedUsage) { - await saveUsage(estimatedUsage) - } - } -} - export default { async fetch( c: Context, @@ -122,15 +35,14 @@ export default { requestBody: any, saveUsage: (usage: Usage) => Promise, ): Promise { - const reqJson = requestBody - const { stream } = reqJson + const { stream } = requestBody - const proxyRequest = buildProxyRequest(c.req.raw, reqJson, config) + const proxyRequest = buildProxyRequest(c.req.raw, requestBody, config) const response = await fetch(proxyRequest) if (stream) { const [streamForClient, streamForServer] = response.body?.tee() || [] - c.executionCtx.waitUntil(handleStreamResponse(c, streamForServer, reqJson, saveUsage)) + c.executionCtx.waitUntil(handleStreamResponse(c, streamForServer, requestBody, saveUsage)) return new Response(streamForClient, { headers: response.headers, status: response.status, @@ -145,7 +57,7 @@ export default { if (usage) { await saveUsage(usage) } else { - const estimatedUsage = estimateUsageFromBodies(reqJson, resJson) + const estimatedUsage = estimateUsageFromBodies(requestBody, resJson) if (estimatedUsage) { await saveUsage(estimatedUsage) } diff --git a/src/providers/shared/auth.ts b/src/providers/shared/auth.ts new file mode 100644 index 0000000..b2b90a0 --- /dev/null +++ b/src/providers/shared/auth.ts @@ -0,0 +1,53 @@ +import { Context } from "hono" + +export const getApiKeyFromHeaders = (c: Context): string | null => { + const authHeader = c.req.raw.headers.get('Authorization'); + const xApiKey = c.req.raw.headers.get('x-api-key'); + + if (authHeader) { + return authHeader.startsWith("Bearer ") ? authHeader.slice(7).trim() : authHeader.trim(); + } + if (xApiKey) { + return xApiKey.trim(); + } + return null; +} + +export const fetchTokenData = async (c: Context, apiKey: string) => { + const tokenResult = await c.env.DB.prepare( + `SELECT * FROM api_token WHERE key = ?` + ).bind(apiKey).first(); + + if (!tokenResult || !tokenResult.value) { + return null; + } + + return { + tokenData: JSON.parse(tokenResult.value as string) as ApiTokenData, + usage: tokenResult.usage as number || 0, + }; +} + +export const fetchChannelsForToken = async ( + c: Context, + tokenData: ApiTokenData +) => { + const channelKeys = tokenData.channel_keys; + + if (!channelKeys || channelKeys.length === 0) { + return await c.env.DB.prepare( + `SELECT key, value FROM channel_config` + ).all(); + } + + const channelQuery = channelKeys.map(() => '?').join(','); + return await c.env.DB.prepare( + `SELECT key, value FROM channel_config WHERE key IN (${channelQuery})` + ).bind(...channelKeys).all(); +} + +export const fetchAllChannels = async (c: Context) => { + return await c.env.DB.prepare( + `SELECT key, value FROM channel_config` + ).all(); +} diff --git a/src/providers/shared/channel-resolver.ts b/src/providers/shared/channel-resolver.ts new file mode 100644 index 0000000..ec2e194 --- /dev/null +++ b/src/providers/shared/channel-resolver.ts @@ -0,0 +1,100 @@ +import { Context } from "hono" +import { getApiKeyFromHeaders, fetchTokenData, fetchChannelsForToken } from "./auth" +import { RouteId, getRoutePolicy } from "./route-policy" +import { findDeploymentMapping } from "../../utils" +import { TokenUtils } from "../../admin/token_utils" + +export type ChannelResolution = { + channel: { key: string; config: ChannelConfig } + requestBody: any + saveUsage: (usage: Usage) => Promise +} + +export const resolveChannel = async ( + c: Context, + routeId: RouteId +): Promise => { + const apiKey = getApiKeyFromHeaders(c); + if (!apiKey) { + return c.text("Authorization header or x-api-key not found", 401); + } + + const tokenInfo = await fetchTokenData(c, apiKey); + if (!tokenInfo) { + return c.text("Invalid API key", 401); + } + + const { tokenData, usage } = tokenInfo; + + if (usage >= tokenData.total_quota) { + return c.text("Quota exceeded", 402); + } + + const channelsResult = await fetchChannelsForToken(c, tokenData); + + if (!channelsResult.results || channelsResult.results.length === 0) { + return c.text("No available channels for this token", 401); + } + + let requestBody: any; + try { + requestBody = await c.req.json(); + } catch (error) { + return c.text("Invalid JSON body", 400); + } + const model = requestBody.model; + if (!model) { + return c.text("Model is required", 400); + } + + const policy = getRoutePolicy(routeId); + const allowedTypes = policy.allowedTypes; + + const availableChannels: Array<{ key: string, config: ChannelConfig, mapping: { pattern: string, deployment: string } }> = []; + + for (const row of channelsResult.results) { + const config = JSON.parse(row.value) as ChannelConfig; + + if (allowedTypes && (!config.type || !allowedTypes.includes(config.type))) { + continue; + } + + const mapping = findDeploymentMapping(config.deployment_mapper, model); + if (mapping) { + availableChannels.push({ + key: row.key, + config: config, + mapping: mapping + }); + } + } + + if (availableChannels.length === 0) { + return c.text(`Model not mapped: ${model}. Please configure deployment_mapper.`, 400); + } + + const randomIndex = Math.floor(Math.random() * availableChannels.length); + const selectedChannel = availableChannels[randomIndex]; + const targetChannelKey = selectedChannel.key; + const targetChannelConfig = selectedChannel.config; + + requestBody.model = selectedChannel.mapping.deployment; + + if (!targetChannelConfig.type) { + return c.text("Channel type invalid", 400); + } + + const saveUsage = async (usage: Usage) => { + try { + await TokenUtils.processUsage(c, apiKey, requestBody.model, targetChannelKey, targetChannelConfig, usage); + } catch (error) { + console.error('Error processing usage:', error); + } + }; + + return { + channel: { key: targetChannelKey, config: targetChannelConfig }, + requestBody, + saveUsage, + }; +} diff --git a/src/providers/shared/openai-stream-utils.ts b/src/providers/shared/openai-stream-utils.ts new file mode 100644 index 0000000..b602530 --- /dev/null +++ b/src/providers/shared/openai-stream-utils.ts @@ -0,0 +1,74 @@ +import { Context } from "hono" + +export const checkoutUsageData = async ( + saveUsage: (usage: Usage) => Promise, + response: Response | OpenAIResponse +): Promise => { + try { + const res = response instanceof Response + ? await response.clone().json() + : response + if (!res.usage) return; + await saveUsage(res.usage) + } catch (error) { + console.error("Error logging usage data:", error) + } +} + +export const processStreamData = async ( + lines: string[], + usageSaved: { value: boolean }, + saveUsage: (usage: Usage) => Promise +): Promise => { + if (usageSaved.value) return + const processedLines = lines + .map(line => line.trim()) + .filter(line => line.length > 0 && line.startsWith('data:')) + .map(line => line.replace('data: ', '').trim()) + .filter(line => line !== '[DONE]') + + for (const jsonContent of processedLines) { + try { + const res = JSON.parse(jsonContent) as OpenAIResponse + if (!res.usage) continue + await saveUsage(res.usage) + usageSaved.value = true + break + } catch (e) { + console.error("Error parsing stream data:", e) + } + } +} + +export const handleStreamResponse = async ( + c: Context, + streamForServer: ReadableStream | undefined, + saveUsage: (usage: Usage) => Promise +): Promise => { + const reader = streamForServer?.getReader() + if (!reader) { + throw new Error("No reader found in response body") + } + + const decoder = new TextDecoder('utf-8') + let buffer = "" + const usageSaved = { value: false } + while (true) { + const { done, value } = await reader.read() + if (done) break + + const chunk = decoder.decode(value, { stream: true }) + buffer += chunk + + if (!chunk.includes('\n')) continue + + const lines = buffer.split('\n') + buffer = lines.pop() || "" + + await processStreamData(lines, usageSaved, saveUsage) + } + + if (buffer.trim()) { + await processStreamData([buffer], usageSaved, saveUsage) + } +} diff --git a/src/providers/shared/provider-registry.ts b/src/providers/shared/provider-registry.ts new file mode 100644 index 0000000..11ec497 --- /dev/null +++ b/src/providers/shared/provider-registry.ts @@ -0,0 +1,28 @@ +import { Context } from "hono" + +import azureOpenaiProxy from "../azure-openai-proxy" +import openaiProxy from "../openai-proxy" +import claudeProxy from "../claude-proxy" +import claudeToOpenaiProxy from "../claude-to-openai-proxy" +import openaiResponsesProxy from "../openai-responses-proxy" +import azureOpenaiResponsesProxy from "../azure-openai-responses-proxy" + +type ProviderFetch = ( + c: Context, + config: ChannelConfig, + requestBody: any, + saveUsage: (usage: Usage) => Promise +) => Promise + +const providerMap: Record = { + "azure-openai": azureOpenaiProxy.fetch, + "openai": openaiProxy.fetch, + "claude": claudeProxy.fetch, + "claude-to-openai": claudeToOpenaiProxy.fetch, + "openai-responses": openaiResponsesProxy.fetch, + "azure-openai-responses": azureOpenaiResponsesProxy.fetch, +} + +export const getProvider = (type: string): ProviderFetch | undefined => { + return providerMap[type] +} diff --git a/src/providers/shared/responses-stream-utils.ts b/src/providers/shared/responses-stream-utils.ts new file mode 100644 index 0000000..c2aa76c --- /dev/null +++ b/src/providers/shared/responses-stream-utils.ts @@ -0,0 +1,98 @@ +import { Context } from "hono" +import { + OpenAIResponsesResponse, + OpenAIResponsesStreamEvent, + extractUsageFromResponse, + estimateUsageFromBodies, +} from "./usage-utils" + +export const processStreamData = async ( + lines: string[], + usageSaved: { value: boolean }, + outputText: { value: string }, + outputLimit: number, + saveUsage: (usage: Usage) => Promise +): Promise => { + if (usageSaved.value) return + const processedLines = lines + .map(line => line.trim()) + .filter(line => line.length > 0 && line.startsWith('data:')) + .map(line => line.replace('data:', '').trim()) + .filter(line => line !== '[DONE]') + + for (const jsonContent of processedLines) { + try { + const event = JSON.parse(jsonContent) as OpenAIResponsesStreamEvent + if (event.type === "response.output_text.delta") { + const deltaText = typeof event.delta === "string" + ? event.delta + : (typeof event.text === "string" ? event.text : "") + if (deltaText) { + if (outputText.value.length < outputLimit) { + outputText.value += deltaText + } + } + continue + } + if (event.type !== "response.completed") { + continue + } + const usage = extractUsageFromResponse(event.response) + if (usage && !usageSaved.value) { + await saveUsage(usage) + usageSaved.value = true + } + } catch (e) { + console.error("Error parsing stream data:", e) + } + } +} + +export const handleStreamResponse = async ( + c: Context, + streamForServer: ReadableStream | undefined, + requestBody: any, + saveUsage: (usage: Usage) => Promise +): Promise => { + const reader = streamForServer?.getReader() + if (!reader) { + throw new Error("No reader found in response body") + } + + const decoder = new TextDecoder('utf-8') + let buffer = "" + const usageSaved = { value: false } + const outputText = { value: "" } + const outputLimit = 200_000 + while (true) { + const { done, value } = await reader.read() + if (done) break + + const chunk = decoder.decode(value, { stream: true }) + buffer += chunk + + if (!chunk.includes('\n')) continue + + const lines = buffer.split('\n') + buffer = lines.pop() || "" + + await processStreamData(lines, usageSaved, outputText, outputLimit, saveUsage) + } + + if (buffer.trim()) { + await processStreamData([buffer], usageSaved, outputText, outputLimit, saveUsage) + } + + if (!usageSaved.value) { + const estimatedUsage = estimateUsageFromBodies( + requestBody, + undefined, + outputText.value + ) + if (estimatedUsage) { + await saveUsage(estimatedUsage) + } + } +} + +export { OpenAIResponsesResponse, extractUsageFromResponse, estimateUsageFromBodies } diff --git a/src/providers/shared/route-policy.ts b/src/providers/shared/route-policy.ts new file mode 100644 index 0000000..4a9f0c6 --- /dev/null +++ b/src/providers/shared/route-policy.ts @@ -0,0 +1,22 @@ +export type RouteId = "chat-completions" | "messages" | "responses" + +type RoutePolicy = { + allowedTypes: ChannelType[] | null +} + +const ROUTE_POLICIES: Record = { + "chat-completions": { allowedTypes: null }, + "messages": { allowedTypes: null }, + "responses": { allowedTypes: ["openai-responses", "azure-openai-responses"] }, +} + +export const resolveRouteId = (pathname: string): RouteId | null => { + if (pathname.endsWith("/chat/completions")) return "chat-completions" + if (pathname.endsWith("/messages")) return "messages" + if (pathname.endsWith("/responses")) return "responses" + return null +} + +export const getRoutePolicy = (routeId: RouteId): RoutePolicy => { + return ROUTE_POLICIES[routeId] +} diff --git a/tests/mock-upstream.ts b/tests/mock-upstream.ts new file mode 100644 index 0000000..323a885 --- /dev/null +++ b/tests/mock-upstream.ts @@ -0,0 +1,160 @@ +/** + * Mock upstream server that simulates OpenAI / Azure OpenAI / OpenAI Responses APIs. + * Usage: npx tsx tests/mock-upstream.ts + * Listens on port 9999. + */ + +const PORT = 9999 + +// --- OpenAI Chat Completions (non-stream) --- +const openaiChatResponse = (model: string) => ({ + id: "chatcmpl-mock-001", + object: "chat.completion", + created: 1700000000, + model, + choices: [{ + index: 0, + message: { role: "assistant", content: "Mock response from " + model }, + finish_reason: "stop", + }], + usage: { prompt_tokens: 10, completion_tokens: 8, total_tokens: 18 }, +}) + +// --- OpenAI Chat Completions (stream) --- +const openaiChatStreamChunks = (model: string) => [ + `data: ${JSON.stringify({ id: "chatcmpl-mock-001", object: "chat.completion.chunk", model, choices: [{ index: 0, delta: { role: "assistant", content: "" }, finish_reason: null }], usage: null })}\n\n`, + `data: ${JSON.stringify({ id: "chatcmpl-mock-001", object: "chat.completion.chunk", model, choices: [{ index: 0, delta: { content: "Mock " }, finish_reason: null }], usage: null })}\n\n`, + `data: ${JSON.stringify({ id: "chatcmpl-mock-001", object: "chat.completion.chunk", model, choices: [{ index: 0, delta: { content: "stream " }, finish_reason: null }], usage: null })}\n\n`, + `data: ${JSON.stringify({ id: "chatcmpl-mock-001", object: "chat.completion.chunk", model, choices: [{ index: 0, delta: { content: "response" }, finish_reason: null }], usage: null })}\n\n`, + `data: ${JSON.stringify({ id: "chatcmpl-mock-001", object: "chat.completion.chunk", model, choices: [{ index: 0, delta: {}, finish_reason: "stop" }], usage: { prompt_tokens: 10, completion_tokens: 3, total_tokens: 13 } })}\n\n`, + `data: [DONE]\n\n`, +] + +// --- OpenAI Responses (non-stream) --- +const openaiResponsesResponse = (model: string) => ({ + id: "resp-mock-001", + object: "response", + created_at: 1700000000, + status: "completed", + model, + output: [{ + id: "msg-mock-001", + type: "message", + status: "completed", + content: [{ type: "output_text", text: "Mock responses output" }], + role: "assistant", + }], + usage: { input_tokens: 8, output_tokens: 5, total_tokens: 13 }, +}) + +// --- OpenAI Responses (stream) --- +const openaiResponsesStreamChunks = (model: string) => { + const respId = "resp-mock-001" + const msgId = "msg-mock-001" + return [ + `event: response.created\ndata: ${JSON.stringify({ type: "response.created", response: { id: respId, object: "response", model, status: "in_progress", output: [], usage: null } })}\n\n`, + `event: response.output_item.added\ndata: ${JSON.stringify({ type: "response.output_item.added", item: { id: msgId, type: "message", status: "in_progress", content: [], role: "assistant" }, output_index: 0 })}\n\n`, + `event: response.output_text.delta\ndata: ${JSON.stringify({ type: "response.output_text.delta", delta: "Mock ", item_id: msgId, output_index: 0, content_index: 0 })}\n\n`, + `event: response.output_text.delta\ndata: ${JSON.stringify({ type: "response.output_text.delta", delta: "stream", item_id: msgId, output_index: 0, content_index: 0 })}\n\n`, + `event: response.completed\ndata: ${JSON.stringify({ type: "response.completed", response: { id: respId, object: "response", model, status: "completed", output: [{ id: msgId, type: "message", status: "completed", content: [{ type: "output_text", text: "Mock stream" }], role: "assistant" }], usage: { input_tokens: 8, output_tokens: 4, total_tokens: 12 } } })}\n\n`, + ] +} + +// --- Claude Messages (non-stream) --- +const claudeMessagesResponse = (model: string) => ({ + id: "msg-mock-claude-001", + type: "message", + role: "assistant", + model, + content: [{ type: "text", text: "Mock Claude response" }], + stop_reason: "end_turn", + stop_sequence: null, + usage: { input_tokens: 12, output_tokens: 6 }, +}) + +// --- Claude Messages (stream) --- +const claudeMessagesStreamChunks = (model: string) => [ + `event: message_start\ndata: ${JSON.stringify({ type: "message_start", message: { id: "msg-mock-claude-001", type: "message", role: "assistant", model, content: [], stop_reason: null, usage: { input_tokens: 12, output_tokens: 0 } } })}\n\n`, + `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: 0, content_block: { type: "text", text: "" } })}\n\n`, + `event: content_block_delta\ndata: ${JSON.stringify({ type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "Mock Claude stream" } })}\n\n`, + `event: content_block_stop\ndata: ${JSON.stringify({ type: "content_block_stop", index: 0 })}\n\n`, + `event: message_delta\ndata: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "end_turn" }, usage: { output_tokens: 4 } })}\n\n`, + `event: message_stop\ndata: ${JSON.stringify({ type: "message_stop" })}\n\n`, +] + +const server = Bun.serve({ + port: PORT, + async fetch(req) { + const url = new URL(req.url) + const path = url.pathname + const body = req.method === "POST" ? await req.json().catch(() => ({})) : {} + const model = (body as any).model || "mock-model" + const isStream = (body as any).stream === true + + console.log(`[MOCK] ${req.method} ${path} model=${model} stream=${isStream}`) + + // OpenAI Chat Completions + if (path.endsWith("/chat/completions")) { + if (isStream) { + const chunks = openaiChatStreamChunks(model) + const stream = new ReadableStream({ + async start(controller) { + for (const chunk of chunks) { + controller.enqueue(new TextEncoder().encode(chunk)) + await new Promise(r => setTimeout(r, 50)) + } + controller.close() + } + }) + return new Response(stream, { + headers: { "content-type": "text/event-stream", "cache-control": "no-cache" }, + }) + } + return Response.json(openaiChatResponse(model)) + } + + // OpenAI Responses API + if (path.endsWith("/responses")) { + if (isStream) { + const chunks = openaiResponsesStreamChunks(model) + const stream = new ReadableStream({ + async start(controller) { + for (const chunk of chunks) { + controller.enqueue(new TextEncoder().encode(chunk)) + await new Promise(r => setTimeout(r, 50)) + } + controller.close() + } + }) + return new Response(stream, { + headers: { "content-type": "text/event-stream", "cache-control": "no-cache" }, + }) + } + return Response.json(openaiResponsesResponse(model)) + } + + // Claude Messages API + if (path.endsWith("/messages")) { + if (isStream) { + const chunks = claudeMessagesStreamChunks(model) + const stream = new ReadableStream({ + async start(controller) { + for (const chunk of chunks) { + controller.enqueue(new TextEncoder().encode(chunk)) + await new Promise(r => setTimeout(r, 50)) + } + controller.close() + } + }) + return new Response(stream, { + headers: { "content-type": "text/event-stream", "cache-control": "no-cache" }, + }) + } + return Response.json(claudeMessagesResponse(model)) + } + + return new Response("Mock: unknown route " + path, { status: 404 }) + }, +}) + +console.log(`[MOCK] Upstream mock server running on http://localhost:${PORT}`) diff --git a/tests/run-e2e.sh b/tests/run-e2e.sh new file mode 100755 index 0000000..4193404 --- /dev/null +++ b/tests/run-e2e.sh @@ -0,0 +1,207 @@ +#!/bin/bash +# E2E tests against mock upstream +# Prerequisites: mock-upstream running on :9999, wrangler dev on :8788 + +API="http://localhost:8788" +PASS=0 +FAIL=0 +TOTAL=0 + +assert() { + local name="$1" expected="$2" actual="$3" + TOTAL=$((TOTAL + 1)) + if echo "$actual" | grep -q "$expected"; then + echo " ✅ #$TOTAL $name" + PASS=$((PASS + 1)) + else + echo " ❌ #$TOTAL $name" + echo " expected: $expected" + echo " actual: $(echo "$actual" | head -1)" + FAIL=$((FAIL + 1)) + fi +} + +assert_http() { + local name="$1" expected_code="$2" expected_body="$3" actual="$4" + local code=$(echo "$actual" | tail -1) + local body=$(echo "$actual" | sed '$d') + TOTAL=$((TOTAL + 1)) + if [[ "$code" == "$expected_code" ]] && echo "$body" | grep -q "$expected_body"; then + echo " ✅ #$TOTAL $name [HTTP $code]" + PASS=$((PASS + 1)) + else + echo " ❌ #$TOTAL $name [HTTP $code]" + echo " expected: HTTP $expected_code + '$expected_body'" + echo " actual: $(echo "$body" | head -1)" + FAIL=$((FAIL + 1)) + fi +} + +do_get() { curl -s -w "\n%{http_code}" "$@"; } +do_post() { curl -s -w "\n%{http_code}" -X POST "$@"; } + +echo "" +echo "=========================================" +echo " E2E Tests with Mock Upstream" +echo "=========================================" + +# ---- /v1/models ---- +echo "" +echo "--- GET /v1/models ---" + +R=$(do_get "$API/v1/models") +assert_http "no key → empty list" "200" '"data":\[\]' "$R" + +R=$(do_get -H "Authorization: Bearer bad" "$API/v1/models") +assert_http "invalid key → 401" "401" "Invalid API key" "$R" + +R=$(do_get -H "Authorization: Bearer sk-mock-all" "$API/v1/models") +assert_http "full token → has gpt-4o" "200" "gpt-4o" "$R" + +R=$(do_get -H "x-api-key: sk-mock-openai" "$API/v1/models") +assert_http "limited token → has gpt-4 only" "200" "gpt-4" "$R" + +# ---- /v1/chat/completions errors ---- +echo "" +echo "--- POST /v1/chat/completions (errors) ---" + +R=$(do_post "$API/v1/chat/completions" -H "Content-Type: application/json" -d '{"model":"gpt-4o","messages":[{"role":"user","content":"hi"}]}') +assert_http "no auth → 401" "401" "Authorization header" "$R" + +R=$(do_post -H "Authorization: Bearer bad" "$API/v1/chat/completions" -H "Content-Type: application/json" -d '{"model":"gpt-4o","messages":[{"role":"user","content":"hi"}]}') +assert_http "invalid key → 401" "401" "Invalid API key" "$R" + +R=$(do_post -H "Authorization: Bearer sk-mock-all" "$API/v1/chat/completions" -H "Content-Type: application/json" -d 'not-json') +assert_http "invalid JSON → 400" "400" "Invalid JSON body" "$R" + +R=$(do_post -H "Authorization: Bearer sk-mock-all" "$API/v1/chat/completions" -H "Content-Type: application/json" -d '{"messages":[]}') +assert_http "no model → 400" "400" "Model is required" "$R" + +R=$(do_post -H "Authorization: Bearer sk-mock-all" "$API/v1/chat/completions" -H "Content-Type: application/json" -d '{"model":"nonexistent","messages":[]}') +assert_http "unmapped model → 400" "400" "Model not mapped" "$R" + +R=$(do_post -H "Authorization: Bearer sk-mock-openai" "$API/v1/chat/completions" -H "Content-Type: application/json" -d '{"model":"claude-3-opus","messages":[]}') +assert_http "limited token, wrong model → 400" "400" "Model not mapped" "$R" + +# ---- /v1/chat/completions via openai provider (non-stream) ---- +echo "" +echo "--- POST /v1/chat/completions (openai, non-stream) ---" + +R=$(do_post -H "Authorization: Bearer sk-mock-openai" "$API/v1/chat/completions" -H "Content-Type: application/json" -d '{"model":"gpt-4o","messages":[{"role":"user","content":"hi"}]}') +assert_http "openai non-stream → 200 + usage" "200" "Mock response" "$R" + +# ---- /v1/chat/completions via openai provider (stream) ---- +echo "" +echo "--- POST /v1/chat/completions (openai, stream) ---" + +R=$(do_post -H "Authorization: Bearer sk-mock-openai" "$API/v1/chat/completions" -H "Content-Type: application/json" -d '{"model":"gpt-4o","stream":true,"messages":[{"role":"user","content":"hi"}]}') +assert_http "openai stream → 200 + SSE" "200" "data:" "$R" + +# ---- /v1/chat/completions via azure-openai (non-stream) ---- +echo "" +echo "--- POST /v1/chat/completions (azure-openai, non-stream) ---" + +R=$(do_post -H "Authorization: Bearer sk-mock-all" "$API/v1/chat/completions" -H "Content-Type: application/json" -d '{"model":"gpt-4o-azure","messages":[{"role":"user","content":"hi"}]}') +assert_http "azure non-stream → 200" "200" "Mock response" "$R" + +# ---- /v1/chat/completions via azure-openai (stream) ---- +echo "" +echo "--- POST /v1/chat/completions (azure-openai, stream) ---" + +R=$(do_post -H "Authorization: Bearer sk-mock-all" "$API/v1/chat/completions" -H "Content-Type: application/json" -d '{"model":"gpt-4o-azure","stream":true,"messages":[{"role":"user","content":"hi"}]}') +assert_http "azure stream → 200 + SSE" "200" "data:" "$R" + +# ---- /v1/messages via claude (non-stream) ---- +echo "" +echo "--- POST /v1/messages (claude, non-stream) ---" + +R=$(do_post -H "x-api-key: sk-mock-all" "$API/v1/messages" -H "Content-Type: application/json" -d '{"model":"claude-3-opus","max_tokens":100,"messages":[{"role":"user","content":"hi"}]}') +assert_http "claude non-stream → 200" "200" "Mock Claude response" "$R" + +# ---- /v1/messages via claude (stream) ---- +echo "" +echo "--- POST /v1/messages (claude, stream) ---" + +R=$(do_post -H "x-api-key: sk-mock-all" "$API/v1/messages" -H "Content-Type: application/json" -d '{"model":"claude-3-opus","stream":true,"max_tokens":100,"messages":[{"role":"user","content":"hi"}]}') +assert_http "claude stream → 200 + SSE" "200" "event:" "$R" + +# ---- /v1/messages via claude-to-openai (non-stream) ---- +echo "" +echo "--- POST /v1/messages (claude-to-openai, non-stream) ---" + +R=$(do_post -H "x-api-key: sk-mock-all" "$API/v1/messages" -H "Content-Type: application/json" -d '{"model":"claude-via-openai","max_tokens":100,"messages":[{"role":"user","content":"hi"}]}') +assert_http "c2o non-stream → 200" "200" "message" "$R" + +# ---- /v1/responses errors ---- +echo "" +echo "--- POST /v1/responses (errors) ---" + +R=$(do_post "$API/v1/responses" -H "Content-Type: application/json" -d '{"model":"gpt-4o-resp","input":"hi"}') +assert_http "no auth → 401" "401" "Authorization header" "$R" + +R=$(do_post -H "Authorization: Bearer sk-mock-all" "$API/v1/responses" -H "Content-Type: application/json" -d '{"model":"gpt-4o","input":"hi"}') +assert_http "non-responses model → 400" "400" "Model not mapped" "$R" + +# ---- /v1/responses via openai-responses (non-stream) ---- +echo "" +echo "--- POST /v1/responses (openai-responses, non-stream) ---" + +R=$(do_post -H "Authorization: Bearer sk-mock-all" "$API/v1/responses" -H "Content-Type: application/json" -d '{"model":"gpt-4o-resp","input":"hi"}') +assert_http "responses non-stream → 200 + usage" "200" "Mock responses output" "$R" + +# ---- /v1/responses via openai-responses (stream) ---- +echo "" +echo "--- POST /v1/responses (openai-responses, stream) ---" + +R=$(do_post -H "Authorization: Bearer sk-mock-all" "$API/v1/responses" -H "Content-Type: application/json" -d '{"model":"gpt-4o-resp","stream":true,"input":"hi"}') +assert_http "responses stream → 200 + SSE" "200" "event:" "$R" + +# ---- /v1/responses via azure-openai-responses (non-stream) ---- +echo "" +echo "--- POST /v1/responses (azure-openai-responses, non-stream) ---" + +R=$(do_post -H "Authorization: Bearer sk-mock-all" "$API/v1/responses" -H "Content-Type: application/json" -d '{"model":"gpt-4o-azure-resp","input":"hi"}') +assert_http "azure responses non-stream → 200" "200" "Mock responses output" "$R" + +# ---- /v1/responses via azure-openai-responses (stream) ---- +echo "" +echo "--- POST /v1/responses (azure-openai-responses, stream) ---" + +R=$(do_post -H "Authorization: Bearer sk-mock-all" "$API/v1/responses" -H "Content-Type: application/json" -d '{"model":"gpt-4o-azure-resp","stream":true,"input":"hi"}') +assert_http "azure responses stream → 200 + SSE" "200" "event:" "$R" + +# ---- Quota ---- +echo "" +echo "--- Quota check ---" + +npx wrangler d1 execute awsl-one-api --local --command "UPDATE api_token SET usage = 99999999 WHERE key = 'sk-mock-all'" > /dev/null 2>&1 +R=$(do_post -H "Authorization: Bearer sk-mock-all" "$API/v1/chat/completions" -H "Content-Type: application/json" -d '{"model":"gpt-4o","messages":[]}') +assert_http "quota exceeded → 402" "402" "Quota exceeded" "$R" +npx wrangler d1 execute awsl-one-api --local --command "UPDATE api_token SET usage = 0 WHERE key = 'sk-mock-all'" > /dev/null 2>&1 + +# ---- Error recovery ---- +echo "" +echo "--- Error recovery ---" + +do_post -H "Authorization: Bearer sk-mock-all" "$API/v1/chat/completions" -H "Content-Type: application/json" -d 'bad' > /dev/null 2>&1 +R=$(do_post -H "Authorization: Bearer sk-mock-openai" "$API/v1/chat/completions" -H "Content-Type: application/json" -d '{"model":"gpt-4o","messages":[{"role":"user","content":"hi"}]}') +assert_http "recover after error → 200" "200" "Mock response" "$R" + +# ---- Summary ---- +echo "" +echo "=========================================" +echo " Results: $PASS/$TOTAL PASS, $FAIL FAIL" +echo "=========================================" + +# Cleanup test data +echo "" +echo "Cleaning up test data..." +for key in test-openai test-azure test-claude test-c2o test-responses test-azure-resp; do + npx wrangler d1 execute awsl-one-api --local --command "DELETE FROM channel_config WHERE key = '$key'" > /dev/null 2>&1 +done +for key in sk-mock-all sk-mock-openai sk-mock-quota; do + npx wrangler d1 execute awsl-one-api --local --command "DELETE FROM api_token WHERE key = '$key'" > /dev/null 2>&1 +done +echo "Cleanup done." + +exit $FAIL diff --git a/tests/setup-test-data.sh b/tests/setup-test-data.sh new file mode 100755 index 0000000..3914b25 --- /dev/null +++ b/tests/setup-test-data.sh @@ -0,0 +1,36 @@ +#!/bin/bash +# Setup test data in local D1 for mock upstream testing +# All channels point to http://localhost:9999 + +MOCK="http://localhost:9999" + +echo "Setting up test channels and tokens..." + +# Channel: openai type +npx wrangler d1 execute awsl-one-api --local --command "INSERT OR REPLACE INTO channel_config (key, value) VALUES ('test-openai', '{\"name\":\"test-openai\",\"type\":\"openai\",\"endpoint\":\"$MOCK\",\"api_key\":\"mock-key\",\"deployment_mapper\":{\"gpt-4o\":\"gpt-4o\",\"gpt-4\":\"gpt-4\"}}')" + +# Channel: azure-openai type +npx wrangler d1 execute awsl-one-api --local --command "INSERT OR REPLACE INTO channel_config (key, value) VALUES ('test-azure', '{\"name\":\"test-azure\",\"type\":\"azure-openai\",\"endpoint\":\"$MOCK\",\"api_key\":\"mock-key\",\"api_version\":\"2024-02-01\",\"deployment_mapper\":{\"gpt-4o-azure\":\"gpt-4o-deployment\"}}')" + +# Channel: claude type +npx wrangler d1 execute awsl-one-api --local --command "INSERT OR REPLACE INTO channel_config (key, value) VALUES ('test-claude', '{\"name\":\"test-claude\",\"type\":\"claude\",\"endpoint\":\"$MOCK\",\"api_key\":\"mock-key\",\"deployment_mapper\":{\"claude-3-opus\":\"claude-3-opus-20240229\"}}')" + +# Channel: claude-to-openai type +npx wrangler d1 execute awsl-one-api --local --command "INSERT OR REPLACE INTO channel_config (key, value) VALUES ('test-c2o', '{\"name\":\"test-c2o\",\"type\":\"claude-to-openai\",\"endpoint\":\"$MOCK\",\"api_key\":\"mock-key\",\"deployment_mapper\":{\"claude-via-openai\":\"claude-3-opus\"}}')" + +# Channel: openai-responses type +npx wrangler d1 execute awsl-one-api --local --command "INSERT OR REPLACE INTO channel_config (key, value) VALUES ('test-responses', '{\"name\":\"test-responses\",\"type\":\"openai-responses\",\"endpoint\":\"$MOCK\",\"api_key\":\"mock-key\",\"deployment_mapper\":{\"gpt-4o-resp\":\"gpt-4o\"}}')" + +# Channel: azure-openai-responses type +npx wrangler d1 execute awsl-one-api --local --command "INSERT OR REPLACE INTO channel_config (key, value) VALUES ('test-azure-resp', '{\"name\":\"test-azure-resp\",\"type\":\"azure-openai-responses\",\"endpoint\":\"$MOCK\",\"api_key\":\"mock-key\",\"api_version\":\"2025-01-01\",\"deployment_mapper\":{\"gpt-4o-azure-resp\":\"gpt-4o-deployment\"}}')" + +# Token: access to all test channels only +npx wrangler d1 execute awsl-one-api --local --command "INSERT OR REPLACE INTO api_token (key, value, usage) VALUES ('sk-mock-all', '{\"name\":\"mock-all\",\"channel_keys\":[\"test-openai\",\"test-azure\",\"test-claude\",\"test-c2o\",\"test-responses\",\"test-azure-resp\"],\"total_quota\":99999999}', 0)" + +# Token: limited to openai only +npx wrangler d1 execute awsl-one-api --local --command "INSERT OR REPLACE INTO api_token (key, value, usage) VALUES ('sk-mock-openai', '{\"name\":\"mock-openai\",\"channel_keys\":[\"test-openai\"],\"total_quota\":99999999}', 0)" + +# Token: quota almost full +npx wrangler d1 execute awsl-one-api --local --command "INSERT OR REPLACE INTO api_token (key, value, usage) VALUES ('sk-mock-quota', '{\"name\":\"mock-quota\",\"channel_keys\":[\"test-openai\"],\"total_quota\":100}', 99)" + +echo "Test data setup complete!"