Merge branch 'feature/esm-and-nx' into develop

6 files changed, 1915 insertions, 0 deletions

diff --git a/packages/tests/src/api/runners/index.ts b/packages/tests/src/api/runners/index.ts
new file mode 100644
index 000000000..441ddc874
--- /dev/null
+++ b/packages/tests/src/api/runners/index.ts
@@ -0,0 +1,5 @@
+export * from './runner-common.js'
+export * from './runner-live-transcoding.js'
+export * from './runner-socket.js'
+export * from './runner-studio-transcoding.js'
+export * from './runner-vod-transcoding.js'
diff --git a/packages/tests/src/api/runners/runner-common.ts b/packages/tests/src/api/runners/runner-common.ts
new file mode 100644
index 000000000..53ea321d0
--- /dev/null
+++ b/packages/tests/src/api/runners/runner-common.ts
@@ -0,0 +1,744 @@
+/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
+
+import { wait } from '@peertube/peertube-core-utils'
+import {
+  HttpStatusCode,
+  Runner,
+  RunnerJob,
+  RunnerJobAdmin,
+  RunnerJobState,
+  RunnerJobStateType,
+  RunnerJobVODWebVideoTranscodingPayload,
+  RunnerRegistrationToken
+} from '@peertube/peertube-models'
+import {
+  PeerTubeServer,
+  cleanupTests,
+  createSingleServer,
+  setAccessTokensToServers,
+  setDefaultVideoChannel,
+  waitJobs
+} from '@peertube/peertube-server-commands'
+import { expect } from 'chai'
+
+describe('Test runner common actions', function () {
+  let server: PeerTubeServer
+  let registrationToken: string
+  let runnerToken: string
+  let jobMaxPriority: string
+
+  before(async function () {
+    this.timeout(120_000)
+
+    server = await createSingleServer(1, {
+      remote_runners: {
+        stalled_jobs: {
+          vod: '5 seconds'
+        }
+      }
+    })
+
+    await setAccessTokensToServers([ server ])
+    await setDefaultVideoChannel([ server ])
+
+    await server.config.enableTranscoding({ hls: true, webVideo: true })
+    await server.config.enableRemoteTranscoding()
+  })
+
+  describe('Managing runner registration tokens', function () {
+    let base: RunnerRegistrationToken[]
+    let registrationTokenToDelete: RunnerRegistrationToken
+
+    it('Should have a default registration token', async function () {
+      const { total, data } = await server.runnerRegistrationTokens.list()
+
+      expect(total).to.equal(1)
+      expect(data).to.have.lengthOf(1)
+
+      const token = data[0]
+      expect(token.id).to.exist
+      expect(token.createdAt).to.exist
+      expect(token.updatedAt).to.exist
+      expect(token.registeredRunnersCount).to.equal(0)
+      expect(token.registrationToken).to.exist
+    })
+
+    it('Should create other registration tokens', async function () {
+      await server.runnerRegistrationTokens.generate()
+      await server.runnerRegistrationTokens.generate()
+
+      const { total, data } = await server.runnerRegistrationTokens.list()
+      expect(total).to.equal(3)
+      expect(data).to.have.lengthOf(3)
+    })
+
+    it('Should list registration tokens', async function () {
+      {
+        const { total, data } = await server.runnerRegistrationTokens.list({ sort: 'createdAt' })
+        expect(total).to.equal(3)
+        expect(data).to.have.lengthOf(3)
+        expect(new Date(data[0].createdAt)).to.be.below(new Date(data[1].createdAt))
+        expect(new Date(data[1].createdAt)).to.be.below(new Date(data[2].createdAt))
+
+        base = data
+
+        registrationTokenToDelete = data[0]
+        registrationToken = data[1].registrationToken
+      }
+
+      {
+        const { total, data } = await server.runnerRegistrationTokens.list({ sort: '-createdAt', start: 2, count: 1 })
+        expect(total).to.equal(3)
+        expect(data).to.have.lengthOf(1)
+        expect(data[0].registrationToken).to.equal(base[0].registrationToken)
+      }
+    })
+
+    it('Should have appropriate registeredRunnersCount for registration tokens', async function () {
+      await server.runners.register({ name: 'to delete 1', registrationToken: registrationTokenToDelete.registrationToken })
+      await server.runners.register({ name: 'to delete 2', registrationToken: registrationTokenToDelete.registrationToken })
+
+      const { data } = await server.runnerRegistrationTokens.list()
+
+      for (const d of data) {
+        if (d.registrationToken === registrationTokenToDelete.registrationToken) {
+          expect(d.registeredRunnersCount).to.equal(2)
+        } else {
+          expect(d.registeredRunnersCount).to.equal(0)
+        }
+      }
+
+      const { data: runners } = await server.runners.list()
+      expect(runners).to.have.lengthOf(2)
+    })
+
+    it('Should delete a registration token', async function () {
+      await server.runnerRegistrationTokens.delete({ id: registrationTokenToDelete.id })
+
+      const { total, data } = await server.runnerRegistrationTokens.list({ sort: 'createdAt' })
+      expect(total).to.equal(2)
+      expect(data).to.have.lengthOf(2)
+
+      for (const d of data) {
+        expect(d.registeredRunnersCount).to.equal(0)
+        expect(d.registrationToken).to.not.equal(registrationTokenToDelete.registrationToken)
+      }
+    })
+
+    it('Should have removed runners of this registration token', async function () {
+      const { data: runners } = await server.runners.list()
+      expect(runners).to.have.lengthOf(0)
+    })
+  })
+
+  describe('Managing runners', function () {
+    let toDelete: Runner
+
+    it('Should not have runners available', async function () {
+      const { total, data } = await server.runners.list()
+
+      expect(data).to.have.lengthOf(0)
+      expect(total).to.equal(0)
+    })
+
+    it('Should register runners', async function () {
+      const now = new Date()
+
+      const result = await server.runners.register({
+        name: 'runner 1',
+        description: 'my super runner 1',
+        registrationToken
+      })
+      expect(result.runnerToken).to.exist
+      runnerToken = result.runnerToken
+
+      await server.runners.register({
+        name: 'runner 2',
+        registrationToken
+      })
+
+      const { total, data } = await server.runners.list({ sort: 'createdAt' })
+      expect(total).to.equal(2)
+      expect(data).to.have.lengthOf(2)
+
+      for (const d of data) {
+        expect(d.id).to.exist
+        expect(d.createdAt).to.exist
+        expect(d.updatedAt).to.exist
+        expect(new Date(d.createdAt)).to.be.above(now)
+        expect(new Date(d.updatedAt)).to.be.above(now)
+        expect(new Date(d.lastContact)).to.be.above(now)
+        expect(d.ip).to.exist
+      }
+
+      expect(data[0].name).to.equal('runner 1')
+      expect(data[0].description).to.equal('my super runner 1')
+
+      expect(data[1].name).to.equal('runner 2')
+      expect(data[1].description).to.be.null
+
+      toDelete = data[1]
+    })
+
+    it('Should list runners', async function () {
+      const { total, data } = await server.runners.list({ sort: '-createdAt', start: 1, count: 1 })
+
+      expect(total).to.equal(2)
+      expect(data).to.have.lengthOf(1)
+      expect(data[0].name).to.equal('runner 1')
+    })
+
+    it('Should delete a runner', async function () {
+      await server.runners.delete({ id: toDelete.id })
+
+      const { total, data } = await server.runners.list()
+
+      expect(total).to.equal(1)
+      expect(data).to.have.lengthOf(1)
+      expect(data[0].name).to.equal('runner 1')
+    })
+
+    it('Should unregister a runner', async function () {
+      const registered = await server.runners.autoRegisterRunner()
+
+      {
+        const { total, data } = await server.runners.list()
+        expect(total).to.equal(2)
+        expect(data).to.have.lengthOf(2)
+      }
+
+      await server.runners.unregister({ runnerToken: registered })
+
+      {
+        const { total, data } = await server.runners.list()
+        expect(total).to.equal(1)
+        expect(data).to.have.lengthOf(1)
+        expect(data[0].name).to.equal('runner 1')
+      }
+    })
+  })
+
+  describe('Managing runner jobs', function () {
+    let jobUUID: string
+    let jobToken: string
+    let lastRunnerContact: Date
+    let failedJob: RunnerJob
+
+    async function checkMainJobState (
+      mainJobState: RunnerJobStateType,
+      otherJobStates: RunnerJobStateType[] = [ RunnerJobState.PENDING, RunnerJobState.WAITING_FOR_PARENT_JOB ]
+    ) {
+      const { data } = await server.runnerJobs.list({ count: 10, sort: '-updatedAt' })
+
+      for (const job of data) {
+        if (job.uuid === jobUUID) {
+          expect(job.state.id).to.equal(mainJobState)
+        } else {
+          expect(otherJobStates).to.include(job.state.id)
+        }
+      }
+    }
+
+    function getMainJob () {
+      return server.runnerJobs.getJob({ uuid: jobUUID })
+    }
+
+    describe('List jobs', function () {
+
+      it('Should not have jobs', async function () {
+        const { total, data } = await server.runnerJobs.list()
+
+        expect(data).to.have.lengthOf(0)
+        expect(total).to.equal(0)
+      })
+
+      it('Should upload a video and have available jobs', async function () {
+        await server.videos.quickUpload({ name: 'to transcode' })
+        await waitJobs([ server ])
+
+        const { total, data } = await server.runnerJobs.list()
+
+        expect(data).to.have.lengthOf(10)
+        expect(total).to.equal(10)
+
+        for (const job of data) {
+          expect(job.startedAt).to.not.exist
+          expect(job.finishedAt).to.not.exist
+          expect(job.payload).to.exist
+          expect(job.privatePayload).to.exist
+        }
+
+        const hlsJobs = data.filter(d => d.type === 'vod-hls-transcoding')
+        const webVideoJobs = data.filter(d => d.type === 'vod-web-video-transcoding')
+
+        expect(hlsJobs).to.have.lengthOf(5)
+        expect(webVideoJobs).to.have.lengthOf(5)
+
+        const pendingJobs = data.filter(d => d.state.id === RunnerJobState.PENDING)
+        const waitingJobs = data.filter(d => d.state.id === RunnerJobState.WAITING_FOR_PARENT_JOB)
+
+        expect(pendingJobs).to.have.lengthOf(1)
+        expect(waitingJobs).to.have.lengthOf(9)
+      })
+
+      it('Should upload another video and list/sort jobs', async function () {
+        await server.videos.quickUpload({ name: 'to transcode 2' })
+        await waitJobs([ server ])
+
+        {
+          const { total, data } = await server.runnerJobs.list({ start: 0, count: 30 })
+
+          expect(data).to.have.lengthOf(20)
+          expect(total).to.equal(20)
+
+          jobUUID = data[16].uuid
+        }
+
+        {
+          const { total, data } = await server.runnerJobs.list({ start: 3, count: 1, sort: 'createdAt' })
+          expect(total).to.equal(20)
+
+          expect(data).to.have.lengthOf(1)
+          expect(data[0].uuid).to.equal(jobUUID)
+        }
+
+        {
+          let previousPriority = Infinity
+          const { total, data } = await server.runnerJobs.list({ start: 0, count: 100, sort: '-priority' })
+          expect(total).to.equal(20)
+
+          for (const job of data) {
+            expect(job.priority).to.be.at.most(previousPriority)
+            previousPriority = job.priority
+
+            if (job.state.id === RunnerJobState.PENDING) {
+              jobMaxPriority = job.uuid
+            }
+          }
+        }
+      })
+
+      it('Should search jobs', async function () {
+        {
+          const { total, data } = await server.runnerJobs.list({ search: jobUUID })
+
+          expect(data).to.have.lengthOf(1)
+          expect(total).to.equal(1)
+
+          expect(data[0].uuid).to.equal(jobUUID)
+        }
+
+        {
+          const { total, data } = await server.runnerJobs.list({ search: 'toto' })
+
+          expect(data).to.have.lengthOf(0)
+          expect(total).to.equal(0)
+        }
+
+        {
+          const { total, data } = await server.runnerJobs.list({ search: 'hls' })
+
+          expect(data).to.not.have.lengthOf(0)
+          expect(total).to.not.equal(0)
+
+          for (const job of data) {
+            expect(job.type).to.include('hls')
+          }
+        }
+      })
+
+      it('Should filter jobs', async function () {
+        {
+          const { total, data } = await server.runnerJobs.list({ stateOneOf: [ RunnerJobState.WAITING_FOR_PARENT_JOB ] })
+
+          expect(data).to.not.have.lengthOf(0)
+          expect(total).to.not.equal(0)
+
+          for (const job of data) {
+            expect(job.state.label).to.equal('Waiting for parent job to finish')
+          }
+        }
+
+        {
+          const { total, data } = await server.runnerJobs.list({ stateOneOf: [ RunnerJobState.COMPLETED ] })
+
+          expect(data).to.have.lengthOf(0)
+          expect(total).to.equal(0)
+        }
+      })
+    })
+
+    describe('Accept/update/abort/process a job', function () {
+
+      it('Should request available jobs', async function () {
+        lastRunnerContact = new Date()
+
+        const { availableJobs } = await server.runnerJobs.request({ runnerToken })
+
+        // Only optimize jobs are available
+        expect(availableJobs).to.have.lengthOf(2)
+
+        for (const job of availableJobs) {
+          expect(job.uuid).to.exist
+          expect(job.payload.input).to.exist
+          expect((job.payload as RunnerJobVODWebVideoTranscodingPayload).output).to.exist
+
+          expect((job as RunnerJobAdmin).privatePayload).to.not.exist
+        }
+
+        const hlsJobs = availableJobs.filter(d => d.type === 'vod-hls-transcoding')
+        const webVideoJobs = availableJobs.filter(d => d.type === 'vod-web-video-transcoding')
+
+        expect(hlsJobs).to.have.lengthOf(0)
+        expect(webVideoJobs).to.have.lengthOf(2)
+
+        jobUUID = webVideoJobs[0].uuid
+      })
+
+      it('Should have sorted available jobs by priority', async function () {
+        const { availableJobs } = await server.runnerJobs.request({ runnerToken })
+
+        expect(availableJobs[0].uuid).to.equal(jobMaxPriority)
+      })
+
+      it('Should have last runner contact updated', async function () {
+        await wait(1000)
+
+        const { data } = await server.runners.list({ sort: 'createdAt' })
+        expect(new Date(data[0].lastContact)).to.be.above(lastRunnerContact)
+      })
+
+      it('Should accept a job', async function () {
+        const startedAt = new Date()
+
+        const { job } = await server.runnerJobs.accept({ runnerToken, jobUUID })
+        jobToken = job.jobToken
+
+        const checkProcessingJob = (job: RunnerJob & { jobToken?: string }, fromAccept: boolean) => {
+          expect(job.uuid).to.equal(jobUUID)
+
+          expect(job.type).to.equal('vod-web-video-transcoding')
+          expect(job.state.label).to.equal('Processing')
+          expect(job.state.id).to.equal(RunnerJobState.PROCESSING)
+
+          expect(job.runner).to.exist
+          expect(job.runner.name).to.equal('runner 1')
+          expect(job.runner.description).to.equal('my super runner 1')
+
+          expect(job.progress).to.be.null
+
+          expect(job.startedAt).to.exist
+          expect(new Date(job.startedAt)).to.be.above(startedAt)
+
+          expect(job.finishedAt).to.not.exist
+
+          expect(job.failures).to.equal(0)
+
+          expect(job.payload).to.exist
+
+          if (fromAccept) {
+            expect(job.jobToken).to.exist
+            expect((job as RunnerJobAdmin).privatePayload).to.not.exist
+          } else {
+            expect(job.jobToken).to.not.exist
+            expect((job as RunnerJobAdmin).privatePayload).to.exist
+          }
+        }
+
+        checkProcessingJob(job, true)
+
+        const { data } = await server.runnerJobs.list({ count: 10, sort: '-updatedAt' })
+
+        const processingJob = data.find(j => j.uuid === jobUUID)
+        checkProcessingJob(processingJob, false)
+
+        await checkMainJobState(RunnerJobState.PROCESSING)
+      })
+
+      it('Should update a job', async function () {
+        await server.runnerJobs.update({ runnerToken, jobUUID, jobToken, progress: 53 })
+
+        const { data } = await server.runnerJobs.list({ count: 10, sort: '-updatedAt' })
+
+        for (const job of data) {
+          if (job.state.id === RunnerJobState.PROCESSING) {
+            expect(job.progress).to.equal(53)
+          } else {
+            expect(job.progress).to.be.null
+          }
+        }
+      })
+
+      it('Should abort a job', async function () {
+        await server.runnerJobs.abort({ runnerToken, jobUUID, jobToken, reason: 'for tests' })
+
+        await checkMainJobState(RunnerJobState.PENDING)
+
+        const { data } = await server.runnerJobs.list({ count: 10, sort: '-updatedAt' })
+        for (const job of data) {
+          expect(job.progress).to.be.null
+        }
+      })
+
+      it('Should accept the same job again and post a success', async function () {
+        const { availableJobs } = await server.runnerJobs.request({ runnerToken })
+        expect(availableJobs.find(j => j.uuid === jobUUID)).to.exist
+
+        const { job } = await server.runnerJobs.accept({ runnerToken, jobUUID })
+        jobToken = job.jobToken
+
+        await checkMainJobState(RunnerJobState.PROCESSING)
+
+        const { data } = await server.runnerJobs.list({ count: 10, sort: '-updatedAt' })
+
+        for (const job of data) {
+          expect(job.progress).to.be.null
+        }
+
+        const payload = {
+          videoFile: 'video_short.mp4'
+        }
+
+        await server.runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+      })
+
+      it('Should not have available jobs anymore', async function () {
+        await checkMainJobState(RunnerJobState.COMPLETED)
+
+        const job = await getMainJob()
+        expect(job.finishedAt).to.exist
+
+        const { availableJobs } = await server.runnerJobs.request({ runnerToken })
+        expect(availableJobs.find(j => j.uuid === jobUUID)).to.not.exist
+      })
+    })
+
+    describe('Error job', function () {
+
+      it('Should accept another job and post an error', async function () {
+        await server.runnerJobs.cancelAllJobs()
+        await server.videos.quickUpload({ name: 'video' })
+        await waitJobs([ server ])
+
+        const { availableJobs } = await server.runnerJobs.request({ runnerToken })
+        jobUUID = availableJobs[0].uuid
+
+        const { job } = await server.runnerJobs.accept({ runnerToken, jobUUID })
+        jobToken = job.jobToken
+
+        await server.runnerJobs.error({ runnerToken, jobUUID, jobToken, message: 'Error' })
+      })
+
+      it('Should have job failures increased', async function () {
+        const job = await getMainJob()
+        expect(job.state.id).to.equal(RunnerJobState.PENDING)
+        expect(job.failures).to.equal(1)
+        expect(job.error).to.be.null
+        expect(job.progress).to.be.null
+        expect(job.finishedAt).to.not.exist
+      })
+
+      it('Should error a job when job attempts is too big', async function () {
+        for (let i = 0; i < 4; i++) {
+          const { job } = await server.runnerJobs.accept({ runnerToken, jobUUID })
+          jobToken = job.jobToken
+
+          await server.runnerJobs.error({ runnerToken, jobUUID, jobToken, message: 'Error ' + i })
+        }
+
+        const job = await getMainJob()
+        expect(job.failures).to.equal(5)
+        expect(job.state.id).to.equal(RunnerJobState.ERRORED)
+        expect(job.state.label).to.equal('Errored')
+        expect(job.error).to.equal('Error 3')
+        expect(job.progress).to.be.null
+        expect(job.finishedAt).to.exist
+
+        failedJob = job
+      })
+
+      it('Should have failed children jobs too', async function () {
+        const { data } = await server.runnerJobs.list({ count: 50, sort: '-updatedAt' })
+
+        const children = data.filter(j => j.parent?.uuid === failedJob.uuid)
+        expect(children).to.have.lengthOf(9)
+
+        for (const child of children) {
+          expect(child.parent.uuid).to.equal(failedJob.uuid)
+          expect(child.parent.type).to.equal(failedJob.type)
+          expect(child.parent.state.id).to.equal(failedJob.state.id)
+          expect(child.parent.state.label).to.equal(failedJob.state.label)
+
+          expect(child.state.id).to.equal(RunnerJobState.PARENT_ERRORED)
+          expect(child.state.label).to.equal('Parent job failed')
+        }
+      })
+    })
+
+    describe('Cancel', function () {
+
+      it('Should cancel a pending job', async function () {
+        await server.videos.quickUpload({ name: 'video' })
+        await waitJobs([ server ])
+
+        {
+          const { data } = await server.runnerJobs.list({ count: 10, sort: '-updatedAt' })
+
+          const pendingJob = data.find(j => j.state.id === RunnerJobState.PENDING)
+          jobUUID = pendingJob.uuid
+
+          await server.runnerJobs.cancelByAdmin({ jobUUID })
+        }
+
+        {
+          const job = await getMainJob()
+          expect(job.state.id).to.equal(RunnerJobState.CANCELLED)
+          expect(job.state.label).to.equal('Cancelled')
+        }
+
+        {
+          const { data } = await server.runnerJobs.list({ count: 10, sort: '-updatedAt' })
+          const children = data.filter(j => j.parent?.uuid === jobUUID)
+          expect(children).to.have.lengthOf(9)
+
+          for (const child of children) {
+            expect(child.state.id).to.equal(RunnerJobState.PARENT_CANCELLED)
+          }
+        }
+      })
+
+      it('Should cancel an already accepted job and skip success/error', async function () {
+        await server.videos.quickUpload({ name: 'video' })
+        await waitJobs([ server ])
+
+        const { availableJobs } = await server.runnerJobs.request({ runnerToken })
+        jobUUID = availableJobs[0].uuid
+
+        const { job } = await server.runnerJobs.accept({ runnerToken, jobUUID })
+        jobToken = job.jobToken
+
+        await server.runnerJobs.cancelByAdmin({ jobUUID })
+
+        await server.runnerJobs.abort({ runnerToken, jobUUID, jobToken, reason: 'aborted', expectedStatus: HttpStatusCode.NOT_FOUND_404 })
+      })
+    })
+
+    describe('Remove', function () {
+
+      it('Should remove a pending job', async function () {
+        await server.videos.quickUpload({ name: 'video' })
+        await waitJobs([ server ])
+
+        {
+          const { data } = await server.runnerJobs.list({ count: 10, sort: '-updatedAt' })
+
+          const pendingJob = data.find(j => j.state.id === RunnerJobState.PENDING)
+          jobUUID = pendingJob.uuid
+
+          await server.runnerJobs.deleteByAdmin({ jobUUID })
+        }
+
+        {
+          const { data } = await server.runnerJobs.list({ count: 10, sort: '-updatedAt' })
+
+          const parent = data.find(j => j.uuid === jobUUID)
+          expect(parent).to.not.exist
+
+          const children = data.filter(j => j.parent?.uuid === jobUUID)
+          expect(children).to.have.lengthOf(0)
+        }
+      })
+    })
+
+    describe('Stalled jobs', function () {
+
+      it('Should abort stalled jobs', async function () {
+        this.timeout(60000)
+
+        await server.videos.quickUpload({ name: 'video' })
+        await server.videos.quickUpload({ name: 'video' })
+        await waitJobs([ server ])
+
+        const { job: job1 } = await server.runnerJobs.autoAccept({ runnerToken })
+        const { job: stalledJob } = await server.runnerJobs.autoAccept({ runnerToken })
+
+        for (let i = 0; i < 6; i++) {
+          await wait(2000)
+
+          await server.runnerJobs.update({ runnerToken, jobToken: job1.jobToken, jobUUID: job1.uuid })
+        }
+
+        const refreshedJob1 = await server.runnerJobs.getJob({ uuid: job1.uuid })
+        const refreshedStalledJob = await server.runnerJobs.getJob({ uuid: stalledJob.uuid })
+
+        expect(refreshedJob1.state.id).to.equal(RunnerJobState.PROCESSING)
+        expect(refreshedStalledJob.state.id).to.equal(RunnerJobState.PENDING)
+      })
+    })
+
+    describe('Rate limit', function () {
+
+      before(async function () {
+        this.timeout(60000)
+
+        await server.kill()
+
+        await server.run({
+          rates_limit: {
+            api: {
+              max: 10
+            }
+          }
+        })
+      })
+
+      it('Should rate limit an unknown runner, but not a registered one', async function () {
+        this.timeout(60000)
+
+        await server.videos.quickUpload({ name: 'video' })
+        await waitJobs([ server ])
+
+        const { job } = await server.runnerJobs.autoAccept({ runnerToken })
+
+        for (let i = 0; i < 20; i++) {
+          try {
+            await server.runnerJobs.request({ runnerToken })
+            await server.runnerJobs.update({ runnerToken, jobToken: job.jobToken, jobUUID: job.uuid })
+          } catch {}
+        }
+
+        // Invalid
+        {
+          await server.runnerJobs.request({ runnerToken: 'toto', expectedStatus: HttpStatusCode.TOO_MANY_REQUESTS_429 })
+          await server.runnerJobs.update({
+            runnerToken: 'toto',
+            jobToken: job.jobToken,
+            jobUUID: job.uuid,
+            expectedStatus: HttpStatusCode.TOO_MANY_REQUESTS_429
+          })
+        }
+
+        // Not provided
+        {
+          await server.runnerJobs.request({ runnerToken: undefined, expectedStatus: HttpStatusCode.TOO_MANY_REQUESTS_429 })
+          await server.runnerJobs.update({
+            runnerToken: undefined,
+            jobToken: job.jobToken,
+            jobUUID: job.uuid,
+            expectedStatus: HttpStatusCode.TOO_MANY_REQUESTS_429
+          })
+        }
+
+        // Registered
+        {
+          await server.runnerJobs.request({ runnerToken })
+          await server.runnerJobs.update({ runnerToken, jobToken: job.jobToken, jobUUID: job.uuid })
+        }
+      })
+    })
+  })
+
+  after(async function () {
+    await cleanupTests([ server ])
+  })
+})
diff --git a/packages/tests/src/api/runners/runner-live-transcoding.ts b/packages/tests/src/api/runners/runner-live-transcoding.ts
new file mode 100644
index 000000000..20c1e5c2a
--- /dev/null
+++ b/packages/tests/src/api/runners/runner-live-transcoding.ts
@@ -0,0 +1,332 @@
+/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
+
+import { expect } from 'chai'
+import { FfmpegCommand } from 'fluent-ffmpeg'
+import { readFile } from 'fs/promises'
+import { wait } from '@peertube/peertube-core-utils'
+import {
+  HttpStatusCode,
+  LiveRTMPHLSTranscodingUpdatePayload,
+  LiveVideo,
+  LiveVideoError,
+  LiveVideoErrorType,
+  RunnerJob,
+  RunnerJobLiveRTMPHLSTranscodingPayload,
+  Video,
+  VideoPrivacy,
+  VideoState
+} from '@peertube/peertube-models'
+import { buildAbsoluteFixturePath } from '@peertube/peertube-node-utils'
+import {
+  cleanupTests,
+  createSingleServer,
+  makeRawRequest,
+  PeerTubeServer,
+  sendRTMPStream,
+  setAccessTokensToServers,
+  setDefaultVideoChannel,
+  stopFfmpeg,
+  testFfmpegStreamError,
+  waitJobs
+} from '@peertube/peertube-server-commands'
+
+describe('Test runner live transcoding', function () {
+  let server: PeerTubeServer
+  let runnerToken: string
+  let baseUrl: string
+
+  before(async function () {
+    this.timeout(120_000)
+
+    server = await createSingleServer(1)
+
+    await setAccessTokensToServers([ server ])
+    await setDefaultVideoChannel([ server ])
+
+    await server.config.enableRemoteTranscoding()
+    await server.config.enableTranscoding()
+    runnerToken = await server.runners.autoRegisterRunner()
+
+    baseUrl = server.url + '/static/streaming-playlists/hls'
+  })
+
+  describe('Without transcoding enabled', function () {
+
+    before(async function () {
+      await server.config.enableLive({
+        allowReplay: false,
+        resolutions: 'min',
+        transcoding: false
+      })
+    })
+
+    it('Should not have available jobs', async function () {
+      this.timeout(120000)
+
+      const { live, video } = await server.live.quickCreate({ permanentLive: true, saveReplay: false, privacy: VideoPrivacy.PUBLIC })
+
+      const ffmpegCommand = sendRTMPStream({ rtmpBaseUrl: live.rtmpUrl, streamKey: live.streamKey })
+      await server.live.waitUntilPublished({ videoId: video.id })
+
+      await waitJobs([ server ])
+
+      const { availableJobs } = await server.runnerJobs.requestLive({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(0)
+
+      await stopFfmpeg(ffmpegCommand)
+    })
+  })
+
+  describe('With transcoding enabled on classic live', function () {
+    let live: LiveVideo
+    let video: Video
+    let ffmpegCommand: FfmpegCommand
+    let jobUUID: string
+    let acceptedJob: RunnerJob & { jobToken: string }
+
+    async function testPlaylistFile (fixture: string, expected: string) {
+      const text = await server.streamingPlaylists.get({ url: `${baseUrl}/${video.uuid}/${fixture}` })
+      expect(await readFile(buildAbsoluteFixturePath(expected), 'utf-8')).to.equal(text)
+
+    }
+
+    async function testTSFile (fixture: string, expected: string) {
+      const { body } = await makeRawRequest({ url: `${baseUrl}/${video.uuid}/${fixture}`, expectedStatus: HttpStatusCode.OK_200 })
+      expect(await readFile(buildAbsoluteFixturePath(expected))).to.deep.equal(body)
+    }
+
+    before(async function () {
+      await server.config.enableLive({
+        allowReplay: true,
+        resolutions: 'max',
+        transcoding: true
+      })
+    })
+
+    it('Should publish a a live and have available jobs', async function () {
+      this.timeout(120000)
+
+      const data = await server.live.quickCreate({ permanentLive: false, saveReplay: false, privacy: VideoPrivacy.PUBLIC })
+      live = data.live
+      video = data.video
+
+      ffmpegCommand = sendRTMPStream({ rtmpBaseUrl: live.rtmpUrl, streamKey: live.streamKey })
+      await waitJobs([ server ])
+
+      const job = await server.runnerJobs.requestLiveJob(runnerToken)
+      jobUUID = job.uuid
+
+      expect(job.type).to.equal('live-rtmp-hls-transcoding')
+      expect(job.payload.input.rtmpUrl).to.exist
+
+      expect(job.payload.output.toTranscode).to.have.lengthOf(5)
+
+      for (const { resolution, fps } of job.payload.output.toTranscode) {
+        expect([ 720, 480, 360, 240, 144 ]).to.contain(resolution)
+
+        expect(fps).to.be.above(25)
+        expect(fps).to.be.below(70)
+      }
+    })
+
+    it('Should update the live with a new chunk', async function () {
+      this.timeout(120000)
+
+      const { job } = await server.runnerJobs.accept<RunnerJobLiveRTMPHLSTranscodingPayload>({ jobUUID, runnerToken })
+      acceptedJob = job
+
+      {
+        const payload: LiveRTMPHLSTranscodingUpdatePayload = {
+          masterPlaylistFile: 'live/master.m3u8',
+          resolutionPlaylistFile: 'live/0.m3u8',
+          resolutionPlaylistFilename: '0.m3u8',
+          type: 'add-chunk',
+          videoChunkFile: 'live/0-000067.ts',
+          videoChunkFilename: '0-000067.ts'
+        }
+        await server.runnerJobs.update({ jobUUID, runnerToken, jobToken: job.jobToken, payload, progress: 50 })
+
+        const updatedJob = await server.runnerJobs.getJob({ uuid: job.uuid })
+        expect(updatedJob.progress).to.equal(50)
+      }
+
+      {
+        const payload: LiveRTMPHLSTranscodingUpdatePayload = {
+          resolutionPlaylistFile: 'live/1.m3u8',
+          resolutionPlaylistFilename: '1.m3u8',
+          type: 'add-chunk',
+          videoChunkFile: 'live/1-000068.ts',
+          videoChunkFilename: '1-000068.ts'
+        }
+        await server.runnerJobs.update({ jobUUID, runnerToken, jobToken: job.jobToken, payload })
+      }
+
+      await wait(1000)
+
+      await testPlaylistFile('master.m3u8', 'live/master.m3u8')
+      await testPlaylistFile('0.m3u8', 'live/0.m3u8')
+      await testPlaylistFile('1.m3u8', 'live/1.m3u8')
+
+      await testTSFile('0-000067.ts', 'live/0-000067.ts')
+      await testTSFile('1-000068.ts', 'live/1-000068.ts')
+    })
+
+    it('Should replace existing m3u8 on update', async function () {
+      this.timeout(120000)
+
+      const payload: LiveRTMPHLSTranscodingUpdatePayload = {
+        masterPlaylistFile: 'live/1.m3u8',
+        resolutionPlaylistFilename: '0.m3u8',
+        resolutionPlaylistFile: 'live/1.m3u8',
+        type: 'add-chunk',
+        videoChunkFile: 'live/1-000069.ts',
+        videoChunkFilename: '1-000068.ts'
+      }
+      await server.runnerJobs.update({ jobUUID, runnerToken, jobToken: acceptedJob.jobToken, payload })
+      await wait(1000)
+
+      await testPlaylistFile('master.m3u8', 'live/1.m3u8')
+      await testPlaylistFile('0.m3u8', 'live/1.m3u8')
+      await testTSFile('1-000068.ts', 'live/1-000069.ts')
+    })
+
+    it('Should update the live with removed chunks', async function () {
+      this.timeout(120000)
+
+      const payload: LiveRTMPHLSTranscodingUpdatePayload = {
+        resolutionPlaylistFile: 'live/0.m3u8',
+        resolutionPlaylistFilename: '0.m3u8',
+        type: 'remove-chunk',
+        videoChunkFilename: '1-000068.ts'
+      }
+      await server.runnerJobs.update({ jobUUID, runnerToken, jobToken: acceptedJob.jobToken, payload })
+
+      await wait(1000)
+
+      await server.streamingPlaylists.get({ url: `${baseUrl}/${video.uuid}/master.m3u8` })
+      await server.streamingPlaylists.get({ url: `${baseUrl}/${video.uuid}/0.m3u8` })
+      await server.streamingPlaylists.get({ url: `${baseUrl}/${video.uuid}/1.m3u8` })
+      await makeRawRequest({ url: `${baseUrl}/${video.uuid}/0-000067.ts`, expectedStatus: HttpStatusCode.OK_200 })
+      await makeRawRequest({ url: `${baseUrl}/${video.uuid}/1-000068.ts`, expectedStatus: HttpStatusCode.NOT_FOUND_404 })
+    })
+
+    it('Should complete the live and save the replay', async function () {
+      this.timeout(120000)
+
+      for (const segment of [ '0-000069.ts', '0-000070.ts' ]) {
+        const payload: LiveRTMPHLSTranscodingUpdatePayload = {
+          masterPlaylistFile: 'live/master.m3u8',
+          resolutionPlaylistFilename: '0.m3u8',
+          resolutionPlaylistFile: 'live/0.m3u8',
+          type: 'add-chunk',
+          videoChunkFile: 'live/' + segment,
+          videoChunkFilename: segment
+        }
+        await server.runnerJobs.update({ jobUUID, runnerToken, jobToken: acceptedJob.jobToken, payload })
+
+        await wait(1000)
+      }
+
+      await waitJobs([ server ])
+
+      {
+        const { state } = await server.videos.get({ id: video.uuid })
+        expect(state.id).to.equal(VideoState.PUBLISHED)
+      }
+
+      await stopFfmpeg(ffmpegCommand)
+
+      await server.runnerJobs.success({ jobUUID, runnerToken, jobToken: acceptedJob.jobToken, payload: {} })
+
+      await wait(1500)
+      await waitJobs([ server ])
+
+      {
+        const { state } = await server.videos.get({ id: video.uuid })
+        expect(state.id).to.equal(VideoState.LIVE_ENDED)
+
+        const session = await server.live.findLatestSession({ videoId: video.uuid })
+        expect(session.error).to.be.null
+      }
+    })
+  })
+
+  describe('With transcoding enabled on cancelled/aborted/errored live', function () {
+    let live: LiveVideo
+    let video: Video
+    let ffmpegCommand: FfmpegCommand
+
+    async function prepare () {
+      ffmpegCommand = sendRTMPStream({ rtmpBaseUrl: live.rtmpUrl, streamKey: live.streamKey })
+      await server.runnerJobs.requestLiveJob(runnerToken)
+
+      const { job } = await server.runnerJobs.autoAccept({ runnerToken, type: 'live-rtmp-hls-transcoding' })
+
+      return job
+    }
+
+    async function checkSessionError (error: LiveVideoErrorType) {
+      await wait(1500)
+      await waitJobs([ server ])
+
+      const session = await server.live.findLatestSession({ videoId: video.uuid })
+      expect(session.error).to.equal(error)
+    }
+
+    before(async function () {
+      await server.config.enableLive({
+        allowReplay: true,
+        resolutions: 'max',
+        transcoding: true
+      })
+
+      const data = await server.live.quickCreate({ permanentLive: true, saveReplay: false, privacy: VideoPrivacy.PUBLIC })
+      live = data.live
+      video = data.video
+    })
+
+    it('Should abort a running live', async function () {
+      this.timeout(120000)
+
+      const job = await prepare()
+
+      await Promise.all([
+        server.runnerJobs.abort({ jobUUID: job.uuid, runnerToken, jobToken: job.jobToken, reason: 'abort' }),
+        testFfmpegStreamError(ffmpegCommand, true)
+      ])
+
+      // Abort is not supported
+      await checkSessionError(LiveVideoError.RUNNER_JOB_ERROR)
+    })
+
+    it('Should cancel a running live', async function () {
+      this.timeout(120000)
+
+      const job = await prepare()
+
+      await Promise.all([
+        server.runnerJobs.cancelByAdmin({ jobUUID: job.uuid }),
+        testFfmpegStreamError(ffmpegCommand, true)
+      ])
+
+      await checkSessionError(LiveVideoError.RUNNER_JOB_CANCEL)
+    })
+
+    it('Should error a running live', async function () {
+      this.timeout(120000)
+
+      const job = await prepare()
+
+      await Promise.all([
+        server.runnerJobs.error({ jobUUID: job.uuid, runnerToken, jobToken: job.jobToken, message: 'error' }),
+        testFfmpegStreamError(ffmpegCommand, true)
+      ])
+
+      await checkSessionError(LiveVideoError.RUNNER_JOB_ERROR)
+    })
+  })
+
+  after(async function () {
+    await cleanupTests([ server ])
+  })
+})
diff --git a/packages/tests/src/api/runners/runner-socket.ts b/packages/tests/src/api/runners/runner-socket.ts
new file mode 100644
index 000000000..726ef084f
--- /dev/null
+++ b/packages/tests/src/api/runners/runner-socket.ts
@@ -0,0 +1,120 @@
+/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
+
+import { expect } from 'chai'
+import { wait } from '@peertube/peertube-core-utils'
+import {
+  cleanupTests,
+  createSingleServer,
+  PeerTubeServer,
+  setAccessTokensToServers,
+  setDefaultVideoChannel,
+  waitJobs
+} from '@peertube/peertube-server-commands'
+
+describe('Test runner socket', function () {
+  let server: PeerTubeServer
+  let runnerToken: string
+
+  before(async function () {
+    this.timeout(120_000)
+
+    server = await createSingleServer(1)
+
+    await setAccessTokensToServers([ server ])
+    await setDefaultVideoChannel([ server ])
+
+    await server.config.enableTranscoding({ hls: true, webVideo: true })
+    await server.config.enableRemoteTranscoding()
+    runnerToken = await server.runners.autoRegisterRunner()
+  })
+
+  it('Should throw an error without runner token', function (done) {
+    const localSocket = server.socketIO.getRunnersSocket({ runnerToken: null })
+    localSocket.on('connect_error', err => {
+      expect(err.message).to.contain('No runner token provided')
+      done()
+    })
+  })
+
+  it('Should throw an error with a bad runner token', function (done) {
+    const localSocket = server.socketIO.getRunnersSocket({ runnerToken: 'ergag' })
+    localSocket.on('connect_error', err => {
+      expect(err.message).to.contain('Invalid runner token')
+      done()
+    })
+  })
+
+  it('Should not send ping if there is no available jobs', async function () {
+    let pings = 0
+    const localSocket = server.socketIO.getRunnersSocket({ runnerToken })
+    localSocket.on('available-jobs', () => pings++)
+
+    expect(pings).to.equal(0)
+  })
+
+  it('Should send a ping on available job', async function () {
+    let pings = 0
+    const localSocket = server.socketIO.getRunnersSocket({ runnerToken })
+    localSocket.on('available-jobs', () => pings++)
+
+    await server.videos.quickUpload({ name: 'video1' })
+    await waitJobs([ server ])
+
+    // eslint-disable-next-line no-unmodified-loop-condition
+    while (pings !== 1) {
+      await wait(500)
+    }
+
+    await server.videos.quickUpload({ name: 'video2' })
+    await waitJobs([ server ])
+
+    // eslint-disable-next-line no-unmodified-loop-condition
+    while ((pings as number) !== 2) {
+      await wait(500)
+    }
+
+    await server.runnerJobs.cancelAllJobs()
+  })
+
+  it('Should send a ping when a child is ready', async function () {
+    let pings = 0
+    const localSocket = server.socketIO.getRunnersSocket({ runnerToken })
+    localSocket.on('available-jobs', () => pings++)
+
+    await server.videos.quickUpload({ name: 'video3' })
+    await waitJobs([ server ])
+
+    // eslint-disable-next-line no-unmodified-loop-condition
+    while (pings !== 1) {
+      await wait(500)
+    }
+
+    await server.runnerJobs.autoProcessWebVideoJob(runnerToken)
+    await waitJobs([ server ])
+
+    // eslint-disable-next-line no-unmodified-loop-condition
+    while ((pings as number) !== 2) {
+      await wait(500)
+    }
+  })
+
+  it('Should not send a ping if the ended job does not have a child', async function () {
+    let pings = 0
+    const localSocket = server.socketIO.getRunnersSocket({ runnerToken })
+    localSocket.on('available-jobs', () => pings++)
+
+    const { availableJobs } = await server.runnerJobs.request({ runnerToken })
+    const job = availableJobs.find(j => j.type === 'vod-web-video-transcoding')
+    await server.runnerJobs.autoProcessWebVideoJob(runnerToken, job.uuid)
+
+    // Wait for debounce
+    await wait(1000)
+    await waitJobs([ server ])
+
+    expect(pings).to.equal(0)
+  })
+
+  after(async function () {
+    await cleanupTests([ server ])
+  })
+})
diff --git a/packages/tests/src/api/runners/runner-studio-transcoding.ts b/packages/tests/src/api/runners/runner-studio-transcoding.ts
new file mode 100644
index 000000000..adf6941c3
--- /dev/null
+++ b/packages/tests/src/api/runners/runner-studio-transcoding.ts
@@ -0,0 +1,169 @@
+/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
+
+import { expect } from 'chai'
+import { readFile } from 'fs/promises'
+import { buildAbsoluteFixturePath } from '@peertube/peertube-node-utils'
+import {
+  RunnerJobStudioTranscodingPayload,
+  VideoStudioTranscodingSuccess,
+  VideoState,
+  VideoStudioTask,
+  VideoStudioTaskIntro
+} from '@peertube/peertube-models'
+import {
+  cleanupTests,
+  createMultipleServers,
+  doubleFollow,
+  PeerTubeServer,
+  setAccessTokensToServers,
+  setDefaultVideoChannel,
+  VideoStudioCommand,
+  waitJobs
+} from '@peertube/peertube-server-commands'
+import { checkVideoDuration } from '@tests/shared/checks.js'
+import { checkPersistentTmpIsEmpty } from '@tests/shared/directories.js'
+
+describe('Test runner video studio transcoding', function () {
+  let servers: PeerTubeServer[] = []
+  let runnerToken: string
+  let videoUUID: string
+  let jobUUID: string
+
+  async function renewStudio (tasks: VideoStudioTask[] = VideoStudioCommand.getComplexTask()) {
+    const { uuid } = await servers[0].videos.quickUpload({ name: 'video' })
+    videoUUID = uuid
+
+    await waitJobs(servers)
+
+    await servers[0].videoStudio.createEditionTasks({ videoId: uuid, tasks })
+    await waitJobs(servers)
+
+    const { availableJobs } = await servers[0].runnerJobs.request({ runnerToken })
+    expect(availableJobs).to.have.lengthOf(1)
+
+    jobUUID = availableJobs[0].uuid
+  }
+
+  before(async function () {
+    this.timeout(120_000)
+
+    servers = await createMultipleServers(2)
+
+    await setAccessTokensToServers(servers)
+    await setDefaultVideoChannel(servers)
+
+    await doubleFollow(servers[0], servers[1])
+
+    await servers[0].config.enableTranscoding({ hls: true, webVideo: true })
+    await servers[0].config.enableStudio()
+    await servers[0].config.enableRemoteStudio()
+
+    runnerToken = await servers[0].runners.autoRegisterRunner()
+  })
+
+  it('Should error a studio transcoding job', async function () {
+    this.timeout(60000)
+
+    await renewStudio()
+
+    for (let i = 0; i < 5; i++) {
+      const { job } = await servers[0].runnerJobs.accept({ runnerToken, jobUUID })
+      const jobToken = job.jobToken
+
+      await servers[0].runnerJobs.error({ runnerToken, jobUUID, jobToken, message: 'Error' })
+    }
+
+    const video = await servers[0].videos.get({ id: videoUUID })
+    expect(video.state.id).to.equal(VideoState.PUBLISHED)
+
+    await checkPersistentTmpIsEmpty(servers[0])
+  })
+
+  it('Should cancel a transcoding job', async function () {
+    this.timeout(60000)
+
+    await renewStudio()
+
+    await servers[0].runnerJobs.cancelByAdmin({ jobUUID })
+
+    const video = await servers[0].videos.get({ id: videoUUID })
+    expect(video.state.id).to.equal(VideoState.PUBLISHED)
+
+    await checkPersistentTmpIsEmpty(servers[0])
+  })
+
+  it('Should execute a remote studio job', async function () {
+    this.timeout(240_000)
+
+    const tasks = [
+      {
+        name: 'add-outro' as 'add-outro',
+        options: {
+          file: 'video_short.webm'
+        }
+      },
+      {
+        name: 'add-watermark' as 'add-watermark',
+        options: {
+          file: 'custom-thumbnail.png'
+        }
+      },
+      {
+        name: 'add-intro' as 'add-intro',
+        options: {
+          file: 'video_very_short_240p.mp4'
+        }
+      }
+    ]
+
+    await renewStudio(tasks)
+
+    for (const server of servers) {
+      await checkVideoDuration(server, videoUUID, 5)
+    }
+
+    const { job } = await servers[0].runnerJobs.accept<RunnerJobStudioTranscodingPayload>({ runnerToken, jobUUID })
+    const jobToken = job.jobToken
+
+    expect(job.type === 'video-studio-transcoding')
+    expect(job.payload.input.videoFileUrl).to.exist
+
+    // Check video input file
+    {
+      await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+    }
+
+    // Check task files
+    for (let i = 0; i < tasks.length; i++) {
+      const task = tasks[i]
+      const payloadTask = job.payload.tasks[i]
+
+      expect(payloadTask.name).to.equal(task.name)
+
+      const inputFile = await readFile(buildAbsoluteFixturePath(task.options.file))
+
+      const { body } = await servers[0].runnerJobs.getJobFile({
+        url: (payloadTask as VideoStudioTaskIntro).options.file as string,
+        jobToken,
+        runnerToken
+      })
+
+      expect(body).to.deep.equal(inputFile)
+    }
+
+    const payload: VideoStudioTranscodingSuccess = { videoFile: 'video_very_short_240p.mp4' }
+    await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+
+    await waitJobs(servers)
+
+    for (const server of servers) {
+      await checkVideoDuration(server, videoUUID, 2)
+    }
+
+    await checkPersistentTmpIsEmpty(servers[0])
+  })
+
+  after(async function () {
+    await cleanupTests(servers)
+  })
+})
diff --git a/packages/tests/src/api/runners/runner-vod-transcoding.ts b/packages/tests/src/api/runners/runner-vod-transcoding.ts
new file mode 100644
index 000000000..fe1c8f0b2
--- /dev/null
+++ b/packages/tests/src/api/runners/runner-vod-transcoding.ts
@@ -0,0 +1,545 @@
+/* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/require-await */
+
+import { expect } from 'chai'
+import { readFile } from 'fs/promises'
+import { completeCheckHlsPlaylist } from '@tests/shared/streaming-playlists.js'
+import { buildAbsoluteFixturePath } from '@peertube/peertube-node-utils'
+import {
+  HttpStatusCode,
+  RunnerJobSuccessPayload,
+  RunnerJobVODAudioMergeTranscodingPayload,
+  RunnerJobVODHLSTranscodingPayload,
+  RunnerJobVODPayload,
+  RunnerJobVODWebVideoTranscodingPayload,
+  VideoState,
+  VODAudioMergeTranscodingSuccess,
+  VODHLSTranscodingSuccess,
+  VODWebVideoTranscodingSuccess
+} from '@peertube/peertube-models'
+import {
+  cleanupTests,
+  createMultipleServers,
+  doubleFollow,
+  makeGetRequest,
+  makeRawRequest,
+  PeerTubeServer,
+  setAccessTokensToServers,
+  setDefaultVideoChannel,
+  waitJobs
+} from '@peertube/peertube-server-commands'
+
+async function processAllJobs (server: PeerTubeServer, runnerToken: string) {
+  do {
+    const { availableJobs } = await server.runnerJobs.requestVOD({ runnerToken })
+    if (availableJobs.length === 0) break
+
+    const { job } = await server.runnerJobs.accept<RunnerJobVODPayload>({ runnerToken, jobUUID: availableJobs[0].uuid })
+
+    const payload: RunnerJobSuccessPayload = {
+      videoFile: `video_short_${job.payload.output.resolution}p.mp4`,
+      resolutionPlaylistFile: `video_short_${job.payload.output.resolution}p.m3u8`
+    }
+    await server.runnerJobs.success({ runnerToken, jobUUID: job.uuid, jobToken: job.jobToken, payload })
+  } while (true)
+
+  await waitJobs([ server ])
+}
+
+describe('Test runner VOD transcoding', function () {
+  let servers: PeerTubeServer[] = []
+  let runnerToken: string
+
+  before(async function () {
+    this.timeout(120_000)
+
+    servers = await createMultipleServers(2)
+
+    await setAccessTokensToServers(servers)
+    await setDefaultVideoChannel(servers)
+
+    await doubleFollow(servers[0], servers[1])
+
+    await servers[0].config.enableRemoteTranscoding()
+    runnerToken = await servers[0].runners.autoRegisterRunner()
+  })
+
+  describe('Without transcoding', function () {
+
+    before(async function () {
+      this.timeout(60000)
+
+      await servers[0].config.disableTranscoding()
+      await servers[0].videos.quickUpload({ name: 'video' })
+
+      await waitJobs(servers)
+    })
+
+    it('Should not have available jobs', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(0)
+    })
+  })
+
+  describe('With classic transcoding enabled', function () {
+
+    before(async function () {
+      this.timeout(60000)
+
+      await servers[0].config.enableTranscoding({ hls: true, webVideo: true })
+    })
+
+    it('Should error a transcoding job', async function () {
+      this.timeout(60000)
+
+      await servers[0].runnerJobs.cancelAllJobs()
+      const { uuid } = await servers[0].videos.quickUpload({ name: 'video' })
+      await waitJobs(servers)
+
+      const { availableJobs } = await servers[0].runnerJobs.request({ runnerToken })
+      const jobUUID = availableJobs[0].uuid
+
+      for (let i = 0; i < 5; i++) {
+        const { job } = await servers[0].runnerJobs.accept({ runnerToken, jobUUID })
+        const jobToken = job.jobToken
+
+        await servers[0].runnerJobs.error({ runnerToken, jobUUID, jobToken, message: 'Error' })
+      }
+
+      const video = await servers[0].videos.get({ id: uuid })
+      expect(video.state.id).to.equal(VideoState.TRANSCODING_FAILED)
+    })
+
+    it('Should cancel a transcoding job', async function () {
+      await servers[0].runnerJobs.cancelAllJobs()
+      const { uuid } = await servers[0].videos.quickUpload({ name: 'video' })
+      await waitJobs(servers)
+
+      const { availableJobs } = await servers[0].runnerJobs.request({ runnerToken })
+      const jobUUID = availableJobs[0].uuid
+
+      await servers[0].runnerJobs.cancelByAdmin({ jobUUID })
+
+      const video = await servers[0].videos.get({ id: uuid })
+      expect(video.state.id).to.equal(VideoState.PUBLISHED)
+    })
+  })
+
+  describe('Web video transcoding only', function () {
+    let videoUUID: string
+    let jobToken: string
+    let jobUUID: string
+
+    before(async function () {
+      this.timeout(60000)
+
+      await servers[0].runnerJobs.cancelAllJobs()
+      await servers[0].config.enableTranscoding({ hls: false, webVideo: true })
+
+      const { uuid } = await servers[0].videos.quickUpload({ name: 'web video', fixture: 'video_short.webm' })
+      videoUUID = uuid
+
+      await waitJobs(servers)
+    })
+
+    it('Should have jobs available for remote runners', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(1)
+
+      jobUUID = availableJobs[0].uuid
+    })
+
+    it('Should have a valid first transcoding job', async function () {
+      const { job } = await servers[0].runnerJobs.accept<RunnerJobVODWebVideoTranscodingPayload>({ runnerToken, jobUUID })
+      jobToken = job.jobToken
+
+      expect(job.type === 'vod-web-video-transcoding')
+      expect(job.payload.input.videoFileUrl).to.exist
+      expect(job.payload.output.resolution).to.equal(720)
+      expect(job.payload.output.fps).to.equal(25)
+
+      const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+      const inputFile = await readFile(buildAbsoluteFixturePath('video_short.webm'))
+
+      expect(body).to.deep.equal(inputFile)
+    })
+
+    it('Should transcode the max video resolution and send it back to the server', async function () {
+      this.timeout(60000)
+
+      const payload: VODWebVideoTranscodingSuccess = {
+        videoFile: 'video_short.mp4'
+      }
+      await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+
+      await waitJobs(servers)
+    })
+
+    it('Should have the video updated', async function () {
+      for (const server of servers) {
+        const video = await server.videos.get({ id: videoUUID })
+        expect(video.files).to.have.lengthOf(1)
+        expect(video.streamingPlaylists).to.have.lengthOf(0)
+
+        const { body } = await makeRawRequest({ url: video.files[0].fileUrl, expectedStatus: HttpStatusCode.OK_200 })
+        expect(body).to.deep.equal(await readFile(buildAbsoluteFixturePath('video_short.mp4')))
+      }
+    })
+
+    it('Should have 4 lower resolution to transcode', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(4)
+
+      for (const resolution of [ 480, 360, 240, 144 ]) {
+        const job = availableJobs.find(j => j.payload.output.resolution === resolution)
+        expect(job).to.exist
+        expect(job.type).to.equal('vod-web-video-transcoding')
+
+        if (resolution === 240) jobUUID = job.uuid
+      }
+    })
+
+    it('Should process one of these transcoding jobs', async function () {
+      const { job } = await servers[0].runnerJobs.accept<RunnerJobVODWebVideoTranscodingPayload>({ runnerToken, jobUUID })
+      jobToken = job.jobToken
+
+      const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+      const inputFile = await readFile(buildAbsoluteFixturePath('video_short.mp4'))
+
+      expect(body).to.deep.equal(inputFile)
+
+      const payload: VODWebVideoTranscodingSuccess = { videoFile: `video_short_${job.payload.output.resolution}p.mp4` }
+      await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+    })
+
+    it('Should process all other jobs', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(3)
+
+      for (const resolution of [ 480, 360, 144 ]) {
+        const availableJob = availableJobs.find(j => j.payload.output.resolution === resolution)
+        expect(availableJob).to.exist
+        jobUUID = availableJob.uuid
+
+        const { job } = await servers[0].runnerJobs.accept<RunnerJobVODWebVideoTranscodingPayload>({ runnerToken, jobUUID })
+        jobToken = job.jobToken
+
+        const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+        const inputFile = await readFile(buildAbsoluteFixturePath('video_short.mp4'))
+        expect(body).to.deep.equal(inputFile)
+
+        const payload: VODWebVideoTranscodingSuccess = { videoFile: `video_short_${resolution}p.mp4` }
+        await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+      }
+
+      await waitJobs(servers)
+    })
+
+    it('Should have the video updated', async function () {
+      for (const server of servers) {
+        const video = await server.videos.get({ id: videoUUID })
+        expect(video.files).to.have.lengthOf(5)
+        expect(video.streamingPlaylists).to.have.lengthOf(0)
+
+        const { body } = await makeRawRequest({ url: video.files[0].fileUrl, expectedStatus: HttpStatusCode.OK_200 })
+        expect(body).to.deep.equal(await readFile(buildAbsoluteFixturePath('video_short.mp4')))
+
+        for (const file of video.files) {
+          await makeRawRequest({ url: file.fileUrl, expectedStatus: HttpStatusCode.OK_200 })
+          await makeRawRequest({ url: file.torrentUrl, expectedStatus: HttpStatusCode.OK_200 })
+        }
+      }
+    })
+
+    it('Should not have available jobs anymore', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(0)
+    })
+  })
+
+  describe('HLS transcoding only', function () {
+    let videoUUID: string
+    let jobToken: string
+    let jobUUID: string
+
+    before(async function () {
+      this.timeout(60000)
+
+      await servers[0].config.enableTranscoding({ hls: true, webVideo: false })
+
+      const { uuid } = await servers[0].videos.quickUpload({ name: 'hls video', fixture: 'video_short.webm' })
+      videoUUID = uuid
+
+      await waitJobs(servers)
+    })
+
+    it('Should run the optimize job', async function () {
+      this.timeout(60000)
+
+      await servers[0].runnerJobs.autoProcessWebVideoJob(runnerToken)
+    })
+
+    it('Should have 5 HLS resolution to transcode', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(5)
+
+      for (const resolution of [ 720, 480, 360, 240, 144 ]) {
+        const job = availableJobs.find(j => j.payload.output.resolution === resolution)
+        expect(job).to.exist
+        expect(job.type).to.equal('vod-hls-transcoding')
+
+        if (resolution === 480) jobUUID = job.uuid
+      }
+    })
+
+    it('Should process one of these transcoding jobs', async function () {
+      this.timeout(60000)
+
+      const { job } = await servers[0].runnerJobs.accept<RunnerJobVODHLSTranscodingPayload>({ runnerToken, jobUUID })
+      jobToken = job.jobToken
+
+      const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+      const inputFile = await readFile(buildAbsoluteFixturePath('video_short.mp4'))
+
+      expect(body).to.deep.equal(inputFile)
+
+      const payload: VODHLSTranscodingSuccess = {
+        videoFile: 'video_short_480p.mp4',
+        resolutionPlaylistFile: 'video_short_480p.m3u8'
+      }
+      await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+
+      await waitJobs(servers)
+    })
+
+    it('Should have the video updated', async function () {
+      for (const server of servers) {
+        const video = await server.videos.get({ id: videoUUID })
+
+        expect(video.files).to.have.lengthOf(1)
+        expect(video.streamingPlaylists).to.have.lengthOf(1)
+
+        const hls = video.streamingPlaylists[0]
+        expect(hls.files).to.have.lengthOf(1)
+
+        await completeCheckHlsPlaylist({ videoUUID, hlsOnly: false, servers, resolutions: [ 480 ] })
+      }
+    })
+
+    it('Should process all other jobs', async function () {
+      this.timeout(60000)
+
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(4)
+
+      let maxQualityFile = 'video_short.mp4'
+
+      for (const resolution of [ 720, 360, 240, 144 ]) {
+        const availableJob = availableJobs.find(j => j.payload.output.resolution === resolution)
+        expect(availableJob).to.exist
+        jobUUID = availableJob.uuid
+
+        const { job } = await servers[0].runnerJobs.accept<RunnerJobVODHLSTranscodingPayload>({ runnerToken, jobUUID })
+        jobToken = job.jobToken
+
+        const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+        const inputFile = await readFile(buildAbsoluteFixturePath(maxQualityFile))
+        expect(body).to.deep.equal(inputFile)
+
+        const payload: VODHLSTranscodingSuccess = {
+          videoFile: `video_short_${resolution}p.mp4`,
+          resolutionPlaylistFile: `video_short_${resolution}p.m3u8`
+        }
+        await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+
+        if (resolution === 720) {
+          maxQualityFile = 'video_short_720p.mp4'
+        }
+      }
+
+      await waitJobs(servers)
+    })
+
+    it('Should have the video updated', async function () {
+      for (const server of servers) {
+        const video = await server.videos.get({ id: videoUUID })
+
+        expect(video.files).to.have.lengthOf(0)
+        expect(video.streamingPlaylists).to.have.lengthOf(1)
+
+        const hls = video.streamingPlaylists[0]
+        expect(hls.files).to.have.lengthOf(5)
+
+        await completeCheckHlsPlaylist({ videoUUID, hlsOnly: true, servers, resolutions: [ 720, 480, 360, 240, 144 ] })
+      }
+    })
+
+    it('Should not have available jobs anymore', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(0)
+    })
+  })
+
+  describe('Web video and HLS transcoding', function () {
+
+    before(async function () {
+      this.timeout(60000)
+
+      await servers[0].config.enableTranscoding({ hls: true, webVideo: true })
+
+      await servers[0].videos.quickUpload({ name: 'web video and hls video', fixture: 'video_short.webm' })
+
+      await waitJobs(servers)
+    })
+
+    it('Should process the first optimize job', async function () {
+      this.timeout(60000)
+
+      await servers[0].runnerJobs.autoProcessWebVideoJob(runnerToken)
+    })
+
+    it('Should have 9 jobs to process', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+
+      expect(availableJobs).to.have.lengthOf(9)
+
+      const webVideoJobs = availableJobs.filter(j => j.type === 'vod-web-video-transcoding')
+      const hlsJobs = availableJobs.filter(j => j.type === 'vod-hls-transcoding')
+
+      expect(webVideoJobs).to.have.lengthOf(4)
+      expect(hlsJobs).to.have.lengthOf(5)
+    })
+
+    it('Should process all available jobs', async function () {
+      await processAllJobs(servers[0], runnerToken)
+    })
+  })
+
+  describe('Audio merge transcoding', function () {
+    let videoUUID: string
+    let jobToken: string
+    let jobUUID: string
+
+    before(async function () {
+      this.timeout(60000)
+
+      await servers[0].config.enableTranscoding({ hls: true, webVideo: true })
+
+      const attributes = { name: 'audio_with_preview', previewfile: 'custom-preview.jpg', fixture: 'sample.ogg' }
+      const { uuid } = await servers[0].videos.upload({ attributes, mode: 'legacy' })
+      videoUUID = uuid
+
+      await waitJobs(servers)
+    })
+
+    it('Should have an audio merge transcoding job', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(1)
+
+      expect(availableJobs[0].type).to.equal('vod-audio-merge-transcoding')
+
+      jobUUID = availableJobs[0].uuid
+    })
+
+    it('Should have a valid remote audio merge transcoding job', async function () {
+      const { job } = await servers[0].runnerJobs.accept<RunnerJobVODAudioMergeTranscodingPayload>({ runnerToken, jobUUID })
+      jobToken = job.jobToken
+
+      expect(job.type === 'vod-audio-merge-transcoding')
+      expect(job.payload.input.audioFileUrl).to.exist
+      expect(job.payload.input.previewFileUrl).to.exist
+      expect(job.payload.output.resolution).to.equal(480)
+
+      {
+        const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.audioFileUrl, jobToken, runnerToken })
+        const inputFile = await readFile(buildAbsoluteFixturePath('sample.ogg'))
+        expect(body).to.deep.equal(inputFile)
+      }
+
+      {
+        const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.previewFileUrl, jobToken, runnerToken })
+
+        const video = await servers[0].videos.get({ id: videoUUID })
+        const { body: inputFile } = await makeGetRequest({
+          url: servers[0].url,
+          path: video.previewPath,
+          expectedStatus: HttpStatusCode.OK_200
+        })
+
+        expect(body).to.deep.equal(inputFile)
+      }
+    })
+
+    it('Should merge the audio', async function () {
+      this.timeout(60000)
+
+      const payload: VODAudioMergeTranscodingSuccess = { videoFile: 'video_short_480p.mp4' }
+      await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+
+      await waitJobs(servers)
+    })
+
+    it('Should have the video updated', async function () {
+      for (const server of servers) {
+        const video = await server.videos.get({ id: videoUUID })
+        expect(video.files).to.have.lengthOf(1)
+        expect(video.streamingPlaylists).to.have.lengthOf(0)
+
+        const { body } = await makeRawRequest({ url: video.files[0].fileUrl, expectedStatus: HttpStatusCode.OK_200 })
+        expect(body).to.deep.equal(await readFile(buildAbsoluteFixturePath('video_short_480p.mp4')))
+      }
+    })
+
+    it('Should have 7 lower resolutions to transcode', async function () {
+      const { availableJobs } = await servers[0].runnerJobs.requestVOD({ runnerToken })
+      expect(availableJobs).to.have.lengthOf(7)
+
+      for (const resolution of [ 360, 240, 144 ]) {
+        const jobs = availableJobs.filter(j => j.payload.output.resolution === resolution)
+        expect(jobs).to.have.lengthOf(2)
+      }
+
+      jobUUID = availableJobs.find(j => j.payload.output.resolution === 480).uuid
+    })
+
+    it('Should process one other job', async function () {
+      this.timeout(60000)
+
+      const { job } = await servers[0].runnerJobs.accept<RunnerJobVODHLSTranscodingPayload>({ runnerToken, jobUUID })
+      jobToken = job.jobToken
+
+      const { body } = await servers[0].runnerJobs.getJobFile({ url: job.payload.input.videoFileUrl, jobToken, runnerToken })
+      const inputFile = await readFile(buildAbsoluteFixturePath('video_short_480p.mp4'))
+      expect(body).to.deep.equal(inputFile)
+
+      const payload: VODHLSTranscodingSuccess = {
+        videoFile: `video_short_480p.mp4`,
+        resolutionPlaylistFile: `video_short_480p.m3u8`
+      }
+      await servers[0].runnerJobs.success({ runnerToken, jobUUID, jobToken, payload })
+
+      await waitJobs(servers)
+    })
+
+    it('Should have the video updated', async function () {
+      for (const server of servers) {
+        const video = await server.videos.get({ id: videoUUID })
+
+        expect(video.files).to.have.lengthOf(1)
+        expect(video.streamingPlaylists).to.have.lengthOf(1)
+
+        const hls = video.streamingPlaylists[0]
+        expect(hls.files).to.have.lengthOf(1)
+
+        await completeCheckHlsPlaylist({ videoUUID, hlsOnly: false, servers, resolutions: [ 480 ] })
+      }
+    })
+
+    it('Should process all available jobs', async function () {
+      await processAllJobs(servers[0], runnerToken)
+    })
+  })
+
+  after(async function () {
+    await cleanupTests(servers)
+  })
+})