LaPaoPao/BoosAPI
1
0
Fork 0
forked from boos/BoosAPI
Code Pull Requests 1 Activity

[codex] Split site API endpoint pool from panel host (#373)

Browse Source 
* feat: add site api endpoint schema scaffolding

* feat: wire site api endpoints into backup migration

* feat: add site api endpoint selection service

* feat: add site api endpoint contracts

* feat: route discovery through site api endpoints

* feat: route proxy traffic through site api endpoints

* feat: add site api endpoint pool editor

* fix: keep proxy video tasks on site endpoint pool

* fix: restore core test compatibility for site endpoint pool

* fix: address endpoint pool review findings
This commit is contained in:
Cita
2026-04-01 20:58:37 +08:00
committed by GitHub
parent ce77bde95a
commit 76c85d0a13
49 changed files with 7226 additions and 986 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/routes/proxy/chat.codex-oauth.test.ts
+25 -1
View File
@@ -55,6 +55,7 @@ vi.mock('../../services/modelPricingService.js', () => ({
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
RETRYABLE_TIMEOUT_PATTERNS: [/(request timed out|connection timed out|read timeout|\btimed out\b)/i],
}));
vi.mock('../../services/proxyUsageFallbackService.js', () => ({
@@ -68,12 +69,33 @@ vi.mock('../../services/oauth/refreshSingleflight.js', () => ({
vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
select: () => ({
from: () => ({
where: () => ({
orderBy: () => ({
all: async () => [],
}),
}),
}),
}),
update: () => ({
set: () => ({
where: () => ({
run: async () => undefined,
}),
}),
}),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
siteApiEndpoints: {
id: {},
siteId: {},
sortOrder: {},
},
},
}));
@@ -144,7 +166,9 @@ describe('chat proxy codex oauth compatibility', () => {
});
afterAll(async () => {
await app.close();
if (app) {
await app.close();
}
});
it('strips codex-unsupported responses fields and aggregates the SSE result for /v1/messages callers', async () => {
src/server/routes/proxy/chat.count-tokens.test.ts
+25 -1
View File
@@ -59,6 +59,7 @@ vi.mock('../../services/modelPricingService.js', () => ({
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
RETRYABLE_TIMEOUT_PATTERNS: [/(request timed out|connection timed out|read timeout|\btimed out\b)/i],
}));
vi.mock('../../services/proxyUsageFallbackService.js', () => ({
@@ -85,12 +86,33 @@ vi.mock('../../services/oauth/quota.js', () => ({
vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
select: () => ({
from: () => ({
where: () => ({
orderBy: () => ({
all: async () => [],
}),
}),
}),
}),
update: () => ({
set: () => ({
where: () => ({
run: async () => undefined,
}),
}),
}),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
siteApiEndpoints: {
id: {},
siteId: {},
sortOrder: {},
},
},
}));
@@ -147,7 +169,9 @@ describe('claude count_tokens proxy route', () => {
});
afterAll(async () => {
await app.close();
if (app) {
await app.close();
}
});
it('forwards /v1/messages/count_tokens to the claude count_tokens upstream path', async () => {
src/server/routes/proxy/chat.siteApiEndpoint.test.ts
+239
View File
@@ -0,0 +1,239 @@
import Fastify, { type FastifyInstance } from 'fastify';
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { mkdtempSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { asc, eq } from 'drizzle-orm';
import { config } from '../../config.js';
import { resetUpstreamEndpointRuntimeState } from '../../services/upstreamEndpointRuntimeMemory.js';
const fetchMock = vi.fn();
const selectChannelMock = vi.fn();
const selectNextChannelMock = vi.fn();
const recordSuccessMock = vi.fn();
const recordFailureMock = vi.fn();
const refreshModelsAndRebuildRoutesMock = vi.fn();
const reportProxyAllFailedMock = vi.fn();
const reportTokenExpiredMock = vi.fn();
const estimateProxyCostMock = vi.fn(async (_arg?: any) => 0);
const buildProxyBillingDetailsMock = vi.fn(async (_arg?: any) => null);
const fetchModelPricingCatalogMock = vi.fn(async (_arg?: any): Promise<any> => null);
const resolveProxyUsageWithSelfLogFallbackMock = vi.fn(async ({ usage }: any) => ({
...usage,
estimatedCostFromQuota: 0,
recoveredFromSelfLog: false,
}));
const insertProxyLogMock = vi.fn();
vi.mock('undici', () => ({
fetch: (...args: unknown[]) => fetchMock(...args),
}));
vi.mock('../../services/tokenRouter.js', () => ({
tokenRouter: {
selectChannel: (...args: unknown[]) => selectChannelMock(...args),
selectNextChannel: (...args: unknown[]) => selectNextChannelMock(...args),
recordSuccess: (...args: unknown[]) => recordSuccessMock(...args),
recordFailure: (...args: unknown[]) => recordFailureMock(...args),
},
}));
vi.mock('../../services/modelService.js', () => ({
refreshModelsAndRebuildRoutes: (...args: unknown[]) => refreshModelsAndRebuildRoutesMock(...args),
}));
vi.mock('../../services/alertService.js', () => ({
reportProxyAllFailed: (...args: unknown[]) => reportProxyAllFailedMock(...args),
reportTokenExpired: (...args: unknown[]) => reportTokenExpiredMock(...args),
}));
vi.mock('../../services/alertRules.js', () => ({
isTokenExpiredError: () => false,
}));
vi.mock('../../services/modelPricingService.js', () => ({
estimateProxyCost: (arg: any) => estimateProxyCostMock(arg),
buildProxyBillingDetails: (arg: any) => buildProxyBillingDetailsMock(arg),
fetchModelPricingCatalog: (arg: any) => fetchModelPricingCatalogMock(arg),
}));
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
RETRYABLE_TIMEOUT_PATTERNS: [/(request timed out|connection timed out|read timeout|\btimed out\b)/i],
}));
vi.mock('../../services/proxyUsageFallbackService.js', () => ({
resolveProxyUsageWithSelfLogFallback: (arg: any) => resolveProxyUsageWithSelfLogFallbackMock(arg),
}));
vi.mock('../../services/proxyLogStore.js', () => ({
insertProxyLog: (...args: unknown[]) => insertProxyLogMock(...args),
}));
type DbModule = typeof import('../../db/index.js');
describe('chat proxy site api endpoint rotation', () => {
let app: FastifyInstance;
let db: DbModule['db'];
let schema: DbModule['schema'];
let dataDir = '';
beforeAll(async () => {
dataDir = mkdtempSync(join(tmpdir(), 'metapi-chat-site-api-endpoint-'));
process.env.DATA_DIR = dataDir;
await import('../../db/migrate.js');
const dbModule = await import('../../db/index.js');
const routesModule = await import('./chat.js');
db = dbModule.db;
schema = dbModule.schema;
app = Fastify();
await app.register(routesModule.chatProxyRoute);
});
beforeEach(async () => {
fetchMock.mockReset();
selectChannelMock.mockReset();
selectNextChannelMock.mockReset();
recordSuccessMock.mockReset();
recordFailureMock.mockReset();
refreshModelsAndRebuildRoutesMock.mockReset();
reportProxyAllFailedMock.mockReset();
reportTokenExpiredMock.mockReset();
estimateProxyCostMock.mockClear();
buildProxyBillingDetailsMock.mockClear();
fetchModelPricingCatalogMock.mockReset();
resolveProxyUsageWithSelfLogFallbackMock.mockClear();
insertProxyLogMock.mockReset();
resetUpstreamEndpointRuntimeState();
await db.delete(schema.proxyLogs).run();
await db.delete(schema.routeChannels).run();
await db.delete(schema.tokenRoutes).run();
await db.delete(schema.tokenModelAvailability).run();
await db.delete(schema.modelAvailability).run();
await db.delete(schema.accountTokens).run();
await db.delete(schema.accounts).run();
await db.delete(schema.siteApiEndpoints).run();
await db.delete(schema.sites).run();
fetchModelPricingCatalogMock.mockResolvedValue(null);
(config as any).codexHeaderDefaults = {
userAgent: '',
betaFeatures: '',
};
(config as any).payloadRules = {
default: [],
defaultRaw: [],
override: [],
overrideRaw: [],
filter: [],
};
(config as any).disableCrossProtocolFallback = false;
config.proxyEmptyContentFailEnabled = false;
config.proxyErrorKeywords = [];
});
afterAll(async () => {
if (app) {
await app.close();
}
delete process.env.DATA_DIR;
});
it('rotates to the next configured ai endpoint for retryable /v1/chat/completions failures', async () => {
const site = await db.insert(schema.sites).values({
name: 'nihao-panel',
url: 'https://nih.cc',
platform: 'openai',
status: 'active',
}).returning().get();
const account = await db.insert(schema.accounts).values({
siteId: site.id,
username: 'nihao-user',
accessToken: '',
apiToken: 'sk-nihao',
status: 'active',
checkinEnabled: false,
extraConfig: JSON.stringify({ credentialMode: 'apikey' }),
}).returning().get();
await db.insert(schema.siteApiEndpoints).values([
{
siteId: site.id,
url: 'https://api-a.nih.cc',
enabled: true,
sortOrder: 0,
},
{
siteId: site.id,
url: 'https://api-b.nih.cc',
enabled: true,
sortOrder: 1,
},
]).run();
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site,
account,
tokenName: 'default',
tokenValue: 'sk-nihao',
actualModel: 'gpt-4o-mini',
});
selectNextChannelMock.mockReturnValue(null);
fetchMock
.mockResolvedValueOnce(new Response('bad gateway', { status: 502 }))
.mockResolvedValueOnce(new Response('bad gateway via responses', { status: 502 }))
.mockResolvedValueOnce(new Response('bad gateway via messages', { status: 502 }))
.mockResolvedValueOnce(new Response(JSON.stringify({
id: 'chatcmpl-ok',
object: 'chat.completion',
created: 1_706_000_000,
model: 'gpt-4o-mini',
choices: [{
index: 0,
message: { role: 'assistant', content: 'ok via api-b' },
finish_reason: 'stop',
}],
usage: { prompt_tokens: 1, completion_tokens: 2, total_tokens: 3 },
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/chat/completions',
payload: {
model: 'gpt-4o-mini',
messages: [{ role: 'user', content: 'hi' }],
},
});
expect(response.statusCode).toBe(200);
expect(response.json()?.choices?.[0]?.message?.content).toBe('ok via api-b');
expect(fetchMock).toHaveBeenCalledTimes(4);
expect(String(fetchMock.mock.calls[0]?.[0] || '')).toBe('https://api-a.nih.cc/v1/chat/completions');
expect(String(fetchMock.mock.calls[1]?.[0] || '')).toBe('https://api-a.nih.cc/v1/responses');
expect(String(fetchMock.mock.calls[2]?.[0] || '')).toBe('https://api-a.nih.cc/v1/messages');
expect(String(fetchMock.mock.calls[3]?.[0] || '')).toBe('https://api-b.nih.cc/v1/chat/completions');
expect(selectNextChannelMock).not.toHaveBeenCalled();
expect(recordFailureMock).not.toHaveBeenCalled();
expect(recordSuccessMock).toHaveBeenCalledTimes(1);
const storedEndpoints = await db.select().from(schema.siteApiEndpoints)
.where(eq(schema.siteApiEndpoints.siteId, site.id))
.orderBy(asc(schema.siteApiEndpoints.sortOrder), asc(schema.siteApiEndpoints.id))
.all();
expect(storedEndpoints[0]).toMatchObject({
url: 'https://api-a.nih.cc',
lastFailureReason: 'HTTP 502: [upstream:/v1/messages] Upstream returned HTTP 502: bad gateway via messages',
});
expect(storedEndpoints[0]?.cooldownUntil).toBeTruthy();
expect(storedEndpoints[1]?.lastSelectedAt).toBeTruthy();
});
});
src/server/routes/proxy/chat.stream.test.ts
+25 -1
View File
@@ -60,6 +60,7 @@ vi.mock('../../services/modelPricingService.js', () => ({
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
RETRYABLE_TIMEOUT_PATTERNS: [/(request timed out|connection timed out|read timeout|\btimed out\b)/i],
}));
vi.mock('../../services/proxyUsageFallbackService.js', () => ({
@@ -69,12 +70,33 @@ vi.mock('../../services/proxyUsageFallbackService.js', () => ({
vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
select: () => ({
from: () => ({
where: () => ({
orderBy: () => ({
all: async () => [],
}),
}),
}),
}),
update: () => ({
set: () => ({
where: () => ({
run: async () => undefined,
}),
}),
}),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
siteApiEndpoints: {
id: {},
siteId: {},
sortOrder: {},
},
},
}));
@@ -133,7 +155,9 @@ describe('chat proxy stream behavior', () => {
});
afterAll(async () => {
await app.close();
if (app) {
await app.close();
}
});
it('converts non-SSE upstream streaming responses into SSE events', async () => {
src/server/routes/proxy/completions.siteApiEndpoint.test.ts
+213
View File
@@ -0,0 +1,213 @@
import Fastify, { type FastifyInstance } from 'fastify';
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { mkdtempSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { asc, eq } from 'drizzle-orm';
const fetchMock = vi.fn();
const selectChannelMock = vi.fn();
const selectNextChannelMock = vi.fn();
const recordSuccessMock = vi.fn();
const recordFailureMock = vi.fn();
const refreshModelsAndRebuildRoutesMock = vi.fn();
const reportProxyAllFailedMock = vi.fn();
const reportTokenExpiredMock = vi.fn();
const insertProxyLogMock = vi.fn();
const resolveProxyUsageWithSelfLogFallbackMock = vi.fn();
const resolveProxyLogBillingMock = vi.fn();
vi.mock('undici', () => ({
fetch: (...args: unknown[]) => fetchMock(...args),
}));
vi.mock('../../services/tokenRouter.js', () => ({
tokenRouter: {
selectChannel: (...args: unknown[]) => selectChannelMock(...args),
selectNextChannel: (...args: unknown[]) => selectNextChannelMock(...args),
recordSuccess: (...args: unknown[]) => recordSuccessMock(...args),
recordFailure: (...args: unknown[]) => recordFailureMock(...args),
},
invalidateTokenRouterCache: vi.fn(),
}));
vi.mock('../../services/routeRefreshWorkflow.js', () => ({
refreshModelsAndRebuildRoutes: (...args: unknown[]) => refreshModelsAndRebuildRoutesMock(...args),
}));
vi.mock('../../services/alertService.js', () => ({
reportProxyAllFailed: (...args: unknown[]) => reportProxyAllFailedMock(...args),
reportTokenExpired: (...args: unknown[]) => reportTokenExpiredMock(...args),
}));
vi.mock('../../services/proxyLogStore.js', () => ({
insertProxyLog: (...args: unknown[]) => insertProxyLogMock(...args),
}));
vi.mock('../../services/proxyUsageFallbackService.js', () => ({
resolveProxyUsageWithSelfLogFallback: (...args: unknown[]) => resolveProxyUsageWithSelfLogFallbackMock(...args),
}));
vi.mock('./proxyBilling.js', () => ({
resolveProxyLogBilling: (...args: unknown[]) => resolveProxyLogBillingMock(...args),
}));
type DbModule = typeof import('../../db/index.js');
describe('/v1/completions site api endpoint rotation', () => {
let app: FastifyInstance;
let db: DbModule['db'];
let schema: DbModule['schema'];
let dataDir = '';
beforeAll(async () => {
dataDir = mkdtempSync(join(tmpdir(), 'metapi-completions-site-api-endpoint-'));
process.env.DATA_DIR = dataDir;
await import('../../db/migrate.js');
const dbModule = await import('../../db/index.js');
const routesModule = await import('./completions.js');
db = dbModule.db;
schema = dbModule.schema;
app = Fastify();
await app.register(routesModule.completionsProxyRoute);
});
beforeEach(async () => {
fetchMock.mockReset();
selectChannelMock.mockReset();
selectNextChannelMock.mockReset();
recordSuccessMock.mockReset();
recordFailureMock.mockReset();
refreshModelsAndRebuildRoutesMock.mockReset();
reportProxyAllFailedMock.mockReset();
reportTokenExpiredMock.mockReset();
insertProxyLogMock.mockReset();
resolveProxyUsageWithSelfLogFallbackMock.mockReset();
resolveProxyLogBillingMock.mockReset();
resolveProxyUsageWithSelfLogFallbackMock.mockResolvedValue({
promptTokens: 1,
completionTokens: 2,
totalTokens: 3,
});
resolveProxyLogBillingMock.mockResolvedValue({
estimatedCost: 0,
billingDetails: null,
});
await db.delete(schema.proxyLogs).run();
await db.delete(schema.routeChannels).run();
await db.delete(schema.tokenRoutes).run();
await db.delete(schema.tokenModelAvailability).run();
await db.delete(schema.modelAvailability).run();
await db.delete(schema.accountTokens).run();
await db.delete(schema.accounts).run();
await db.delete(schema.siteApiEndpoints).run();
await db.delete(schema.sites).run();
});
afterAll(async () => {
await app.close();
delete process.env.DATA_DIR;
});
it('cools down a retryable failed endpoint and retries the next endpoint within the same site', async () => {
const site = await db.insert(schema.sites).values({
name: 'nihao-panel',
url: 'https://nih.cc',
platform: 'new-api',
status: 'active',
}).returning().get();
const account = await db.insert(schema.accounts).values({
siteId: site.id,
username: 'nihao-user',
accessToken: '',
apiToken: 'sk-nihao',
status: 'active',
checkinEnabled: false,
extraConfig: JSON.stringify({ credentialMode: 'apikey' }),
}).returning().get();
await db.insert(schema.siteApiEndpoints).values([
{
siteId: site.id,
url: 'https://api-a.nih.cc',
enabled: true,
sortOrder: 0,
},
{
siteId: site.id,
url: 'https://api-b.nih.cc',
enabled: true,
sortOrder: 1,
},
]).run();
selectChannelMock.mockResolvedValue({
channel: { id: 11, routeId: 22 },
site,
account,
tokenName: 'default',
tokenValue: 'sk-nihao',
actualModel: 'gpt-4o-mini',
});
selectNextChannelMock.mockResolvedValue(null);
fetchMock
.mockResolvedValueOnce(new Response('bad gateway', { status: 502 }))
.mockResolvedValueOnce(new Response(JSON.stringify({
id: 'cmpl-ok',
object: 'text_completion',
choices: [{ text: 'ok' }],
usage: {
prompt_tokens: 1,
completion_tokens: 2,
total_tokens: 3,
},
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/completions',
headers: {
authorization: 'Bearer sk-downstream',
},
payload: {
model: 'gpt-4o-mini',
prompt: 'hello',
},
});
expect(response.statusCode).toBe(200);
expect(response.json()).toMatchObject({
id: 'cmpl-ok',
object: 'text_completion',
});
expect(fetchMock).toHaveBeenCalledTimes(2);
expect(String(fetchMock.mock.calls[0]?.[0] || '')).toBe('https://api-a.nih.cc/v1/completions');
expect(String(fetchMock.mock.calls[1]?.[0] || '')).toBe('https://api-b.nih.cc/v1/completions');
expect(selectNextChannelMock).not.toHaveBeenCalled();
expect(recordFailureMock).not.toHaveBeenCalled();
expect(recordSuccessMock).toHaveBeenCalledTimes(1);
const storedEndpoints = await db.select().from(schema.siteApiEndpoints)
.where(eq(schema.siteApiEndpoints.siteId, site.id))
.orderBy(asc(schema.siteApiEndpoints.sortOrder), asc(schema.siteApiEndpoints.id))
.all();
expect(storedEndpoints[0]).toMatchObject({
url: 'https://api-a.nih.cc',
lastFailureReason: 'HTTP 502: bad gateway',
});
expect(storedEndpoints[0]?.cooldownUntil).toBeTruthy();
expect(storedEndpoints[1]).toMatchObject({
url: 'https://api-b.nih.cc',
});
expect(storedEndpoints[1]?.lastSelectedAt).toBeTruthy();
});
});
src/server/routes/proxy/completions.ts
+47 -74
View File
@@ -10,14 +10,15 @@ import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordD
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';
import { resolveProxyLogBilling } from './proxyBilling.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { detectProxyFailure } from './proxyFailureJudge.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 { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js';
import { runWithSiteApiEndpointPool, SiteApiEndpointRequestError } from '../../services/siteApiEndpointService.js';
import {
buildForcedChannelUnavailableMessage,
canRetryChannelSelection,
@@ -72,68 +73,30 @@ export async function completionsProxyRoute(app: FastifyInstance) {
excludeChannelIds.push(selected.channel.id);
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/completions');
const upstreamModel = selected.actualModel || requestedModel;
const forwardBody = { ...body, model: upstreamModel };
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),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
if (!upstream.ok) {
const errText = await upstream.text().catch(() => 'unknown error');
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: errText,
modelName: upstreamModel,
}));
logProxy(
selected,
requestedModel,
'failed',
upstream.status,
Date.now() - startTime,
errText,
retryCount,
downstreamApiKeyId,
0,
0,
0,
0,
null,
clientContext,
downstreamPath,
);
if (isTokenExpiredError({ status: upstream.status, message: errText })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${upstream.status}`,
});
}
if (shouldRetryProxyRequest(upstream.status, errText) && canRetryChannelSelection(retryCount, forcedChannelId)) {
retryCount++;
continue;
try {
const upstream = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const targetUrl = buildUpstreamUrl(target.baseUrl, '/v1/completions');
const response = 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)));
if (!response.ok) {
const errText = await response.text().catch(() => 'unknown error');
throw new SiteApiEndpointRequestError(errText || 'unknown error', {
status: response.status,
});
}
await reportProxyAllFailed({
model: requestedModel,
reason: `upstream returned HTTP ${upstream.status}`,
});
return reply.code(upstream.status).send({ error: { message: errText, type: 'upstream_error' } });
}
if (isStream) {
return response;
});
if (isStream) {
reply.raw.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
@@ -328,18 +291,20 @@ export async function completionsProxyRoute(app: FastifyInstance) {
);
return reply.send(data);
} catch (err: any) {
const status = err instanceof SiteApiEndpointRequestError ? (err.status || 0) : 0;
const errorText = err?.message || 'network failure';
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
status,
errorText,
modelName: upstreamModel,
}));
logProxy(
selected,
requestedModel,
'failed',
0,
status,
Date.now() - startTime,
err.message,
errorText,
retryCount,
downstreamApiKeyId,
0,
@@ -350,18 +315,26 @@ export async function completionsProxyRoute(app: FastifyInstance) {
clientContext,
downstreamPath,
);
if (canRetryChannelSelection(retryCount, forcedChannelId)) {
if (status > 0 && isTokenExpiredError({ status, message: errorText })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${status}`,
});
}
if ((status > 0 ? shouldRetryProxyRequest(status, errorText) : true) && canRetryChannelSelection(retryCount, forcedChannelId)) {
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' },
});
}
await reportProxyAllFailed({
model: requestedModel,
reason: errorText || 'network failure',
});
return reply.code(status || 502).send({
error: { message: status > 0 ? errorText : `Upstream error: ${errorText}`, type: 'upstream_error' },
});
}
}
});
}
src/server/routes/proxy/downstreamClientContext.routes.test.ts
+25 -1
View File
@@ -58,6 +58,7 @@ vi.mock('../../services/modelPricingService.js', () => ({
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
RETRYABLE_TIMEOUT_PATTERNS: [/(request timed out|connection timed out|read timeout|\btimed out\b)/i],
}));
vi.mock('../../services/proxyUsageFallbackService.js', () => ({
@@ -67,9 +68,30 @@ vi.mock('../../services/proxyUsageFallbackService.js', () => ({
vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
select: () => ({
from: () => ({
where: () => ({
orderBy: () => ({
all: async () => [],
}),
}),
}),
}),
update: () => ({
set: () => ({
where: () => ({
run: async () => undefined,
}),
}),
}),
},
schema: {
proxyLogs: {},
siteApiEndpoints: {
id: {},
siteId: {},
sortOrder: {},
},
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
@@ -116,7 +138,9 @@ describe('downstream client context route logging', () => {
});
afterAll(async () => {
await app.close();
if (app) {
await app.close();
}
});
it('includes Codex client and session metadata in /v1/responses failure logs', async () => {
src/server/routes/proxy/embeddings.ts
+63 -70
View File
@@ -12,11 +12,12 @@ 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 { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.js';
import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js';
import { runWithSiteApiEndpointPool, SiteApiEndpointRequestError } from '../../services/siteApiEndpointService.js';
import {
buildForcedChannelUnavailableMessage,
canRetryChannelSelection,
@@ -66,68 +67,45 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
return reply.code(503).send({ error: { message: noChannelMessage, type: 'server_error' } });
}
excludeChannelIds.push(selected.channel.id);
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/embeddings');
const upstreamModel = selected.actualModel || requestedModel;
const forwardBody = { ...body, model: upstreamModel };
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),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
const text = await upstream.text();
if (!upstream.ok) {
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: upstreamModel,
}));
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({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${upstream.status}`,
});
}
if (shouldRetryProxyRequest(upstream.status, text) && canRetryChannelSelection(retryCount, forcedChannelId)) {
retryCount++;
continue;
try {
const { upstream, text } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const targetUrl = buildUpstreamUrl(target.baseUrl, '/v1/embeddings');
const response = 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 status = response.status;
let responseText = '';
try {
responseText = await response.text();
} catch (error) {
if (!response.ok) {
throw new SiteApiEndpointRequestError('unknown error', {
status,
cause: error,
});
}
throw error;
}
await reportProxyAllFailed({
model: requestedModel,
reason: `upstream returned HTTP ${upstream.status}`,
});
return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } });
}
if (!response.ok) {
throw new SiteApiEndpointRequestError(responseText || 'unknown error', {
status,
rawErrText: responseText || null,
});
}
return {
upstream: response,
text: responseText,
};
});
let data: any = {};
try { data = JSON.parse(text); } catch { data = {}; }
const latency = Date.now() - startTime;
@@ -165,18 +143,20 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
);
return reply.code(upstream.status).send(data);
} catch (err: any) {
const status = err instanceof SiteApiEndpointRequestError ? (err.status || 0) : 0;
const errorText = err?.message || 'network failure';
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
status,
errorText,
modelName: upstreamModel,
}));
logProxy(
selected,
requestedModel,
'failed',
0,
status,
Date.now() - startTime,
err.message,
errorText,
retryCount,
downstreamApiKeyId,
0,
@@ -187,16 +167,29 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
clientContext,
downstreamPath,
);
if (canRetryChannelSelection(retryCount, forcedChannelId)) {
if (status > 0 && isTokenExpiredError({ status, message: errorText })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${status}`,
});
}
if ((status > 0 ? shouldRetryProxyRequest(status, errorText) : true) && canRetryChannelSelection(retryCount, forcedChannelId)) {
retryCount++;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: err.message || 'network failure',
});
return reply.code(502).send({ error: { message: err.message, type: 'upstream_error' } });
}
await reportProxyAllFailed({
model: requestedModel,
reason: errorText || 'network failure',
});
return reply.code(status || 502).send({
error: {
message: status > 0 ? errorText : `Upstream error: ${errorText}`,
type: 'upstream_error',
},
});
}
}
});
}
src/server/routes/proxy/images.edits.test.ts
+25 -1
View File
@@ -52,17 +52,39 @@ vi.mock('../../services/modelPricingService.js', () => ({
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
RETRYABLE_TIMEOUT_PATTERNS: [/(request timed out|connection timed out|read timeout|\btimed out\b)/i],
}));
vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
select: () => ({
from: () => ({
where: () => ({
orderBy: () => ({
all: async () => [],
}),
}),
}),
}),
update: () => ({
set: () => ({
where: () => ({
run: async () => undefined,
}),
}),
}),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
siteApiEndpoints: {
id: {},
siteId: {},
sortOrder: {},
},
},
}));
@@ -115,7 +137,9 @@ describe('/v1/images/edits route', () => {
});
afterAll(async () => {
await app.close();
if (app) {
await app.close();
}
});
it('accepts multipart image edit requests and forwards them to /v1/images/edits', async () => {
src/server/routes/proxy/images.ts
+95 -123
View File
@@ -16,6 +16,7 @@ import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.js';
import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js';
import { runWithSiteApiEndpointPool, SiteApiEndpointRequestError } from '../../services/siteApiEndpointService.js';
import {
buildForcedChannelUnavailableMessage,
canRetryChannelSelection,
@@ -66,60 +67,32 @@ export async function imagesProxyRoute(app: FastifyInstance) {
}
excludeChannelIds.push(selected.channel.id);
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/images/generations');
const upstreamModel = selected.actualModel || requestedModel;
const forwardBody = { ...body, model: upstreamModel };
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),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
const text = await upstream.text();
if (!upstream.ok) {
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: upstreamModel,
}));
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}`,
const { upstream, text } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const targetUrl = buildUpstreamUrl(target.baseUrl, '/v1/images/generations');
const response = 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 responseText = await response.text();
if (!response.ok) {
throw new SiteApiEndpointRequestError(responseText || 'unknown error', {
status: response.status,
});
}
if (shouldRetryProxyRequest(upstream.status, text) && canRetryChannelSelection(retryCount, forcedChannelId)) {
retryCount++;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: `upstream returned HTTP ${upstream.status}`,
});
return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } });
}
return {
upstream: response,
text: responseText,
};
});
const data = parseUpstreamImageResponse(text);
if (!data.ok) {
@@ -175,34 +148,47 @@ export async function imagesProxyRoute(app: FastifyInstance) {
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId, estimatedCost, downstreamPath, clientContext);
return reply.code(upstream.status).send(data.value);
} catch (err: any) {
const status = err instanceof SiteApiEndpointRequestError ? (err.status || 0) : 0;
const errorText = err?.message || 'network failure';
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
status,
errorText,
modelName: upstreamModel,
}));
logProxy(
selected,
requestedModel,
'failed',
0,
status,
Date.now() - startTime,
err.message,
errorText,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
if (canRetryChannelSelection(retryCount, forcedChannelId)) {
if (status > 0 && isTokenExpiredError({ status, message: errorText })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${status}`,
});
}
if ((status > 0 ? shouldRetryProxyRequest(status, errorText) : true) && canRetryChannelSelection(retryCount, forcedChannelId)) {
retryCount++;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: err.message || 'network failure',
reason: errorText || 'network failure',
});
return reply.code(502).send({
error: { message: `Upstream error: ${err.message}`, type: 'upstream_error' },
return reply.code(status || 502).send({
error: {
message: status > 0 ? errorText : `Upstream error: ${errorText}`,
type: 'upstream_error',
},
});
}
}
@@ -254,72 +240,45 @@ export async function imagesProxyRoute(app: FastifyInstance) {
}
excludeChannelIds.push(selected.channel.id);
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/images/edits');
const upstreamModel = selected.actualModel || requestedModel;
const startTime = Date.now();
try {
const requestInit = multipartForm
? withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
Authorization: `Bearer ${selected.tokenValue}`,
},
body: cloneFormDataWithOverrides(multipartForm, {
model: upstreamModel,
}) 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: upstreamModel,
}),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig));
const upstream = await fetch(targetUrl, requestInit);
const text = await upstream.text();
if (!upstream.ok) {
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: upstreamModel,
}));
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}`,
const { upstream, text } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const targetUrl = buildUpstreamUrl(target.baseUrl, '/v1/images/edits');
const requestInit = multipartForm
? withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
Authorization: `Bearer ${selected.tokenValue}`,
},
body: cloneFormDataWithOverrides(multipartForm, {
model: upstreamModel,
}) 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: upstreamModel,
}),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig));
const response = await fetch(targetUrl, requestInit);
const responseText = await response.text();
if (!response.ok) {
throw new SiteApiEndpointRequestError(responseText || 'unknown error', {
status: response.status,
});
}
if (shouldRetryProxyRequest(upstream.status, text) && canRetryChannelSelection(retryCount, forcedChannelId)) {
retryCount++;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: `upstream returned HTTP ${upstream.status}`,
});
return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } });
}
return {
upstream: response,
text: responseText,
};
});
const data = parseUpstreamImageResponse(text);
if (!data.ok) {
@@ -375,34 +334,47 @@ export async function imagesProxyRoute(app: FastifyInstance) {
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId, estimatedCost, downstreamPath, clientContext);
return reply.code(upstream.status).send(data.value);
} catch (err: any) {
const status = err instanceof SiteApiEndpointRequestError ? (err.status || 0) : 0;
const errorText = err?.message || 'network failure';
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
status,
errorText,
modelName: upstreamModel,
}));
logProxy(
selected,
requestedModel,
'failed',
0,
status,
Date.now() - startTime,
err.message,
errorText,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
if (canRetryChannelSelection(retryCount, forcedChannelId)) {
if (status > 0 && isTokenExpiredError({ status, message: errorText })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${status}`,
});
}
if ((status > 0 ? shouldRetryProxyRequest(status, errorText) : true) && canRetryChannelSelection(retryCount, forcedChannelId)) {
retryCount++;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: err.message || 'network failure',
reason: errorText || 'network failure',
});
return reply.code(502).send({
error: { message: `Upstream error: ${err.message}`, type: 'upstream_error' },
return reply.code(status || 502).send({
error: {
message: status > 0 ? errorText : `Upstream error: ${errorText}`,
type: 'upstream_error',
},
});
}
}
src/server/routes/proxy/responses.codex-oauth.test.ts
+25 -1
View File
@@ -69,6 +69,7 @@ vi.mock('../../services/modelPricingService.js', () => ({
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
RETRYABLE_TIMEOUT_PATTERNS: [/(request timed out|connection timed out|read timeout|\btimed out\b)/i],
}));
vi.mock('../../services/proxyUsageFallbackService.js', () => ({
@@ -86,12 +87,33 @@ vi.mock('../../services/oauth/quota.js', () => ({
vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
select: () => ({
from: () => ({
where: () => ({
orderBy: () => ({
all: async () => [],
}),
}),
}),
}),
update: () => ({
set: () => ({
where: () => ({
run: async () => undefined,
}),
}),
}),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
siteApiEndpoints: {
id: {},
siteId: {},
sortOrder: {},
},
},
}));
@@ -205,7 +227,9 @@ describe('responses proxy codex oauth refresh', () => {
config.proxyStickySessionEnabled = originalProxyStickySessionEnabled;
config.proxySessionChannelConcurrencyLimit = originalProxySessionChannelConcurrencyLimit;
config.proxySessionChannelQueueWaitMs = originalProxySessionChannelQueueWaitMs;
await app.close();
if (app) {
await app.close();
}
});
it('refreshes codex oauth token and retries the same responses request on 401', async () => {
src/server/routes/proxy/responses.compact-upstream.test.ts
+25 -1
View File
@@ -57,6 +57,7 @@ vi.mock('../../services/modelPricingService.js', () => ({
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
RETRYABLE_TIMEOUT_PATTERNS: [/(request timed out|connection timed out|read timeout|\btimed out\b)/i],
}));
vi.mock('../../services/proxyUsageFallbackService.js', () => ({
@@ -66,12 +67,33 @@ vi.mock('../../services/proxyUsageFallbackService.js', () => ({
vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
select: () => ({
from: () => ({
where: () => ({
orderBy: () => ({
all: async () => [],
}),
}),
}),
}),
update: () => ({
set: () => ({
where: () => ({
run: async () => undefined,
}),
}),
}),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
siteApiEndpoints: {
id: {},
siteId: {},
sortOrder: {},
},
},
}));
@@ -112,7 +134,9 @@ describe('responses proxy compact upstream routing', () => {
});
afterAll(async () => {
await app.close();
if (app) {
await app.close();
}
});
it('forwards compact requests to the upstream /v1/responses/compact path first', async () => {
src/server/routes/proxy/responses.websocket.test.ts
+31 -3
View File
@@ -78,6 +78,7 @@ vi.mock('../../services/modelPricingService.js', () => ({
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
RETRYABLE_TIMEOUT_PATTERNS: [/(request timed out|connection timed out|read timeout|\btimed out\b)/i],
}));
vi.mock('../../services/proxyUsageFallbackService.js', () => ({
@@ -91,12 +92,33 @@ vi.mock('../../services/oauth/quota.js', () => ({
vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
select: () => ({
from: () => ({
where: () => ({
orderBy: () => ({
all: async () => [],
}),
}),
}),
}),
update: () => ({
set: () => ({
where: () => ({
run: async () => undefined,
}),
}),
}),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
siteApiEndpoints: {
id: {},
siteId: {},
sortOrder: {},
},
},
}));
@@ -366,9 +388,15 @@ describe('responses websocket transport', () => {
afterAll(async () => {
(config as any).codexUpstreamWebsocketEnabled = originalCodexUpstreamWebsocketEnabled;
await new Promise<void>((resolve) => rejectedUpgradeServer.close(() => resolve()));
await new Promise<void>((resolve) => upstreamServer.close(() => resolve()));
await app.close();
if (rejectedUpgradeServer) {
await new Promise<void>((resolve) => rejectedUpgradeServer.close(() => resolve()));
}
if (upstreamServer) {
await new Promise<void>((resolve) => upstreamServer.close(() => resolve()));
}
if (app) {
await app.close();
}
});
it('accepts response.create over GET /v1/responses websocket and forwards streamed responses events', async () => {
src/server/routes/proxy/search.test.ts
+25 -1
View File
@@ -48,17 +48,39 @@ vi.mock('../../services/modelPricingService.js', () => ({
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
RETRYABLE_TIMEOUT_PATTERNS: [/(request timed out|connection timed out|read timeout|\btimed out\b)/i],
}));
vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
select: () => ({
from: () => ({
where: () => ({
orderBy: () => ({
all: async () => [],
}),
}),
}),
}),
update: () => ({
set: () => ({
where: () => ({
run: async () => undefined,
}),
}),
}),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
siteApiEndpoints: {
id: {},
siteId: {},
sortOrder: {},
},
},
}));
@@ -95,7 +117,9 @@ describe('/v1/search route', () => {
});
afterAll(async () => {
await app.close();
if (app) {
await app.close();
}
});
it('defaults model to __search and forwards to the upstream /v1/search endpoint', async () => {
src/server/routes/proxy/search.ts
+41 -53
View File
@@ -14,6 +14,7 @@ import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.js';
import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js';
import { runWithSiteApiEndpointPool, SiteApiEndpointRequestError } from '../../services/siteApiEndpointService.js';
import {
buildForcedChannelUnavailableMessage,
canRetryChannelSelection,
@@ -94,7 +95,6 @@ export async function searchProxyRoute(app: FastifyInstance) {
}
excludeChannelIds.push(selected.channel.id);
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/search');
const upstreamModel = selected.actualModel || requestedModel;
const forwardBody = {
...body,
@@ -104,52 +104,27 @@ export async function searchProxyRoute(app: FastifyInstance) {
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),
}, getProxyUrlFromExtraConfig(selected.account.extraConfig)));
const text = await upstream.text();
if (!upstream.ok) {
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: upstreamModel,
}));
logProxy(
selected,
requestedModel,
'failed',
upstream.status,
Date.now() - startTime,
text,
retryCount,
downstreamApiKeyId,
clientContext,
downstreamPath,
);
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}`,
const { upstream, text } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const targetUrl = buildUpstreamUrl(target.baseUrl, '/v1/search');
const response = 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 responseText = await response.text();
if (!response.ok) {
throw new SiteApiEndpointRequestError(responseText || 'unknown error', {
status: response.status,
});
}
if (shouldRetryProxyRequest(upstream.status, text) && canRetryChannelSelection(retryCount, forcedChannelId)) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: `upstream returned HTTP ${upstream.status}`,
});
return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } });
}
return {
upstream: response,
text: responseText,
};
});
let data: any = {};
try { data = JSON.parse(text); } catch { data = { data: [] }; }
@@ -162,33 +137,46 @@ export async function searchProxyRoute(app: FastifyInstance) {
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId, clientContext, downstreamPath);
return reply.code(upstream.status).send(data);
} catch (error: any) {
const status = error instanceof SiteApiEndpointRequestError ? (error.status || 0) : 0;
const errorText = error?.message || 'network error';
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: error?.message || 'network error',
status,
errorText,
modelName: upstreamModel,
}));
logProxy(
selected,
requestedModel,
'failed',
0,
status,
Date.now() - startTime,
error?.message || 'network error',
errorText,
retryCount,
downstreamApiKeyId,
clientContext,
downstreamPath,
);
if (canRetryChannelSelection(retryCount, forcedChannelId)) {
if (status > 0 && isTokenExpiredError({ status, message: errorText })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${status}`,
});
}
if ((status > 0 ? shouldRetryProxyRequest(status, errorText) : true) && canRetryChannelSelection(retryCount, forcedChannelId)) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: error?.message || 'network failure',
reason: errorText || 'network failure',
});
return reply.code(502).send({
error: { message: error?.message || 'network failure', type: 'upstream_error' },
return reply.code(status || 502).send({
error: {
message: status > 0 ? errorText : `Upstream error: ${errorText}`,
type: 'upstream_error',
},
});
}
}
src/server/routes/proxy/videos.test.ts
+238 -2
View File
@@ -10,10 +10,13 @@ const refreshModelsAndRebuildRoutesMock = vi.fn();
const reportProxyAllFailedMock = vi.fn();
const reportTokenExpiredMock = vi.fn();
const estimateProxyCostMock = vi.fn(async () => 0);
const shouldRetryProxyRequestMock = vi.fn();
const saveProxyVideoTaskMock = vi.fn();
const getProxyVideoTaskByPublicIdMock = vi.fn();
const deleteProxyVideoTaskByPublicIdMock = vi.fn();
const refreshProxyVideoTaskSnapshotMock = vi.fn();
const resolveProxyVideoTaskSiteMock = vi.fn();
let siteApiEndpointRows: Array<Record<string, unknown>> = [];
vi.mock('undici', async () => {
const actual = await vi.importActual<typeof import('undici')>('undici');
@@ -50,7 +53,8 @@ vi.mock('../../services/modelPricingService.js', () => ({
}));
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
shouldRetryProxyRequest: (...args: unknown[]) => shouldRetryProxyRequestMock(...args),
RETRYABLE_TIMEOUT_PATTERNS: [/(request timed out|connection timed out|read timeout|\btimed out\b)/i],
}));
vi.mock('../../services/proxyVideoTaskStore.js', () => ({
@@ -58,6 +62,35 @@ vi.mock('../../services/proxyVideoTaskStore.js', () => ({
getProxyVideoTaskByPublicId: (...args: unknown[]) => getProxyVideoTaskByPublicIdMock(...args),
deleteProxyVideoTaskByPublicId: (...args: unknown[]) => deleteProxyVideoTaskByPublicIdMock(...args),
refreshProxyVideoTaskSnapshot: (...args: unknown[]) => refreshProxyVideoTaskSnapshotMock(...args),
resolveProxyVideoTaskSite: (...args: unknown[]) => resolveProxyVideoTaskSiteMock(...args),
}));
vi.mock('../../db/index.js', () => ({
db: {
select: () => ({
from: () => ({
where: () => ({
orderBy: () => ({
all: async () => siteApiEndpointRows,
}),
}),
}),
}),
update: () => ({
set: () => ({
where: () => ({
run: async () => undefined,
}),
}),
}),
},
schema: {
siteApiEndpoints: {
id: {},
siteId: {},
sortOrder: {},
},
},
}));
describe('/v1/videos routes', () => {
@@ -93,10 +126,14 @@ describe('/v1/videos routes', () => {
reportProxyAllFailedMock.mockReset();
reportTokenExpiredMock.mockReset();
estimateProxyCostMock.mockClear();
shouldRetryProxyRequestMock.mockReset();
saveProxyVideoTaskMock.mockReset();
getProxyVideoTaskByPublicIdMock.mockReset();
deleteProxyVideoTaskByPublicIdMock.mockReset();
refreshProxyVideoTaskSnapshotMock.mockReset();
resolveProxyVideoTaskSiteMock.mockReset();
siteApiEndpointRows = [];
shouldRetryProxyRequestMock.mockReturnValue(false);
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
@@ -110,7 +147,9 @@ describe('/v1/videos routes', () => {
});
afterAll(async () => {
await app.close();
if (app) {
await app.close();
}
});
it('creates an upstream video task and stores a local public id mapping', async () => {
@@ -189,7 +228,55 @@ describe('/v1/videos routes', () => {
});
});
it('stores the selected api endpoint base url for split-host video tasks', async () => {
siteApiEndpointRows = [
{
id: 91,
siteId: 44,
url: 'https://api-videos.example.com',
enabled: true,
sortOrder: 0,
},
];
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { id: 44, name: 'demo-site', url: 'https://panel.example.com', platform: 'openai' },
account: { id: 33, username: 'demo-user' },
tokenName: 'default',
tokenValue: 'sk-demo',
actualModel: 'sora-2',
});
saveProxyVideoTaskMock.mockResolvedValue({
publicId: 'vid_local_split',
upstreamVideoId: 'vid_upstream_split',
});
fetchMock.mockResolvedValue(new Response(JSON.stringify({
id: 'vid_upstream_split',
object: 'video',
status: 'queued',
model: 'sora-2',
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/videos',
payload: {
model: 'sora-2',
prompt: 'split-host create',
},
});
expect(response.statusCode).toBe(200);
expect(saveProxyVideoTaskMock).toHaveBeenCalledWith(expect.objectContaining({
siteUrl: 'https://api-videos.example.com',
}));
});
it('resolves local video ids back to the upstream task on GET', async () => {
resolveProxyVideoTaskSiteMock.mockResolvedValue(null);
getProxyVideoTaskByPublicIdMock.mockResolvedValue({
publicId: 'vid_local_123',
upstreamVideoId: 'vid_upstream_123',
@@ -228,6 +315,116 @@ describe('/v1/videos routes', () => {
});
});
it('re-resolves account-backed video tasks through the site api endpoint pool on GET', async () => {
siteApiEndpointRows = [
{
id: 91,
siteId: 44,
url: 'https://api-videos.example.com',
enabled: true,
sortOrder: 0,
},
];
resolveProxyVideoTaskSiteMock.mockResolvedValue({
id: 44,
name: 'demo-site',
url: 'https://panel.example.com',
platform: 'openai',
});
getProxyVideoTaskByPublicIdMock.mockResolvedValue({
publicId: 'vid_local_456',
upstreamVideoId: 'vid_upstream_456',
siteUrl: 'https://panel.example.com',
tokenValue: 'sk-demo',
accountId: 33,
});
fetchMock.mockResolvedValue(new Response(JSON.stringify({
id: 'vid_upstream_456',
object: 'video',
status: 'queued',
model: 'sora-2',
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'GET',
url: '/v1/videos/vid_local_456',
});
expect(response.statusCode).toBe(200);
const [targetUrl] = fetchMock.mock.calls[0] as [string, RequestInit];
expect(targetUrl).toBe('https://api-videos.example.com/v1/videos/vid_upstream_456');
});
it('uses the persisted api base url when the backing site can no longer be resolved', async () => {
resolveProxyVideoTaskSiteMock.mockResolvedValue(null);
getProxyVideoTaskByPublicIdMock.mockResolvedValue({
publicId: 'vid_local_fallback',
upstreamVideoId: 'vid_upstream_fallback',
siteUrl: 'https://api-fallback.example.com',
tokenValue: 'sk-demo',
accountId: 33,
});
fetchMock.mockResolvedValue(new Response(JSON.stringify({
id: 'vid_upstream_fallback',
object: 'video',
status: 'running',
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'GET',
url: '/v1/videos/vid_local_fallback',
});
expect(response.statusCode).toBe(200);
const [targetUrl] = fetchMock.mock.calls[0] as [string, RequestInit];
expect(targetUrl).toBe('https://api-fallback.example.com/v1/videos/vid_upstream_fallback');
});
it('preserves retryable upstream failures after mapped GET retries exhaust', async () => {
shouldRetryProxyRequestMock.mockReturnValue(true);
siteApiEndpointRows = [
{
id: 91,
siteId: 44,
url: 'https://api-videos.example.com',
enabled: true,
sortOrder: 0,
},
];
resolveProxyVideoTaskSiteMock.mockResolvedValue({
id: 44,
name: 'demo-site',
url: 'https://panel.example.com',
platform: 'openai',
});
getProxyVideoTaskByPublicIdMock.mockResolvedValue({
publicId: 'vid_local_retry_get',
upstreamVideoId: 'vid_upstream_retry_get',
siteUrl: 'https://api-videos.example.com',
tokenValue: 'sk-demo',
accountId: 33,
});
fetchMock.mockResolvedValue(new Response('temporary unavailable', {
status: 502,
headers: { 'content-type': 'text/plain' },
}));
const response = await app.inject({
method: 'GET',
url: '/v1/videos/vid_local_retry_get',
});
expect(response.statusCode).toBe(502);
expect(response.headers['content-type']).toContain('text/plain');
expect(response.body).toBe('temporary unavailable');
});
it('deletes the upstream task and local mapping on DELETE', async () => {
getProxyVideoTaskByPublicIdMock.mockResolvedValue({
publicId: 'vid_local_123',
@@ -245,4 +442,43 @@ describe('/v1/videos routes', () => {
expect(response.statusCode).toBe(204);
expect(deleteProxyVideoTaskByPublicIdMock).toHaveBeenCalledWith('vid_local_123');
});
it('preserves retryable upstream failures after mapped DELETE retries exhaust', async () => {
shouldRetryProxyRequestMock.mockReturnValue(true);
siteApiEndpointRows = [
{
id: 91,
siteId: 44,
url: 'https://api-videos.example.com',
enabled: true,
sortOrder: 0,
},
];
resolveProxyVideoTaskSiteMock.mockResolvedValue({
id: 44,
name: 'demo-site',
url: 'https://panel.example.com',
platform: 'openai',
});
getProxyVideoTaskByPublicIdMock.mockResolvedValue({
publicId: 'vid_local_retry_delete',
upstreamVideoId: 'vid_upstream_retry_delete',
siteUrl: 'https://api-videos.example.com',
tokenValue: 'sk-demo',
accountId: 33,
});
fetchMock.mockResolvedValue(new Response('temporary unavailable', {
status: 502,
headers: { 'content-type': 'text/plain' },
}));
const response = await app.inject({
method: 'DELETE',
url: '/v1/videos/vid_local_retry_delete',
});
expect(response.statusCode).toBe(502);
expect(response.headers['content-type']).toContain('text/plain');
expect(response.body).toBe('temporary unavailable');
});
});
src/server/routes/proxy/videos.ts
+133 -69
View File
@@ -14,6 +14,7 @@ import {
deleteProxyVideoTaskByPublicId,
getProxyVideoTaskByPublicId,
refreshProxyVideoTaskSnapshot,
resolveProxyVideoTaskSite,
saveProxyVideoTask,
} from '../../services/proxyVideoTaskStore.js';
import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js';
@@ -23,6 +24,7 @@ import {
getTesterForcedChannelId,
selectProxyChannelForAttempt,
} from '../../proxy-core/channelSelection.js';
import { runWithSiteApiEndpointPool, SiteApiEndpointRequestError } from '../../services/siteApiEndpointService.js';
function rewriteVideoResponsePublicId(payload: unknown, publicId: string): unknown {
if (!payload || typeof payload !== 'object') return payload;
@@ -80,60 +82,48 @@ export async function videosProxyRoute(app: FastifyInstance) {
}
excludeChannelIds.push(selected.channel.id);
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/videos');
const upstreamModel = selected.actualModel || requestedModel;
const startTime = Date.now();
try {
const accountProxy = getProxyUrlFromExtraConfig(selected.account.extraConfig);
const requestInit = multipartForm
? withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
Authorization: `Bearer ${selected.tokenValue}`,
},
body: cloneFormDataWithOverrides(multipartForm, {
model: upstreamModel,
}) as any,
}, accountProxy)
: withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify({
...(jsonBody || {}),
model: upstreamModel,
}),
}, accountProxy);
const upstream = await fetch(targetUrl, requestInit);
const text = await upstream.text();
if (!upstream.ok) {
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: upstreamModel,
}));
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}`,
const { upstream, text, baseUrl } = await runWithSiteApiEndpointPool(selected.site, async (target) => {
const targetUrl = buildUpstreamUrl(target.baseUrl, '/v1/videos');
const accountProxy = getProxyUrlFromExtraConfig(selected.account.extraConfig);
const requestInit = multipartForm
? withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
Authorization: `Bearer ${selected.tokenValue}`,
},
body: cloneFormDataWithOverrides(multipartForm, {
model: upstreamModel,
}) as any,
}, accountProxy)
: withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify({
...(jsonBody || {}),
model: upstreamModel,
}),
}, accountProxy);
const response = await fetch(targetUrl, requestInit);
const responseText = await response.text();
if (!response.ok) {
throw new SiteApiEndpointRequestError(responseText || 'unknown error', {
status: response.status,
rawErrText: responseText || null,
});
}
if (shouldRetryProxyRequest(upstream.status, text) && canRetryChannelSelection(retryCount, forcedChannelId)) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: `upstream returned HTTP ${upstream.status}`,
});
return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } });
}
return {
baseUrl: target.baseUrl,
upstream: response,
text: responseText,
};
});
let data: any = {};
try { data = JSON.parse(text); } catch { data = {}; }
@@ -146,7 +136,7 @@ export async function videosProxyRoute(app: FastifyInstance) {
const mapping = await saveProxyVideoTask({
upstreamVideoId,
siteUrl: selected.site.url,
siteUrl: baseUrl,
tokenValue: selected.tokenValue,
requestedModel,
actualModel: upstreamModel,
@@ -174,21 +164,34 @@ export async function videosProxyRoute(app: FastifyInstance) {
recordDownstreamCostUsage(request, estimatedCost);
return reply.code(upstream.status).send(rewriteVideoResponsePublicId(data, mapping.publicId));
} catch (error: any) {
const status = error instanceof SiteApiEndpointRequestError ? (error.status || 0) : 0;
const errorText = error?.message || 'network failure';
await recordTokenRouterEventBestEffort('record channel failure', () => tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: error?.message || 'network failure',
status,
errorText,
modelName: upstreamModel,
}));
if (canRetryChannelSelection(retryCount, forcedChannelId)) {
if (status > 0 && isTokenExpiredError({ status, message: errorText })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${status}`,
});
}
if ((status > 0 ? shouldRetryProxyRequest(status, errorText) : true) && canRetryChannelSelection(retryCount, forcedChannelId)) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: error?.message || 'network failure',
reason: errorText || 'network failure',
});
return reply.code(502).send({
error: { message: error?.message || 'network failure', type: 'upstream_error' },
return reply.code(status || 502).send({
error: {
message: status > 0 ? errorText : `Upstream error: ${errorText}`,
type: 'upstream_error',
},
});
}
}
@@ -202,13 +205,15 @@ export async function videosProxyRoute(app: FastifyInstance) {
});
}
const targetUrl = buildUpstreamUrl(mapping.siteUrl, `/v1/videos/${encodeURIComponent(mapping.upstreamVideoId)}`);
const upstream = await fetch(targetUrl, await withSiteProxyRequestInit(targetUrl, {
method: 'GET',
headers: {
Authorization: `Bearer ${mapping.tokenValue}`,
},
}));
let upstream: Awaited<ReturnType<typeof fetch>>;
try {
({ upstream } = await requestMappedVideoTaskUpstream(mapping, 'GET'));
} catch (error) {
if (isSiteApiEndpointFailure(error)) {
return sendVideoTaskEndpointFailure(reply, error);
}
throw error;
}
const text = await upstream.text();
try {
const data = JSON.parse(text);
@@ -233,13 +238,15 @@ export async function videosProxyRoute(app: FastifyInstance) {
});
}
const targetUrl = buildUpstreamUrl(mapping.siteUrl, `/v1/videos/${encodeURIComponent(mapping.upstreamVideoId)}`);
const upstream = await fetch(targetUrl, await withSiteProxyRequestInit(targetUrl, {
method: 'DELETE',
headers: {
Authorization: `Bearer ${mapping.tokenValue}`,
},
}));
let upstream: Awaited<ReturnType<typeof fetch>>;
try {
({ upstream } = await requestMappedVideoTaskUpstream(mapping, 'DELETE'));
} catch (error) {
if (isSiteApiEndpointFailure(error)) {
return sendVideoTaskEndpointFailure(reply, error);
}
throw error;
}
if (upstream.ok) {
await deleteProxyVideoTaskByPublicId(mapping.publicId);
return reply.code(upstream.status).send();
@@ -252,6 +259,63 @@ export async function videosProxyRoute(app: FastifyInstance) {
});
}
async function requestMappedVideoTaskUpstream(
mapping: NonNullable<Awaited<ReturnType<typeof getProxyVideoTaskByPublicId>>>,
method: 'GET' | 'DELETE',
): Promise<{ upstream: Awaited<ReturnType<typeof fetch>> }> {
const buildRequest = async (baseUrl: string) => {
const targetUrl = buildUpstreamUrl(baseUrl, `/v1/videos/${encodeURIComponent(mapping.upstreamVideoId)}`);
const upstream = await fetch(targetUrl, await withSiteProxyRequestInit(targetUrl, {
method,
headers: {
Authorization: `Bearer ${mapping.tokenValue}`,
},
}));
if (!upstream.ok) {
const errorText = await upstream.clone().text().catch(() => '');
if (shouldRetryProxyRequest(upstream.status, errorText || `HTTP ${upstream.status}`)) {
throw new SiteApiEndpointRequestError(errorText || `HTTP ${upstream.status}`, {
status: upstream.status,
rawErrText: errorText || null,
});
}
}
return { upstream };
};
const site = await resolveProxyVideoTaskSite(mapping.accountId);
if (site) {
return runWithSiteApiEndpointPool(site, (target) => buildRequest(target.baseUrl));
}
return buildRequest(mapping.siteUrl);
}
function isSiteApiEndpointFailure(error: unknown): error is SiteApiEndpointRequestError {
return error instanceof SiteApiEndpointRequestError
|| (typeof error === 'object' && error !== null && (error as { name?: unknown }).name === 'SiteApiEndpointRequestError');
}
function sendVideoTaskEndpointFailure(
reply: FastifyReply,
error: { status?: number | null; rawErrText?: string | null; message?: string | null },
) {
const status = typeof error.status === 'number' && error.status > 0 ? error.status : 502;
const rawText = (typeof error.rawErrText === 'string' && error.rawErrText.trim())
? error.rawErrText
: (typeof error.message === 'string' ? error.message.trim() : '');
if (!rawText) {
return reply.code(status).send({
error: { message: 'Upstream request failed', type: 'upstream_error' },
});
}
try {
return reply.code(status).send(JSON.parse(rawText));
} catch {
return reply.code(status).type('text/plain').send(rawText);
}
}
async function recordTokenRouterEventBestEffort(
label: string,
operation: () => Promise<unknown>,