858 lines
27 KiB
TypeScript
858 lines
27 KiB
TypeScript
/**
|
||
* 录制服务：基于 mediasoup PlainTransport + ffmpeg 子进程
|
||
*
|
||
* 录制管线：
|
||
* ┌─────────────┐ RTP/RTCP ┌──────────────┐ SDP+stdin ┌────────┐ fs ┌─────────┐
|
||
* │ Producer X │ ───────────> │ PlainTransport│ ───────────────> │ ffmpeg │ ──────> │ *.mp4 │
|
||
* │ (audio/video)│ │ + Consumer │ (UDP loopback) │ (子进程)│ └─────────┘
|
||
* └─────────────┘ └──────────────┘ └────────┘
|
||
*
|
||
* 启动顺序（关键）：
|
||
* 1) 为每个 Producer 创建 Consumer 拿到 negotiated rtpParameters
|
||
* （Consumer 立即 pause，避免 RTP 在 ffmpeg 起来前空打）
|
||
* 2) 给每个 Consumer 分配一对 UDP 端口（ffmpeg 监听端）
|
||
* 3) 写 SDP 文件描述所有流
|
||
* 4) spawn ffmpeg 进程，listen 在那些端口
|
||
* 5) 创建对应 PlainTransport（rtcpMux=false, comedia=false）
|
||
* 并 connect 到 ffmpeg 监听端口
|
||
* 6) resume Consumer，RTP 开始流向 ffmpeg
|
||
*
|
||
* 停止顺序：
|
||
* 1) SIGINT ffmpeg（让 ffmpeg 关闭 mp4 moov box；SIGKILL 会导致文件损坏）
|
||
* 2) 等 ffmpeg 进程退出（超时 5s 兜底 SIGKILL）
|
||
* 3) close Consumer / PlainTransport
|
||
* 4) 释放端口
|
||
* 5) stat 输出文件，返回大小 + 时长
|
||
*
|
||
* 关键约束：
|
||
* - PlainTransport 用 rtcpMux=false：避免 RTCP 与 RTP 同端口，ffmpeg 解析更可靠
|
||
* - comedia=false：由 server 主动 connect，IP/Port 是已知的本地回环
|
||
* - 不开启 SRTP（loopback 不需要加密，降低 CPU 与配置复杂度）
|
||
* - Consumer 必须设置 paused=true 创建，否则 RTP 会先于 ffmpeg 启动到达
|
||
*/
|
||
import { spawn, type ChildProcess } from 'node:child_process';
|
||
import { randomUUID } from 'node:crypto';
|
||
import { promises as fs } from 'node:fs';
|
||
import { tmpdir } from 'node:os';
|
||
import { join } from 'node:path';
|
||
|
||
import type {
|
||
Consumer,
|
||
PlainTransport,
|
||
RtpCodecParameters,
|
||
RtpParameters,
|
||
} from 'mediasoup/types';
|
||
|
||
import { config } from '../config.js';
|
||
import { AppError, notFound } from '../utils/errors.js';
|
||
import { childLogger } from '../utils/logger.js';
|
||
import { allocatePortPair, releasePortPair } from '../utils/port-pool.js';
|
||
import { assertTestOnly } from '../utils/test-guard.js';
|
||
|
||
import { getProducer } from './producer.service.js';
|
||
import { getRouter } from './router.service.js';
|
||
|
||
const log = childLogger({ module: 'services.recording' });
|
||
|
||
// ============ 类型 ============
|
||
|
||
interface RecordingTrack {
|
||
/** 来源 producer ID（供日志/排障） */
|
||
producerId: string;
|
||
/** Consumer ID（停止时关闭） */
|
||
consumerId: string;
|
||
/** PlainTransport ID（停止时关闭） */
|
||
transportId: string;
|
||
/** ffmpeg 监听端口（RTP） */
|
||
ffmpegRtpPort: number;
|
||
/** ffmpeg 监听端口（RTCP） */
|
||
ffmpegRtcpPort: number;
|
||
kind: 'audio' | 'video';
|
||
/** payload type，用于 SDP / 跨进程串联 */
|
||
payloadType: number;
|
||
/** Codec mime，例如 audio/opus / video/VP8 */
|
||
codecMime: string;
|
||
/** Codec clock rate */
|
||
clockRate: number;
|
||
/** Codec channels（音频专用） */
|
||
channels?: number;
|
||
/** Codec sdpFmtp 字段（如 minptime=10;useinbandfec=1） */
|
||
sdpFmtp?: string;
|
||
}
|
||
|
||
interface RecordingEntry {
|
||
id: string;
|
||
roomCode: string;
|
||
routerId: string;
|
||
startedAt: number;
|
||
/** 各路 producer 对应的 track */
|
||
tracks: RecordingTrack[];
|
||
/** mediasoup Consumer 句柄（停止时 close） */
|
||
consumers: Consumer[];
|
||
/** mediasoup PlainTransport 句柄（停止时 close） */
|
||
transports: PlainTransport[];
|
||
/** ffmpeg 子进程 */
|
||
ffmpeg: ChildProcess;
|
||
/** SDP 临时文件 */
|
||
sdpFilePath: string;
|
||
/** 输出 mp4 文件 */
|
||
outputPath: string;
|
||
/** ffmpeg 退出 promise，stop 时 await */
|
||
ffmpegExited: Promise<void>;
|
||
/** 停止时回填，避免重复 stop */
|
||
stopping: boolean;
|
||
/** 退出码，stop 后填充供 callback 返回 */
|
||
exitCode: number | null;
|
||
/** 失败原因（spawn 阶段或 stop 超时），供 finalize 上报 */
|
||
failureReason: string | null;
|
||
/** ffmpeg 是否在 stop 之前异常退出（watchdog 标记，避免重复触发回调） */
|
||
unexpectedExit: boolean;
|
||
}
|
||
|
||
// ==================== Watchdog / 失败回调 ====================
|
||
|
||
/**
|
||
* ffmpeg 异常退出回调签名
|
||
*
|
||
* 触发条件：ffmpeg 在 stop 调用之前自行退出（非零退出码或 spawn 失败）
|
||
* 调用方可注册 webhook、上报告警、写入 DB 标记失败等
|
||
*
|
||
* 注意：回调在 ffmpeg 'exit' 事件后异步触发（setImmediate），不会阻塞退出处理；
|
||
* 实现方需保证内部错误不抛出到 Node 主循环，否则会被全局 unhandledRejection 接住。
|
||
*/
|
||
export type RecordingFailureCallback = (info: {
|
||
recordingId: string;
|
||
roomCode: string;
|
||
routerId: string;
|
||
exitCode: number | null;
|
||
failureReason: string;
|
||
}) => void | Promise<void>;
|
||
|
||
let failureCallback: RecordingFailureCallback | null = null;
|
||
|
||
/**
|
||
* 注册录制失败回调（覆盖式：只保留最近一次注册）
|
||
* 传 null 取消注册
|
||
*/
|
||
export function setRecordingFailureCallback(cb: RecordingFailureCallback | null): void {
|
||
failureCallback = cb;
|
||
}
|
||
|
||
/**
|
||
* 默认 webhook 实现：当 process.env.RECORDING_FAILURE_WEBHOOK_URL 设置时，
|
||
* 通过 fetch POST JSON 上报失败。Node 18+ 自带 fetch
|
||
*/
|
||
async function defaultWebhookCallback(info: Parameters<RecordingFailureCallback>[0]): Promise<void> {
|
||
const url = process.env.RECORDING_FAILURE_WEBHOOK_URL;
|
||
if (!url) return;
|
||
// go-service 端校验 X-Internal-Secret 与其 INTERNAL_WEBHOOK_SECRET 是否一致，
|
||
// 缺失/不匹配会直接 403。生产部署务必两侧使用相同值
|
||
const secret = process.env.INTERNAL_WEBHOOK_SECRET ?? '';
|
||
const headers: Record<string, string> = { 'Content-Type': 'application/json' };
|
||
if (secret) headers['X-Internal-Secret'] = secret;
|
||
try {
|
||
const res = await fetch(url, {
|
||
method: 'POST',
|
||
headers,
|
||
body: JSON.stringify(info),
|
||
});
|
||
if (!res.ok) {
|
||
log.warn(
|
||
{ recordingId: info.recordingId, status: res.status },
|
||
'recording failure webhook returned non-2xx',
|
||
);
|
||
}
|
||
} catch (err) {
|
||
log.warn(
|
||
{ recordingId: info.recordingId, err: err instanceof Error ? err.message : String(err) },
|
||
'recording failure webhook POST failed',
|
||
);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 实际异步派发失败回调：同时跑 default webhook + 用户注册的 callback
|
||
* 抽出此函数是为了让 spawn 阶段（entry 还没进 recordingMap）也能复用
|
||
*/
|
||
function dispatchFailureInfo(info: Parameters<RecordingFailureCallback>[0]): void {
|
||
setImmediate(() => {
|
||
void defaultWebhookCallback(info);
|
||
if (failureCallback) {
|
||
try {
|
||
const r = failureCallback(info);
|
||
if (r && typeof (r as Promise<void>).catch === 'function') {
|
||
(r as Promise<void>).catch((err) => {
|
||
log.warn(
|
||
{ recordingId: info.recordingId, err: err instanceof Error ? err.message : String(err) },
|
||
'recording failure callback rejected',
|
||
);
|
||
});
|
||
}
|
||
} catch (err) {
|
||
log.warn(
|
||
{ recordingId: info.recordingId, err: err instanceof Error ? err.message : String(err) },
|
||
'recording failure callback threw',
|
||
);
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 内部触发失败回调：异步、错误吞掉
|
||
* 入口语义"已 active 的 ffmpeg 进程异常退出"，依赖已落到 recordingMap 的 entry
|
||
*/
|
||
function fireFailureCallback(entry: RecordingEntry): void {
|
||
if (entry.unexpectedExit) return; // 幂等：每条录制最多触发一次
|
||
entry.unexpectedExit = true;
|
||
dispatchFailureInfo({
|
||
recordingId: entry.id,
|
||
roomCode: entry.roomCode,
|
||
routerId: entry.routerId,
|
||
exitCode: entry.exitCode,
|
||
failureReason: entry.failureReason ?? 'ffmpeg exited unexpectedly',
|
||
});
|
||
}
|
||
|
||
/**
|
||
* spawn 阶段失败兜底：ffmpeg 二进制找不到 / 无权限 / 启动后立刻退出
|
||
* 此时 entry 可能还没入 recordingMap（startRecording 在末尾才插入），
|
||
* 依旧需要把失败上报给 go-service，否则 startRecording 同步 throw 之后
|
||
* 万一上层 DB 行已经存在（异常时序、重试场景）就会卡在 recording 状态。
|
||
*
|
||
* 与 fireFailureCallback 互斥：只要其中一个跑过就标记 entry.unexpectedExit=true（如果有 entry）
|
||
*/
|
||
function fireSpawnFailure(args: {
|
||
recordingId: string;
|
||
roomCode: string;
|
||
routerId: string;
|
||
exitCode: number | null;
|
||
failureReason: string;
|
||
}): void {
|
||
const entry = recordingMap.get(args.recordingId);
|
||
if (entry) {
|
||
if (entry.unexpectedExit) return;
|
||
entry.unexpectedExit = true;
|
||
if (!entry.failureReason) entry.failureReason = args.failureReason;
|
||
if (entry.exitCode === null) entry.exitCode = args.exitCode;
|
||
}
|
||
dispatchFailureInfo(args);
|
||
}
|
||
|
||
const recordingMap = new Map<string, RecordingEntry>();
|
||
|
||
// 默认输出目录；由 RECORDING_OUTPUT_DIR 环境变量覆盖，便于 docker volume 挂载
|
||
const OUTPUT_DIR = process.env.RECORDING_OUTPUT_DIR
|
||
? process.env.RECORDING_OUTPUT_DIR
|
||
: join(tmpdir(), 'echochat-recordings');
|
||
|
||
const STOP_GRACE_MS = 5000;
|
||
|
||
// ============ 工具函数 ============
|
||
|
||
async function ensureOutputDir(): Promise<void> {
|
||
await fs.mkdir(OUTPUT_DIR, { recursive: true });
|
||
}
|
||
|
||
function recordingId(): string {
|
||
// crypto.randomUUID 在 node 20+ 可用；移除连字符以让文件名更紧凑、URL-safe
|
||
return randomUUID().replace(/-/g, '');
|
||
}
|
||
|
||
/**
|
||
* 从 Consumer 的 rtpParameters 提取录制需要的关键字段
|
||
* - 取第一个 codec：mediasoup consumer 默认每个 Consumer 只协商一个主 codec
|
||
* - sdpFmtp：把 codec.parameters 拼成 SDP a=fmtp 行格式
|
||
*/
|
||
function pickCodec(rtpParameters: RtpParameters): {
|
||
payloadType: number;
|
||
codecMime: string;
|
||
clockRate: number;
|
||
channels?: number;
|
||
sdpFmtp?: string;
|
||
} {
|
||
const codec: RtpCodecParameters | undefined = rtpParameters.codecs[0];
|
||
if (!codec) {
|
||
throw new AppError('MEDIASOUP_ERROR', 'consumer has no codecs in rtpParameters');
|
||
}
|
||
const params = codec.parameters || {};
|
||
const fmtpParts = Object.entries(params)
|
||
.filter(([, v]) => v !== undefined && v !== null && v !== '')
|
||
.map(([k, v]) => `${k}=${typeof v === 'boolean' ? (v ? 1 : 0) : String(v)}`);
|
||
return {
|
||
payloadType: codec.payloadType,
|
||
codecMime: codec.mimeType,
|
||
clockRate: codec.clockRate,
|
||
channels: codec.channels,
|
||
sdpFmtp: fmtpParts.length > 0 ? fmtpParts.join(';') : undefined,
|
||
};
|
||
}
|
||
|
||
/**
|
||
* 把 mime（audio/opus / video/VP8）转成 SDP rtpmap 编码名
|
||
* SDP a=rtpmap 不接受 `video/` 前缀，要单独取编码名
|
||
*/
|
||
function mimeToCodecName(mime: string): string {
|
||
const idx = mime.indexOf('/');
|
||
return idx >= 0 ? mime.slice(idx + 1) : mime;
|
||
}
|
||
|
||
/**
|
||
* 生成 SDP 内容
|
||
*
|
||
* 注意 SDP 是从 ffmpeg 视角写的：c= 行的地址、m= 行的端口都是 ffmpeg 监听处。
|
||
* mediasoup PlainTransport 之后 connect({ ip, port: ffmpegPort, rtcpPort }) 把流送过来。
|
||
*
|
||
* 一份 SDP 描述了一个会话里 1~N 条流（每条对应一个 producer 的 consumer）
|
||
*/
|
||
function buildSdp(tracks: RecordingTrack[], localIp: string): string {
|
||
const lines: string[] = [
|
||
'v=0',
|
||
`o=- 0 0 IN IP4 ${localIp}`,
|
||
's=echochat-recording',
|
||
`c=IN IP4 ${localIp}`,
|
||
't=0 0',
|
||
];
|
||
for (const t of tracks) {
|
||
const codecName = mimeToCodecName(t.codecMime);
|
||
const rtpmapValue =
|
||
t.kind === 'audio'
|
||
? `${codecName}/${t.clockRate}/${t.channels ?? 2}`
|
||
: `${codecName}/${t.clockRate}`;
|
||
lines.push(`m=${t.kind} ${t.ffmpegRtpPort} RTP/AVP ${t.payloadType}`);
|
||
lines.push(`a=rtcp:${t.ffmpegRtcpPort}`);
|
||
lines.push(`a=rtpmap:${t.payloadType} ${rtpmapValue}`);
|
||
if (t.sdpFmtp) {
|
||
lines.push(`a=fmtp:${t.payloadType} ${t.sdpFmtp}`);
|
||
}
|
||
lines.push('a=recvonly');
|
||
}
|
||
// SDP 必须以 \r\n 分隔，ffmpeg 解析行为更稳定
|
||
return lines.join('\r\n') + '\r\n';
|
||
}
|
||
|
||
/**
|
||
* 构造 ffmpeg 命令行参数
|
||
*
|
||
* -protocol_whitelist file,udp,rtp：默认只允许 file，需要显式开 udp/rtp
|
||
* -fflags +genpts：从 RTP 时间戳生成 PTS
|
||
* -use_wallclock_as_timestamps 1：避免 RTP 时间戳跳变导致 mp4 时长异常
|
||
* -c copy：不重编码，直接复用 RTP 载荷
|
||
* - 优点：CPU 占用极低
|
||
* - 缺点：浏览器侧 VP8 进 mp4 容器在某些播放器不支持；后续可改 libx264
|
||
* -y：覆盖已存在文件（防止 recordingId 碰撞）
|
||
*/
|
||
function buildFfmpegArgs(sdpFilePath: string, outputPath: string): string[] {
|
||
return [
|
||
'-loglevel',
|
||
'info',
|
||
'-protocol_whitelist',
|
||
'file,udp,rtp',
|
||
'-fflags',
|
||
'+genpts',
|
||
'-use_wallclock_as_timestamps',
|
||
'1',
|
||
'-i',
|
||
sdpFilePath,
|
||
'-map',
|
||
'0:a?',
|
||
'-map',
|
||
'0:v?',
|
||
'-c',
|
||
'copy',
|
||
'-y',
|
||
outputPath,
|
||
];
|
||
}
|
||
|
||
// ============ 对外 API ============
|
||
|
||
export interface StartRecordingInput {
|
||
routerId: string;
|
||
roomCode: string;
|
||
/**
|
||
* 要录制的 producer ID 列表。
|
||
* 通常是 host 的 audio + video，或者屏幕共享 video。
|
||
* Phase B1 阶段只录这些指定的 producer，不混合所有成员；后续 Phase B3 再扩展。
|
||
*/
|
||
producerIds: string[];
|
||
}
|
||
|
||
export interface StartRecordingOutput {
|
||
recordingId: string;
|
||
outputPath: string;
|
||
tracks: Array<{
|
||
producerId: string;
|
||
kind: 'audio' | 'video';
|
||
}>;
|
||
}
|
||
|
||
/**
|
||
* 启动录制
|
||
*
|
||
* 失败处理：
|
||
* - 任一步骤失败都会回滚已创建的资源（端口 / Transport / Consumer / ffmpeg 进程）
|
||
* - 不在 recordingMap 中留下"半态"条目，调用方可直接重试
|
||
*/
|
||
export async function startRecording(input: StartRecordingInput): Promise<StartRecordingOutput> {
|
||
const router = getRouter(input.routerId);
|
||
if (input.producerIds.length === 0) {
|
||
throw new AppError('CONFLICT', 'producerIds is empty');
|
||
}
|
||
await ensureOutputDir();
|
||
|
||
const id = recordingId();
|
||
const startedAt = Date.now();
|
||
const localIp = '127.0.0.1';
|
||
const outputPath = join(OUTPUT_DIR, `${id}.mp4`);
|
||
const sdpFilePath = join(OUTPUT_DIR, `${id}.sdp`);
|
||
|
||
// 已分配资源，用于失败时回滚
|
||
const allocatedPortPairs: Array<{ rtp: number; rtcp: number }> = [];
|
||
const createdTransports: PlainTransport[] = [];
|
||
const createdConsumers: Consumer[] = [];
|
||
let ffmpegProc: ChildProcess | null = null;
|
||
|
||
const rollback = async () => {
|
||
for (const c of createdConsumers) {
|
||
try {
|
||
c.close();
|
||
} catch {
|
||
/* ignored */
|
||
}
|
||
}
|
||
for (const t of createdTransports) {
|
||
try {
|
||
t.close();
|
||
} catch {
|
||
/* ignored */
|
||
}
|
||
}
|
||
for (const p of allocatedPortPairs) {
|
||
releasePortPair(p.rtp, p.rtcp);
|
||
}
|
||
if (ffmpegProc && ffmpegProc.exitCode === null) {
|
||
try {
|
||
ffmpegProc.kill('SIGKILL');
|
||
} catch {
|
||
/* ignored */
|
||
}
|
||
}
|
||
try {
|
||
await fs.unlink(sdpFilePath);
|
||
} catch {
|
||
/* ignored */
|
||
}
|
||
try {
|
||
await fs.unlink(outputPath);
|
||
} catch {
|
||
/* ignored */
|
||
}
|
||
};
|
||
|
||
try {
|
||
// ---------- Step 1: 预校验所有 producer 存在且可消费 ----------
|
||
// 不在这里建 Consumer：Consumer 必须依附 PlainTransport，留到 Step 2 一起做
|
||
// 但 canConsume 失败要尽早抛错，避免分配资源后再失败导致回滚开销
|
||
const tracks: RecordingTrack[] = [];
|
||
for (const producerId of input.producerIds) {
|
||
// getProducer 不存在会抛 notFound，等价于校验 producerId 合法
|
||
getProducer(producerId);
|
||
if (!router.canConsume({ producerId, rtpCapabilities: router.rtpCapabilities })) {
|
||
throw new AppError(
|
||
'CONFLICT',
|
||
`router cannot consume producer ${producerId} with its own rtpCapabilities`,
|
||
);
|
||
}
|
||
}
|
||
|
||
// ---------- Step 2: 为每个 producer 分配端口对 + 创建 PlainTransport ----------
|
||
for (const producerId of input.producerIds) {
|
||
const producer = getProducer(producerId);
|
||
const portPair = await allocatePortPair();
|
||
allocatedPortPairs.push(portPair);
|
||
|
||
const transport = await router.createPlainTransport({
|
||
listenIp: { ip: config.mediasoup.listenIp, announcedIp: undefined },
|
||
rtcpMux: false,
|
||
comedia: false,
|
||
enableSrtp: false,
|
||
appData: {
|
||
purpose: 'recording',
|
||
recordingId: id,
|
||
producerId,
|
||
},
|
||
});
|
||
createdTransports.push(transport);
|
||
|
||
// 创建 consumer：必须传 router 自己的 rtpCapabilities，consumer 会与 producer 协商
|
||
const consumer = await transport.consume({
|
||
producerId,
|
||
rtpCapabilities: router.rtpCapabilities,
|
||
paused: true,
|
||
});
|
||
createdConsumers.push(consumer);
|
||
|
||
const codec = pickCodec(consumer.rtpParameters);
|
||
tracks.push({
|
||
producerId,
|
||
consumerId: consumer.id,
|
||
transportId: transport.id,
|
||
ffmpegRtpPort: portPair.rtp,
|
||
ffmpegRtcpPort: portPair.rtcp,
|
||
kind: producer.kind as 'audio' | 'video',
|
||
...codec,
|
||
});
|
||
}
|
||
|
||
// ---------- Step 3: 写 SDP 文件 ----------
|
||
const sdp = buildSdp(tracks, localIp);
|
||
await fs.writeFile(sdpFilePath, sdp, 'utf-8');
|
||
log.info({ recordingId: id, sdp }, 'SDP file generated');
|
||
|
||
// ---------- Step 4: spawn ffmpeg ----------
|
||
const ffmpegBin = process.env.FFMPEG_BIN || 'ffmpeg';
|
||
const args = buildFfmpegArgs(sdpFilePath, outputPath);
|
||
ffmpegProc = spawn(ffmpegBin, args, { stdio: ['ignore', 'pipe', 'pipe'] });
|
||
|
||
// ffmpeg stderr 环形缓冲：正常运行时 debug 级 + 不落盘，
|
||
// 进程异常退出时把最后 N KB stderr 一次性以 error 级 dump 出来，方便定位 code=1 类问题
|
||
// 上限 16 KB 足以覆盖典型 ffmpeg 启动失败诊断（codec/端口/SDP 解析等），且容量可控
|
||
const STDERR_TAIL_LIMIT = 16 * 1024;
|
||
let stderrTail = '';
|
||
const appendStderrTail = (text: string) => {
|
||
stderrTail += text;
|
||
if (stderrTail.length > STDERR_TAIL_LIMIT) {
|
||
stderrTail = stderrTail.slice(stderrTail.length - STDERR_TAIL_LIMIT);
|
||
}
|
||
};
|
||
|
||
const exited = new Promise<void>((resolve) => {
|
||
ffmpegProc!.once('exit', (code, signal) => {
|
||
const entry = recordingMap.get(id);
|
||
if (entry) {
|
||
entry.exitCode = code;
|
||
if (!entry.stopping && code !== 0) {
|
||
entry.failureReason = `ffmpeg exited unexpectedly code=${code} signal=${signal ?? ''}`;
|
||
// 异常退出：把 stderr tail 落 error 级，便于线上排查 ffmpeg 失败根因
|
||
log.error(
|
||
{ recordingId: id, code, signal, stderrTail: stderrTail.trim() },
|
||
'ffmpeg exited unexpectedly before stop',
|
||
);
|
||
// Phase B watchdog：异步触发失败回调（webhook + 用户回调），不阻塞 exit 处理
|
||
fireFailureCallback(entry);
|
||
}
|
||
} else if (code !== 0) {
|
||
// entry 还没入 map（startRecording 末尾才插入）→ 走 spawn 失败兜底，避免 webhook 漏报
|
||
// startRecording 同步路径会自己 throw + rollback，这里上报只为防止上层 DB 行已存在的边缘场景
|
||
log.error(
|
||
{ recordingId: id, code, signal, stderrTail: stderrTail.trim() },
|
||
'ffmpeg exited before recording entry registered',
|
||
);
|
||
fireSpawnFailure({
|
||
recordingId: id,
|
||
roomCode: input.roomCode,
|
||
routerId: input.routerId,
|
||
exitCode: code,
|
||
failureReason: `ffmpeg exited during spawn code=${code} signal=${signal ?? ''}; stderr_tail=${stderrTail.trim().slice(-512)}`,
|
||
});
|
||
}
|
||
resolve();
|
||
});
|
||
});
|
||
|
||
ffmpegProc.stderr?.on('data', (chunk: Buffer) => {
|
||
const text = chunk.toString('utf-8');
|
||
// 1) 永远写入环形缓冲，供进程异常退出时 dump
|
||
appendStderrTail(text);
|
||
// 2) 正常实时日志仍走 debug 级，避免高频日志爆炸
|
||
log.debug({ recordingId: id }, `[ffmpeg] ${text.trim()}`);
|
||
});
|
||
|
||
ffmpegProc.once('error', (err) => {
|
||
// spawn 阶段错误：ffmpeg 二进制找不到 / 无权限 / OS 资源耗尽等
|
||
// 此前仅记日志，go-service 完全感知不到。这里立刻触发 failure webhook 兜底，
|
||
// 让上层（即使 DB 已写入 recording 行）能把状态推进到 failed
|
||
log.error({ recordingId: id, err: err.message }, 'ffmpeg spawn error');
|
||
fireSpawnFailure({
|
||
recordingId: id,
|
||
roomCode: input.roomCode,
|
||
routerId: input.routerId,
|
||
exitCode: null,
|
||
failureReason: `ffmpeg spawn error: ${err.message}`,
|
||
});
|
||
});
|
||
|
||
// 等 ffmpeg 进入监听就绪：实践经验 200ms 足够 ffmpeg 解析 SDP + 绑定端口
|
||
// 这里加一个 250ms 的小延迟，避免过早 connect 导致前几个 RTP 包丢失
|
||
await new Promise((r) => setTimeout(r, 250));
|
||
|
||
if (ffmpegProc.exitCode !== null) {
|
||
throw new AppError(
|
||
'MEDIASOUP_ERROR',
|
||
`ffmpeg exited prematurely with code ${ffmpegProc.exitCode}`,
|
||
);
|
||
}
|
||
|
||
// ---------- Step 5: PlainTransport.connect 到 ffmpeg ----------
|
||
for (let i = 0; i < input.producerIds.length; i++) {
|
||
const transport = createdTransports[i];
|
||
const t = tracks[i];
|
||
await transport.connect({
|
||
ip: localIp,
|
||
port: t.ffmpegRtpPort,
|
||
rtcpPort: t.ffmpegRtcpPort,
|
||
});
|
||
}
|
||
|
||
// ---------- Step 6: resume consumers，RTP 开始流向 ffmpeg ----------
|
||
for (const c of createdConsumers) {
|
||
await c.resume();
|
||
}
|
||
|
||
const entry: RecordingEntry = {
|
||
id,
|
||
roomCode: input.roomCode,
|
||
routerId: input.routerId,
|
||
startedAt,
|
||
tracks,
|
||
consumers: createdConsumers,
|
||
transports: createdTransports,
|
||
ffmpeg: ffmpegProc,
|
||
sdpFilePath,
|
||
outputPath,
|
||
ffmpegExited: exited,
|
||
stopping: false,
|
||
exitCode: null,
|
||
failureReason: null,
|
||
unexpectedExit: false,
|
||
};
|
||
recordingMap.set(id, entry);
|
||
|
||
log.info(
|
||
{
|
||
recordingId: id,
|
||
roomCode: input.roomCode,
|
||
producerCount: input.producerIds.length,
|
||
outputPath,
|
||
},
|
||
'recording started',
|
||
);
|
||
|
||
return {
|
||
recordingId: id,
|
||
outputPath,
|
||
tracks: tracks.map((t) => ({ producerId: t.producerId, kind: t.kind })),
|
||
};
|
||
} catch (err) {
|
||
log.error(
|
||
{ recordingId: id, err: err instanceof Error ? err.message : String(err) },
|
||
'startRecording failed, rolling back',
|
||
);
|
||
await rollback();
|
||
if (err instanceof AppError) throw err;
|
||
throw new AppError(
|
||
'MEDIASOUP_ERROR',
|
||
`startRecording failed: ${err instanceof Error ? err.message : String(err)}`,
|
||
);
|
||
}
|
||
}
|
||
|
||
export interface StopRecordingOutput {
|
||
recordingId: string;
|
||
outputPath: string;
|
||
sizeBytes: number;
|
||
durationMs: number;
|
||
exitCode: number | null;
|
||
failureReason: string | null;
|
||
}
|
||
|
||
/**
|
||
* 停止录制
|
||
*
|
||
* 幂等：重复调用同一 recordingId 仅第一次执行清理，后续返回 notFound
|
||
*
|
||
* 时间线：
|
||
* - t=0: SIGINT ffmpeg
|
||
* - t≤5s: ffmpeg 优雅退出，写完 moov box
|
||
* - t>5s: 强制 SIGKILL，文件可能损坏（罕见，记录 failureReason）
|
||
* - 之后 close consumers/transports/释放端口
|
||
*/
|
||
export async function stopRecording(id: string): Promise<StopRecordingOutput> {
|
||
const entry = recordingMap.get(id);
|
||
if (!entry) {
|
||
throw notFound('recording', id);
|
||
}
|
||
if (entry.stopping) {
|
||
// 并发重复 stop：返回当前快照，避免重复 SIGINT
|
||
log.warn({ recordingId: id }, 'stopRecording called while already stopping');
|
||
}
|
||
entry.stopping = true;
|
||
|
||
// 1) SIGINT ffmpeg
|
||
let killed = false;
|
||
if (entry.ffmpeg.exitCode === null) {
|
||
try {
|
||
entry.ffmpeg.kill('SIGINT');
|
||
killed = true;
|
||
} catch (err) {
|
||
log.warn(
|
||
{ recordingId: id, err: err instanceof Error ? err.message : String(err) },
|
||
'failed to send SIGINT to ffmpeg',
|
||
);
|
||
}
|
||
}
|
||
|
||
// 2) 等 ffmpeg 退出 + 超时兜底
|
||
if (killed) {
|
||
const timeout = new Promise<void>((resolve) => {
|
||
setTimeout(() => {
|
||
if (entry.ffmpeg.exitCode === null) {
|
||
try {
|
||
entry.ffmpeg.kill('SIGKILL');
|
||
} catch {
|
||
/* ignored */
|
||
}
|
||
entry.failureReason = `ffmpeg did not exit within ${STOP_GRACE_MS}ms, forced SIGKILL`;
|
||
log.error({ recordingId: id }, entry.failureReason);
|
||
}
|
||
resolve();
|
||
}, STOP_GRACE_MS);
|
||
});
|
||
await Promise.race([entry.ffmpegExited, timeout]);
|
||
// race 后再 await exited 拿到最终退出码
|
||
await entry.ffmpegExited;
|
||
}
|
||
|
||
// 3) close consumers + transports
|
||
for (const c of entry.consumers) {
|
||
try {
|
||
c.close();
|
||
} catch {
|
||
/* ignored */
|
||
}
|
||
}
|
||
for (const t of entry.transports) {
|
||
try {
|
||
t.close();
|
||
} catch {
|
||
/* ignored */
|
||
}
|
||
}
|
||
|
||
// 4) 释放端口
|
||
for (const track of entry.tracks) {
|
||
releasePortPair(track.ffmpegRtpPort, track.ffmpegRtcpPort);
|
||
}
|
||
|
||
// 5) 删除 SDP 临时文件（保留 mp4）
|
||
try {
|
||
await fs.unlink(entry.sdpFilePath);
|
||
} catch {
|
||
/* ignored */
|
||
}
|
||
|
||
// 6) stat 输出文件
|
||
let sizeBytes = 0;
|
||
try {
|
||
const st = await fs.stat(entry.outputPath);
|
||
sizeBytes = st.size;
|
||
} catch (err) {
|
||
if (!entry.failureReason) {
|
||
entry.failureReason = `output file missing: ${err instanceof Error ? err.message : String(err)}`;
|
||
}
|
||
}
|
||
|
||
const durationMs = Date.now() - entry.startedAt;
|
||
recordingMap.delete(id);
|
||
|
||
log.info(
|
||
{
|
||
recordingId: id,
|
||
durationMs,
|
||
sizeBytes,
|
||
exitCode: entry.exitCode,
|
||
failureReason: entry.failureReason,
|
||
},
|
||
'recording stopped',
|
||
);
|
||
|
||
return {
|
||
recordingId: id,
|
||
outputPath: entry.outputPath,
|
||
sizeBytes,
|
||
durationMs,
|
||
exitCode: entry.exitCode,
|
||
failureReason: entry.failureReason,
|
||
};
|
||
}
|
||
|
||
export interface RecordingInfo {
|
||
recordingId: string;
|
||
roomCode: string;
|
||
routerId: string;
|
||
startedAt: number;
|
||
durationMs: number;
|
||
outputPath: string;
|
||
trackCount: number;
|
||
ffmpegAlive: boolean;
|
||
}
|
||
|
||
export function getRecording(id: string): RecordingInfo {
|
||
const entry = recordingMap.get(id);
|
||
if (!entry) {
|
||
throw notFound('recording', id);
|
||
}
|
||
return {
|
||
recordingId: entry.id,
|
||
roomCode: entry.roomCode,
|
||
routerId: entry.routerId,
|
||
startedAt: entry.startedAt,
|
||
durationMs: Date.now() - entry.startedAt,
|
||
outputPath: entry.outputPath,
|
||
trackCount: entry.tracks.length,
|
||
ffmpegAlive: entry.ffmpeg.exitCode === null,
|
||
};
|
||
}
|
||
|
||
export function getRecordingStats(): { total: number } {
|
||
return { total: recordingMap.size };
|
||
}
|
||
|
||
/**
|
||
* 列出指定 roomCode 下的所有活跃录制（一般 ≤1）
|
||
* 供 Go 后端 EndRoom / 资源清理时反查
|
||
*/
|
||
export function listRecordingsByRoom(roomCode: string): RecordingInfo[] {
|
||
const out: RecordingInfo[] = [];
|
||
for (const entry of recordingMap.values()) {
|
||
if (entry.roomCode === roomCode) {
|
||
out.push({
|
||
recordingId: entry.id,
|
||
roomCode: entry.roomCode,
|
||
routerId: entry.routerId,
|
||
startedAt: entry.startedAt,
|
||
durationMs: Date.now() - entry.startedAt,
|
||
outputPath: entry.outputPath,
|
||
trackCount: entry.tracks.length,
|
||
ffmpegAlive: entry.ffmpeg.exitCode === null,
|
||
});
|
||
}
|
||
}
|
||
return out;
|
||
}
|
||
|
||
/** 测试专用：复位 recordingMap（停止所有进行中的录制） */
|
||
export async function _clearRecordings(): Promise<void> {
|
||
assertTestOnly('_clearRecordings');
|
||
const ids = [...recordingMap.keys()];
|
||
for (const id of ids) {
|
||
try {
|
||
await stopRecording(id);
|
||
} catch {
|
||
/* ignored */
|
||
}
|
||
}
|
||
recordingMap.clear();
|
||
}
|