server/lib/job-queue/handlers/video-transcoding.ts

   1 import { Job } from 'bullmq'
   2 import { onTranscodingEnded } from '@server/lib/transcoding/ended-transcoding'
   3 import { generateHlsPlaylistResolution } from '@server/lib/transcoding/hls-transcoding'
   4 import { mergeAudioVideofile, optimizeOriginalVideofile, transcodeNewWebTorrentResolution } from '@server/lib/transcoding/web-transcoding'
   5 import { removeAllWebTorrentFiles } from '@server/lib/video-file'
   6 import { VideoPathManager } from '@server/lib/video-path-manager'
   7 import { moveToFailedTranscodingState } from '@server/lib/video-state'
   8 import { UserModel } from '@server/models/user/user'
   9 import { VideoJobInfoModel } from '@server/models/video/video-job-info'
  10 import { MUser, MUserId, MVideoFullLight } from '@server/types/models'
  11 import {
  12   HLSTranscodingPayload,
  13   MergeAudioTranscodingPayload,
  14   NewWebTorrentResolutionTranscodingPayload,
  15   OptimizeTranscodingPayload,
  16   VideoTranscodingPayload
  17 } from '@shared/models'
  18 import { logger, loggerTagsFactory } from '../../../helpers/logger'
  19 import { VideoModel } from '../../../models/video/video'
  20
  21 type HandlerFunction = (job: Job, payload: VideoTranscodingPayload, video: MVideoFullLight, user: MUser) => Promise<void>
  22
  23 const handlers: { [ id in VideoTranscodingPayload['type'] ]: HandlerFunction } = {
  24   'new-resolution-to-hls': handleHLSJob,
  25   'new-resolution-to-webtorrent': handleNewWebTorrentResolutionJob,
  26   'merge-audio-to-webtorrent': handleWebTorrentMergeAudioJob,
  27   'optimize-to-webtorrent': handleWebTorrentOptimizeJob
  28 }
  29
  30 const lTags = loggerTagsFactory('transcoding')
  31
  32 async function processVideoTranscoding (job: Job) {
  33   const payload = job.data as VideoTranscodingPayload
  34   logger.info('Processing transcoding job %s.', job.id, lTags(payload.videoUUID))
  35
  36   const video = await VideoModel.loadFull(payload.videoUUID)
  37   // No video, maybe deleted?
  38   if (!video) {
  39     logger.info('Do not process job %d, video does not exist.', job.id, lTags(payload.videoUUID))
  40     return undefined
  41   }
  42
  43   const user = await UserModel.loadByChannelActorId(video.VideoChannel.actorId)
  44
  45   const handler = handlers[payload.type]
  46
  47   if (!handler) {
  48     await moveToFailedTranscodingState(video)
  49     await VideoJobInfoModel.decrease(video.uuid, 'pendingTranscode')
  50
  51     throw new Error('Cannot find transcoding handler for ' + payload.type)
  52   }
  53
  54   try {
  55     await handler(job, payload, video, user)
  56   } catch (error) {
  57     await moveToFailedTranscodingState(video)
  58
  59     await VideoJobInfoModel.decrease(video.uuid, 'pendingTranscode')
  60
  61     throw error
  62   }
  63
  64   return video
  65 }
  66
  67 // ---------------------------------------------------------------------------
  68
  69 export {
  70   processVideoTranscoding
  71 }
  72
  73 // ---------------------------------------------------------------------------
  74 // Job handlers
  75 // ---------------------------------------------------------------------------
  76
  77 async function handleWebTorrentMergeAudioJob (job: Job, payload: MergeAudioTranscodingPayload, video: MVideoFullLight, user: MUserId) {
  78   logger.info('Handling merge audio transcoding job for %s.', video.uuid, lTags(video.uuid))
  79
  80   await mergeAudioVideofile({ video, resolution: payload.resolution, fps: payload.fps, job })
  81
  82   logger.info('Merge audio transcoding job for %s ended.', video.uuid, lTags(video.uuid))
  83
  84   await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: !payload.hasChildren, video })
  85 }
  86
  87 async function handleWebTorrentOptimizeJob (job: Job, payload: OptimizeTranscodingPayload, video: MVideoFullLight, user: MUserId) {
  88   logger.info('Handling optimize transcoding job for %s.', video.uuid, lTags(video.uuid))
  89
  90   await optimizeOriginalVideofile({ video, inputVideoFile: video.getMaxQualityFile(), quickTranscode: payload.quickTranscode, job })
  91
  92   logger.info('Optimize transcoding job for %s ended.', video.uuid, lTags(video.uuid))
  93
  94   await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: !payload.hasChildren, video })
  95 }
  96
  97 // ---------------------------------------------------------------------------
  98
  99 async function handleNewWebTorrentResolutionJob (job: Job, payload: NewWebTorrentResolutionTranscodingPayload, video: MVideoFullLight) {
 100   logger.info('Handling WebTorrent transcoding job for %s.', video.uuid, lTags(video.uuid))
 101
 102   await transcodeNewWebTorrentResolution({ video, resolution: payload.resolution, fps: payload.fps, job })
 103
 104   logger.info('WebTorrent transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 105
 106   await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: true, video })
 107 }
 108
 109 // ---------------------------------------------------------------------------
 110
 111 async function handleHLSJob (job: Job, payload: HLSTranscodingPayload, videoArg: MVideoFullLight) {
 112   logger.info('Handling HLS transcoding job for %s.', videoArg.uuid, lTags(videoArg.uuid))
 113
 114   const inputFileMutexReleaser = await VideoPathManager.Instance.lockFiles(videoArg.uuid)
 115   let video: MVideoFullLight
 116
 117   try {
 118     video = await VideoModel.loadFull(videoArg.uuid)
 119
 120     const videoFileInput = payload.copyCodecs
 121       ? video.getWebTorrentFile(payload.resolution)
 122       : video.getMaxQualityFile()
 123
 124     const videoOrStreamingPlaylist = videoFileInput.getVideoOrStreamingPlaylist()
 125
 126     await VideoPathManager.Instance.makeAvailableVideoFile(videoFileInput.withVideoOrPlaylist(videoOrStreamingPlaylist), videoInputPath => {
 127       return generateHlsPlaylistResolution({
 128         video,
 129         videoInputPath,
 130         inputFileMutexReleaser,
 131         resolution: payload.resolution,
 132         fps: payload.fps,
 133         copyCodecs: payload.copyCodecs,
 134         job
 135       })
 136     })
 137   } finally {
 138     inputFileMutexReleaser()
 139   }
 140
 141   logger.info('HLS transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 142
 143   if (payload.deleteWebTorrentFiles === true) {
 144     logger.info('Removing WebTorrent files of %s now we have a HLS version of it.', video.uuid, lTags(video.uuid))
 145
 146     await removeAllWebTorrentFiles(video)
 147   }
 148
 149   await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: true, video })
 150 }