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: true, 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: true, video })
  95 }
  96
  97 async function handleNewWebTorrentResolutionJob (job: Job, payload: NewWebTorrentResolutionTranscodingPayload, video: MVideoFullLight) {
  98   logger.info('Handling WebTorrent transcoding job for %s.', video.uuid, lTags(video.uuid))
  99
 100   await transcodeNewWebTorrentResolution({ video, resolution: payload.resolution, fps: payload.fps, job })
 101
 102   logger.info('WebTorrent transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 103
 104   await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: true, video })
 105 }
 106
 107 async function handleHLSJob (job: Job, payload: HLSTranscodingPayload, video: MVideoFullLight) {
 108   logger.info('Handling HLS transcoding job for %s.', video.uuid, lTags(video.uuid))
 109
 110   const videoFileInput = payload.copyCodecs
 111     ? video.getWebTorrentFile(payload.resolution)
 112     : video.getMaxQualityFile()
 113
 114   const videoOrStreamingPlaylist = videoFileInput.getVideoOrStreamingPlaylist()
 115
 116   const inputFileMutexReleaser = await VideoPathManager.Instance.lockFiles(video.uuid)
 117
 118   try {
 119     await videoFileInput.getVideo().reload()
 120
 121     await VideoPathManager.Instance.makeAvailableVideoFile(videoFileInput.withVideoOrPlaylist(videoOrStreamingPlaylist), videoInputPath => {
 122       return generateHlsPlaylistResolution({
 123         video,
 124         videoInputPath,
 125         inputFileMutexReleaser,
 126         resolution: payload.resolution,
 127         fps: payload.fps,
 128         copyCodecs: payload.copyCodecs,
 129         job
 130       })
 131     })
 132   } finally {
 133     inputFileMutexReleaser()
 134   }
 135
 136   logger.info('HLS transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 137
 138   if (payload.deleteWebTorrentFiles === true) {
 139     logger.info('Removing WebTorrent files of %s now we have a HLS version of it.', video.uuid, lTags(video.uuid))
 140
 141     await removeAllWebTorrentFiles(video)
 142   }
 143
 144   await onTranscodingEnded({ isNewVideo: payload.isNewVideo, moveVideoToNextState: true, video })
 145 }