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

   1 import * as Bull from 'bull'
   2 import { VideoResolution, VideoState } from '../../../../shared'
   3 import { logger } from '../../../helpers/logger'
   4 import { computeResolutionsToTranscode } from '../../../helpers/utils'
   5 import { VideoModel } from '../../../models/video/video'
   6 import { JobQueue } from '../job-queue'
   7 import { federateVideoIfNeeded } from '../../activitypub'
   8 import { retryTransactionWrapper } from '../../../helpers/database-utils'
   9 import { sequelizeTypescript } from '../../../initializers'
  10 import * as Bluebird from 'bluebird'
  11
  12 export type VideoFilePayload = {
  13   videoUUID: string
  14   isNewVideo?: boolean
  15   resolution?: VideoResolution
  16   isPortraitMode?: boolean
  17 }
  18
  19 export type VideoFileImportPayload = {
  20   videoUUID: string,
  21   filePath: string
  22 }
  23
  24 async function processVideoFileImport (job: Bull.Job) {
  25   const payload = job.data as VideoFileImportPayload
  26   logger.info('Processing video file import in job %d.', job.id)
  27
  28   const video = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(payload.videoUUID)
  29   // No video, maybe deleted?
  30   if (!video) {
  31     logger.info('Do not process job %d, video does not exist.', job.id)
  32     return undefined
  33   }
  34
  35   await video.importVideoFile(payload.filePath)
  36
  37   await onVideoFileTranscoderOrImportSuccess(video)
  38   return video
  39 }
  40
  41 async function processVideoFile (job: Bull.Job) {
  42   const payload = job.data as VideoFilePayload
  43   logger.info('Processing video file in job %d.', job.id)
  44
  45   const video = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(payload.videoUUID)
  46   // No video, maybe deleted?
  47   if (!video) {
  48     logger.info('Do not process job %d, video does not exist.', job.id)
  49     return undefined
  50   }
  51
  52   // Transcoding in other resolution
  53   if (payload.resolution) {
  54     await video.transcodeOriginalVideofile(payload.resolution, payload.isPortraitMode || false)
  55
  56     await retryTransactionWrapper(onVideoFileTranscoderOrImportSuccess, video)
  57   } else {
  58     await video.optimizeOriginalVideofile()
  59
  60     await retryTransactionWrapper(onVideoFileOptimizerSuccess, video, payload.isNewVideo)
  61   }
  62
  63   return video
  64 }
  65
  66 async function onVideoFileTranscoderOrImportSuccess (video: VideoModel) {
  67   if (video === undefined) return undefined
  68
  69   return sequelizeTypescript.transaction(async t => {
  70     // Maybe the video changed in database, refresh it
  71     let videoDatabase = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(video.uuid, t)
  72     // Video does not exist anymore
  73     if (!videoDatabase) return undefined
  74
  75     let isNewVideo = false
  76
  77     // We transcoded the video file in another format, now we can publish it
  78     if (videoDatabase.state !== VideoState.PUBLISHED) {
  79       isNewVideo = true
  80
  81       videoDatabase.state = VideoState.PUBLISHED
  82       videoDatabase.publishedAt = new Date()
  83       videoDatabase = await videoDatabase.save({ transaction: t })
  84     }
  85
  86     // If the video was not published, we consider it is a new one for other instances
  87     await federateVideoIfNeeded(videoDatabase, isNewVideo, t)
  88
  89     return undefined
  90   })
  91 }
  92
  93 async function onVideoFileOptimizerSuccess (video: VideoModel, isNewVideo: boolean) {
  94   if (video === undefined) return undefined
  95
  96   // Outside the transaction (IO on disk)
  97   const { videoFileResolution } = await video.getOriginalFileResolution()
  98
  99   return sequelizeTypescript.transaction(async t => {
 100     // Maybe the video changed in database, refresh it
 101     const videoDatabase = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(video.uuid, t)
 102     // Video does not exist anymore
 103     if (!videoDatabase) return undefined
 104
 105     // Create transcoding jobs if there are enabled resolutions
 106     const resolutionsEnabled = computeResolutionsToTranscode(videoFileResolution)
 107     logger.info(
 108       'Resolutions computed for video %s and origin file height of %d.', videoDatabase.uuid, videoFileResolution,
 109       { resolutions: resolutionsEnabled }
 110     )
 111
 112     if (resolutionsEnabled.length !== 0) {
 113       const tasks: Bluebird<any>[] = []
 114
 115       for (const resolution of resolutionsEnabled) {
 116         const dataInput = {
 117           videoUUID: videoDatabase.uuid,
 118           resolution
 119         }
 120
 121         const p = JobQueue.Instance.createJob({ type: 'video-file', payload: dataInput })
 122         tasks.push(p)
 123       }
 124
 125       await Promise.all(tasks)
 126
 127       logger.info('Transcoding jobs created for uuid %s.', videoDatabase.uuid, { resolutionsEnabled })
 128     } else {
 129       // No transcoding to do, it's now published
 130       video.state = VideoState.PUBLISHED
 131       video = await video.save({ transaction: t })
 132
 133       logger.info('No transcoding jobs created for video %s (no resolutions).', video.uuid)
 134     }
 135
 136     return federateVideoIfNeeded(video, isNewVideo, t)
 137   })
 138 }
 139
 140 // ---------------------------------------------------------------------------
 141
 142 export {
 143   processVideoFile,
 144   processVideoFileImport
 145 }