next-tjwater-drainage-frontend/server/edge-tts-service.ts

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);
    }
  };
}