forked from boos/BoosAPI
refactor: consolidate single-source routing and protocol helpers
This commit is contained in:
@@ -4,7 +4,6 @@ import { db, schema } from '../../db/index.js';
|
||||
import {
|
||||
ACCOUNT_TOKEN_VALUE_STATUS_MASKED_PENDING,
|
||||
ACCOUNT_TOKEN_VALUE_STATUS_READY,
|
||||
ensureDefaultTokenForAccount,
|
||||
isMaskedPendingAccountToken,
|
||||
isMaskedTokenValue,
|
||||
isUsableAccountToken,
|
||||
@@ -14,17 +13,17 @@ import {
|
||||
repairDefaultToken,
|
||||
resolveAccountTokenValueStatus,
|
||||
setDefaultToken,
|
||||
syncTokensFromUpstream,
|
||||
} from '../../services/accountTokenService.js';
|
||||
import { getAdapter } from '../../services/platforms/index.js';
|
||||
import { getCredentialModeFromExtraConfig, getProxyUrlFromExtraConfig, resolvePlatformUserId } from '../../services/accountExtraConfig.js';
|
||||
import { startBackgroundTask } from '../../services/backgroundTaskService.js';
|
||||
import { withAccountProxyOverride } from '../../services/siteProxy.js';
|
||||
import { type ModelRefreshResult } from '../../services/modelService.js';
|
||||
import {
|
||||
rebuildTokenRoutesFromAvailability,
|
||||
refreshModelsForAccount,
|
||||
type ModelRefreshResult,
|
||||
} from '../../services/modelService.js';
|
||||
type CoverageBatchRebuildResult,
|
||||
convergeAccountMutation,
|
||||
refreshAccountCoverageBatch,
|
||||
} from '../../services/accountMutationWorkflow.js';
|
||||
|
||||
type AccountWithSiteRow = {
|
||||
accounts: typeof schema.accounts.$inferSelect;
|
||||
@@ -65,9 +64,7 @@ type CoverageRefreshFailureItem = {
|
||||
};
|
||||
|
||||
type CoverageRefreshItem = ModelRefreshResult | CoverageRefreshFailureItem;
|
||||
type CoverageRefreshRebuildResult =
|
||||
| { success: true; result: Awaited<ReturnType<typeof rebuildTokenRoutesFromAvailability>> }
|
||||
| { success: false; error: string };
|
||||
type CoverageRefreshRebuildResult = CoverageBatchRebuildResult;
|
||||
|
||||
const TOKEN_SYNC_TIMEOUT_MS = 15_000;
|
||||
const SYNC_ALL_BATCH_SIZE = 3;
|
||||
@@ -231,7 +228,11 @@ async function executeAccountTokenSync(row: AccountWithSiteRow): Promise<SyncExe
|
||||
|
||||
if (!row.accounts.accessToken) {
|
||||
if (row.accounts.apiToken) {
|
||||
ensureDefaultTokenForAccount(accountId, row.accounts.apiToken, { name: 'default', source: 'legacy' });
|
||||
await convergeAccountMutation({
|
||||
accountId,
|
||||
preferredApiToken: row.accounts.apiToken,
|
||||
defaultTokenSource: 'legacy',
|
||||
});
|
||||
}
|
||||
return {
|
||||
...base,
|
||||
@@ -296,7 +297,11 @@ async function executeAccountTokenSync(row: AccountWithSiteRow): Promise<SyncExe
|
||||
};
|
||||
}
|
||||
|
||||
const synced = await syncTokensFromUpstream(accountId, tokens);
|
||||
const convergence = await convergeAccountMutation({
|
||||
accountId,
|
||||
upstreamTokens: tokens,
|
||||
});
|
||||
const synced = convergence.tokenSync!;
|
||||
if ((synced.maskedPending || 0) > 0) {
|
||||
return {
|
||||
...base,
|
||||
@@ -390,49 +395,26 @@ async function executeSyncAllAccountTokens() {
|
||||
}
|
||||
|
||||
async function refreshCoverageForAccounts(accountIds: number[]) {
|
||||
const uniqueAccountIds = Array.from(new Set(
|
||||
accountIds.filter((id) => Number.isFinite(id) && id > 0),
|
||||
));
|
||||
const result = await refreshAccountCoverageBatch({
|
||||
accountIds,
|
||||
batchSize: SYNC_ALL_BATCH_SIZE,
|
||||
mapFailure: buildCoverageRefreshFailureItem,
|
||||
});
|
||||
|
||||
if (uniqueAccountIds.length === 0) {
|
||||
return { refresh: [], rebuild: null };
|
||||
result.refresh.forEach((item) => {
|
||||
if ((item as CoverageRefreshFailureItem).reason === 'coverage_refresh_failed') {
|
||||
const failed = item as CoverageRefreshFailureItem;
|
||||
console.warn(`[account-tokens] coverage refresh failed for account ${failed.accountId}: ${failed.errorMessage}`);
|
||||
}
|
||||
});
|
||||
if (result.rebuild && !result.rebuild.success) {
|
||||
console.warn(`[account-tokens] token route rebuild failed after coverage refresh: ${result.rebuild.error}`);
|
||||
}
|
||||
|
||||
const refresh: CoverageRefreshItem[] = [];
|
||||
for (let offset = 0; offset < uniqueAccountIds.length; offset += SYNC_ALL_BATCH_SIZE) {
|
||||
const batch = uniqueAccountIds.slice(offset, offset + SYNC_ALL_BATCH_SIZE);
|
||||
const settled = await Promise.allSettled(
|
||||
batch.map(async (accountId) => refreshModelsForAccount(accountId)),
|
||||
);
|
||||
settled.forEach((result, index) => {
|
||||
if (result.status === 'fulfilled') {
|
||||
refresh.push(result.value);
|
||||
return;
|
||||
}
|
||||
|
||||
const accountId = batch[index] || 0;
|
||||
const errorMessage = result.reason instanceof Error ? result.reason.message : String(result.reason || 'coverage refresh failed');
|
||||
refresh.push(buildCoverageRefreshFailureItem(accountId, errorMessage));
|
||||
console.warn(`[account-tokens] coverage refresh failed for account ${accountId}: ${errorMessage}`);
|
||||
});
|
||||
}
|
||||
|
||||
let rebuild: CoverageRefreshRebuildResult | null = null;
|
||||
try {
|
||||
rebuild = {
|
||||
success: true,
|
||||
result: await rebuildTokenRoutesFromAvailability(),
|
||||
};
|
||||
} catch (error) {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error || 'route rebuild failed');
|
||||
rebuild = {
|
||||
success: false,
|
||||
error: errorMessage,
|
||||
};
|
||||
console.warn(`[account-tokens] token route rebuild failed after coverage refresh: ${errorMessage}`);
|
||||
}
|
||||
|
||||
return { refresh, rebuild };
|
||||
return {
|
||||
refresh: result.refresh as CoverageRefreshItem[],
|
||||
rebuild: result.rebuild as CoverageRefreshRebuildResult | null,
|
||||
};
|
||||
}
|
||||
|
||||
function buildCoverageRefreshFailureItem(
|
||||
|
||||
@@ -3,8 +3,8 @@ import { db, schema, runtimeDbDialect } from '../../db/index.js';
|
||||
import { and, eq, gte, lt, sql } from 'drizzle-orm';
|
||||
import { refreshBalance } from '../../services/balanceService.js';
|
||||
import { getAdapter } from '../../services/platforms/index.js';
|
||||
import { refreshModelsForAccount, rebuildTokenRoutesFromAvailability } from '../../services/modelService.js';
|
||||
import { ensureDefaultTokenForAccount, syncTokensFromUpstream } from '../../services/accountTokenService.js';
|
||||
import { rebuildTokenRoutesFromAvailability } from '../../services/modelService.js';
|
||||
import { convergeAccountMutation } from '../../services/accountMutationWorkflow.js';
|
||||
import {
|
||||
getCredentialModeFromExtraConfig,
|
||||
getProxyUrlFromExtraConfig,
|
||||
@@ -153,37 +153,29 @@ async function initializeAccountInBackground({
|
||||
rebuiltRoutes: false,
|
||||
};
|
||||
|
||||
if (tokenType === 'session' && apiToken) {
|
||||
try {
|
||||
await ensureDefaultTokenForAccount(accountId, apiToken, { name: 'default', source: 'manual' });
|
||||
} catch {}
|
||||
}
|
||||
|
||||
let fetchedUpstreamTokens: Array<{ name?: string | null; key?: string | null; enabled?: boolean | null; tokenGroup?: string | null }> = [];
|
||||
if (tokenType === 'session' && accessToken) {
|
||||
try {
|
||||
const syncedTokens = await adapter.getApiTokens(site.url, accessToken, platformUserId);
|
||||
summary.syncedTokenCount = Array.isArray(syncedTokens) ? syncedTokens.length : 0;
|
||||
if (summary.syncedTokenCount > 0) {
|
||||
await syncTokensFromUpstream(accountId, syncedTokens);
|
||||
}
|
||||
fetchedUpstreamTokens = Array.isArray(syncedTokens) ? syncedTokens : [];
|
||||
} catch {}
|
||||
}
|
||||
|
||||
if (tokenType === 'session') {
|
||||
try {
|
||||
await refreshBalance(accountId);
|
||||
summary.refreshedBalance = true;
|
||||
} catch {}
|
||||
}
|
||||
|
||||
if (skipModelFetch !== true) {
|
||||
try {
|
||||
await refreshModelsForAccount(accountId);
|
||||
summary.refreshedModels = true;
|
||||
await rebuildTokenRoutesFromAvailability();
|
||||
summary.rebuiltRoutes = true;
|
||||
} catch {}
|
||||
}
|
||||
const convergence = await convergeAccountMutation({
|
||||
accountId,
|
||||
preferredApiToken: tokenType === 'session' ? apiToken : null,
|
||||
defaultTokenSource: 'manual',
|
||||
ensurePreferredTokenBeforeSync: tokenType === 'session',
|
||||
upstreamTokens: fetchedUpstreamTokens,
|
||||
refreshBalance: tokenType === 'session',
|
||||
refreshModels: skipModelFetch !== true,
|
||||
rebuildRoutes: skipModelFetch !== true,
|
||||
continueOnError: true,
|
||||
});
|
||||
summary.refreshedBalance = convergence.refreshedBalance;
|
||||
summary.refreshedModels = convergence.refreshedModels;
|
||||
summary.rebuiltRoutes = convergence.rebuiltRoutes;
|
||||
|
||||
return summary;
|
||||
}
|
||||
@@ -612,22 +604,16 @@ export async function accountsRoutes(app: FastifyInstance) {
|
||||
return { success: false, message: 'account create failed' };
|
||||
}
|
||||
|
||||
if (apiTokens.length > 0) {
|
||||
try {
|
||||
await syncTokensFromUpstream(result.id, apiTokens);
|
||||
} catch { }
|
||||
} else if (preferredApiToken) {
|
||||
try {
|
||||
await ensureDefaultTokenForAccount(result.id, preferredApiToken, { name: 'default', source: 'sync' });
|
||||
} catch { }
|
||||
}
|
||||
|
||||
// Auto-refresh balance
|
||||
try { await refreshBalance(result.id); } catch { }
|
||||
try {
|
||||
await refreshModelsForAccount(result.id);
|
||||
await rebuildTokenRoutesFromAvailability();
|
||||
} catch { }
|
||||
await convergeAccountMutation({
|
||||
accountId: result.id,
|
||||
preferredApiToken,
|
||||
defaultTokenSource: 'sync',
|
||||
upstreamTokens: apiTokens,
|
||||
refreshBalance: true,
|
||||
refreshModels: true,
|
||||
rebuildRoutes: true,
|
||||
continueOnError: true,
|
||||
});
|
||||
|
||||
const account = await db.select().from(schema.accounts).where(eq(schema.accounts.id, result.id)).get();
|
||||
return {
|
||||
@@ -1043,19 +1029,15 @@ export async function accountsRoutes(app: FastifyInstance) {
|
||||
|
||||
await db.update(schema.accounts).set(updates).where(eq(schema.accounts.id, accountId)).run();
|
||||
|
||||
if (nextApiToken) {
|
||||
try {
|
||||
await ensureDefaultTokenForAccount(accountId, nextApiToken, { name: 'default', source: 'sync' });
|
||||
} catch {}
|
||||
}
|
||||
|
||||
try {
|
||||
await refreshBalance(accountId);
|
||||
} catch {}
|
||||
try {
|
||||
await refreshModelsForAccount(accountId);
|
||||
await rebuildTokenRoutesFromAvailability();
|
||||
} catch {}
|
||||
await convergeAccountMutation({
|
||||
accountId,
|
||||
preferredApiToken: nextApiToken,
|
||||
defaultTokenSource: 'sync',
|
||||
refreshBalance: true,
|
||||
refreshModels: true,
|
||||
rebuildRoutes: true,
|
||||
continueOnError: true,
|
||||
});
|
||||
|
||||
const latest = await db.select().from(schema.accounts).where(eq(schema.accounts.id, accountId)).get();
|
||||
return {
|
||||
@@ -1332,16 +1314,14 @@ export async function accountsRoutes(app: FastifyInstance) {
|
||||
? explicitNextMode
|
||||
: (hasSessionTokenValue(nextAccessToken) ? 'session' : 'apikey');
|
||||
|
||||
if (nextCredentialMode !== 'apikey' && typeof updates.apiToken === 'string' && updates.apiToken.trim()) {
|
||||
try {
|
||||
await ensureDefaultTokenForAccount(id, updates.apiToken, { name: 'default', source: 'manual' });
|
||||
} catch { }
|
||||
}
|
||||
|
||||
try {
|
||||
await refreshModelsForAccount(id);
|
||||
await rebuildTokenRoutesFromAvailability();
|
||||
} catch { }
|
||||
await convergeAccountMutation({
|
||||
accountId: id,
|
||||
preferredApiToken: nextCredentialMode !== 'apikey' ? updates.apiToken : null,
|
||||
defaultTokenSource: 'manual',
|
||||
refreshModels: true,
|
||||
rebuildRoutes: true,
|
||||
continueOnError: true,
|
||||
});
|
||||
|
||||
return await db.select().from(schema.accounts).where(eq(schema.accounts.id, id)).get();
|
||||
});
|
||||
|
||||
@@ -243,11 +243,11 @@ describe('oauth routes', { timeout: 15_000 }, () => {
|
||||
|
||||
const parsed = JSON.parse(accounts[0]?.extraConfig || '{}');
|
||||
expect(parsed.oauth).toMatchObject({
|
||||
provider: 'antigravity',
|
||||
email: 'antigravity-user@example.com',
|
||||
projectId: 'antigravity-auto-project',
|
||||
refreshToken: 'antigravity-refresh-token',
|
||||
});
|
||||
expect(parsed.oauth).not.toHaveProperty('provider');
|
||||
expect(parsed.oauth).not.toHaveProperty('projectId');
|
||||
|
||||
expect(String(fetchMock.mock.calls[2]?.[0] || '')).toContain('/v1internal:loadCodeAssist');
|
||||
expect(String(fetchMock.mock.calls[3]?.[0] || '')).toContain('/v1internal:onboardUser');
|
||||
@@ -419,14 +419,15 @@ describe('oauth routes', { timeout: 15_000 }, () => {
|
||||
expect(JSON.parse(accounts[0]?.extraConfig || '{}')).toMatchObject({
|
||||
credentialMode: 'session',
|
||||
oauth: {
|
||||
provider: 'codex',
|
||||
accountId: 'chatgpt-account-123',
|
||||
email: 'codex-user@example.com',
|
||||
planType: 'plus',
|
||||
refreshToken: 'oauth-refresh-token',
|
||||
idToken: jwt,
|
||||
},
|
||||
});
|
||||
const codexStoredOauth = JSON.parse(accounts[0]?.extraConfig || '{}').oauth;
|
||||
expect(codexStoredOauth).not.toHaveProperty('provider');
|
||||
expect(codexStoredOauth).not.toHaveProperty('accountId');
|
||||
|
||||
const models = await db.select().from(schema.modelAvailability).all();
|
||||
const modelNames = models.map((row) => row.modelName);
|
||||
@@ -932,11 +933,11 @@ describe('oauth routes', { timeout: 15_000 }, () => {
|
||||
|
||||
const parsed = JSON.parse(accounts[0]?.extraConfig || '{}');
|
||||
expect(parsed.oauth).toMatchObject({
|
||||
provider: 'gemini-cli',
|
||||
email: 'gemini-user@example.com',
|
||||
projectId: 'first-project-id',
|
||||
refreshToken: 'gemini-refresh-token',
|
||||
});
|
||||
expect(parsed.oauth).not.toHaveProperty('provider');
|
||||
expect(parsed.oauth).not.toHaveProperty('projectId');
|
||||
|
||||
expect(String(fetchMock.mock.calls[1]?.[0] || '')).toContain('cloudresourcemanager.googleapis.com/v1/projects');
|
||||
expect(String(fetchMock.mock.calls[2]?.[0] || '')).toContain('/v1internal:loadCodeAssist');
|
||||
@@ -1048,8 +1049,9 @@ describe('oauth routes', { timeout: 15_000 }, () => {
|
||||
|
||||
const parsed = JSON.parse(accounts[0]?.extraConfig || '{}');
|
||||
expect(parsed.oauth).toMatchObject({
|
||||
projectId: 'gen-lang-client-0123456789',
|
||||
refreshToken: 'gemini-refresh-token',
|
||||
});
|
||||
expect(parsed.oauth).not.toHaveProperty('projectId');
|
||||
|
||||
expect(String(fetchMock.mock.calls[2]?.[0] || '')).toContain('/v1internal:loadCodeAssist');
|
||||
expect(String(fetchMock.mock.calls[3]?.[0] || '')).toContain('/v1internal:onboardUser');
|
||||
@@ -1525,12 +1527,10 @@ describe('oauth routes', { timeout: 15_000 }, () => {
|
||||
items: expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
provider: 'codex',
|
||||
email: 'team-user@example.com',
|
||||
accountKey: 'chatgpt-team-account-a',
|
||||
}),
|
||||
expect.objectContaining({
|
||||
provider: 'codex',
|
||||
email: 'team-user@example.com',
|
||||
accountKey: 'chatgpt-team-account-b',
|
||||
}),
|
||||
]),
|
||||
|
||||
@@ -9,12 +9,13 @@ import {
|
||||
import { normalizeRouteRoutingStrategy } from '../../services/routeRoutingStrategy.js';
|
||||
import { invalidateTokenRouterCache, matchesModelPattern, tokenRouter } from '../../services/tokenRouter.js';
|
||||
import { startBackgroundTask } from '../../services/backgroundTaskService.js';
|
||||
import {
|
||||
clearRouteDecisionSnapshot,
|
||||
clearRouteDecisionSnapshots,
|
||||
parseRouteDecisionSnapshot,
|
||||
saveRouteDecisionSnapshots,
|
||||
} from '../../services/routeDecisionSnapshotStore.js';
|
||||
import {
|
||||
clearRouteDecisionSnapshot,
|
||||
clearRouteDecisionSnapshots,
|
||||
parseRouteDecisionSnapshot,
|
||||
saveRouteDecisionSnapshots,
|
||||
} from '../../services/routeDecisionSnapshotStore.js';
|
||||
import { normalizeTokenRouteMode, type RouteMode } from '../../../shared/tokenRouteContract.js';
|
||||
|
||||
function isExactModelPattern(modelPattern: string): boolean {
|
||||
const normalized = modelPattern.trim();
|
||||
@@ -23,14 +24,13 @@ function isExactModelPattern(modelPattern: string): boolean {
|
||||
return !/[\*\?]/.test(normalized);
|
||||
}
|
||||
|
||||
type RouteMode = 'pattern' | 'explicit_group';
|
||||
type RouteRow = typeof schema.tokenRoutes.$inferSelect & {
|
||||
routeMode: RouteMode;
|
||||
sourceRouteIds: number[];
|
||||
};
|
||||
|
||||
function normalizeRouteMode(routeMode: unknown): RouteMode {
|
||||
return routeMode === 'explicit_group' ? 'explicit_group' : 'pattern';
|
||||
return normalizeTokenRouteMode(routeMode);
|
||||
}
|
||||
|
||||
function isExplicitGroupRoute(route: Pick<RouteRow, 'routeMode'> | Pick<typeof schema.tokenRoutes.$inferSelect, 'routeMode'>): boolean {
|
||||
|
||||
Reference in New Issue
Block a user