From 3877f3a3f15f562a28b8d293c72d36dba076fbb0 Mon Sep 17 00:00:00 2001 From: Cita Date: Mon, 23 Mar 2026 11:53:59 +0800 Subject: [PATCH] refactor: centralize oauth identity and route refresh workflow --- src/server/index.ts | 6 +- .../proxy-core/surfaces/geminiSurface.ts | 4 +- .../proxy-core/surfaces/sharedSurface.ts | 4 +- src/server/routes/api/oauth.test.ts | 63 ++++++++++++++ .../routeRefreshWorkflow.architecture.test.ts | 57 +++++++++++++ src/server/routes/api/settings.ts | 4 +- src/server/routes/api/stats.ts | 11 +-- src/server/routes/api/tokens.ts | 10 +-- src/server/routes/proxy/completions.ts | 4 +- src/server/routes/proxy/embeddings.ts | 4 +- src/server/routes/proxy/images.ts | 6 +- src/server/routes/proxy/models.ts | 4 +- src/server/routes/proxy/search.ts | 4 +- src/server/routes/proxy/videos.ts | 4 +- .../services/accountMutationWorkflow.ts | 19 ++--- src/server/services/backgroundTaskService.ts | 13 ++- src/server/services/checkinScheduler.ts | 4 +- .../services/oauth/oauthAccount.test.ts | 37 ++++++++ src/server/services/oauth/oauthAccount.ts | 84 +++++++++++++------ .../services/oauth/oauthIdentityBackfill.ts | 43 ++++++++++ src/server/services/oauth/service.ts | 11 ++- src/server/services/routeRefreshWorkflow.ts | 43 ++++++++++ 22 files changed, 361 insertions(+), 78 deletions(-) create mode 100644 src/server/routes/api/routeRefreshWorkflow.architecture.test.ts create mode 100644 src/server/services/oauth/oauthIdentityBackfill.ts create mode 100644 src/server/services/routeRefreshWorkflow.ts diff --git a/src/server/index.ts b/src/server/index.ts index 5eb80ad..8ee121a 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -22,13 +22,14 @@ import { oauthRoutes } from './routes/api/oauth.js'; import { siteAnnouncementsRoutes } from './routes/api/siteAnnouncements.js'; import { proxyRoutes } from './routes/proxy/router.js'; import { startScheduler } from './services/checkinScheduler.js'; -import { rebuildTokenRoutesFromAvailability } from './services/modelService.js'; +import * as routeRefreshWorkflow from './services/routeRefreshWorkflow.js'; import { startProxyFileRetentionService, stopProxyFileRetentionService } from './services/proxyFileRetentionService.js'; import { setLegacyProxyLogRetentionFallbackEnabled, stopProxyLogRetentionService } from './services/proxyLogRetentionService.js'; import { buildStartupSummaryLines } from './services/startupInfo.js'; import { repairStoredCreatedAtValues } from './services/storedTimestampRepairService.js'; import { migrateSiteApiKeysToAccounts } from './services/siteApiKeyMigrationService.js'; import { ensureDefaultSitesSeeded } from './services/defaultSiteSeedService.js'; +import { ensureOauthIdentityBackfill } from './services/oauth/oauthIdentityBackfill.js'; import { startOAuthLoopbackCallbackServers, stopOAuthLoopbackCallbackServers } from './services/oauth/localCallbackServer.js'; import { startSiteAnnouncementPolling } from './services/siteAnnouncementPollingService.js'; import { reloadBackupWebdavScheduler } from './services/backupService.js'; @@ -332,7 +333,8 @@ try { await repairStoredCreatedAtValues(); await migrateSiteApiKeysToAccounts(); await ensureDefaultSitesSeeded(); - await rebuildTokenRoutesFromAvailability(); + await ensureOauthIdentityBackfill(); + await routeRefreshWorkflow.rebuildRoutesOnly(); console.log('Loaded runtime settings overrides'); } catch (error) { diff --git a/src/server/proxy-core/surfaces/geminiSurface.ts b/src/server/proxy-core/surfaces/geminiSurface.ts index f531c3f..b1b4208 100644 --- a/src/server/proxy-core/surfaces/geminiSurface.ts +++ b/src/server/proxy-core/surfaces/geminiSurface.ts @@ -11,7 +11,7 @@ import { buildOauthProviderHeaders } from '../../services/oauth/service.js'; import { getOauthInfoFromAccount } from '../../services/oauth/oauthAccount.js'; import { refreshOauthAccessTokenSingleflight } from '../../services/oauth/refreshSingleflight.js'; import { resolveChannelProxyUrl, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; -import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; +import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { getDownstreamRoutingPolicy } from '../../routes/proxy/downstreamPolicy.js'; import { executeEndpointFlow, type BuiltEndpointRequest } from '../../routes/proxy/endpointFlow.js'; import { composeProxyLogMessage } from '../../routes/proxy/logPathMeta.js'; @@ -278,7 +278,7 @@ export async function geminiProxyRoute(app: FastifyInstance) { if (!isDirectGeminiFamilyPlatform(selected.site.platform)) { let models = await readRouteAwareGeminiModels(request); if (models.length <= 0) { - await refreshModelsAndRebuildRoutes(); + await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); models = await readRouteAwareGeminiModels(request); } return reply.code(200).send({ models }); diff --git a/src/server/proxy-core/surfaces/sharedSurface.ts b/src/server/proxy-core/surfaces/sharedSurface.ts index e2a41f7..f1f91b9 100644 --- a/src/server/proxy-core/surfaces/sharedSurface.ts +++ b/src/server/proxy-core/surfaces/sharedSurface.ts @@ -1,4 +1,4 @@ -import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; +import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { formatUtcSqlDateTime } from '../../services/localTimeService.js'; import { resolveChannelProxyUrl, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; import type { SiteProxyConfigLike } from '../../services/siteProxy.js'; @@ -86,7 +86,7 @@ export async function selectSurfaceChannelForAttempt(input: { ); if (!selected && input.retryCount === 0) { - await refreshModelsAndRebuildRoutes(); + await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); selected = await tokenRouter.selectChannel(input.requestedModel, input.downstreamPolicy); } diff --git a/src/server/routes/api/oauth.test.ts b/src/server/routes/api/oauth.test.ts index 4229d4d..c6d6350 100644 --- a/src/server/routes/api/oauth.test.ts +++ b/src/server/routes/api/oauth.test.ts @@ -1537,6 +1537,69 @@ describe('oauth routes', { timeout: 15_000 }, () => { }); }); + it('backfills structured oauth identity columns for legacy oauth rows before listing connections', async () => { + const site = await db.insert(schema.sites).values({ + name: 'Legacy Codex', + url: 'https://codex.example.com', + platform: 'codex', + status: 'active', + useSystemProxy: false, + isPinned: false, + globalWeight: 1, + sortOrder: 0, + }).returning().get(); + + const account = await db.insert(schema.accounts).values({ + siteId: site.id, + username: 'legacy-user@example.com', + accessToken: 'legacy-oauth-access-token', + status: 'active', + extraConfig: JSON.stringify({ + oauth: { + provider: 'codex', + accountKey: 'legacy-chatgpt-account', + projectId: 'legacy-project', + refreshToken: 'legacy-refresh-token', + modelDiscoveryStatus: 'healthy', + }, + }), + isPinned: false, + sortOrder: 0, + }).returning().get(); + + expect(account.oauthProvider).toBeNull(); + expect(account.oauthAccountKey).toBeNull(); + expect(account.oauthProjectId).toBeNull(); + + const response = await app.inject({ + method: 'GET', + url: '/api/oauth/connections', + }); + + expect(response.statusCode).toBe(200); + expect(response.json()).toMatchObject({ + total: 1, + items: [ + expect.objectContaining({ + provider: 'codex', + accountKey: 'legacy-chatgpt-account', + projectId: 'legacy-project', + username: 'legacy-user@example.com', + }), + ], + }); + + const backfilled = await db.select().from(schema.accounts) + .where(eq(schema.accounts.id, account.id)) + .get(); + + expect(backfilled).toEqual(expect.objectContaining({ + oauthProvider: 'codex', + oauthAccountKey: 'legacy-chatgpt-account', + oauthProjectId: 'legacy-project', + })); + }); + it('rejects malformed manual callback submissions', async () => { const startResponse = await app.inject({ method: 'POST', diff --git a/src/server/routes/api/routeRefreshWorkflow.architecture.test.ts b/src/server/routes/api/routeRefreshWorkflow.architecture.test.ts new file mode 100644 index 0000000..7d2efd5 --- /dev/null +++ b/src/server/routes/api/routeRefreshWorkflow.architecture.test.ts @@ -0,0 +1,57 @@ +import { readFileSync } from 'node:fs'; +import { describe, expect, it } from 'vitest'; + +function readSource(relativePath: string): string { + return readFileSync(new URL(relativePath, import.meta.url), 'utf8'); +} + +function expectNoDirectModelServiceRouteRefresh(source: string): void { + expect(source).not.toMatch(/import\s*\{[^}]*\brefreshModelsAndRebuildRoutes\b[^}]*\}\s*from\s*['"][^'"]*modelService\.js['"]/m); + expect(source).not.toMatch(/import\s*\{[^}]*\brebuildTokenRoutesFromAvailability\b[^}]*\}\s*from\s*['"][^'"]*modelService\.js['"]/m); +} + +describe('route refresh workflow architecture boundaries', () => { + it('keeps api controllers on the shared route refresh workflow instead of modelService', () => { + const tokensSource = readSource('./tokens.ts'); + const settingsSource = readSource('./settings.ts'); + const statsSource = readSource('./stats.ts'); + + expect(tokensSource).toContain("from '../../services/routeRefreshWorkflow.js'"); + expect(tokensSource).not.toContain("from '../../services/modelService.js'"); + + expect(settingsSource).toContain("from '../../services/routeRefreshWorkflow.js'"); + expect(settingsSource).not.toContain("from '../../services/modelService.js'"); + + expect(statsSource).toContain("from '../../services/routeRefreshWorkflow.js'"); + expectNoDirectModelServiceRouteRefresh(statsSource); + }); + + it('keeps proxy fallback refreshes and scheduler hooks on the route refresh workflow', () => { + const completionsSource = readSource('../proxy/completions.ts'); + const embeddingsSource = readSource('../proxy/embeddings.ts'); + const imagesSource = readSource('../proxy/images.ts'); + const modelsRouteSource = readSource('../proxy/models.ts'); + const searchSource = readSource('../proxy/search.ts'); + const videosSource = readSource('../proxy/videos.ts'); + const schedulerSource = readSource('../../services/checkinScheduler.ts'); + const oauthServiceSource = readSource('../../services/oauth/service.ts'); + const sharedSurfaceSource = readSource('../../proxy-core/surfaces/sharedSurface.ts'); + const geminiSurfaceSource = readSource('../../proxy-core/surfaces/geminiSurface.ts'); + + for (const source of [ + completionsSource, + embeddingsSource, + imagesSource, + modelsRouteSource, + searchSource, + videosSource, + schedulerSource, + oauthServiceSource, + sharedSurfaceSource, + geminiSurfaceSource, + ]) { + expect(source).toContain('routeRefreshWorkflow'); + expectNoDirectModelServiceRouteRefresh(source); + } + }); +}); diff --git a/src/server/routes/api/settings.ts b/src/server/routes/api/settings.ts index f99f2ba..4b2bda9 100644 --- a/src/server/routes/api/settings.ts +++ b/src/server/routes/api/settings.ts @@ -4,7 +4,7 @@ import { fetch } from 'undici'; import { config } from '../../config.js'; import { db, runtimeDbDialect, schema } from '../../db/index.js'; import { upsertSetting } from '../../db/upsertSetting.js'; -import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; +import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { updateBalanceRefreshCron, updateCheckinSchedule, updateLogCleanupSettings } from '../../services/checkinScheduler.js'; import { sendNotification } from '../../services/notifyService.js'; import { @@ -1404,7 +1404,7 @@ export async function settingsRoutes(app: FastifyInstance) { }, failureMessage: (currentTask) => `缓存清理后重建失败：${currentTask.error || 'unknown error'}`, }, - async () => refreshModelsAndRebuildRoutes(), + async () => routeRefreshWorkflow.refreshModelsAndRebuildRoutes(), ); return reply.code(202).send({ diff --git a/src/server/routes/api/stats.ts b/src/server/routes/api/stats.ts index 385f59d..1485e6c 100644 --- a/src/server/routes/api/stats.ts +++ b/src/server/routes/api/stats.ts @@ -1,11 +1,8 @@ import { FastifyInstance } from 'fastify'; import { db, schema } from '../../db/index.js'; import { and, desc, eq, gte, lt, sql } from 'drizzle-orm'; -import { - refreshModelsForAccount, - refreshModelsAndRebuildRoutes, - rebuildTokenRoutesFromAvailability, -} from '../../services/modelService.js'; +import { refreshModelsForAccount } from '../../services/modelService.js'; +import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { buildModelAnalysis } from '../../services/modelAnalysisService.js'; import { fallbackTokenCost, fetchModelPricingCatalog } from '../../services/modelPricingService.js'; import { getUpstreamModelDescriptionsCached } from '../../services/upstreamModelDescriptionService.js'; @@ -847,7 +844,7 @@ export async function statsRoutes(app: FastifyInstance) { }, failureMessage: (currentTask) => `模型广场刷新失败：${currentTask.error || 'unknown error'}`, }, - async () => refreshModelsAndRebuildRoutes(), + async () => routeRefreshWorkflow.refreshModelsAndRebuildRoutes(), ); refreshQueued = !reused; refreshReused = reused; @@ -1390,7 +1387,7 @@ export async function statsRoutes(app: FastifyInstance) { } const refresh = await refreshModelsForAccount(accountId); - const rebuild = rebuildTokenRoutesFromAvailability(); + const rebuild = routeRefreshWorkflow.rebuildRoutesOnly(); return { success: true, refresh, rebuild }; }); diff --git a/src/server/routes/api/tokens.ts b/src/server/routes/api/tokens.ts index 9b11928..5af88e8 100644 --- a/src/server/routes/api/tokens.ts +++ b/src/server/routes/api/tokens.ts @@ -1,7 +1,7 @@ import { FastifyInstance } from 'fastify'; import { and, eq, inArray } from 'drizzle-orm'; import { db, schema } from '../../db/index.js'; -import { rebuildTokenRoutesFromAvailability, refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; +import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { ACCOUNT_TOKEN_VALUE_STATUS_READY, isUsableAccountToken, @@ -1170,12 +1170,12 @@ export async function tokensRoutes(app: FastifyInstance) { app.post<{ Body?: { refreshModels?: boolean; wait?: boolean } }>('/api/routes/rebuild', async (request, reply) => { const body = (request.body || {}) as { refreshModels?: boolean }; if (body.refreshModels === false) { - const rebuild = rebuildTokenRoutesFromAvailability(); + const rebuild = routeRefreshWorkflow.rebuildRoutesOnly(); return { success: true, rebuild }; } if ((request.body as { wait?: boolean } | undefined)?.wait) { - const result = await refreshModelsAndRebuildRoutes(); + const result = await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); return { success: true, ...result }; } @@ -1192,8 +1192,8 @@ export async function tokensRoutes(app: FastifyInstance) { }, failureMessage: (currentTask) => `刷新模型并重建路由失败：${currentTask.error || 'unknown error'}`, }, - async () => refreshModelsAndRebuildRoutes(), - ); + async () => routeRefreshWorkflow.refreshModelsAndRebuildRoutes(), + ); return reply.code(202).send({ success: true, diff --git a/src/server/routes/proxy/completions.ts b/src/server/routes/proxy/completions.ts index 5d49328..6aca4ed 100644 --- a/src/server/routes/proxy/completions.ts +++ b/src/server/routes/proxy/completions.ts @@ -1,7 +1,7 @@ import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; import { fetch } from 'undici'; import { tokenRouter } from '../../services/tokenRouter.js'; -import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; +import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; import { isTokenExpiredError } from '../../services/alertRules.js'; import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; @@ -48,7 +48,7 @@ export async function completionsProxyRoute(app: FastifyInstance) { : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); if (!selected && retryCount === 0) { - await refreshModelsAndRebuildRoutes(); + await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); } diff --git a/src/server/routes/proxy/embeddings.ts b/src/server/routes/proxy/embeddings.ts index 3741523..1ea22b7 100644 --- a/src/server/routes/proxy/embeddings.ts +++ b/src/server/routes/proxy/embeddings.ts @@ -1,7 +1,7 @@ import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; import { fetch } from 'undici'; import { tokenRouter } from '../../services/tokenRouter.js'; -import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; +import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; import { isTokenExpiredError } from '../../services/alertRules.js'; import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; @@ -46,7 +46,7 @@ export async function embeddingsProxyRoute(app: FastifyInstance) { : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); if (!selected && retryCount === 0) { - await refreshModelsAndRebuildRoutes(); + await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); } diff --git a/src/server/routes/proxy/images.ts b/src/server/routes/proxy/images.ts index 4d12c31..54c8d8f 100644 --- a/src/server/routes/proxy/images.ts +++ b/src/server/routes/proxy/images.ts @@ -1,7 +1,7 @@ import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; import { fetch } from 'undici'; import { tokenRouter } from '../../services/tokenRouter.js'; -import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; +import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; import { isTokenExpiredError } from '../../services/alertRules.js'; import { estimateProxyCost } from '../../services/modelPricingService.js'; @@ -43,7 +43,7 @@ export async function imagesProxyRoute(app: FastifyInstance) { : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); if (!selected && retryCount === 0) { - await refreshModelsAndRebuildRoutes(); + await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); } @@ -189,7 +189,7 @@ export async function imagesProxyRoute(app: FastifyInstance) { : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); if (!selected && retryCount === 0) { - await refreshModelsAndRebuildRoutes(); + await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); } diff --git a/src/server/routes/proxy/models.ts b/src/server/routes/proxy/models.ts index 708a7a3..e9e7aff 100644 --- a/src/server/routes/proxy/models.ts +++ b/src/server/routes/proxy/models.ts @@ -1,6 +1,6 @@ import { FastifyInstance } from 'fastify'; import { listModelsSurface } from '../../proxy-core/surfaces/modelsSurface.js'; -import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; +import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { getDownstreamRoutingPolicy } from './downstreamPolicy.js'; import { tokenRouter } from '../../services/tokenRouter.js'; import { isModelAllowedByPolicyOrAllowedRoutes } from '../../services/downstreamApiKeyService.js'; @@ -14,7 +14,7 @@ export async function modelsProxyRoute(app: FastifyInstance) { downstreamPolicy, responseFormat: wantsClaudeFormat ? 'claude' : 'openai', tokenRouter, - refreshModelsAndRebuildRoutes, + refreshModelsAndRebuildRoutes: routeRefreshWorkflow.refreshModelsAndRebuildRoutes, isModelAllowed: isModelAllowedByPolicyOrAllowedRoutes, }); }); diff --git a/src/server/routes/proxy/search.ts b/src/server/routes/proxy/search.ts index d91cbd4..a609ff2 100644 --- a/src/server/routes/proxy/search.ts +++ b/src/server/routes/proxy/search.ts @@ -1,7 +1,7 @@ import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; import { fetch } from 'undici'; import { tokenRouter } from '../../services/tokenRouter.js'; -import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; +import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; import { isTokenExpiredError } from '../../services/alertRules.js'; import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; @@ -71,7 +71,7 @@ export async function searchProxyRoute(app: FastifyInstance) { : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); if (!selected && retryCount === 0) { - await refreshModelsAndRebuildRoutes(); + await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); } diff --git a/src/server/routes/proxy/videos.ts b/src/server/routes/proxy/videos.ts index 10e5a96..6c3f218 100644 --- a/src/server/routes/proxy/videos.ts +++ b/src/server/routes/proxy/videos.ts @@ -1,7 +1,7 @@ import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; import { fetch } from 'undici'; import { tokenRouter } from '../../services/tokenRouter.js'; -import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js'; +import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; import { isTokenExpiredError } from '../../services/alertRules.js'; import { estimateProxyCost } from '../../services/modelPricingService.js'; @@ -57,7 +57,7 @@ export async function videosProxyRoute(app: FastifyInstance) { : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); if (!selected && retryCount === 0) { - await refreshModelsAndRebuildRoutes(); + await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); } diff --git a/src/server/services/accountMutationWorkflow.ts b/src/server/services/accountMutationWorkflow.ts index b4c1575..7749ab5 100644 --- a/src/server/services/accountMutationWorkflow.ts +++ b/src/server/services/accountMutationWorkflow.ts @@ -4,10 +4,10 @@ import { syncTokensFromUpstream, } from './accountTokenService.js'; import { - rebuildTokenRoutesFromAvailability, refreshModelsForAccount, type ModelRefreshResult, } from './modelService.js'; +import * as routeRefreshWorkflow from './routeRefreshWorkflow.js'; type UpstreamTokenLike = { name?: string | null; @@ -17,16 +17,11 @@ type UpstreamTokenLike = { }; export type CoverageBatchRebuildResult = - | { success: true; result: Awaited> } + | { success: true; result: Awaited> } | { success: false; error: string }; export async function rebuildRoutesBestEffort(): Promise { - try { - await rebuildTokenRoutesFromAvailability(); - return true; - } catch { - return false; - } + return routeRefreshWorkflow.rebuildRoutesBestEffort(); } export async function convergeAccountMutation(input: { @@ -47,7 +42,7 @@ export async function convergeAccountMutation(input: { rebuiltRoutes: boolean; balanceResult: Awaited> | null; modelRefreshResult: ModelRefreshResult | null; - rebuildResult: Awaited> | null; + rebuildResult: Awaited> | null; }> { const result = { defaultTokenId: null as number | null, @@ -57,7 +52,7 @@ export async function convergeAccountMutation(input: { rebuiltRoutes: false, balanceResult: null as Awaited> | null, modelRefreshResult: null as ModelRefreshResult | null, - rebuildResult: null as Awaited> | null, + rebuildResult: null as Awaited> | null, }; const runStep = async (fn: () => Promise): Promise => { @@ -124,7 +119,7 @@ export async function convergeAccountMutation(input: { } if (input.rebuildRoutes) { - const rebuildResult = await runStep(() => rebuildTokenRoutesFromAvailability()); + const rebuildResult = await runStep(() => routeRefreshWorkflow.rebuildRoutesOnly()); if (rebuildResult) { result.rebuildResult = rebuildResult; result.rebuiltRoutes = true; @@ -180,7 +175,7 @@ export async function refreshAccountCoverageBatch(input: { refresh, rebuild: { success: true, - result: await rebuildTokenRoutesFromAvailability(), + result: await routeRefreshWorkflow.rebuildRoutesOnly(), }, }; } catch (error) { diff --git a/src/server/services/backgroundTaskService.ts b/src/server/services/backgroundTaskService.ts index 140554c..04e2740 100644 --- a/src/server/services/backgroundTaskService.ts +++ b/src/server/services/backgroundTaskService.ts @@ -40,6 +40,7 @@ const TASK_CLEANUP_INTERVAL_MS = 60 * 1000; const tasks = new Map(); const dedupeTaskIds = new Map(); +let cleanupTimer: ReturnType | null = null; function nowIso() { return new Date().toISOString(); @@ -157,13 +158,17 @@ function cleanupExpiredTasks() { } } -const cleanupTimer = setInterval(cleanupExpiredTasks, TASK_CLEANUP_INTERVAL_MS); -cleanupTimer.unref?.(); +function ensureCleanupTimer() { + if (cleanupTimer) return; + cleanupTimer = setInterval(cleanupExpiredTasks, TASK_CLEANUP_INTERVAL_MS); + cleanupTimer.unref?.(); +} export function startBackgroundTask( options: BackgroundTaskStartOptions, runner: () => Promise, ): { task: BackgroundTask; reused: boolean } { + ensureCleanupTimer(); const dedupeKey = options.dedupeKey?.trim() || ''; if (dedupeKey) { const existingTaskId = dedupeTaskIds.get(dedupeKey); @@ -241,4 +246,8 @@ export function summarizeCheckinResults(results: Array<{ result?: any }>): { tot export function __resetBackgroundTasksForTests() { tasks.clear(); dedupeTaskIds.clear(); + if (cleanupTimer) { + clearInterval(cleanupTimer); + cleanupTimer = null; + } } diff --git a/src/server/services/checkinScheduler.ts b/src/server/services/checkinScheduler.ts index e758ec1..09e9e2d 100644 --- a/src/server/services/checkinScheduler.ts +++ b/src/server/services/checkinScheduler.ts @@ -4,7 +4,7 @@ import { config } from '../config.js'; import { db, schema } from '../db/index.js'; import { refreshAllBalances } from './balanceService.js'; import { checkinAll } from './checkinService.js'; -import { refreshModelsAndRebuildRoutes } from './modelService.js'; +import * as routeRefreshWorkflow from './routeRefreshWorkflow.js'; import { sendNotification } from './notifyService.js'; import { buildDailySummaryNotification, collectDailySummaryMetrics } from './dailySummaryService.js'; import { cleanupConfiguredLogs, normalizeLogCleanupRetentionDays } from './logCleanupService.js'; @@ -162,7 +162,7 @@ function createBalanceTask(cronExpr: string) { console.log(`[Scheduler] Refreshing balances at ${new Date().toISOString()}`); try { await refreshAllBalances(); - await refreshModelsAndRebuildRoutes(); + await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); console.log('[Scheduler] Balance refresh complete'); } catch (err) { console.error('[Scheduler] Balance refresh error:', err); diff --git a/src/server/services/oauth/oauthAccount.test.ts b/src/server/services/oauth/oauthAccount.test.ts index 6ca83f3..904c55a 100644 --- a/src/server/services/oauth/oauthAccount.test.ts +++ b/src/server/services/oauth/oauthAccount.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from 'vitest'; import { + buildOauthIdentityBackfillPatch, buildOauthInfoFromAccount, buildStoredOauthStateFromAccount, getOauthInfoFromAccount, @@ -133,4 +134,40 @@ describe('oauth account identity helpers', () => { quota: { status: 'supported' }, }); }); + + it('builds a structured identity backfill patch from legacy oauth metadata only for missing columns', () => { + expect(buildOauthIdentityBackfillPatch({ + oauthProvider: null, + oauthAccountKey: null, + oauthProjectId: null, + extraConfig: JSON.stringify({ + oauth: { + provider: 'codex', + accountKey: 'legacy-account', + projectId: 'legacy-project', + refreshToken: 'refresh-token', + }, + }), + })).toEqual({ + oauthProvider: 'codex', + oauthAccountKey: 'legacy-account', + oauthProjectId: 'legacy-project', + }); + + expect(buildOauthIdentityBackfillPatch({ + oauthProvider: 'codex', + oauthAccountKey: 'structured-account', + oauthProjectId: null, + extraConfig: JSON.stringify({ + oauth: { + provider: 'legacy-provider-ignored', + accountKey: 'legacy-account-ignored', + projectId: 'legacy-project', + refreshToken: 'refresh-token', + }, + }), + })).toEqual({ + oauthProjectId: 'legacy-project', + }); + }); }); diff --git a/src/server/services/oauth/oauthAccount.ts b/src/server/services/oauth/oauthAccount.ts index 79f3a86..77710b8 100644 --- a/src/server/services/oauth/oauthAccount.ts +++ b/src/server/services/oauth/oauthAccount.ts @@ -52,6 +52,8 @@ type OauthIdentityCarrier = { oauthProjectId?: string | null; }; +type StoredOauthIdentity = Pick; + function parseExtraConfig(extraConfig?: string | null): ParsedExtraConfig { if (!extraConfig) return {}; try { @@ -113,17 +115,26 @@ function asModelDiscoveryStatus(value: unknown): OauthModelDiscoveryStatus | und return undefined; } -function parseStoredOauthState(extraConfig?: string | null): Partial | null { +function parseStoredOauthIdentity(extraConfig?: string | null): StoredOauthIdentity | null { const parsed = parseExtraConfig(extraConfig).oauth; if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) return null; const accountKey = asTrimmedString(parsed.accountKey) || asTrimmedString(parsed.accountId); + const provider = asTrimmedString(parsed.provider); + if (!provider) return null; return { - provider: asTrimmedString(parsed.provider), + provider, accountId: asTrimmedString(parsed.accountId) || accountKey, accountKey, + projectId: asTrimmedString(parsed.projectId), + }; +} + +function parseStoredOauthRuntimeState(extraConfig?: string | null): Partial | null { + const parsed = parseExtraConfig(extraConfig).oauth; + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) return null; + return { email: asTrimmedString(parsed.email), planType: asTrimmedString(parsed.planType), - projectId: asTrimmedString(parsed.projectId), tokenExpiresAt: asPositiveInteger(parsed.tokenExpiresAt), refreshToken: asTrimmedString(parsed.refreshToken), idToken: asTrimmedString(parsed.idToken), @@ -137,45 +148,68 @@ function parseStoredOauthState(extraConfig?: string | null): Partial } export function getOauthInfoFromExtraConfig(extraConfig?: string | null): OauthInfo | null { - const parsed = parseStoredOauthState(extraConfig); - const provider = parsed?.provider; + const identity = parseStoredOauthIdentity(extraConfig); + const runtime = parseStoredOauthRuntimeState(extraConfig); + const provider = identity?.provider; if (!provider) return null; return { provider, - accountId: parsed.accountId || parsed.accountKey, - accountKey: parsed.accountKey, - email: parsed.email, - planType: parsed.planType, - projectId: parsed.projectId, - tokenExpiresAt: parsed.tokenExpiresAt, - refreshToken: parsed.refreshToken, - idToken: parsed.idToken, - providerData: parsed.providerData, - quota: parsed.quota, - modelDiscoveryStatus: parsed.modelDiscoveryStatus, - lastModelSyncAt: parsed.lastModelSyncAt, - lastModelSyncError: parsed.lastModelSyncError, - lastDiscoveredModels: parsed.lastDiscoveredModels, + accountId: identity?.accountId || identity?.accountKey, + accountKey: identity?.accountKey, + projectId: identity?.projectId, + email: runtime?.email, + planType: runtime?.planType, + tokenExpiresAt: runtime?.tokenExpiresAt, + refreshToken: runtime?.refreshToken, + idToken: runtime?.idToken, + providerData: runtime?.providerData, + quota: runtime?.quota, + modelDiscoveryStatus: runtime?.modelDiscoveryStatus, + lastModelSyncAt: runtime?.lastModelSyncAt, + lastModelSyncError: runtime?.lastModelSyncError, + lastDiscoveredModels: runtime?.lastDiscoveredModels, }; } export function getOauthInfoFromAccount(account?: OauthIdentityCarrier | null): OauthInfo | null { if (!account) return null; - const stored = parseStoredOauthState(account.extraConfig); - const provider = asTrimmedString(account.oauthProvider) || stored?.provider; + const storedIdentity = parseStoredOauthIdentity(account.extraConfig); + const storedRuntime = parseStoredOauthRuntimeState(account.extraConfig); + const provider = asTrimmedString(account.oauthProvider) || storedIdentity?.provider; if (!provider) return null; const structuredAccountKey = asTrimmedString(account.oauthAccountKey); - const accountKey = structuredAccountKey || stored?.accountKey || stored?.accountId; - const projectId = asTrimmedString(account.oauthProjectId) || stored?.projectId; + const accountKey = structuredAccountKey || storedIdentity?.accountKey || storedIdentity?.accountId; + const projectId = asTrimmedString(account.oauthProjectId) || storedIdentity?.projectId; return { - ...(stored || {}), + ...(storedRuntime || {}), provider, - accountId: structuredAccountKey || stored?.accountId || accountKey, + accountId: structuredAccountKey || storedIdentity?.accountId || accountKey, accountKey, projectId, }; } +export function buildOauthIdentityBackfillPatch( + account?: OauthIdentityCarrier | null, +): Partial> | null { + if (!account) return null; + const legacyIdentity = parseStoredOauthIdentity(account.extraConfig); + if (!legacyIdentity?.provider) return null; + + const patch: Partial> = {}; + if (!asTrimmedString(account.oauthProvider)) { + patch.oauthProvider = legacyIdentity.provider; + } + if (!asTrimmedString(account.oauthAccountKey) && (legacyIdentity.accountKey || legacyIdentity.accountId)) { + patch.oauthAccountKey = legacyIdentity.accountKey || legacyIdentity.accountId; + } + if (!asTrimmedString(account.oauthProjectId) && legacyIdentity.projectId) { + patch.oauthProjectId = legacyIdentity.projectId; + } + + return Object.keys(patch).length > 0 ? patch : null; +} + export function buildOauthInfo( extraConfig?: string | null, patch: Partial = {}, diff --git a/src/server/services/oauth/oauthIdentityBackfill.ts b/src/server/services/oauth/oauthIdentityBackfill.ts new file mode 100644 index 0000000..9f5114e --- /dev/null +++ b/src/server/services/oauth/oauthIdentityBackfill.ts @@ -0,0 +1,43 @@ +import { eq, isNotNull, isNull, or } from 'drizzle-orm'; +import { db, schema } from '../../db/index.js'; +import { buildOauthIdentityBackfillPatch } from './oauthAccount.js'; + +let inFlightOauthIdentityBackfill: Promise | null = null; + +async function runOauthIdentityBackfill(): Promise { + const rows = await db.select().from(schema.accounts) + .where(or( + isNotNull(schema.accounts.extraConfig), + isNull(schema.accounts.oauthProvider), + )) + .all(); + + let updated = 0; + for (const row of rows) { + const patch = buildOauthIdentityBackfillPatch(row); + if (!patch) continue; + await db.update(schema.accounts).set({ + ...patch, + updatedAt: new Date().toISOString(), + }).where(eq(schema.accounts.id, row.id)).run(); + updated += 1; + } + + return updated; +} + +export async function ensureOauthIdentityBackfill(): Promise { + if (inFlightOauthIdentityBackfill) { + return inFlightOauthIdentityBackfill; + } + + inFlightOauthIdentityBackfill = (async () => { + try { + return await runOauthIdentityBackfill(); + } finally { + inFlightOauthIdentityBackfill = null; + } + })(); + + return inFlightOauthIdentityBackfill; +} diff --git a/src/server/services/oauth/service.ts b/src/server/services/oauth/service.ts index daf22b4..0294c01 100644 --- a/src/server/services/oauth/service.ts +++ b/src/server/services/oauth/service.ts @@ -1,7 +1,8 @@ import { and, desc, eq, inArray, isNull, or, sql } from 'drizzle-orm'; import { db, schema } from '../../db/index.js'; import { mergeAccountExtraConfig } from '../accountExtraConfig.js'; -import { refreshModelsForAccount, rebuildTokenRoutesFromAvailability } from '../modelService.js'; +import { refreshModelsForAccount } from '../modelService.js'; +import * as routeRefreshWorkflow from '../routeRefreshWorkflow.js'; import { createOauthSession, getOauthSession, @@ -22,6 +23,7 @@ import { getOauthInfoFromAccount, } from './oauthAccount.js'; import { buildCodexOauthInfo } from './codexAccount.js'; +import { ensureOauthIdentityBackfill } from './oauthIdentityBackfill.js'; import { buildQuotaSnapshotFromOauthInfo, refreshOauthQuotaSnapshot } from './quota.js'; type OAuthProviderMetadata = ReturnType[number]; @@ -404,7 +406,7 @@ export async function handleOauthCallback(input: { updatedAt: previousAccount.updatedAt, }).where(eq(schema.accounts.id, previousAccount.id)).run(); } - await rebuildTokenRoutesFromAvailability(); + await routeRefreshWorkflow.rebuildRoutesOnly(); const errorMessage = refreshResult.errorMessage || `${input.provider} model discovery failed`; markOauthSessionError(input.state, errorMessage); throw new Error(errorMessage); @@ -417,7 +419,7 @@ export async function handleOauthCallback(input: { }).where(eq(schema.accounts.id, account.id)).run(); } - await rebuildTokenRoutesFromAvailability(); + await routeRefreshWorkflow.rebuildRoutesOnly(); markOauthSessionSuccess(input.state, { accountId: account.id, siteId: site.id, @@ -464,6 +466,7 @@ export async function listOauthConnections(options: { limit?: number; offset?: number; } = {}) { + await ensureOauthIdentityBackfill(); const limit = Math.max(1, Math.min(200, Math.trunc(options.limit ?? 50))); const offset = Math.max(0, Math.trunc(options.offset ?? 0)); @@ -565,7 +568,7 @@ export async function deleteOauthConnection(accountId: number) { throw new Error('account is not managed by oauth'); } await db.delete(schema.accounts).where(eq(schema.accounts.id, accountId)).run(); - await rebuildTokenRoutesFromAvailability(); + await routeRefreshWorkflow.rebuildRoutesOnly(); return { success: true }; } diff --git a/src/server/services/routeRefreshWorkflow.ts b/src/server/services/routeRefreshWorkflow.ts new file mode 100644 index 0000000..32c9bd6 --- /dev/null +++ b/src/server/services/routeRefreshWorkflow.ts @@ -0,0 +1,43 @@ +import { startBackgroundTask } from './backgroundTaskService.js'; +import { + rebuildTokenRoutesFromAvailability, + refreshModelsAndRebuildRoutes as refreshModelsAndRebuildRoutesViaModelService, +} from './modelService.js'; + +export async function rebuildRoutesOnly() { + return rebuildTokenRoutesFromAvailability(); +} + +export async function rebuildRoutesBestEffort() { + try { + await rebuildRoutesOnly(); + return true; + } catch { + return false; + } +} + +export async function refreshModelsAndRebuildRoutes() { + return refreshModelsAndRebuildRoutesViaModelService(); +} + +export function queueRefreshModelsAndRebuildRoutesTask(input: { + type: string; + title: string; + dedupeKey?: string; + notifyOnFailure?: boolean; + successMessage: (currentTask: { result?: unknown }) => string; + failureMessage: (currentTask: { error?: string | null }) => string; +}) { + return startBackgroundTask( + { + type: input.type, + title: input.title, + dedupeKey: input.dedupeKey || 'refresh-models-and-rebuild-routes', + notifyOnFailure: input.notifyOnFailure ?? true, + successMessage: input.successMessage, + failureMessage: input.failureMessage, + }, + async () => refreshModelsAndRebuildRoutes(), + ); +}