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

   1 import { Job } from 'bull'
   2 import { TranscodeOptionsType } from '@server/helpers/ffmpeg-utils'
   3 import { addTranscodingJob, getTranscodingJobPriority } from '@server/lib/video'
   4 import { VideoPathManager } from '@server/lib/video-path-manager'
   5 import { moveToFailedTranscodingState, moveToNextState } from '@server/lib/video-state'
   6 import { UserModel } from '@server/models/user/user'
   7 import { VideoJobInfoModel } from '@server/models/video/video-job-info'
   8 import { MUser, MUserId, MVideo, MVideoFullLight, MVideoWithFile } from '@server/types/models'
   9 import { pick } from '@shared/core-utils'
  10 import {
  11   HLSTranscodingPayload,
  12   MergeAudioTranscodingPayload,
  13   NewResolutionTranscodingPayload,
  14   OptimizeTranscodingPayload,
  15   VideoResolution,
  16   VideoTranscodingPayload
  17 } from '@shared/models'
  18 import { retryTransactionWrapper } from '../../../helpers/database-utils'
  19 import { computeLowerResolutionsToTranscode } from '../../../helpers/ffprobe-utils'
  20 import { logger, loggerTagsFactory } from '../../../helpers/logger'
  21 import { CONFIG } from '../../../initializers/config'
  22 import { VideoModel } from '../../../models/video/video'
  23 import {
  24   generateHlsPlaylistResolution,
  25   mergeAudioVideofile,
  26   optimizeOriginalVideofile,
  27   transcodeNewWebTorrentResolution
  28 } from '../../transcoding/video-transcoding'
  29
  30 type HandlerFunction = (job: Job, payload: VideoTranscodingPayload, video: MVideoFullLight, user: MUser) => Promise<void>
  31
  32 const handlers: { [ id in VideoTranscodingPayload['type'] ]: HandlerFunction } = {
  33   'new-resolution-to-hls': handleHLSJob,
  34   'new-resolution-to-webtorrent': handleNewWebTorrentResolutionJob,
  35   'merge-audio-to-webtorrent': handleWebTorrentMergeAudioJob,
  36   'optimize-to-webtorrent': handleWebTorrentOptimizeJob
  37 }
  38
  39 const lTags = loggerTagsFactory('transcoding')
  40
  41 async function processVideoTranscoding (job: Job) {
  42   const payload = job.data as VideoTranscodingPayload
  43   logger.info('Processing transcoding job %d.', job.id, lTags(payload.videoUUID))
  44
  45   const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(payload.videoUUID)
  46   // No video, maybe deleted?
  47   if (!video) {
  48     logger.info('Do not process job %d, video does not exist.', job.id, lTags(payload.videoUUID))
  49     return undefined
  50   }
  51
  52   const user = await UserModel.loadByChannelActorId(video.VideoChannel.actorId)
  53
  54   const handler = handlers[payload.type]
  55
  56   if (!handler) {
  57     await moveToFailedTranscodingState(video)
  58     await VideoJobInfoModel.decrease(video.uuid, 'pendingTranscode')
  59
  60     throw new Error('Cannot find transcoding handler for ' + payload.type)
  61   }
  62
  63   try {
  64     await handler(job, payload, video, user)
  65   } catch (error) {
  66     await moveToFailedTranscodingState(video)
  67
  68     await VideoJobInfoModel.decrease(video.uuid, 'pendingTranscode')
  69
  70     throw error
  71   }
  72
  73   return video
  74 }
  75
  76 // ---------------------------------------------------------------------------
  77
  78 export {
  79   processVideoTranscoding
  80 }
  81
  82 // ---------------------------------------------------------------------------
  83 // Job handlers
  84 // ---------------------------------------------------------------------------
  85
  86 async function handleHLSJob (job: Job, payload: HLSTranscodingPayload, video: MVideoFullLight, user: MUser) {
  87   logger.info('Handling HLS transcoding job for %s.', video.uuid, lTags(video.uuid))
  88
  89   const videoFileInput = payload.copyCodecs
  90     ? video.getWebTorrentFile(payload.resolution)
  91     : video.getMaxQualityFile()
  92
  93   const videoOrStreamingPlaylist = videoFileInput.getVideoOrStreamingPlaylist()
  94
  95   await VideoPathManager.Instance.makeAvailableVideoFile(videoFileInput.withVideoOrPlaylist(videoOrStreamingPlaylist), videoInputPath => {
  96     return generateHlsPlaylistResolution({
  97       video,
  98       videoInputPath,
  99       resolution: payload.resolution,
 100       copyCodecs: payload.copyCodecs,
 101       isPortraitMode: payload.isPortraitMode || false,
 102       job
 103     })
 104   })
 105
 106   logger.info('HLS transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 107
 108   await onHlsPlaylistGeneration(video, user, payload)
 109 }
 110
 111 async function handleNewWebTorrentResolutionJob (
 112   job: Job,
 113   payload: NewResolutionTranscodingPayload,
 114   video: MVideoFullLight,
 115   user: MUserId
 116 ) {
 117   logger.info('Handling WebTorrent transcoding job for %s.', video.uuid, lTags(video.uuid))
 118
 119   await transcodeNewWebTorrentResolution(video, payload.resolution, payload.isPortraitMode || false, job)
 120
 121   logger.info('WebTorrent transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 122
 123   await onNewWebTorrentFileResolution(video, user, payload)
 124 }
 125
 126 async function handleWebTorrentMergeAudioJob (job: Job, payload: MergeAudioTranscodingPayload, video: MVideoFullLight, user: MUserId) {
 127   logger.info('Handling merge audio transcoding job for %s.', video.uuid, lTags(video.uuid))
 128
 129   await mergeAudioVideofile(video, payload.resolution, job)
 130
 131   logger.info('Merge audio transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 132
 133   await onVideoFirstWebTorrentTranscoding(video, payload, 'video', user)
 134 }
 135
 136 async function handleWebTorrentOptimizeJob (job: Job, payload: OptimizeTranscodingPayload, video: MVideoFullLight, user: MUserId) {
 137   logger.info('Handling optimize transcoding job for %s.', video.uuid, lTags(video.uuid))
 138
 139   const { transcodeType } = await optimizeOriginalVideofile(video, video.getMaxQualityFile(), job)
 140
 141   logger.info('Optimize transcoding job for %s ended.', video.uuid, lTags(video.uuid))
 142
 143   await onVideoFirstWebTorrentTranscoding(video, payload, transcodeType, user)
 144 }
 145
 146 // ---------------------------------------------------------------------------
 147
 148 async function onHlsPlaylistGeneration (video: MVideoFullLight, user: MUser, payload: HLSTranscodingPayload) {
 149   if (payload.isMaxQuality && payload.autoDeleteWebTorrentIfNeeded && CONFIG.TRANSCODING.WEBTORRENT.ENABLED === false) {
 150     // Remove webtorrent files if not enabled
 151     for (const file of video.VideoFiles) {
 152       await video.removeWebTorrentFileAndTorrent(file)
 153       await file.destroy()
 154     }
 155
 156     video.VideoFiles = []
 157
 158     // Create HLS new resolution jobs
 159     await createLowerResolutionsJobs({
 160       video,
 161       user,
 162       videoFileResolution: payload.resolution,
 163       isPortraitMode: payload.isPortraitMode,
 164       hasAudio: payload.hasAudio,
 165       isNewVideo: payload.isNewVideo ?? true,
 166       type: 'hls'
 167     })
 168   }
 169
 170   await VideoJobInfoModel.decrease(video.uuid, 'pendingTranscode')
 171   await retryTransactionWrapper(moveToNextState, video, payload.isNewVideo)
 172 }
 173
 174 async function onVideoFirstWebTorrentTranscoding (
 175   videoArg: MVideoWithFile,
 176   payload: OptimizeTranscodingPayload | MergeAudioTranscodingPayload,
 177   transcodeType: TranscodeOptionsType,
 178   user: MUserId
 179 ) {
 180   const { resolution, isPortraitMode, audioStream } = await videoArg.getMaxQualityFileInfo()
 181
 182   // Maybe the video changed in database, refresh it
 183   const videoDatabase = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoArg.uuid)
 184   // Video does not exist anymore
 185   if (!videoDatabase) return undefined
 186
 187   // Generate HLS version of the original file
 188   const originalFileHLSPayload = {
 189     ...payload,
 190
 191     isPortraitMode,
 192     hasAudio: !!audioStream,
 193     resolution: videoDatabase.getMaxQualityFile().resolution,
 194     // If we quick transcoded original file, force transcoding for HLS to avoid some weird playback issues
 195     copyCodecs: transcodeType !== 'quick-transcode',
 196     isMaxQuality: true
 197   }
 198   const hasHls = await createHlsJobIfEnabled(user, originalFileHLSPayload)
 199   const hasNewResolutions = await createLowerResolutionsJobs({
 200     video: videoDatabase,
 201     user,
 202     videoFileResolution: resolution,
 203     hasAudio: !!audioStream,
 204     isPortraitMode,
 205     type: 'webtorrent',
 206     isNewVideo: payload.isNewVideo ?? true
 207   })
 208
 209   await VideoJobInfoModel.decrease(videoDatabase.uuid, 'pendingTranscode')
 210
 211   // Move to next state if there are no other resolutions to generate
 212   if (!hasHls && !hasNewResolutions) {
 213     await retryTransactionWrapper(moveToNextState, videoDatabase, payload.isNewVideo)
 214   }
 215 }
 216
 217 async function onNewWebTorrentFileResolution (
 218   video: MVideo,
 219   user: MUserId,
 220   payload: NewResolutionTranscodingPayload | MergeAudioTranscodingPayload
 221 ) {
 222   await createHlsJobIfEnabled(user, { hasAudio: true, copyCodecs: true, isMaxQuality: false, ...payload })
 223   await VideoJobInfoModel.decrease(video.uuid, 'pendingTranscode')
 224
 225   await retryTransactionWrapper(moveToNextState, video, payload.isNewVideo)
 226 }
 227
 228 // ---------------------------------------------------------------------------
 229
 230 async function createHlsJobIfEnabled (user: MUserId, payload: {
 231   videoUUID: string
 232   resolution: number
 233   hasAudio: boolean
 234   isPortraitMode?: boolean
 235   copyCodecs: boolean
 236   isMaxQuality: boolean
 237   isNewVideo?: boolean
 238 }) {
 239   if (!payload || CONFIG.TRANSCODING.ENABLED !== true || CONFIG.TRANSCODING.HLS.ENABLED !== true) return false
 240
 241   const jobOptions = {
 242     priority: await getTranscodingJobPriority(user)
 243   }
 244
 245   const hlsTranscodingPayload: HLSTranscodingPayload = {
 246     type: 'new-resolution-to-hls',
 247     autoDeleteWebTorrentIfNeeded: true,
 248
 249     ...pick(payload, [ 'videoUUID', 'resolution', 'isPortraitMode', 'copyCodecs', 'isMaxQuality', 'isNewVideo', 'hasAudio' ])
 250   }
 251
 252   await addTranscodingJob(hlsTranscodingPayload, jobOptions)
 253
 254   return true
 255 }
 256
 257 async function createLowerResolutionsJobs (options: {
 258   video: MVideoFullLight
 259   user: MUserId
 260   videoFileResolution: number
 261   isPortraitMode: boolean
 262   hasAudio: boolean
 263   isNewVideo: boolean
 264   type: 'hls' | 'webtorrent'
 265 }) {
 266   const { video, user, videoFileResolution, isPortraitMode, isNewVideo, hasAudio, type } = options
 267
 268   // Create transcoding jobs if there are enabled resolutions
 269   const resolutionsEnabled = computeLowerResolutionsToTranscode(videoFileResolution, 'vod')
 270   const resolutionCreated: string[] = []
 271
 272   for (const resolution of resolutionsEnabled) {
 273     if (resolution === VideoResolution.H_NOVIDEO && hasAudio === false) continue
 274
 275     let dataInput: VideoTranscodingPayload
 276
 277     if (CONFIG.TRANSCODING.WEBTORRENT.ENABLED && type === 'webtorrent') {
 278       // WebTorrent will create subsequent HLS job
 279       dataInput = {
 280         type: 'new-resolution-to-webtorrent',
 281         videoUUID: video.uuid,
 282         resolution,
 283         isPortraitMode,
 284         hasAudio,
 285         isNewVideo
 286       }
 287
 288       resolutionCreated.push('webtorrent-' + resolution)
 289     }
 290
 291     if (CONFIG.TRANSCODING.HLS.ENABLED && type === 'hls') {
 292       dataInput = {
 293         type: 'new-resolution-to-hls',
 294         videoUUID: video.uuid,
 295         resolution,
 296         isPortraitMode,
 297         hasAudio,
 298         copyCodecs: false,
 299         isMaxQuality: false,
 300         autoDeleteWebTorrentIfNeeded: true,
 301         isNewVideo
 302       }
 303
 304       resolutionCreated.push('hls-' + resolution)
 305     }
 306
 307     if (!dataInput) continue
 308
 309     const jobOptions = {
 310       priority: await getTranscodingJobPriority(user)
 311     }
 312
 313     await addTranscodingJob(dataInput, jobOptions)
 314   }
 315
 316   if (resolutionCreated.length === 0) {
 317     logger.info('No transcoding jobs created for video %s (no resolutions).', video.uuid, lTags(video.uuid))
 318
 319     return false
 320   }
 321
 322   logger.info(
 323     'New resolutions %s transcoding jobs created for video %s and origin file resolution of %d.', type, video.uuid, videoFileResolution,
 324     { resolutionCreated, ...lTags(video.uuid) }
 325   )
 326
 327   return true
 328 }