diff --git a/src/server/proxy-core/surfaces/chatSurface.ts b/src/server/proxy-core/surfaces/chatSurface.ts index bb3e20f..b1809ef 100644 --- a/src/server/proxy-core/surfaces/chatSurface.ts +++ b/src/server/proxy-core/surfaces/chatSurface.ts @@ -7,7 +7,7 @@ import { isTokenExpiredError } from '../../services/alertRules.js'; import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js'; import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js'; -import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; +import { resolveChannelProxyUrl, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; import { type DownstreamFormat } from '../../transformers/shared/normalized.js'; import { buildClaudeCountTokensUpstreamRequest, @@ -204,6 +204,7 @@ export async function handleChatSurfaceRequest( runtime: endpointRequest.runtime, }; }; + const channelProxyUrl = resolveChannelProxyUrl(selected.site, selected.account.extraConfig); const dispatchRequest = ( compatibilityRequest: BuiltEndpointRequest, targetUrl?: string, @@ -216,7 +217,7 @@ export async function handleChatSurfaceRequest( method: 'POST', headers: requestForFetch.headers, body: JSON.stringify(requestForFetch.body), - }), + }, channelProxyUrl), }) ); const endpointStrategy = downstreamTransformer.compatibility.createEndpointStrategy({ @@ -912,7 +913,7 @@ export async function handleClaudeCountTokensSurfaceRequest( method: 'POST', headers: requestForFetch.headers, body: JSON.stringify(requestForFetch.body), - }), + }, resolveChannelProxyUrl(selected.site, selected.account.extraConfig)), }); if (upstream.status === 401 && oauth) { @@ -932,7 +933,7 @@ export async function handleClaudeCountTokensSurfaceRequest( method: 'POST', headers: requestForFetch.headers, body: JSON.stringify(requestForFetch.body), - }), + }, resolveChannelProxyUrl(selected.site, selected.account.extraConfig)), }); } catch { // Fall through to the regular upstream error handling below. diff --git a/src/server/proxy-core/surfaces/geminiSurface.ts b/src/server/proxy-core/surfaces/geminiSurface.ts index bbeb0be..03fae42 100644 --- a/src/server/proxy-core/surfaces/geminiSurface.ts +++ b/src/server/proxy-core/surfaces/geminiSurface.ts @@ -9,7 +9,7 @@ import { isModelAllowedByPolicyOrAllowedRoutes } from '../../services/downstream import { tokenRouter } from '../../services/tokenRouter.js'; import { buildOauthProviderHeaders } from '../../services/oauth/service.js'; import { getOauthInfoFromExtraConfig } from '../../services/oauth/oauthAccount.js'; -import { withSiteProxyRequestInit } from '../../services/siteProxy.js'; +import { resolveChannelProxyUrl, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; import { getDownstreamRoutingPolicy } from '../../routes/proxy/downstreamPolicy.js'; import { executeEndpointFlow, type BuiltEndpointRequest } from '../../routes/proxy/endpointFlow.js'; @@ -470,6 +470,7 @@ export async function geminiProxyRoute(app: FastifyInstance) { selected.tokenValue, query, ); + const channelProxyUrl = resolveChannelProxyUrl(selected.site, selected.account.extraConfig); const upstream = isInternalGemini ? await dispatchRuntimeRequest({ siteUrl: selected.site.url, @@ -489,11 +490,11 @@ export async function geminiProxyRoute(app: FastifyInstance) { : (isStreamAction ? 'streamGenerateContent' : 'generateContent'), }, }, - buildInit: async (requestUrl, requestForFetch) => withSiteProxyRequestInit(requestUrl, { + buildInit: async (_requestUrl, requestForFetch) => withSiteRecordProxyRequestInit(selected.site, { method: 'POST', headers: requestForFetch.headers, body: JSON.stringify(requestForFetch.body), - }), + }, channelProxyUrl), }) : await fetch(targetUrl, { method: 'POST', @@ -741,16 +742,17 @@ export async function geminiProxyRoute(app: FastifyInstance) { runtime: endpointRequest.runtime, }; }; + const channelProxyUrl = resolveChannelProxyUrl(selected.site, selected.account.extraConfig); const dispatchRequest = (compatibilityRequest: BuiltEndpointRequest, targetUrl?: string) => ( dispatchRuntimeRequest({ siteUrl: selected.site.url, targetUrl, request: compatibilityRequest, - buildInit: async (requestUrl, requestForFetch) => withSiteProxyRequestInit(requestUrl, { + buildInit: async (_requestUrl, requestForFetch) => withSiteRecordProxyRequestInit(selected.site, { method: 'POST', headers: requestForFetch.headers, body: JSON.stringify(requestForFetch.body), - }), + }, channelProxyUrl), }) ); const endpointStrategy = createChatEndpointStrategy({ diff --git a/src/server/proxy-core/surfaces/openAiResponsesSurface.ts b/src/server/proxy-core/surfaces/openAiResponsesSurface.ts index 3f07437..006284e 100644 --- a/src/server/proxy-core/surfaces/openAiResponsesSurface.ts +++ b/src/server/proxy-core/surfaces/openAiResponsesSurface.ts @@ -7,7 +7,7 @@ import { isTokenExpiredError } from '../../services/alertRules.js'; import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js'; import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js'; -import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; +import { resolveChannelProxyUrl, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; import { openAiResponsesTransformer } from '../../transformers/openai/responses/index.js'; import { buildUpstreamEndpointRequest, @@ -304,6 +304,7 @@ export async function handleOpenAiResponsesSurfaceRequest( runtime: endpointRequest.runtime, }; }; + const channelProxyUrl = resolveChannelProxyUrl(selected.site, selected.account.extraConfig); const dispatchRequest = (compatibilityRequest: BuiltEndpointRequest, targetUrl?: string) => ( dispatchRuntimeRequest({ siteUrl: selected.site.url, @@ -313,7 +314,7 @@ export async function handleOpenAiResponsesSurfaceRequest( method: 'POST', headers: requestForFetch.headers, body: JSON.stringify(requestForFetch.body), - }), + }, channelProxyUrl), }) ); const endpointStrategy = openAiResponsesTransformer.compatibility.createEndpointStrategy({ diff --git a/src/server/routes/api/accountTokens.ts b/src/server/routes/api/accountTokens.ts index 770ef41..2bf36b4 100644 --- a/src/server/routes/api/accountTokens.ts +++ b/src/server/routes/api/accountTokens.ts @@ -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 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) { diff --git a/src/server/routes/api/accounts.ts b/src/server/routes/api/accounts.ts index 689241e..5ee9b7e 100644 --- a/src/server/routes/api/accounts.ts +++ b/src/server/routes/api/accounts.ts @@ -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(); diff --git a/src/server/routes/proxy/completions.ts b/src/server/routes/proxy/completions.ts index 138410c..5d49328 100644 --- a/src/server/routes/proxy/completions.ts +++ b/src/server/routes/proxy/completions.ts @@ -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'); diff --git a/src/server/routes/proxy/embeddings.ts b/src/server/routes/proxy/embeddings.ts index a5f8236..3741523 100644 --- a/src/server/routes/proxy/embeddings.ts +++ b/src/server/routes/proxy/embeddings.ts @@ -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, - 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, + 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); + } +} diff --git a/src/server/routes/proxy/images.ts b/src/server/routes/proxy/images.ts index b0d3488..4d12c31 100644 --- a/src/server/routes/proxy/images.ts +++ b/src/server/routes/proxy/images.ts @@ -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, - 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, + 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 - : 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, - 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 + : 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, + 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); + } +} diff --git a/src/server/routes/proxy/search.ts b/src/server/routes/proxy/search.ts index 23ee34c..d91cbd4 100644 --- a/src/server/routes/proxy/search.ts +++ b/src/server/routes/proxy/search.ts @@ -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) { diff --git a/src/server/routes/proxy/videos.ts b/src/server/routes/proxy/videos.ts index 3ca359e..10e5a96 100644 --- a/src/server/routes/proxy/videos.ts +++ b/src/server/routes/proxy/videos.ts @@ -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(); diff --git a/src/server/services/accountExtraConfig.test.ts b/src/server/services/accountExtraConfig.test.ts index 3189481..12fd432 100644 --- a/src/server/services/accountExtraConfig.test.ts +++ b/src/server/services/accountExtraConfig.test.ts @@ -3,6 +3,7 @@ import { buildStoredSub2ApiSubscriptionSummary, getCredentialModeFromExtraConfig, getPlatformUserIdFromExtraConfig, + getProxyUrlFromExtraConfig, getSub2ApiAuthFromExtraConfig, getSub2ApiSubscriptionFromExtraConfig, guessPlatformUserIdFromUsername, @@ -72,6 +73,20 @@ describe('accountExtraConfig', () => { }))).toBeNull(); }); + it('reads proxyUrl from extra config', () => { + expect(getProxyUrlFromExtraConfig(JSON.stringify({ proxyUrl: 'http://127.0.0.1:7890' }))).toBe('http://127.0.0.1:7890'); + expect(getProxyUrlFromExtraConfig(JSON.stringify({ proxyUrl: ' socks5://proxy.local:1080 ' }))).toBe('socks5://proxy.local:1080'); + }); + + it('returns null for missing or empty proxyUrl', () => { + expect(getProxyUrlFromExtraConfig(JSON.stringify({}))).toBeNull(); + expect(getProxyUrlFromExtraConfig(JSON.stringify({ proxyUrl: '' }))).toBeNull(); + expect(getProxyUrlFromExtraConfig(JSON.stringify({ proxyUrl: ' ' }))).toBeNull(); + expect(getProxyUrlFromExtraConfig(null)).toBeNull(); + expect(getProxyUrlFromExtraConfig(undefined)).toBeNull(); + expect(getProxyUrlFromExtraConfig('invalid-json')).toBeNull(); + }); + it('treats auto-mode api token connections as direct-account routable', () => { expect(supportsDirectAccountRoutingConnection({ accessToken: '', diff --git a/src/server/services/accountExtraConfig.ts b/src/server/services/accountExtraConfig.ts index 28e1ce7..393abf8 100644 --- a/src/server/services/accountExtraConfig.ts +++ b/src/server/services/accountExtraConfig.ts @@ -114,6 +114,11 @@ export function normalizeCredentialMode(raw: unknown): AccountCredentialMode | u return normalized as AccountCredentialMode; } +export function getProxyUrlFromExtraConfig(extraConfig?: string | null): string | null { + const parsed = parseExtraConfig(extraConfig); + return normalizeNonEmptyString(parsed.proxyUrl) ?? null; +} + export function getPlatformUserIdFromExtraConfig(extraConfig?: string | null): number | undefined { const parsed = parseExtraConfig(extraConfig); return normalizeUserId(parsed.platformUserId); diff --git a/src/server/services/balanceService.ts b/src/server/services/balanceService.ts index 1662580..5540a12 100644 --- a/src/server/services/balanceService.ts +++ b/src/server/services/balanceService.ts @@ -7,6 +7,7 @@ import { buildStoredSub2ApiSubscriptionSummary, getAutoReloginConfig, getCredentialModeFromExtraConfig, + getProxyUrlFromExtraConfig, getSub2ApiAuthFromExtraConfig, mergeAccountExtraConfig, resolvePlatformUserId, @@ -15,7 +16,7 @@ import { decryptAccountPassword } from './accountCredentialService.js'; import { extractRuntimeHealth, setAccountRuntimeHealth } from './accountHealthService.js'; import { updateTodayIncomeSnapshot } from './todayIncomeRewardService.js'; import type { BalanceInfo } from './platforms/base.js'; -import { withSiteProxyRequestInit, withSiteRecordProxyRequestInit } from './siteProxy.js'; +import { withAccountProxyOverride, withSiteProxyRequestInit, withSiteRecordProxyRequestInit } from './siteProxy.js'; function isSiteDisabled(status?: string | null): boolean { return (status || 'active') === 'disabled'; @@ -200,7 +201,7 @@ async function refreshSub2ApiManagedSession(params: { method: 'POST', headers, body: JSON.stringify({ refresh_token: refreshToken }), - })); + }, getProxyUrlFromExtraConfig(params.account.extraConfig))); payload = await response.json().catch(() => null); } catch (err: any) { throw new Error(err?.message || 'sub2api token refresh request failed'); @@ -317,7 +318,10 @@ async function tryAutoRelogin(account: any, site: any): Promise { const password = decryptAccountPassword(relogin.passwordCipher); if (!password) return null; - const loginResult = await adapter.login(site.url, relogin.username, password); + const loginResult = await withAccountProxyOverride( + getProxyUrlFromExtraConfig(account.extraConfig), + () => adapter.login(site.url, relogin.username, password), + ); if (!loginResult.success || !loginResult.accessToken) return null; await db.update(schema.accounts) @@ -378,23 +382,25 @@ export async function refreshBalance(accountId: number) { let activeExtraConfig = account.extraConfig; let balanceInfo: BalanceInfo | null = null; + const accountProxyUrl = getProxyUrlFromExtraConfig(account.extraConfig); + if (isSub2ApiPlatform(site.platform)) { const managedAuth = getSub2ApiAuthFromExtraConfig(activeExtraConfig); if (managedAuth?.refreshToken && isNearTokenExpiry(managedAuth.tokenExpiresAt)) { try { - const refreshed = await refreshSub2ApiManagedSession({ + const refreshed = await withAccountProxyOverride(accountProxyUrl, () => refreshSub2ApiManagedSession({ account, site, currentAccessToken: activeAccessToken, currentExtraConfig: activeExtraConfig, - }); + })); activeAccessToken = refreshed.accessToken; activeExtraConfig = refreshed.extraConfig; } catch {} } } - - const readBalance = async (token: string) => adapter.getBalance(site.url, token, platformUserId); + const readBalance = async (token: string) => withAccountProxyOverride(accountProxyUrl, + () => adapter.getBalance(site.url, token, platformUserId)); const handleBalanceError = async (err: any) => { const message = appendSessionTokenRebindHint(err?.message || 'unknown error'); setAccountRuntimeHealth(account.id, { @@ -424,12 +430,12 @@ export async function refreshBalance(accountId: number) { if (canTryManagedSub2ApiRefresh) { try { - const refreshed = await refreshSub2ApiManagedSession({ + const refreshed = await withAccountProxyOverride(accountProxyUrl, () => refreshSub2ApiManagedSession({ account, site, currentAccessToken: activeAccessToken, currentExtraConfig: activeExtraConfig, - }); + })); activeAccessToken = refreshed.accessToken; activeExtraConfig = refreshed.extraConfig; balanceInfo = await readBalance(activeAccessToken); @@ -462,12 +468,12 @@ export async function refreshBalance(accountId: number) { supportsTodayIncomeLogFallback(site.platform) ) { try { - const fallbackIncome = await fetchTodayIncomeFromLogs({ + const fallbackIncome = await withAccountProxyOverride(accountProxyUrl, () => fetchTodayIncomeFromLogs({ baseUrl: site.url, accessToken: activeAccessToken, platform: site.platform, platformUserId, - }); + })); if (typeof fallbackIncome === 'number' && Number.isFinite(fallbackIncome)) { balanceInfo.todayIncome = fallbackIncome; } diff --git a/src/server/services/checkinService.ts b/src/server/services/checkinService.ts index 6688b41..46fd1cd 100644 --- a/src/server/services/checkinService.ts +++ b/src/server/services/checkinService.ts @@ -8,6 +8,7 @@ import { refreshBalance } from './balanceService.js'; import { parseCheckinRewardAmount } from './checkinRewardParser.js'; import { getAutoReloginConfig, + getProxyUrlFromExtraConfig, getPlatformUserIdFromExtraConfig, guessPlatformUserIdFromUsername, mergeAccountExtraConfig, @@ -16,6 +17,7 @@ import { import { decryptAccountPassword } from './accountCredentialService.js'; import { setAccountRuntimeHealth } from './accountHealthService.js'; import { formatUtcSqlDateTime } from './localTimeService.js'; +import { withAccountProxyOverride } from './siteProxy.js'; type CheckinExecutionStatus = 'success' | 'failed' | 'skipped'; @@ -100,7 +102,10 @@ async function tryAutoRelogin(account: any, site: any): Promise { const password = decryptAccountPassword(relogin.passwordCipher); if (!password) return null; - const result = await adapter.login(site.url, relogin.username, password); + const result = await withAccountProxyOverride( + getProxyUrlFromExtraConfig(account.extraConfig), + () => adapter.login(site.url, relogin.username, password), + ); if (!result.success || !result.accessToken) return null; await db.update(schema.accounts) @@ -172,14 +177,17 @@ export async function checkinAccount(accountId: number, options?: { skipEvent?: : guessPlatformUserIdFromUsername(account.username); const platformUserId = resolvePlatformUserId(account.extraConfig, account.username); + const accountProxyUrl = getProxyUrlFromExtraConfig(account.extraConfig); let activeAccessToken = account.accessToken; - let result = await adapter.checkin(site.url, activeAccessToken, platformUserId); + let result = await withAccountProxyOverride(accountProxyUrl, + () => adapter.checkin(site.url, activeAccessToken, platformUserId)); if (!result.success && shouldAttemptAutoRelogin(result.message)) { const refreshedAccessToken = await tryAutoRelogin(account, site); if (refreshedAccessToken) { activeAccessToken = refreshedAccessToken; - result = await adapter.checkin(site.url, activeAccessToken, platformUserId); + result = await withAccountProxyOverride(accountProxyUrl, + () => adapter.checkin(site.url, activeAccessToken, platformUserId)); } } diff --git a/src/server/services/modelService.discovery.test.ts b/src/server/services/modelService.discovery.test.ts index 9de6827..3cef888 100644 --- a/src/server/services/modelService.discovery.test.ts +++ b/src/server/services/modelService.discovery.test.ts @@ -7,6 +7,18 @@ import { eq } from 'drizzle-orm'; const getApiTokenMock = vi.fn(); const getModelsMock = vi.fn(); const undiciFetchMock = vi.fn(); +const proxyAgentCtorMock = vi.fn(); + +class MockProxyAgent { + readonly proxyUrl: string; + + constructor(proxyUrl: string) { + this.proxyUrl = proxyUrl; + proxyAgentCtorMock(proxyUrl); + } +} + +class MockAgent {} vi.mock('./platforms/index.js', () => ({ getAdapter: () => ({ @@ -17,6 +29,8 @@ vi.mock('./platforms/index.js', () => ({ vi.mock('undici', () => ({ fetch: (...args: unknown[]) => undiciFetchMock(...args), + ProxyAgent: MockProxyAgent, + Agent: MockAgent, })); type DbModule = typeof import('../db/index.js'); @@ -47,6 +61,7 @@ describe('refreshModelsForAccount credential discovery', () => { getApiTokenMock.mockReset(); getModelsMock.mockReset(); undiciFetchMock.mockReset(); + proxyAgentCtorMock.mockReset(); await db.delete(schema.routeChannels).run(); await db.delete(schema.tokenRoutes).run(); @@ -595,6 +610,115 @@ describe('refreshModelsForAccount credential discovery', () => { expect(parsed.runtimeHealth?.state).toBe('unhealthy'); }); + it('applies account proxy override to codex oauth cloud discovery requests', async () => { + getApiTokenMock.mockResolvedValue(null); + getModelsMock.mockRejectedValue(new Error('codex oauth discovery should not call adapter.getModels')); + undiciFetchMock.mockResolvedValue({ + ok: true, + status: 200, + json: async () => ({ + models: [ + { id: 'gpt-5.4' }, + ], + }), + text: async () => JSON.stringify({ ok: true }), + }); + + const site = await db.insert(schema.sites).values({ + name: 'codex-account-proxy-site', + url: 'https://chatgpt.com/backend-api/codex', + platform: 'codex', + status: 'active', + }).returning().get(); + + const account = await db.insert(schema.accounts).values({ + siteId: site.id, + username: 'codex-proxy-user@example.com', + accessToken: 'oauth-access-token', + apiToken: null, + status: 'active', + extraConfig: JSON.stringify({ + credentialMode: 'session', + proxyUrl: 'http://127.0.0.1:7890', + oauth: { + provider: 'codex', + accountId: 'chatgpt-account-proxy', + email: 'codex-proxy-user@example.com', + planType: 'plus', + }, + }), + }).returning().get(); + + const result = await refreshModelsForAccount(account.id); + + expect(result).toMatchObject({ + accountId: account.id, + refreshed: true, + status: 'success', + modelCount: 1, + modelsPreview: ['gpt-5.4'], + }); + expect(undiciFetchMock).toHaveBeenCalledTimes(1); + expect(proxyAgentCtorMock).toHaveBeenCalledWith('http://127.0.0.1:7890'); + expect(undiciFetchMock.mock.calls[0]?.[1]).toMatchObject({ + method: 'GET', + dispatcher: expect.any(MockProxyAgent), + }); + }); + + it('applies account proxy override to gemini oauth validation requests', async () => { + getApiTokenMock.mockResolvedValue(null); + getModelsMock.mockRejectedValue(new Error('gemini oauth validation should not call adapter.getModels')); + undiciFetchMock.mockResolvedValue({ + ok: true, + status: 200, + json: async () => ({ + state: 'ENABLED', + }), + text: async () => JSON.stringify({ state: 'ENABLED' }), + }); + + const site = await db.insert(schema.sites).values({ + name: 'gemini-account-proxy-site', + url: 'https://gemini.example.com', + platform: 'gemini', + status: 'active', + }).returning().get(); + + const account = await db.insert(schema.accounts).values({ + siteId: site.id, + username: 'gemini-proxy-user@example.com', + accessToken: 'gemini-access-token', + apiToken: null, + status: 'active', + extraConfig: JSON.stringify({ + credentialMode: 'session', + proxyUrl: 'http://127.0.0.1:1080', + oauth: { + provider: 'gemini-cli', + projectId: 'project-proxy-demo', + email: 'gemini-proxy-user@example.com', + }, + }), + }).returning().get(); + + const result = await refreshModelsForAccount(account.id); + + expect(result).toMatchObject({ + accountId: account.id, + refreshed: true, + status: 'success', + modelCount: expect.any(Number), + }); + expect(undiciFetchMock).toHaveBeenCalledTimes(1); + expect(proxyAgentCtorMock).toHaveBeenCalledWith('http://127.0.0.1:1080'); + expect(String(undiciFetchMock.mock.calls[0]?.[0] || '')).toContain('/projects/project-proxy-demo/services/'); + expect(undiciFetchMock.mock.calls[0]?.[1]).toMatchObject({ + method: 'GET', + dispatcher: expect.any(MockProxyAgent), + }); + }); + it('discovers antigravity oauth models via fetchAvailableModels fallback using the oauth project id', async () => { getApiTokenMock.mockResolvedValue(null); getModelsMock.mockRejectedValue(new Error('antigravity oauth discovery should not call adapter.getModels')); diff --git a/src/server/services/modelService.ts b/src/server/services/modelService.ts index 321b171..9f5dbbf 100644 --- a/src/server/services/modelService.ts +++ b/src/server/services/modelService.ts @@ -10,6 +10,8 @@ import { isUsableAccountToken, } from './accountTokenService.js'; import { + getCredentialModeFromExtraConfig, + getProxyUrlFromExtraConfig, mergeAccountExtraConfig, requiresManagedAccountTokens, resolvePlatformUserId, @@ -18,7 +20,7 @@ import { import { invalidateTokenRouterCache } from './tokenRouter.js'; import { setAccountRuntimeHealth } from './accountHealthService.js'; import { clearAllRouteDecisionSnapshots } from './routeDecisionSnapshotStore.js'; -import { withSiteRecordProxyRequestInit } from './siteProxy.js'; +import { withAccountProxyOverride, withExplicitProxyRequestInit, withSiteRecordProxyRequestInit } from './siteProxy.js'; import { getCodexOauthInfoFromExtraConfig, isCodexPlatform } from './oauth/codexAccount.js'; import { buildOauthInfo, getOauthInfoFromExtraConfig } from './oauth/oauthAccount.js'; import { CLAUDE_DEFAULT_ANTHROPIC_VERSION } from './oauth/claudeProvider.js'; @@ -321,7 +323,7 @@ async function validateGeminiCliOauthConnection(input: { } const response = await fetch( `https://serviceusage.googleapis.com/v1/projects/${encodeURIComponent(projectId)}/services/${encodeURIComponent(GEMINI_CLI_REQUIRED_SERVICE)}`, - { + withExplicitProxyRequestInit(getProxyUrlFromExtraConfig(input.account.extraConfig), { method: 'GET', headers: { Authorization: `Bearer ${accessToken}`, @@ -329,7 +331,7 @@ async function validateGeminiCliOauthConnection(input: { 'User-Agent': GEMINI_CLI_USER_AGENT, 'X-Goog-Api-Client': GEMINI_CLI_GOOGLE_API_CLIENT, }, - }, + }, true), ); if (!response.ok) { const text = await response.text().catch(() => ''); @@ -493,6 +495,7 @@ export async function refreshModelsForAccount(accountId: number): Promise discoverCodexModelsFromCloud({ site, account }), + () => withAccountProxyOverride(accountProxyUrl, + () => discoverCodexModelsFromCloud({ site, account })), MODEL_DISCOVERY_TIMEOUT_MS, `codex model discovery timeout (${Math.round(MODEL_DISCOVERY_TIMEOUT_MS / 1000)}s)`, ); @@ -592,7 +596,8 @@ export async function refreshModelsForAccount(accountId: number): Promise validateClaudeOauthConnection({ site, account }), + () => withAccountProxyOverride(accountProxyUrl, + () => validateClaudeOauthConnection({ site, account })), MODEL_DISCOVERY_TIMEOUT_MS, `claude oauth validation timeout (${Math.round(MODEL_DISCOVERY_TIMEOUT_MS / 1000)}s)`, ); @@ -658,7 +663,8 @@ export async function refreshModelsForAccount(accountId: number): Promise validateGeminiCliOauthConnection({ account }), + () => withAccountProxyOverride(accountProxyUrl, + () => validateGeminiCliOauthConnection({ account })), MODEL_DISCOVERY_TIMEOUT_MS, `gemini cli oauth validation timeout (${Math.round(MODEL_DISCOVERY_TIMEOUT_MS / 1000)}s)`, ); @@ -724,7 +730,8 @@ export async function refreshModelsForAccount(accountId: number): Promise discoverAntigravityModelsFromCloud({ site, account }), + () => withAccountProxyOverride(accountProxyUrl, + () => discoverAntigravityModelsFromCloud({ site, account })), MODEL_DISCOVERY_TIMEOUT_MS, `antigravity model discovery timeout (${Math.round(MODEL_DISCOVERY_TIMEOUT_MS / 1000)}s)`, ); @@ -799,7 +806,8 @@ export async function refreshModelsForAccount(accountId: number): Promise adapter.getApiToken(site.url, account.accessToken, platformUserId), + () => withAccountProxyOverride(accountProxyUrl, + () => adapter.getApiToken(site.url, account.accessToken, platformUserId)), API_TOKEN_DISCOVERY_TIMEOUT_MS, `api token discovery timeout (${Math.round(API_TOKEN_DISCOVERY_TIMEOUT_MS / 1000)}s)`, ); @@ -881,7 +889,8 @@ export async function refreshModelsForAccount(accountId: number): Promise adapter.getModels(site.url, credential, platformUserId), + () => withAccountProxyOverride(accountProxyUrl, + () => adapter.getModels(site.url, credential, platformUserId)), MODEL_DISCOVERY_TIMEOUT_MS, `model discovery timeout (${Math.round(MODEL_DISCOVERY_TIMEOUT_MS / 1000)}s)`, ), @@ -908,7 +917,8 @@ export async function refreshModelsForAccount(accountId: number): Promise adapter.getModels(site.url, token.token, platformUserId), + () => withAccountProxyOverride(accountProxyUrl, + () => adapter.getModels(site.url, token.token, platformUserId)), MODEL_DISCOVERY_TIMEOUT_MS, `model discovery timeout (${Math.round(MODEL_DISCOVERY_TIMEOUT_MS / 1000)}s)`, ), diff --git a/src/server/services/siteProxy.test.ts b/src/server/services/siteProxy.test.ts index ae327e6..218e72a 100644 --- a/src/server/services/siteProxy.test.ts +++ b/src/server/services/siteProxy.test.ts @@ -217,4 +217,77 @@ describe('siteProxy', () => { expect(headers.get('x-request-id')).toBe('req-1'); expect('dispatcher' in requestInit).toBe(true); }); + + it('resolveChannelProxyUrl prefers account proxy over site proxy', async () => { + const { resolveChannelProxyUrl } = await import('./siteProxy.js'); + + const accountConfig = JSON.stringify({ proxyUrl: 'http://account-proxy:8080' }); + const siteWithProxy = { useSystemProxy: true }; + + expect(resolveChannelProxyUrl(siteWithProxy, accountConfig)).toBe('http://account-proxy:8080'); + expect(resolveChannelProxyUrl(siteWithProxy, null)).toBeNull(); + expect(resolveChannelProxyUrl(siteWithProxy, JSON.stringify({}))).toBeNull(); + }); + + it('withSiteRecordProxyRequestInit uses account proxy when provided', async () => { + const { withSiteRecordProxyRequestInit } = await import('./siteProxy.js'); + const result = withSiteRecordProxyRequestInit( + { useSystemProxy: false }, + { method: 'POST' }, + 'http://account-proxy:8080', + ); + expect('dispatcher' in result).toBe(true); + }); + + it('withSiteRecordProxyRequestInit ignores invalid account proxy', async () => { + const { withSiteRecordProxyRequestInit } = await import('./siteProxy.js'); + const result = withSiteRecordProxyRequestInit( + { useSystemProxy: false }, + { method: 'POST' }, + 'not-a-url', + ); + expect('dispatcher' in result).toBe(false); + }); + + it('withAccountProxyOverride sets ALS context for nested proxy calls', async () => { + const { withAccountProxyOverride, withSiteProxyRequestInit } = await import('./siteProxy.js'); + + await db.insert(schema.sites).values({ + name: 'als-site', + url: 'https://als-site.example.com', + platform: 'new-api', + }).run(); + + const result = await withAccountProxyOverride( + 'http://account-als-proxy:9090', + async () => { + return withSiteProxyRequestInit('https://als-site.example.com/v1/models', { + method: 'GET', + }); + }, + ); + + expect('dispatcher' in result).toBe(true); + }); + + it('withAccountProxyOverride skips ALS when proxy is null', async () => { + const { withAccountProxyOverride, withSiteProxyRequestInit } = await import('./siteProxy.js'); + + await db.insert(schema.sites).values({ + name: 'als-null-site', + url: 'https://als-null-site.example.com', + platform: 'new-api', + }).run(); + + const result = await withAccountProxyOverride( + null, + async () => { + return withSiteProxyRequestInit('https://als-null-site.example.com/v1/models', { + method: 'GET', + }); + }, + ); + + expect('dispatcher' in result).toBe(false); + }); }); diff --git a/src/server/services/siteProxy.ts b/src/server/services/siteProxy.ts index 75b0949..59ac42f 100644 --- a/src/server/services/siteProxy.ts +++ b/src/server/services/siteProxy.ts @@ -1,3 +1,4 @@ +import { AsyncLocalStorage } from 'node:async_hooks'; import { db, schema } from '../db/index.js'; import { eq } from 'drizzle-orm'; import { config } from '../config.js'; @@ -8,6 +9,7 @@ import { SocksClient } from 'socks'; import type { Dispatcher, RequestInit as UndiciRequestInit } from 'undici'; import { Agent as UndiciAgent, ProxyAgent } from 'undici'; import { mergeHeadersWithSiteCustomHeaders } from './siteCustomHeaders.js'; +import { getProxyUrlFromExtraConfig } from './accountExtraConfig.js'; const SITE_PROXY_CACHE_TTL_MS = 3_000; const SUPPORTED_PROXY_PROTOCOLS = new Set([ @@ -60,6 +62,17 @@ let siteProxyCache: { const dispatcherCache = new Map(); +const accountProxyOverride = new AsyncLocalStorage(); + +export function withAccountProxyOverride( + proxyUrl: string | null | undefined, + fn: () => Promise, +): Promise { + const normalized = normalizeSiteProxyUrl(proxyUrl); + if (!normalized) return fn(); + return accountProxyOverride.run(normalized, fn); +} + type ParsedSocksProxyConfig = { shouldLookup: boolean; proxy: { @@ -145,19 +158,23 @@ async function getCachedSiteProxyRows(nowMs = Date.now()): Promise(null); @@ -652,21 +653,23 @@ export default function Accounts() { } }; + const parseAccountExtraConfig = (account: any): Record => { + try { return JSON.parse(account?.extraConfig || '{}') || {}; } + catch { return {}; } + }; + const extractManagedSub2ApiAuth = (account: any) => { - try { - const parsed = JSON.parse(account?.extraConfig || '{}'); - const auth = parsed?.sub2apiAuth || {}; - return { - refreshToken: typeof auth.refreshToken === 'string' ? auth.refreshToken : '', - tokenExpiresAt: auth.tokenExpiresAt ? String(auth.tokenExpiresAt) : '', - }; - } catch { - return { refreshToken: '', tokenExpiresAt: '' }; - } + const parsed = parseAccountExtraConfig(account); + const auth = parsed?.sub2apiAuth || {}; + return { + refreshToken: typeof auth.refreshToken === 'string' ? auth.refreshToken : '', + tokenExpiresAt: auth.tokenExpiresAt ? String(auth.tokenExpiresAt) : '', + }; }; const openEditPanel = (account: any) => { const managedAuth = extractManagedSub2ApiAuth(account); + const proxyUrl = parseAccountExtraConfig(account)?.proxyUrl || ''; closeAddPanel(); setRebindTarget(null); setEditingAccount(account); @@ -680,6 +683,7 @@ export default function Accounts() { isPinned: !!account?.isPinned, refreshToken: managedAuth.refreshToken, tokenExpiresAt: managedAuth.tokenExpiresAt, + proxyUrl, }); }; @@ -702,6 +706,7 @@ export default function Accounts() { isPinned: editForm.isPinned, refreshToken: editForm.refreshToken.trim() || null, tokenExpiresAt: editForm.tokenExpiresAt.trim() ? Number.parseInt(editForm.tokenExpiresAt.trim(), 10) : null, + proxyUrl: editForm.proxyUrl.trim() || null, }); toast.success('账号已更新'); closeEditPanel(); @@ -786,12 +791,10 @@ export default function Accounts() { }; const extractPlatformUserId = (account: any): string => { - try { - const parsed = JSON.parse(account?.extraConfig || '{}'); - const raw = parsed?.platformUserId; - const value = Number.parseInt(String(raw ?? ''), 10); - if (Number.isFinite(value) && value > 0) return String(value); - } catch { } + const parsed = parseAccountExtraConfig(account); + const raw = parsed?.platformUserId; + const value = Number.parseInt(String(raw ?? ''), 10); + if (Number.isFinite(value) && value > 0) return String(value); const guessed = Number.parseInt(String(account?.username || '').match(/(\d{3,8})$/)?.[1] || '', 10); return Number.isFinite(guessed) && guessed > 0 ? String(guessed) : ''; }; @@ -1542,6 +1545,15 @@ export default function Accounts() { onChange={(e) => setEditForm((prev) => ({ ...prev, apiToken: e.target.value }))} style={{ ...inputStyle, fontFamily: 'var(--font-mono)' }} /> + setEditForm((prev) => ({ ...prev, proxyUrl: e.target.value }))} + style={inputStyle} + /> +
+ 覆盖站点和系统代理，留空则使用站点设置。支持 http/https/socks5 协议。 +
{((editingAccount?.site?.platform || '').toLowerCase() === 'sub2api') && ( <> {connectionMode === 'apikey' ? 'API Key' : 'Session'} + {parseAccountExtraConfig(a)?.proxyUrl && ( + 代理 + )} )} > @@ -1799,6 +1814,9 @@ export default function Accounts() { {connectionMode === 'apikey' ? 'API Key' : 'Session'} + {parseAccountExtraConfig(a)?.proxyUrl && ( + 代理 + )}