feat(engine): 实现同步引擎与 engine_status 快照

apps/engine 独立进程（syncRound + 防重叠调度）；上游 log/token；QPU 采样；
server 读 engine_status；时刻出参 toUnixSec；中文日志；zod-openapi 扩展顺序修复。

apps/engine/package.json

@@ -0,0 +1,13 @@
+{
+  "name": "@new-api-key-usage/engine",
+  "private": true,
+  "version": "0.0.0",
+  "type": "module",
+  "scripts": {
+    "dev": "bun run src/index.ts"
+  },
+  "dependencies": {
+    "@new-api-key-usage/core": "workspace:*",
+    "@new-api-key-usage/db": "workspace:*"
+  }
+}

apps/engine/src/env.ts

@@ -0,0 +1,43 @@
+export interface EngineEnv {
+  DATABASE_URL: string;
+  SK_KEY_ENCRYPTION_SECRET: string;
+  NEWAPI_BASE: string;
+  SYNC_INTERVAL_MS: number;
+  LOG_LEVEL: "debug" | "info" | "warning" | "error" | "fatal";
+}
+
+function parseLogLevel(
+  value: string | undefined,
+): "debug" | "info" | "warning" | "error" | "fatal" {
+  const v = (value ?? "info").trim().toLowerCase();
+  if (
+    v === "debug" ||
+    v === "info" ||
+    v === "warning" ||
+    v === "error" ||
+    v === "fatal"
+  ) {
+    return v;
+  }
+  throw new Error("LOG_LEVEL must be debug|info|warning|error|fatal");
+}
+
+export function loadEnv(source: Record<string, string | undefined>): EngineEnv {
+  const databaseUrl = source.DATABASE_URL;
+  const secret = source.SK_KEY_ENCRYPTION_SECRET;
+  if (!databaseUrl) throw new Error("DATABASE_URL is required");
+  if (!secret) throw new Error("SK_KEY_ENCRYPTION_SECRET is required");
+
+  const interval = Number(source.SYNC_INTERVAL_MS ?? 300_000);
+  if (!Number.isInteger(interval) || interval < 1000) {
+    throw new Error("SYNC_INTERVAL_MS must be an integer >= 1000");
+  }
+
+  return {
+    DATABASE_URL: databaseUrl,
+    SK_KEY_ENCRYPTION_SECRET: secret,
+    NEWAPI_BASE: source.NEWAPI_BASE ?? "http://127.0.0.1:8080",
+    SYNC_INTERVAL_MS: interval,
+    LOG_LEVEL: parseLogLevel(source.LOG_LEVEL),
+  };
+}

apps/engine/src/index.ts

@@ -0,0 +1,29 @@
+import { appLogger, setupLogging } from "@new-api-key-usage/core";
+import { createDb, createRepos } from "@new-api-key-usage/db";
+import { loadEnv } from "./env.ts";
+import { createSyncLoop } from "./loop.ts";
+
+const env = loadEnv(process.env);
+setupLogging(env.LOG_LEVEL);
+const log = appLogger("engine");
+
+const db = createDb(env.DATABASE_URL);
+const repos = createRepos(db);
+
+const loop = createSyncLoop({
+  repos,
+  newapiBase: env.NEWAPI_BASE,
+  skKeyEncryptionSecret: env.SK_KEY_ENCRYPTION_SECRET,
+  intervalMs: env.SYNC_INTERVAL_MS,
+});
+
+loop.start();
+
+log.info`apps/engine 已就绪`;
+
+process.on("SIGINT", () => {
+  void loop.stop().then(() => process.exit(0));
+});
+process.on("SIGTERM", () => {
+  void loop.stop().then(() => process.exit(0));
+});

apps/engine/src/loop.ts

@@ -0,0 +1,75 @@
+/**
+ * 调度循环（方案 A）：一轮结束 → setTimeout 下一枪；inFlight 防重叠。
+ * 启停不得抹掉已有 last_round。
+ */
+import { appLogger } from "@new-api-key-usage/core";
+import { syncRound, type SyncRoundDeps } from "./sync-round.ts";
+
+const log = appLogger("engine");
+
+export interface LoopOptions extends SyncRoundDeps {
+  intervalMs: number;
+}
+
+export function createSyncLoop(opts: LoopOptions) {
+  let inFlight = false;
+  let stopped = false;
+  let timer: ReturnType<typeof setTimeout> | null = null;
+  let last_round: unknown = null;
+
+  async function persist(running: boolean) {
+    await opts.repos.engine.upsertSnapshot({
+      running,
+      sync_interval_ms: opts.intervalMs,
+      last_round,
+    });
+  }
+
+  async function runOnce(): Promise<void> {
+    if (stopped) return;
+    if (inFlight) {
+      log.warn("上一轮尚未结束，跳过本枪");
+      schedule();
+      return;
+    }
+    inFlight = true;
+    try {
+      last_round = await syncRound(opts);
+      await persist(!stopped);
+    } catch (e) {
+      log.error(`同步轮次异常：${e}`);
+      await persist(!stopped).catch(() => {});
+    } finally {
+      inFlight = false;
+      schedule();
+    }
+  }
+
+  function schedule() {
+    if (stopped) return;
+    timer = setTimeout(() => {
+      void runOnce();
+    }, opts.intervalMs);
+  }
+
+  return {
+    start() {
+      log.info(`同步引擎启动：周期=${opts.intervalMs}ms`);
+      void (async () => {
+        const snap = await opts.repos.engine.getSnapshot();
+        last_round = snap?.last_round ?? null;
+        await persist(true).catch(() => {});
+        void runOnce();
+      })();
+    },
+    async stop() {
+      stopped = true;
+      if (timer) clearTimeout(timer);
+      // 等在跑的一轮结束，避免 stop 与 upsert 交错
+      for (let i = 0; i < 50 && inFlight; i++) {
+        await new Promise((r) => setTimeout(r, 100));
+      }
+      await persist(false).catch(() => {});
+    },
+  };
+}

apps/engine/src/sync-round.ts

@@ -0,0 +1,174 @@
+/**
+ * 同步轮次：按《同步引擎设计》§3 编排。
+ * 只经 packages/core（上游 + 去重）与 packages/db（仓储）写库。
+ */
+import {
+  appLogger,
+  buildDedupeKey,
+  createUpstreamClient,
+  KeyCrypto,
+  parseOtherJson,
+  toUnixSec,
+} from "@new-api-key-usage/core";
+import type { Repos } from "@new-api-key-usage/db";
+
+const log = appLogger("engine");
+
+export interface SyncRoundDeps {
+  repos: Repos;
+  newapiBase: string;
+  skKeyEncryptionSecret: string;
+}
+
+export interface LastRound {
+  started_at: number;
+  keys_total: number;
+  succeeded: number;
+  failed: number;
+  failed_api_key_ids: number[];
+}
+
+function parseOther(other: string | undefined): {
+  billing_source: string | null;
+  subscription_id: number | null;
+  raw: Record<string, unknown> | null;
+} {
+  return parseOtherJson(other);
+}
+
+export async function syncRound(deps: SyncRoundDeps): Promise<LastRound> {
+  const started = Date.now();
+  const started_at = toUnixSec(new Date())!;
+  const client = createUpstreamClient({ baseUrl: deps.newapiBase });
+  const crypto = await KeyCrypto.fromEncryptionSecret(
+    deps.skKeyEncryptionSecret,
+  );
+
+  const keys = await deps.repos.keys.listActive();
+  const failed_api_key_ids: number[] = [];
+  let succeeded = 0;
+  let pulled = 0;
+  let inserted = 0;
+  let skipped = 0;
+
+  for (const key of keys) {
+    try {
+      const sk = await crypto.openKey(key.key_ciphertext);
+
+      const usage = await client.fetchTokenUsage(sk);
+      if (!usage.ok) {
+        failed_api_key_ids.push(key.id);
+        log.warn(`同步失败 key_id=${key.id} 原因=usage/${usage.kind}`);
+        continue;
+      }
+      await deps.repos.keys.update(key.id, {
+        token_name: usage.data.name ?? key.token_name,
+        total_granted: usage.data.total_granted,
+        total_used: usage.data.total_used,
+        total_available: usage.data.total_available,
+        unlimited_quota: usage.data.unlimited_quota,
+        model_limits: usage.data.model_limits,
+        model_limits_enabled: usage.data.model_limits_enabled,
+        expires_at: normalizeExpires(usage.data.expires_at),
+        usage_synced_at: new Date(),
+      });
+
+      const logs = await client.fetchTokenLogs(sk);
+      if (!logs.ok) {
+        failed_api_key_ids.push(key.id);
+        log.warn(`同步失败 key_id=${key.id} 原因=logs/${logs.kind}`);
+        continue;
+      }
+
+      const rows = [...(logs.data ?? [])].sort((a, b) => {
+        const ta = a.created_at ?? 0;
+        const tb = b.created_at ?? 0;
+        if (ta !== tb) return ta - tb;
+        return String(a.request_id ?? "").localeCompare(String(b.request_id ?? ""));
+      });
+      pulled += rows.length;
+
+      for (const row of rows) {
+        const other = parseOther(row.other);
+        const created_at = new Date(row.created_at * 1000);
+        const dedupe_key = await buildDedupeKey({
+          api_key_id: key.id,
+          request_id: row.request_id,
+          type: row.type ?? 2,
+          created_at: row.created_at,
+          model_name: row.model_name ?? "",
+          quota: row.quota ?? 0,
+          prompt_tokens: row.prompt_tokens ?? 0,
+          completion_tokens: row.completion_tokens ?? 0,
+          use_time: row.use_time ?? 0,
+          content: row.content ?? "",
+        });
+        const ok = await deps.repos.keys.insertLog({
+          api_key_id: key.id,
+          dedupe_key,
+          billing_source: other.billing_source,
+          subscription_id: other.subscription_id,
+          user_id: row.user_id ?? 0,
+          created_at,
+          type: row.type ?? 2,
+          content: row.content ?? "",
+          username: row.username ?? "",
+          token_name: row.token_name ?? "",
+          model_name: row.model_name ?? "",
+          quota: row.quota ?? 0,
+          prompt_tokens: row.prompt_tokens ?? 0,
+          completion_tokens: row.completion_tokens ?? 0,
+          use_time: row.use_time ?? 0,
+          is_stream: !!row.is_stream,
+          channel: row.channel ?? 0,
+          channel_name: row.channel_name ?? "",
+          token_id: row.token_id ?? 0,
+          group_name: row.group ?? "",
+          ip: row.ip ?? "",
+          request_id: row.request_id ?? "",
+          upstream_request_id: row.upstream_request_id ?? null,
+          other: other.raw,
+        });
+        if (ok) inserted += 1;
+        else skipped += 1;
+      }
+      await deps.repos.keys.update(key.id, { log_synced_at: new Date() });
+      succeeded += 1;
+    } catch (e) {
+      failed_api_key_ids.push(key.id);
+      log.error(`同步异常 key_id=${key.id} ${e}`);
+    }
+  }
+
+  // QPU 采样：值变化才追加
+  const qpuRes = await client.fetchQuotaPerUnit();
+  if (qpuRes.ok) {
+    const latest = await deps.repos.qpu.latest();
+    const next = qpuRes.data.quota_per_unit;
+    if (!latest || latest.qpu !== next) {
+      await deps.repos.qpu.insertChange(next, new Date());
+      log.info(`QPU 变更 ${latest?.qpu ?? "-"} -> ${next}`);
+    }
+  } else {
+    log.warn(`QPU 采样失败 ${qpuRes.kind}`);
+  }
+
+  const last_round: LastRound = {
+    started_at,
+    keys_total: keys.length,
+    succeeded,
+    failed: failed_api_key_ids.length,
+    failed_api_key_ids,
+  };
+
+  log.info(
+    `同步轮次完成 key=${keys.length} 成功=${succeeded} 失败=${failed_api_key_ids.length} 拉取=${pulled} 新写=${inserted} 去重跳过=${skipped} 耗时=${Date.now() - started}ms`,
+  );
+
+  return last_round;
+}
+
+function normalizeExpires(unixSeconds: number | undefined | null): Date | null {
+  if (!unixSeconds) return null;
+  return new Date(unixSeconds * 1000);
+}

apps/engine/tsconfig.json

@@ -0,0 +1,8 @@
+{
+  "extends": "../../tsconfig.json",
+  "compilerOptions": {
+    "types": ["bun"],
+    "allowImportingTsExtensions": true
+  },
+  "include": ["src/**/*.ts"]
+}