Implement remote runner jobs in server

Move ffmpeg functions to @shared

1 files changed, 42 insertions, 240 deletions

diff --git a/server/lib/job-queue/handlers/video-transcoding.ts b/server/lib/job-queue/handlers/video-transcoding.ts
index 3e6d23363..17b717275 100644
--- a/server/lib/job-queue/handlers/video-transcoding.ts
+++ b/server/lib/job-queue/handlers/video-transcoding.ts
@@ -1,13 +1,13 @@
 import { Job } from 'bullmq'
-import { TranscodeVODOptionsType } from '@server/helpers/ffmpeg'
-import { Hooks } from '@server/lib/plugins/hooks'
-import { buildTranscodingJob, getTranscodingJobPriority } from '@server/lib/video'
+import { onTranscodingEnded } from '@server/lib/transcoding/ended-transcoding'
+import { generateHlsPlaylistResolution } from '@server/lib/transcoding/hls-transcoding'
+import { mergeAudioVideofile, optimizeOriginalVideofile, transcodeNewWebTorrentResolution } from '@server/lib/transcoding/web-transcoding'
+import { removeAllWebTorrentFiles } from '@server/lib/video-file'
 import { VideoPathManager } from '@server/lib/video-path-manager'
-import { moveToFailedTranscodingState, moveToNextState } from '@server/lib/video-state'
+import { moveToFailedTranscodingState } from '@server/lib/video-state'
 import { UserModel } from '@server/models/user/user'
 import { VideoJobInfoModel } from '@server/models/video/video-job-info'
-import { MUser, MUserId, MVideo, MVideoFullLight, MVideoWithFile } from '@server/types/models'
-import { pick } from '@shared/core-utils'
+import { MUser, MUserId, MVideoFullLight } from '@server/types/models'
 import {
   HLSTranscodingPayload,
   MergeAudioTranscodingPayload,
@@ -15,18 +15,8 @@ import {
   OptimizeTranscodingPayload,
   VideoTranscodingPayload
 } from '@shared/models'
-import { retryTransactionWrapper } from '../../../helpers/database-utils'
-import { computeResolutionsToTranscode } from '../../../helpers/ffmpeg'
 import { logger, loggerTagsFactory } from '../../../helpers/logger'
-import { CONFIG } from '../../../initializers/config'
 import { VideoModel } from '../../../models/video/video'
-import {
-  generateHlsPlaylistResolution,
-  mergeAudioVideofile,
-  optimizeOriginalVideofile,
-  transcodeNewWebTorrentResolution
-} from '../../transcoding/transcoding'
-import { JobQueue } from '../job-queue'
 
 type HandlerFunction = (job: Job, payload: VideoTranscodingPayload, video: MVideoFullLight, user: MUser) => Promise<void>
 
@@ -84,260 +74,72 @@ export {
 // Job handlers
 // ---------------------------------------------------------------------------
 
-async function handleHLSJob (job: Job, payload: HLSTranscodingPayload, video: MVideoFullLight, user: MUser) {
-  logger.info('Handling HLS transcoding job for %s.', video.uuid, lTags(video.uuid))
-
-  const videoFileInput = payload.copyCodecs
-    ? video.getWebTorrentFile(payload.resolution)
-    : video.getMaxQualityFile()
-
-  const videoOrStreamingPlaylist = videoFileInput.getVideoOrStreamingPlaylist()
-
-  const inputFileMutexReleaser = await VideoPathManager.Instance.lockFiles(video.uuid)
-
-  try {
-    await videoFileInput.getVideo().reload()
-
-    await VideoPathManager.Instance.makeAvailableVideoFile(videoFileInput.withVideoOrPlaylist(videoOrStreamingPlaylist), videoInputPath => {
-      return generateHlsPlaylistResolution({
-        video,
-        videoInputPath,
-        inputFileMutexReleaser,
-        resolution: payload.resolution,
-        copyCodecs: payload.copyCodecs,
-        job
-      })
-    })
-  } finally {
-    inputFileMutexReleaser()
-  }
-
-  logger.info('HLS transcoding job for %s ended.', video.uuid, lTags(video.uuid))
-
-  await onHlsPlaylistGeneration(video, user, payload)
-}
-
-async function handleNewWebTorrentResolutionJob (
-  job: Job,
-  payload: NewWebTorrentResolutionTranscodingPayload,
-  video: MVideoFullLight,
-  user: MUserId
-) {
-  logger.info('Handling WebTorrent transcoding job for %s.', video.uuid, lTags(video.uuid))
-
-  await transcodeNewWebTorrentResolution({ video, resolution: payload.resolution, job })
-
-  logger.info('WebTorrent transcoding job for %s ended.', video.uuid, lTags(video.uuid))
-
-  await onNewWebTorrentFileResolution(video, user, payload)
-}
-
 async function handleWebTorrentMergeAudioJob (job: Job, payload: MergeAudioTranscodingPayload, video: MVideoFullLight, user: MUserId) {
   logger.info('Handling merge audio transcoding job for %s.', video.uuid, lTags(video.uuid))
 
-  await mergeAudioVideofile({ video, resolution: payload.resolution, job })
+  await mergeAudioVideofile({ video, resolution: payload.resolution, fps: payload.fps, job })
 
   logger.info('Merge audio transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 
-  await onVideoFirstWebTorrentTranscoding(video, payload, 'video', user)
+  await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: true, video })
 }
 
 async function handleWebTorrentOptimizeJob (job: Job, payload: OptimizeTranscodingPayload, video: MVideoFullLight, user: MUserId) {
   logger.info('Handling optimize transcoding job for %s.', video.uuid, lTags(video.uuid))
 
-  const { transcodeType } = await optimizeOriginalVideofile({ video, inputVideoFile: video.getMaxQualityFile(), job })
+  await optimizeOriginalVideofile({ video, inputVideoFile: video.getMaxQualityFile(), quickTranscode: payload.quickTranscode, job })
 
   logger.info('Optimize transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 
-  await onVideoFirstWebTorrentTranscoding(video, payload, transcodeType, user)
+  await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: true, video })
 }
 
-// ---------------------------------------------------------------------------
-
-async function onHlsPlaylistGeneration (video: MVideoFullLight, user: MUser, payload: HLSTranscodingPayload) {
-  if (payload.isMaxQuality && payload.autoDeleteWebTorrentIfNeeded && CONFIG.TRANSCODING.WEBTORRENT.ENABLED === false) {
-    // Remove webtorrent files if not enabled
-    for (const file of video.VideoFiles) {
-      await video.removeWebTorrentFile(file)
-      await file.destroy()
-    }
-
-    video.VideoFiles = []
-
-    // Create HLS new resolution jobs
-    await createLowerResolutionsJobs({
-      video,
-      user,
-      videoFileResolution: payload.resolution,
-      hasAudio: payload.hasAudio,
-      isNewVideo: payload.isNewVideo ?? true,
-      type: 'hls'
-    })
-  }
-
-  await VideoJobInfoModel.decrease(video.uuid, 'pendingTranscode')
-  await retryTransactionWrapper(moveToNextState, { video, isNewVideo: payload.isNewVideo })
-}
+async function handleNewWebTorrentResolutionJob (job: Job, payload: NewWebTorrentResolutionTranscodingPayload, video: MVideoFullLight) {
+  logger.info('Handling WebTorrent transcoding job for %s.', video.uuid, lTags(video.uuid))
 
-async function onVideoFirstWebTorrentTranscoding (
-  videoArg: MVideoWithFile,
-  payload: OptimizeTranscodingPayload | MergeAudioTranscodingPayload,
-  transcodeType: TranscodeVODOptionsType,
-  user: MUserId
-) {
-  const mutexReleaser = await VideoPathManager.Instance.lockFiles(videoArg.uuid)
+  await transcodeNewWebTorrentResolution({ video, resolution: payload.resolution, fps: payload.fps, job })
 
-  try {
-    // Maybe the video changed in database, refresh it
-    const videoDatabase = await VideoModel.loadFull(videoArg.uuid)
-    // Video does not exist anymore
-    if (!videoDatabase) return undefined
-
-    const { resolution, audioStream } = await videoDatabase.probeMaxQualityFile()
-
-    // Generate HLS version of the original file
-    const originalFileHLSPayload = {
-      ...payload,
-
-      hasAudio: !!audioStream,
-      resolution: videoDatabase.getMaxQualityFile().resolution,
-      // If we quick transcoded original file, force transcoding for HLS to avoid some weird playback issues
-      copyCodecs: transcodeType !== 'quick-transcode',
-      isMaxQuality: true
-    }
-    const hasHls = await createHlsJobIfEnabled(user, originalFileHLSPayload)
-    const hasNewResolutions = await createLowerResolutionsJobs({
-      video: videoDatabase,
-      user,
-      videoFileResolution: resolution,
-      hasAudio: !!audioStream,
-      type: 'webtorrent',
-      isNewVideo: payload.isNewVideo ?? true
-    })
-
-    await VideoJobInfoModel.decrease(videoDatabase.uuid, 'pendingTranscode')
+  logger.info('WebTorrent transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 
-    // Move to next state if there are no other resolutions to generate
-    if (!hasHls && !hasNewResolutions) {
-      await retryTransactionWrapper(moveToNextState, { video: videoDatabase, isNewVideo: payload.isNewVideo })
-    }
-  } finally {
-    mutexReleaser()
-  }
+  await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: true, video })
 }
 
-async function onNewWebTorrentFileResolution (
-  video: MVideo,
-  user: MUserId,
-  payload: NewWebTorrentResolutionTranscodingPayload | MergeAudioTranscodingPayload
-) {
-  if (payload.createHLSIfNeeded) {
-    await createHlsJobIfEnabled(user, { hasAudio: true, copyCodecs: true, isMaxQuality: false, ...payload })
-  }
-
-  await VideoJobInfoModel.decrease(video.uuid, 'pendingTranscode')
+async function handleHLSJob (job: Job, payload: HLSTranscodingPayload, video: MVideoFullLight) {
+  logger.info('Handling HLS transcoding job for %s.', video.uuid, lTags(video.uuid))
 
-  await retryTransactionWrapper(moveToNextState, { video, isNewVideo: payload.isNewVideo })
-}
+  const videoFileInput = payload.copyCodecs
+    ? video.getWebTorrentFile(payload.resolution)
+    : video.getMaxQualityFile()
 
-// ---------------------------------------------------------------------------
+  const videoOrStreamingPlaylist = videoFileInput.getVideoOrStreamingPlaylist()
 
-async function createHlsJobIfEnabled (user: MUserId, payload: {
-  videoUUID: string
-  resolution: number
-  hasAudio: boolean
-  copyCodecs: boolean
-  isMaxQuality: boolean
-  isNewVideo?: boolean
-}) {
-  if (!payload || CONFIG.TRANSCODING.ENABLED !== true || CONFIG.TRANSCODING.HLS.ENABLED !== true) return false
-
-  const jobOptions = {
-    priority: await getTranscodingJobPriority(user)
-  }
+  const inputFileMutexReleaser = await VideoPathManager.Instance.lockFiles(video.uuid)
 
-  const hlsTranscodingPayload: HLSTranscodingPayload = {
-    type: 'new-resolution-to-hls',
-    autoDeleteWebTorrentIfNeeded: true,
+  try {
+    await videoFileInput.getVideo().reload()
 
-    ...pick(payload, [ 'videoUUID', 'resolution', 'copyCodecs', 'isMaxQuality', 'isNewVideo', 'hasAudio' ])
+    await VideoPathManager.Instance.makeAvailableVideoFile(videoFileInput.withVideoOrPlaylist(videoOrStreamingPlaylist), videoInputPath => {
+      return generateHlsPlaylistResolution({
+        video,
+        videoInputPath,
+        inputFileMutexReleaser,
+        resolution: payload.resolution,
+        fps: payload.fps,
+        copyCodecs: payload.copyCodecs,
+        job
+      })
+    })
+  } finally {
+    inputFileMutexReleaser()
   }
 
-  await JobQueue.Instance.createJob(await buildTranscodingJob(hlsTranscodingPayload, jobOptions))
-
-  return true
-}
-
-async function createLowerResolutionsJobs (options: {
-  video: MVideoFullLight
-  user: MUserId
-  videoFileResolution: number
-  hasAudio: boolean
-  isNewVideo: boolean
-  type: 'hls' | 'webtorrent'
-}) {
-  const { video, user, videoFileResolution, isNewVideo, hasAudio, type } = options
-
-  // Create transcoding jobs if there are enabled resolutions
-  const resolutionsEnabled = await Hooks.wrapObject(
-    computeResolutionsToTranscode({ input: videoFileResolution, type: 'vod', includeInput: false, strictLower: true, hasAudio }),
-    'filter:transcoding.auto.resolutions-to-transcode.result',
-    options
-  )
-
-  const resolutionCreated: string[] = []
-
-  for (const resolution of resolutionsEnabled) {
-    let dataInput: VideoTranscodingPayload
-
-    if (CONFIG.TRANSCODING.WEBTORRENT.ENABLED && type === 'webtorrent') {
-      // WebTorrent will create subsequent HLS job
-      dataInput = {
-        type: 'new-resolution-to-webtorrent',
-        videoUUID: video.uuid,
-        resolution,
-        hasAudio,
-        createHLSIfNeeded: true,
-        isNewVideo
-      }
-
-      resolutionCreated.push('webtorrent-' + resolution)
-    }
-
-    if (CONFIG.TRANSCODING.HLS.ENABLED && type === 'hls') {
-      dataInput = {
-        type: 'new-resolution-to-hls',
-        videoUUID: video.uuid,
-        resolution,
-        hasAudio,
-        copyCodecs: false,
-        isMaxQuality: false,
-        autoDeleteWebTorrentIfNeeded: true,
-        isNewVideo
-      }
-
-      resolutionCreated.push('hls-' + resolution)
-    }
-
-    if (!dataInput) continue
-
-    const jobOptions = {
-      priority: await getTranscodingJobPriority(user)
-    }
-
-    await JobQueue.Instance.createJob(await buildTranscodingJob(dataInput, jobOptions))
-  }
+  logger.info('HLS transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 
-  if (resolutionCreated.length === 0) {
-    logger.info('No transcoding jobs created for video %s (no resolutions).', video.uuid, lTags(video.uuid))
+  if (payload.deleteWebTorrentFiles === true) {
+    logger.info('Removing WebTorrent files of %s now we have a HLS version of it.', video.uuid, lTags(video.uuid))
 
-    return false
+    await removeAllWebTorrentFiles(video)
   }
 
-  logger.info(
-    'New resolutions %s transcoding jobs created for video %s and origin file resolution of %d.', type, video.uuid, videoFileResolution,
-    { resolutionCreated, ...lTags(video.uuid) }
-  )
-
-  return true
+  await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: true, video })
 }