[codex] add oauth route pools and proxy save flow (#440)

* feat: add oauth route pools and proxy save flow

* fix: align oauth route pool schema and rate limits

* fix: harden oauth route pool review fixes

* fix: address oauth ci regressions

* fix: close oauth codeql alerts

* fix: harden oauth route unit recovery

* fix: split oauth route limiters for codeql

* fix: address remaining oauth review comments

src/server/routes/api/oauth.ts

@@ -2,6 +2,7 @@ import type { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
 import { RateLimiterMemory, RateLimiterRes } from 'rate-limiter-flexible';
 import { createRateLimitGuard } from '../../middleware/requestRateLimit.js';
 import {
+  getOauthProviderDefaults,
   deleteOauthConnection,
   importOauthConnectionsFromNativeJson,
   getOauthSessionStatus,
@@ -14,13 +15,22 @@ import {
   startOauthProviderFlow,
   startOauthRebindFlow,
   submitOauthManualCallback,
+  updateOauthConnectionProxySettings,
 } from '../../services/oauth/service.js';
+import {
+  createOauthRouteUnit,
+  deleteOauthRouteUnit,
+  updateOauthRouteUnit,
+} from '../../services/oauth/routeUnitService.js';
 import { parseSiteProxyUrlInput } from '../../services/siteProxy.js';
 import {
   parseOauthConnectionRebindPayload,
+  parseOauthConnectionProxyUpdatePayload,
   parseOauthImportPayload,
   parseOauthManualCallbackPayload,
   parseOauthQuotaBatchRefreshPayload,
+  parseOauthRouteUnitCreatePayload,
+  parseOauthRouteUnitUpdatePayload,
   parseOauthStartPayload,
 } from '../../contracts/supportRoutePayloads.js';
 
@@ -60,21 +70,47 @@ const limitOauthConnectionMutate = createRateLimitGuard({
   windowMs: 60_000,
 });
 
-const oauthSensitiveRouteLimiter = new RateLimiterMemory({
-  keyPrefix: 'oauth-connection-sensitive',
-  points: 20,
-  duration: 60,
-});
+function createOauthSensitiveRouteLimiter(keyPrefix: string, points = 20) {
+  return new RateLimiterMemory({
+    keyPrefix,
+    points,
+    duration: 60,
+  });
+}
+
+let oauthQuotaBatchRefreshLimiter = createOauthSensitiveRouteLimiter('oauth-connection-sensitive-quota-batch');
+let oauthProxyUpdateLimiter = createOauthSensitiveRouteLimiter('oauth-connection-sensitive-proxy');
+let oauthImportLimiter = createOauthSensitiveRouteLimiter('oauth-connection-sensitive-import');
+let oauthRouteUnitCreateLimiter = createOauthSensitiveRouteLimiter('oauth-connection-sensitive-route-unit-create');
+let oauthRouteUnitUpdateLimiter = createOauthSensitiveRouteLimiter('oauth-connection-sensitive-route-unit-update');
+let oauthRouteUnitDeleteLimiter = createOauthSensitiveRouteLimiter('oauth-connection-sensitive-route-unit-delete');
 const MAX_OAUTH_QUOTA_BATCH_SIZE = 100;
 
+export function resetOauthSensitiveRouteLimiterForTests(options: {
+  points?: number;
+} = {}): void {
+  const points = options.points ?? 20;
+  oauthQuotaBatchRefreshLimiter = createOauthSensitiveRouteLimiter('oauth-connection-sensitive-quota-batch', points);
+  oauthProxyUpdateLimiter = createOauthSensitiveRouteLimiter('oauth-connection-sensitive-proxy', points);
+  oauthImportLimiter = createOauthSensitiveRouteLimiter('oauth-connection-sensitive-import', points);
+  oauthRouteUnitCreateLimiter = createOauthSensitiveRouteLimiter('oauth-connection-sensitive-route-unit-create', points);
+  oauthRouteUnitUpdateLimiter = createOauthSensitiveRouteLimiter('oauth-connection-sensitive-route-unit-update', points);
+  oauthRouteUnitDeleteLimiter = createOauthSensitiveRouteLimiter('oauth-connection-sensitive-route-unit-delete', points);
+}
+
+function sendOauthSensitiveRateLimit(reply: FastifyReply, error: unknown): void {
+  const retryState = error instanceof RateLimiterRes ? error : null;
+  const retryAfterSec = Math.max(1, Math.ceil((retryState?.msBeforeNext ?? 60_000) / 1000));
+  reply.code(429).header('retry-after', String(retryAfterSec))
+    .send({ message: '请求过于频繁，请稍后再试' });
+}
+
 async function limitOauthSensitiveRoute(request: FastifyRequest, reply: FastifyReply) {
   try {
-    await oauthSensitiveRouteLimiter.consume(request.ip);
+    await oauthQuotaBatchRefreshLimiter.consume(request.ip);
   } catch (error) {
-    const retryState = error instanceof RateLimiterRes ? error : null;
-    const retryAfterSec = Math.max(1, Math.ceil((retryState?.msBeforeNext ?? 60_000) / 1000));
-    reply.code(429).header('retry-after', String(retryAfterSec))
-      .send({ message: '请求过于频繁，请稍后再试' });
+    sendOauthSensitiveRateLimit(reply, error);
+    return reply;
   }
 }
 
@@ -137,6 +173,7 @@ function resolveRequestOrigin(request: FastifyRequest): string | undefined {
 
 export async function oauthRoutes(app: FastifyInstance) {
   app.get('/api/oauth/providers', { preHandler: [limitOauthProviderRead] }, async () => ({
+    defaults: getOauthProviderDefaults(),
     providers: listOauthProviders(),
   }));
 
@@ -282,6 +319,46 @@ export async function oauthRoutes(app: FastifyInstance) {
     },
   );
 
+  app.patch<{ Params: { accountId: string }; Body: unknown }>(
+    '/api/oauth/connections/:accountId/proxy',
+    { preHandler: [limitOauthConnectionMutate] },
+    async (request, reply) => {
+      try {
+        await oauthProxyUpdateLimiter.consume(request.ip);
+      } catch (error) {
+        sendOauthSensitiveRateLimit(reply, error);
+        return;
+      }
+      const parsedBody = parseOauthConnectionProxyUpdatePayload(request.body);
+      if (!parsedBody.success) {
+        return reply.code(400).send({ message: parsedBody.error });
+      }
+
+      const accountId = parsePositiveInteger(request.params.accountId);
+      if (accountId === null) {
+        return reply.code(400).send({ message: 'invalid account id' });
+      }
+      const normalizedProxyUrl = parseSiteProxyUrlInput(parsedBody.data.proxyUrl);
+      if (normalizedProxyUrl.present && !normalizedProxyUrl.valid) {
+        return reply.code(400).send({ message: 'invalid proxy url' });
+      }
+      try {
+        return await updateOauthConnectionProxySettings({
+          accountId,
+          proxyUrl: normalizedProxyUrl.present ? normalizedProxyUrl.proxyUrl : undefined,
+          useSystemProxy: parsedBody.data.useSystemProxy,
+        });
+      } catch (error: any) {
+        const message = error?.message || 'oauth account not found';
+        if (message === 'oauth account not found' || message === 'account is not managed by oauth') {
+          return reply.code(404).send({ message });
+        }
+        request.log.error({ err: error }, 'oauth proxy update failed');
+        return reply.code(500).send({ message });
+      }
+    },
+  );
+
   app.delete<{ Params: { accountId: string } }>(
     '/api/oauth/connections/:accountId',
     { preHandler: [limitOauthConnectionMutate] },
@@ -337,18 +414,34 @@ export async function oauthRoutes(app: FastifyInstance) {
 
   app.post<{ Body: unknown }>(
     '/api/oauth/import',
-    { preHandler: [limitOauthConnectionMutate, limitOauthSensitiveRoute] },
+    { preHandler: [limitOauthConnectionMutate] },
     async (request, reply) => {
+      try {
+        await oauthImportLimiter.consume(request.ip);
+      } catch (error) {
+        sendOauthSensitiveRateLimit(reply, error);
+        return;
+      }
       const parsedBody = parseOauthImportPayload(request.body);
       if (!parsedBody.success) {
         return reply.code(400).send({ message: parsedBody.error });
       }
+      const hasBatchItems = Array.isArray(parsedBody.data.items) && parsedBody.data.items.length > 0;
       const data = parsedBody.data.data;
-      if (!data || typeof data !== 'object' || Array.isArray(data)) {
+      if (!hasBatchItems && (!data || typeof data !== 'object' || Array.isArray(data))) {
         return reply.code(400).send({ message: 'data must be a native oauth json object' });
       }
+      const normalizedProxyUrl = parseSiteProxyUrlInput(parsedBody.data.proxyUrl);
+      if (normalizedProxyUrl.present && !normalizedProxyUrl.valid) {
+        return reply.code(400).send({ message: 'invalid proxy url' });
+      }
       try {
-        return await importOauthConnectionsFromNativeJson(data);
+        return await importOauthConnectionsFromNativeJson({
+          data,
+          items: hasBatchItems ? parsedBody.data.items : undefined,
+          proxyUrl: normalizedProxyUrl.present ? normalizedProxyUrl.proxyUrl : undefined,
+          useSystemProxy: parsedBody.data.useSystemProxy,
+        });
       } catch (error: any) {
         const message = error?.message || 'oauth import failed';
         if (error instanceof OauthImportValidationError) {
@@ -360,6 +453,113 @@ export async function oauthRoutes(app: FastifyInstance) {
     },
   );
 
+  app.post<{ Body: unknown }>(
+    '/api/oauth/route-units',
+    { preHandler: [limitOauthConnectionMutate] },
+    async (request, reply) => {
+      try {
+        await oauthRouteUnitCreateLimiter.consume(request.ip);
+      } catch (error) {
+        sendOauthSensitiveRateLimit(reply, error);
+        return;
+      }
+      const parsedBody = parseOauthRouteUnitCreatePayload(request.body);
+      if (!parsedBody.success) {
+        return reply.code(400).send({ message: parsedBody.error });
+      }
+      try {
+        return await createOauthRouteUnit({
+          accountIds: Array.isArray(parsedBody.data.accountIds) ? parsedBody.data.accountIds : [],
+          name: parsedBody.data.name || '',
+          strategy: parsedBody.data.strategy || 'round_robin',
+        });
+      } catch (error: any) {
+        const message = error?.message || 'oauth route unit creation failed';
+        if (message === 'oauth route unit accounts not found' || message === 'oauth route unit not found') {
+          return reply.code(404).send({ message });
+        }
+        if (
+          message === 'oauth route unit requires at least 2 accounts'
+          || message === 'oauth route unit name is required'
+          || message === 'invalid oauth route unit strategy'
+          || message === 'oauth route unit accounts already grouped'
+          || message === 'oauth route unit only supports oauth accounts'
+          || message === 'oauth route unit accounts must belong to the same site'
+          || message === 'oauth route unit accounts must share the same provider'
+        ) {
+          return reply.code(400).send({ message });
+        }
+        request.log.error({ err: error }, 'oauth route unit creation failed');
+        return reply.code(500).send({ message });
+      }
+    },
+  );
+
+  app.patch<{ Params: { routeUnitId: string }; Body: unknown }>(
+    '/api/oauth/route-units/:routeUnitId',
+    { preHandler: [limitOauthConnectionMutate] },
+    async (request, reply) => {
+      try {
+        await oauthRouteUnitUpdateLimiter.consume(request.ip);
+      } catch (error) {
+        sendOauthSensitiveRateLimit(reply, error);
+        return;
+      }
+      const parsedBody = parseOauthRouteUnitUpdatePayload(request.body);
+      if (!parsedBody.success) {
+        return reply.code(400).send({ message: parsedBody.error });
+      }
+      const routeUnitId = parsePositiveInteger(request.params.routeUnitId);
+      if (routeUnitId === null) {
+        return reply.code(400).send({ message: 'invalid route unit id' });
+      }
+      try {
+        return await updateOauthRouteUnit({
+          routeUnitId,
+          name: parsedBody.data.name,
+          strategy: parsedBody.data.strategy,
+        });
+      } catch (error: any) {
+        const message = error?.message || 'oauth route unit update failed';
+        if (message === 'oauth route unit not found') {
+          return reply.code(404).send({ message });
+        }
+        if (message === 'oauth route unit name is required' || message === 'invalid oauth route unit strategy') {
+          return reply.code(400).send({ message });
+        }
+        request.log.error({ err: error }, 'oauth route unit update failed');
+        return reply.code(500).send({ message });
+      }
+    },
+  );
+
+  app.delete<{ Params: { routeUnitId: string } }>(
+    '/api/oauth/route-units/:routeUnitId',
+    { preHandler: [limitOauthConnectionMutate] },
+    async (request, reply) => {
+      try {
+        await oauthRouteUnitDeleteLimiter.consume(request.ip);
+      } catch (error) {
+        sendOauthSensitiveRateLimit(reply, error);
+        return;
+      }
+      const routeUnitId = parsePositiveInteger(request.params.routeUnitId);
+      if (routeUnitId === null) {
+        return reply.code(400).send({ message: 'invalid route unit id' });
+      }
+      try {
+        return await deleteOauthRouteUnit(routeUnitId);
+      } catch (error: any) {
+        const message = error?.message || 'oauth route unit not found';
+        if (message === 'oauth route unit not found') {
+          return reply.code(404).send({ message });
+        }
+        request.log.error({ err: error }, 'oauth route unit deletion failed');
+        return reply.code(500).send({ message });
+      }
+    },
+  );
+
   app.get<{ Params: { provider: string }; Querystring: { state?: string; code?: string; error?: string } }>(
     '/api/oauth/callback/:provider',
     { preHandler: [limitOauthCallback] },

src/server/routes/api/tokens.route-update-rebuild.test.ts

@@ -11,6 +11,7 @@ describe('PUT /api/routes/:id route rebuild', () => {
   let app: FastifyInstance;
   let db: DbModule['db'];
   let schema: DbModule['schema'];
+  let resetTokenRouteReadLimitersForTests: (options?: { summaryPoints?: number; listPoints?: number }) => void;
   let dataDir = '';
   let seedId = 0;
 
@@ -61,12 +62,14 @@ describe('PUT /api/routes/:id route rebuild', () => {
     const routesModule = await import('./tokens.js');
     db = dbModule.db;
     schema = dbModule.schema;
+    resetTokenRouteReadLimitersForTests = routesModule.resetTokenRouteReadLimitersForTests;
 
     app = Fastify();
     await app.register(routesModule.tokensRoutes);
   });
 
   beforeEach(async () => {
+    resetTokenRouteReadLimitersForTests();
     await db.delete(schema.routeChannels).run();
     await db.delete(schema.tokenRoutes).run();
     await db.delete(schema.tokenModelAvailability).run();
@@ -150,6 +153,35 @@ describe('PUT /api/routes/:id route rebuild', () => {
     expect(rebuiltAuto?.weight).toBe(10);
   });
 
+  it('rate limits repeated route overview reads', async () => {
+    resetTokenRouteReadLimitersForTests({
+      summaryPoints: 1,
+      listPoints: 1,
+    });
+
+    const firstSummary = await app.inject({
+      method: 'GET',
+      url: '/api/routes/summary',
+    });
+    const secondSummary = await app.inject({
+      method: 'GET',
+      url: '/api/routes/summary',
+    });
+    const firstRoutes = await app.inject({
+      method: 'GET',
+      url: '/api/routes',
+    });
+    const secondRoutes = await app.inject({
+      method: 'GET',
+      url: '/api/routes',
+    });
+
+    expect(firstSummary.statusCode).toBe(200);
+    expect(secondSummary.statusCode).toBe(429);
+    expect(firstRoutes.statusCode).toBe(200);
+    expect(secondRoutes.statusCode).toBe(429);
+  });
+
   it('creates explicit-group routes with sourceRouteIds and aggregates source channels', async () => {
     const sourceA = await seedAccountWithToken('claude-opus-4-5');
     const sourceB = await seedAccountWithToken('claude-sonnet-4-5');

src/server/routes/api/tokens.ts

@@ -1,5 +1,6 @@
-import { FastifyInstance } from 'fastify';
+import { FastifyInstance, type FastifyReply } from 'fastify';
 import { and, eq, inArray } from 'drizzle-orm';
+import { RateLimiterMemory, RateLimiterRes } from 'rate-limiter-flexible';
 import { db, schema } from '../../db/index.js';
 import { requireInsertedRowId } from '../../db/insertHelpers.js';
 import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js';
@@ -26,6 +27,10 @@ import {
   ROUTE_DECISION_REFRESH_DEDUPE_KEY,
   ROUTE_DECISION_REFRESH_TASK_TYPE,
 } from '../../services/routeDecisionRefreshService.js';
+import {
+  listOauthRouteUnitMembersByUnitIds,
+  loadOauthRouteUnitSummariesByIds,
+} from '../../services/oauth/routeUnitService.js';
 import { normalizeTokenRouteMode, type RouteMode } from '../../../shared/tokenRouteContract.js';
 import {
   parseRouteChannelBatchCreatePayload,
@@ -36,7 +41,33 @@ import {
   parseTokenRouteCreatePayload,
   parseTokenRouteUpdatePayload,
 } from '../../contracts/tokenRoutePayloads.js';
-
+
+function createTokenRouteReadLimiter(keyPrefix: string, points = 60) {
+  return new RateLimiterMemory({
+    keyPrefix,
+    points,
+    duration: 60,
+  });
+}
+
+let routeSummaryReadLimiter = createTokenRouteReadLimiter('token-routes-summary-read');
+let routeListReadLimiter = createTokenRouteReadLimiter('token-routes-list-read');
+
+export function resetTokenRouteReadLimitersForTests(options: {
+  summaryPoints?: number;
+  listPoints?: number;
+} = {}): void {
+  routeSummaryReadLimiter = createTokenRouteReadLimiter('token-routes-summary-read', options.summaryPoints ?? 60);
+  routeListReadLimiter = createTokenRouteReadLimiter('token-routes-list-read', options.listPoints ?? 60);
+}
+
+function sendTokenRouteRateLimit(reply: FastifyReply, error: unknown): void {
+  const retryState = error instanceof RateLimiterRes ? error : null;
+  const retryAfterSec = Math.max(1, Math.ceil((retryState?.msBeforeNext ?? 60_000) / 1000));
+  reply.code(429).header('retry-after', String(retryAfterSec))
+    .send({ success: false, message: '请求过于频繁，请稍后再试' });
+}
+
 function isExactModelPattern(modelPattern: string): boolean {
   const normalized = modelPattern.trim();
   if (!normalized) return false;
@@ -610,8 +641,14 @@ type RouteChannelSummary = {
   siteNames: Set<string>;
 };
 
-async function fetchChannelsForRouteRows(routes: RouteRow[]): Promise<Map<number, any[]>> {
+async function fetchChannelsForRouteRows(
+  routes: RouteRow[],
+  options: {
+    includeRouteUnitDetails?: boolean;
+  } = {},
+): Promise<Map<number, any[]>> {
   if (routes.length === 0) return new Map();
+  const includeRouteUnitDetails = options.includeRouteUnitDetails !== false;
 
   const explicitSourceRouteIds = Array.from(new Set(routes
     .filter((route) => isExplicitGroupRoute(route))
@@ -652,6 +689,18 @@ async function fetchChannelsForRouteRows(routes: RouteRow[]): Promise<Map<number
     .where(inArray(schema.routeChannels.routeId, actualRouteIds))
     .all();
 
+  const oauthRouteUnitIds: number[] = Array.from(new Set<number>(
+    channelRows
+      .map((row) => Number(row.route_channels.oauthRouteUnitId))
+      .filter((id): id is number => Number.isFinite(id) && id > 0),
+  ));
+  const routeUnitSummaries = includeRouteUnitDetails
+    ? await loadOauthRouteUnitSummariesByIds(oauthRouteUnitIds)
+    : new Map();
+  const routeUnitMembersByUnitId = includeRouteUnitDetails
+    ? await listOauthRouteUnitMembersByUnitIds(oauthRouteUnitIds)
+    : new Map();
+
   const channelsByActualRouteId = new Map<number, any[]>();
 
   for (const row of channelRows) {
@@ -662,6 +711,9 @@ async function fetchChannelsForRouteRows(routes: RouteRow[]): Promise<Map<number
       : null;
     const resolvedSourceModel = (row.route_channels.sourceModel || fallbackSourceModel || '').trim();
     if (!channelsByActualRouteId.has(routeId)) channelsByActualRouteId.set(routeId, []);
+    const routeUnit = row.route_channels.oauthRouteUnitId
+      ? routeUnitSummaries.get(row.route_channels.oauthRouteUnitId) || null
+      : null;
     channelsByActualRouteId.get(routeId)!.push({
       ...row.route_channels,
       sourceModel: resolvedSourceModel || null,
@@ -672,9 +724,22 @@ async function fetchChannelsForRouteRows(routes: RouteRow[]): Promise<Map<number
           id: row.account_tokens.id,
           name: row.account_tokens.name,
           accountId: row.account_tokens.accountId,
-          enabled: row.account_tokens.enabled,
-          isDefault: row.account_tokens.isDefault,
-        }
+          enabled: row.account_tokens.enabled,
+          isDefault: row.account_tokens.isDefault,
+        }
+        : null,
+      routeUnit: includeRouteUnitDetails && routeUnit
+        ? {
+          id: routeUnit.id,
+          name: routeUnit.name,
+          strategy: routeUnit.strategy,
+          memberCount: routeUnit.memberCount,
+          members: (routeUnitMembersByUnitId.get(routeUnit.id) || []).map((member) => ({
+            accountId: member.account.id,
+            username: member.account.username,
+            siteName: member.site.name,
+          })),
+        }
         : null,
     });
   }
@@ -703,7 +768,7 @@ async function fetchChannelsForRoutes(routeIds: number[]): Promise<Map<number, a
 }
 
 async function buildRouteChannelSummaryMap(routes: RouteRow[]): Promise<Map<number, RouteChannelSummary>> {
-  const channelsByRoute = await fetchChannelsForRouteRows(routes);
+  const channelsByRoute = await fetchChannelsForRouteRows(routes, { includeRouteUnitDetails: false });
   const summaryByRoute = new Map<number, RouteChannelSummary>();
   for (const route of routes) {
     const channels = channelsByRoute.get(route.id) || [];
@@ -738,7 +803,13 @@ export async function tokensRoutes(app: FastifyInstance) {
   });
 
   // Route summary (no channel details) for first-screen rendering
-  app.get('/api/routes/summary', async () => {
+  app.get('/api/routes/summary', async (request, reply) => {
+    try {
+      await routeSummaryReadLimiter.consume(request.ip);
+    } catch (error) {
+      sendTokenRouteRateLimit(reply, error);
+      return;
+    }
     const routes = await listRoutesWithSources();
     if (routes.length === 0) return [];
     const aggByRoute = await buildRouteChannelSummaryMap(routes);
@@ -862,7 +933,13 @@ export async function tokensRoutes(app: FastifyInstance) {
   });
 
   // List all routes
-  app.get('/api/routes', async () => {
+  app.get('/api/routes', async (request, reply) => {
+    try {
+      await routeListReadLimiter.consume(request.ip);
+    } catch (error) {
+      sendTokenRouteRateLimit(reply, error);
+      return;
+    }
     const routes = await listRoutesWithSources();
     if (routes.length === 0) return [];