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

   1 import * as Bull from 'bull'
   2 import { VideoResolution, VideoState } from '../../../../shared'
   3 import { logger } from '../../../helpers/logger'
   4 import { VideoModel } from '../../../models/video/video'
   5 import { JobQueue } from '../job-queue'
   6 import { federateVideoIfNeeded } from '../../activitypub'
   7 import { retryTransactionWrapper } from '../../../helpers/database-utils'
   8 import { sequelizeTypescript } from '../../../initializers'
   9 import * as Bluebird from 'bluebird'
  10 import { computeResolutionsToTranscode } from '../../../helpers/ffmpeg-utils'
  11 import { generateHlsPlaylist, optimizeVideofile, transcodeOriginalVideofile, mergeAudioVideofile } from '../../video-transcoding'
  12 import { Notifier } from '../../notifier'
  13 import { CONFIG } from '../../../initializers/config'
  14
  15 interface BaseTranscodingPayload {
  16   videoUUID: string
  17   isNewVideo?: boolean
  18 }
  19
  20 interface HLSTranscodingPayload extends BaseTranscodingPayload {
  21   type: 'hls'
  22   isPortraitMode?: boolean
  23   resolution: VideoResolution
  24 }
  25
  26 interface NewResolutionTranscodingPayload extends BaseTranscodingPayload {
  27   type: 'new-resolution'
  28   isPortraitMode?: boolean
  29   resolution: VideoResolution
  30 }
  31
  32 interface MergeAudioTranscodingPayload extends BaseTranscodingPayload {
  33   type: 'merge-audio'
  34   resolution: VideoResolution
  35 }
  36
  37 interface OptimizeTranscodingPayload extends BaseTranscodingPayload {
  38   type: 'optimize'
  39 }
  40
  41 export type VideoTranscodingPayload = HLSTranscodingPayload | NewResolutionTranscodingPayload
  42   | OptimizeTranscodingPayload | MergeAudioTranscodingPayload
  43
  44 async function processVideoTranscoding (job: Bull.Job) {
  45   const payload = job.data as VideoTranscodingPayload
  46   logger.info('Processing video file in job %d.', job.id)
  47
  48   const video = await VideoModel.loadAndPopulateAccountAndServerAndTags(payload.videoUUID)
  49   // No video, maybe deleted?
  50   if (!video) {
  51     logger.info('Do not process job %d, video does not exist.', job.id)
  52     return undefined
  53   }
  54
  55   if (payload.type === 'hls') {
  56     await generateHlsPlaylist(video, payload.resolution, payload.isPortraitMode || false)
  57
  58     await retryTransactionWrapper(onHlsPlaylistGenerationSuccess, video)
  59   } else if (payload.type === 'new-resolution') {
  60     await transcodeOriginalVideofile(video, payload.resolution, payload.isPortraitMode || false)
  61
  62     await retryTransactionWrapper(publishNewResolutionIfNeeded, video, payload)
  63   } else if (payload.type === 'merge-audio') {
  64     await mergeAudioVideofile(video, payload.resolution)
  65
  66     await retryTransactionWrapper(publishNewResolutionIfNeeded, video, payload)
  67   } else {
  68     await optimizeVideofile(video)
  69
  70     await retryTransactionWrapper(onVideoFileOptimizerSuccess, video, payload)
  71   }
  72
  73   return video
  74 }
  75
  76 async function onHlsPlaylistGenerationSuccess (video: VideoModel) {
  77   if (video === undefined) return undefined
  78
  79   await sequelizeTypescript.transaction(async t => {
  80     // Maybe the video changed in database, refresh it
  81     let videoDatabase = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.uuid, t)
  82     // Video does not exist anymore
  83     if (!videoDatabase) return undefined
  84
  85     // If the video was not published, we consider it is a new one for other instances
  86     await federateVideoIfNeeded(videoDatabase, false, t)
  87   })
  88 }
  89
  90 async function publishNewResolutionIfNeeded (video: VideoModel, payload?: NewResolutionTranscodingPayload | MergeAudioTranscodingPayload) {
  91   const { videoDatabase, videoPublished } = await sequelizeTypescript.transaction(async t => {
  92     // Maybe the video changed in database, refresh it
  93     let videoDatabase = await VideoModel.loadAndPopulateAccountAndServerAndTags(video.uuid, t)
  94     // Video does not exist anymore
  95     if (!videoDatabase) return undefined
  96
  97     let videoPublished = false
  98
  99     // We transcoded the video file in another format, now we can publish it
 100     if (videoDatabase.state !== VideoState.PUBLISHED) {
 101       videoPublished = true
 102
 103       videoDatabase.state = VideoState.PUBLISHED
 104       videoDatabase.publishedAt = new Date()
 105       videoDatabase = await videoDatabase.save({ transaction: t })
 106     }
 107
 108     // If the video was not published, we consider it is a new one for other instances
 109     await federateVideoIfNeeded(videoDatabase, videoPublished, t)
 110
 111     return { videoDatabase, videoPublished }
 112   })
 113
 114   if (videoPublished) {
 115     Notifier.Instance.notifyOnNewVideoIfNeeded(videoDatabase)
 116     Notifier.Instance.notifyOnVideoPublishedAfterTranscoding(videoDatabase)
 117   }
 118
 119   await createHlsJobIfEnabled(payload)
 120 }
 121
 122 async function onVideoFileOptimizerSuccess (videoArg: VideoModel, payload: OptimizeTranscodingPayload) {
 123   if (videoArg === undefined) return undefined
 124
 125   // Outside the transaction (IO on disk)
 126   const { videoFileResolution } = await videoArg.getOriginalFileResolution()
 127
 128   const { videoDatabase, videoPublished } = await sequelizeTypescript.transaction(async t => {
 129     // Maybe the video changed in database, refresh it
 130     let videoDatabase = await VideoModel.loadAndPopulateAccountAndServerAndTags(videoArg.uuid, t)
 131     // Video does not exist anymore
 132     if (!videoDatabase) return undefined
 133
 134     // Create transcoding jobs if there are enabled resolutions
 135     const resolutionsEnabled = computeResolutionsToTranscode(videoFileResolution)
 136     logger.info(
 137       'Resolutions computed for video %s and origin file height of %d.', videoDatabase.uuid, videoFileResolution,
 138       { resolutions: resolutionsEnabled }
 139     )
 140
 141     let videoPublished = false
 142
 143     if (resolutionsEnabled.length !== 0) {
 144       const tasks: (Bluebird<Bull.Job<any>> | Promise<Bull.Job<any>>)[] = []
 145
 146       for (const resolution of resolutionsEnabled) {
 147         const dataInput = {
 148           type: 'new-resolution' as 'new-resolution',
 149           videoUUID: videoDatabase.uuid,
 150           resolution
 151         }
 152
 153         const p = JobQueue.Instance.createJob({ type: 'video-transcoding', payload: dataInput })
 154         tasks.push(p)
 155       }
 156
 157       await Promise.all(tasks)
 158
 159       logger.info('Transcoding jobs created for uuid %s.', videoDatabase.uuid, { resolutionsEnabled })
 160     } else {
 161       videoPublished = true
 162
 163       // No transcoding to do, it's now published
 164       videoDatabase.state = VideoState.PUBLISHED
 165       videoDatabase = await videoDatabase.save({ transaction: t })
 166
 167       logger.info('No transcoding jobs created for video %s (no resolutions).', videoDatabase.uuid, { privacy: videoDatabase.privacy })
 168     }
 169
 170     await federateVideoIfNeeded(videoDatabase, payload.isNewVideo, t)
 171
 172     return { videoDatabase, videoPublished }
 173   })
 174
 175   if (payload.isNewVideo) Notifier.Instance.notifyOnNewVideoIfNeeded(videoDatabase)
 176   if (videoPublished) Notifier.Instance.notifyOnVideoPublishedAfterTranscoding(videoDatabase)
 177
 178   const hlsPayload = Object.assign({}, payload, { resolution: videoDatabase.getOriginalFile().resolution })
 179   await createHlsJobIfEnabled(hlsPayload)
 180 }
 181
 182 // ---------------------------------------------------------------------------
 183
 184 export {
 185   processVideoTranscoding,
 186   publishNewResolutionIfNeeded
 187 }
 188
 189 // ---------------------------------------------------------------------------
 190
 191 function createHlsJobIfEnabled (payload?: { videoUUID: string, resolution: number, isPortraitMode?: boolean }) {
 192   // Generate HLS playlist?
 193   if (payload && CONFIG.TRANSCODING.HLS.ENABLED) {
 194     const hlsTranscodingPayload = {
 195       type: 'hls' as 'hls',
 196       videoUUID: payload.videoUUID,
 197       resolution: payload.resolution,
 198       isPortraitMode: payload.isPortraitMode
 199     }
 200
 201     return JobQueue.Instance.createJob({ type: 'video-transcoding', payload: hlsTranscodingPayload })
 202   }
 203 }