10 files changed, 967 insertions, 0 deletions

diff --git a/packages/peertube-runner/server/index.ts b/packages/peertube-runner/server/index.ts
new file mode 100644
index 000000000..371836515
--- /dev/null
+++ b/packages/peertube-runner/server/index.ts
@@ -0,0 +1 @@
+export * from './server'
diff --git a/packages/peertube-runner/server/process/index.ts b/packages/peertube-runner/server/process/index.ts
new file mode 100644
index 000000000..6caedbdaf
--- /dev/null
+++ b/packages/peertube-runner/server/process/index.ts
@@ -0,0 +1,2 @@
+export * from './shared'
+export * from './process'
diff --git a/packages/peertube-runner/server/process/process.ts b/packages/peertube-runner/server/process/process.ts
new file mode 100644
index 000000000..39a929c59
--- /dev/null
+++ b/packages/peertube-runner/server/process/process.ts
@@ -0,0 +1,30 @@
+import { logger } from 'packages/peertube-runner/shared/logger'
+import {
+  RunnerJobLiveRTMPHLSTranscodingPayload,
+  RunnerJobVODAudioMergeTranscodingPayload,
+  RunnerJobVODHLSTranscodingPayload,
+  RunnerJobVODWebVideoTranscodingPayload
+} from '@shared/models'
+import { processAudioMergeTranscoding, processHLSTranscoding, ProcessOptions, processWebVideoTranscoding } from './shared'
+import { ProcessLiveRTMPHLSTranscoding } from './shared/process-live'
+
+export async function processJob (options: ProcessOptions) {
+  const { server, job } = options
+
+  logger.info(`[${server.url}] Processing job of type ${job.type}: ${job.uuid}`, { payload: job.payload })
+
+  if (job.type === 'vod-audio-merge-transcoding') {
+    await processAudioMergeTranscoding(options as ProcessOptions<RunnerJobVODAudioMergeTranscodingPayload>)
+  } else if (job.type === 'vod-web-video-transcoding') {
+    await processWebVideoTranscoding(options as ProcessOptions<RunnerJobVODWebVideoTranscodingPayload>)
+  } else if (job.type === 'vod-hls-transcoding') {
+    await processHLSTranscoding(options as ProcessOptions<RunnerJobVODHLSTranscodingPayload>)
+  } else if (job.type === 'live-rtmp-hls-transcoding') {
+    await new ProcessLiveRTMPHLSTranscoding(options as ProcessOptions<RunnerJobLiveRTMPHLSTranscodingPayload>).process()
+  } else {
+    logger.error(`Unknown job ${job.type} to process`)
+    return
+  }
+
+  logger.info(`[${server.url}] Finished processing job of type ${job.type}: ${job.uuid}`)
+}
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`
+  ]
+}
diff --git a/packages/peertube-runner/server/server.ts b/packages/peertube-runner/server/server.ts
new file mode 100644
index 000000000..724f359bd
--- /dev/null
+++ b/packages/peertube-runner/server/server.ts
@@ -0,0 +1,269 @@
+import { ensureDir, readdir, remove } from 'fs-extra'
+import { join } from 'path'
+import { io, Socket } from 'socket.io-client'
+import { pick } from '@shared/core-utils'
+import { PeerTubeProblemDocument, ServerErrorCode } from '@shared/models'
+import { PeerTubeServer as PeerTubeServerCommand } from '@shared/server-commands'
+import { ConfigManager } from '../shared'
+import { IPCServer } from '../shared/ipc'
+import { logger } from '../shared/logger'
+import { JobWithToken, processJob } from './process'
+
+type PeerTubeServer = PeerTubeServerCommand & {
+  runnerToken: string
+  runnerName: string
+  runnerDescription?: string
+}
+
+export class RunnerServer {
+  private static instance: RunnerServer
+
+  private servers: PeerTubeServer[] = []
+  private processingJobs: { job: JobWithToken, server: PeerTubeServer }[] = []
+
+  private checkingAvailableJobs = false
+
+  private readonly sockets = new Map<PeerTubeServer, Socket>()
+
+  private constructor () {}
+
+  async run () {
+    logger.info('Running PeerTube runner in server mode')
+
+    await ConfigManager.Instance.load()
+
+    for (const registered of ConfigManager.Instance.getConfig().registeredInstances) {
+      const serverCommand = new PeerTubeServerCommand({ url: registered.url })
+
+      this.loadServer(Object.assign(serverCommand, registered))
+
+      logger.info(`Loading registered instance ${registered.url}`)
+    }
+
+    // Run IPC
+    const ipcServer = new IPCServer()
+    try {
+      await ipcServer.run(this)
+    } catch (err) {
+      console.error('Cannot start local socket for IPC communication', err)
+      process.exit(-1)
+    }
+
+    // Cleanup on exit
+    for (const code of [ 'SIGINT', 'SIGUSR1', 'SIGUSR2', 'uncaughtException' ]) {
+      process.on(code, async () => {
+        await this.onExit()
+      })
+    }
+
+    // Process jobs
+    await ensureDir(ConfigManager.Instance.getTranscodingDirectory())
+    await this.cleanupTMP()
+
+    logger.info(`Using ${ConfigManager.Instance.getTranscodingDirectory()} for transcoding directory`)
+
+    await this.checkAvailableJobs()
+  }
+
+  // ---------------------------------------------------------------------------
+
+  async registerRunner (options: {
+    url: string
+    registrationToken: string
+    runnerName: string
+    runnerDescription?: string
+  }) {
+    const { url, registrationToken, runnerName, runnerDescription } = options
+
+    logger.info(`Registering runner ${runnerName} on ${url}...`)
+
+    const serverCommand = new PeerTubeServerCommand({ url })
+    const { runnerToken } = await serverCommand.runners.register({ name: runnerName, description: runnerDescription, registrationToken })
+
+    const server: PeerTubeServer = Object.assign(serverCommand, {
+      runnerToken,
+      runnerName,
+      runnerDescription
+    })
+
+    this.loadServer(server)
+    await this.saveRegisteredInstancesInConf()
+
+    logger.info(`Registered runner ${runnerName} on ${url}`)
+
+    await this.checkAvailableJobs()
+  }
+
+  private loadServer (server: PeerTubeServer) {
+    this.servers.push(server)
+
+    const url = server.url + '/runners'
+    const socket = io(url, {
+      auth: {
+        runnerToken: server.runnerToken
+      },
+      transports: [ 'websocket' ]
+    })
+
+    socket.on('connect_error', err => logger.warn({ err }, `Cannot connect to ${url} socket`))
+    socket.on('connect', () => logger.info(`Connected to ${url} socket`))
+    socket.on('available-jobs', () => this.checkAvailableJobs())
+
+    this.sockets.set(server, socket)
+  }
+
+  async unregisterRunner (options: {
+    url: string
+  }) {
+    const { url } = options
+
+    const server = this.servers.find(s => s.url === url)
+    if (!server) {
+      logger.error(`Unknown server ${url} to unregister`)
+      return
+    }
+
+    logger.info(`Unregistering runner ${server.runnerName} on ${url}...`)
+
+    try {
+      await server.runners.unregister({ runnerToken: server.runnerToken })
+    } catch (err) {
+      logger.error({ err }, `Cannot unregister runner ${server.runnerName} on ${url}`)
+    }
+
+    this.unloadServer(server)
+    await this.saveRegisteredInstancesInConf()
+
+    logger.info(`Unregistered runner ${server.runnerName} on ${server.url}`)
+  }
+
+  private unloadServer (server: PeerTubeServer) {
+    this.servers = this.servers.filter(s => s !== server)
+
+    const socket = this.sockets.get(server)
+    socket.disconnect()
+
+    this.sockets.delete(server)
+  }
+
+  listRegistered () {
+    return {
+      servers: this.servers.map(s => {
+        return {
+          url: s.url,
+          runnerName: s.runnerName,
+          runnerDescription: s.runnerDescription
+        }
+      })
+    }
+  }
+
+  // ---------------------------------------------------------------------------
+
+  private async checkAvailableJobs () {
+    if (this.checkingAvailableJobs) return
+
+    logger.info('Checking available jobs')
+
+    this.checkingAvailableJobs = true
+
+    for (const server of this.servers) {
+      try {
+        const job = await this.requestJob(server)
+        if (!job) continue
+
+        await this.tryToExecuteJobAsync(server, job)
+      } catch (err) {
+        if ((err.res?.body as PeerTubeProblemDocument)?.code === ServerErrorCode.UNKNOWN_RUNNER_TOKEN) {
+          logger.error({ err }, `Unregistering ${server.url} as the runner token ${server.runnerToken} is invalid`)
+
+          await this.unregisterRunner({ url: server.url })
+          return
+        }
+
+        logger.error({ err }, `Cannot request/accept job on ${server.url} for runner ${server.runnerName}`)
+      }
+    }
+
+    this.checkingAvailableJobs = false
+  }
+
+  private async requestJob (server: PeerTubeServer) {
+    logger.debug(`Requesting jobs on ${server.url} for runner ${server.runnerName}`)
+
+    const { availableJobs } = await server.runnerJobs.request({ runnerToken: server.runnerToken })
+
+    if (availableJobs.length === 0) {
+      logger.debug(`No job available on ${server.url} for runner ${server.runnerName}`)
+      return undefined
+    }
+
+    return availableJobs[0]
+  }
+
+  private async tryToExecuteJobAsync (server: PeerTubeServer, jobToAccept: { uuid: string }) {
+    if (this.processingJobs.length >= ConfigManager.Instance.getConfig().jobs.concurrency) return
+
+    const { job } = await server.runnerJobs.accept({ runnerToken: server.runnerToken, jobUUID: jobToAccept.uuid })
+
+    const processingJob = { job, server }
+    this.processingJobs.push(processingJob)
+
+    processJob({ server, job, runnerToken: server.runnerToken })
+      .catch(err => {
+        logger.error({ err }, 'Cannot process job')
+
+        server.runnerJobs.error({ jobToken: job.jobToken, jobUUID: job.uuid, runnerToken: server.runnerToken, message: err.message })
+          .catch(err2 => logger.error({ err: err2 }, 'Cannot abort job after error'))
+      })
+      .finally(() => {
+        this.processingJobs = this.processingJobs.filter(p => p !== processingJob)
+
+        return this.checkAvailableJobs()
+      })
+  }
+
+  // ---------------------------------------------------------------------------
+
+  private saveRegisteredInstancesInConf () {
+    const data = this.servers.map(s => {
+      return pick(s, [ 'url', 'runnerToken', 'runnerName', 'runnerDescription' ])
+    })
+
+    return ConfigManager.Instance.setRegisteredInstances(data)
+  }
+
+  // ---------------------------------------------------------------------------
+
+  private async cleanupTMP () {
+    const files = await readdir(ConfigManager.Instance.getTranscodingDirectory())
+
+    for (const file of files) {
+      await remove(join(ConfigManager.Instance.getTranscodingDirectory(), file))
+    }
+  }
+
+  private async onExit () {
+    try {
+      for (const { server, job } of this.processingJobs) {
+        await server.runnerJobs.abort({
+          jobToken: job.jobToken,
+          jobUUID: job.uuid,
+          reason: 'Runner stopped',
+          runnerToken: server.runnerToken
+        })
+      }
+
+      await this.cleanupTMP()
+    } catch (err) {
+      console.error(err)
+      process.exit(-1)
+    }
+
+    process.exit()
+  }
+
+  static get Instance () {
+    return this.instance || (this.instance = new this())
+  }
+}