Support studio transcoding in peertube runner

9 files changed, 177 insertions, 16 deletions

diff --git a/server/lib/runners/job-handlers/abstract-job-handler.ts b/server/lib/runners/job-handlers/abstract-job-handler.ts
index 74b455107..76fd1c5ac 100644
--- a/server/lib/runners/job-handlers/abstract-job-handler.ts
+++ b/server/lib/runners/job-handlers/abstract-job-handler.ts
@@ -1,3 +1,4 @@
+import { throttle } from 'lodash'
 import { retryTransactionWrapper } from '@server/helpers/database-utils'
 import { logger, loggerTagsFactory } from '@server/helpers/logger'
 import { RUNNER_JOBS } from '@server/initializers/constants'
@@ -14,6 +15,8 @@ import {
   RunnerJobSuccessPayload,
   RunnerJobType,
   RunnerJobUpdatePayload,
+  RunnerJobVideoEditionTranscodingPayload,
+  RunnerJobVideoEditionTranscodingPrivatePayload,
   RunnerJobVODAudioMergeTranscodingPayload,
   RunnerJobVODAudioMergeTranscodingPrivatePayload,
   RunnerJobVODHLSTranscodingPayload,
@@ -21,7 +24,6 @@ import {
   RunnerJobVODWebVideoTranscodingPayload,
   RunnerJobVODWebVideoTranscodingPrivatePayload
 } from '@shared/models'
-import { throttle } from 'lodash'
 
 type CreateRunnerJobArg =
   {
@@ -43,6 +45,11 @@ type CreateRunnerJobArg =
     type: Extract<RunnerJobType, 'live-rtmp-hls-transcoding'>
     payload: RunnerJobLiveRTMPHLSTranscodingPayload
     privatePayload: RunnerJobLiveRTMPHLSTranscodingPrivatePayload
+  } |
+  {
+    type: Extract<RunnerJobType, 'video-edition-transcoding'>
+    payload: RunnerJobVideoEditionTranscodingPayload
+    privatePayload: RunnerJobVideoEditionTranscodingPrivatePayload
   }
 
 export abstract class AbstractJobHandler <C, U extends RunnerJobUpdatePayload, S extends RunnerJobSuccessPayload> {
@@ -62,6 +69,8 @@ export abstract class AbstractJobHandler <C, U extends RunnerJobUpdatePayload, S
   }): Promise<MRunnerJob> {
     const { priority, dependsOnRunnerJob } = options
 
+    logger.debug('Creating runner job', { options, ...this.lTags(options.type) })
+
     const runnerJob = new RunnerJobModel({
       ...pick(options, [ 'type', 'payload', 'privatePayload' ]),
 
diff --git a/server/lib/runners/job-handlers/abstract-vod-transcoding-job-handler.ts b/server/lib/runners/job-handlers/abstract-vod-transcoding-job-handler.ts
index 517645848..a910ae383 100644
--- a/server/lib/runners/job-handlers/abstract-vod-transcoding-job-handler.ts
+++ b/server/lib/runners/job-handlers/abstract-vod-transcoding-job-handler.ts
@@ -4,27 +4,19 @@ import { logger } from '@server/helpers/logger'
 import { moveToFailedTranscodingState, moveToNextState } from '@server/lib/video-state'
 import { VideoJobInfoModel } from '@server/models/video/video-job-info'
 import { MRunnerJob } from '@server/types/models/runners'
-import {
-  LiveRTMPHLSTranscodingUpdatePayload,
-  RunnerJobSuccessPayload,
-  RunnerJobUpdatePayload,
-  RunnerJobVODPrivatePayload
-} from '@shared/models'
+import { RunnerJobSuccessPayload, RunnerJobUpdatePayload, RunnerJobVODPrivatePayload } from '@shared/models'
 import { AbstractJobHandler } from './abstract-job-handler'
 import { loadTranscodingRunnerVideo } from './shared'
 
 // eslint-disable-next-line max-len
 export abstract class AbstractVODTranscodingJobHandler <C, U extends RunnerJobUpdatePayload, S extends RunnerJobSuccessPayload> extends AbstractJobHandler<C, U, S> {
 
-  // ---------------------------------------------------------------------------
-
   protected isAbortSupported () {
     return true
   }
 
   protected specificUpdate (_options: {
     runnerJob: MRunnerJob
-    updatePayload?: LiveRTMPHLSTranscodingUpdatePayload
   }) {
     // empty
   }
diff --git a/server/lib/runners/job-handlers/index.ts b/server/lib/runners/job-handlers/index.ts
index 0fca72b9a..a40cee865 100644
--- a/server/lib/runners/job-handlers/index.ts
+++ b/server/lib/runners/job-handlers/index.ts
@@ -1,6 +1,7 @@
 export * from './abstract-job-handler'
 export * from './live-rtmp-hls-transcoding-job-handler'
+export * from './runner-job-handlers'
+export * from './video-edition-transcoding-job-handler'
 export * from './vod-audio-merge-transcoding-job-handler'
 export * from './vod-hls-transcoding-job-handler'
 export * from './vod-web-video-transcoding-job-handler'
-export * from './runner-job-handlers'
diff --git a/server/lib/runners/job-handlers/live-rtmp-hls-transcoding-job-handler.ts b/server/lib/runners/job-handlers/live-rtmp-hls-transcoding-job-handler.ts
index c3d0e427d..48a70d891 100644
--- a/server/lib/runners/job-handlers/live-rtmp-hls-transcoding-job-handler.ts
+++ b/server/lib/runners/job-handlers/live-rtmp-hls-transcoding-job-handler.ts
@@ -70,7 +70,7 @@ export class LiveRTMPHLSTranscodingJobHandler extends AbstractJobHandler<CreateO
 
   // ---------------------------------------------------------------------------
 
-  async specificUpdate (options: {
+  protected async specificUpdate (options: {
     runnerJob: MRunnerJob
     updatePayload: LiveRTMPHLSTranscodingUpdatePayload
   }) {
diff --git a/server/lib/runners/job-handlers/runner-job-handlers.ts b/server/lib/runners/job-handlers/runner-job-handlers.ts
index 7bad1bc77..4ea6684ea 100644
--- a/server/lib/runners/job-handlers/runner-job-handlers.ts
+++ b/server/lib/runners/job-handlers/runner-job-handlers.ts
@@ -2,6 +2,7 @@ import { MRunnerJob } from '@server/types/models/runners'
 import { RunnerJobSuccessPayload, RunnerJobType, RunnerJobUpdatePayload } from '@shared/models'
 import { AbstractJobHandler } from './abstract-job-handler'
 import { LiveRTMPHLSTranscodingJobHandler } from './live-rtmp-hls-transcoding-job-handler'
+import { VideoEditionTranscodingJobHandler } from './video-edition-transcoding-job-handler'
 import { VODAudioMergeTranscodingJobHandler } from './vod-audio-merge-transcoding-job-handler'
 import { VODHLSTranscodingJobHandler } from './vod-hls-transcoding-job-handler'
 import { VODWebVideoTranscodingJobHandler } from './vod-web-video-transcoding-job-handler'
@@ -10,7 +11,8 @@ const processors: Record<RunnerJobType, new() => AbstractJobHandler<unknown, Run
   'vod-web-video-transcoding': VODWebVideoTranscodingJobHandler,
   'vod-hls-transcoding': VODHLSTranscodingJobHandler,
   'vod-audio-merge-transcoding': VODAudioMergeTranscodingJobHandler,
-  'live-rtmp-hls-transcoding': LiveRTMPHLSTranscodingJobHandler
+  'live-rtmp-hls-transcoding': LiveRTMPHLSTranscodingJobHandler,
+  'video-edition-transcoding': VideoEditionTranscodingJobHandler
 }
 
 export function getRunnerJobHandlerClass (job: MRunnerJob) {
diff --git a/server/lib/runners/job-handlers/video-edition-transcoding-job-handler.ts b/server/lib/runners/job-handlers/video-edition-transcoding-job-handler.ts
new file mode 100644
index 000000000..39a755c48
--- /dev/null
+++ b/server/lib/runners/job-handlers/video-edition-transcoding-job-handler.ts
@@ -0,0 +1,157 @@
+
+import { basename } from 'path'
+import { logger } from '@server/helpers/logger'
+import { onVideoEditionEnded, safeCleanupStudioTMPFiles } from '@server/lib/video-studio'
+import { MVideo } from '@server/types/models'
+import { MRunnerJob } from '@server/types/models/runners'
+import { buildUUID } from '@shared/extra-utils'
+import {
+  isVideoStudioTaskIntro,
+  isVideoStudioTaskOutro,
+  isVideoStudioTaskWatermark,
+  RunnerJobState,
+  RunnerJobUpdatePayload,
+  RunnerJobVideoEditionTranscodingPayload,
+  RunnerJobVideoEditionTranscodingPrivatePayload,
+  VideoEditionTranscodingSuccess,
+  VideoState,
+  VideoStudioTaskPayload
+} from '@shared/models'
+import { generateRunnerEditionTranscodingVideoInputFileUrl, generateRunnerTranscodingVideoInputFileUrl } from '../runner-urls'
+import { AbstractJobHandler } from './abstract-job-handler'
+import { loadTranscodingRunnerVideo } from './shared'
+
+type CreateOptions = {
+  video: MVideo
+  tasks: VideoStudioTaskPayload[]
+  priority: number
+}
+
+// eslint-disable-next-line max-len
+export class VideoEditionTranscodingJobHandler extends AbstractJobHandler<CreateOptions, RunnerJobUpdatePayload, VideoEditionTranscodingSuccess> {
+
+  async create (options: CreateOptions) {
+    const { video, priority, tasks } = options
+
+    const jobUUID = buildUUID()
+    const payload: RunnerJobVideoEditionTranscodingPayload = {
+      input: {
+        videoFileUrl: generateRunnerTranscodingVideoInputFileUrl(jobUUID, video.uuid)
+      },
+      tasks: tasks.map(t => {
+        if (isVideoStudioTaskIntro(t) || isVideoStudioTaskOutro(t)) {
+          return {
+            ...t,
+
+            options: {
+              ...t.options,
+
+              file: generateRunnerEditionTranscodingVideoInputFileUrl(jobUUID, video.uuid, basename(t.options.file))
+            }
+          }
+        }
+
+        if (isVideoStudioTaskWatermark(t)) {
+          return {
+            ...t,
+
+            options: {
+              ...t.options,
+
+              file: generateRunnerEditionTranscodingVideoInputFileUrl(jobUUID, video.uuid, basename(t.options.file))
+            }
+          }
+        }
+
+        return t
+      })
+    }
+
+    const privatePayload: RunnerJobVideoEditionTranscodingPrivatePayload = {
+      videoUUID: video.uuid,
+      originalTasks: tasks
+    }
+
+    const job = await this.createRunnerJob({
+      type: 'video-edition-transcoding',
+      jobUUID,
+      payload,
+      privatePayload,
+      priority
+    })
+
+    return job
+  }
+
+  // ---------------------------------------------------------------------------
+
+  protected isAbortSupported () {
+    return true
+  }
+
+  protected specificUpdate (_options: {
+    runnerJob: MRunnerJob
+  }) {
+    // empty
+  }
+
+  protected specificAbort (_options: {
+    runnerJob: MRunnerJob
+  }) {
+    // empty
+  }
+
+  protected async specificComplete (options: {
+    runnerJob: MRunnerJob
+    resultPayload: VideoEditionTranscodingSuccess
+  }) {
+    const { runnerJob, resultPayload } = options
+    const privatePayload = runnerJob.privatePayload as RunnerJobVideoEditionTranscodingPrivatePayload
+
+    const video = await loadTranscodingRunnerVideo(runnerJob, this.lTags)
+    if (!video) {
+      await safeCleanupStudioTMPFiles(privatePayload.originalTasks)
+
+    }
+
+    const videoFilePath = resultPayload.videoFile as string
+
+    await onVideoEditionEnded({ video, editionResultPath: videoFilePath, tasks: privatePayload.originalTasks })
+
+    logger.info(
+      'Runner video edition transcoding job %s for %s ended.',
+      runnerJob.uuid, video.uuid, this.lTags(video.uuid, runnerJob.uuid)
+    )
+  }
+
+  protected specificError (options: {
+    runnerJob: MRunnerJob
+    nextState: RunnerJobState
+  }) {
+    if (options.nextState === RunnerJobState.ERRORED) {
+      return this.specificErrorOrCancel(options)
+    }
+
+    return Promise.resolve()
+  }
+
+  protected specificCancel (options: {
+    runnerJob: MRunnerJob
+  }) {
+    return this.specificErrorOrCancel(options)
+  }
+
+  private async specificErrorOrCancel (options: {
+    runnerJob: MRunnerJob
+  }) {
+    const { runnerJob } = options
+
+    const payload = runnerJob.privatePayload as RunnerJobVideoEditionTranscodingPrivatePayload
+    await safeCleanupStudioTMPFiles(payload.originalTasks)
+
+    const video = await loadTranscodingRunnerVideo(options.runnerJob, this.lTags)
+    if (!video) return
+
+    return video.setNewState(VideoState.PUBLISHED, false, undefined)
+  }
+}
diff --git a/server/lib/runners/job-handlers/vod-audio-merge-transcoding-job-handler.ts b/server/lib/runners/job-handlers/vod-audio-merge-transcoding-job-handler.ts
index a7b33f87e..5f247d792 100644
--- a/server/lib/runners/job-handlers/vod-audio-merge-transcoding-job-handler.ts
+++ b/server/lib/runners/job-handlers/vod-audio-merge-transcoding-job-handler.ts
@@ -64,7 +64,7 @@ export class VODAudioMergeTranscodingJobHandler extends AbstractVODTranscodingJo
 
   // ---------------------------------------------------------------------------
 
-  async specificComplete (options: {
+  protected async specificComplete (options: {
     runnerJob: MRunnerJob
     resultPayload: VODAudioMergeTranscodingSuccess
   }) {
diff --git a/server/lib/runners/job-handlers/vod-hls-transcoding-job-handler.ts b/server/lib/runners/job-handlers/vod-hls-transcoding-job-handler.ts
index 02566b9d5..cc94bcbda 100644
--- a/server/lib/runners/job-handlers/vod-hls-transcoding-job-handler.ts
+++ b/server/lib/runners/job-handlers/vod-hls-transcoding-job-handler.ts
@@ -71,7 +71,7 @@ export class VODHLSTranscodingJobHandler extends AbstractVODTranscodingJobHandle
 
   // ---------------------------------------------------------------------------
 
-  async specificComplete (options: {
+  protected async specificComplete (options: {
     runnerJob: MRunnerJob
     resultPayload: VODHLSTranscodingSuccess
   }) {
diff --git a/server/lib/runners/job-handlers/vod-web-video-transcoding-job-handler.ts b/server/lib/runners/job-handlers/vod-web-video-transcoding-job-handler.ts
index 57761a7a1..663d3306e 100644
--- a/server/lib/runners/job-handlers/vod-web-video-transcoding-job-handler.ts
+++ b/server/lib/runners/job-handlers/vod-web-video-transcoding-job-handler.ts
@@ -62,7 +62,7 @@ export class VODWebVideoTranscodingJobHandler extends AbstractVODTranscodingJobH
 
   // ---------------------------------------------------------------------------
 
-  async specificComplete (options: {
+  protected async specificComplete (options: {
     runnerJob: MRunnerJob
     resultPayload: VODWebVideoTranscodingSuccess
   }) {