feat(proxy): pass account proxy through all proxy route handlers

Thread account-level proxyUrl from extraConfig into
resolveChannelProxyUrl for endpointFlow and the third parameter
of withSiteRecordProxyRequestInit for direct fetch calls across
chat, responses, completions, embeddings, images, search, and
videos proxy routes.
This commit is contained in:
Hureru
2026-03-16 01:06:16 +08:00
parent b43a4fd2ea
commit 6d2f1c4b81
7 changed files with 535 additions and 526 deletions
+4 -3
View File
@@ -8,7 +8,8 @@ import { isTokenExpiredError } from '../../services/alertRules.js';
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js'; import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js'; import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js';
import { resolveProxyUrlForSite, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; import { resolveChannelProxyUrl, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { getProxyUrlFromExtraConfig } from '../../services/accountExtraConfig.js';
import { type DownstreamFormat } from '../../transformers/shared/normalized.js'; import { type DownstreamFormat } from '../../transformers/shared/normalized.js';
import { import {
buildUpstreamEndpointRequest, buildUpstreamEndpointRequest,
@@ -168,7 +169,7 @@ async function handleChatProxyRequest(
method: 'POST', method: 'POST',
headers: compatibilityRequest.headers, headers: compatibilityRequest.headers,
body: JSON.stringify(compatibilityRequest.body), body: JSON.stringify(compatibilityRequest.body),
}), }, getProxyUrlFromExtraConfig(selected.account.extraConfig)),
), ),
}); });
let startTime = Date.now(); let startTime = Date.now();
@@ -176,7 +177,7 @@ async function handleChatProxyRequest(
try { try {
const endpointResult = await executeEndpointFlow({ const endpointResult = await executeEndpointFlow({
siteUrl: selected.site.url, siteUrl: selected.site.url,
proxyUrl: resolveProxyUrlForSite(selected.site), proxyUrl: resolveChannelProxyUrl(selected.site, selected.account.extraConfig),
endpointCandidates, endpointCandidates,
buildRequest: (endpoint) => buildEndpointRequest(endpoint), buildRequest: (endpoint) => buildEndpointRequest(endpoint),
tryRecover: endpointStrategy.tryRecover, tryRecover: endpointStrategy.tryRecover,
+185 -184
View File
@@ -1,45 +1,46 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici'; import { fetch } from 'undici';
import { db, schema } from '../../db/index.js'; import { db, schema } from '../../db/index.js';
import { tokenRouter } from '../../services/tokenRouter.js'; import { tokenRouter } from '../../services/tokenRouter.js';
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
import { isTokenExpiredError } from '../../services/alertRules.js'; import { isTokenExpiredError } from '../../services/alertRules.js';
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js'; import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
import { mergeProxyUsage, parseProxyUsage, pullSseDataEvents } from '../../services/proxyUsageParser.js'; import { mergeProxyUsage, parseProxyUsage, pullSseDataEvents } from '../../services/proxyUsageParser.js';
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js'; import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { composeProxyLogMessage } from './logPathMeta.js'; import { getProxyUrlFromExtraConfig } from '../../services/accountExtraConfig.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js'; import { composeProxyLogMessage } from './logPathMeta.js';
import { resolveProxyLogBilling } from './proxyBilling.js'; import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { buildUpstreamUrl } from './upstreamUrl.js'; import { resolveProxyLogBilling } from './proxyBilling.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
const MAX_RETRIES = 2; const MAX_RETRIES = 2;
export async function completionsProxyRoute(app: FastifyInstance) { export async function completionsProxyRoute(app: FastifyInstance) {
app.post('/v1/completions', async (request: FastifyRequest, reply: FastifyReply) => { app.post('/v1/completions', async (request: FastifyRequest, reply: FastifyReply) => {
const body = request.body as any; const body = request.body as any;
const requestedModel = body?.model; const requestedModel = body?.model;
if (!requestedModel) { if (!requestedModel) {
return reply.code(400).send({ error: { message: 'model is required', type: 'invalid_request_error' } }); return reply.code(400).send({ error: { message: 'model is required', type: 'invalid_request_error' } });
} }
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return; if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request); const downstreamPolicy = getDownstreamRoutingPolicy(request);
const isStream = body.stream === true; const isStream = body.stream === true;
const excludeChannelIds: number[] = []; const excludeChannelIds: number[] = [];
let retryCount = 0; let retryCount = 0;
while (retryCount <= MAX_RETRIES) { while (retryCount <= MAX_RETRIES) {
let selected = retryCount === 0 let selected = retryCount === 0
? await tokenRouter.selectChannel(requestedModel, downstreamPolicy) ? await tokenRouter.selectChannel(requestedModel, downstreamPolicy)
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy);
if (!selected && retryCount === 0) { if (!selected && retryCount === 0) {
await refreshModelsAndRebuildRoutes(); await refreshModelsAndRebuildRoutes();
selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy);
} }
if (!selected) { if (!selected) {
await reportProxyAllFailed({ await reportProxyAllFailed({
@@ -51,21 +52,21 @@ export async function completionsProxyRoute(app: FastifyInstance) {
}); });
} }
excludeChannelIds.push(selected.channel.id); excludeChannelIds.push(selected.channel.id);
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/completions'); const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/completions');
const forwardBody = { ...body, model: selected.actualModel }; const forwardBody = { ...body, model: selected.actualModel };
const startTime = Date.now(); const startTime = Date.now();
try { try {
const upstream = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, { const upstream = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST', method: 'POST',
headers: { headers: {
'Content-Type': 'application/json', 'Content-Type': 'application/json',
'Authorization': `Bearer ${selected.tokenValue}`, 'Authorization': `Bearer ${selected.tokenValue}`,
}, },
body: JSON.stringify(forwardBody), body: JSON.stringify(forwardBody),
})); }, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
if (!upstream.ok) { if (!upstream.ok) {
const errText = await upstream.text().catch(() => 'unknown error'); const errText = await upstream.text().catch(() => 'unknown error');
@@ -81,10 +82,10 @@ export async function completionsProxyRoute(app: FastifyInstance) {
}); });
} }
if (shouldRetryProxyRequest(upstream.status, errText) && retryCount < MAX_RETRIES) { if (shouldRetryProxyRequest(upstream.status, errText) && retryCount < MAX_RETRIES) {
retryCount++; retryCount++;
continue; continue;
} }
await reportProxyAllFailed({ await reportProxyAllFailed({
model: requestedModel, model: requestedModel,
@@ -106,110 +107,110 @@ export async function completionsProxyRoute(app: FastifyInstance) {
return; return;
} }
const decoder = new TextDecoder(); const decoder = new TextDecoder();
let parsedUsage: ReturnType<typeof parseProxyUsage> = { let parsedUsage: ReturnType<typeof parseProxyUsage> = {
promptTokens: 0, promptTokens: 0,
completionTokens: 0, completionTokens: 0,
totalTokens: 0, totalTokens: 0,
cacheReadTokens: 0, cacheReadTokens: 0,
cacheCreationTokens: 0, cacheCreationTokens: 0,
promptTokensIncludeCache: null, promptTokensIncludeCache: null,
}; };
let sseBuffer = ''; let sseBuffer = '';
try { try {
while (true) { while (true) {
const { done, value } = await reader.read(); const { done, value } = await reader.read();
if (done) break; if (done) break;
const chunk = decoder.decode(value, { stream: true }); const chunk = decoder.decode(value, { stream: true });
reply.raw.write(chunk); 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; sseBuffer += chunk;
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({ const pulled = pullSseDataEvents(sseBuffer);
site: selected.site, sseBuffer = pulled.rest;
account: selected.account, for (const eventPayload of pulled.events) {
tokenValue: selected.tokenValue, try {
tokenName: selected.tokenName, parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(JSON.parse(eventPayload)));
modelName: selected.actualModel || requestedModel, } catch {}
requestStartedAtMs: startTime, }
requestEndedAtMs: startTime + latency, }
localLatencyMs: latency, if (sseBuffer.trim().length > 0) {
usage: { const pulled = pullSseDataEvents(`${sseBuffer}\n\n`);
promptTokens: parsedUsage.promptTokens, for (const eventPayload of pulled.events) {
completionTokens: parsedUsage.completionTokens, try {
totalTokens: parsedUsage.totalTokens, parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(JSON.parse(eventPayload)));
}, } catch {}
}); }
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({ }
site: selected.site, } finally {
account: selected.account, reader.releaseLock();
modelName: selected.actualModel || requestedModel, reply.raw.end();
parsedUsage, }
resolvedUsage,
}); const latency = Date.now() - startTime;
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost); const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
recordDownstreamCostUsage(request, estimatedCost); site: selected.site,
logProxy( account: selected.account,
selected, requestedModel, 'success', 200, latency, null, retryCount, tokenValue: selected.tokenValue,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails, tokenName: selected.tokenName,
); modelName: selected.actualModel || requestedModel,
return; requestStartedAtMs: startTime,
} requestEndedAtMs: startTime + latency,
localLatencyMs: latency,
const data = await upstream.json() as any; usage: {
const latency = Date.now() - startTime; promptTokens: parsedUsage.promptTokens,
const parsedUsage = parseProxyUsage(data); completionTokens: parsedUsage.completionTokens,
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({ totalTokens: parsedUsage.totalTokens,
site: selected.site, },
account: selected.account, });
tokenValue: selected.tokenValue, const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
tokenName: selected.tokenName, site: selected.site,
modelName: selected.actualModel || requestedModel, account: selected.account,
requestStartedAtMs: startTime, modelName: selected.actualModel || requestedModel,
requestEndedAtMs: startTime + latency, parsedUsage,
localLatencyMs: latency, resolvedUsage,
usage: { });
promptTokens: parsedUsage.promptTokens, tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
completionTokens: parsedUsage.completionTokens, recordDownstreamCostUsage(request, estimatedCost);
totalTokens: parsedUsage.totalTokens, logProxy(
}, selected, requestedModel, 'success', 200, latency, null, retryCount,
}); resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails,
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({ );
site: selected.site, return;
account: selected.account, }
modelName: selected.actualModel || requestedModel,
parsedUsage, const data = await upstream.json() as any;
resolvedUsage, const latency = Date.now() - startTime;
}); const parsedUsage = parseProxyUsage(data);
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost); site: selected.site,
recordDownstreamCostUsage(request, estimatedCost); account: selected.account,
logProxy( tokenValue: selected.tokenValue,
selected, requestedModel, 'success', 200, latency, null, retryCount, tokenName: selected.tokenName,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails, modelName: selected.actualModel || requestedModel,
); requestStartedAtMs: startTime,
return reply.send(data); requestEndedAtMs: startTime + latency,
localLatencyMs: latency,
usage: {
promptTokens: parsedUsage.promptTokens,
completionTokens: parsedUsage.completionTokens,
totalTokens: parsedUsage.totalTokens,
},
});
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
site: selected.site,
account: selected.account,
modelName: selected.actualModel || requestedModel,
parsedUsage,
resolvedUsage,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(
selected, requestedModel, 'success', 200, latency, null, retryCount,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails,
);
return reply.send(data);
} catch (err: any) { } catch (err: any) {
tokenRouter.recordFailure(selected.channel.id); tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount); logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount);
@@ -237,38 +238,38 @@ async function logProxy(
latencyMs: number, latencyMs: number,
errorMessage: string | null, errorMessage: string | null,
retryCount: number, retryCount: number,
promptTokens = 0, promptTokens = 0,
completionTokens = 0, completionTokens = 0,
totalTokens = 0, totalTokens = 0,
estimatedCost = 0, estimatedCost = 0,
billingDetails: unknown = null, billingDetails: unknown = null,
) { ) {
try { try {
const createdAt = formatUtcSqlDateTime(new Date()); const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({ const normalizedErrorMessage = composeProxyLogMessage({
downstreamPath: '/v1/completions', downstreamPath: '/v1/completions',
errorMessage, errorMessage,
}); });
await db.insert(schema.proxyLogs).values({ await db.insert(schema.proxyLogs).values({
routeId: selected.channel.routeId, routeId: selected.channel.routeId,
channelId: selected.channel.id, channelId: selected.channel.id,
accountId: selected.account.id, accountId: selected.account.id,
modelRequested, modelRequested,
modelActual: selected.actualModel, modelActual: selected.actualModel,
status, status,
httpStatus, httpStatus,
latencyMs, latencyMs,
promptTokens, promptTokens,
completionTokens, completionTokens,
totalTokens, totalTokens,
estimatedCost, estimatedCost,
billingDetails: billingDetails ? JSON.stringify(billingDetails) : null, billingDetails: billingDetails ? JSON.stringify(billingDetails) : null,
errorMessage: normalizedErrorMessage, errorMessage: normalizedErrorMessage,
retryCount, retryCount,
createdAt, createdAt,
}).run(); }).run();
} catch (error) { } catch (error) {
console.warn('[proxy/completions] failed to write proxy log', error); console.warn('[proxy/completions] failed to write proxy log', error);
} }
} }
+116 -115
View File
@@ -1,44 +1,45 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici'; import { fetch } from 'undici';
import { db, schema } from '../../db/index.js'; import { db, schema } from '../../db/index.js';
import { tokenRouter } from '../../services/tokenRouter.js'; import { tokenRouter } from '../../services/tokenRouter.js';
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
import { isTokenExpiredError } from '../../services/alertRules.js'; import { isTokenExpiredError } from '../../services/alertRules.js';
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js'; import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
import { parseProxyUsage } from '../../services/proxyUsageParser.js'; import { parseProxyUsage } from '../../services/proxyUsageParser.js';
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js'; import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { composeProxyLogMessage } from './logPathMeta.js'; import { getProxyUrlFromExtraConfig } from '../../services/accountExtraConfig.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js'; import { composeProxyLogMessage } from './logPathMeta.js';
import { resolveProxyLogBilling } from './proxyBilling.js'; import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { buildUpstreamUrl } from './upstreamUrl.js'; import { resolveProxyLogBilling } from './proxyBilling.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
const MAX_RETRIES = 2; const MAX_RETRIES = 2;
export async function embeddingsProxyRoute(app: FastifyInstance) { export async function embeddingsProxyRoute(app: FastifyInstance) {
app.post('/v1/embeddings', async (request: FastifyRequest, reply: FastifyReply) => { app.post('/v1/embeddings', async (request: FastifyRequest, reply: FastifyReply) => {
const body = request.body as any; const body = request.body as any;
const requestedModel = body?.model; const requestedModel = body?.model;
if (!requestedModel) { if (!requestedModel) {
return reply.code(400).send({ error: { message: 'model is required', type: 'invalid_request_error' } }); return reply.code(400).send({ error: { message: 'model is required', type: 'invalid_request_error' } });
} }
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return; if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request); const downstreamPolicy = getDownstreamRoutingPolicy(request);
const excludeChannelIds: number[] = []; const excludeChannelIds: number[] = [];
let retryCount = 0; let retryCount = 0;
while (retryCount <= MAX_RETRIES) { while (retryCount <= MAX_RETRIES) {
let selected = retryCount === 0 let selected = retryCount === 0
? await tokenRouter.selectChannel(requestedModel, downstreamPolicy) ? await tokenRouter.selectChannel(requestedModel, downstreamPolicy)
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy);
if (!selected && retryCount === 0) { if (!selected && retryCount === 0) {
await refreshModelsAndRebuildRoutes(); await refreshModelsAndRebuildRoutes();
selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy);
} }
if (!selected) { if (!selected) {
await reportProxyAllFailed({ await reportProxyAllFailed({
@@ -48,21 +49,21 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
return reply.code(503).send({ error: { message: 'No available channels', type: 'server_error' } }); return reply.code(503).send({ error: { message: 'No available channels', type: 'server_error' } });
} }
excludeChannelIds.push(selected.channel.id); excludeChannelIds.push(selected.channel.id);
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/embeddings'); const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/embeddings');
const forwardBody = { ...body, model: selected.actualModel }; const forwardBody = { ...body, model: selected.actualModel };
const startTime = Date.now(); const startTime = Date.now();
try { try {
const upstream = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, { const upstream = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST', method: 'POST',
headers: { headers: {
'Content-Type': 'application/json', 'Content-Type': 'application/json',
'Authorization': `Bearer ${selected.tokenValue}`, 'Authorization': `Bearer ${selected.tokenValue}`,
}, },
body: JSON.stringify(forwardBody), body: JSON.stringify(forwardBody),
})); }, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
const text = await upstream.text(); const text = await upstream.text();
if (!upstream.ok) { if (!upstream.ok) {
@@ -78,10 +79,10 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
}); });
} }
if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) { if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) {
retryCount++; retryCount++;
continue; continue;
} }
await reportProxyAllFailed({ await reportProxyAllFailed({
model: requestedModel, model: requestedModel,
@@ -90,40 +91,40 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } }); return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } });
} }
let data: any = {}; let data: any = {};
try { data = JSON.parse(text); } catch { data = {}; } try { data = JSON.parse(text); } catch { data = {}; }
const latency = Date.now() - startTime; const latency = Date.now() - startTime;
const parsedUsage = parseProxyUsage(data); const parsedUsage = parseProxyUsage(data);
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({ const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site, site: selected.site,
account: selected.account, account: selected.account,
tokenValue: selected.tokenValue, tokenValue: selected.tokenValue,
tokenName: selected.tokenName, tokenName: selected.tokenName,
modelName: selected.actualModel || requestedModel, modelName: selected.actualModel || requestedModel,
requestStartedAtMs: startTime, requestStartedAtMs: startTime,
requestEndedAtMs: startTime + latency, requestEndedAtMs: startTime + latency,
localLatencyMs: latency, localLatencyMs: latency,
usage: { usage: {
promptTokens: parsedUsage.promptTokens, promptTokens: parsedUsage.promptTokens,
completionTokens: parsedUsage.completionTokens, completionTokens: parsedUsage.completionTokens,
totalTokens: parsedUsage.totalTokens, totalTokens: parsedUsage.totalTokens,
}, },
}); });
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({ const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
site: selected.site, site: selected.site,
account: selected.account, account: selected.account,
modelName: selected.actualModel || requestedModel, modelName: selected.actualModel || requestedModel,
parsedUsage, parsedUsage,
resolvedUsage, resolvedUsage,
}); });
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost); tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
recordDownstreamCostUsage(request, estimatedCost); recordDownstreamCostUsage(request, estimatedCost);
logProxy( logProxy(
selected, requestedModel, 'success', upstream.status, latency, null, retryCount, selected, requestedModel, 'success', upstream.status, latency, null, retryCount,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails, resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails,
); );
return reply.code(upstream.status).send(data); return reply.code(upstream.status).send(data);
} catch (err: any) { } catch (err: any) {
tokenRouter.recordFailure(selected.channel.id); tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount); logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount);
@@ -149,38 +150,38 @@ async function logProxy(
latencyMs: number, latencyMs: number,
errorMessage: string | null, errorMessage: string | null,
retryCount: number, retryCount: number,
promptTokens = 0, promptTokens = 0,
completionTokens = 0, completionTokens = 0,
totalTokens = 0, totalTokens = 0,
estimatedCost = 0, estimatedCost = 0,
billingDetails: unknown = null, billingDetails: unknown = null,
) { ) {
try { try {
const createdAt = formatUtcSqlDateTime(new Date()); const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({ const normalizedErrorMessage = composeProxyLogMessage({
downstreamPath: '/v1/embeddings', downstreamPath: '/v1/embeddings',
errorMessage, errorMessage,
}); });
await db.insert(schema.proxyLogs).values({ await db.insert(schema.proxyLogs).values({
routeId: selected.channel.routeId, routeId: selected.channel.routeId,
channelId: selected.channel.id, channelId: selected.channel.id,
accountId: selected.account.id, accountId: selected.account.id,
modelRequested, modelRequested,
modelActual: selected.actualModel, modelActual: selected.actualModel,
status, status,
httpStatus, httpStatus,
latencyMs, latencyMs,
promptTokens, promptTokens,
completionTokens, completionTokens,
totalTokens, totalTokens,
estimatedCost, estimatedCost,
billingDetails: billingDetails ? JSON.stringify(billingDetails) : null, billingDetails: billingDetails ? JSON.stringify(billingDetails) : null,
errorMessage: normalizedErrorMessage, errorMessage: normalizedErrorMessage,
retryCount, retryCount,
createdAt, createdAt,
}).run(); }).run();
} catch (error) { } catch (error) {
console.warn('[proxy/embeddings] failed to write proxy log', error); console.warn('[proxy/embeddings] failed to write proxy log', error);
} }
} }
+220 -218
View File
@@ -2,40 +2,41 @@
import { fetch } from 'undici'; import { fetch } from 'undici';
import { db, schema } from '../../db/index.js'; import { db, schema } from '../../db/index.js';
import { tokenRouter } from '../../services/tokenRouter.js'; import { tokenRouter } from '../../services/tokenRouter.js';
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
import { isTokenExpiredError } from '../../services/alertRules.js'; import { isTokenExpiredError } from '../../services/alertRules.js';
import { estimateProxyCost } from '../../services/modelPricingService.js'; import { estimateProxyCost } from '../../services/modelPricingService.js';
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js'; import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { composeProxyLogMessage } from './logPathMeta.js'; import { getProxyUrlFromExtraConfig } from '../../services/accountExtraConfig.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js'; import { composeProxyLogMessage } from './logPathMeta.js';
import { cloneFormDataWithOverrides, ensureMultipartBufferParser, parseMultipartFormData } from './multipart.js'; import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { buildUpstreamUrl } from './upstreamUrl.js'; import { cloneFormDataWithOverrides, ensureMultipartBufferParser, parseMultipartFormData } from './multipart.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
const MAX_RETRIES = 2; const MAX_RETRIES = 2;
export async function imagesProxyRoute(app: FastifyInstance) { export async function imagesProxyRoute(app: FastifyInstance) {
ensureMultipartBufferParser(app); ensureMultipartBufferParser(app);
app.post('/v1/images/generations', async (request: FastifyRequest, reply: FastifyReply) => { app.post('/v1/images/generations', async (request: FastifyRequest, reply: FastifyReply) => {
const body = request.body as any; const body = request.body as any;
const requestedModel = body?.model || 'gpt-image-1'; const requestedModel = body?.model || 'gpt-image-1';
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return; if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request); const downstreamPolicy = getDownstreamRoutingPolicy(request);
const excludeChannelIds: number[] = []; const excludeChannelIds: number[] = [];
let retryCount = 0; let retryCount = 0;
while (retryCount <= MAX_RETRIES) { while (retryCount <= MAX_RETRIES) {
let selected = retryCount === 0 let selected = retryCount === 0
? await tokenRouter.selectChannel(requestedModel, downstreamPolicy) ? await tokenRouter.selectChannel(requestedModel, downstreamPolicy)
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy);
if (!selected && retryCount === 0) { if (!selected && retryCount === 0) {
await refreshModelsAndRebuildRoutes(); await refreshModelsAndRebuildRoutes();
selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy);
} }
if (!selected) { if (!selected) {
await reportProxyAllFailed({ await reportProxyAllFailed({
@@ -49,19 +50,19 @@ export async function imagesProxyRoute(app: FastifyInstance) {
excludeChannelIds.push(selected.channel.id); excludeChannelIds.push(selected.channel.id);
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/images/generations'); const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/images/generations');
const forwardBody = { ...body, model: selected.actualModel }; const forwardBody = { ...body, model: selected.actualModel };
const startTime = Date.now(); const startTime = Date.now();
try { try {
const upstream = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, { const upstream = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST', method: 'POST',
headers: { headers: {
'Content-Type': 'application/json', 'Content-Type': 'application/json',
'Authorization': `Bearer ${selected.tokenValue}`, 'Authorization': `Bearer ${selected.tokenValue}`,
}, },
body: JSON.stringify(forwardBody), body: JSON.stringify(forwardBody),
})); }, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
const text = await upstream.text(); const text = await upstream.text();
if (!upstream.ok) { if (!upstream.ok) {
@@ -75,10 +76,10 @@ export async function imagesProxyRoute(app: FastifyInstance) {
detail: `HTTP ${upstream.status}`, detail: `HTTP ${upstream.status}`,
}); });
} }
if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) { if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) {
retryCount++; retryCount++;
continue; continue;
} }
await reportProxyAllFailed({ await reportProxyAllFailed({
model: requestedModel, model: requestedModel,
reason: `upstream returned HTTP ${upstream.status}`, reason: `upstream returned HTTP ${upstream.status}`,
@@ -90,18 +91,18 @@ export async function imagesProxyRoute(app: FastifyInstance) {
try { data = JSON.parse(text); } catch { data = { data: [] }; } try { data = JSON.parse(text); } catch { data = { data: [] }; }
const latency = Date.now() - startTime; const latency = Date.now() - startTime;
const estimatedCost = await estimateProxyCost({ const estimatedCost = await estimateProxyCost({
site: selected.site, site: selected.site,
account: selected.account, account: selected.account,
modelName: selected.actualModel || requestedModel, modelName: selected.actualModel || requestedModel,
promptTokens: 0, promptTokens: 0,
completionTokens: 0, completionTokens: 0,
totalTokens: 0, totalTokens: 0,
}); });
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost); tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
recordDownstreamCostUsage(request, estimatedCost); recordDownstreamCostUsage(request, estimatedCost);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, estimatedCost); logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, estimatedCost);
return reply.code(upstream.status).send(data); return reply.code(upstream.status).send(data);
} catch (err: any) { } catch (err: any) {
tokenRouter.recordFailure(selected.channel.id); tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount); logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount);
@@ -116,175 +117,176 @@ export async function imagesProxyRoute(app: FastifyInstance) {
return reply.code(502).send({ return reply.code(502).send({
error: { message: `Upstream error: ${err.message}`, type: 'upstream_error' }, error: { message: `Upstream error: ${err.message}`, type: 'upstream_error' },
}); });
} }
} }
}); });
app.post('/v1/images/edits', async (request: FastifyRequest, reply: FastifyReply) => { app.post('/v1/images/edits', async (request: FastifyRequest, reply: FastifyReply) => {
const multipartForm = await parseMultipartFormData(request); const multipartForm = await parseMultipartFormData(request);
const jsonBody = (!multipartForm && request.body && typeof request.body === 'object') const jsonBody = (!multipartForm && request.body && typeof request.body === 'object')
? request.body as Record<string, unknown> ? request.body as Record<string, unknown>
: null; : null;
const requestedModel = typeof multipartForm?.get('model') === 'string' const requestedModel = typeof multipartForm?.get('model') === 'string'
? String(multipartForm.get('model')).trim() ? String(multipartForm.get('model')).trim()
: (typeof jsonBody?.model === 'string' ? jsonBody.model.trim() : '') || 'gpt-image-1'; : (typeof jsonBody?.model === 'string' ? jsonBody.model.trim() : '') || 'gpt-image-1';
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return; if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request); const downstreamPolicy = getDownstreamRoutingPolicy(request);
const excludeChannelIds: number[] = []; const excludeChannelIds: number[] = [];
let retryCount = 0; let retryCount = 0;
while (retryCount <= MAX_RETRIES) { while (retryCount <= MAX_RETRIES) {
let selected = retryCount === 0 let selected = retryCount === 0
? await tokenRouter.selectChannel(requestedModel, downstreamPolicy) ? await tokenRouter.selectChannel(requestedModel, downstreamPolicy)
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy);
if (!selected && retryCount === 0) { if (!selected && retryCount === 0) {
await refreshModelsAndRebuildRoutes(); await refreshModelsAndRebuildRoutes();
selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy);
} }
if (!selected) { if (!selected) {
await reportProxyAllFailed({ await reportProxyAllFailed({
model: requestedModel, model: requestedModel,
reason: 'No available channels after retries', reason: 'No available channels after retries',
}); });
return reply.code(503).send({ return reply.code(503).send({
error: { message: 'No available channels for this model', type: 'server_error' }, error: { message: 'No available channels for this model', type: 'server_error' },
}); });
} }
excludeChannelIds.push(selected.channel.id); excludeChannelIds.push(selected.channel.id);
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/images/edits'); const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/images/edits');
const startTime = Date.now(); const startTime = Date.now();
try { try {
const requestInit = multipartForm const accountProxy = getProxyUrlFromExtraConfig(selected.account.extraConfig);
? withSiteRecordProxyRequestInit(selected.site, { const requestInit = multipartForm
method: 'POST', ? withSiteRecordProxyRequestInit(selected.site, {
headers: { method: 'POST',
Authorization: `Bearer ${selected.tokenValue}`, headers: {
}, Authorization: `Bearer ${selected.tokenValue}`,
body: cloneFormDataWithOverrides(multipartForm, { },
model: selected.actualModel || requestedModel, body: cloneFormDataWithOverrides(multipartForm, {
}) as any, model: selected.actualModel || requestedModel,
}) }) as any,
: withSiteRecordProxyRequestInit(selected.site, { }, accountProxy)
method: 'POST', : withSiteRecordProxyRequestInit(selected.site, {
headers: { method: 'POST',
'Content-Type': 'application/json', headers: {
Authorization: `Bearer ${selected.tokenValue}`, 'Content-Type': 'application/json',
}, Authorization: `Bearer ${selected.tokenValue}`,
body: JSON.stringify({ },
...(jsonBody || {}), body: JSON.stringify({
model: selected.actualModel || requestedModel, ...(jsonBody || {}),
}), model: selected.actualModel || requestedModel,
}); }),
}, accountProxy);
const upstream = await fetch(targetUrl, requestInit);
const text = await upstream.text(); const upstream = await fetch(targetUrl, requestInit);
if (!upstream.ok) { const text = await upstream.text();
tokenRouter.recordFailure(selected.channel.id); if (!upstream.ok) {
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, text, retryCount, 0, '/v1/images/edits'); tokenRouter.recordFailure(selected.channel.id);
if (isTokenExpiredError({ status: upstream.status, message: text })) { logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, text, retryCount, 0, '/v1/images/edits');
await reportTokenExpired({ if (isTokenExpiredError({ status: upstream.status, message: text })) {
accountId: selected.account.id, await reportTokenExpired({
username: selected.account.username, accountId: selected.account.id,
siteName: selected.site.name, username: selected.account.username,
detail: `HTTP ${upstream.status}`, siteName: selected.site.name,
}); detail: `HTTP ${upstream.status}`,
} });
if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) { }
retryCount++; if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) {
continue; retryCount++;
} continue;
await reportProxyAllFailed({ }
model: requestedModel, await reportProxyAllFailed({
reason: `upstream returned HTTP ${upstream.status}`, model: requestedModel,
}); reason: `upstream returned HTTP ${upstream.status}`,
return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } }); });
} return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } });
}
let data: any = {};
try { data = JSON.parse(text); } catch { data = { data: [] }; } let data: any = {};
try { data = JSON.parse(text); } catch { data = { data: [] }; }
const latency = Date.now() - startTime;
const estimatedCost = await estimateProxyCost({ const latency = Date.now() - startTime;
site: selected.site, const estimatedCost = await estimateProxyCost({
account: selected.account, site: selected.site,
modelName: selected.actualModel || requestedModel, account: selected.account,
promptTokens: 0, modelName: selected.actualModel || requestedModel,
completionTokens: 0, promptTokens: 0,
totalTokens: 0, completionTokens: 0,
}); totalTokens: 0,
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost); });
recordDownstreamCostUsage(request, estimatedCost); tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, estimatedCost, '/v1/images/edits'); recordDownstreamCostUsage(request, estimatedCost);
return reply.code(upstream.status).send(data); logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, estimatedCost, '/v1/images/edits');
} catch (err: any) { return reply.code(upstream.status).send(data);
tokenRouter.recordFailure(selected.channel.id); } catch (err: any) {
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount, 0, '/v1/images/edits'); tokenRouter.recordFailure(selected.channel.id);
if (retryCount < MAX_RETRIES) { logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount, 0, '/v1/images/edits');
retryCount++; if (retryCount < MAX_RETRIES) {
continue; retryCount++;
} continue;
await reportProxyAllFailed({ }
model: requestedModel, await reportProxyAllFailed({
reason: err.message || 'network failure', model: requestedModel,
}); reason: err.message || 'network failure',
return reply.code(502).send({ });
error: { message: `Upstream error: ${err.message}`, type: 'upstream_error' }, return reply.code(502).send({
}); error: { message: `Upstream error: ${err.message}`, type: 'upstream_error' },
} });
} }
}); }
});
app.post('/v1/images/variations', async (_request: FastifyRequest, reply: FastifyReply) => {
return reply.code(400).send({ app.post('/v1/images/variations', async (_request: FastifyRequest, reply: FastifyReply) => {
error: { return reply.code(400).send({
message: 'Image variations are not supported', error: {
type: 'invalid_request_error', message: 'Image variations are not supported',
}, type: 'invalid_request_error',
}); },
}); });
} });
}
async function logProxy( async function logProxy(
selected: any, selected: any,
modelRequested: string, modelRequested: string,
status: string, status: string,
httpStatus: number, httpStatus: number,
latencyMs: number, latencyMs: number,
errorMessage: string | null, errorMessage: string | null,
retryCount: number, retryCount: number,
estimatedCost = 0, estimatedCost = 0,
downstreamPath = '/v1/images/generations', downstreamPath = '/v1/images/generations',
) { ) {
try { try {
const createdAt = formatUtcSqlDateTime(new Date()); const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({ const normalizedErrorMessage = composeProxyLogMessage({
downstreamPath, downstreamPath,
errorMessage, errorMessage,
}); });
await db.insert(schema.proxyLogs).values({ await db.insert(schema.proxyLogs).values({
routeId: selected.channel.routeId, routeId: selected.channel.routeId,
channelId: selected.channel.id, channelId: selected.channel.id,
accountId: selected.account.id, accountId: selected.account.id,
modelRequested, modelRequested,
modelActual: selected.actualModel, modelActual: selected.actualModel,
status, status,
httpStatus, httpStatus,
latencyMs, latencyMs,
promptTokens: 0, promptTokens: 0,
completionTokens: 0, completionTokens: 0,
totalTokens: 0, totalTokens: 0,
estimatedCost, estimatedCost,
errorMessage: normalizedErrorMessage, errorMessage: normalizedErrorMessage,
retryCount, retryCount,
createdAt, createdAt,
}).run(); }).run();
} catch (error) { } catch (error) {
console.warn('[proxy/images] failed to write proxy log', error); console.warn('[proxy/images] failed to write proxy log', error);
} }
} }
+4 -3
View File
@@ -8,7 +8,8 @@ import { isTokenExpiredError } from '../../services/alertRules.js';
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js'; import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js'; import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js';
import { resolveProxyUrlForSite, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; import { resolveChannelProxyUrl, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { getProxyUrlFromExtraConfig } from '../../services/accountExtraConfig.js';
import { openAiResponsesTransformer } from '../../transformers/openai/responses/index.js'; import { openAiResponsesTransformer } from '../../transformers/openai/responses/index.js';
import { import {
buildUpstreamEndpointRequest, buildUpstreamEndpointRequest,
@@ -259,7 +260,7 @@ export async function responsesProxyRoute(app: FastifyInstance) {
method: 'POST', method: 'POST',
headers: compatibilityRequest.headers, headers: compatibilityRequest.headers,
body: JSON.stringify(compatibilityRequest.body), body: JSON.stringify(compatibilityRequest.body),
}), }, getProxyUrlFromExtraConfig(selected.account.extraConfig)),
), ),
}); });
@@ -268,7 +269,7 @@ export async function responsesProxyRoute(app: FastifyInstance) {
try { try {
const endpointResult = await executeEndpointFlow({ const endpointResult = await executeEndpointFlow({
siteUrl: selected.site.url, siteUrl: selected.site.url,
proxyUrl: resolveProxyUrlForSite(selected.site), proxyUrl: resolveChannelProxyUrl(selected.site, selected.account.extraConfig),
endpointCandidates, endpointCandidates,
buildRequest: (endpoint) => buildEndpointRequest(endpoint), buildRequest: (endpoint) => buildEndpointRequest(endpoint),
tryRecover: endpointStrategy.tryRecover, tryRecover: endpointStrategy.tryRecover,
+2 -1
View File
@@ -8,6 +8,7 @@ import { isTokenExpiredError } from '../../services/alertRules.js';
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js'; import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { getProxyUrlFromExtraConfig } from '../../services/accountExtraConfig.js';
import { composeProxyLogMessage } from './logPathMeta.js'; import { composeProxyLogMessage } from './logPathMeta.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js'; import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { buildUpstreamUrl } from './upstreamUrl.js'; import { buildUpstreamUrl } from './upstreamUrl.js';
@@ -92,7 +93,7 @@ export async function searchProxyRoute(app: FastifyInstance) {
Authorization: `Bearer ${selected.tokenValue}`, Authorization: `Bearer ${selected.tokenValue}`,
}, },
body: JSON.stringify(forwardBody), body: JSON.stringify(forwardBody),
})); }, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
const text = await upstream.text(); const text = await upstream.text();
if (!upstream.ok) { if (!upstream.ok) {
+4 -2
View File
@@ -8,6 +8,7 @@ import { estimateProxyCost } from '../../services/modelPricingService.js';
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js'; import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteProxyRequestInit, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; import { withSiteProxyRequestInit, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { getProxyUrlFromExtraConfig } from '../../services/accountExtraConfig.js';
import { cloneFormDataWithOverrides, ensureMultipartBufferParser, parseMultipartFormData } from './multipart.js'; import { cloneFormDataWithOverrides, ensureMultipartBufferParser, parseMultipartFormData } from './multipart.js';
import { buildUpstreamUrl } from './upstreamUrl.js'; import { buildUpstreamUrl } from './upstreamUrl.js';
import { import {
@@ -75,6 +76,7 @@ export async function videosProxyRoute(app: FastifyInstance) {
const startTime = Date.now(); const startTime = Date.now();
try { try {
const accountProxy = getProxyUrlFromExtraConfig(selected.account.extraConfig);
const requestInit = multipartForm const requestInit = multipartForm
? withSiteRecordProxyRequestInit(selected.site, { ? withSiteRecordProxyRequestInit(selected.site, {
method: 'POST', method: 'POST',
@@ -84,7 +86,7 @@ export async function videosProxyRoute(app: FastifyInstance) {
body: cloneFormDataWithOverrides(multipartForm, { body: cloneFormDataWithOverrides(multipartForm, {
model: selected.actualModel || requestedModel, model: selected.actualModel || requestedModel,
}) as any, }) as any,
}) }, accountProxy)
: withSiteRecordProxyRequestInit(selected.site, { : withSiteRecordProxyRequestInit(selected.site, {
method: 'POST', method: 'POST',
headers: { headers: {
@@ -95,7 +97,7 @@ export async function videosProxyRoute(app: FastifyInstance) {
...(jsonBody || {}), ...(jsonBody || {}),
model: selected.actualModel || requestedModel, model: selected.actualModel || requestedModel,
}), }),
}); }, accountProxy);
const upstream = await fetch(targetUrl, requestInit); const upstream = await fetch(targetUrl, requestInit);
const text = await upstream.text(); const text = await upstream.text();