6 files changed, 665 insertions, 0 deletions

diff --git a/packages/peertube-runner/server/process/shared/common.ts b/packages/peertube-runner/server/process/shared/common.ts
new file mode 100644
index 000000000..9b2c40728
--- /dev/null
+++ b/packages/peertube-runner/server/process/shared/common.ts
@@ -0,0 +1,91 @@
+import { throttle } from 'lodash'
+import { ConfigManager, downloadFile, logger } from 'packages/peertube-runner/shared'
+import { join } from 'path'
+import { buildUUID } from '@shared/extra-utils'
+import { FFmpegLive, FFmpegVOD } from '@shared/ffmpeg'
+import { RunnerJob, RunnerJobPayload } from '@shared/models'
+import { PeerTubeServer } from '@shared/server-commands'
+import { getTranscodingLogger } from './transcoding-logger'
+import { getAvailableEncoders, getEncodersToTry } from './transcoding-profiles'
+
+export type JobWithToken <T extends RunnerJobPayload = RunnerJobPayload> = RunnerJob<T> & { jobToken: string }
+
+export type ProcessOptions <T extends RunnerJobPayload = RunnerJobPayload> = {
+  server: PeerTubeServer
+  job: JobWithToken<T>
+  runnerToken: string
+}
+
+export async function downloadInputFile (options: {
+  url: string
+  job: JobWithToken
+  runnerToken: string
+}) {
+  const { url, job, runnerToken } = options
+  const destination = join(ConfigManager.Instance.getTranscodingDirectory(), buildUUID())
+
+  await downloadFile({ url, jobToken: job.jobToken, runnerToken, destination })
+
+  return destination
+}
+
+export async function updateTranscodingProgress (options: {
+  server: PeerTubeServer
+  runnerToken: string
+  job: JobWithToken
+  progress: number
+}) {
+  const { server, job, runnerToken, progress } = options
+
+  return server.runnerJobs.update({ jobToken: job.jobToken, jobUUID: job.uuid, runnerToken, progress })
+}
+
+export function buildFFmpegVOD (options: {
+  server: PeerTubeServer
+  runnerToken: string
+  job: JobWithToken
+}) {
+  const { server, job, runnerToken } = options
+
+  const updateInterval = ConfigManager.Instance.isTestInstance()
+    ? 500
+    : 60000
+
+  const updateJobProgress = throttle((progress: number) => {
+    if (progress < 0 || progress > 100) progress = undefined
+
+    updateTranscodingProgress({ server, job, runnerToken, progress })
+      .catch(err => logger.error({ err }, 'Cannot send job progress'))
+  }, updateInterval, { trailing: false })
+
+  const config = ConfigManager.Instance.getConfig()
+
+  return new FFmpegVOD({
+    niceness: config.ffmpeg.nice,
+    threads: config.ffmpeg.threads,
+    tmpDirectory: ConfigManager.Instance.getTranscodingDirectory(),
+    profile: 'default',
+    availableEncoders: {
+      available: getAvailableEncoders(),
+      encodersToTry: getEncodersToTry()
+    },
+    logger: getTranscodingLogger(),
+    updateJobProgress
+  })
+}
+
+export function buildFFmpegLive () {
+  const config = ConfigManager.Instance.getConfig()
+
+  return new FFmpegLive({
+    niceness: config.ffmpeg.nice,
+    threads: config.ffmpeg.threads,
+    tmpDirectory: ConfigManager.Instance.getTranscodingDirectory(),
+    profile: 'default',
+    availableEncoders: {
+      available: getAvailableEncoders(),
+      encodersToTry: getEncodersToTry()
+    },
+    logger: getTranscodingLogger()
+  })
+}
diff --git a/packages/peertube-runner/server/process/shared/index.ts b/packages/peertube-runner/server/process/shared/index.ts
new file mode 100644
index 000000000..8e09a7869
--- /dev/null
+++ b/packages/peertube-runner/server/process/shared/index.ts
@@ -0,0 +1,4 @@
+export * from './common'
+export * from './process-vod'
+export * from './transcoding-logger'
+export * from './transcoding-profiles'
diff --git a/packages/peertube-runner/server/process/shared/process-live.ts b/packages/peertube-runner/server/process/shared/process-live.ts
new file mode 100644
index 000000000..5a3b596a2
--- /dev/null
+++ b/packages/peertube-runner/server/process/shared/process-live.ts
@@ -0,0 +1,295 @@
+import { FSWatcher, watch } from 'chokidar'
+import { FfmpegCommand } from 'fluent-ffmpeg'
+import { ensureDir, remove } from 'fs-extra'
+import { logger } from 'packages/peertube-runner/shared'
+import { basename, join } from 'path'
+import { wait } from '@shared/core-utils'
+import { buildUUID } from '@shared/extra-utils'
+import { ffprobePromise, getVideoStreamBitrate, getVideoStreamDimensionsInfo, hasAudioStream } from '@shared/ffmpeg'
+import {
+  LiveRTMPHLSTranscodingSuccess,
+  LiveRTMPHLSTranscodingUpdatePayload,
+  PeerTubeProblemDocument,
+  RunnerJobLiveRTMPHLSTranscodingPayload,
+  ServerErrorCode
+} from '@shared/models'
+import { ConfigManager } from '../../../shared/config-manager'
+import { buildFFmpegLive, ProcessOptions } from './common'
+
+export class ProcessLiveRTMPHLSTranscoding {
+
+  private readonly outputPath: string
+  private readonly fsWatchers: FSWatcher[] = []
+
+  private readonly playlistsCreated = new Set<string>()
+  private allPlaylistsCreated = false
+
+  private ffmpegCommand: FfmpegCommand
+
+  private ended = false
+  private errored = false
+
+  constructor (private readonly options: ProcessOptions<RunnerJobLiveRTMPHLSTranscodingPayload>) {
+    this.outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), buildUUID())
+  }
+
+  process () {
+    const job = this.options.job
+    const payload = job.payload
+
+    return new Promise<void>(async (res, rej) => {
+      try {
+        await ensureDir(this.outputPath)
+
+        logger.info(`Probing ${payload.input.rtmpUrl}`)
+        const probe = await ffprobePromise(payload.input.rtmpUrl)
+        logger.info({ probe }, `Probed ${payload.input.rtmpUrl}`)
+
+        const hasAudio = await hasAudioStream(payload.input.rtmpUrl, probe)
+        const bitrate = await getVideoStreamBitrate(payload.input.rtmpUrl, probe)
+        const { ratio } = await getVideoStreamDimensionsInfo(payload.input.rtmpUrl, probe)
+
+        const m3u8Watcher = watch(this.outputPath + '/*.m3u8')
+        this.fsWatchers.push(m3u8Watcher)
+
+        const tsWatcher = watch(this.outputPath + '/*.ts')
+        this.fsWatchers.push(tsWatcher)
+
+        m3u8Watcher.on('change', p => {
+          logger.debug(`${p} m3u8 playlist changed`)
+        })
+
+        m3u8Watcher.on('add', p => {
+          this.playlistsCreated.add(p)
+
+          if (this.playlistsCreated.size === this.options.job.payload.output.toTranscode.length + 1) {
+            this.allPlaylistsCreated = true
+            logger.info('All m3u8 playlists are created.')
+          }
+        })
+
+        tsWatcher.on('add', p => {
+          this.sendAddedChunkUpdate(p)
+            .catch(err => this.onUpdateError(err, rej))
+        })
+
+        tsWatcher.on('unlink', p => {
+          this.sendDeletedChunkUpdate(p)
+            .catch(err => this.onUpdateError(err, rej))
+        })
+
+        this.ffmpegCommand = await buildFFmpegLive().getLiveTranscodingCommand({
+          inputUrl: payload.input.rtmpUrl,
+
+          outPath: this.outputPath,
+          masterPlaylistName: 'master.m3u8',
+
+          segmentListSize: payload.output.segmentListSize,
+          segmentDuration: payload.output.segmentDuration,
+
+          toTranscode: payload.output.toTranscode,
+
+          bitrate,
+          ratio,
+
+          hasAudio
+        })
+
+        logger.info(`Running live transcoding for ${payload.input.rtmpUrl}`)
+
+        this.ffmpegCommand.on('error', (err, stdout, stderr) => {
+          this.onFFmpegError({ err, stdout, stderr })
+
+          res()
+        })
+
+        this.ffmpegCommand.on('end', () => {
+          this.onFFmpegEnded()
+            .catch(err => logger.error({ err }, 'Error in FFmpeg end handler'))
+
+          res()
+        })
+
+        this.ffmpegCommand.run()
+      } catch (err) {
+        rej(err)
+      }
+    })
+  }
+
+  // ---------------------------------------------------------------------------
+
+  private onUpdateError (err: Error, reject: (reason?: any) => void) {
+    if (this.errored) return
+    if (this.ended) return
+
+    this.errored = true
+
+    reject(err)
+    this.ffmpegCommand.kill('SIGINT')
+
+    const type = ((err as any).res?.body as PeerTubeProblemDocument)?.code
+    if (type === ServerErrorCode.RUNNER_JOB_NOT_IN_PROCESSING_STATE) {
+      logger.info({ err }, 'Stopping transcoding as the job is not in processing state anymore')
+    } else {
+      logger.error({ err }, 'Cannot send update after added/deleted chunk, stopping live transcoding')
+
+      this.sendError(err)
+        .catch(subErr => logger.error({ err: subErr }, 'Cannot send error'))
+    }
+
+    this.cleanup()
+  }
+
+  // ---------------------------------------------------------------------------
+
+  private onFFmpegError (options: {
+    err: any
+    stdout: string
+    stderr: string
+  }) {
+    const { err, stdout, stderr } = options
+
+    // Don't care that we killed the ffmpeg process
+    if (err?.message?.includes('Exiting normally')) return
+    if (this.errored) return
+    if (this.ended) return
+
+    this.errored = true
+
+    logger.error({ err, stdout, stderr }, 'FFmpeg transcoding error.')
+
+    this.sendError(err)
+      .catch(subErr => logger.error({ err: subErr }, 'Cannot send error'))
+
+    this.cleanup()
+  }
+
+  private async sendError (err: Error) {
+    await this.options.server.runnerJobs.error({
+      jobToken: this.options.job.jobToken,
+      jobUUID: this.options.job.uuid,
+      runnerToken: this.options.runnerToken,
+      message: err.message
+    })
+  }
+
+  // ---------------------------------------------------------------------------
+
+  private async onFFmpegEnded () {
+    if (this.ended) return
+
+    this.ended = true
+    logger.info('FFmpeg ended, sending success to server')
+
+    // Wait last ffmpeg chunks generation
+    await wait(1500)
+
+    this.sendSuccess()
+      .catch(err => logger.error({ err }, 'Cannot send success'))
+
+    this.cleanup()
+  }
+
+  private async sendSuccess () {
+    const successBody: LiveRTMPHLSTranscodingSuccess = {}
+
+    await this.options.server.runnerJobs.success({
+      jobToken: this.options.job.jobToken,
+      jobUUID: this.options.job.uuid,
+      runnerToken: this.options.runnerToken,
+      payload: successBody
+    })
+  }
+
+  // ---------------------------------------------------------------------------
+
+  private sendDeletedChunkUpdate (deletedChunk: string) {
+    if (this.ended) return
+
+    logger.debug(`Sending removed live chunk ${deletedChunk} update`)
+
+    const videoChunkFilename = basename(deletedChunk)
+
+    let payload: LiveRTMPHLSTranscodingUpdatePayload = {
+      type: 'remove-chunk',
+      videoChunkFilename
+    }
+
+    if (this.allPlaylistsCreated) {
+      const playlistName = this.getPlaylistName(videoChunkFilename)
+
+      payload = {
+        ...payload,
+        masterPlaylistFile: join(this.outputPath, 'master.m3u8'),
+        resolutionPlaylistFilename: playlistName,
+        resolutionPlaylistFile: join(this.outputPath, playlistName)
+      }
+    }
+
+    return this.updateWithRetry(payload)
+  }
+
+  private sendAddedChunkUpdate (addedChunk: string) {
+    if (this.ended) return
+
+    logger.debug(`Sending added live chunk ${addedChunk} update`)
+
+    const videoChunkFilename = basename(addedChunk)
+
+    let payload: LiveRTMPHLSTranscodingUpdatePayload = {
+      type: 'add-chunk',
+      videoChunkFilename,
+      videoChunkFile: addedChunk
+    }
+
+    if (this.allPlaylistsCreated) {
+      const playlistName = this.getPlaylistName(videoChunkFilename)
+
+      payload = {
+        ...payload,
+        masterPlaylistFile: join(this.outputPath, 'master.m3u8'),
+        resolutionPlaylistFilename: playlistName,
+        resolutionPlaylistFile: join(this.outputPath, playlistName)
+      }
+    }
+
+    return this.updateWithRetry(payload)
+  }
+
+  private async updateWithRetry (payload: LiveRTMPHLSTranscodingUpdatePayload, currentTry = 1) {
+    if (this.ended || this.errored) return
+
+    try {
+      await this.options.server.runnerJobs.update({
+        jobToken: this.options.job.jobToken,
+        jobUUID: this.options.job.uuid,
+        runnerToken: this.options.runnerToken,
+        payload
+      })
+    } catch (err) {
+      if (currentTry >= 3) throw err
+
+      logger.warn({ err }, 'Will retry update after error')
+      await wait(250)
+
+      return this.updateWithRetry(payload, currentTry + 1)
+    }
+  }
+
+  private getPlaylistName (videoChunkFilename: string) {
+    return `${videoChunkFilename.split('-')[0]}.m3u8`
+  }
+
+  // ---------------------------------------------------------------------------
+
+  private cleanup () {
+    for (const fsWatcher of this.fsWatchers) {
+      fsWatcher.close()
+        .catch(err => logger.error({ err }, 'Cannot close watcher'))
+    }
+
+    remove(this.outputPath)
+      .catch(err => logger.error({ err }, `Cannot remove ${this.outputPath}`))
+  }
+}
diff --git a/packages/peertube-runner/server/process/shared/process-vod.ts b/packages/peertube-runner/server/process/shared/process-vod.ts
new file mode 100644
index 000000000..aae61e9c5
--- /dev/null
+++ b/packages/peertube-runner/server/process/shared/process-vod.ts
@@ -0,0 +1,131 @@
+import { remove } from 'fs-extra'
+import { join } from 'path'
+import { buildUUID } from '@shared/extra-utils'
+import {
+  RunnerJobVODAudioMergeTranscodingPayload,
+  RunnerJobVODHLSTranscodingPayload,
+  RunnerJobVODWebVideoTranscodingPayload,
+  VODAudioMergeTranscodingSuccess,
+  VODHLSTranscodingSuccess,
+  VODWebVideoTranscodingSuccess
+} from '@shared/models'
+import { ConfigManager } from '../../../shared/config-manager'
+import { buildFFmpegVOD, downloadInputFile, ProcessOptions } from './common'
+
+export async function processWebVideoTranscoding (options: ProcessOptions<RunnerJobVODWebVideoTranscodingPayload>) {
+  const { server, job, runnerToken } = options
+  const payload = job.payload
+
+  const inputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
+
+  const ffmpegVod = buildFFmpegVOD({ job, server, runnerToken })
+
+  const outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), `output-${buildUUID()}.mp4`)
+
+  await ffmpegVod.transcode({
+    type: 'video',
+
+    inputPath,
+
+    outputPath,
+
+    inputFileMutexReleaser: () => {},
+
+    resolution: payload.output.resolution,
+    fps: payload.output.fps
+  })
+
+  const successBody: VODWebVideoTranscodingSuccess = {
+    videoFile: outputPath
+  }
+
+  await server.runnerJobs.success({
+    jobToken: job.jobToken,
+    jobUUID: job.uuid,
+    runnerToken,
+    payload: successBody
+  })
+
+  await remove(outputPath)
+}
+
+export async function processHLSTranscoding (options: ProcessOptions<RunnerJobVODHLSTranscodingPayload>) {
+  const { server, job, runnerToken } = options
+  const payload = job.payload
+
+  const inputPath = await downloadInputFile({ url: payload.input.videoFileUrl, runnerToken, job })
+  const uuid = buildUUID()
+
+  const outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), `${uuid}-${payload.output.resolution}.m3u8`)
+  const videoFilename = `${uuid}-${payload.output.resolution}-fragmented.mp4`
+  const videoPath = join(join(ConfigManager.Instance.getTranscodingDirectory(), videoFilename))
+
+  const ffmpegVod = buildFFmpegVOD({ job, server, runnerToken })
+
+  await ffmpegVod.transcode({
+    type: 'hls',
+    copyCodecs: false,
+    inputPath,
+    hlsPlaylist: { videoFilename },
+    outputPath,
+
+    inputFileMutexReleaser: () => {},
+
+    resolution: payload.output.resolution,
+    fps: payload.output.fps
+  })
+
+  const successBody: VODHLSTranscodingSuccess = {
+    resolutionPlaylistFile: outputPath,
+    videoFile: videoPath
+  }
+
+  await server.runnerJobs.success({
+    jobToken: job.jobToken,
+    jobUUID: job.uuid,
+    runnerToken,
+    payload: successBody
+  })
+
+  await remove(outputPath)
+  await remove(videoPath)
+}
+
+export async function processAudioMergeTranscoding (options: ProcessOptions<RunnerJobVODAudioMergeTranscodingPayload>) {
+  const { server, job, runnerToken } = options
+  const payload = job.payload
+
+  const audioPath = await downloadInputFile({ url: payload.input.audioFileUrl, runnerToken, job })
+  const inputPath = await downloadInputFile({ url: payload.input.previewFileUrl, runnerToken, job })
+
+  const outputPath = join(ConfigManager.Instance.getTranscodingDirectory(), `output-${buildUUID()}.mp4`)
+
+  const ffmpegVod = buildFFmpegVOD({ job, server, runnerToken })
+
+  await ffmpegVod.transcode({
+    type: 'merge-audio',
+
+    audioPath,
+    inputPath,
+
+    outputPath,
+
+    inputFileMutexReleaser: () => {},
+
+    resolution: payload.output.resolution,
+    fps: payload.output.fps
+  })
+
+  const successBody: VODAudioMergeTranscodingSuccess = {
+    videoFile: outputPath
+  }
+
+  await server.runnerJobs.success({
+    jobToken: job.jobToken,
+    jobUUID: job.uuid,
+    runnerToken,
+    payload: successBody
+  })
+
+  await remove(outputPath)
+}
diff --git a/packages/peertube-runner/server/process/shared/transcoding-logger.ts b/packages/peertube-runner/server/process/shared/transcoding-logger.ts
new file mode 100644
index 000000000..d0f928914
--- /dev/null
+++ b/packages/peertube-runner/server/process/shared/transcoding-logger.ts
@@ -0,0 +1,10 @@
+import { logger } from 'packages/peertube-runner/shared/logger'
+
+export function getTranscodingLogger () {
+  return {
+    info: logger.info.bind(logger),
+    debug: logger.debug.bind(logger),
+    warn: logger.warn.bind(logger),
+    error: logger.error.bind(logger)
+  }
+}
diff --git a/packages/peertube-runner/server/process/shared/transcoding-profiles.ts b/packages/peertube-runner/server/process/shared/transcoding-profiles.ts
new file mode 100644
index 000000000..492d17d6a
--- /dev/null
+++ b/packages/peertube-runner/server/process/shared/transcoding-profiles.ts
@@ -0,0 +1,134 @@
+import { getAverageBitrate, getMinLimitBitrate } from '@shared/core-utils'
+import { buildStreamSuffix, ffprobePromise, getAudioStream, getMaxAudioBitrate } from '@shared/ffmpeg'
+import { EncoderOptionsBuilder, EncoderOptionsBuilderParams, VideoResolution } from '@shared/models'
+
+const defaultX264VODOptionsBuilder: EncoderOptionsBuilder = (options: EncoderOptionsBuilderParams) => {
+  const { fps, inputRatio, inputBitrate, resolution } = options
+
+  const targetBitrate = getTargetBitrate({ inputBitrate, ratio: inputRatio, fps, resolution })
+
+  return {
+    outputOptions: [
+      ...getCommonOutputOptions(targetBitrate),
+
+      `-r ${fps}`
+    ]
+  }
+}
+
+const defaultX264LiveOptionsBuilder: EncoderOptionsBuilder = (options: EncoderOptionsBuilderParams) => {
+  const { streamNum, fps, inputBitrate, inputRatio, resolution } = options
+
+  const targetBitrate = getTargetBitrate({ inputBitrate, ratio: inputRatio, fps, resolution })
+
+  return {
+    outputOptions: [
+      ...getCommonOutputOptions(targetBitrate, streamNum),
+
+      `${buildStreamSuffix('-r:v', streamNum)} ${fps}`,
+      `${buildStreamSuffix('-b:v', streamNum)} ${targetBitrate}`
+    ]
+  }
+}
+
+const defaultAACOptionsBuilder: EncoderOptionsBuilder = async ({ input, streamNum, canCopyAudio }) => {
+  const probe = await ffprobePromise(input)
+
+  const parsedAudio = await getAudioStream(input, probe)
+
+  // We try to reduce the ceiling bitrate by making rough matches of bitrates
+  // Of course this is far from perfect, but it might save some space in the end
+
+  const audioCodecName = parsedAudio.audioStream['codec_name']
+
+  const bitrate = getMaxAudioBitrate(audioCodecName, parsedAudio.bitrate)
+
+  // Force stereo as it causes some issues with HLS playback in Chrome
+  const base = [ '-channel_layout', 'stereo' ]
+
+  if (bitrate !== -1) {
+    return { outputOptions: base.concat([ buildStreamSuffix('-b:a', streamNum), bitrate + 'k' ]) }
+  }
+
+  return { outputOptions: base }
+}
+
+const defaultLibFDKAACVODOptionsBuilder: EncoderOptionsBuilder = ({ streamNum }) => {
+  return { outputOptions: [ buildStreamSuffix('-q:a', streamNum), '5' ] }
+}
+
+export function getAvailableEncoders () {
+  return {
+    vod: {
+      libx264: {
+        default: defaultX264VODOptionsBuilder
+      },
+      aac: {
+        default: defaultAACOptionsBuilder
+      },
+      libfdk_aac: {
+        default: defaultLibFDKAACVODOptionsBuilder
+      }
+    },
+    live: {
+      libx264: {
+        default: defaultX264LiveOptionsBuilder
+      },
+      aac: {
+        default: defaultAACOptionsBuilder
+      }
+    }
+  }
+}
+
+export function getEncodersToTry () {
+  return {
+    vod: {
+      video: [ 'libx264' ],
+      audio: [ 'libfdk_aac', 'aac' ]
+    },
+
+    live: {
+      video: [ 'libx264' ],
+      audio: [ 'libfdk_aac', 'aac' ]
+    }
+  }
+}
+
+// ---------------------------------------------------------------------------
+
+function getTargetBitrate (options: {
+  inputBitrate: number
+  resolution: VideoResolution
+  ratio: number
+  fps: number
+}) {
+  const { inputBitrate, resolution, ratio, fps } = options
+
+  const capped = capBitrate(inputBitrate, getAverageBitrate({ resolution, fps, ratio }))
+  const limit = getMinLimitBitrate({ resolution, fps, ratio })
+
+  return Math.max(limit, capped)
+}
+
+function capBitrate (inputBitrate: number, targetBitrate: number) {
+  if (!inputBitrate) return targetBitrate
+
+  // Add 30% margin to input bitrate
+  const inputBitrateWithMargin = inputBitrate + (inputBitrate * 0.3)
+
+  return Math.min(targetBitrate, inputBitrateWithMargin)
+}
+
+function getCommonOutputOptions (targetBitrate: number, streamNum?: number) {
+  return [
+    `-preset veryfast`,
+    `${buildStreamSuffix('-maxrate:v', streamNum)} ${targetBitrate}`,
+    `${buildStreamSuffix('-bufsize:v', streamNum)} ${targetBitrate * 2}`,
+
+    // NOTE: b-strategy 1 - heuristic algorithm, 16 is optimal B-frames for it
+    `-b_strategy 1`,
+    // NOTE: Why 16: https://github.com/Chocobozzz/PeerTube/pull/774. b-strategy 2 -> B-frames<16
+    `-bf 16`
+  ]
+}