[codex] background route decision refresh (#370)

* background route decision refresh * address route decision refresh review * fix route decision refresh typing
2026-04-01 18:24:14 +08:00
parent 6e21f911ea
commit 37a86d1380
12 changed files with 647 additions and 90 deletions
@@ -0,0 +1,199 @@
+import Fastify, { type FastifyInstance } from 'fastify';
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
+import { mkdtempSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+
+type DbModule = typeof import('../../db/index.js');
+type BackgroundTaskModule = typeof import('../../services/backgroundTaskService.js');
+type TokenRouterModule = typeof import('../../services/tokenRouter.js');
+
+describe('POST /api/routes/decision/refresh', () => {
+  let app: FastifyInstance;
+  let previousDataDir: string | undefined;
+  let db: DbModule['db'];
+  let schema: DbModule['schema'];
+  let invalidateTokenRouterCache: TokenRouterModule['invalidateTokenRouterCache'];
+  let tokenRouter: TokenRouterModule['tokenRouter'];
+  let getBackgroundTask: BackgroundTaskModule['getBackgroundTask'];
+  let resetBackgroundTasks: BackgroundTaskModule['__resetBackgroundTasksForTests'];
+  let dataDir = '';
+
+  beforeAll(async () => {
+    previousDataDir = process.env.DATA_DIR;
+    vi.resetModules();
+    dataDir = mkdtempSync(join(tmpdir(), 'metapi-route-decision-refresh-task-'));
+    process.env.DATA_DIR = dataDir;
+
+    await import('../../db/migrate.js');
+    const dbModule = await import('../../db/index.js');
+    const routesModule = await import('./tokens.js');
+    const tokenRouterModule = await import('../../services/tokenRouter.js');
+    const backgroundTaskModule = await import('../../services/backgroundTaskService.js');
+    db = dbModule.db;
+    schema = dbModule.schema;
+    invalidateTokenRouterCache = tokenRouterModule.invalidateTokenRouterCache;
+    tokenRouter = tokenRouterModule.tokenRouter;
+    getBackgroundTask = backgroundTaskModule.getBackgroundTask;
+    resetBackgroundTasks = backgroundTaskModule.__resetBackgroundTasksForTests;
+    vi.spyOn(tokenRouter, 'refreshPricingReferenceCosts').mockResolvedValue(undefined);
+    vi.spyOn(tokenRouter, 'refreshRouteWidePricingReferenceCosts').mockResolvedValue(undefined);
+
+    app = Fastify();
+    await app.register(routesModule.tokensRoutes);
+  });
+
+  beforeEach(async () => {
+    resetBackgroundTasks();
+    await db.delete(schema.routeChannels).run();
+    await db.delete(schema.tokenRoutes).run();
+    await db.delete(schema.accountTokens).run();
+    await db.delete(schema.accounts).run();
+    await db.delete(schema.sites).run();
+    await db.delete(schema.settings).run();
+    await db.delete(schema.events).run();
+    invalidateTokenRouterCache();
+  });
+
+  afterAll(async () => {
+    await app.close();
+    invalidateTokenRouterCache();
+    vi.restoreAllMocks();
+    if (dataDir) {
+      try {
+        rmSync(dataDir, { recursive: true, force: true });
+      } catch {}
+    }
+    if (previousDataDir === undefined) {
+      delete process.env.DATA_DIR;
+    } else {
+      process.env.DATA_DIR = previousDataDir;
+    }
+    vi.resetModules();
+  });
+
+  it('queues a background refresh task and persists refreshed snapshots for exact and wildcard routes', async () => {
+    const site = await db.insert(schema.sites).values({
+      name: 'refresh-task-site',
+      url: 'https://refresh-task-site.example.com',
+      platform: 'new-api',
+      status: 'active',
+    }).returning().get();
+
+    const account = await db.insert(schema.accounts).values({
+      siteId: site.id,
+      username: 'refresh-task-user',
+      accessToken: 'refresh-task-access',
+      apiToken: 'refresh-task-api',
+      status: 'active',
+    }).returning().get();
+
+    const exactRoute = await db.insert(schema.tokenRoutes).values({
+      modelPattern: 'gpt-4o-mini',
+      enabled: true,
+    }).returning().get();
+
+    const duplicateExactRoute = await db.insert(schema.tokenRoutes).values({
+      modelPattern: 'gpt-4o-mini',
+      enabled: true,
+    }).returning().get();
+
+    const wildcardRoute = await db.insert(schema.tokenRoutes).values({
+      modelPattern: 're:^claude-(opus|sonnet)-4-6$',
+      enabled: true,
+    }).returning().get();
+
+    const disabledRoute = await db.insert(schema.tokenRoutes).values({
+      modelPattern: 'disabled-model',
+      enabled: false,
+    }).returning().get();
+
+    await db.insert(schema.routeChannels).values([
+      {
+        routeId: exactRoute.id,
+        accountId: account.id,
+        tokenId: null,
+        priority: 0,
+        weight: 10,
+        enabled: true,
+      },
+      {
+        routeId: duplicateExactRoute.id,
+        accountId: account.id,
+        tokenId: null,
+        sourceModel: 'gpt-4o-mini',
+        priority: 0,
+        weight: 5,
+        enabled: true,
+      },
+      {
+        routeId: wildcardRoute.id,
+        accountId: account.id,
+        tokenId: null,
+        sourceModel: 'claude-opus-4-6',
+        priority: 0,
+        weight: 10,
+        enabled: true,
+      },
+      {
+        routeId: disabledRoute.id,
+        accountId: account.id,
+        tokenId: null,
+        priority: 0,
+        weight: 10,
+        enabled: true,
+      },
+    ]).run();
+
+    const response = await app.inject({
+      method: 'POST',
+      url: '/api/routes/decision/refresh',
+      payload: {},
+    });
+
+    expect(response.statusCode).toBe(202);
+    const body = response.json() as {
+      success: boolean;
+      queued: boolean;
+      jobId: string;
+      status: string;
+    };
+    expect(body.success).toBe(true);
+    expect(body.queued).toBe(true);
+    expect(body.jobId.length).toBeGreaterThan(10);
+    expect(body.status).toBe('pending');
+
+    for (let attempt = 0; attempt < 200; attempt += 1) {
+      const task = getBackgroundTask(body.jobId);
+      if (task?.status === 'succeeded') break;
+      await new Promise((resolve) => setTimeout(resolve, 0));
+    }
+
+    expect(getBackgroundTask(body.jobId)).toMatchObject({ status: 'succeeded' });
+
+    const routesResponse = await app.inject({
+      method: 'GET',
+      url: '/api/routes',
+    });
+
+    expect(routesResponse.statusCode).toBe(200);
+    const routes = routesResponse.json() as Array<{
+      id: number;
+      decisionSnapshot?: { matched?: boolean; routeId?: number; candidates?: Array<unknown> } | null;
+    }>;
+
+    const refreshedExactRoute = routes.find((route) => route.id === exactRoute.id);
+    const refreshedDuplicateExactRoute = routes.find((route) => route.id === duplicateExactRoute.id);
+    const refreshedWildcardRoute = routes.find((route) => route.id === wildcardRoute.id);
+    const refreshedDisabledRoute = routes.find((route) => route.id === disabledRoute.id);
+
+    expect(refreshedExactRoute?.decisionSnapshot?.matched).toBe(true);
+    expect(refreshedExactRoute?.decisionSnapshot?.routeId).toBe(exactRoute.id);
+    expect(Array.isArray(refreshedExactRoute?.decisionSnapshot?.candidates)).toBe(true);
+    expect(refreshedDuplicateExactRoute?.decisionSnapshot?.matched).toBe(true);
+    expect(refreshedDuplicateExactRoute?.decisionSnapshot?.routeId).toBe(duplicateExactRoute.id);
+    expect(refreshedWildcardRoute?.decisionSnapshot?.matched).toBe(true);
+    expect(refreshedWildcardRoute?.decisionSnapshot?.routeId).toBe(wildcardRoute.id);
+    expect(refreshedDisabledRoute?.decisionSnapshot).toBeNull();
+  });
+});
@@ -13,7 +13,7 @@ import {
  type RouteRoutingStrategy,
 } from '../../services/routeRoutingStrategy.js';
 import { invalidateTokenRouterCache, matchesModelPattern, tokenRouter } from '../../services/tokenRouter.js';
-import { startBackgroundTask } from '../../services/backgroundTaskService.js';
+import { appendBackgroundTaskLog, startBackgroundTask } from '../../services/backgroundTaskService.js';
 import {
  clearRouteDecisionSnapshot,
  clearRouteDecisionSnapshots,
@@ -21,6 +21,11 @@ import {
  saveRouteDecisionSnapshots,
 } from '../../services/routeDecisionSnapshotStore.js';
 import { clearRouteCooldown } from '../../services/routeCooldownService.js';
+import {
+  refreshAllRouteDecisionSnapshots,
+  ROUTE_DECISION_REFRESH_DEDUPE_KEY,
+  ROUTE_DECISION_REFRESH_TASK_TYPE,
+} from '../../services/routeDecisionRefreshService.js';
 import { normalizeTokenRouteMode, type RouteMode } from '../../../shared/tokenRouteContract.js';
 import {
  parseRouteChannelBatchCreatePayload,
@@ -945,11 +950,11 @@ export async function tokensRoutes(app: FastifyInstance) {
    return { success: true, decisions };
  });

-  app.post<{ Body: BatchRouteWideDecisionRouteIds }>('/api/routes/decision/route-wide/batch', async (request, reply) => {
-    const parsed = parseBatchRouteWideDecisionRouteIds(request.body);
-    if (!parsed.ok) {
-      return reply.code(400).send({ success: false, message: parsed.message });
-    }
+  app.post<{ Body: BatchRouteWideDecisionRouteIds }>('/api/routes/decision/route-wide/batch', async (request, reply) => {
+    const parsed = parseBatchRouteWideDecisionRouteIds(request.body);
+    if (!parsed.ok) {
+      return reply.code(400).send({ success: false, message: parsed.message });
+    }

    const decisions: Record<string, Awaited<ReturnType<typeof tokenRouter.explainSelectionRouteWide>>> = {};
    const refreshedKeys = parsed.refreshPricingCatalog ? new Set<string>() : undefined;
@@ -966,10 +971,49 @@ export async function tokensRoutes(app: FastifyInstance) {
        snapshot: decisions[String(routeId)] ?? null,
      })));
    }
-
-    return { success: true, decisions };
-  });
-
+
+    return { success: true, decisions };
+  });
+
+  app.post('/api/routes/decision/refresh', async (_request, reply) => {
+    let taskId = '';
+    const { task, reused } = startBackgroundTask(
+      {
+        type: ROUTE_DECISION_REFRESH_TASK_TYPE,
+        title: '刷新路由选中概率',
+        dedupeKey: ROUTE_DECISION_REFRESH_DEDUPE_KEY,
+        successMessage: (currentTask) => {
+          const result = currentTask.result as { exactModelCount?: number; wildcardRouteCount?: number } | null;
+          const exactModelCount = result?.exactModelCount ?? 0;
+          const wildcardRouteCount = result?.wildcardRouteCount ?? 0;
+          return `路由选中概率刷新完成：精确模型 ${exactModelCount}，通配符路由 ${wildcardRouteCount}`;
+        },
+        failureMessage: (currentTask) => `路由选中概率刷新失败：${currentTask.error || 'unknown error'}`,
+      },
+      async () => {
+        await Promise.resolve();
+        return await refreshAllRouteDecisionSnapshots({
+          refreshPricingCatalog: true,
+          onProgress: (message) => {
+            appendBackgroundTaskLog(taskId, message);
+          },
+        });
+      },
+    );
+    taskId = task.id;
+
+    return reply.code(202).send({
+      success: true,
+      queued: true,
+      reused,
+      jobId: task.id,
+      status: task.status,
+      message: reused
+        ? '路由选中概率刷新任务执行中，可稍后返回查看'
+        : '已开始后台刷新路由选中概率，可稍后返回查看',
+    });
+  });
+
  // Create a route
  app.post<{ Body: unknown }>('/api/routes', async (request, reply) => {
    const parsedBody = parseTokenRouteCreatePayload(request.body);
@@ -0,0 +1,103 @@
+import { eq } from 'drizzle-orm';
+import { db, schema } from '../db/index.js';
+import { saveRouteDecisionSnapshots } from './routeDecisionSnapshotStore.js';
+import { matchesModelPattern, tokenRouter } from './tokenRouter.js';
+import { ROUTE_DECISION_REFRESH_TASK_TYPE } from '../../shared/tokenRouteContract.js';
+
+export { ROUTE_DECISION_REFRESH_TASK_TYPE };
+export const ROUTE_DECISION_REFRESH_DEDUPE_KEY = 'refresh-route-decision-snapshots';
+
+function isExactModelPattern(modelPattern: string): boolean {
+  const normalized = modelPattern.trim();
+  if (!normalized) return false;
+  if (normalized.toLowerCase().startsWith('re:')) return false;
+  return !/[\*\?]/.test(normalized);
+}
+
+function normalizeModels(models: string[]): string[] {
+  return Array.from(new Set(
+    models
+      .map((model) => String(model || '').trim())
+      .filter(Boolean),
+  ));
+}
+
+function normalizeRouteIds(routeIds: number[]): number[] {
+  return Array.from(new Set(
+    routeIds
+      .map((routeId) => Math.trunc(routeId))
+      .filter((routeId) => routeId > 0),
+  ));
+}
+
+type RefreshOptions = {
+  refreshPricingCatalog?: boolean;
+  onProgress?: (message: string) => void;
+};
+
+export async function refreshAllRouteDecisionSnapshots(options: RefreshOptions = {}): Promise<{
+  exactModelCount: number;
+  wildcardRouteCount: number;
+}> {
+  const routes = await db.select({
+    id: schema.tokenRoutes.id,
+    modelPattern: schema.tokenRoutes.modelPattern,
+  }).from(schema.tokenRoutes)
+    .where(eq(schema.tokenRoutes.enabled, true))
+    .all();
+
+  const exactModels = normalizeModels(
+    routes
+      .filter((route) => isExactModelPattern(route.modelPattern))
+      .map((route) => route.modelPattern),
+  );
+  const wildcardRouteIds = normalizeRouteIds(
+    routes
+      .filter((route) => !isExactModelPattern(route.modelPattern))
+      .map((route) => route.id),
+  );
+  const refreshedKeys = options.refreshPricingCatalog ? new Set<string>() : undefined;
+
+  options.onProgress?.(`开始刷新路由概率：精确模型 ${exactModels.length}，通配符路由 ${wildcardRouteIds.length}`);
+
+  for (const [index, model] of exactModels.entries()) {
+    options.onProgress?.(`刷新精确模型概率 ${index + 1}/${exactModels.length}：${model}`);
+    const matchingRoutes = routes.filter((route) => (
+      isExactModelPattern(route.modelPattern) && matchesModelPattern(model, route.modelPattern)
+    ));
+    const snapshotWrites: Array<{ routeId: number; snapshot: unknown }> = [];
+    for (const route of matchingRoutes) {
+      if (options.refreshPricingCatalog) {
+        await tokenRouter.refreshPricingReferenceCostsForRoute(route.id, model, { refreshedKeys });
+      }
+      const decision = await tokenRouter.explainSelectionForRoute(route.id, model);
+      snapshotWrites.push({
+        routeId: route.id,
+        snapshot: decision,
+      });
+    }
+    if (snapshotWrites.length > 0) {
+      await saveRouteDecisionSnapshots(snapshotWrites);
+    }
+  }
+
+  for (const [index, routeId] of wildcardRouteIds.entries()) {
+    options.onProgress?.(`刷新通配符路由概率 ${index + 1}/${wildcardRouteIds.length}：#${routeId}`);
+    if (options.refreshPricingCatalog) {
+      await tokenRouter.refreshRouteWidePricingReferenceCosts(routeId, { refreshedKeys });
+    }
+
+    const decision = await tokenRouter.explainSelectionRouteWide(routeId);
+    await saveRouteDecisionSnapshots([
+      {
+        routeId,
+        snapshot: decision,
+      },
+    ]);
+  }
+
+  return {
+    exactModelCount: exactModels.length,
+    wildcardRouteCount: wildcardRouteIds.length,
+  };
+}