whipwhep

2026-09-16 01:21:20 -04:00 · 2026-03-20 01:43:06 -04:00
parent 5b8704ccb3
commit 7ad3dc7015
12 changed files with 529 additions and 187 deletions
@@ -8,13 +8,16 @@ const { loadConfig } = require('../helpers/configLoader');
 const roverManager = require('./roverManager');
 const { isVerified } = require('./verificationService');
 const turnService = require('./turnService');
+const videoSessions = require('./videoSessions');

 const audioForwardEvents = new EventEmitter();
 const config = loadConfig();
 const audioForwardConfig = config.audioForward || {};
+const mediaConfig = config.media || {};
 const serviceEnabled = audioForwardConfig.enabled !== false;
 const ffmpegBin = audioForwardConfig.ffmpegBin || 'ffmpeg';
 const streamSuffix = typeof audioForwardConfig.streamSuffix === 'string' ? audioForwardConfig.streamSuffix : '-fwd';
+const micSuffix = typeof audioForwardConfig.micSuffix === 'string' ? audioForwardConfig.micSuffix : '-mic';
 const runtimeDir = path.resolve(audioForwardConfig.runtimeDir || '/tmp/mrr-audio-forward');
 const uploadsDir = path.join(runtimeDir, 'uploads');
 const maxUploadBytes = Number.isFinite(audioForwardConfig.maxUploadBytes)
@@ -143,6 +146,34 @@ function resolveForwardUrl(roverId) {
  return `srt://127.0.0.1:9000?streamid=#!::r=${encodeURIComponent(roverId + streamSuffix)},m=publish&latency=10&mode=caller&transtype=live&pkt_size=1316`;
 }

+function resolveMicPathId(roverId) {
+  return `${roverId}${micSuffix}`;
+}
+
+function resolveMicReadUrl(roverId) {
+  const pathId = resolveMicPathId(roverId);
+  return `srt://127.0.0.1:9000?streamid=#!::r=${encodeURIComponent(pathId)},m=request&latency=10&mode=caller&transtype=live&pkt_size=1316`;
+}
+
+function getMediaPrefix() {
+  const base = mediaConfig.whepBaseUrl;
+  if (!base) return '';
+  try {
+    const parsed = new URL(base);
+    return `${parsed.origin}${parsed.pathname}`.replace(/\/+$/, '');
+  } catch {
+    return String(base).replace(/\/+$/, '');
+  }
+}
+
+function buildWhipUrl(pathId) {
+  const prefix = getMediaPrefix();
+  if (!prefix) {
+    throw new Error('Server media base URL missing');
+  }
+  return `${prefix}/${encodeURIComponent(pathId)}/whip`;
+}
+
 function spawnProcess(roverId, tag, args, options = {}) {
  const proc = spawn(ffmpegBin, args, {
    stdio: [options.captureStdin ? 'pipe' : 'ignore', options.captureStdout ? 'pipe' : 'ignore', 'pipe'],
@@ -258,6 +289,34 @@ function buildUploadWriterArgs(filePath) {
  ];
 }

+function buildWhipRelayReaderArgs(inputUrl) {
+  return [
+    '-hide_banner',
+    '-loglevel',
+    'warning',
+    '-fflags',
+    'nobuffer',
+    '-flags',
+    'low_delay',
+    '-analyzeduration',
+    '0',
+    '-probesize',
+    '32',
+    '-i',
+    inputUrl,
+    '-vn',
+    '-af',
+    'aresample=16000',
+    '-f',
+    's16le',
+    '-ac',
+    '1',
+    '-ar',
+    '16000',
+    'pipe:1',
+  ];
+}
+
 function attachWriterPipe(worker, proc) {
  const writer = fs.createWriteStream(worker.fifoPath, { flags: 'w' });
  writer.on('error', (err) => {
@@ -324,6 +383,7 @@ function stopContentWriter(worker) {
  worker.contentProc = null;
  worker.contentKind = null;
  worker.activeOwnerSocketId = null;
+  worker.micWhipPathId = null;
 }

 function startSilenceWriter(roverId) {
@@ -407,7 +467,7 @@ function startMicWriter(roverId, ownerSocketId = null) {
  if (!worker || worker.stopping) return;
  if (
    worker.contentKind === 'mic' &&
-    worker.contentProc &&
+    worker.micWriter &&
    worker.activeOwnerSocketId === ownerSocketId
  ) {
    return;
@@ -445,6 +505,50 @@ function startMicWriter(roverId, ownerSocketId = null) {
  setState(roverId, { state: 'playing', source: 'mic', error: null, startedAt: Date.now() });
 }

+function startMicWhipRelay(roverId, ownerSocketId = null) {
+  const worker = workers.get(roverId);
+  if (!worker || worker.stopping) return;
+  if (
+    worker.contentKind === 'mic_whip' &&
+    worker.contentProc &&
+    worker.activeOwnerSocketId === ownerSocketId
+  ) {
+    return;
+  }
+
+  stopContentWriter(worker);
+  cleanupUploadFile(worker);
+  const inputUrl = resolveMicReadUrl(roverId);
+  const proc = spawnProcess(roverId, 'mic-whip-reader', buildWhipRelayReaderArgs(inputUrl), {
+    captureStdout: true,
+  });
+  worker.contentProc = proc;
+  worker.contentKind = 'mic_whip';
+  worker.activeOwnerSocketId = ownerSocketId;
+  worker.micWhipPathId = resolveMicPathId(roverId);
+  const seq = ++worker.writerSeq;
+  attachWriterPipe(worker, proc);
+  setState(roverId, { state: 'playing', source: 'mic-whip', error: null, startedAt: Date.now() });
+
+  proc.on('exit', (code, signal) => {
+    const current = workers.get(roverId);
+    if (!current || current.stopping) return;
+    if (current.writerSeq !== seq || current.contentProc !== proc) return;
+    current.contentProc = null;
+    current.contentKind = null;
+    current.activeOwnerSocketId = null;
+    current.micWhipPathId = null;
+    if (code != null && code !== 0 && signal !== 'SIGTERM') {
+      setState(roverId, {
+        state: 'error',
+        source: 'mic-whip',
+        error: `mic whip reader exited code=${code} signal=${signal || 'none'}`,
+      });
+    }
+    startSilenceWriter(roverId);
+  });
+}
+
 function decodeMicChunk(payload = {}) {
  const binary = payload?.data;
  if (Buffer.isBuffer(binary)) {
@@ -531,6 +635,7 @@ function ensureWorker(roverId) {
    activeOwnerSocketId: null,
    activeUploadPath: null,
    micWriter: null,
+    micWhipPathId: null,
    micLastChunkAt: 0,
    micIdleTimer: null,
    micBackpressured: false,
@@ -639,12 +744,18 @@ function stopOwnedAudioIfUnauthorized(roverId, ownerSocketId, reason = 'driver_c
  if (!roverId || !ownerSocketId) return;
  const worker = workers.get(roverId);
  if (!worker) return;
-  if (worker.contentKind !== 'upload' && worker.contentKind !== 'mic') return;
+  if (worker.contentKind !== 'upload' && worker.contentKind !== 'mic' && worker.contentKind !== 'mic_whip') return;
  if (worker.activeOwnerSocketId !== ownerSocketId) return;
  const ownerSocket = io.sockets.sockets.get(ownerSocketId);
  const ownerIsDriver = ownerSocket ? roverManager.isDriver(roverId, ownerSocket) : false;
  const ownerCanDrive = ownerSocket ? turnService.canDrive(roverId, ownerSocket) : false;
  if (ownerIsDriver && ownerCanDrive) return;
+  if (worker.contentKind === 'mic_whip' && worker.micWhipPathId) {
+    const pathId = worker.micWhipPathId;
+    videoSessions.revokeWhere(
+      (info) => info?.socketId === ownerSocketId && info?.sourceType === 'roverMic' && info?.sourceId === pathId,
+    );
+  }
  logger.info('Stopping audio forward due to ownership/driver change', { roverId, ownerSocketId, reason, source: worker.contentKind });
  startSilenceWriter(roverId);
 }
@@ -659,7 +770,7 @@ roverManager.managerEvents.on('driver', ({ socketId, roverId, action } = {}) =>
 turnService.turnEvents.on('activeDriver', ({ roverId } = {}) => {
  if (!roverId) return;
  const worker = workers.get(roverId);
-  if (!worker || (worker.contentKind !== 'upload' && worker.contentKind !== 'mic')) return;
+  if (!worker || (worker.contentKind !== 'upload' && worker.contentKind !== 'mic' && worker.contentKind !== 'mic_whip')) return;
  stopOwnedAudioIfUnauthorized(roverId, worker.activeOwnerSocketId, 'turn_change');
 });

@@ -725,13 +836,51 @@ io.on('connection', (socket) => {
    }
  });

+  socket.on('audio:micWhipStart', ({ roverId } = {}, cb = () => {}) => {
+    try {
+      const normalized = String(roverId || '').trim();
+      ensureAudioForwardPermission(socket, normalized);
+      ensureWorker(normalized);
+      const pathId = resolveMicPathId(normalized);
+      videoSessions.revokeWhere(
+        (info) => info?.socketId === socket.id && info?.sourceType === 'roverMic' && info?.sourceId === pathId,
+      );
+      startMicWhipRelay(normalized, socket.id);
+      const token = videoSessions.createSession(socket, { type: 'roverMic', id: pathId });
+      const whipUrl = buildWhipUrl(pathId);
+      cb({ success: true, roverId: normalized, pathId, token, whipUrl });
+    } catch (err) {
+      cb({ error: err.message });
+    }
+  });
+
+  socket.on('audio:micWhipStop', ({ roverId } = {}, cb = () => {}) => {
+    try {
+      const normalized = String(roverId || '').trim();
+      ensureAudioForwardPermission(socket, normalized);
+      const worker = workers.get(normalized);
+      if (worker && worker.contentKind === 'mic_whip' && worker.activeOwnerSocketId !== socket.id) {
+        throw new Error('Mic forwarding is owned by another session');
+      }
+      const pathId = resolveMicPathId(normalized);
+      videoSessions.revokeWhere(
+        (info) => info?.socketId === socket.id && info?.sourceType === 'roverMic' && info?.sourceId === pathId,
+      );
+      stopPlayback(normalized);
+      cb({ success: true, roverId: normalized });
+    } catch (err) {
+      cb({ error: err.message });
+    }
+  });
+
  socket.on('disconnect', () => {
    workers.forEach((worker, roverId) => {
      if (!worker || worker.activeOwnerSocketId !== socket.id) return;
-      if (worker.contentKind !== 'upload' && worker.contentKind !== 'mic') return;
+      if (worker.contentKind !== 'upload' && worker.contentKind !== 'mic' && worker.contentKind !== 'mic_whip') return;
      logger.info('Stopping owned audio forward due to socket disconnect', { roverId, socketId: socket.id, source: worker.contentKind });
      startSilenceWriter(roverId);
    });
+    videoSessions.revokeWhere((info) => info?.socketId === socket.id && info?.sourceType === 'roverMic');
  });
 });