feat: align frontend runtime and Edge TTS

2026-07-22 15:01:25 +08:00
parent d2c278f0ea
commit 699a0bced4
43 changed files with 2000 additions and 73 deletions
@@ -0,0 +1,24 @@
+import { createServer } from "node:http";
+import { createEdgeTtsMiddleware } from "./edge-tts-service";
+
+const host = "127.0.0.1";
+const port = 8790;
+const handleEdgeTts = createEdgeTtsMiddleware({ trustProxy: true });
+
+const server = createServer((request, response) => {
+  void handleEdgeTts(request, response, () => {
+    response.statusCode = 404;
+    response.end("Not Found");
+  });
+});
+
+server.listen(port, host, () => {
+  console.log(`[Agent TTS] Edge TTS adapter listening on http://${host}:${port}`);
+});
+
+function closeServer() {
+  server.close(() => process.exit(0));
+}
+
+process.on("SIGINT", closeServer);
+process.on("SIGTERM", closeServer);
@@ -0,0 +1,102 @@
+import { createServer } from "node:http";
+import type { AddressInfo } from "node:net";
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+import {
+  createEdgeTtsMiddleware,
+  type EdgeTtsMiddlewareOptions,
+  type EdgeTtsSynthesizer
+} from "./edge-tts-service";
+
+const servers: ReturnType<typeof createServer>[] = [];
+
+afterEach(async () => {
+  await Promise.all(
+    servers.splice(0).map((server) => new Promise<void>((resolve) => server.close(() => resolve())))
+  );
+});
+
+async function startServer(options: EdgeTtsMiddlewareOptions) {
+  const middleware = createEdgeTtsMiddleware(options);
+  const server = createServer((request, response) => {
+    void middleware(request, response, () => {
+      response.statusCode = 404;
+      response.end();
+    });
+  });
+  servers.push(server);
+  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+  const { port } = server.address() as AddressInfo;
+  return `http://127.0.0.1:${port}`;
+}
+
+function postSpeech(baseUrl: string, body: unknown, headers = true) {
+  return fetch(`${baseUrl}/api/tts/edge`, {
+    method: "POST",
+    ...(headers ? { headers: { "Content-Type": "application/json" } } : {}),
+    body: typeof body === "string" ? body : JSON.stringify(body)
+  });
+}
+
+describe("Edge TTS adapter", () => {
+  it("returns synthesized MP3 audio with the server-owned voice", async () => {
+    const synthesize = vi.fn<EdgeTtsSynthesizer>(async () => Uint8Array.from([73, 68, 51]));
+    const baseUrl = await startServer({ synthesize, defaultVoice: "zh-CN-XiaoxiaoNeural" });
+
+    const response = await postSpeech(baseUrl, { text: "  朗读调度结果  " });
+
+    expect(response.status).toBe(200);
+    expect(response.headers.get("content-type")).toBe("audio/mpeg");
+    expect(new Uint8Array(await response.arrayBuffer())).toEqual(Uint8Array.from([73, 68, 51]));
+    expect(synthesize).toHaveBeenCalledWith("朗读调度结果", "zh-CN-XiaoxiaoNeural");
+  });
+
+  it("rejects invalid, unsafe and oversized requests before synthesis", async () => {
+    const synthesize = vi.fn<EdgeTtsSynthesizer>(async () => Uint8Array.from([1]));
+    const baseUrl = await startServer({ synthesize });
+
+    expect((await postSpeech(baseUrl, "not-json")).status).toBe(400);
+    expect((await postSpeech(baseUrl, { text: "测试" }, false)).status).toBe(415);
+    expect((await postSpeech(baseUrl, { text: "测试", voice: "unsafe" })).status).toBe(400);
+    expect((await postSpeech(baseUrl, { text: "排".repeat(521) })).status).toBe(413);
+    expect(synthesize).not.toHaveBeenCalled();
+  });
+
+  it("limits requests by client and releases the window", async () => {
+    let currentTime = 1_000;
+    const synthesize = vi.fn<EdgeTtsSynthesizer>(async () => Uint8Array.from([1]));
+    const baseUrl = await startServer({
+      synthesize,
+      maxRequestsPerWindow: 1,
+      rateWindowMs: 1_000,
+      now: () => currentTime
+    });
+
+    expect((await postSpeech(baseUrl, { text: "第一次" })).status).toBe(200);
+    expect((await postSpeech(baseUrl, { text: "第二次" })).status).toBe(429);
+    currentTime += 1_000;
+    expect((await postSpeech(baseUrl, { text: "新窗口" })).status).toBe(200);
+  });
+
+  it("caps concurrent synthesis work", async () => {
+    let releaseFirst: ((audio: Uint8Array) => void) | undefined;
+    const firstAudio = new Promise<Uint8Array>((resolve) => {
+      releaseFirst = resolve;
+    });
+    const synthesize = vi.fn<EdgeTtsSynthesizer>(() => firstAudio);
+    const baseUrl = await startServer({ synthesize, maxConcurrent: 1 });
+
+    const firstRequest = postSpeech(baseUrl, { text: "第一段" });
+    await vi.waitFor(() => expect(synthesize).toHaveBeenCalledTimes(1));
+    expect((await postSpeech(baseUrl, { text: "第二段" })).status).toBe(429);
+    releaseFirst?.(Uint8Array.from([1]));
+    expect((await firstRequest).status).toBe(200);
+  });
+
+  it("returns a gateway timeout without releasing the occupied slot early", async () => {
+    const synthesize = vi.fn<EdgeTtsSynthesizer>(() => new Promise(() => undefined));
+    const baseUrl = await startServer({ synthesize, timeoutMs: 5 });
+
+    expect((await postSpeech(baseUrl, { text: "超时测试" })).status).toBe(504);
+  });
+});
@@ -0,0 +1,210 @@
+import type { IncomingMessage, ServerResponse } from "node:http";
+import { Communicate } from "edge-tts-ts";
+
+const EDGE_TTS_PATH = "/api/tts/edge";
+const DEFAULT_VOICE = "zh-CN-XiaoxiaoNeural";
+const MAX_REQUEST_BYTES = 8 * 1024;
+const MAX_TEXT_LENGTH = 520;
+const MAX_AUDIO_BYTES = 5 * 1024 * 1024;
+const DEFAULT_MAX_CONCURRENT = 8;
+const DEFAULT_RATE_LIMIT = 60;
+const DEFAULT_RATE_WINDOW_MS = 60_000;
+const DEFAULT_TIMEOUT_MS = 30_000;
+
+interface EdgeTtsPayload {
+  text?: unknown;
+  voice?: unknown;
+}
+
+interface RateBucket {
+  count: number;
+  startedAt: number;
+}
+
+export type EdgeTtsSynthesizer = (text: string, voice: string) => Promise<Uint8Array>;
+
+export type EdgeTtsMiddlewareOptions = {
+  defaultVoice?: string;
+  maxConcurrent?: number;
+  maxRequestsPerWindow?: number;
+  now?: () => number;
+  rateWindowMs?: number;
+  synthesize?: EdgeTtsSynthesizer;
+  timeoutMs?: number;
+  trustProxy?: boolean;
+};
+
+function sendJson(response: ServerResponse, status: number, message: string) {
+  response.statusCode = status;
+  response.setHeader("Content-Type", "application/json; charset=utf-8");
+  response.setHeader("Cache-Control", "no-store");
+  response.end(JSON.stringify({ error: message }));
+}
+
+async function readJson(request: IncomingMessage): Promise<EdgeTtsPayload> {
+  const chunks: Buffer[] = [];
+  let byteLength = 0;
+
+  for await (const chunk of request) {
+    const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
+    byteLength += buffer.byteLength;
+    if (byteLength > MAX_REQUEST_BYTES) {
+      throw new RangeError("请求内容过大");
+    }
+    chunks.push(buffer);
+  }
+
+  const payload = JSON.parse(Buffer.concat(chunks).toString("utf8")) as unknown;
+  if (!payload || typeof payload !== "object" || Array.isArray(payload)) {
+    throw new SyntaxError("请求内容必须是 JSON 对象");
+  }
+  return payload as EdgeTtsPayload;
+}
+
+function isLoopback(address: string | undefined) {
+  return address === "127.0.0.1" || address === "::1" || address === "::ffff:127.0.0.1";
+}
+
+function getClientKey(request: IncomingMessage, trustProxy: boolean) {
+  const remoteAddress = request.socket.remoteAddress;
+  if (trustProxy && isLoopback(remoteAddress)) {
+    const forwardedFor = request.headers["x-forwarded-for"];
+    const firstAddress = Array.isArray(forwardedFor) ? forwardedFor[0] : forwardedFor?.split(",", 1)[0];
+    if (firstAddress?.trim()) return firstAddress.trim();
+  }
+  return remoteAddress || "unknown";
+}
+
+export async function synthesizeEdgeSpeech(text: string, voice: string): Promise<Uint8Array> {
+  const communicate = new Communicate(text, { voice });
+  const chunks: Uint8Array[] = [];
+  let byteLength = 0;
+
+  for await (const chunk of communicate.stream()) {
+    if (chunk.type !== "audio") continue;
+    chunks.push(chunk.data);
+    byteLength += chunk.data.byteLength;
+  }
+
+  if (!byteLength) throw new Error("语音服务返回了空音频");
+
+  const audio = new Uint8Array(byteLength);
+  let offset = 0;
+  for (const chunk of chunks) {
+    audio.set(chunk, offset);
+    offset += chunk.byteLength;
+  }
+  return audio;
+}
+
+export function createEdgeTtsMiddleware(options: EdgeTtsMiddlewareOptions = {}) {
+  const synthesize = options.synthesize ?? synthesizeEdgeSpeech;
+  const defaultVoice = options.defaultVoice ?? process.env.EDGE_TTS_VOICE ?? DEFAULT_VOICE;
+  const maxConcurrent = options.maxConcurrent ?? DEFAULT_MAX_CONCURRENT;
+  const maxRequestsPerWindow = options.maxRequestsPerWindow ?? DEFAULT_RATE_LIMIT;
+  const rateWindowMs = options.rateWindowMs ?? DEFAULT_RATE_WINDOW_MS;
+  const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS;
+  const trustProxy = options.trustProxy ?? false;
+  const now = options.now ?? Date.now;
+  const rateBuckets = new Map<string, RateBucket>();
+  let activeRequests = 0;
+
+  return async (request: IncomingMessage, response: ServerResponse, next?: () => void) => {
+    if (request.url?.split("?", 1)[0] !== EDGE_TTS_PATH) {
+      next?.();
+      return;
+    }
+    if (request.method !== "POST") {
+      response.setHeader("Allow", "POST");
+      sendJson(response, 405, "仅支持 POST 请求");
+      return;
+    }
+    if (!request.headers["content-type"]?.toLowerCase().startsWith("application/json")) {
+      sendJson(response, 415, "请求必须使用 application/json");
+      return;
+    }
+
+    const currentTime = now();
+    const clientKey = getClientKey(request, trustProxy);
+    const existingBucket = rateBuckets.get(clientKey);
+    const bucket = !existingBucket || currentTime - existingBucket.startedAt >= rateWindowMs
+      ? { count: 0, startedAt: currentTime }
+      : existingBucket;
+    bucket.count += 1;
+    rateBuckets.set(clientKey, bucket);
+    if (rateBuckets.size > 1_024) {
+      for (const [key, value] of rateBuckets) {
+        if (currentTime - value.startedAt >= rateWindowMs) rateBuckets.delete(key);
+      }
+    }
+    if (bucket.count > maxRequestsPerWindow) {
+      response.setHeader("Retry-After", String(Math.max(1, Math.ceil(rateWindowMs / 1_000))));
+      sendJson(response, 429, "语音请求过于频繁，请稍后重试");
+      return;
+    }
+
+    let payload: EdgeTtsPayload;
+    try {
+      payload = await readJson(request);
+    } catch (error) {
+      sendJson(response, error instanceof RangeError ? 413 : 400, "请求内容不是有效 JSON");
+      return;
+    }
+
+    if (Object.prototype.hasOwnProperty.call(payload, "voice")) {
+      sendJson(response, 400, "语音角色只能由服务端配置");
+      return;
+    }
+    const text = typeof payload.text === "string" ? payload.text.trim() : "";
+    if (!text) {
+      sendJson(response, 400, "缺少待朗读文本");
+      return;
+    }
+    if (text.length > MAX_TEXT_LENGTH) {
+      sendJson(response, 413, `单次文本不能超过 ${MAX_TEXT_LENGTH} 字`);
+      return;
+    }
+    if (activeRequests >= maxConcurrent) {
+      response.setHeader("Retry-After", "1");
+      sendJson(response, 429, "语音服务繁忙，请稍后重试");
+      return;
+    }
+
+    activeRequests += 1;
+    const synthesisTask = Promise.resolve().then(() => synthesize(text, defaultVoice));
+    void synthesisTask.finally(() => {
+      activeRequests -= 1;
+    }).catch(() => undefined);
+
+    let timeout: ReturnType<typeof setTimeout> | undefined;
+    try {
+      const audio = await Promise.race([
+        synthesisTask,
+        new Promise<never>((_, reject) => {
+          timeout = setTimeout(() => reject(new DOMException("语音服务响应超时", "TimeoutError")), timeoutMs);
+        })
+      ]);
+      if (!audio.byteLength) {
+        sendJson(response, 502, "语音服务返回了空音频");
+        return;
+      }
+      if (audio.byteLength > MAX_AUDIO_BYTES) {
+        sendJson(response, 502, "语音服务返回的音频过大");
+        return;
+      }
+      response.statusCode = 200;
+      response.setHeader("Content-Type", "audio/mpeg");
+      response.setHeader("Cache-Control", "no-store");
+      response.end(Buffer.from(audio));
+    } catch (error) {
+      if (error instanceof DOMException && error.name === "TimeoutError") {
+        sendJson(response, 504, "语音服务响应超时");
+        return;
+      }
+      console.error("[Agent TTS] 语音生成失败", error);
+      sendJson(response, 502, "语音生成失败");
+    } finally {
+      if (timeout) clearTimeout(timeout);
+    }
+  };
+}