whipwhep

server/src/services/audioForwardService.js

@@ -8,10 +8,12 @@ const { loadConfig } = require('../helpers/configLoader');
 const roverManager = require('./roverManager');
 const turnService = require('./turnService');
 const { isVerified } = require('./verificationService');
+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 =
@@ -26,6 +28,7 @@ const maxUploadBytes = Number.isFinite(audioForwardConfig.maxUploadBytes)
 
 const states = new Map(); // roverId -> { state, source, error, startedAt, updatedAt }
 const workers = new Map(); // roverId -> worker
+const whipOwners = new Map(); // roverId -> socketId
 
 function publishStateChange(roverId) {
   audioForwardEvents.emit('change', { roverId, state: states.get(roverId) || null });
@@ -126,6 +129,29 @@ function resolveForwardUrl(roverId) {
   )},m=publish&latency=10&mode=caller&transtype=live&pkt_size=1316`;
 }
 
+function resolveForwardPathId(roverId) {
+  return `${roverId}${streamSuffix}`;
+}
+
+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 spawnFfmpeg(roverId, tag, args, options = {}) {
   const proc = spawn(ffmpegBin, args, {
     stdio: [options.captureStdin ? 'pipe' : 'ignore', options.captureStdout ? 'pipe' : 'ignore', 'pipe'],
@@ -399,6 +425,12 @@ function ensureWorker(roverId) {
 }
 
 function stopWorker(roverId) {
+  const whipOwner = whipOwners.get(roverId);
+  if (whipOwner) {
+    whipOwners.delete(roverId);
+    revokeWhipSessionForRover(roverId, whipOwner);
+  }
+
   const worker = workers.get(roverId);
   if (!worker) return;
 
@@ -426,16 +458,12 @@ function writeUploadFile(roverId, payload = {}) {
   const { name, mime, dataBase64 } = payload || {};
   const ext = extFromUpload(name, mime);
   const encoded = typeof dataBase64 === 'string' ? dataBase64.trim() : '';
-  if (!encoded) {
-    throw new Error('Upload payload missing');
-  }
+  if (!encoded) throw new Error('Upload payload missing');
+
   const bytes = Buffer.from(encoded, 'base64');
-  if (!bytes.length) {
-    throw new Error('Upload decode failed');
-  }
-  if (bytes.length > maxUploadBytes) {
-    throw new Error(`Upload too large (max ${maxUploadBytes} bytes)`);
-  }
+  if (!bytes.length) throw new Error('Upload decode failed');
+  if (bytes.length > maxUploadBytes) throw new Error(`Upload too large (max ${maxUploadBytes} bytes)`);
+
   ensureRuntimeDir();
   const stem = sanitizeFileStem(name || `upload-${Date.now()}`);
   const filePath = path.join(uploadsDir, `${sanitizeRoverId(roverId)}-${Date.now()}-${stem}${ext}`);
@@ -444,18 +472,52 @@ function writeUploadFile(roverId, payload = {}) {
 }
 
 function playUploadedAudio(roverId, payload = {}, ownerSocketId = null) {
+  stopWhipForRover(roverId, 'upload_override');
   ensureWorker(roverId);
   const uploadPath = writeUploadFile(roverId, payload);
   startUploadWriter(roverId, uploadPath, ownerSocketId);
 }
 
 function stopPlayback(roverId) {
+  stopWhipForRover(roverId, 'stop_playback');
   ensureWorker(roverId);
   startSilenceWriter(roverId);
 }
 
-function stopOwnedUploadIfUnauthorized(roverId, ownerSocketId, reason = 'driver_change') {
+function revokeWhipSessionForRover(roverId, ownerSocketId) {
   if (!roverId || !ownerSocketId) return;
+  const pathId = resolveForwardPathId(roverId);
+  videoSessions.revokeWhere(
+    (info) => info?.socketId === ownerSocketId && info?.sourceType === 'roverMic' && info?.sourceId === pathId,
+  );
+}
+
+function stopWhipForRover(roverId, reason = 'unknown') {
+  const ownerSocketId = whipOwners.get(roverId);
+  if (!ownerSocketId) return;
+  whipOwners.delete(roverId);
+  revokeWhipSessionForRover(roverId, ownerSocketId);
+  logger.info('Stopping WHIP mic session', { roverId, ownerSocketId, reason });
+  try {
+    ensureWorker(roverId);
+    startSilenceWriter(roverId);
+  } catch (err) {
+    setState(roverId, { state: 'error', source: 'mic-whip', error: err?.message || String(err), startedAt: null });
+  }
+}
+
+function stopOwnedAudioIfUnauthorized(roverId, ownerSocketId, reason = 'driver_change') {
+  if (!roverId || !ownerSocketId) return;
+
+  if (whipOwners.get(roverId) === ownerSocketId) {
+    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) {
+      stopWhipForRover(roverId, reason);
+    }
+  }
+
   const worker = workers.get(roverId);
   if (!worker || worker.contentKind !== 'upload' || worker.activeOwnerSocketId !== ownerSocketId) return;
 
@@ -464,11 +526,7 @@ function stopOwnedUploadIfUnauthorized(roverId, ownerSocketId, reason = 'driver_
   const ownerCanDrive = ownerSocket ? turnService.canDrive(roverId, ownerSocket) : false;
   if (ownerIsDriver && ownerCanDrive) return;
 
-  logger.info('Stopping upload audio due to ownership/driver change', {
-    roverId,
-    ownerSocketId,
-    reason,
-  });
+  logger.info('Stopping upload audio due to ownership/driver change', { roverId, ownerSocketId, reason });
   startSilenceWriter(roverId);
 }
 
@@ -479,6 +537,9 @@ roverManager.managerEvents.on('rover', ({ roverId, action } = {}) => {
     return;
   }
   if (action === 'upsert' && serviceEnabled) {
+    if (whipOwners.has(roverId)) {
+      return;
+    }
     try {
       ensureWorker(roverId);
     } catch (err) {
@@ -490,15 +551,19 @@ roverManager.managerEvents.on('rover', ({ roverId, action } = {}) => {
 roverManager.managerEvents.on('driver', ({ socketId, roverId, action } = {}) => {
   if (!socketId || !roverId) return;
   if (action === 'remove' || action === 'add') {
-    stopOwnedUploadIfUnauthorized(roverId, socketId, action);
+    stopOwnedAudioIfUnauthorized(roverId, socketId, action);
   }
 });
 
 turnService.turnEvents.on('activeDriver', ({ roverId } = {}) => {
   if (!roverId) return;
+  const whipOwner = whipOwners.get(roverId);
+  if (whipOwner) {
+    stopOwnedAudioIfUnauthorized(roverId, whipOwner, 'turn_change');
+  }
   const worker = workers.get(roverId);
   if (!worker || worker.contentKind !== 'upload') return;
-  stopOwnedUploadIfUnauthorized(roverId, worker.activeOwnerSocketId, 'turn_change');
+  stopOwnedAudioIfUnauthorized(roverId, worker.activeOwnerSocketId, 'turn_change');
 });
 
 io.on('connection', (socket) => {
@@ -528,12 +593,61 @@ io.on('connection', (socket) => {
     }
   });
 
+  socket.on('audio:micWhipStart', ({ roverId } = {}, cb = () => {}) => {
+    try {
+      const normalized = String(roverId || '').trim();
+      ensureAudioForwardPermission(socket, normalized);
+      stopWorker(normalized);
+      whipOwners.set(normalized, socket.id);
+      const pathId = resolveForwardPathId(normalized);
+      revokeWhipSessionForRover(normalized, socket.id);
+      const token = videoSessions.createSession(socket, { type: 'roverMic', id: pathId });
+      const whipUrl = buildWhipUrl(pathId);
+      setState(normalized, { state: 'starting', source: 'mic-whip', error: null, startedAt: Date.now() });
+      cb({ success: true, roverId: normalized, pathId, token, whipUrl });
+    } catch (err) {
+      cb({ error: err.message });
+    }
+  });
+
+  socket.on('audio:micWhipReady', ({ roverId } = {}, cb = () => {}) => {
+    try {
+      const normalized = String(roverId || '').trim();
+      ensureAudioForwardPermission(socket, normalized);
+      if (whipOwners.get(normalized) !== socket.id) {
+        throw new Error('WHIP session not owned by this client');
+      }
+      setState(normalized, { state: 'playing', source: 'mic-whip', error: null, startedAt: Date.now() });
+      cb({ success: true, roverId: normalized });
+    } catch (err) {
+      cb({ error: err.message });
+    }
+  });
+
+  socket.on('audio:micWhipStop', ({ roverId } = {}, cb = () => {}) => {
+    try {
+      const normalized = String(roverId || '').trim();
+      ensureAudioForwardPermission(socket, normalized);
+      if (whipOwners.get(normalized) && whipOwners.get(normalized) !== socket.id) {
+        throw new Error('Mic forwarding is owned by another session');
+      }
+      stopWhipForRover(normalized, 'client_stop');
+      cb({ success: true, roverId: normalized });
+    } catch (err) {
+      cb({ error: err.message });
+    }
+  });
+
   socket.on('disconnect', () => {
     workers.forEach((worker, roverId) => {
       if (!worker || worker.contentKind !== 'upload' || worker.activeOwnerSocketId !== socket.id) return;
       logger.info('Stopping owned upload audio due to socket disconnect', { roverId, socketId: socket.id });
       startSilenceWriter(roverId);
     });
+    for (const [roverId, ownerSocketId] of whipOwners.entries()) {
+      if (ownerSocketId !== socket.id) continue;
+      stopWhipForRover(roverId, 'socket_disconnect');
+    }
   });
 });
 

server/src/services/videoAuthService.js

@@ -4,6 +4,8 @@ const logger = require('../globals/logger').child('videoAuth');
 const videoSessions = require('./videoSessions');
 const { getMode, MODES } = require('./modeManager');
 const { isAdmin, isLockdownAdmin, getRole } = require('./roleService');
+const { isVerified } = require('./verificationService');
+const turnService = require('./turnService');
 const roverManager = require('./roverManager');
 const { loadConfig } = require('../helpers/configLoader');
 const { getRequestIp, getSocketIp, isLocalNetwork } = require('../helpers/ipResolver');
@@ -145,7 +147,9 @@ app.post('/mediamtx/auth', (req, res) => {
   }
 
   const info = videoSessions.getSession(sessionId);
-  const streamTypeMatches = info && info.sourceType === streamInfo.type;
+  const streamTypeMatches =
+    info &&
+    (info.sourceType === streamInfo.type || (info.sourceType === 'roverMic' && streamInfo.type === 'rover'));
   if (!info || !streamTypeMatches || info.sourceId !== streamInfo.id) {
     logger.warn('invalid session %s for stream %s:%s', sessionId, streamInfo.type, streamInfo.id);
     return res.status(401).end();
@@ -158,6 +162,20 @@ app.post('/mediamtx/auth', (req, res) => {
   if (!canView(socket)) {
     return res.status(401).end();
   }
+  if (info.sourceType === 'roverMic' && action === 'publish') {
+    const roverId = streamInfo.baseId || streamInfo.id;
+    if (!isVerified(socket)) {
+      return res.status(401).end();
+    }
+    if (!roverManager.isDriver(roverId, socket)) {
+      return res.status(401).end();
+    }
+    if (!turnService.canDrive(roverId, socket)) {
+      return res.status(401).end();
+    }
+    return res.status(200).end();
+  }
+
   const role = getRole(socket);
   const isAudio = streamInfo.id?.endsWith('-audio');
   if (role === 'spectator' && !isAdmin(socket) && !isAudio) {