feat(phase2e-2): media-server Task 0/1/2 落地 + 代码审查修复

本次提交一次性落盘 Phase 2e-2 的 Task 0（PoC Spike）、Task 1（骨架 + Fastify 5 升级）和 Task 2（9 个内部 REST API + code-reviewer 审查修复），覆盖 media-server 子项目从零到可用的全部工作。【Task 0 - PoC Spike】 - media-server/poc/：Node + mediasoup + fastify-websocket + 前端 mediasoup-client - Playwright 双 tab 自动化验证 2 人会议：4 transports / 4 producers / 4 consumers / RSS 61MB - media-server/docs/poc-notes.md 归档 7 项关键坑 + 启动步骤 + Task 1/2/9 复用映射 - 锁定技术栈：mediasoup + fastify + mediasoup-client，不改用 livekit-server 【Task 1 - 骨架 + Fastify 5 升级】 - 依赖版本：mediasoup@3.19.0 + fastify@5.8.5 + fastify-plugin@5.1.0 + @fastify/sensible@6.0.4 + @fastify/websocket@11.2.0 + pino@9.3.2 + zod@3.23.8 - src 五件套：app.ts / config.ts / utils/logger.ts / mediasoup/worker.ts / middlewares/internal-auth.ts - /healthz + /readyz + /internal/info 三端点实测通过 - X-Internal-Token 鉴权用 timingSafeEqual 防侧信道 - mediasoup Worker died 指数退避自愈通过 kill -9 验证 - 多阶段 Dockerfile：非 root + curl HEALTHCHECK + 暴露 40000-40199 UDP/TCP - Fastify 4 → 5 升级：loggerInstance: logger 消灭两个 pino 实例【Task 2 - 9 个内部 REST API + 审查修复】接口全部挂 /internal/v1/* 前缀，覆盖 Router/Transport/Producer/Consumer 完整生命周期： - POST /routers, DELETE /routers/:id - POST /transports, POST /transports/:id/connect - POST /producers, DELETE /producers/:id - POST /consumers, POST /consumers/:id/resume, DELETE /consumers/:id 工程特性： - zod 手动 parse + 全局 errorHandler（不引入 fastify-type-provider-zod 避免 zod v4 依赖冲突） - AppError 统一错误码（NOT_FOUND/CONFLICT/CAN_NOT_CONSUME/ROUTER_LIMIT_EXCEEDED /MEDIASOUP_ERROR/VALIDATION_ERROR/UNAUTHORIZED/INTERNAL_ERROR） - 所有资源用 Map + observer.once('close') 自清理，Consumer 监听 producerclose 级联关闭 - Consumer 强制 paused:true 创建，/resume 独立接口 - Transport direction 强约束：recv 拒 produce、send 拒 consume code-reviewer 子代理审查"有条件通过"，同步修复： - M1 _clearXxxMap 新增 src/utils/test-guard.ts#assertTestOnly 守卫（生产误调用抛错） - M2 新增 src/schemas/rtp.ts 对 rtpParameters / rtpCapabilities 做 codecs 浅层校验（mimeType/clockRate/payloadType 必填、codecs ≥1），消除 as unknown as 双跳断言 - m1 connectTransport 改乐观锁：先置位再 await，失败回退 - m2 改读 consumer.producerPaused（更符合 mediasoup 语义） - m3 producerclose 改为 once（风格一致） - m5 internal-auth 改为反向白名单 PRIVATE_PATH_PREFIXES = ['/internal/'] 验证： - typecheck / lint 0 错误 - vitest：65 passed / 8 spec 文件 - 覆盖率 stmts 82.87% / branches 75.83% / funcs 91.3% / lines 82.87% - 9 接口 happy path + 6 类错误路径 curl 手测全部按预期返回【文档同步】 - CURRENT_STATUS.md +232 行：新增 Task 0/1/2 完整记录 + 代码审查修复章节 + 延后清单 - project-context.mdc：Task 0/1/2 状态同步 - phase2e-2-design.md +20 行：Fastify 5 升级相关决策记录 - phase2e-2-implementation.plan.md +117 行：Task 0/1/2 实际产出 + 修复记录余下 Minor/Nits（m4/m6~m10/n1~n10）登记至 Task 16 收尾清单一次性清扫。 Made-with: Cursor
2026-04-21 15:31:01 +08:00
parent df538e8df6
commit 6e2e792db0
56 changed files with 11123 additions and 51 deletions
--- a/media-server/src/services/consumer.service.ts
+++ b/media-server/src/services/consumer.service.ts
@@ -0,0 +1,150 @@
+import type { Consumer, MediaKind, RtpCapabilities, RtpParameters } from 'mediasoup/types';
+
+import { AppError, notFound } from '../utils/errors.js';
+import { childLogger } from '../utils/logger.js';
+import { assertTestOnly } from '../utils/test-guard.js';
+
+// 注：getProducer 仍然保留，既用于 notFound 语义（消费不存在的 producer 直接返回 404），
+// 也用于后续 Phase 2e-3 需要预读 producer.appData 做资格校验
+import { getProducer } from './producer.service.js';
+import { getRouter } from './router.service.js';
+import { getTransportEntry } from './transport.service.js';
+
+const log = childLogger({ module: 'services.consumer' });
+
+interface ConsumerEntry {
+  consumer: Consumer;
+  transportId: string;
+  producerId: string;
+  routerId: string;
+  userId: string;
+  createdAt: number;
+}
+
+const consumerMap = new Map<string, ConsumerEntry>();
+
+export interface CreatedConsumerInfo {
+  id: string;
+  kind: MediaKind;
+  rtpParameters: RtpParameters;
+  producerPaused: boolean;
+}
+
+export async function createConsumer(params: {
+  routerId: string;
+  transportId: string;
+  producerId: string;
+  rtpCapabilities: RtpCapabilities;
+}): Promise<CreatedConsumerInfo> {
+  const router = getRouter(params.routerId);
+  // 预检 producer 存在性；后续不再读 producer.paused，改用 consumer.producerPaused 更符合 mediasoup 语义
+  getProducer(params.producerId);
+  const transportEntry = getTransportEntry(params.transportId);
+
+  if (transportEntry.direction !== 'recv') {
+    throw new AppError(
+      'CONFLICT',
+      `transport ${params.transportId} is not a recv transport (direction=${transportEntry.direction})`,
+      { transportId: params.transportId, direction: transportEntry.direction },
+    );
+  }
+
+  if (!router.canConsume({ producerId: params.producerId, rtpCapabilities: params.rtpCapabilities })) {
+    throw new AppError(
+      'CAN_NOT_CONSUME',
+      `router cannot consume producer ${params.producerId} with given rtpCapabilities`,
+      { routerId: params.routerId, producerId: params.producerId },
+    );
+  }
+
+  try {
+    const consumer = await transportEntry.transport.consume({
+      producerId: params.producerId,
+      rtpCapabilities: params.rtpCapabilities,
+      paused: true,
+    });
+
+    consumer.observer.once('close', () => {
+      consumerMap.delete(consumer.id);
+      log.info(
+        { consumerId: consumer.id, transportId: params.transportId },
+        'consumer closed and removed from map',
+      );
+    });
+
+    consumer.once('producerclose', () => {
+      log.info({ consumerId: consumer.id }, 'upstream producer closed, closing consumer');
+      consumer.close();
+    });
+
+    consumerMap.set(consumer.id, {
+      consumer,
+      transportId: params.transportId,
+      producerId: params.producerId,
+      routerId: params.routerId,
+      userId: transportEntry.userId,
+      createdAt: Date.now(),
+    });
+
+    log.info(
+      {
+        consumerId: consumer.id,
+        transportId: params.transportId,
+        producerId: params.producerId,
+        kind: consumer.kind,
+      },
+      'consumer created (paused)',
+    );
+
+    return {
+      id: consumer.id,
+      kind: consumer.kind,
+      rtpParameters: consumer.rtpParameters,
+      producerPaused: consumer.producerPaused,
+    };
+  } catch (err) {
+    if (err instanceof AppError) {
+      throw err;
+    }
+    const message = err instanceof Error ? err.message : String(err);
+    throw new AppError('MEDIASOUP_ERROR', `failed to create consumer: ${message}`, {
+      transportId: params.transportId,
+      producerId: params.producerId,
+    });
+  }
+}
+
+export async function resumeConsumer(consumerId: string): Promise<void> {
+  const entry = consumerMap.get(consumerId);
+  if (!entry) {
+    throw notFound('consumer', consumerId);
+  }
+  await entry.consumer.resume();
+  log.info({ consumerId }, 'consumer resumed');
+}
+
+export async function closeConsumer(consumerId: string): Promise<void> {
+  const entry = consumerMap.get(consumerId);
+  if (!entry) {
+    throw notFound('consumer', consumerId);
+  }
+  entry.consumer.close();
+  log.info({ consumerId }, 'consumer closed explicitly');
+}
+
+export function getConsumerStats(): { total: number } {
+  return { total: consumerMap.size };
+}
+
+/** 测试专用：复位 consumerMap，生产环境调用会抛错 */
+export function _clearConsumerMap(): void {
+  assertTestOnly('_clearConsumerMap');
+  for (const entry of consumerMap.values()) {
+    try {
+      entry.consumer.close();
+    } catch {
+      // already closed
+    }
+  }
+  consumerMap.clear();
+}
--- a/media-server/src/services/producer.service.ts
+++ b/media-server/src/services/producer.service.ts
@@ -0,0 +1,112 @@
+import type { MediaKind, Producer, RtpParameters } from 'mediasoup/types';
+
+import { AppError, notFound } from '../utils/errors.js';
+import { childLogger } from '../utils/logger.js';
+import { assertTestOnly } from '../utils/test-guard.js';
+
+import { getTransportEntry } from './transport.service.js';
+
+const log = childLogger({ module: 'services.producer' });
+
+interface ProducerEntry {
+  producer: Producer;
+  transportId: string;
+  userId: string;
+  createdAt: number;
+}
+
+const producerMap = new Map<string, ProducerEntry>();
+
+export async function createProducer(params: {
+  transportId: string;
+  kind: MediaKind;
+  rtpParameters: RtpParameters;
+  appData?: Record<string, unknown>;
+}): Promise<{ id: string }> {
+  const entry = getTransportEntry(params.transportId);
+  if (entry.direction !== 'send') {
+    throw new AppError(
+      'CONFLICT',
+      `transport ${params.transportId} is not a send transport (direction=${entry.direction})`,
+      { transportId: params.transportId, direction: entry.direction },
+    );
+  }
+
+  try {
+    const producer = await entry.transport.produce({
+      kind: params.kind,
+      rtpParameters: params.rtpParameters,
+      appData: {
+        ...(params.appData ?? {}),
+        userId: entry.userId,
+      },
+    });
+
+    producer.observer.once('close', () => {
+      producerMap.delete(producer.id);
+      log.info(
+        { producerId: producer.id, transportId: params.transportId },
+        'producer closed and removed from map',
+      );
+    });
+
+    producerMap.set(producer.id, {
+      producer,
+      transportId: params.transportId,
+      userId: entry.userId,
+      createdAt: Date.now(),
+    });
+
+    log.info(
+      {
+        producerId: producer.id,
+        transportId: params.transportId,
+        userId: entry.userId,
+        kind: params.kind,
+      },
+      'producer created',
+    );
+
+    return { id: producer.id };
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err);
+    throw new AppError('MEDIASOUP_ERROR', `failed to create producer: ${message}`, {
+      transportId: params.transportId,
+      kind: params.kind,
+    });
+  }
+}
+
+export function getProducer(producerId: string): Producer {
+  const entry = producerMap.get(producerId);
+  if (!entry) {
+    throw notFound('producer', producerId);
+  }
+  return entry.producer;
+}
+
+export async function closeProducer(producerId: string): Promise<void> {
+  const entry = producerMap.get(producerId);
+  if (!entry) {
+    throw notFound('producer', producerId);
+  }
+  entry.producer.close();
+  log.info({ producerId }, 'producer closed explicitly');
+}
+
+export function getProducerStats(): { total: number } {
+  return { total: producerMap.size };
+}
+
+/** 测试专用：复位 producerMap，生产环境调用会抛错 */
+export function _clearProducerMap(): void {
+  assertTestOnly('_clearProducerMap');
+  for (const entry of producerMap.values()) {
+    try {
+      entry.producer.close();
+    } catch {
+      // already closed
+    }
+  }
+  producerMap.clear();
+}
--- a/media-server/src/services/router.service.ts
+++ b/media-server/src/services/router.service.ts
@@ -0,0 +1,100 @@
+import type { Router, RtpCapabilities } from 'mediasoup/types';
+
+import { config } from '../config.js';
+import { MEDIA_CODECS } from '../mediasoup/codecs.js';
+import { getWorker } from '../mediasoup/worker.js';
+import { AppError, notFound } from '../utils/errors.js';
+import { childLogger } from '../utils/logger.js';
+import { assertTestOnly } from '../utils/test-guard.js';
+
+const log = childLogger({ module: 'services.router' });
+
+interface RouterEntry {
+  router: Router;
+  roomCode: string;
+  createdAt: number;
+}
+
+const routerMap = new Map<string, RouterEntry>();
+
+export async function createRouter(roomCode: string): Promise<{
+  routerId: string;
+  rtpCapabilities: RtpCapabilities;
+}> {
+  if (routerMap.size >= config.mediasoup.maxRouters) {
+    throw new AppError(
+      'ROUTER_LIMIT_EXCEEDED',
+      `router limit reached: ${config.mediasoup.maxRouters}`,
+      { current: routerMap.size, max: config.mediasoup.maxRouters },
+    );
+  }
+
+  const worker = getWorker();
+  const router = await worker.createRouter({ mediaCodecs: MEDIA_CODECS });
+
+  router.observer.once('close', () => {
+    routerMap.delete(router.id);
+    log.info({ routerId: router.id, roomCode }, 'router closed and removed from map');
+  });
+
+  routerMap.set(router.id, {
+    router,
+    roomCode,
+    createdAt: Date.now(),
+  });
+
+  log.info({ routerId: router.id, roomCode, total: routerMap.size }, 'router created');
+  return {
+    routerId: router.id,
+    rtpCapabilities: router.rtpCapabilities,
+  };
+}
+
+export function getRouter(routerId: string): Router {
+  const entry = routerMap.get(routerId);
+  if (!entry) {
+    throw notFound('router', routerId);
+  }
+  return entry.router;
+}
+
+export function tryGetRouter(routerId: string): Router | undefined {
+  return routerMap.get(routerId)?.router;
+}
+
+export async function closeRouter(routerId: string): Promise<void> {
+  const entry = routerMap.get(routerId);
+  if (!entry) {
+    throw notFound('router', routerId);
+  }
+  entry.router.close();
+  log.info({ routerId, roomCode: entry.roomCode }, 'router closed explicitly');
+}
+
+export function getRouterStats(): {
+  total: number;
+  rooms: Array<{ routerId: string; roomCode: string; ageMs: number }>;
+} {
+  const now = Date.now();
+  return {
+    total: routerMap.size,
+    rooms: [...routerMap.entries()].map(([routerId, entry]) => ({
+      routerId,
+      roomCode: entry.roomCode,
+      ageMs: now - entry.createdAt,
+    })),
+  };
+}
+
+/** 测试专用：复位 routerMap，生产环境调用会抛错 */
+export function _clearRouterMap(): void {
+  assertTestOnly('_clearRouterMap');
+  for (const entry of routerMap.values()) {
+    try {
+      entry.router.close();
+    } catch {
+      // already closed
+    }
+  }
+  routerMap.clear();
+}
--- a/media-server/src/services/transport.service.ts
+++ b/media-server/src/services/transport.service.ts
@@ -0,0 +1,157 @@
+import type {
+  DtlsParameters,
+  IceCandidate,
+  IceParameters,
+  WebRtcTransport,
+} from 'mediasoup/types';
+
+import { config } from '../config.js';
+import { AppError, conflict, notFound } from '../utils/errors.js';
+import { childLogger } from '../utils/logger.js';
+import { assertTestOnly } from '../utils/test-guard.js';
+
+import { getRouter } from './router.service.js';
+import type { TransportDirection } from '../schemas/transport.schema.js';
+
+const log = childLogger({ module: 'services.transport' });
+
+interface TransportEntry {
+  transport: WebRtcTransport;
+  routerId: string;
+  userId: string;
+  direction: TransportDirection;
+  connected: boolean;
+  createdAt: number;
+}
+
+const transportMap = new Map<string, TransportEntry>();
+
+export interface CreatedTransportInfo {
+  id: string;
+  iceParameters: IceParameters;
+  iceCandidates: IceCandidate[];
+  dtlsParameters: DtlsParameters;
+}
+
+function buildListenIps(): Array<{ ip: string; announcedIp?: string }> {
+  const entry: { ip: string; announcedIp?: string } = {
+    ip: config.mediasoup.listenIp,
+  };
+  if (config.mediasoup.announcedIp) {
+    entry.announcedIp = config.mediasoup.announcedIp;
+  }
+  return [entry];
+}
+
+export async function createWebRtcTransport(params: {
+  routerId: string;
+  userId: string;
+  direction: TransportDirection;
+}): Promise<CreatedTransportInfo> {
+  const router = getRouter(params.routerId);
+
+  const transport = await router.createWebRtcTransport({
+    listenIps: buildListenIps(),
+    enableUdp: true,
+    enableTcp: true,
+    preferUdp: true,
+    initialAvailableOutgoingBitrate: 1_000_000,
+    appData: {
+      userId: params.userId,
+      direction: params.direction,
+      routerId: params.routerId,
+    },
+  });
+
+  transport.observer.once('close', () => {
+    transportMap.delete(transport.id);
+    log.info(
+      { transportId: transport.id, routerId: params.routerId, userId: params.userId },
+      'transport closed and removed from map',
+    );
+  });
+
+  transportMap.set(transport.id, {
+    transport,
+    routerId: params.routerId,
+    userId: params.userId,
+    direction: params.direction,
+    connected: false,
+    createdAt: Date.now(),
+  });
+
+  log.info(
+    {
+      transportId: transport.id,
+      routerId: params.routerId,
+      userId: params.userId,
+      direction: params.direction,
+    },
+    'webrtc transport created',
+  );
+
+  return {
+    id: transport.id,
+    iceParameters: transport.iceParameters,
+    iceCandidates: transport.iceCandidates,
+    dtlsParameters: transport.dtlsParameters,
+  };
+}
+
+export function getTransport(transportId: string): WebRtcTransport {
+  const entry = transportMap.get(transportId);
+  if (!entry) {
+    throw notFound('transport', transportId);
+  }
+  return entry.transport;
+}
+
+export function getTransportEntry(transportId: string): TransportEntry {
+  const entry = transportMap.get(transportId);
+  if (!entry) {
+    throw notFound('transport', transportId);
+  }
+  return entry;
+}
+
+export async function connectTransport(params: {
+  transportId: string;
+  dtlsParameters: DtlsParameters;
+}): Promise<void> {
+  const entry = transportMap.get(params.transportId);
+  if (!entry) {
+    throw notFound('transport', params.transportId);
+  }
+  // 乐观锁：先置位再 await，防止并发重复 connect 被 mediasoup 层包成 500
+  if (entry.connected) {
+    throw conflict(`transport already connected: ${params.transportId}`);
+  }
+  entry.connected = true;
+  try {
+    await entry.transport.connect({ dtlsParameters: params.dtlsParameters });
+    log.info({ transportId: params.transportId }, 'transport connected');
+  } catch (err) {
+    entry.connected = false;
+    const message = err instanceof Error ? err.message : String(err);
+    throw new AppError('MEDIASOUP_ERROR', `failed to connect transport: ${message}`, {
+      transportId: params.transportId,
+    });
+  }
+}
+
+export function getTransportStats(): { total: number } {
+  return { total: transportMap.size };
+}
+
+/** 测试专用：复位 transportMap，生产环境调用会抛错 */
+export function _clearTransportMap(): void {
+  assertTestOnly('_clearTransportMap');
+  for (const entry of transportMap.values()) {
+    try {
+      entry.transport.close();
+    } catch {
+      // already closed
+    }
+  }
+  transportMap.clear();
+}