はじまりの大地

server/core/lib/runners/job-handlers/abstract-job-handler.ts

@@ -0,0 +1,280 @@
+import { pick } from '@peertube/peertube-core-utils'
+import {
+  RunnerJobLiveRTMPHLSTranscodingPayload,
+  RunnerJobLiveRTMPHLSTranscodingPrivatePayload,
+  RunnerJobState,
+  RunnerJobStateType,
+  RunnerJobStudioTranscodingPayload,
+  RunnerJobSuccessPayload,
+  RunnerJobTranscriptionPayload,
+  RunnerJobTranscriptionPrivatePayload,
+  RunnerJobType,
+  RunnerJobUpdatePayload,
+  RunnerJobVODAudioMergeTranscodingPayload,
+  RunnerJobVODAudioMergeTranscodingPrivatePayload,
+  RunnerJobVODHLSTranscodingPayload,
+  RunnerJobVODHLSTranscodingPrivatePayload,
+  RunnerJobVODWebVideoTranscodingPayload,
+  RunnerJobVODWebVideoTranscodingPrivatePayload,
+  RunnerJobVideoStudioTranscodingPrivatePayload
+} from '@peertube/peertube-models'
+import { saveInTransactionWithRetries } from '@server/helpers/database-utils.js'
+import { logger, loggerTagsFactory } from '@server/helpers/logger.js'
+import { RUNNER_JOBS } from '@server/initializers/constants.js'
+import { sequelizeTypescript } from '@server/initializers/database.js'
+import { PeerTubeSocket } from '@server/lib/peertube-socket.js'
+import { RunnerJobModel } from '@server/models/runner/runner-job.js'
+import { setAsUpdated } from '@server/models/shared/index.js'
+import { MRunnerJob } from '@server/types/models/runners/index.js'
+import throttle from 'lodash-es/throttle.js'
+
+type CreateRunnerJobArg =
+  {
+    type: Extract<RunnerJobType, 'vod-web-video-transcoding'>
+    payload: RunnerJobVODWebVideoTranscodingPayload
+    privatePayload: RunnerJobVODWebVideoTranscodingPrivatePayload
+  } |
+  {
+    type: Extract<RunnerJobType, 'vod-hls-transcoding'>
+    payload: RunnerJobVODHLSTranscodingPayload
+    privatePayload: RunnerJobVODHLSTranscodingPrivatePayload
+  } |
+  {
+    type: Extract<RunnerJobType, 'vod-audio-merge-transcoding'>
+    payload: RunnerJobVODAudioMergeTranscodingPayload
+    privatePayload: RunnerJobVODAudioMergeTranscodingPrivatePayload
+  } |
+  {
+    type: Extract<RunnerJobType, 'live-rtmp-hls-transcoding'>
+    payload: RunnerJobLiveRTMPHLSTranscodingPayload
+    privatePayload: RunnerJobLiveRTMPHLSTranscodingPrivatePayload
+  } |
+  {
+    type: Extract<RunnerJobType, 'video-studio-transcoding'>
+    payload: RunnerJobStudioTranscodingPayload
+    privatePayload: RunnerJobVideoStudioTranscodingPrivatePayload
+  } |
+  {
+    type: Extract<RunnerJobType, 'video-transcription'>
+    payload: RunnerJobTranscriptionPayload
+    privatePayload: RunnerJobTranscriptionPrivatePayload
+  }
+
+export abstract class AbstractJobHandler <C, U extends RunnerJobUpdatePayload, S extends RunnerJobSuccessPayload> {
+
+  protected readonly lTags = loggerTagsFactory('runner')
+
+  static setJobAsUpdatedThrottled = throttle(setAsUpdated, 2000)
+
+  // ---------------------------------------------------------------------------
+
+  abstract create (options: C): Promise<MRunnerJob>
+
+  protected async createRunnerJob (options: CreateRunnerJobArg & {
+    jobUUID: string
+    priority: number
+    dependsOnRunnerJob?: MRunnerJob
+  }): 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' ]),
+
+      uuid: options.jobUUID,
+
+      state: dependsOnRunnerJob
+        ? RunnerJobState.WAITING_FOR_PARENT_JOB
+        : RunnerJobState.PENDING,
+
+      dependsOnRunnerJobId: dependsOnRunnerJob?.id,
+
+      priority
+    })
+
+    await saveInTransactionWithRetries(runnerJob)
+
+    if (runnerJob.state === RunnerJobState.PENDING) {
+      PeerTubeSocket.Instance.sendAvailableJobsPingToRunners()
+    }
+
+    return runnerJob
+  }
+
+  // ---------------------------------------------------------------------------
+
+  protected abstract specificUpdate (options: {
+    runnerJob: MRunnerJob
+    updatePayload?: U
+  }): Promise<void> | void
+
+  async update (options: {
+    runnerJob: MRunnerJob
+    progress?: number
+    updatePayload?: U
+  }) {
+    const { runnerJob, progress } = options
+
+    await this.specificUpdate(options)
+
+    if (progress) runnerJob.progress = progress
+
+    if (!runnerJob.changed()) {
+      try {
+        await AbstractJobHandler.setJobAsUpdatedThrottled({ sequelize: sequelizeTypescript, table: 'runnerJob', id: runnerJob.id })
+      } catch (err) {
+        logger.warn('Cannot set remote job as updated', { err, ...this.lTags(runnerJob.id, runnerJob.type) })
+      }
+
+      return
+    }
+
+    await saveInTransactionWithRetries(runnerJob)
+  }
+
+  // ---------------------------------------------------------------------------
+
+  async complete (options: {
+    runnerJob: MRunnerJob
+    resultPayload: S
+  }) {
+    const { runnerJob } = options
+
+    runnerJob.state = RunnerJobState.COMPLETING
+    await saveInTransactionWithRetries(runnerJob)
+
+    try {
+      await this.specificComplete(options)
+
+      runnerJob.state = RunnerJobState.COMPLETED
+    } catch (err) {
+      logger.error('Cannot complete runner job', { err, ...this.lTags(runnerJob.id, runnerJob.type) })
+
+      runnerJob.state = RunnerJobState.ERRORED
+      runnerJob.error = err.message
+    }
+
+    runnerJob.progress = null
+    runnerJob.finishedAt = new Date()
+
+    await saveInTransactionWithRetries(runnerJob)
+
+    const [ affectedCount ] = await RunnerJobModel.updateDependantJobsOf(runnerJob)
+
+    if (affectedCount !== 0) PeerTubeSocket.Instance.sendAvailableJobsPingToRunners()
+  }
+
+  protected abstract specificComplete (options: {
+    runnerJob: MRunnerJob
+    resultPayload: S
+  }): Promise<void> | void
+
+  // ---------------------------------------------------------------------------
+
+  async cancel (options: {
+    runnerJob: MRunnerJob
+    fromParent?: boolean
+  }) {
+    const { runnerJob, fromParent } = options
+
+    await this.specificCancel(options)
+
+    const cancelState = fromParent
+      ? RunnerJobState.PARENT_CANCELLED
+      : RunnerJobState.CANCELLED
+
+    runnerJob.setToErrorOrCancel(cancelState)
+
+    await saveInTransactionWithRetries(runnerJob)
+
+    const children = await RunnerJobModel.listChildrenOf(runnerJob)
+    for (const child of children) {
+      logger.info(`Cancelling child job ${child.uuid} of ${runnerJob.uuid} because of parent cancel`, this.lTags(child.uuid))
+
+      await this.cancel({ runnerJob: child, fromParent: true })
+    }
+  }
+
+  protected abstract specificCancel (options: {
+    runnerJob: MRunnerJob
+  }): Promise<void> | void
+
+  // ---------------------------------------------------------------------------
+
+  protected abstract isAbortSupported (): boolean
+
+  async abort (options: {
+    runnerJob: MRunnerJob
+    abortNotSupportedErrorMessage?: string
+  }) {
+    const { runnerJob, abortNotSupportedErrorMessage = 'Job has been aborted but it is not supported by this job type' } = options
+
+    if (this.isAbortSupported() !== true) {
+      return this.error({ runnerJob, message: abortNotSupportedErrorMessage })
+    }
+
+    await this.specificAbort(options)
+
+    runnerJob.resetToPending()
+
+    await saveInTransactionWithRetries(runnerJob)
+
+    PeerTubeSocket.Instance.sendAvailableJobsPingToRunners()
+  }
+
+  protected setAbortState (runnerJob: MRunnerJob) {
+    runnerJob.resetToPending()
+  }
+
+  protected abstract specificAbort (options: {
+    runnerJob: MRunnerJob
+  }): Promise<void> | void
+
+  // ---------------------------------------------------------------------------
+
+  async error (options: {
+    runnerJob: MRunnerJob
+    message: string
+    fromParent?: boolean
+  }) {
+    const { runnerJob, message, fromParent } = options
+
+    const errorState = fromParent
+      ? RunnerJobState.PARENT_ERRORED
+      : RunnerJobState.ERRORED
+
+    const nextState = errorState === RunnerJobState.ERRORED && this.isAbortSupported() && runnerJob.failures < RUNNER_JOBS.MAX_FAILURES
+      ? RunnerJobState.PENDING
+      : errorState
+
+    await this.specificError({ ...options, nextState })
+
+    if (nextState === errorState) {
+      runnerJob.setToErrorOrCancel(nextState)
+      runnerJob.error = message
+    } else {
+      runnerJob.resetToPending()
+    }
+
+    await saveInTransactionWithRetries(runnerJob)
+
+    if (runnerJob.state === errorState) {
+      const children = await RunnerJobModel.listChildrenOf(runnerJob)
+
+      for (const child of children) {
+        logger.info(`Erroring child job ${child.uuid} of ${runnerJob.uuid} because of parent error`, this.lTags(child.uuid))
+
+        await this.error({ runnerJob: child, message: 'Parent error', fromParent: true })
+      }
+    } else {
+      PeerTubeSocket.Instance.sendAvailableJobsPingToRunners()
+    }
+  }
+
+  protected abstract specificError (options: {
+    runnerJob: MRunnerJob
+    message: string
+    nextState: RunnerJobStateType
+  }): Promise<void> | void
+}

server/core/lib/runners/job-handlers/abstract-vod-transcoding-job-handler.ts

@@ -0,0 +1,71 @@
+import {
+  RunnerJobState,
+  RunnerJobStateType,
+  RunnerJobSuccessPayload,
+  RunnerJobUpdatePayload,
+  RunnerJobVODPrivatePayload
+} from '@peertube/peertube-models'
+import { retryTransactionWrapper } from '@server/helpers/database-utils.js'
+import { logger } from '@server/helpers/logger.js'
+import { moveToFailedTranscodingState, moveToNextState } from '@server/lib/video-state.js'
+import { VideoJobInfoModel } from '@server/models/video/video-job-info.js'
+import { MRunnerJob } from '@server/types/models/runners/index.js'
+import { AbstractJobHandler } from './abstract-job-handler.js'
+import { loadRunnerVideo } from './shared/utils.js'
+
+// 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
+  }) {
+    // empty
+  }
+
+  protected specificAbort (_options: {
+    runnerJob: MRunnerJob
+  }) {
+    // empty
+  }
+
+  protected async specificError (options: {
+    runnerJob: MRunnerJob
+    nextState: RunnerJobStateType
+  }) {
+    if (options.nextState !== RunnerJobState.ERRORED) return
+
+    const video = await loadRunnerVideo(options.runnerJob, this.lTags)
+    if (!video) return
+
+    await moveToFailedTranscodingState(video)
+
+    await VideoJobInfoModel.decrease(video.uuid, 'pendingTranscode')
+  }
+
+  protected async specificCancel (options: {
+    runnerJob: MRunnerJob
+  }) {
+    const { runnerJob } = options
+
+    const video = await loadRunnerVideo(options.runnerJob, this.lTags)
+    if (!video) return
+
+    const pending = await VideoJobInfoModel.decrease(video.uuid, 'pendingTranscode')
+
+    logger.debug(`Pending transcode decreased to ${pending} after cancel`, this.lTags(video.uuid))
+
+    if (pending === 0) {
+      logger.info(
+        `All transcoding jobs of ${video.uuid} have been processed or canceled, moving it to its next state`,
+        this.lTags(video.uuid)
+      )
+
+      const privatePayload = runnerJob.privatePayload as RunnerJobVODPrivatePayload
+      await retryTransactionWrapper(moveToNextState, { video, isNewVideo: privatePayload.isNewVideo })
+    }
+  }
+}

server/core/lib/runners/job-handlers/index.ts

@@ -0,0 +1,8 @@
+export * from './abstract-job-handler.js'
+export * from './live-rtmp-hls-transcoding-job-handler.js'
+export * from './runner-job-handlers.js'
+export * from './transcription-job-handler.js'
+export * from './video-studio-transcoding-job-handler.js'
+export * from './vod-audio-merge-transcoding-job-handler.js'
+export * from './vod-hls-transcoding-job-handler.js'
+export * from './vod-web-video-transcoding-job-handler.js'

server/core/lib/runners/job-handlers/live-rtmp-hls-transcoding-job-handler.ts

@@ -0,0 +1,176 @@
+import {
+  LiveRTMPHLSTranscodingSuccess,
+  LiveRTMPHLSTranscodingUpdatePayload,
+  LiveVideoError,
+  RunnerJobLiveRTMPHLSTranscodingPayload,
+  RunnerJobLiveRTMPHLSTranscodingPrivatePayload,
+  RunnerJobStateType
+} from '@peertube/peertube-models'
+import { buildUUID } from '@peertube/peertube-node-utils'
+import { tryAtomicMove } from '@server/helpers/fs.js'
+import { logger } from '@server/helpers/logger.js'
+import { JOB_PRIORITY } from '@server/initializers/constants.js'
+import { LiveManager } from '@server/lib/live/index.js'
+import { MStreamingPlaylist, MVideo } from '@server/types/models/index.js'
+import { MRunnerJob } from '@server/types/models/runners/index.js'
+import { remove } from 'fs-extra/esm'
+import { join } from 'path'
+import { AbstractJobHandler } from './abstract-job-handler.js'
+
+type CreateOptions = {
+  video: MVideo
+  playlist: MStreamingPlaylist
+
+  sessionId: string
+  rtmpUrl: string
+
+  toTranscode: {
+    resolution: number
+    fps: number
+  }[]
+
+  segmentListSize: number
+  segmentDuration: number
+
+  outputDirectory: string
+}
+
+// eslint-disable-next-line max-len
+export class LiveRTMPHLSTranscodingJobHandler extends AbstractJobHandler<CreateOptions, LiveRTMPHLSTranscodingUpdatePayload, LiveRTMPHLSTranscodingSuccess> {
+
+  async create (options: CreateOptions) {
+    const { video, rtmpUrl, toTranscode, playlist, segmentDuration, segmentListSize, outputDirectory, sessionId } = options
+
+    const jobUUID = buildUUID()
+    const payload: RunnerJobLiveRTMPHLSTranscodingPayload = {
+      input: {
+        rtmpUrl
+      },
+      output: {
+        toTranscode,
+        segmentListSize,
+        segmentDuration
+      }
+    }
+
+    const privatePayload: RunnerJobLiveRTMPHLSTranscodingPrivatePayload = {
+      videoUUID: video.uuid,
+      masterPlaylistName: playlist.playlistFilename,
+      sessionId,
+      outputDirectory
+    }
+
+    const job = await this.createRunnerJob({
+      type: 'live-rtmp-hls-transcoding',
+      jobUUID,
+      payload,
+      privatePayload,
+      priority: JOB_PRIORITY.TRANSCODING
+    })
+
+    return job
+  }
+
+  // ---------------------------------------------------------------------------
+
+  protected async specificUpdate (options: {
+    runnerJob: MRunnerJob
+    updatePayload: LiveRTMPHLSTranscodingUpdatePayload
+  }) {
+    const { runnerJob, updatePayload } = options
+
+    const privatePayload = runnerJob.privatePayload as RunnerJobLiveRTMPHLSTranscodingPrivatePayload
+    const outputDirectory = privatePayload.outputDirectory
+    const videoUUID = privatePayload.videoUUID
+
+    // Always process the chunk first before moving m3u8 that references this chunk
+    if (updatePayload.type === 'add-chunk') {
+      await tryAtomicMove(
+        updatePayload.videoChunkFile as string,
+        join(outputDirectory, updatePayload.videoChunkFilename)
+      )
+    } else if (updatePayload.type === 'remove-chunk') {
+      await remove(join(outputDirectory, updatePayload.videoChunkFilename))
+    }
+
+    if (updatePayload.resolutionPlaylistFile && updatePayload.resolutionPlaylistFilename) {
+      await tryAtomicMove(
+        updatePayload.resolutionPlaylistFile as string,
+        join(outputDirectory, updatePayload.resolutionPlaylistFilename)
+      )
+    }
+
+    if (updatePayload.masterPlaylistFile) {
+      await tryAtomicMove(updatePayload.masterPlaylistFile as string, join(outputDirectory, privatePayload.masterPlaylistName))
+    }
+
+    logger.debug(
+      'Runner live RTMP to HLS job %s for %s updated.',
+      runnerJob.uuid, videoUUID, { updatePayload, ...this.lTags(videoUUID, runnerJob.uuid) }
+    )
+  }
+
+  // ---------------------------------------------------------------------------
+
+  protected specificComplete (options: {
+    runnerJob: MRunnerJob
+  }) {
+    return this.stopLive({
+      runnerJob: options.runnerJob,
+      type: 'ended'
+    })
+  }
+
+  // ---------------------------------------------------------------------------
+
+  protected isAbortSupported () {
+    return false
+  }
+
+  protected specificAbort () {
+    throw new Error('Not implemented')
+  }
+
+  protected specificError (options: {
+    runnerJob: MRunnerJob
+    nextState: RunnerJobStateType
+  }) {
+    return this.stopLive({
+      runnerJob: options.runnerJob,
+      type: 'errored'
+    })
+  }
+
+  protected specificCancel (options: {
+    runnerJob: MRunnerJob
+  }) {
+    return this.stopLive({
+      runnerJob: options.runnerJob,
+      type: 'cancelled'
+    })
+  }
+
+  private stopLive (options: {
+    runnerJob: MRunnerJob
+    type: 'ended' | 'errored' | 'cancelled'
+  }) {
+    const { runnerJob, type } = options
+
+    const privatePayload = runnerJob.privatePayload as RunnerJobLiveRTMPHLSTranscodingPrivatePayload
+    const videoUUID = privatePayload.videoUUID
+
+    const errorType = {
+      ended: null,
+      errored: LiveVideoError.RUNNER_JOB_ERROR,
+      cancelled: LiveVideoError.RUNNER_JOB_CANCEL
+    }
+
+    LiveManager.Instance.stopSessionOfVideo({
+      videoUUID: privatePayload.videoUUID,
+      expectedSessionId: privatePayload.sessionId,
+      error: errorType[type]
+    })
+
+    logger.info('Runner live RTMP to HLS job %s for video %s %s.', runnerJob.uuid, videoUUID, type, this.lTags(runnerJob.uuid, videoUUID))
+  }
+}

server/core/lib/runners/job-handlers/runner-job-handlers.ts

@@ -0,0 +1,22 @@
+import { RunnerJobSuccessPayload, RunnerJobType, RunnerJobUpdatePayload } from '@peertube/peertube-models'
+import { MRunnerJob } from '@server/types/models/runners/index.js'
+import { AbstractJobHandler } from './abstract-job-handler.js'
+import { LiveRTMPHLSTranscodingJobHandler } from './live-rtmp-hls-transcoding-job-handler.js'
+import { TranscriptionJobHandler } from './transcription-job-handler.js'
+import { VideoStudioTranscodingJobHandler } from './video-studio-transcoding-job-handler.js'
+import { VODAudioMergeTranscodingJobHandler } from './vod-audio-merge-transcoding-job-handler.js'
+import { VODHLSTranscodingJobHandler } from './vod-hls-transcoding-job-handler.js'
+import { VODWebVideoTranscodingJobHandler } from './vod-web-video-transcoding-job-handler.js'
+
+const processors: Record<RunnerJobType, new() => AbstractJobHandler<unknown, RunnerJobUpdatePayload, RunnerJobSuccessPayload>> = {
+  'vod-web-video-transcoding': VODWebVideoTranscodingJobHandler,
+  'vod-hls-transcoding': VODHLSTranscodingJobHandler,
+  'vod-audio-merge-transcoding': VODAudioMergeTranscodingJobHandler,
+  'live-rtmp-hls-transcoding': LiveRTMPHLSTranscodingJobHandler,
+  'video-studio-transcoding': VideoStudioTranscodingJobHandler,
+  'video-transcription': TranscriptionJobHandler
+}
+
+export function getRunnerJobHandlerClass (job: MRunnerJob) {
+  return processors[job.type]
+}

server/core/lib/runners/job-handlers/shared/utils.ts

@@ -0,0 +1,36 @@
+import { logger, LoggerTagsFn } from '@server/helpers/logger.js'
+import { onTranscodingEnded } from '@server/lib/transcoding/ended-transcoding.js'
+import { onWebVideoFileTranscoding } from '@server/lib/transcoding/web-transcoding.js'
+import { VideoModel } from '@server/models/video/video.js'
+import { MVideoFullLight } from '@server/types/models/index.js'
+import { MRunnerJob } from '@server/types/models/runners/index.js'
+import { RunnerJobVODAudioMergeTranscodingPrivatePayload, RunnerJobVODWebVideoTranscodingPrivatePayload } from '@peertube/peertube-models'
+
+export async function onVODWebVideoOrAudioMergeTranscodingJob (options: {
+  video: MVideoFullLight
+  videoFilePath: string
+  privatePayload: RunnerJobVODWebVideoTranscodingPrivatePayload | RunnerJobVODAudioMergeTranscodingPrivatePayload
+  wasAudioFile: boolean
+}) {
+  const { video, videoFilePath, privatePayload, wasAudioFile } = options
+
+  const deleteWebInputVideoFile = privatePayload.deleteInputFileId
+    ? video.VideoFiles.find(f => f.id === privatePayload.deleteInputFileId)
+    : undefined
+
+  await onWebVideoFileTranscoding({ video, videoOutputPath: videoFilePath, deleteWebInputVideoFile, wasAudioFile })
+
+  await onTranscodingEnded({ isNewVideo: privatePayload.isNewVideo, moveVideoToNextState: true, video })
+}
+
+export async function loadRunnerVideo (runnerJob: MRunnerJob, lTags: LoggerTagsFn) {
+  const videoUUID = runnerJob.privatePayload.videoUUID
+
+  const video = await VideoModel.loadFull(videoUUID)
+  if (!video) {
+    logger.info('Video %s does not exist anymore after runner job.', videoUUID, lTags(videoUUID))
+    return undefined
+  }
+
+  return video
+}

server/core/lib/runners/job-handlers/transcription-job-handler.ts

@@ -0,0 +1,99 @@
+import {
+  RunnerJobState,
+  RunnerJobStateType,
+  RunnerJobTranscriptionPayload,
+  RunnerJobTranscriptionPrivatePayload,
+  RunnerJobUpdatePayload,
+  TranscriptionSuccess
+} from '@peertube/peertube-models'
+import { buildUUID } from '@peertube/peertube-node-utils'
+import { JOB_PRIORITY } from '@server/initializers/constants.js'
+import { onTranscriptionEnded } from '@server/lib/video-captions.js'
+import { VideoJobInfoModel } from '@server/models/video/video-job-info.js'
+import { MVideoUUID } from '@server/types/models/index.js'
+import { MRunnerJob } from '@server/types/models/runners/index.js'
+import { generateRunnerTranscodingVideoInputFileUrl } from '../runner-urls.js'
+import { AbstractJobHandler } from './abstract-job-handler.js'
+import { loadRunnerVideo } from './shared/utils.js'
+
+type CreateOptions = {
+  video: MVideoUUID
+}
+
+export class TranscriptionJobHandler extends AbstractJobHandler<CreateOptions, RunnerJobUpdatePayload, TranscriptionSuccess> {
+
+  protected isAbortSupported () {
+    return true
+  }
+
+  protected specificUpdate (_options: {
+    runnerJob: MRunnerJob
+  }) {
+    // empty
+  }
+
+  protected specificAbort (_options: {
+    runnerJob: MRunnerJob
+  }) {
+    // empty
+  }
+
+  protected async specificError (options: {
+    runnerJob: MRunnerJob
+    nextState: RunnerJobStateType
+  }) {
+    if (options.nextState !== RunnerJobState.ERRORED) return
+
+    await VideoJobInfoModel.decrease(options.runnerJob.privatePayload.videoUUID, 'pendingTranscription')
+  }
+
+  protected async specificCancel (options: {
+    runnerJob: MRunnerJob
+  }) {
+    await VideoJobInfoModel.decrease(options.runnerJob.privatePayload.videoUUID, 'pendingTranscription')
+  }
+
+  async create (options: CreateOptions) {
+    const { video } = options
+
+    const jobUUID = buildUUID()
+    const payload: RunnerJobTranscriptionPayload = {
+      input: {
+        videoFileUrl: generateRunnerTranscodingVideoInputFileUrl(jobUUID, video.uuid)
+      }
+    }
+
+    const privatePayload: RunnerJobTranscriptionPrivatePayload = {
+      videoUUID: video.uuid
+    }
+
+    const job = await this.createRunnerJob({
+      type: 'video-transcription',
+      jobUUID,
+      payload,
+      privatePayload,
+      priority: JOB_PRIORITY.TRANSCODING
+    })
+
+    return job
+  }
+
+  // ---------------------------------------------------------------------------
+
+  protected async specificComplete (options: {
+    runnerJob: MRunnerJob
+    resultPayload: TranscriptionSuccess
+  }) {
+    const { runnerJob, resultPayload } = options
+
+    const video = await loadRunnerVideo(runnerJob, this.lTags)
+    if (!video) return
+
+    await onTranscriptionEnded({
+      video,
+      language: resultPayload.inputLanguage,
+      vttPath: resultPayload.vttFile as string,
+      lTags: this.lTags().tags
+    })
+  }
+}

server/core/lib/runners/job-handlers/video-studio-transcoding-job-handler.ts

@@ -0,0 +1,157 @@
+import {
+  RunnerJobState,
+  RunnerJobStateType,
+  RunnerJobStudioTranscodingPayload,
+  RunnerJobUpdatePayload,
+  RunnerJobVideoStudioTranscodingPrivatePayload,
+  VideoState,
+  VideoStudioTaskPayload,
+  VideoStudioTranscodingSuccess,
+  isVideoStudioTaskIntro,
+  isVideoStudioTaskOutro,
+  isVideoStudioTaskWatermark
+} from '@peertube/peertube-models'
+import { buildUUID } from '@peertube/peertube-node-utils'
+import { logger } from '@server/helpers/logger.js'
+import { onVideoStudioEnded, safeCleanupStudioTMPFiles } from '@server/lib/video-studio.js'
+import { MVideo } from '@server/types/models/index.js'
+import { MRunnerJob } from '@server/types/models/runners/index.js'
+import { basename } from 'path'
+import { generateRunnerEditionTranscodingVideoInputFileUrl, generateRunnerTranscodingVideoInputFileUrl } from '../runner-urls.js'
+import { AbstractJobHandler } from './abstract-job-handler.js'
+import { loadRunnerVideo } from './shared/utils.js'
+
+type CreateOptions = {
+  video: MVideo
+  tasks: VideoStudioTaskPayload[]
+  priority: number
+}
+
+// eslint-disable-next-line max-len
+export class VideoStudioTranscodingJobHandler extends AbstractJobHandler<CreateOptions, RunnerJobUpdatePayload, VideoStudioTranscodingSuccess> {
+
+  async create (options: CreateOptions) {
+    const { video, priority, tasks } = options
+
+    const jobUUID = buildUUID()
+    const payload: RunnerJobStudioTranscodingPayload = {
+      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: RunnerJobVideoStudioTranscodingPrivatePayload = {
+      videoUUID: video.uuid,
+      originalTasks: tasks
+    }
+
+    const job = await this.createRunnerJob({
+      type: 'video-studio-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: VideoStudioTranscodingSuccess
+  }) {
+    const { runnerJob, resultPayload } = options
+    const privatePayload = runnerJob.privatePayload as RunnerJobVideoStudioTranscodingPrivatePayload
+
+    const video = await loadRunnerVideo(runnerJob, this.lTags)
+    if (!video) {
+      await safeCleanupStudioTMPFiles(privatePayload.originalTasks)
+
+    }
+
+    const videoFilePath = resultPayload.videoFile as string
+
+    await onVideoStudioEnded({ 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: RunnerJobStateType
+  }) {
+    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 RunnerJobVideoStudioTranscodingPrivatePayload
+    await safeCleanupStudioTMPFiles(payload.originalTasks)
+
+    const video = await loadRunnerVideo(options.runnerJob, this.lTags)
+    if (!video) return
+
+    return video.setNewState(VideoState.PUBLISHED, false, undefined)
+  }
+}

server/core/lib/runners/job-handlers/vod-audio-merge-transcoding-job-handler.ts

@@ -0,0 +1,86 @@
+import { pick } from '@peertube/peertube-core-utils'
+import {
+  RunnerJobUpdatePayload,
+  RunnerJobVODAudioMergeTranscodingPayload,
+  RunnerJobVODWebVideoTranscodingPrivatePayload,
+  VODAudioMergeTranscodingSuccess
+} from '@peertube/peertube-models'
+import { buildUUID } from '@peertube/peertube-node-utils'
+import { logger } from '@server/helpers/logger.js'
+import { VideoJobInfoModel } from '@server/models/video/video-job-info.js'
+import { MVideo } from '@server/types/models/index.js'
+import { MRunnerJob } from '@server/types/models/runners/index.js'
+import { generateRunnerTranscodingVideoInputFileUrl, generateRunnerTranscodingVideoPreviewFileUrl } from '../runner-urls.js'
+import { AbstractVODTranscodingJobHandler } from './abstract-vod-transcoding-job-handler.js'
+import { loadRunnerVideo, onVODWebVideoOrAudioMergeTranscodingJob } from './shared/utils.js'
+
+type CreateOptions = {
+  video: MVideo
+  isNewVideo: boolean
+  resolution: number
+  fps: number
+  priority: number
+  deleteInputFileId: number | null
+  dependsOnRunnerJob?: MRunnerJob
+}
+
+// eslint-disable-next-line max-len
+export class VODAudioMergeTranscodingJobHandler extends AbstractVODTranscodingJobHandler<CreateOptions, RunnerJobUpdatePayload, VODAudioMergeTranscodingSuccess> {
+
+  async create (options: CreateOptions) {
+    const { video, resolution, fps, priority, dependsOnRunnerJob } = options
+
+    const jobUUID = buildUUID()
+    const payload: RunnerJobVODAudioMergeTranscodingPayload = {
+      input: {
+        audioFileUrl: generateRunnerTranscodingVideoInputFileUrl(jobUUID, video.uuid),
+        previewFileUrl: generateRunnerTranscodingVideoPreviewFileUrl(jobUUID, video.uuid)
+      },
+      output: {
+        resolution,
+        fps
+      }
+    }
+
+    const privatePayload: RunnerJobVODWebVideoTranscodingPrivatePayload = {
+      ...pick(options, [ 'isNewVideo', 'deleteInputFileId' ]),
+
+      videoUUID: video.uuid
+    }
+
+    const job = await this.createRunnerJob({
+      type: 'vod-audio-merge-transcoding',
+      jobUUID,
+      payload,
+      privatePayload,
+      priority,
+      dependsOnRunnerJob
+    })
+
+    await VideoJobInfoModel.increaseOrCreate(video.uuid, 'pendingTranscode')
+
+    return job
+  }
+
+  // ---------------------------------------------------------------------------
+
+  protected async specificComplete (options: {
+    runnerJob: MRunnerJob
+    resultPayload: VODAudioMergeTranscodingSuccess
+  }) {
+    const { runnerJob, resultPayload } = options
+    const privatePayload = runnerJob.privatePayload as RunnerJobVODWebVideoTranscodingPrivatePayload
+
+    const video = await loadRunnerVideo(runnerJob, this.lTags)
+    if (!video) return
+
+    const videoFilePath = resultPayload.videoFile as string
+
+    await onVODWebVideoOrAudioMergeTranscodingJob({ video, videoFilePath, privatePayload, wasAudioFile: true })
+
+    logger.info(
+      'Runner VOD audio merge transcoding job %s for %s ended.',
+      runnerJob.uuid, video.uuid, this.lTags(video.uuid, runnerJob.uuid)
+    )
+  }
+}

server/core/lib/runners/job-handlers/vod-hls-transcoding-job-handler.ts

@@ -0,0 +1,99 @@
+import { pick } from '@peertube/peertube-core-utils'
+import {
+  RunnerJobUpdatePayload,
+  RunnerJobVODHLSTranscodingPayload,
+  RunnerJobVODHLSTranscodingPrivatePayload,
+  VODHLSTranscodingSuccess
+} from '@peertube/peertube-models'
+import { buildUUID } from '@peertube/peertube-node-utils'
+import { logger } from '@server/helpers/logger.js'
+import { onTranscodingEnded } from '@server/lib/transcoding/ended-transcoding.js'
+import { onHLSVideoFileTranscoding } from '@server/lib/transcoding/hls-transcoding.js'
+import { removeAllWebVideoFiles } from '@server/lib/video-file.js'
+import { VideoJobInfoModel } from '@server/models/video/video-job-info.js'
+import { MVideo } from '@server/types/models/index.js'
+import { MRunnerJob } from '@server/types/models/runners/index.js'
+import { generateRunnerTranscodingVideoInputFileUrl } from '../runner-urls.js'
+import { AbstractVODTranscodingJobHandler } from './abstract-vod-transcoding-job-handler.js'
+import { loadRunnerVideo } from './shared/utils.js'
+
+type CreateOptions = {
+  video: MVideo
+  isNewVideo: boolean
+  deleteWebVideoFiles: boolean
+  resolution: number
+  fps: number
+  priority: number
+  dependsOnRunnerJob?: MRunnerJob
+}
+
+// eslint-disable-next-line max-len
+export class VODHLSTranscodingJobHandler extends AbstractVODTranscodingJobHandler<CreateOptions, RunnerJobUpdatePayload, VODHLSTranscodingSuccess> {
+
+  async create (options: CreateOptions) {
+    const { video, resolution, fps, dependsOnRunnerJob, priority } = options
+
+    const jobUUID = buildUUID()
+
+    const payload: RunnerJobVODHLSTranscodingPayload = {
+      input: {
+        videoFileUrl: generateRunnerTranscodingVideoInputFileUrl(jobUUID, video.uuid)
+      },
+      output: {
+        resolution,
+        fps
+      }
+    }
+
+    const privatePayload: RunnerJobVODHLSTranscodingPrivatePayload = {
+      ...pick(options, [ 'isNewVideo', 'deleteWebVideoFiles' ]),
+
+      videoUUID: video.uuid
+    }
+
+    const job = await this.createRunnerJob({
+      type: 'vod-hls-transcoding',
+      jobUUID,
+      payload,
+      privatePayload,
+      priority,
+      dependsOnRunnerJob
+    })
+
+    await VideoJobInfoModel.increaseOrCreate(video.uuid, 'pendingTranscode')
+
+    return job
+  }
+
+  // ---------------------------------------------------------------------------
+
+  protected async specificComplete (options: {
+    runnerJob: MRunnerJob
+    resultPayload: VODHLSTranscodingSuccess
+  }) {
+    const { runnerJob, resultPayload } = options
+    const privatePayload = runnerJob.privatePayload as RunnerJobVODHLSTranscodingPrivatePayload
+
+    const video = await loadRunnerVideo(runnerJob, this.lTags)
+    if (!video) return
+
+    const videoFilePath = resultPayload.videoFile as string
+    const resolutionPlaylistFilePath = resultPayload.resolutionPlaylistFile as string
+
+    await onHLSVideoFileTranscoding({
+      video,
+      m3u8OutputPath: resolutionPlaylistFilePath,
+      videoOutputPath: videoFilePath
+    })
+
+    await onTranscodingEnded({ isNewVideo: privatePayload.isNewVideo, moveVideoToNextState: true, video })
+
+    if (privatePayload.deleteWebVideoFiles === true) {
+      logger.info('Removing web video files of %s now we have a HLS version of it.', video.uuid, this.lTags(video.uuid))
+
+      await removeAllWebVideoFiles(video)
+    }
+
+    logger.info('Runner VOD HLS job %s for %s ended.', runnerJob.uuid, video.uuid, this.lTags(runnerJob.uuid, video.uuid))
+  }
+}

server/core/lib/runners/job-handlers/vod-web-video-transcoding-job-handler.ts

@@ -0,0 +1,85 @@
+import { pick } from '@peertube/peertube-core-utils'
+import {
+  RunnerJobUpdatePayload,
+  RunnerJobVODWebVideoTranscodingPayload,
+  RunnerJobVODWebVideoTranscodingPrivatePayload,
+  VODWebVideoTranscodingSuccess
+} from '@peertube/peertube-models'
+import { buildUUID } from '@peertube/peertube-node-utils'
+import { logger } from '@server/helpers/logger.js'
+import { VideoJobInfoModel } from '@server/models/video/video-job-info.js'
+import { MVideo } from '@server/types/models/index.js'
+import { MRunnerJob } from '@server/types/models/runners/index.js'
+import { generateRunnerTranscodingVideoInputFileUrl } from '../runner-urls.js'
+import { AbstractVODTranscodingJobHandler } from './abstract-vod-transcoding-job-handler.js'
+import { loadRunnerVideo, onVODWebVideoOrAudioMergeTranscodingJob } from './shared/utils.js'
+
+type CreateOptions = {
+  video: MVideo
+  isNewVideo: boolean
+  resolution: number
+  fps: number
+  priority: number
+  deleteInputFileId: number | null
+  dependsOnRunnerJob?: MRunnerJob
+}
+
+// eslint-disable-next-line max-len
+export class VODWebVideoTranscodingJobHandler extends AbstractVODTranscodingJobHandler<CreateOptions, RunnerJobUpdatePayload, VODWebVideoTranscodingSuccess> {
+
+  async create (options: CreateOptions) {
+    const { video, resolution, fps, priority, dependsOnRunnerJob } = options
+
+    const jobUUID = buildUUID()
+    const payload: RunnerJobVODWebVideoTranscodingPayload = {
+      input: {
+        videoFileUrl: generateRunnerTranscodingVideoInputFileUrl(jobUUID, video.uuid)
+      },
+      output: {
+        resolution,
+        fps
+      }
+    }
+
+    const privatePayload: RunnerJobVODWebVideoTranscodingPrivatePayload = {
+      ...pick(options, [ 'isNewVideo', 'deleteInputFileId' ]),
+
+      videoUUID: video.uuid
+    }
+
+    const job = await this.createRunnerJob({
+      type: 'vod-web-video-transcoding',
+      jobUUID,
+      payload,
+      privatePayload,
+      dependsOnRunnerJob,
+      priority
+    })
+
+    await VideoJobInfoModel.increaseOrCreate(video.uuid, 'pendingTranscode')
+
+    return job
+  }
+
+  // ---------------------------------------------------------------------------
+
+  protected async specificComplete (options: {
+    runnerJob: MRunnerJob
+    resultPayload: VODWebVideoTranscodingSuccess
+  }) {
+    const { runnerJob, resultPayload } = options
+    const privatePayload = runnerJob.privatePayload as RunnerJobVODWebVideoTranscodingPrivatePayload
+
+    const video = await loadRunnerVideo(runnerJob, this.lTags)
+    if (!video) return
+
+    const videoFilePath = resultPayload.videoFile as string
+
+    await onVODWebVideoOrAudioMergeTranscodingJob({ video, videoFilePath, privatePayload, wasAudioFile: false })
+
+    logger.info(
+      'Runner VOD web video transcoding job %s for %s ended.',
+      runnerJob.uuid, video.uuid, this.lTags(video.uuid, runnerJob.uuid)
+    )
+  }
+}