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

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