SolTube/server/core/lib/live-videos/live-video-process-manager.ts

import { logger, loggerTagsFactory } from '@server/helpers/logger.js'
import { stopScheduledLiveSession } from '@server/controllers/api/live-stream.js'
import { CONFIG } from '@server/initializers/config.js'
import { PeerTubeSocket } from '@server/lib/peertube-socket.js'
import { LiveVideoModel, LiveVideoStatus } from '@server/models/video/live-video.js'
import { getLiveVideoPlaybackInputPath } from './preprocessed-video.js'
import { buildRtspSourceUrlForTargetCamera } from './rtsp-url.js'
import { ChildProcessWithoutNullStreams, spawn } from 'node:child_process'
import { existsSync } from 'node:fs'

type StopReason = 'end-time' | 'cancelled' | 'deleted' | 'manual' | 'shutdown'

type RunningProcess = {
  endTimer?: NodeJS.Timeout
  ffmpegProcess: ChildProcessWithoutNullStreams
  liveVideoId: number
  stderrBuffer: string
  stopReason?: StopReason
  targetCamera: string
}

const lTags = loggerTagsFactory('live-videos', 'process-manager')

export class LiveVideoProcessManager {
  private static instance: LiveVideoProcessManager

  private readonly activeByLiveVideoId = new Map<number, RunningProcess>()
  private readonly liveVideoIdByTargetCamera = new Map<string, number>()

  async start (liveVideo: LiveVideoModel) {
    const existing = this.activeByLiveVideoId.get(liveVideo.id)
    if (existing) return false

    const otherLiveVideoId = this.liveVideoIdByTargetCamera.get(liveVideo.targetCamera)
    if (otherLiveVideoId && otherLiveVideoId !== liveVideo.id) {
      throw new Error(`Target camera ${liveVideo.targetCamera} is already streaming another live video.`)
    }

    const inputPath = getLiveVideoPlaybackInputPath(liveVideo.videoStoragePath)
    if (!existsSync(inputPath)) {
      throw new Error(`Input file does not exist: ${inputPath}`)
    }

    const ffmpegPath = CONFIG.LIVE.FFMPEG.PATH?.trim() || 'ffmpeg'
    const rtspTargetUrl = buildRtspSourceUrlForTargetCamera(liveVideo.targetCamera)
    const ffmpegArgs = [
      '-re',
      '-i', inputPath,
      '-vf', 'scale=-2:\'min(480,ih)\'',
      '-c:v', 'libx264',
      '-preset', 'ultrafast',
      '-tune', 'zerolatency',
      '-b:v', '2500k',
      '-maxrate', '2800k',
      '-bufsize', '5000k',
      '-profile:v', 'main',
      '-force_key_frames', 'expr:gte(t,n_forced*1)',
      '-c:a', 'aac',
      '-b:a', '128k',
      '-f', 'rtsp',
      '-rtsp_transport', 'tcp',
      rtspTargetUrl
    ]

    logger.info('Starting scheduled live video %d on camera %s.', liveVideo.id, liveVideo.targetCamera, {
      inputPath,
      usesPreprocessedInput: inputPath !== liveVideo.videoStoragePath,
      ...lTags()
    })
    const ffmpegProcess = spawn(ffmpegPath, ffmpegArgs, {
      stdio: [ 'ignore', 'ignore', 'pipe' ]
    })

    const runningProcess: RunningProcess = {
      ffmpegProcess,
      liveVideoId: liveVideo.id,
      stderrBuffer: '',
      targetCamera: liveVideo.targetCamera
    }

    this.activeByLiveVideoId.set(liveVideo.id, runningProcess)
    this.liveVideoIdByTargetCamera.set(liveVideo.targetCamera, liveVideo.id)

    ffmpegProcess.stderr.on('data', chunk => {
      const line = chunk?.toString?.() || ''
      runningProcess.stderrBuffer = (runningProcess.stderrBuffer + line).slice(-4_000)
    })

    ffmpegProcess.once('error', err => {
      logger.error('Cannot start ffmpeg for scheduled live video %d.', liveVideo.id, { err, ...lTags() })
    })

    ffmpegProcess.once('exit', (code, signal) => {
      void this.onProcessExit(runningProcess, code, signal)
    })

    const stopDelay = Math.max(0, liveVideo.endTime.getTime() - Date.now())
    runningProcess.endTimer = setTimeout(() => {
      void this.stop(liveVideo.id, 'end-time')
    }, stopDelay)

    return true
  }

  async stop (liveVideoId: number, reason: StopReason) {
    const runningProcess = this.activeByLiveVideoId.get(liveVideoId)
    if (!runningProcess) return false

    runningProcess.stopReason = reason

    if (runningProcess.endTimer) {
      clearTimeout(runningProcess.endTimer)
      runningProcess.endTimer = undefined
    }

    if (!runningProcess.ffmpegProcess.killed) {
      runningProcess.ffmpegProcess.kill('SIGTERM')

      setTimeout(() => {
        const existing = this.activeByLiveVideoId.get(liveVideoId)
        if (!existing || existing.ffmpegProcess.killed) return

        existing.ffmpegProcess.kill('SIGKILL')
      }, 5_000)
    }

    return true
  }

  async stopAll () {
    await Promise.all(Array.from(this.activeByLiveVideoId.keys()).map(id => this.stop(id, 'shutdown')))
  }

  private async onProcessExit (runningProcess: RunningProcess, code: number | null, signal: NodeJS.Signals | null) {
    if (runningProcess.endTimer) {
      clearTimeout(runningProcess.endTimer)
      runningProcess.endTimer = undefined
    }

    this.activeByLiveVideoId.delete(runningProcess.liveVideoId)
    this.liveVideoIdByTargetCamera.delete(runningProcess.targetCamera)

    const stderr = runningProcess.stderrBuffer.trim()
    if (code !== 0 && runningProcess.stopReason === undefined) {
      logger.error('Scheduled live video %d ffmpeg exited with code %s and signal %s.', runningProcess.liveVideoId, code, signal, {
        stderr,
        ...lTags()
      })
    } else {
      logger.info('Scheduled live video %d ffmpeg exited.', runningProcess.liveVideoId, {
        code,
        signal,
        reason: runningProcess.stopReason,
        ...lTags()
      })
    }

    const liveVideo = await LiveVideoModel.loadById(runningProcess.liveVideoId)
    const targetCamera = runningProcess.targetCamera

    await stopScheduledLiveSession(targetCamera)
      .catch(err => logger.error('Cannot stop scheduled sink stream session after ffmpeg exit.', Object.assign({ err, liveVideoId: runningProcess.liveVideoId, targetCamera }, lTags())))

    if (!liveVideo) return
    if (liveVideo.status !== LiveVideoStatus.SCHEDULED) return
    if (code !== 0 && runningProcess.stopReason === undefined) return
    if (runningProcess.stopReason && runningProcess.stopReason !== 'end-time') return

    await liveVideo.update({
      status: LiveVideoStatus.COMPLETED
    })

    await PeerTubeSocket.Instance.closeLiveChatRoom(targetCamera, 'live-ended')
      .catch(err => logger.error(
        'Cannot reset live chat room after scheduled live video completion.',
        Object.assign({ err, liveVideoId: runningProcess.liveVideoId, targetCamera }, lTags())
      ))
  }

  static get Instance () {
    return this.instance || (this.instance = new this())
  }
}