kestrelos/server/utils/mediasoup.js

/**
 * Mediasoup SFU (Selective Forwarding Unit) setup and management.
 * Handles WebRTC router, transport, producer, and consumer creation.
 */

import os from 'node:os'
import mediasoup from 'mediasoup'

let worker = null
const routers = new Map() // sessionId -> Router
const transports = new Map() // transportId -> WebRtcTransport
export const producers = new Map() // producerId -> Producer

/**
 * Initialize Mediasoup worker (singleton).
 * @returns {Promise<mediasoup.types.Worker>} The Mediasoup worker.
 */
export async function getWorker() {
  if (worker) return worker
  worker = await mediasoup.createWorker({
    logLevel: process.env.NODE_ENV === 'development' ? 'debug' : 'warn',
    logTags: ['info', 'ice', 'dtls', 'rtp', 'srtp', 'rtcp'],
    rtcMinPort: 40000,
    rtcMaxPort: 49999,
  })
  worker.on('died', () => {
    console.error('[mediasoup] Worker died, exiting')
    process.exit(1)
  })
  return worker
}

/**
 * Create or get a router for a live session.
 * @param {string} sessionId
 * @returns {Promise<mediasoup.types.Router>} Router for the session.
 */
export async function getRouter(sessionId) {
  if (routers.has(sessionId)) {
    return routers.get(sessionId)
  }
  const w = await getWorker()
  const router = await w.createRouter({
    mediaCodecs: [
      {
        kind: 'video',
        mimeType: 'video/H264',
        clockRate: 90000,
        parameters: {
          'packetization-mode': 1,
          'profile-level-id': '42e01f',
        },
      },
      {
        kind: 'video',
        mimeType: 'video/VP8',
        clockRate: 90000,
      },
      {
        kind: 'video',
        mimeType: 'video/VP9',
        clockRate: 90000,
      },
    ],
  })
  routers.set(sessionId, router)
  return router
}

/**
 * True if the string is a valid IPv4 address (numeric a.b.c.d, each octet 0-255).
 * Used to accept request Host as announced IP only when it's safe (no hostnames/DNS rebinding).
 * @param {string} host
 * @returns {boolean} True if host is a valid IPv4 address.
 */
function isIPv4(host) {
  if (typeof host !== 'string' || !host) return false
  const parts = host.split('.')
  if (parts.length !== 4) return false
  for (const p of parts) {
    const n = Number.parseInt(p, 10)
    if (Number.isNaN(n) || n < 0 || n > 255 || String(n) !== p) return false
  }
  return true
}

/**
 * First non-internal IPv4 from network interfaces (no env read).
 * @returns {string | null} First non-internal IPv4 address or null.
 */
function getAnnouncedIpFromInterfaces() {
  const ifaces = os.networkInterfaces()
  for (const addrs of Object.values(ifaces)) {
    if (!addrs) continue
    for (const addr of addrs) {
      if (addr.family === 'IPv4' && !addr.internal) {
        return addr.address
      }
    }
  }
  return null
}

/**
 * Resolve announced IP: env override, then request host if IPv4, then auto-detect. Pure and deterministic.
 * @param {string | undefined} requestHost - Host header from the client.
 * @returns {string | null} The IP to announce in ICE, or null for localhost-only.
 */
function resolveAnnouncedIp(requestHost) {
  const envIp = process.env.MEDIASOUP_ANNOUNCED_IP?.trim()
  if (envIp) return envIp
  if (requestHost && isIPv4(requestHost)) return requestHost
  return getAnnouncedIpFromInterfaces()
}

/**
 * Create a WebRTC transport for a router.
 * @param {mediasoup.types.Router} router
 * @param {boolean} _isProducer - true for publisher, false for consumer (reserved for future use)
 * @param {string} [requestHost] - Hostname from the request (e.g. getRequestURL(event).hostname). If a valid IPv4, used as announced IP so the client can reach the server.
 * @returns {Promise<{ transport: mediasoup.types.WebRtcTransport, params: object }>} Transport and connection params.
 */
// eslint-disable-next-line no-unused-vars
export async function createTransport(router, _isProducer = false, requestHost = undefined) {
  // LAN first so the phone (and remote viewers) try the reachable IP before 127.0.0.1 (loopback on the client).
  const announcedIp = resolveAnnouncedIp(requestHost)
  const listenIps = announcedIp
    ? [{ ip: '0.0.0.0', announcedIp }, { ip: '127.0.0.1' }]
    : [{ ip: '127.0.0.1' }]

  const transport = await router.createWebRtcTransport({
    listenIps,
    enableUdp: true,
    enableTcp: true,
    preferUdp: true,
    initialAvailableOutgoingBitrate: 1_000_000,
  }).catch((err) => {
    console.error('[mediasoup] Transport creation failed:', err)
    throw new Error(`Failed to create transport: ${err.message || String(err)}`)
  })
  transports.set(transport.id, transport)
  transport.on('close', () => {
    transports.delete(transport.id)
  })
  return {
    transport,
    params: {
      id: transport.id,
      iceParameters: transport.iceParameters,
      iceCandidates: transport.iceCandidates,
      dtlsParameters: transport.dtlsParameters,
    },
  }
}

/**
 * Get transport by ID.
 * @param {string} transportId
 * @returns {mediasoup.types.WebRtcTransport | undefined} Transport or undefined.
 */
export function getTransport(transportId) {
  return transports.get(transportId)
}

/**
 * Create a producer (publisher's video track).
 * @param {mediasoup.types.WebRtcTransport} transport
 * @param {MediaStreamTrack} track
 * @returns {Promise<mediasoup.types.Producer>} The producer.
 */
export async function createProducer(transport, track) {
  const producer = await transport.produce({ track })
  producers.set(producer.id, producer)
  producer.on('close', () => {
    producers.delete(producer.id)
  })
  return producer
}

/**
 * Get producer by ID.
 * @param {string} producerId
 * @returns {mediasoup.types.Producer | undefined} Producer or undefined.
 */
export function getProducer(producerId) {
  return producers.get(producerId)
}

/**
 * Get transports Map (for cleanup).
 * @returns {Map<string, mediasoup.types.WebRtcTransport>} Map of transport ID to transport.
 */
export function getTransports() {
  return transports
}

/**
 * Create a consumer (viewer subscribes to producer's stream).
 * @param {mediasoup.types.WebRtcTransport} transport
 * @param {mediasoup.types.Producer} producer
 * @param {boolean} rtpCapabilities
 * @returns {Promise<{ consumer: mediasoup.types.Consumer, params: object }>} Consumer and connection params.
 */
export async function createConsumer(transport, producer, rtpCapabilities) {
  if (producer.closed) {
    throw new Error('Producer is closed')
  }
  if (producer.paused) {
    await producer.resume()
  }

  const consumer = await transport.consume({
    producerId: producer.id,
    rtpCapabilities,
    paused: false,
  })

  consumer.on('transportclose', () => {})
  consumer.on('producerclose', () => {})

  return {
    consumer,
    params: {
      id: consumer.id,
      producerId: consumer.producerId,
      kind: consumer.kind,
      rtpParameters: consumer.rtpParameters,
    },
  }
}

/**
 * Clean up router for a session.
 * @param {string} sessionId
 */
export async function closeRouter(sessionId) {
  const router = routers.get(sessionId)
  if (router) {
    router.close()
    routers.delete(sessionId)
  }
}

/**
 * Get all active routers (for debugging/monitoring).
 * @returns {Array<string>} Session IDs with active routers
 */
export function getActiveRouters() {
  return Array.from(routers.keys())
}