chore: initialize clean metapi repository
This commit is contained in:
cita
@@ -0,0 +1,317 @@
|
||||
import Fastify, { type FastifyInstance } from 'fastify';
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
|
||||
const fetchMock = vi.fn();
|
||||
const selectChannelMock = vi.fn();
|
||||
const selectNextChannelMock = vi.fn();
|
||||
const recordSuccessMock = vi.fn();
|
||||
const recordFailureMock = vi.fn();
|
||||
const refreshModelsAndRebuildRoutesMock = vi.fn();
|
||||
const reportProxyAllFailedMock = vi.fn();
|
||||
const reportTokenExpiredMock = vi.fn();
|
||||
const estimateProxyCostMock = vi.fn(async (_arg?: any) => 0);
|
||||
const resolveProxyUsageWithSelfLogFallbackMock = vi.fn(async ({ usage }: any) => ({
|
||||
...usage,
|
||||
estimatedCostFromQuota: 0,
|
||||
recoveredFromSelfLog: false,
|
||||
}));
|
||||
const dbInsertMock = vi.fn((_arg?: any) => ({
|
||||
values: () => ({
|
||||
run: () => undefined,
|
||||
}),
|
||||
}));
|
||||
|
||||
vi.mock('undici', () => ({
|
||||
fetch: (...args: unknown[]) => fetchMock(...args),
|
||||
}));
|
||||
|
||||
vi.mock('../../services/tokenRouter.js', () => ({
|
||||
tokenRouter: {
|
||||
selectChannel: (...args: unknown[]) => selectChannelMock(...args),
|
||||
selectNextChannel: (...args: unknown[]) => selectNextChannelMock(...args),
|
||||
recordSuccess: (...args: unknown[]) => recordSuccessMock(...args),
|
||||
recordFailure: (...args: unknown[]) => recordFailureMock(...args),
|
||||
},
|
||||
}));
|
||||
|
||||
vi.mock('../../services/modelService.js', () => ({
|
||||
refreshModelsAndRebuildRoutes: (...args: unknown[]) => refreshModelsAndRebuildRoutesMock(...args),
|
||||
}));
|
||||
|
||||
vi.mock('../../services/alertService.js', () => ({
|
||||
reportProxyAllFailed: (...args: unknown[]) => reportProxyAllFailedMock(...args),
|
||||
reportTokenExpired: (...args: unknown[]) => reportTokenExpiredMock(...args),
|
||||
}));
|
||||
|
||||
vi.mock('../../services/alertRules.js', () => ({
|
||||
isTokenExpiredError: () => false,
|
||||
}));
|
||||
|
||||
vi.mock('../../services/modelPricingService.js', () => ({
|
||||
estimateProxyCost: (arg: any) => estimateProxyCostMock(arg),
|
||||
}));
|
||||
|
||||
vi.mock('../../services/proxyRetryPolicy.js', () => ({
|
||||
shouldRetryProxyRequest: () => false,
|
||||
}));
|
||||
|
||||
vi.mock('../../services/proxyUsageFallbackService.js', () => ({
|
||||
resolveProxyUsageWithSelfLogFallback: (arg: any) => resolveProxyUsageWithSelfLogFallbackMock(arg),
|
||||
}));
|
||||
|
||||
vi.mock('../../db/index.js', () => ({
|
||||
db: {
|
||||
insert: (arg: any) => dbInsertMock(arg),
|
||||
},
|
||||
schema: {
|
||||
proxyLogs: {},
|
||||
},
|
||||
}));
|
||||
|
||||
describe('chat proxy stream behavior', () => {
|
||||
let app: FastifyInstance;
|
||||
|
||||
beforeAll(async () => {
|
||||
const { chatProxyRoute, claudeMessagesProxyRoute } = await import('./chat.js');
|
||||
app = Fastify();
|
||||
await app.register(chatProxyRoute);
|
||||
await app.register(claudeMessagesProxyRoute);
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
fetchMock.mockReset();
|
||||
selectChannelMock.mockReset();
|
||||
selectNextChannelMock.mockReset();
|
||||
recordSuccessMock.mockReset();
|
||||
recordFailureMock.mockReset();
|
||||
refreshModelsAndRebuildRoutesMock.mockReset();
|
||||
reportProxyAllFailedMock.mockReset();
|
||||
reportTokenExpiredMock.mockReset();
|
||||
estimateProxyCostMock.mockClear();
|
||||
resolveProxyUsageWithSelfLogFallbackMock.mockClear();
|
||||
dbInsertMock.mockClear();
|
||||
|
||||
selectChannelMock.mockReturnValue({
|
||||
channel: { id: 11, routeId: 22 },
|
||||
site: { name: 'demo-site', url: 'https://upstream.example.com' },
|
||||
account: { id: 33, username: 'demo-user' },
|
||||
tokenName: 'default',
|
||||
tokenValue: 'sk-demo',
|
||||
actualModel: 'upstream-gpt',
|
||||
});
|
||||
selectNextChannelMock.mockReturnValue(null);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await app.close();
|
||||
});
|
||||
|
||||
it('converts non-SSE upstream streaming responses into SSE events', async () => {
|
||||
fetchMock.mockResolvedValue(new Response(JSON.stringify({
|
||||
id: 'chatcmpl-demo',
|
||||
object: 'chat.completion',
|
||||
created: 1_706_000_000,
|
||||
model: 'upstream-gpt',
|
||||
choices: [{
|
||||
index: 0,
|
||||
message: { role: 'assistant', content: 'hello from upstream' },
|
||||
finish_reason: 'stop',
|
||||
}],
|
||||
usage: { prompt_tokens: 11, completion_tokens: 7, total_tokens: 18 },
|
||||
}), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/chat/completions',
|
||||
payload: {
|
||||
model: 'gpt-4o-mini',
|
||||
stream: true,
|
||||
messages: [{ role: 'user', content: 'hi' }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.headers['content-type']).toContain('text/event-stream');
|
||||
expect(response.body).toContain('data: ');
|
||||
expect(response.body).toContain('"chat.completion.chunk"');
|
||||
expect(response.body).toContain('hello from upstream');
|
||||
expect(response.body).toContain('data: [DONE]');
|
||||
});
|
||||
|
||||
it('sets anti-buffering SSE headers for streamed chat responses', async () => {
|
||||
const encoder = new TextEncoder();
|
||||
const upstreamBody = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(encoder.encode('data: {"choices":[{"delta":{"content":"hello"}}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: [DONE]\n\n'));
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(upstreamBody, {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'text/event-stream; charset=utf-8' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/chat/completions',
|
||||
payload: {
|
||||
model: 'gpt-4o-mini',
|
||||
stream: true,
|
||||
messages: [{ role: 'user', content: 'hello' }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.headers['content-type']).toContain('text/event-stream');
|
||||
expect(response.headers['cache-control']).toContain('no-transform');
|
||||
expect(response.headers['x-accel-buffering']).toBe('no');
|
||||
expect(response.body).toContain('"chat.completion.chunk"');
|
||||
expect(response.body).toContain('"delta":{"role":"assistant","content":"hello"}');
|
||||
expect(response.body).toContain('data: [DONE]');
|
||||
});
|
||||
|
||||
it('normalizes anthropic-style SSE events into OpenAI chunks for clients like OpenWebUI', async () => {
|
||||
const encoder = new TextEncoder();
|
||||
const upstreamBody = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(encoder.encode('event: message_start\ndata: {"type":"message_start","message":{"id":"msg_123","model":"claude-opus-4-6"}}\n\n'));
|
||||
controller.enqueue(encoder.encode('event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":"hello"}}\n\n'));
|
||||
controller.enqueue(encoder.encode('event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn"}}\n\n'));
|
||||
controller.enqueue(encoder.encode('event: message_stop\ndata: {"type":"message_stop"}\n\n'));
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(upstreamBody, {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'text/event-stream; charset=utf-8' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/chat/completions',
|
||||
payload: {
|
||||
model: 'claude-opus-4-6',
|
||||
stream: true,
|
||||
messages: [{ role: 'user', content: 'who are you' }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.headers['content-type']).toContain('text/event-stream');
|
||||
expect(response.body).toContain('"chat.completion.chunk"');
|
||||
expect(response.body).toContain('"delta":{"content":"hello"}');
|
||||
expect(response.body).toContain('"finish_reason":"stop"');
|
||||
expect(response.body).toContain('data: [DONE]');
|
||||
});
|
||||
|
||||
it('emits OpenAI-compatible assistant starter chunk for anthropic message_start events', async () => {
|
||||
const encoder = new TextEncoder();
|
||||
const upstreamBody = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(encoder.encode('event: message_start\ndata: {"type":"message_start","message":{"id":"msg_compat","model":"claude-opus-4-6"}}\n\n'));
|
||||
controller.enqueue(encoder.encode('event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"type":"text_delta","text":"compat"}}\n\n'));
|
||||
controller.enqueue(encoder.encode('event: message_stop\ndata: {"type":"message_stop"}\n\n'));
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(upstreamBody, {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'text/event-stream; charset=utf-8' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/chat/completions',
|
||||
payload: {
|
||||
model: 'claude-opus-4-6',
|
||||
stream: true,
|
||||
messages: [{ role: 'user', content: 'compat test' }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.body).toContain('"delta":{"role":"assistant","content":""}');
|
||||
expect(response.body).toContain('"delta":{"content":"compat"}');
|
||||
expect(response.body).toContain('data: [DONE]');
|
||||
});
|
||||
|
||||
it('converts OpenAI non-stream responses into Claude message format on /v1/messages', async () => {
|
||||
fetchMock.mockResolvedValue(new Response(JSON.stringify({
|
||||
id: 'chatcmpl-upstream',
|
||||
object: 'chat.completion',
|
||||
created: 1_706_000_001,
|
||||
model: 'claude-opus-4-6',
|
||||
choices: [{
|
||||
index: 0,
|
||||
message: { role: 'assistant', content: 'hello from claude format' },
|
||||
finish_reason: 'stop',
|
||||
}],
|
||||
usage: { prompt_tokens: 120, completion_tokens: 16, total_tokens: 136 },
|
||||
}), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/messages',
|
||||
payload: {
|
||||
model: 'claude-opus-4-6',
|
||||
max_tokens: 256,
|
||||
messages: [{ role: 'user', content: 'hello' }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
const body = response.json();
|
||||
expect(body.type).toBe('message');
|
||||
expect(body.role).toBe('assistant');
|
||||
expect(body.model).toBe('claude-opus-4-6');
|
||||
expect(body.content?.[0]?.type).toBe('text');
|
||||
expect(body.content?.[0]?.text).toContain('hello from claude format');
|
||||
expect(body.stop_reason).toBe('end_turn');
|
||||
});
|
||||
|
||||
it('converts OpenAI SSE chunks into Claude stream events on /v1/messages', async () => {
|
||||
const encoder = new TextEncoder();
|
||||
const upstreamBody = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-1","model":"claude-opus-4-6","choices":[{"delta":{"role":"assistant"},"finish_reason":null}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-1","model":"claude-opus-4-6","choices":[{"delta":{"content":"hello"},"finish_reason":null}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: {"id":"chatcmpl-1","model":"claude-opus-4-6","choices":[{"delta":{},"finish_reason":"stop"}]}\n\n'));
|
||||
controller.enqueue(encoder.encode('data: [DONE]\n\n'));
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(upstreamBody, {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'text/event-stream; charset=utf-8' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/messages',
|
||||
payload: {
|
||||
model: 'claude-opus-4-6',
|
||||
stream: true,
|
||||
max_tokens: 256,
|
||||
messages: [{ role: 'user', content: 'hello' }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(200);
|
||||
expect(response.headers['content-type']).toContain('text/event-stream');
|
||||
expect(response.body).toContain('event: message_start');
|
||||
expect(response.body).toContain('event: content_block_delta');
|
||||
expect(response.body).toContain('\"text\":\"hello\"');
|
||||
expect(response.body).toContain('event: message_stop');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,468 @@
|
||||
import { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify';
|
||||
import { tokenRouter } from '../../services/tokenRouter.js';
|
||||
import { db, schema } from '../../db/index.js';
|
||||
import { fetch } from 'undici';
|
||||
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
|
||||
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
|
||||
import { isTokenExpiredError } from '../../services/alertRules.js';
|
||||
import { estimateProxyCost } from '../../services/modelPricingService.js';
|
||||
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
|
||||
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
|
||||
import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js';
|
||||
import {
|
||||
type DownstreamFormat,
|
||||
createStreamTransformContext,
|
||||
createClaudeDownstreamContext,
|
||||
parseDownstreamChatRequest,
|
||||
pullSseEventsWithDone,
|
||||
normalizeUpstreamStreamEvent,
|
||||
serializeNormalizedStreamEvent,
|
||||
serializeStreamDone,
|
||||
normalizeUpstreamFinalResponse,
|
||||
serializeFinalResponse,
|
||||
buildSyntheticOpenAiChunks,
|
||||
} from './chatFormats.js';
|
||||
|
||||
const MAX_RETRIES = 2;
|
||||
|
||||
export async function chatProxyRoute(app: FastifyInstance) {
|
||||
app.post('/v1/chat/completions', async (request: FastifyRequest, reply: FastifyReply) =>
|
||||
handleChatProxyRequest(request, reply, 'openai'));
|
||||
}
|
||||
|
||||
export async function claudeMessagesProxyRoute(app: FastifyInstance) {
|
||||
app.post('/v1/messages', async (request: FastifyRequest, reply: FastifyReply) =>
|
||||
handleChatProxyRequest(request, reply, 'claude'));
|
||||
}
|
||||
|
||||
async function handleChatProxyRequest(
|
||||
request: FastifyRequest,
|
||||
reply: FastifyReply,
|
||||
downstreamFormat: DownstreamFormat,
|
||||
) {
|
||||
const parsedRequest = parseDownstreamChatRequest(request.body, downstreamFormat);
|
||||
if (parsedRequest.error) {
|
||||
return reply.code(parsedRequest.error.statusCode).send(parsedRequest.error.payload);
|
||||
}
|
||||
|
||||
const { requestedModel, isStream, upstreamBody } = parsedRequest.value!;
|
||||
|
||||
const excludeChannelIds: number[] = [];
|
||||
let retryCount = 0;
|
||||
|
||||
while (retryCount <= MAX_RETRIES) {
|
||||
let selected = retryCount === 0
|
||||
? tokenRouter.selectChannel(requestedModel)
|
||||
: tokenRouter.selectNextChannel(requestedModel, excludeChannelIds);
|
||||
|
||||
if (!selected && retryCount === 0) {
|
||||
await refreshModelsAndRebuildRoutes();
|
||||
selected = tokenRouter.selectChannel(requestedModel);
|
||||
}
|
||||
|
||||
if (!selected) {
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: 'No available channels after retries',
|
||||
});
|
||||
return reply.code(503).send({
|
||||
error: { message: 'No available channels for this model', type: 'server_error' },
|
||||
});
|
||||
}
|
||||
|
||||
excludeChannelIds.push(selected.channel.id);
|
||||
|
||||
const targetUrl = `${selected.site.url}/v1/chat/completions`;
|
||||
const forwardBody = {
|
||||
...upstreamBody,
|
||||
model: selected.actualModel,
|
||||
stream: isStream,
|
||||
};
|
||||
const startTime = Date.now();
|
||||
|
||||
try {
|
||||
const upstream = await fetch(targetUrl, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
Authorization: `Bearer ${selected.tokenValue}`,
|
||||
},
|
||||
body: JSON.stringify(forwardBody),
|
||||
});
|
||||
|
||||
if (!upstream.ok) {
|
||||
const errText = await upstream.text().catch(() => 'unknown error');
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, errText, retryCount);
|
||||
|
||||
if (isTokenExpiredError({ status: upstream.status, message: errText })) {
|
||||
await reportTokenExpired({
|
||||
accountId: selected.account.id,
|
||||
username: selected.account.username,
|
||||
siteName: selected.site.name,
|
||||
detail: `HTTP ${upstream.status}`,
|
||||
});
|
||||
}
|
||||
|
||||
if (shouldRetryProxyRequest(upstream.status, errText) && retryCount < MAX_RETRIES) {
|
||||
retryCount += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: `upstream returned HTTP ${upstream.status}`,
|
||||
});
|
||||
|
||||
return reply.code(upstream.status).send({
|
||||
error: { message: errText, type: 'upstream_error' },
|
||||
});
|
||||
}
|
||||
|
||||
const modelName = selected.actualModel || requestedModel;
|
||||
|
||||
if (isStream) {
|
||||
reply.hijack();
|
||||
reply.raw.statusCode = 200;
|
||||
reply.raw.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
|
||||
reply.raw.setHeader('Cache-Control', 'no-cache, no-transform');
|
||||
reply.raw.setHeader('Connection', 'keep-alive');
|
||||
reply.raw.setHeader('X-Accel-Buffering', 'no');
|
||||
|
||||
const streamContext = createStreamTransformContext(modelName);
|
||||
const claudeContext = createClaudeDownstreamContext();
|
||||
let parsedUsage = { promptTokens: 0, completionTokens: 0, totalTokens: 0 };
|
||||
|
||||
const writeLines = (lines: string[]) => {
|
||||
for (const line of lines) {
|
||||
reply.raw.write(line);
|
||||
}
|
||||
};
|
||||
|
||||
const writeDone = () => {
|
||||
writeLines(serializeStreamDone(downstreamFormat, streamContext, claudeContext));
|
||||
};
|
||||
|
||||
const emitNormalizedFinalAsStream = (upstreamData: unknown, fallbackText = '') => {
|
||||
const normalizedFinal = normalizeUpstreamFinalResponse(upstreamData, modelName, fallbackText);
|
||||
streamContext.id = normalizedFinal.id;
|
||||
streamContext.model = normalizedFinal.model;
|
||||
streamContext.created = normalizedFinal.created;
|
||||
|
||||
if (downstreamFormat === 'openai') {
|
||||
const syntheticChunks = buildSyntheticOpenAiChunks(normalizedFinal);
|
||||
for (const chunk of syntheticChunks) {
|
||||
reply.raw.write(`data: ${JSON.stringify(chunk)}\n\n`);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
writeLines(serializeNormalizedStreamEvent('claude', { role: 'assistant' }, streamContext, claudeContext));
|
||||
|
||||
const combinedText = [normalizedFinal.reasoningContent, normalizedFinal.content]
|
||||
.filter((item) => typeof item === 'string' && item.trim().length > 0)
|
||||
.join('\n\n');
|
||||
|
||||
if (combinedText) {
|
||||
writeLines(serializeNormalizedStreamEvent('claude', {
|
||||
contentDelta: combinedText,
|
||||
}, streamContext, claudeContext));
|
||||
}
|
||||
|
||||
writeLines(serializeNormalizedStreamEvent('claude', {
|
||||
finishReason: normalizedFinal.finishReason,
|
||||
}, streamContext, claudeContext));
|
||||
};
|
||||
|
||||
const upstreamContentType = (upstream.headers.get('content-type') || '').toLowerCase();
|
||||
if (!upstreamContentType.includes('text/event-stream')) {
|
||||
const fallbackText = await upstream.text();
|
||||
let fallbackData: unknown = null;
|
||||
try {
|
||||
fallbackData = JSON.parse(fallbackText);
|
||||
} catch {
|
||||
fallbackData = fallbackText;
|
||||
}
|
||||
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(fallbackData));
|
||||
emitNormalizedFinalAsStream(fallbackData, fallbackText);
|
||||
writeDone();
|
||||
reply.raw.end();
|
||||
|
||||
const latency = Date.now() - startTime;
|
||||
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
tokenValue: selected.tokenValue,
|
||||
tokenName: selected.tokenName,
|
||||
modelName,
|
||||
requestStartedAtMs: startTime,
|
||||
requestEndedAtMs: startTime + latency,
|
||||
localLatencyMs: latency,
|
||||
usage: {
|
||||
promptTokens: parsedUsage.promptTokens,
|
||||
completionTokens: parsedUsage.completionTokens,
|
||||
totalTokens: parsedUsage.totalTokens,
|
||||
},
|
||||
});
|
||||
|
||||
let estimatedCost = await estimateProxyCost({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
modelName,
|
||||
promptTokens: resolvedUsage.promptTokens,
|
||||
completionTokens: resolvedUsage.completionTokens,
|
||||
totalTokens: resolvedUsage.totalTokens,
|
||||
});
|
||||
|
||||
if (resolvedUsage.estimatedCostFromQuota > 0 && (resolvedUsage.recoveredFromSelfLog || estimatedCost <= 0)) {
|
||||
estimatedCost = resolvedUsage.estimatedCostFromQuota;
|
||||
}
|
||||
|
||||
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
|
||||
logProxy(
|
||||
selected,
|
||||
requestedModel,
|
||||
'success',
|
||||
200,
|
||||
latency,
|
||||
null,
|
||||
retryCount,
|
||||
resolvedUsage.promptTokens,
|
||||
resolvedUsage.completionTokens,
|
||||
resolvedUsage.totalTokens,
|
||||
estimatedCost,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const reader = upstream.body?.getReader();
|
||||
if (!reader) {
|
||||
writeDone();
|
||||
reply.raw.end();
|
||||
return;
|
||||
}
|
||||
|
||||
const decoder = new TextDecoder();
|
||||
let sseBuffer = '';
|
||||
|
||||
const consumeSseBuffer = (incoming: string): string => {
|
||||
const pulled = pullSseEventsWithDone(incoming);
|
||||
for (const eventBlock of pulled.events) {
|
||||
if (eventBlock.data === '[DONE]') {
|
||||
writeDone();
|
||||
continue;
|
||||
}
|
||||
|
||||
let parsedPayload: unknown = null;
|
||||
try {
|
||||
parsedPayload = JSON.parse(eventBlock.data);
|
||||
} catch {
|
||||
parsedPayload = null;
|
||||
}
|
||||
|
||||
if (parsedPayload && typeof parsedPayload === 'object') {
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(parsedPayload));
|
||||
const normalizedEvent = normalizeUpstreamStreamEvent(parsedPayload, streamContext, modelName);
|
||||
writeLines(serializeNormalizedStreamEvent(
|
||||
downstreamFormat,
|
||||
normalizedEvent,
|
||||
streamContext,
|
||||
claudeContext,
|
||||
));
|
||||
continue;
|
||||
}
|
||||
|
||||
if (downstreamFormat === 'openai') {
|
||||
reply.raw.write(`data: ${eventBlock.data}\n\n`);
|
||||
} else {
|
||||
writeLines(serializeNormalizedStreamEvent('claude', {
|
||||
contentDelta: eventBlock.data,
|
||||
}, streamContext, claudeContext));
|
||||
}
|
||||
}
|
||||
|
||||
return pulled.rest;
|
||||
};
|
||||
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
if (!value) continue;
|
||||
|
||||
sseBuffer += decoder.decode(value, { stream: true });
|
||||
sseBuffer = consumeSseBuffer(sseBuffer);
|
||||
}
|
||||
|
||||
sseBuffer += decoder.decode();
|
||||
if (sseBuffer.trim().length > 0) {
|
||||
sseBuffer = consumeSseBuffer(`${sseBuffer}\n\n`);
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
writeDone();
|
||||
reply.raw.end();
|
||||
}
|
||||
|
||||
const latency = Date.now() - startTime;
|
||||
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
tokenValue: selected.tokenValue,
|
||||
tokenName: selected.tokenName,
|
||||
modelName,
|
||||
requestStartedAtMs: startTime,
|
||||
requestEndedAtMs: startTime + latency,
|
||||
localLatencyMs: latency,
|
||||
usage: {
|
||||
promptTokens: parsedUsage.promptTokens,
|
||||
completionTokens: parsedUsage.completionTokens,
|
||||
totalTokens: parsedUsage.totalTokens,
|
||||
},
|
||||
});
|
||||
|
||||
let estimatedCost = await estimateProxyCost({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
modelName,
|
||||
promptTokens: resolvedUsage.promptTokens,
|
||||
completionTokens: resolvedUsage.completionTokens,
|
||||
totalTokens: resolvedUsage.totalTokens,
|
||||
});
|
||||
|
||||
if (resolvedUsage.estimatedCostFromQuota > 0 && (resolvedUsage.recoveredFromSelfLog || estimatedCost <= 0)) {
|
||||
estimatedCost = resolvedUsage.estimatedCostFromQuota;
|
||||
}
|
||||
|
||||
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
|
||||
logProxy(
|
||||
selected,
|
||||
requestedModel,
|
||||
'success',
|
||||
200,
|
||||
latency,
|
||||
null,
|
||||
retryCount,
|
||||
resolvedUsage.promptTokens,
|
||||
resolvedUsage.completionTokens,
|
||||
resolvedUsage.totalTokens,
|
||||
estimatedCost,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const rawText = await upstream.text();
|
||||
let upstreamData: unknown = rawText;
|
||||
try {
|
||||
upstreamData = JSON.parse(rawText);
|
||||
} catch {
|
||||
upstreamData = rawText;
|
||||
}
|
||||
|
||||
const latency = Date.now() - startTime;
|
||||
const parsedUsage = parseProxyUsage(upstreamData);
|
||||
const normalizedFinal = normalizeUpstreamFinalResponse(upstreamData, modelName, rawText);
|
||||
const downstreamResponse = serializeFinalResponse(downstreamFormat, normalizedFinal, parsedUsage);
|
||||
|
||||
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
tokenValue: selected.tokenValue,
|
||||
tokenName: selected.tokenName,
|
||||
modelName,
|
||||
requestStartedAtMs: startTime,
|
||||
requestEndedAtMs: startTime + latency,
|
||||
localLatencyMs: latency,
|
||||
usage: {
|
||||
promptTokens: parsedUsage.promptTokens,
|
||||
completionTokens: parsedUsage.completionTokens,
|
||||
totalTokens: parsedUsage.totalTokens,
|
||||
},
|
||||
});
|
||||
|
||||
let estimatedCost = await estimateProxyCost({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
modelName,
|
||||
promptTokens: resolvedUsage.promptTokens,
|
||||
completionTokens: resolvedUsage.completionTokens,
|
||||
totalTokens: resolvedUsage.totalTokens,
|
||||
});
|
||||
|
||||
if (resolvedUsage.estimatedCostFromQuota > 0 && (resolvedUsage.recoveredFromSelfLog || estimatedCost <= 0)) {
|
||||
estimatedCost = resolvedUsage.estimatedCostFromQuota;
|
||||
}
|
||||
|
||||
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
|
||||
logProxy(
|
||||
selected,
|
||||
requestedModel,
|
||||
'success',
|
||||
200,
|
||||
latency,
|
||||
null,
|
||||
retryCount,
|
||||
resolvedUsage.promptTokens,
|
||||
resolvedUsage.completionTokens,
|
||||
resolvedUsage.totalTokens,
|
||||
estimatedCost,
|
||||
);
|
||||
|
||||
return reply.send(downstreamResponse);
|
||||
} catch (err: any) {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err?.message || 'network error', retryCount);
|
||||
|
||||
if (retryCount < MAX_RETRIES) {
|
||||
retryCount += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: err?.message || 'network failure',
|
||||
});
|
||||
|
||||
return reply.code(502).send({
|
||||
error: {
|
||||
message: `Upstream error: ${err?.message || 'network failure'}`,
|
||||
type: 'upstream_error',
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function logProxy(
|
||||
selected: any,
|
||||
modelRequested: string,
|
||||
status: string,
|
||||
httpStatus: number,
|
||||
latencyMs: number,
|
||||
errorMessage: string | null,
|
||||
retryCount: number,
|
||||
promptTokens = 0,
|
||||
completionTokens = 0,
|
||||
totalTokens = 0,
|
||||
estimatedCost = 0,
|
||||
) {
|
||||
try {
|
||||
db.insert(schema.proxyLogs).values({
|
||||
routeId: selected.channel.routeId,
|
||||
channelId: selected.channel.id,
|
||||
accountId: selected.account.id,
|
||||
modelRequested,
|
||||
modelActual: selected.actualModel,
|
||||
status,
|
||||
httpStatus,
|
||||
latencyMs,
|
||||
promptTokens,
|
||||
completionTokens,
|
||||
totalTokens,
|
||||
estimatedCost,
|
||||
errorMessage,
|
||||
retryCount,
|
||||
}).run();
|
||||
} catch {}
|
||||
}
|
||||
@@ -0,0 +1,846 @@
|
||||
export type DownstreamFormat = 'openai' | 'claude';
|
||||
|
||||
export type ParsedSseEvent = {
|
||||
event: string;
|
||||
data: string;
|
||||
};
|
||||
|
||||
export type StreamTransformContext = {
|
||||
id: string;
|
||||
model: string;
|
||||
created: number;
|
||||
roleSent: boolean;
|
||||
doneSent: boolean;
|
||||
};
|
||||
|
||||
export type ClaudeDownstreamContext = {
|
||||
messageStarted: boolean;
|
||||
contentBlockStarted: boolean;
|
||||
doneSent: boolean;
|
||||
};
|
||||
|
||||
export type NormalizedStreamEvent = {
|
||||
role?: 'assistant';
|
||||
contentDelta?: string;
|
||||
reasoningDelta?: string;
|
||||
finishReason?: string | null;
|
||||
done?: boolean;
|
||||
};
|
||||
|
||||
export type NormalizedFinalResponse = {
|
||||
id: string;
|
||||
model: string;
|
||||
created: number;
|
||||
content: string;
|
||||
reasoningContent: string;
|
||||
finishReason: string;
|
||||
};
|
||||
|
||||
export type ParsedDownstreamChatRequest = {
|
||||
requestedModel: string;
|
||||
isStream: boolean;
|
||||
upstreamBody: Record<string, unknown>;
|
||||
};
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return !!value && typeof value === 'object';
|
||||
}
|
||||
|
||||
function isNonEmptyString(value: unknown): value is string {
|
||||
return typeof value === 'string' && value.trim().length > 0;
|
||||
}
|
||||
|
||||
function pickFiniteNumber(value: unknown): number | undefined {
|
||||
return typeof value === 'number' && Number.isFinite(value) ? value : undefined;
|
||||
}
|
||||
|
||||
function ensureIntegerTimestamp(value: unknown, fallback: number): number {
|
||||
const n = Number(value);
|
||||
if (!Number.isFinite(n) || n <= 0) return fallback;
|
||||
return Math.trunc(n);
|
||||
}
|
||||
|
||||
function joinNonEmpty(parts: string[]): string {
|
||||
return parts.map((item) => item.trim()).filter((item) => item.length > 0).join('\n\n');
|
||||
}
|
||||
|
||||
function textFromPart(part: unknown): string {
|
||||
if (typeof part === 'string') return part;
|
||||
if (!isRecord(part)) return '';
|
||||
|
||||
if (typeof part.text === 'string') return part.text;
|
||||
if (typeof part.content === 'string') return part.content;
|
||||
if (typeof part.output_text === 'string') return part.output_text;
|
||||
if (typeof part.completion === 'string') return part.completion;
|
||||
if (typeof part.partial_json === 'string') return part.partial_json;
|
||||
if (typeof part.reasoning_content === 'string') return part.reasoning_content;
|
||||
if (typeof part.reasoning === 'string') return part.reasoning;
|
||||
|
||||
if (Array.isArray(part.content)) {
|
||||
return part.content.map((item) => textFromPart(item)).join('');
|
||||
}
|
||||
|
||||
if (isRecord(part.delta)) {
|
||||
const fromDelta = textFromPart(part.delta);
|
||||
if (fromDelta) return fromDelta;
|
||||
}
|
||||
|
||||
return '';
|
||||
}
|
||||
|
||||
function extractTextAndReasoning(value: unknown): { content: string; reasoning: string } {
|
||||
if (typeof value === 'string') return { content: value, reasoning: '' };
|
||||
if (Array.isArray(value)) {
|
||||
const contentParts: string[] = [];
|
||||
const reasoningParts: string[] = [];
|
||||
for (const item of value) {
|
||||
if (typeof item === 'string') {
|
||||
contentParts.push(item);
|
||||
continue;
|
||||
}
|
||||
if (!isRecord(item)) continue;
|
||||
const type = typeof item.type === 'string' ? item.type : '';
|
||||
|
||||
if (type === 'thinking' && typeof item.thinking === 'string') {
|
||||
reasoningParts.push(item.thinking);
|
||||
continue;
|
||||
}
|
||||
if (type === 'thinking_delta' && typeof item.text === 'string') {
|
||||
reasoningParts.push(item.text);
|
||||
continue;
|
||||
}
|
||||
if (typeof item.thought === 'boolean' && item.thought && typeof item.text === 'string') {
|
||||
reasoningParts.push(item.text);
|
||||
continue;
|
||||
}
|
||||
|
||||
const text = textFromPart(item);
|
||||
if (text) contentParts.push(text);
|
||||
}
|
||||
|
||||
return {
|
||||
content: contentParts.join(''),
|
||||
reasoning: reasoningParts.join(''),
|
||||
};
|
||||
}
|
||||
|
||||
if (!isRecord(value)) return { content: '', reasoning: '' };
|
||||
|
||||
if (Array.isArray(value.parts)) {
|
||||
return extractTextAndReasoning(value.parts);
|
||||
}
|
||||
|
||||
return {
|
||||
content: textFromPart(value),
|
||||
reasoning: '',
|
||||
};
|
||||
}
|
||||
|
||||
export function normalizeStopReason(raw: unknown): string | null {
|
||||
const value = typeof raw === 'string' ? raw.trim().toLowerCase() : '';
|
||||
if (!value) return null;
|
||||
|
||||
if (
|
||||
value === 'end_turn'
|
||||
|| value === 'stop'
|
||||
|| value === 'end'
|
||||
|| value === 'eos'
|
||||
|| value === 'finished'
|
||||
|| value === 'stop_sequence'
|
||||
) {
|
||||
return 'stop';
|
||||
}
|
||||
|
||||
if (
|
||||
value === 'max_tokens'
|
||||
|| value === 'length'
|
||||
|| value === 'max_output_tokens'
|
||||
|| value === 'max_tokens_exceeded'
|
||||
|| value.includes('max')
|
||||
) {
|
||||
return 'length';
|
||||
}
|
||||
|
||||
if (value === 'tool_use' || value === 'tool_calls' || value.includes('tool')) {
|
||||
return 'tool_calls';
|
||||
}
|
||||
|
||||
return 'stop';
|
||||
}
|
||||
|
||||
export function toClaudeStopReason(finishReason: string | null | undefined): string {
|
||||
const value = normalizeStopReason(finishReason);
|
||||
if (value === 'length') return 'max_tokens';
|
||||
if (value === 'tool_calls') return 'tool_use';
|
||||
return 'end_turn';
|
||||
}
|
||||
|
||||
export function createStreamTransformContext(modelName: string): StreamTransformContext {
|
||||
return {
|
||||
id: `chatcmpl-meta-${Date.now()}`,
|
||||
model: modelName,
|
||||
created: Math.floor(Date.now() / 1000),
|
||||
roleSent: false,
|
||||
doneSent: false,
|
||||
};
|
||||
}
|
||||
|
||||
export function createClaudeDownstreamContext(): ClaudeDownstreamContext {
|
||||
return {
|
||||
messageStarted: false,
|
||||
contentBlockStarted: false,
|
||||
doneSent: false,
|
||||
};
|
||||
}
|
||||
|
||||
function buildClaudeMessageId(sourceId: string): string {
|
||||
if (sourceId.startsWith('msg_')) return sourceId;
|
||||
const sanitized = sourceId.replace(/[^A-Za-z0-9_-]/g, '_');
|
||||
return `msg_${sanitized || Date.now()}`;
|
||||
}
|
||||
|
||||
function serializeSse(event: string, data: unknown): string {
|
||||
const payload = typeof data === 'string' ? data : JSON.stringify(data);
|
||||
if (event) {
|
||||
return `event: ${event}\ndata: ${payload}\n\n`;
|
||||
}
|
||||
return `data: ${payload}\n\n`;
|
||||
}
|
||||
|
||||
function extractAssistantContent(choice: any): string {
|
||||
const messageContent = choice?.message?.content;
|
||||
const parsedMessage = extractTextAndReasoning(messageContent).content;
|
||||
if (parsedMessage) return parsedMessage;
|
||||
|
||||
const content = extractTextAndReasoning(choice?.content).content;
|
||||
if (content) return content;
|
||||
|
||||
if (typeof choice?.text === 'string' && choice.text.length > 0) return choice.text;
|
||||
if (typeof choice?.completion === 'string' && choice.completion.length > 0) return choice.completion;
|
||||
if (typeof choice?.output_text === 'string' && choice.output_text.length > 0) return choice.output_text;
|
||||
if (typeof choice?.delta?.content === 'string' && choice.delta.content.length > 0) return choice.delta.content;
|
||||
|
||||
return '';
|
||||
}
|
||||
|
||||
function extractAssistantReasoning(choice: any): string {
|
||||
const message = choice?.message || {};
|
||||
const direct = [
|
||||
message.reasoning_content,
|
||||
message.reasoning,
|
||||
choice?.reasoning_content,
|
||||
choice?.reasoning,
|
||||
].find((item) => typeof item === 'string' && item.length > 0);
|
||||
|
||||
if (typeof direct === 'string') return direct;
|
||||
|
||||
const parsed = extractTextAndReasoning(message.content);
|
||||
if (parsed.reasoning) return parsed.reasoning;
|
||||
|
||||
const nested = extractTextAndReasoning(choice?.content).reasoning;
|
||||
if (nested) return nested;
|
||||
|
||||
return '';
|
||||
}
|
||||
|
||||
function parseClaudeMessageContent(content: unknown): string {
|
||||
return extractTextAndReasoning(content).content;
|
||||
}
|
||||
|
||||
function convertClaudeRequestToOpenAiBody(body: Record<string, unknown>): {
|
||||
model: string;
|
||||
stream: boolean;
|
||||
messages: Array<{ role: string; content: string }>;
|
||||
payload: Record<string, unknown>;
|
||||
} {
|
||||
const model = typeof body.model === 'string' ? body.model.trim() : '';
|
||||
const stream = body.stream === true;
|
||||
|
||||
const messages: Array<{ role: string; content: string }> = [];
|
||||
|
||||
const appendMessage = (role: string, content: unknown) => {
|
||||
const text = parseClaudeMessageContent(content);
|
||||
if (!text) return;
|
||||
messages.push({ role, content: text });
|
||||
};
|
||||
|
||||
const system = body.system;
|
||||
if (typeof system === 'string') {
|
||||
appendMessage('system', system);
|
||||
} else if (Array.isArray(system)) {
|
||||
const merged = system.map((item) => parseClaudeMessageContent(item)).filter((item) => item.length > 0).join('\n\n');
|
||||
if (merged) appendMessage('system', merged);
|
||||
}
|
||||
|
||||
const rawMessages = Array.isArray(body.messages) ? body.messages : [];
|
||||
for (const message of rawMessages) {
|
||||
if (!isRecord(message)) continue;
|
||||
const role = typeof message.role === 'string' ? message.role : 'user';
|
||||
const mappedRole = role === 'assistant' || role === 'system' ? role : 'user';
|
||||
appendMessage(mappedRole, message.content);
|
||||
}
|
||||
|
||||
const payload: Record<string, unknown> = {
|
||||
model,
|
||||
stream,
|
||||
messages,
|
||||
};
|
||||
|
||||
const temperature = pickFiniteNumber(body.temperature);
|
||||
if (temperature !== undefined) payload.temperature = temperature;
|
||||
|
||||
const topP = pickFiniteNumber(body.top_p);
|
||||
if (topP !== undefined) payload.top_p = topP;
|
||||
|
||||
const maxTokens = pickFiniteNumber(body.max_tokens);
|
||||
if (maxTokens !== undefined) {
|
||||
payload.max_tokens = maxTokens;
|
||||
} else {
|
||||
payload.max_tokens = 4096;
|
||||
}
|
||||
|
||||
if (Array.isArray(body.stop_sequences) && body.stop_sequences.length > 0) {
|
||||
payload.stop = body.stop_sequences;
|
||||
}
|
||||
|
||||
if (body.tools !== undefined) payload.tools = body.tools;
|
||||
if (body.tool_choice !== undefined) payload.tool_choice = body.tool_choice;
|
||||
|
||||
return { model, stream, messages, payload };
|
||||
}
|
||||
|
||||
export function parseDownstreamChatRequest(
|
||||
body: unknown,
|
||||
format: DownstreamFormat,
|
||||
): { value?: ParsedDownstreamChatRequest; error?: { statusCode: number; payload: unknown } } {
|
||||
const raw = isRecord(body) ? body : {};
|
||||
|
||||
if (format === 'claude') {
|
||||
const converted = convertClaudeRequestToOpenAiBody(raw);
|
||||
if (!converted.model) {
|
||||
return {
|
||||
error: {
|
||||
statusCode: 400,
|
||||
payload: { error: { message: 'model is required', type: 'invalid_request_error' } },
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
if (converted.messages.length <= 0) {
|
||||
return {
|
||||
error: {
|
||||
statusCode: 400,
|
||||
payload: { error: { message: 'messages is required', type: 'invalid_request_error' } },
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
value: {
|
||||
requestedModel: converted.model,
|
||||
isStream: converted.stream,
|
||||
upstreamBody: converted.payload,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const model = typeof raw.model === 'string' ? raw.model.trim() : '';
|
||||
if (!model) {
|
||||
return {
|
||||
error: {
|
||||
statusCode: 400,
|
||||
payload: { error: { message: 'model is required', type: 'invalid_request_error' } },
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
value: {
|
||||
requestedModel: model,
|
||||
isStream: raw.stream === true,
|
||||
upstreamBody: raw,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function normalizeUpstreamFinalResponse(
|
||||
payload: unknown,
|
||||
fallbackModel: string,
|
||||
fallbackText = '',
|
||||
): NormalizedFinalResponse {
|
||||
const now = Math.floor(Date.now() / 1000);
|
||||
const fallbackId = `chatcmpl-meta-${Date.now()}`;
|
||||
|
||||
if (isRecord(payload) && Array.isArray(payload.choices)) {
|
||||
const choice = payload.choices[0] ?? {};
|
||||
const content = extractAssistantContent(choice) || extractAssistantContent(payload);
|
||||
const reasoning = extractAssistantReasoning(choice) || extractAssistantReasoning(payload);
|
||||
return {
|
||||
id: isNonEmptyString(payload.id) ? payload.id : fallbackId,
|
||||
model: isNonEmptyString(payload.model) ? payload.model : fallbackModel,
|
||||
created: ensureIntegerTimestamp(payload.created, now),
|
||||
content: content || fallbackText,
|
||||
reasoningContent: reasoning,
|
||||
finishReason: normalizeStopReason(choice?.finish_reason ?? payload.stop_reason) || 'stop',
|
||||
};
|
||||
}
|
||||
|
||||
if (isRecord(payload) && typeof payload.type === 'string' && payload.type === 'message') {
|
||||
return {
|
||||
id: isNonEmptyString(payload.id) ? payload.id : fallbackId,
|
||||
model: isNonEmptyString(payload.model) ? payload.model : fallbackModel,
|
||||
created: now,
|
||||
content: parseClaudeMessageContent(payload.content) || fallbackText,
|
||||
reasoningContent: extractTextAndReasoning(payload.content).reasoning,
|
||||
finishReason: normalizeStopReason(payload.stop_reason) || 'stop',
|
||||
};
|
||||
}
|
||||
|
||||
if (isRecord(payload) && Array.isArray(payload.candidates)) {
|
||||
const candidate = payload.candidates[0] || {};
|
||||
const parsedCandidate = extractTextAndReasoning(candidate?.content?.parts || candidate?.content);
|
||||
return {
|
||||
id: isNonEmptyString((payload as any).responseId) ? (payload as any).responseId : fallbackId,
|
||||
model: isNonEmptyString((payload as any).modelVersion)
|
||||
? (payload as any).modelVersion
|
||||
: fallbackModel,
|
||||
created: now,
|
||||
content: parsedCandidate.content || fallbackText,
|
||||
reasoningContent: parsedCandidate.reasoning,
|
||||
finishReason: normalizeStopReason(candidate?.finishReason || (payload as any).finishReason) || 'stop',
|
||||
};
|
||||
}
|
||||
|
||||
if (typeof payload === 'string' && payload.trim()) {
|
||||
return {
|
||||
id: fallbackId,
|
||||
model: fallbackModel,
|
||||
created: now,
|
||||
content: payload,
|
||||
reasoningContent: '',
|
||||
finishReason: 'stop',
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
id: fallbackId,
|
||||
model: fallbackModel,
|
||||
created: now,
|
||||
content: fallbackText,
|
||||
reasoningContent: '',
|
||||
finishReason: 'stop',
|
||||
};
|
||||
}
|
||||
|
||||
export function normalizeUpstreamStreamEvent(
|
||||
payload: unknown,
|
||||
context: StreamTransformContext,
|
||||
fallbackModel: string,
|
||||
): NormalizedStreamEvent {
|
||||
if (!isRecord(payload)) return {};
|
||||
|
||||
if (Array.isArray(payload.choices)) {
|
||||
if (isNonEmptyString(payload.id)) context.id = payload.id;
|
||||
if (isNonEmptyString(payload.model)) context.model = payload.model;
|
||||
context.created = ensureIntegerTimestamp(payload.created, context.created);
|
||||
|
||||
const choice = payload.choices[0] ?? {};
|
||||
const delta = isRecord(choice?.delta) ? choice.delta : {};
|
||||
const deltaParsed = extractTextAndReasoning(delta.content ?? delta);
|
||||
|
||||
const contentDelta =
|
||||
deltaParsed.content
|
||||
|| (typeof choice?.message?.content === 'string' ? choice.message.content : '')
|
||||
|| '';
|
||||
|
||||
const reasoningDelta =
|
||||
(typeof (delta as any).reasoning_content === 'string' ? (delta as any).reasoning_content : '')
|
||||
|| (typeof (delta as any).reasoning === 'string' ? (delta as any).reasoning : '')
|
||||
|| deltaParsed.reasoning
|
||||
|| '';
|
||||
|
||||
return {
|
||||
role: (delta as any).role === 'assistant' ? 'assistant' : undefined,
|
||||
contentDelta: contentDelta || undefined,
|
||||
reasoningDelta: reasoningDelta || undefined,
|
||||
finishReason: normalizeStopReason(choice?.finish_reason),
|
||||
};
|
||||
}
|
||||
|
||||
const type = typeof payload.type === 'string' ? payload.type : '';
|
||||
const message = isRecord(payload.message) ? payload.message : null;
|
||||
|
||||
if (message) {
|
||||
if (isNonEmptyString(message.id)) context.id = message.id;
|
||||
if (isNonEmptyString(message.model)) context.model = message.model;
|
||||
}
|
||||
if (!context.model) context.model = fallbackModel;
|
||||
|
||||
if (type === 'message_start') {
|
||||
return { role: 'assistant' };
|
||||
}
|
||||
|
||||
if (type === 'content_block_start') {
|
||||
const parsed = extractTextAndReasoning(payload.content_block);
|
||||
return {
|
||||
contentDelta: parsed.content || undefined,
|
||||
reasoningDelta: parsed.reasoning || undefined,
|
||||
};
|
||||
}
|
||||
|
||||
if (type === 'content_block_delta') {
|
||||
const delta = isRecord(payload.delta) ? payload.delta : {};
|
||||
const deltaType = typeof delta.type === 'string' ? delta.type : '';
|
||||
const parsed = extractTextAndReasoning(delta);
|
||||
|
||||
if (deltaType === 'thinking_delta') {
|
||||
return {
|
||||
reasoningDelta: parsed.content || parsed.reasoning || undefined,
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
contentDelta: parsed.content || undefined,
|
||||
reasoningDelta: parsed.reasoning || undefined,
|
||||
};
|
||||
}
|
||||
|
||||
if (type === 'message_delta') {
|
||||
const delta = isRecord(payload.delta) ? payload.delta : {};
|
||||
return {
|
||||
finishReason: normalizeStopReason(delta.stop_reason ?? payload.stop_reason),
|
||||
};
|
||||
}
|
||||
|
||||
if (type === 'message_stop') {
|
||||
return { done: true };
|
||||
}
|
||||
|
||||
if (Array.isArray(payload.candidates)) {
|
||||
const candidate = payload.candidates[0] || {};
|
||||
const parsed = extractTextAndReasoning((candidate as any).content?.parts || (candidate as any).content);
|
||||
|
||||
if (isNonEmptyString((payload as any).modelVersion)) {
|
||||
context.model = (payload as any).modelVersion;
|
||||
} else if (!context.model) {
|
||||
context.model = fallbackModel;
|
||||
}
|
||||
|
||||
return {
|
||||
contentDelta: parsed.content || undefined,
|
||||
reasoningDelta: parsed.reasoning || undefined,
|
||||
finishReason: normalizeStopReason((candidate as any).finishReason || (payload as any).finishReason),
|
||||
};
|
||||
}
|
||||
|
||||
const fallback = extractTextAndReasoning(payload);
|
||||
return {
|
||||
contentDelta: fallback.content || undefined,
|
||||
reasoningDelta: fallback.reasoning || undefined,
|
||||
};
|
||||
}
|
||||
|
||||
function buildOpenAiStreamChunk(
|
||||
context: StreamTransformContext,
|
||||
event: NormalizedStreamEvent,
|
||||
): Record<string, unknown> | null {
|
||||
const delta: Record<string, unknown> = {};
|
||||
const isInitialAssistantRoleOnlyEvent = (
|
||||
!context.roleSent
|
||||
&& event.role === 'assistant'
|
||||
&& !event.contentDelta
|
||||
&& !event.reasoningDelta
|
||||
);
|
||||
|
||||
if (!context.roleSent && (event.role === 'assistant' || event.contentDelta || event.reasoningDelta)) {
|
||||
delta.role = 'assistant';
|
||||
context.roleSent = true;
|
||||
} else if (event.role === 'assistant') {
|
||||
delta.role = 'assistant';
|
||||
context.roleSent = true;
|
||||
}
|
||||
|
||||
if (event.contentDelta) {
|
||||
delta.content = event.contentDelta;
|
||||
}
|
||||
|
||||
if (event.reasoningDelta) {
|
||||
delta.reasoning_content = event.reasoningDelta;
|
||||
}
|
||||
|
||||
// Some OpenAI-compatible clients (e.g. OpenWebUI) expect starter chunk to include empty content.
|
||||
if (isInitialAssistantRoleOnlyEvent) {
|
||||
delta.content = '';
|
||||
}
|
||||
|
||||
const finishReason = event.finishReason || null;
|
||||
const hasDelta = Object.keys(delta).length > 0;
|
||||
if (!hasDelta && !finishReason) return null;
|
||||
|
||||
return {
|
||||
id: context.id,
|
||||
object: 'chat.completion.chunk',
|
||||
created: context.created,
|
||||
model: context.model,
|
||||
choices: [{
|
||||
index: 0,
|
||||
delta,
|
||||
finish_reason: finishReason,
|
||||
}],
|
||||
};
|
||||
}
|
||||
|
||||
function ensureClaudeStartEvents(
|
||||
context: StreamTransformContext,
|
||||
claudeContext: ClaudeDownstreamContext,
|
||||
): string[] {
|
||||
if (claudeContext.messageStarted) return [];
|
||||
|
||||
claudeContext.messageStarted = true;
|
||||
const payload = {
|
||||
type: 'message_start',
|
||||
message: {
|
||||
id: buildClaudeMessageId(context.id),
|
||||
type: 'message',
|
||||
role: 'assistant',
|
||||
model: context.model,
|
||||
content: [],
|
||||
stop_reason: null,
|
||||
stop_sequence: null,
|
||||
usage: {
|
||||
input_tokens: 0,
|
||||
output_tokens: 0,
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
return [serializeSse('message_start', payload)];
|
||||
}
|
||||
|
||||
function ensureClaudeContentBlockStart(
|
||||
claudeContext: ClaudeDownstreamContext,
|
||||
): string[] {
|
||||
if (claudeContext.contentBlockStarted) return [];
|
||||
claudeContext.contentBlockStarted = true;
|
||||
|
||||
return [serializeSse('content_block_start', {
|
||||
type: 'content_block_start',
|
||||
index: 0,
|
||||
content_block: {
|
||||
type: 'text',
|
||||
text: '',
|
||||
},
|
||||
})];
|
||||
}
|
||||
|
||||
function buildClaudeDoneEvents(
|
||||
context: StreamTransformContext,
|
||||
claudeContext: ClaudeDownstreamContext,
|
||||
finishReason?: string | null,
|
||||
): string[] {
|
||||
if (claudeContext.doneSent) return [];
|
||||
|
||||
const events: string[] = [];
|
||||
events.push(...ensureClaudeStartEvents(context, claudeContext));
|
||||
|
||||
if (claudeContext.contentBlockStarted) {
|
||||
events.push(serializeSse('content_block_stop', {
|
||||
type: 'content_block_stop',
|
||||
index: 0,
|
||||
}));
|
||||
claudeContext.contentBlockStarted = false;
|
||||
}
|
||||
|
||||
events.push(serializeSse('message_delta', {
|
||||
type: 'message_delta',
|
||||
delta: {
|
||||
stop_reason: toClaudeStopReason(finishReason),
|
||||
stop_sequence: null,
|
||||
},
|
||||
usage: {
|
||||
output_tokens: 0,
|
||||
},
|
||||
}));
|
||||
events.push(serializeSse('message_stop', { type: 'message_stop' }));
|
||||
|
||||
claudeContext.doneSent = true;
|
||||
return events;
|
||||
}
|
||||
|
||||
export function serializeNormalizedStreamEvent(
|
||||
downstreamFormat: DownstreamFormat,
|
||||
event: NormalizedStreamEvent,
|
||||
context: StreamTransformContext,
|
||||
claudeContext: ClaudeDownstreamContext,
|
||||
): string[] {
|
||||
if (downstreamFormat === 'openai') {
|
||||
const chunk = buildOpenAiStreamChunk(context, event);
|
||||
return chunk ? [serializeSse('', chunk)] : [];
|
||||
}
|
||||
|
||||
if (event.done || event.finishReason) {
|
||||
return buildClaudeDoneEvents(context, claudeContext, event.finishReason);
|
||||
}
|
||||
|
||||
const events: string[] = [];
|
||||
if (event.role === 'assistant' || event.contentDelta || event.reasoningDelta) {
|
||||
events.push(...ensureClaudeStartEvents(context, claudeContext));
|
||||
}
|
||||
|
||||
const mergedText = joinNonEmpty([
|
||||
event.reasoningDelta || '',
|
||||
event.contentDelta || '',
|
||||
]);
|
||||
|
||||
if (mergedText) {
|
||||
events.push(...ensureClaudeContentBlockStart(claudeContext));
|
||||
events.push(serializeSse('content_block_delta', {
|
||||
type: 'content_block_delta',
|
||||
index: 0,
|
||||
delta: {
|
||||
type: 'text_delta',
|
||||
text: mergedText,
|
||||
},
|
||||
}));
|
||||
}
|
||||
|
||||
return events;
|
||||
}
|
||||
|
||||
export function serializeStreamDone(
|
||||
downstreamFormat: DownstreamFormat,
|
||||
context: StreamTransformContext,
|
||||
claudeContext: ClaudeDownstreamContext,
|
||||
): string[] {
|
||||
if (context.doneSent) return [];
|
||||
context.doneSent = true;
|
||||
|
||||
if (downstreamFormat === 'openai') {
|
||||
return [serializeSse('', '[DONE]')];
|
||||
}
|
||||
|
||||
return buildClaudeDoneEvents(context, claudeContext, 'stop');
|
||||
}
|
||||
|
||||
export function serializeFinalResponse(
|
||||
downstreamFormat: DownstreamFormat,
|
||||
normalized: NormalizedFinalResponse,
|
||||
usage: { promptTokens: number; completionTokens: number; totalTokens: number },
|
||||
): Record<string, unknown> {
|
||||
if (downstreamFormat === 'claude') {
|
||||
return {
|
||||
id: buildClaudeMessageId(normalized.id),
|
||||
type: 'message',
|
||||
role: 'assistant',
|
||||
model: normalized.model,
|
||||
content: [{
|
||||
type: 'text',
|
||||
text: normalized.content,
|
||||
}],
|
||||
stop_reason: toClaudeStopReason(normalized.finishReason),
|
||||
stop_sequence: null,
|
||||
usage: {
|
||||
input_tokens: usage.promptTokens,
|
||||
output_tokens: usage.completionTokens,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const message: Record<string, unknown> = {
|
||||
role: 'assistant',
|
||||
content: normalized.content,
|
||||
};
|
||||
if (normalized.reasoningContent) {
|
||||
message.reasoning_content = normalized.reasoningContent;
|
||||
}
|
||||
|
||||
return {
|
||||
id: normalized.id,
|
||||
object: 'chat.completion',
|
||||
created: normalized.created,
|
||||
model: normalized.model,
|
||||
choices: [{
|
||||
index: 0,
|
||||
message,
|
||||
finish_reason: normalizeStopReason(normalized.finishReason) || 'stop',
|
||||
}],
|
||||
usage: {
|
||||
prompt_tokens: usage.promptTokens,
|
||||
completion_tokens: usage.completionTokens,
|
||||
total_tokens: usage.totalTokens,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function buildSyntheticOpenAiChunks(normalized: NormalizedFinalResponse): Array<Record<string, unknown>> {
|
||||
const startChunk: Record<string, unknown> = {
|
||||
id: normalized.id,
|
||||
object: 'chat.completion.chunk',
|
||||
created: normalized.created,
|
||||
model: normalized.model,
|
||||
choices: [{
|
||||
index: 0,
|
||||
delta: normalized.content
|
||||
? { role: 'assistant', content: normalized.content }
|
||||
: { role: 'assistant' },
|
||||
finish_reason: null,
|
||||
}],
|
||||
};
|
||||
|
||||
if (normalized.reasoningContent) {
|
||||
(startChunk.choices as any[])[0].delta.reasoning_content = normalized.reasoningContent;
|
||||
}
|
||||
|
||||
const endChunk = {
|
||||
id: normalized.id,
|
||||
object: 'chat.completion.chunk',
|
||||
created: normalized.created,
|
||||
model: normalized.model,
|
||||
choices: [{
|
||||
index: 0,
|
||||
delta: {},
|
||||
finish_reason: normalizeStopReason(normalized.finishReason) || 'stop',
|
||||
}],
|
||||
};
|
||||
|
||||
return [startChunk, endChunk];
|
||||
}
|
||||
|
||||
export function pullSseEventsWithDone(buffer: string): { events: ParsedSseEvent[]; rest: string } {
|
||||
const normalized = buffer.replace(/\r\n/g, '\n');
|
||||
const events: ParsedSseEvent[] = [];
|
||||
let rest = normalized;
|
||||
|
||||
while (true) {
|
||||
const boundary = rest.indexOf('\n\n');
|
||||
if (boundary < 0) break;
|
||||
|
||||
const block = rest.slice(0, boundary);
|
||||
rest = rest.slice(boundary + 2);
|
||||
|
||||
if (!block.trim()) continue;
|
||||
|
||||
const lines = block.split('\n');
|
||||
let eventName = '';
|
||||
const dataLines: string[] = [];
|
||||
|
||||
for (const line of lines) {
|
||||
if (line.startsWith('event:')) {
|
||||
eventName = line.slice(6).trim();
|
||||
continue;
|
||||
}
|
||||
if (line.startsWith('data:')) {
|
||||
dataLines.push(line.slice(5).trimStart());
|
||||
}
|
||||
}
|
||||
|
||||
if (dataLines.length <= 0) continue;
|
||||
|
||||
events.push({
|
||||
event: eventName,
|
||||
data: dataLines.join('\n').trim(),
|
||||
});
|
||||
}
|
||||
|
||||
return { events, rest };
|
||||
}
|
||||
@@ -0,0 +1,256 @@
|
||||
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
|
||||
import { fetch } from 'undici';
|
||||
import { db, schema } from '../../db/index.js';
|
||||
import { tokenRouter } from '../../services/tokenRouter.js';
|
||||
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
|
||||
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
|
||||
import { isTokenExpiredError } from '../../services/alertRules.js';
|
||||
import { estimateProxyCost } from '../../services/modelPricingService.js';
|
||||
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
|
||||
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
|
||||
import { mergeProxyUsage, parseProxyUsage, pullSseDataEvents } from '../../services/proxyUsageParser.js';
|
||||
|
||||
const MAX_RETRIES = 2;
|
||||
|
||||
export async function completionsProxyRoute(app: FastifyInstance) {
|
||||
app.post('/v1/completions', async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
const body = request.body as any;
|
||||
const requestedModel = body?.model;
|
||||
if (!requestedModel) {
|
||||
return reply.code(400).send({ error: { message: 'model is required', type: 'invalid_request_error' } });
|
||||
}
|
||||
|
||||
const isStream = body.stream === true;
|
||||
const excludeChannelIds: number[] = [];
|
||||
let retryCount = 0;
|
||||
|
||||
while (retryCount <= MAX_RETRIES) {
|
||||
let selected = retryCount === 0
|
||||
? tokenRouter.selectChannel(requestedModel)
|
||||
: tokenRouter.selectNextChannel(requestedModel, excludeChannelIds);
|
||||
|
||||
if (!selected && retryCount === 0) {
|
||||
await refreshModelsAndRebuildRoutes();
|
||||
selected = tokenRouter.selectChannel(requestedModel);
|
||||
}
|
||||
|
||||
if (!selected) {
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: 'No available channels after retries',
|
||||
});
|
||||
return reply.code(503).send({
|
||||
error: { message: 'No available channels for this model', type: 'server_error' },
|
||||
});
|
||||
}
|
||||
|
||||
excludeChannelIds.push(selected.channel.id);
|
||||
|
||||
const targetUrl = `${selected.site.url}/v1/completions`;
|
||||
const forwardBody = { ...body, model: selected.actualModel };
|
||||
const startTime = Date.now();
|
||||
|
||||
try {
|
||||
const upstream = await fetch(targetUrl, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'Authorization': `Bearer ${selected.tokenValue}`,
|
||||
},
|
||||
body: JSON.stringify(forwardBody),
|
||||
});
|
||||
|
||||
if (!upstream.ok) {
|
||||
const errText = await upstream.text().catch(() => 'unknown error');
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, errText, retryCount);
|
||||
|
||||
if (isTokenExpiredError({ status: upstream.status, message: errText })) {
|
||||
await reportTokenExpired({
|
||||
accountId: selected.account.id,
|
||||
username: selected.account.username,
|
||||
siteName: selected.site.name,
|
||||
detail: `HTTP ${upstream.status}`,
|
||||
});
|
||||
}
|
||||
|
||||
if (shouldRetryProxyRequest(upstream.status, errText) && retryCount < MAX_RETRIES) {
|
||||
retryCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: `upstream returned HTTP ${upstream.status}`,
|
||||
});
|
||||
return reply.code(upstream.status).send({ error: { message: errText, type: 'upstream_error' } });
|
||||
}
|
||||
|
||||
if (isStream) {
|
||||
reply.raw.writeHead(200, {
|
||||
'Content-Type': 'text/event-stream',
|
||||
'Cache-Control': 'no-cache',
|
||||
'Connection': 'keep-alive',
|
||||
});
|
||||
|
||||
const reader = upstream.body?.getReader();
|
||||
if (!reader) {
|
||||
reply.raw.end();
|
||||
return;
|
||||
}
|
||||
|
||||
const decoder = new TextDecoder();
|
||||
let parsedUsage = { promptTokens: 0, completionTokens: 0, totalTokens: 0 };
|
||||
let sseBuffer = '';
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
const chunk = decoder.decode(value, { stream: true });
|
||||
reply.raw.write(chunk);
|
||||
|
||||
sseBuffer += chunk;
|
||||
const pulled = pullSseDataEvents(sseBuffer);
|
||||
sseBuffer = pulled.rest;
|
||||
for (const eventPayload of pulled.events) {
|
||||
try {
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(JSON.parse(eventPayload)));
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
if (sseBuffer.trim().length > 0) {
|
||||
const pulled = pullSseDataEvents(`${sseBuffer}\n\n`);
|
||||
for (const eventPayload of pulled.events) {
|
||||
try {
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(JSON.parse(eventPayload)));
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
reply.raw.end();
|
||||
}
|
||||
|
||||
const latency = Date.now() - startTime;
|
||||
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
tokenValue: selected.tokenValue,
|
||||
tokenName: selected.tokenName,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
requestStartedAtMs: startTime,
|
||||
requestEndedAtMs: startTime + latency,
|
||||
localLatencyMs: latency,
|
||||
usage: {
|
||||
promptTokens: parsedUsage.promptTokens,
|
||||
completionTokens: parsedUsage.completionTokens,
|
||||
totalTokens: parsedUsage.totalTokens,
|
||||
},
|
||||
});
|
||||
let estimatedCost = await estimateProxyCost({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
promptTokens: resolvedUsage.promptTokens,
|
||||
completionTokens: resolvedUsage.completionTokens,
|
||||
totalTokens: resolvedUsage.totalTokens,
|
||||
});
|
||||
if (resolvedUsage.estimatedCostFromQuota > 0 && (resolvedUsage.recoveredFromSelfLog || estimatedCost <= 0)) {
|
||||
estimatedCost = resolvedUsage.estimatedCostFromQuota;
|
||||
}
|
||||
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
|
||||
logProxy(
|
||||
selected, requestedModel, 'success', 200, latency, null, retryCount,
|
||||
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const data = await upstream.json() as any;
|
||||
const latency = Date.now() - startTime;
|
||||
const parsedUsage = parseProxyUsage(data);
|
||||
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
tokenValue: selected.tokenValue,
|
||||
tokenName: selected.tokenName,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
requestStartedAtMs: startTime,
|
||||
requestEndedAtMs: startTime + latency,
|
||||
localLatencyMs: latency,
|
||||
usage: {
|
||||
promptTokens: parsedUsage.promptTokens,
|
||||
completionTokens: parsedUsage.completionTokens,
|
||||
totalTokens: parsedUsage.totalTokens,
|
||||
},
|
||||
});
|
||||
let estimatedCost = await estimateProxyCost({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
promptTokens: resolvedUsage.promptTokens,
|
||||
completionTokens: resolvedUsage.completionTokens,
|
||||
totalTokens: resolvedUsage.totalTokens,
|
||||
});
|
||||
if (resolvedUsage.estimatedCostFromQuota > 0 && (resolvedUsage.recoveredFromSelfLog || estimatedCost <= 0)) {
|
||||
estimatedCost = resolvedUsage.estimatedCostFromQuota;
|
||||
}
|
||||
|
||||
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
|
||||
logProxy(
|
||||
selected, requestedModel, 'success', 200, latency, null, retryCount,
|
||||
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost,
|
||||
);
|
||||
return reply.send(data);
|
||||
} catch (err: any) {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount);
|
||||
if (retryCount < MAX_RETRIES) {
|
||||
retryCount++;
|
||||
continue;
|
||||
}
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: err.message || 'network failure',
|
||||
});
|
||||
return reply.code(502).send({
|
||||
error: { message: `Upstream error: ${err.message}`, type: 'upstream_error' },
|
||||
});
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function logProxy(
|
||||
selected: any,
|
||||
modelRequested: string,
|
||||
status: string,
|
||||
httpStatus: number,
|
||||
latencyMs: number,
|
||||
errorMessage: string | null,
|
||||
retryCount: number,
|
||||
promptTokens = 0,
|
||||
completionTokens = 0,
|
||||
totalTokens = 0,
|
||||
estimatedCost = 0,
|
||||
) {
|
||||
try {
|
||||
db.insert(schema.proxyLogs).values({
|
||||
routeId: selected.channel.routeId,
|
||||
channelId: selected.channel.id,
|
||||
accountId: selected.account.id,
|
||||
modelRequested,
|
||||
modelActual: selected.actualModel,
|
||||
status,
|
||||
httpStatus,
|
||||
latencyMs,
|
||||
promptTokens,
|
||||
completionTokens,
|
||||
totalTokens,
|
||||
estimatedCost,
|
||||
errorMessage,
|
||||
retryCount,
|
||||
}).run();
|
||||
} catch {}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,172 @@
|
||||
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
|
||||
import { fetch } from 'undici';
|
||||
import { db, schema } from '../../db/index.js';
|
||||
import { tokenRouter } from '../../services/tokenRouter.js';
|
||||
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
|
||||
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
|
||||
import { isTokenExpiredError } from '../../services/alertRules.js';
|
||||
import { estimateProxyCost } from '../../services/modelPricingService.js';
|
||||
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
|
||||
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
|
||||
import { parseProxyUsage } from '../../services/proxyUsageParser.js';
|
||||
|
||||
const MAX_RETRIES = 2;
|
||||
|
||||
export async function embeddingsProxyRoute(app: FastifyInstance) {
|
||||
app.post('/v1/embeddings', async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
const body = request.body as any;
|
||||
const requestedModel = body?.model;
|
||||
if (!requestedModel) {
|
||||
return reply.code(400).send({ error: { message: 'model is required', type: 'invalid_request_error' } });
|
||||
}
|
||||
|
||||
const excludeChannelIds: number[] = [];
|
||||
let retryCount = 0;
|
||||
|
||||
while (retryCount <= MAX_RETRIES) {
|
||||
let selected = retryCount === 0
|
||||
? tokenRouter.selectChannel(requestedModel)
|
||||
: tokenRouter.selectNextChannel(requestedModel, excludeChannelIds);
|
||||
|
||||
if (!selected && retryCount === 0) {
|
||||
await refreshModelsAndRebuildRoutes();
|
||||
selected = tokenRouter.selectChannel(requestedModel);
|
||||
}
|
||||
|
||||
if (!selected) {
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: 'No available channels after retries',
|
||||
});
|
||||
return reply.code(503).send({ error: { message: 'No available channels', type: 'server_error' } });
|
||||
}
|
||||
|
||||
excludeChannelIds.push(selected.channel.id);
|
||||
|
||||
const targetUrl = `${selected.site.url}/v1/embeddings`;
|
||||
const forwardBody = { ...body, model: selected.actualModel };
|
||||
const startTime = Date.now();
|
||||
|
||||
try {
|
||||
const upstream = await fetch(targetUrl, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'Authorization': `Bearer ${selected.tokenValue}`,
|
||||
},
|
||||
body: JSON.stringify(forwardBody),
|
||||
});
|
||||
|
||||
const text = await upstream.text();
|
||||
if (!upstream.ok) {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, text, retryCount);
|
||||
|
||||
if (isTokenExpiredError({ status: upstream.status, message: text })) {
|
||||
await reportTokenExpired({
|
||||
accountId: selected.account.id,
|
||||
username: selected.account.username,
|
||||
siteName: selected.site.name,
|
||||
detail: `HTTP ${upstream.status}`,
|
||||
});
|
||||
}
|
||||
|
||||
if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) {
|
||||
retryCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: `upstream returned HTTP ${upstream.status}`,
|
||||
});
|
||||
return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } });
|
||||
}
|
||||
|
||||
let data: any = {};
|
||||
try { data = JSON.parse(text); } catch { data = {}; }
|
||||
const latency = Date.now() - startTime;
|
||||
const parsedUsage = parseProxyUsage(data);
|
||||
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
tokenValue: selected.tokenValue,
|
||||
tokenName: selected.tokenName,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
requestStartedAtMs: startTime,
|
||||
requestEndedAtMs: startTime + latency,
|
||||
localLatencyMs: latency,
|
||||
usage: {
|
||||
promptTokens: parsedUsage.promptTokens,
|
||||
completionTokens: parsedUsage.completionTokens,
|
||||
totalTokens: parsedUsage.totalTokens,
|
||||
},
|
||||
});
|
||||
let estimatedCost = await estimateProxyCost({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
promptTokens: resolvedUsage.promptTokens,
|
||||
completionTokens: resolvedUsage.completionTokens,
|
||||
totalTokens: resolvedUsage.totalTokens,
|
||||
});
|
||||
if (resolvedUsage.estimatedCostFromQuota > 0 && (resolvedUsage.recoveredFromSelfLog || estimatedCost <= 0)) {
|
||||
estimatedCost = resolvedUsage.estimatedCostFromQuota;
|
||||
}
|
||||
|
||||
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
|
||||
logProxy(
|
||||
selected, requestedModel, 'success', upstream.status, latency, null, retryCount,
|
||||
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost,
|
||||
);
|
||||
return reply.code(upstream.status).send(data);
|
||||
} catch (err: any) {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount);
|
||||
if (retryCount < MAX_RETRIES) {
|
||||
retryCount++;
|
||||
continue;
|
||||
}
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: err.message || 'network failure',
|
||||
});
|
||||
return reply.code(502).send({ error: { message: err.message, type: 'upstream_error' } });
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function logProxy(
|
||||
selected: any,
|
||||
modelRequested: string,
|
||||
status: string,
|
||||
httpStatus: number,
|
||||
latencyMs: number,
|
||||
errorMessage: string | null,
|
||||
retryCount: number,
|
||||
promptTokens = 0,
|
||||
completionTokens = 0,
|
||||
totalTokens = 0,
|
||||
estimatedCost = 0,
|
||||
) {
|
||||
try {
|
||||
db.insert(schema.proxyLogs).values({
|
||||
routeId: selected.channel.routeId,
|
||||
channelId: selected.channel.id,
|
||||
accountId: selected.account.id,
|
||||
modelRequested,
|
||||
modelActual: selected.actualModel,
|
||||
status,
|
||||
httpStatus,
|
||||
latencyMs,
|
||||
promptTokens,
|
||||
completionTokens,
|
||||
totalTokens,
|
||||
estimatedCost,
|
||||
errorMessage,
|
||||
retryCount,
|
||||
}).run();
|
||||
} catch {}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
|
||||
import { fetch } from 'undici';
|
||||
import { db, schema } from '../../db/index.js';
|
||||
import { tokenRouter } from '../../services/tokenRouter.js';
|
||||
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
|
||||
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
|
||||
import { isTokenExpiredError } from '../../services/alertRules.js';
|
||||
import { estimateProxyCost } from '../../services/modelPricingService.js';
|
||||
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
|
||||
|
||||
const MAX_RETRIES = 2;
|
||||
|
||||
export async function imagesProxyRoute(app: FastifyInstance) {
|
||||
app.post('/v1/images/generations', async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
const body = request.body as any;
|
||||
const requestedModel = body?.model || 'gpt-image-1';
|
||||
const excludeChannelIds: number[] = [];
|
||||
let retryCount = 0;
|
||||
|
||||
while (retryCount <= MAX_RETRIES) {
|
||||
let selected = retryCount === 0
|
||||
? tokenRouter.selectChannel(requestedModel)
|
||||
: tokenRouter.selectNextChannel(requestedModel, excludeChannelIds);
|
||||
|
||||
if (!selected && retryCount === 0) {
|
||||
await refreshModelsAndRebuildRoutes();
|
||||
selected = tokenRouter.selectChannel(requestedModel);
|
||||
}
|
||||
|
||||
if (!selected) {
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: 'No available channels after retries',
|
||||
});
|
||||
return reply.code(503).send({
|
||||
error: { message: 'No available channels for this model', type: 'server_error' },
|
||||
});
|
||||
}
|
||||
|
||||
excludeChannelIds.push(selected.channel.id);
|
||||
|
||||
const targetUrl = `${selected.site.url}/v1/images/generations`;
|
||||
const forwardBody = { ...body, model: selected.actualModel };
|
||||
const startTime = Date.now();
|
||||
|
||||
try {
|
||||
const upstream = await fetch(targetUrl, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'Authorization': `Bearer ${selected.tokenValue}`,
|
||||
},
|
||||
body: JSON.stringify(forwardBody),
|
||||
});
|
||||
|
||||
const text = await upstream.text();
|
||||
if (!upstream.ok) {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, text, retryCount);
|
||||
if (isTokenExpiredError({ status: upstream.status, message: text })) {
|
||||
await reportTokenExpired({
|
||||
accountId: selected.account.id,
|
||||
username: selected.account.username,
|
||||
siteName: selected.site.name,
|
||||
detail: `HTTP ${upstream.status}`,
|
||||
});
|
||||
}
|
||||
if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) {
|
||||
retryCount++;
|
||||
continue;
|
||||
}
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: `upstream returned HTTP ${upstream.status}`,
|
||||
});
|
||||
return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } });
|
||||
}
|
||||
|
||||
let data: any = {};
|
||||
try { data = JSON.parse(text); } catch { data = { data: [] }; }
|
||||
|
||||
const latency = Date.now() - startTime;
|
||||
const estimatedCost = await estimateProxyCost({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
promptTokens: 0,
|
||||
completionTokens: 0,
|
||||
totalTokens: 0,
|
||||
});
|
||||
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
|
||||
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, estimatedCost);
|
||||
return reply.code(upstream.status).send(data);
|
||||
} catch (err: any) {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount);
|
||||
if (retryCount < MAX_RETRIES) {
|
||||
retryCount++;
|
||||
continue;
|
||||
}
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: err.message || 'network failure',
|
||||
});
|
||||
return reply.code(502).send({
|
||||
error: { message: `Upstream error: ${err.message}`, type: 'upstream_error' },
|
||||
});
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function logProxy(
|
||||
selected: any,
|
||||
modelRequested: string,
|
||||
status: string,
|
||||
httpStatus: number,
|
||||
latencyMs: number,
|
||||
errorMessage: string | null,
|
||||
retryCount: number,
|
||||
estimatedCost = 0,
|
||||
) {
|
||||
try {
|
||||
db.insert(schema.proxyLogs).values({
|
||||
routeId: selected.channel.routeId,
|
||||
channelId: selected.channel.id,
|
||||
accountId: selected.account.id,
|
||||
modelRequested,
|
||||
modelActual: selected.actualModel,
|
||||
status,
|
||||
httpStatus,
|
||||
latencyMs,
|
||||
promptTokens: 0,
|
||||
completionTokens: 0,
|
||||
totalTokens: 0,
|
||||
estimatedCost,
|
||||
errorMessage,
|
||||
retryCount,
|
||||
}).run();
|
||||
} catch {}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
import { FastifyInstance } from 'fastify';
|
||||
import { db, schema } from '../../db/index.js';
|
||||
import { and, eq } from 'drizzle-orm';
|
||||
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
|
||||
|
||||
export async function modelsProxyRoute(app: FastifyInstance) {
|
||||
app.get('/v1/models', async (request) => {
|
||||
const readModels = () => {
|
||||
const rows = db.select({ modelName: schema.modelAvailability.modelName })
|
||||
.from(schema.modelAvailability)
|
||||
.innerJoin(schema.accounts, eq(schema.modelAvailability.accountId, schema.accounts.id))
|
||||
.innerJoin(schema.sites, eq(schema.accounts.siteId, schema.sites.id))
|
||||
.where(
|
||||
and(
|
||||
eq(schema.modelAvailability.available, true),
|
||||
eq(schema.accounts.status, 'active'),
|
||||
eq(schema.sites.status, 'active'),
|
||||
),
|
||||
)
|
||||
.all();
|
||||
return Array.from(new Set(rows.map((r) => r.modelName))).sort();
|
||||
};
|
||||
|
||||
let models = readModels();
|
||||
if (models.length === 0) {
|
||||
await refreshModelsAndRebuildRoutes();
|
||||
models = readModels();
|
||||
}
|
||||
|
||||
const wantsClaudeFormat = typeof request.headers['anthropic-version'] === 'string'
|
||||
|| typeof request.headers['x-api-key'] === 'string';
|
||||
if (wantsClaudeFormat) {
|
||||
const data = models.map((id) => ({
|
||||
id,
|
||||
type: 'model',
|
||||
display_name: id,
|
||||
created_at: new Date().toISOString(),
|
||||
}));
|
||||
return {
|
||||
data,
|
||||
first_id: data[0]?.id || null,
|
||||
last_id: data[data.length - 1]?.id || null,
|
||||
has_more: false,
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
object: 'list',
|
||||
data: models.map(id => ({
|
||||
id,
|
||||
object: 'model',
|
||||
created: Math.floor(Date.now() / 1000),
|
||||
owned_by: 'metapi',
|
||||
})),
|
||||
};
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
import { FastifyInstance } from 'fastify';
|
||||
import { proxyAuthMiddleware } from '../../middleware/auth.js';
|
||||
import { chatProxyRoute, claudeMessagesProxyRoute } from './chat.js';
|
||||
import { modelsProxyRoute } from './models.js';
|
||||
import { embeddingsProxyRoute } from './embeddings.js';
|
||||
import { completionsProxyRoute } from './completions.js';
|
||||
import { imagesProxyRoute } from './images.js';
|
||||
|
||||
export async function proxyRoutes(app: FastifyInstance) {
|
||||
// Auth middleware for all /v1 routes
|
||||
app.addHook('onRequest', async (request, reply) => {
|
||||
await proxyAuthMiddleware(request, reply);
|
||||
});
|
||||
|
||||
await app.register(chatProxyRoute);
|
||||
await app.register(claudeMessagesProxyRoute);
|
||||
await app.register(completionsProxyRoute);
|
||||
await app.register(modelsProxyRoute);
|
||||
await app.register(embeddingsProxyRoute);
|
||||
await app.register(imagesProxyRoute);
|
||||
}
|
||||
Reference in New Issue
Block a user