Nvidia/BossAI
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #159 from Hureru/feat/account-level-proxy

Browse Source 
feat(proxy): account-level proxy URL override
This commit is contained in:
Cita
2026-03-21 13:14:08 +08:00
committed by GitHub
parent 8ced774e29 04655a8b17
commit 9b17c3ebdf
20 changed files with 960 additions and 615 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/routes/api/accountTokens.ts
+36 -23
View File
@@ -17,8 +17,9 @@ import {
syncTokensFromUpstream,
} from '../../services/accountTokenService.js';
import { getAdapter } from '../../services/platforms/index.js';
import { getCredentialModeFromExtraConfig, resolvePlatformUserId } from '../../services/accountExtraConfig.js';
import { getCredentialModeFromExtraConfig, getProxyUrlFromExtraConfig, resolvePlatformUserId } from '../../services/accountExtraConfig.js';
import { startBackgroundTask } from '../../services/backgroundTaskService.js';
import { withAccountProxyOverride } from '../../services/siteProxy.js';
import {
rebuildTokenRoutesFromAvailability,
refreshModelsForAccount,
@@ -261,15 +262,18 @@ async function executeAccountTokenSync(row: AccountWithSiteRow): Promise<SyncExe
try {
const platformUserId = resolvePlatformUserId(row.accounts.extraConfig, row.accounts.username);
const accountProxyUrl = getProxyUrlFromExtraConfig(row.accounts.extraConfig);
let tokens = await withTimeout(
() => adapter.getApiTokens(row.sites.url, row.accounts.accessToken, platformUserId),
() => withAccountProxyOverride(accountProxyUrl,
() => adapter.getApiTokens(row.sites.url, row.accounts.accessToken, platformUserId)),
TOKEN_SYNC_TIMEOUT_MS,
`token sync timeout (${Math.round(TOKEN_SYNC_TIMEOUT_MS / 1000)}s)`,
);
if (tokens.length === 0) {
const fallback = await withTimeout(
() => adapter.getApiToken(row.sites.url, row.accounts.accessToken, platformUserId),
() => withAccountProxyOverride(accountProxyUrl,
() => adapter.getApiToken(row.sites.url, row.accounts.accessToken, platformUserId)),
TOKEN_SYNC_TIMEOUT_MS,
`token sync timeout (${Math.round(TOKEN_SYNC_TIMEOUT_MS / 1000)}s)`,
);
@@ -579,20 +583,23 @@ export async function accountTokensRoutes(app: FastifyInstance) {
}
const platformUserId = resolvePlatformUserId(account.extraConfig, account.username);
const createdViaUpstream = await adapter.createApiToken(
site.url,
account.accessToken,
platformUserId,
{
name: asTrimmedString(body.name),
group: asTrimmedString(body.group),
unlimitedQuota,
remainQuota,
expiredTime,
allowIps: asTrimmedString(body.allowIps),
modelLimitsEnabled,
modelLimits: asTrimmedString(body.modelLimits),
},
const createdViaUpstream = await withAccountProxyOverride(
getProxyUrlFromExtraConfig(account.extraConfig),
() => adapter.createApiToken(
site.url,
account.accessToken,
platformUserId,
{
name: asTrimmedString(body.name),
group: asTrimmedString(body.group),
unlimitedQuota,
remainQuota,
expiredTime,
allowIps: asTrimmedString(body.allowIps),
modelLimitsEnabled,
modelLimits: asTrimmedString(body.modelLimits),
},
),
);
if (!createdViaUpstream) {
return reply.code(502).send({ success: false, message: '站点创建令牌失败' });
@@ -652,11 +659,14 @@ export async function accountTokensRoutes(app: FastifyInstance) {
if (shouldDeleteUpstream) {
const platformUserId = resolvePlatformUserId(account.extraConfig, account.username);
const upstreamDeleted = await adapter!.deleteApiToken(
site.url,
account.accessToken,
existing.token,
platformUserId,
const upstreamDeleted = await withAccountProxyOverride(
getProxyUrlFromExtraConfig(account.extraConfig),
() => adapter!.deleteApiToken(
site.url,
account.accessToken,
existing.token,
platformUserId,
),
);
if (!upstreamDeleted) {
return { success: false, message: '站点删除令牌失败，本地未删除' };
@@ -904,7 +914,10 @@ export async function accountTokensRoutes(app: FastifyInstance) {
try {
const platformUserId = resolvePlatformUserId(account.extraConfig, account.username);
const groups = await adapter.getUserGroups(site.url, account.accessToken, platformUserId);
const groups = await withAccountProxyOverride(
getProxyUrlFromExtraConfig(account.extraConfig),
() => adapter.getUserGroups(site.url, account.accessToken, platformUserId),
);
const normalized = Array.from(new Set((groups || []).map((item) => String(item || '').trim()).filter(Boolean)));
return { success: true, groups: normalized.length > 0 ? normalized : ['default'] };
} catch (error: any) {
src/server/routes/api/accounts.ts
+18 -2
View File
@@ -7,6 +7,7 @@ import { refreshModelsForAccount, rebuildTokenRoutesFromAvailability } from '../
import { ensureDefaultTokenForAccount, syncTokensFromUpstream } from '../../services/accountTokenService.js';
import {
getCredentialModeFromExtraConfig,
getProxyUrlFromExtraConfig,
guessPlatformUserIdFromUsername,
hasOauthProvider,
getSub2ApiAuthFromExtraConfig,
@@ -26,7 +27,7 @@ import {
type RuntimeHealthState,
} from '../../services/accountHealthService.js';
import { appendSessionTokenRebindHint } from '../../services/alertRules.js';
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { parseSiteProxyUrlInput, withAccountProxyOverride, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { createRateLimitGuard } from '../../middleware/requestRateLimit.js';
type AccountWithSiteRow = {
@@ -981,7 +982,10 @@ export async function accountsRoutes(app: FastifyInstance) {
let verifyResult: any;
try {
verifyResult = await adapter.verifyToken(site.url, nextAccessToken, candidatePlatformUserId);
verifyResult = await withAccountProxyOverride(
getProxyUrlFromExtraConfig(account.extraConfig),
() => adapter.verifyToken(site.url, nextAccessToken, candidatePlatformUserId),
);
} catch (err: any) {
return reply.code(400).send({
success: false,
@@ -1305,6 +1309,18 @@ export async function accountsRoutes(app: FastifyInstance) {
updates.sortOrder = normalizedSortOrder;
}
if (Object.prototype.hasOwnProperty.call(body, 'proxyUrl')) {
const baseExtraConfig = typeof updates.extraConfig === 'string'
? updates.extraConfig : account.extraConfig;
const { present, valid, proxyUrl: normalizedProxy } = parseSiteProxyUrlInput(body.proxyUrl);
if (present && !valid) {
return reply.code(400).send({ message: 'Invalid proxy URL format' });
}
updates.extraConfig = mergeAccountExtraConfig(baseExtraConfig, {
proxyUrl: normalizedProxy ?? undefined,
});
}
updates.updatedAt = new Date().toISOString();
await db.update(schema.accounts).set(updates).where(eq(schema.accounts.id, id)).run();
src/server/routes/proxy/completions.ts
+2 -1
View File
@@ -9,6 +9,7 @@ import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageF
import { mergeProxyUsage, parseProxyUsage, pullSseDataEvents } from '../../services/proxyUsageParser.js';
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { getProxyUrlFromExtraConfig } from '../../services/accountExtraConfig.js';
import { composeProxyLogMessage } from './logPathMeta.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { detectProxyFailure } from './proxyFailureJudge.js';
@@ -75,7 +76,7 @@ export async function completionsProxyRoute(app: FastifyInstance) {
'Authorization': `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify(forwardBody),
}));
}, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
if (!upstream.ok) {
const errText = await upstream.text().catch(() => 'unknown error');
src/server/routes/proxy/embeddings.ts
+195 -194
View File
@@ -1,53 +1,54 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici';
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 { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
import { parseProxyUsage } from '../../services/proxyUsageParser.js';
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { composeProxyLogMessage } from './logPathMeta.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { resolveProxyLogBilling } from './proxyBilling.js';
import { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.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 { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
import { parseProxyUsage } from '../../services/proxyUsageParser.js';
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { getProxyUrlFromExtraConfig } from '../../services/accountExtraConfig.js';
import { composeProxyLogMessage } from './logPathMeta.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { resolveProxyLogBilling } from './proxyBilling.js';
import { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.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' } });
}
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const downstreamPath = '/v1/embeddings';
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body,
});
const excludeChannelIds: number[] = [];
let retryCount = 0;
const requestedModel = body?.model;
if (!requestedModel) {
return reply.code(400).send({ error: { message: 'model is required', type: 'invalid_request_error' } });
}
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const downstreamPath = '/v1/embeddings';
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body,
});
while (retryCount <= MAX_RETRIES) {
let selected = retryCount === 0
? await tokenRouter.selectChannel(requestedModel, downstreamPolicy)
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy);
if (!selected && retryCount === 0) {
await refreshModelsAndRebuildRoutes();
selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy);
}
const excludeChannelIds: number[] = [];
let retryCount = 0;
while (retryCount <= MAX_RETRIES) {
let selected = retryCount === 0
? await tokenRouter.selectChannel(requestedModel, downstreamPolicy)
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy);
if (!selected && retryCount === 0) {
await refreshModelsAndRebuildRoutes();
selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy);
}
if (!selected) {
await reportProxyAllFailed({
@@ -57,46 +58,46 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
return reply.code(503).send({ error: { message: 'No available channels', type: 'server_error' } });
}
excludeChannelIds.push(selected.channel.id);
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/embeddings');
const forwardBody = { ...body, model: selected.actualModel };
const startTime = Date.now();
excludeChannelIds.push(selected.channel.id);
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/embeddings');
const forwardBody = { ...body, model: selected.actualModel || requestedModel };
const startTime = Date.now();
try {
const upstream = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify(forwardBody),
}));
const upstream = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify(forwardBody),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
const text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
upstream.status,
Date.now() - startTime,
text,
retryCount,
downstreamApiKeyId,
0,
0,
0,
0,
null,
clientContext,
downstreamPath,
);
const text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
upstream.status,
Date.now() - startTime,
text,
retryCount,
downstreamApiKeyId,
0,
0,
0,
0,
null,
clientContext,
downstreamPath,
);
if (isTokenExpiredError({ status: upstream.status, message: text })) {
await reportTokenExpired({
@@ -107,10 +108,10 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
});
}
if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) {
retryCount++;
continue;
}
if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) {
retryCount++;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
@@ -119,63 +120,63 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
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,
},
});
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
site: selected.site,
account: selected.account,
modelName: selected.actualModel || requestedModel,
parsedUsage,
resolvedUsage,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost, selected.actualModel);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(
selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails, clientContext, downstreamPath,
);
return reply.code(upstream.status).send(data);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
0,
Date.now() - startTime,
err.message,
retryCount,
downstreamApiKeyId,
0,
0,
0,
0,
null,
clientContext,
downstreamPath,
);
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,
},
});
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
site: selected.site,
account: selected.account,
modelName: selected.actualModel || requestedModel,
parsedUsage,
resolvedUsage,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost, selected.actualModel);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(
selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails, clientContext, downstreamPath,
);
return reply.code(upstream.status).send(data);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
0,
Date.now() - startTime,
err.message,
retryCount,
downstreamApiKeyId,
0,
0,
0,
0,
null,
clientContext,
downstreamPath,
);
if (retryCount < MAX_RETRIES) {
retryCount++;
continue;
@@ -190,59 +191,59 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
});
}
async function logProxy(
selected: any,
modelRequested: string,
status: string,
httpStatus: number,
latencyMs: number,
errorMessage: string | null,
retryCount: number,
downstreamApiKeyId: number | null = null,
promptTokens = 0,
completionTokens = 0,
totalTokens = 0,
estimatedCost = 0,
billingDetails: unknown = null,
clientContext: DownstreamClientContext | null = null,
downstreamPath = '/v1/embeddings',
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({
clientKind: clientContext?.clientKind && clientContext.clientKind !== 'generic'
? clientContext.clientKind
: null,
sessionId: clientContext?.sessionId || null,
traceHint: clientContext?.traceHint || null,
downstreamPath,
errorMessage,
});
await insertProxyLog({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
accountId: selected.account.id,
downstreamApiKeyId,
modelRequested,
modelActual: selected.actualModel,
status,
httpStatus,
latencyMs,
promptTokens,
completionTokens,
totalTokens,
estimatedCost,
billingDetails,
clientFamily: clientContext?.clientKind || null,
clientAppId: clientContext?.clientAppId || null,
clientAppName: clientContext?.clientAppName || null,
clientConfidence: clientContext?.clientConfidence || null,
errorMessage: normalizedErrorMessage,
retryCount,
createdAt,
});
} catch (error) {
console.warn('[proxy/embeddings] failed to write proxy log', error);
}
}
async function logProxy(
selected: any,
modelRequested: string,
status: string,
httpStatus: number,
latencyMs: number,
errorMessage: string | null,
retryCount: number,
downstreamApiKeyId: number | null = null,
promptTokens = 0,
completionTokens = 0,
totalTokens = 0,
estimatedCost = 0,
billingDetails: unknown = null,
clientContext: DownstreamClientContext | null = null,
downstreamPath = '/v1/embeddings',
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({
clientKind: clientContext?.clientKind && clientContext.clientKind !== 'generic'
? clientContext.clientKind
: null,
sessionId: clientContext?.sessionId || null,
traceHint: clientContext?.traceHint || null,
downstreamPath,
errorMessage,
});
await insertProxyLog({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
accountId: selected.account.id,
downstreamApiKeyId,
modelRequested,
modelActual: selected.actualModel,
status,
httpStatus,
latencyMs,
promptTokens,
completionTokens,
totalTokens,
estimatedCost,
billingDetails,
clientFamily: clientContext?.clientKind || null,
clientAppId: clientContext?.clientAppId || null,
clientAppName: clientContext?.clientAppName || null,
clientConfidence: clientContext?.clientConfidence || null,
errorMessage: normalizedErrorMessage,
retryCount,
createdAt,
});
} catch (error) {
console.warn('[proxy/embeddings] failed to write proxy log', error);
}
}
src/server/routes/proxy/images.ts
+334 -333
View File
@@ -1,50 +1,51 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici';
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 { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { composeProxyLogMessage } from './logPathMeta.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { cloneFormDataWithOverrides, ensureMultipartBufferParser, parseMultipartFormData } from './multipart.js';
import { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.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 { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { getProxyUrlFromExtraConfig } from '../../services/accountExtraConfig.js';
import { composeProxyLogMessage } from './logPathMeta.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { cloneFormDataWithOverrides, ensureMultipartBufferParser, parseMultipartFormData } from './multipart.js';
import { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.js';
const MAX_RETRIES = 2;
export async function imagesProxyRoute(app: FastifyInstance) {
ensureMultipartBufferParser(app);
app.post('/v1/images/generations', async (request: FastifyRequest, reply: FastifyReply) => {
const body = request.body as any;
const requestedModel = body?.model || 'gpt-image-1';
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const downstreamPath = '/v1/images/generations';
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body,
});
const excludeChannelIds: number[] = [];
let retryCount = 0;
while (retryCount <= MAX_RETRIES) {
let selected = retryCount === 0
? await tokenRouter.selectChannel(requestedModel, downstreamPolicy)
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy);
if (!selected && retryCount === 0) {
await refreshModelsAndRebuildRoutes();
selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy);
}
export async function imagesProxyRoute(app: FastifyInstance) {
ensureMultipartBufferParser(app);
app.post('/v1/images/generations', async (request: FastifyRequest, reply: FastifyReply) => {
const body = request.body as any;
const requestedModel = body?.model || 'gpt-image-1';
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const downstreamPath = '/v1/images/generations';
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body,
});
const excludeChannelIds: number[] = [];
let retryCount = 0;
while (retryCount <= MAX_RETRIES) {
let selected = retryCount === 0
? await tokenRouter.selectChannel(requestedModel, downstreamPolicy)
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy);
if (!selected && retryCount === 0) {
await refreshModelsAndRebuildRoutes();
selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy);
}
if (!selected) {
await reportProxyAllFailed({
@@ -58,52 +59,52 @@ export async function imagesProxyRoute(app: FastifyInstance) {
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 startTime = Date.now();
try {
const upstream = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify(forwardBody),
}));
const upstream = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify(forwardBody),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
const text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
upstream.status,
Date.now() - startTime,
text,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
if (isTokenExpiredError({ status: upstream.status, message: text })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
const text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
upstream.status,
Date.now() - startTime,
text,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
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;
}
if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) {
retryCount++;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: `upstream returned HTTP ${upstream.status}`,
@@ -115,41 +116,41 @@ export async function imagesProxyRoute(app: FastifyInstance) {
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,
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, selected.actualModel);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId, estimatedCost, downstreamPath, clientContext);
return reply.code(upstream.status).send(data);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
0,
Date.now() - startTime,
err.message,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
if (retryCount < MAX_RETRIES) {
retryCount++;
continue;
}
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost, selected.actualModel);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId, estimatedCost, downstreamPath, clientContext);
return reply.code(upstream.status).send(data);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
0,
Date.now() - startTime,
err.message,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
if (retryCount < MAX_RETRIES) {
retryCount++;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: err.message || 'network failure',
@@ -157,226 +158,226 @@ export async function imagesProxyRoute(app: FastifyInstance) {
return reply.code(502).send({
error: { message: `Upstream error: ${err.message}`, type: 'upstream_error' },
});
}
}
});
app.post('/v1/images/edits', async (request: FastifyRequest, reply: FastifyReply) => {
const multipartForm = await parseMultipartFormData(request);
const jsonBody = (!multipartForm && request.body && typeof request.body === 'object')
? request.body as Record<string, unknown>
: null;
const requestedModel = typeof multipartForm?.get('model') === 'string'
? String(multipartForm.get('model')).trim()
: (typeof jsonBody?.model === 'string' ? jsonBody.model.trim() : '') || 'gpt-image-1';
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const downstreamPath = '/v1/images/edits';
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body: jsonBody || Object.fromEntries(multipartForm?.entries?.() || []),
});
const excludeChannelIds: number[] = [];
let retryCount = 0;
while (retryCount <= MAX_RETRIES) {
let selected = retryCount === 0
? await tokenRouter.selectChannel(requestedModel, downstreamPolicy)
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy);
if (!selected && retryCount === 0) {
await refreshModelsAndRebuildRoutes();
selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy);
}
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 = buildUpstreamUrl(selected.site.url, '/v1/images/edits');
const startTime = Date.now();
try {
const requestInit = multipartForm
? withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
Authorization: `Bearer ${selected.tokenValue}`,
},
body: cloneFormDataWithOverrides(multipartForm, {
model: selected.actualModel || requestedModel,
}) as any,
})
: withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify({
...(jsonBody || {}),
model: selected.actualModel || requestedModel,
}),
});
const upstream = await fetch(targetUrl, requestInit);
const text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
upstream.status,
Date.now() - startTime,
text,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
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, selected.actualModel);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId, estimatedCost, downstreamPath, clientContext);
return reply.code(upstream.status).send(data);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
0,
Date.now() - startTime,
err.message,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
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' },
});
}
}
});
app.post('/v1/images/variations', async (_request: FastifyRequest, reply: FastifyReply) => {
return reply.code(400).send({
error: {
message: 'Image variations are not supported',
type: 'invalid_request_error',
},
});
});
}
}
}
});
async function logProxy(
selected: any,
modelRequested: string,
status: string,
httpStatus: number,
latencyMs: number,
errorMessage: string | null,
retryCount: number,
downstreamApiKeyId: number | null = null,
estimatedCost = 0,
downstreamPath = '/v1/images/generations',
clientContext: DownstreamClientContext | null = null,
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({
clientKind: clientContext?.clientKind && clientContext.clientKind !== 'generic'
? clientContext.clientKind
: null,
sessionId: clientContext?.sessionId || null,
traceHint: clientContext?.traceHint || null,
downstreamPath,
errorMessage,
});
await insertProxyLog({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
accountId: selected.account.id,
downstreamApiKeyId,
modelRequested,
modelActual: selected.actualModel,
status,
httpStatus,
latencyMs,
promptTokens: 0,
completionTokens: 0,
totalTokens: 0,
estimatedCost,
clientFamily: clientContext?.clientKind || null,
clientAppId: clientContext?.clientAppId || null,
clientAppName: clientContext?.clientAppName || null,
clientConfidence: clientContext?.clientConfidence || null,
errorMessage: normalizedErrorMessage,
retryCount,
createdAt,
});
} catch (error) {
console.warn('[proxy/images] failed to write proxy log', error);
}
}
app.post('/v1/images/edits', async (request: FastifyRequest, reply: FastifyReply) => {
const multipartForm = await parseMultipartFormData(request);
const jsonBody = (!multipartForm && request.body && typeof request.body === 'object')
? request.body as Record<string, unknown>
: null;
const requestedModel = typeof multipartForm?.get('model') === 'string'
? String(multipartForm.get('model')).trim()
: (typeof jsonBody?.model === 'string' ? jsonBody.model.trim() : '') || 'gpt-image-1';
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const downstreamPath = '/v1/images/edits';
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body: jsonBody || Object.fromEntries(multipartForm?.entries?.() || []),
});
const excludeChannelIds: number[] = [];
let retryCount = 0;
while (retryCount <= MAX_RETRIES) {
let selected = retryCount === 0
? await tokenRouter.selectChannel(requestedModel, downstreamPolicy)
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy);
if (!selected && retryCount === 0) {
await refreshModelsAndRebuildRoutes();
selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy);
}
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 = buildUpstreamUrl(selected.site.url, '/v1/images/edits');
const startTime = Date.now();
try {
const requestInit = multipartForm
? withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
Authorization: `Bearer ${selected.tokenValue}`,
},
body: cloneFormDataWithOverrides(multipartForm, {
model: selected.actualModel || requestedModel,
}) as any,
}, getProxyUrlFromExtraConfig(selected.account.extraConfig))
: withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify({
...(jsonBody || {}),
model: selected.actualModel || requestedModel,
}),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig));
const upstream = await fetch(targetUrl, requestInit);
const text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
upstream.status,
Date.now() - startTime,
text,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
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, selected.actualModel);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId, estimatedCost, downstreamPath, clientContext);
return reply.code(upstream.status).send(data);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
0,
Date.now() - startTime,
err.message,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
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' },
});
}
}
});
app.post('/v1/images/variations', async (_request: FastifyRequest, reply: FastifyReply) => {
return reply.code(400).send({
error: {
message: 'Image variations are not supported',
type: 'invalid_request_error',
},
});
});
}
async function logProxy(
selected: any,
modelRequested: string,
status: string,
httpStatus: number,
latencyMs: number,
errorMessage: string | null,
retryCount: number,
downstreamApiKeyId: number | null = null,
estimatedCost = 0,
downstreamPath = '/v1/images/generations',
clientContext: DownstreamClientContext | null = null,
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({
clientKind: clientContext?.clientKind && clientContext.clientKind !== 'generic'
? clientContext.clientKind
: null,
sessionId: clientContext?.sessionId || null,
traceHint: clientContext?.traceHint || null,
downstreamPath,
errorMessage,
});
await insertProxyLog({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
accountId: selected.account.id,
downstreamApiKeyId,
modelRequested,
modelActual: selected.actualModel,
status,
httpStatus,
latencyMs,
promptTokens: 0,
completionTokens: 0,
totalTokens: 0,
estimatedCost,
clientFamily: clientContext?.clientKind || null,
clientAppId: clientContext?.clientAppId || null,
clientAppName: clientContext?.clientAppName || null,
clientConfidence: clientContext?.clientConfidence || null,
errorMessage: normalizedErrorMessage,
retryCount,
createdAt,
});
} catch (error) {
console.warn('[proxy/images] failed to write proxy log', error);
}
}
src/server/routes/proxy/search.ts
+2 -1
View File
@@ -7,6 +7,7 @@ import { isTokenExpiredError } from '../../services/alertRules.js';
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { getProxyUrlFromExtraConfig } from '../../services/accountExtraConfig.js';
import { composeProxyLogMessage } from './logPathMeta.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { getProxyAuthContext } from '../../middleware/auth.js';
@@ -101,7 +102,7 @@ export async function searchProxyRoute(app: FastifyInstance) {
Authorization: `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify(forwardBody),
}));
}, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
const text = await upstream.text();
if (!upstream.ok) {
src/server/routes/proxy/videos.ts
+4 -2
View File
@@ -8,6 +8,7 @@ import { estimateProxyCost } from '../../services/modelPricingService.js';
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteProxyRequestInit, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { getProxyUrlFromExtraConfig } from '../../services/accountExtraConfig.js';
import { cloneFormDataWithOverrides, ensureMultipartBufferParser, parseMultipartFormData } from './multipart.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import {
@@ -75,6 +76,7 @@ export async function videosProxyRoute(app: FastifyInstance) {
const startTime = Date.now();
try {
const accountProxy = getProxyUrlFromExtraConfig(selected.account.extraConfig);
const requestInit = multipartForm
? withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
@@ -84,7 +86,7 @@ export async function videosProxyRoute(app: FastifyInstance) {
body: cloneFormDataWithOverrides(multipartForm, {
model: selected.actualModel || requestedModel,
}) as any,
})
}, accountProxy)
: withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
@@ -95,7 +97,7 @@ export async function videosProxyRoute(app: FastifyInstance) {
...(jsonBody || {}),
model: selected.actualModel || requestedModel,
}),
});
}, accountProxy);
const upstream = await fetch(targetUrl, requestInit);
const text = await upstream.text();