Resumable video uploads (#3933)

* WIP: resumable video uploads

relates to #324

* fix review comments

* video upload: error handling

* fix audio upload

* fixes after self review

* Update server/controllers/api/videos/index.ts

Co-authored-by: Rigel Kent <par@rigelk.eu>

* Update server/middlewares/validators/videos/videos.ts

Co-authored-by: Rigel Kent <par@rigelk.eu>

* Update server/controllers/api/videos/index.ts

Co-authored-by: Rigel Kent <par@rigelk.eu>

* update after code review

* refactor upload route

- restore multipart upload route
- move resumable to dedicated upload-resumable route
- move checks to middleware
- do not leak internal fs structure in response

* fix yarn.lock upon rebase

* factorize addVideo for reuse in both endpoints

* add resumable upload API to openapi spec

* add initial test and test helper for resumable upload

* typings for videoAddResumable middleware

* avoid including aws and google packages via node-uploadx, by only including uploadx/core

* rename ex-isAudioBg to more explicit name mentioning it is a preview file for audio

* add video-upload-tmp-folder-cleaner job

* stronger typing of video upload middleware

* reduce dependency to @uploadx/core

* add audio upload test

* refactor resumable uploads cleanup from job to scheduler

* refactor resumable uploads scheduler to compare to last execution time

* make resumable upload validator to always cleanup on failure

* move legacy upload request building outside of uploadVideo test helper

* filter upload-resumable middlewares down to POST, PUT, DELETE

also begin to type metadata

* merge add duration functions

* stronger typings and documentation for uploadx behaviour, move init validator up

* refactor(client/video-edit): options > uploadxOptions

* refactor(client/video-edit): remove obsolete else

* scheduler/remove-dangling-resum: rename tag

* refactor(server/video): add UploadVideoFiles type

* refactor(mw/validators): restructure eslint disable

* refactor(mw/validators/videos): rename import

* refactor(client/vid-upload): rename html elem id

* refactor(sched/remove-dangl): move fn to method

* refactor(mw/async): add method typing

* refactor(mw/vali/video): double quote > single

* refactor(server/upload-resum): express use > all

* proper http methud enum server/middlewares/async.ts

* properly type http methods

* factorize common video upload validation steps

* add check for maximum partially uploaded file size

* fix audioBg use

* fix extname(filename) in addVideo

* document parameters for uploadx's resumable protocol

* clear META files in scheduler

* last audio refactor before cramming preview in the initial POST form data

* refactor as mulitpart/form-data initial post request

this allows preview/thumbnail uploads alongside the initial request,
and cleans up the upload form

* Add more tests for resumable uploads

* Refactor remove dangling resumable uploads

* Prepare changelog

* Add more resumable upload tests

* Remove user quota check for resumable uploads

* Fix upload error handler

* Update nginx template for upload-resumable

* Cleanup comment

* Remove unused express methods

* Prefer to use got instead of raw http

* Don't retry on error 500

Co-authored-by: Rigel Kent <par@rigelk.eu>
Co-authored-by: Rigel Kent <sendmemail@rigelk.eu>
Co-authored-by: Chocobozzz <me@florianbigard.com>

4 files changed, 230 insertions, 59 deletions

diff --git a/shared/extra-utils/server/debug.ts b/shared/extra-utils/server/debug.ts
index 5cf80a5fb..f196812b7 100644
--- a/shared/extra-utils/server/debug.ts
+++ b/shared/extra-utils/server/debug.ts
@@ -1,5 +1,6 @@
-import { makeGetRequest } from '../requests/requests'
+import { makeGetRequest, makePostBodyRequest } from '../requests/requests'
 import { HttpStatusCode } from '../../core-utils/miscs/http-error-codes'
+import { SendDebugCommand } from '@shared/models'
 
 function getDebug (url: string, token: string) {
   const path = '/api/v1/server/debug'
@@ -12,8 +13,21 @@ function getDebug (url: string, token: string) {
   })
 }
 
+function sendDebugCommand (url: string, token: string, body: SendDebugCommand) {
+  const path = '/api/v1/server/debug/run-command'
+
+  return makePostBodyRequest({
+    url,
+    path,
+    token,
+    fields: body,
+    statusCodeExpected: HttpStatusCode.NO_CONTENT_204
+  })
+}
+
 // ---------------------------------------------------------------------------
 
 export {
-  getDebug
+  getDebug,
+  sendDebugCommand
 }
diff --git a/shared/extra-utils/server/servers.ts b/shared/extra-utils/server/servers.ts
index 779a3cc36..479f08e12 100644
--- a/shared/extra-utils/server/servers.ts
+++ b/shared/extra-utils/server/servers.ts
@@ -274,7 +274,7 @@ async function reRunServer (server: ServerInfo, configOverride?: any) {
 }
 
 async function checkTmpIsEmpty (server: ServerInfo) {
-  await checkDirectoryIsEmpty(server, 'tmp', [ 'plugins-global.css', 'hls' ])
+  await checkDirectoryIsEmpty(server, 'tmp', [ 'plugins-global.css', 'hls', 'resumable-uploads' ])
 
   if (await pathExists(join('test' + server.internalServerNumber, 'tmp', 'hls'))) {
     await checkDirectoryIsEmpty(server, 'tmp/hls')
diff --git a/shared/extra-utils/videos/video-channels.ts b/shared/extra-utils/videos/video-channels.ts
index d0dfb5856..0aab93e52 100644
--- a/shared/extra-utils/videos/video-channels.ts
+++ b/shared/extra-utils/videos/video-channels.ts
@@ -5,7 +5,7 @@ import { VideoChannelUpdate } from '../../models/videos/channel/video-channel-up
 import { VideoChannelCreate } from '../../models/videos/channel/video-channel-create.model'
 import { makeDeleteRequest, makeGetRequest, updateImageRequest } from '../requests/requests'
 import { ServerInfo } from '../server/servers'
-import { User } from '../../models/users/user.model'
+import { MyUser, User } from '../../models/users/user.model'
 import { getMyUserInformation } from '../users/users'
 import { HttpStatusCode } from '../../../shared/core-utils/miscs/http-error-codes'
 
@@ -170,6 +170,12 @@ function setDefaultVideoChannel (servers: ServerInfo[]) {
   return Promise.all(tasks)
 }
 
+async function getDefaultVideoChannel (url: string, token: string) {
+  const res = await getMyUserInformation(url, token)
+
+  return (res.body as MyUser).videoChannels[0].id
+}
+
 // ---------------------------------------------------------------------------
 
 export {
@@ -181,5 +187,6 @@ export {
   deleteVideoChannel,
   getVideoChannel,
   setDefaultVideoChannel,
-  deleteVideoChannelImage
+  deleteVideoChannelImage,
+  getDefaultVideoChannel
 }
diff --git a/shared/extra-utils/videos/videos.ts b/shared/extra-utils/videos/videos.ts
index a0143b0ef..e88256ac0 100644
--- a/shared/extra-utils/videos/videos.ts
+++ b/shared/extra-utils/videos/videos.ts
@@ -1,7 +1,8 @@
 /* eslint-disable @typescript-eslint/no-unused-expressions,@typescript-eslint/no-floating-promises */
 
 import { expect } from 'chai'
-import { pathExists, readdir, readFile } from 'fs-extra'
+import { createReadStream, pathExists, readdir, readFile, stat } from 'fs-extra'
+import got, { Response as GotResponse } from 'got/dist/source'
 import * as parseTorrent from 'parse-torrent'
 import { extname, join } from 'path'
 import * as request from 'supertest'
@@ -42,6 +43,7 @@ type VideoAttributes = {
   channelId?: number
   privacy?: VideoPrivacy
   fixture?: string
+  support?: string
   thumbnailfile?: string
   previewfile?: string
   scheduleUpdate?: {
@@ -364,8 +366,13 @@ async function checkVideoFilesWereRemoved (
   }
 }
 
-async function uploadVideo (url: string, accessToken: string, videoAttributesArg: VideoAttributes, specialStatus = HttpStatusCode.OK_200) {
-  const path = '/api/v1/videos/upload'
+async function uploadVideo (
+  url: string,
+  accessToken: string,
+  videoAttributesArg: VideoAttributes,
+  specialStatus = HttpStatusCode.OK_200,
+  mode: 'legacy' | 'resumable' = 'legacy'
+) {
   let defaultChannelId = '1'
 
   try {
@@ -391,74 +398,170 @@ async function uploadVideo (url: string, accessToken: string, videoAttributesArg
     fixture: 'video_short.webm'
   }, videoAttributesArg)
 
+  const res = mode === 'legacy'
+    ? await buildLegacyUpload(url, accessToken, attributes, specialStatus)
+    : await buildResumeUpload(url, accessToken, attributes, specialStatus)
+
+  // Wait torrent generation
+  if (specialStatus === HttpStatusCode.OK_200) {
+    let video: VideoDetails
+    do {
+      const resVideo = await getVideoWithToken(url, accessToken, res.body.video.uuid)
+      video = resVideo.body
+
+      await wait(50)
+    } while (!video.files[0].torrentUrl)
+  }
+
+  return res
+}
+
+function checkUploadVideoParam (
+  url: string,
+  token: string,
+  attributes: Partial<VideoAttributes>,
+  specialStatus = HttpStatusCode.OK_200,
+  mode: 'legacy' | 'resumable' = 'legacy'
+) {
+  return mode === 'legacy'
+    ? buildLegacyUpload(url, token, attributes, specialStatus)
+    : buildResumeUpload(url, token, attributes, specialStatus)
+}
+
+async function buildLegacyUpload (url: string, token: string, attributes: VideoAttributes, specialStatus = HttpStatusCode.OK_200) {
+  const path = '/api/v1/videos/upload'
   const req = request(url)
               .post(path)
               .set('Accept', 'application/json')
-              .set('Authorization', 'Bearer ' + accessToken)
-              .field('name', attributes.name)
-              .field('nsfw', JSON.stringify(attributes.nsfw))
-              .field('commentsEnabled', JSON.stringify(attributes.commentsEnabled))
-              .field('downloadEnabled', JSON.stringify(attributes.downloadEnabled))
-              .field('waitTranscoding', JSON.stringify(attributes.waitTranscoding))
-              .field('privacy', attributes.privacy.toString())
-              .field('channelId', attributes.channelId)
-
-  if (attributes.support !== undefined) {
-    req.field('support', attributes.support)
-  }
+              .set('Authorization', 'Bearer ' + token)
 
-  if (attributes.description !== undefined) {
-    req.field('description', attributes.description)
-  }
-  if (attributes.language !== undefined) {
-    req.field('language', attributes.language.toString())
-  }
-  if (attributes.category !== undefined) {
-    req.field('category', attributes.category.toString())
-  }
-  if (attributes.licence !== undefined) {
-    req.field('licence', attributes.licence.toString())
-  }
+  buildUploadReq(req, attributes)
 
-  const tags = attributes.tags || []
-  for (let i = 0; i < tags.length; i++) {
-    req.field('tags[' + i + ']', attributes.tags[i])
+  if (attributes.fixture !== undefined) {
+    req.attach('videofile', buildAbsoluteFixturePath(attributes.fixture))
   }
 
-  if (attributes.thumbnailfile !== undefined) {
-    req.attach('thumbnailfile', buildAbsoluteFixturePath(attributes.thumbnailfile))
-  }
-  if (attributes.previewfile !== undefined) {
-    req.attach('previewfile', buildAbsoluteFixturePath(attributes.previewfile))
-  }
+  return req.expect(specialStatus)
+}
 
-  if (attributes.scheduleUpdate) {
-    req.field('scheduleUpdate[updateAt]', attributes.scheduleUpdate.updateAt)
+async function buildResumeUpload (url: string, token: string, attributes: VideoAttributes, specialStatus = HttpStatusCode.OK_200) {
+  let size = 0
+  let videoFilePath: string
+  let mimetype = 'video/mp4'
 
-    if (attributes.scheduleUpdate.privacy) {
-      req.field('scheduleUpdate[privacy]', attributes.scheduleUpdate.privacy)
+  if (attributes.fixture) {
+    videoFilePath = buildAbsoluteFixturePath(attributes.fixture)
+    size = (await stat(videoFilePath)).size
+
+    if (videoFilePath.endsWith('.mkv')) {
+      mimetype = 'video/x-matroska'
+    } else if (videoFilePath.endsWith('.webm')) {
+      mimetype = 'video/webm'
     }
   }
 
-  if (attributes.originallyPublishedAt !== undefined) {
-    req.field('originallyPublishedAt', attributes.originallyPublishedAt)
+  const initializeSessionRes = await prepareResumableUpload({ url, token, attributes, size, mimetype })
+  const initStatus = initializeSessionRes.status
+
+  if (videoFilePath && initStatus === HttpStatusCode.CREATED_201) {
+    const locationHeader = initializeSessionRes.header['location']
+    expect(locationHeader).to.not.be.undefined
+
+    const pathUploadId = locationHeader.split('?')[1]
+
+    return sendResumableChunks({ url, token, pathUploadId, videoFilePath, size, specialStatus })
   }
 
-  const res = await req.attach('videofile', buildAbsoluteFixturePath(attributes.fixture))
-            .expect(specialStatus)
+  const expectedInitStatus = specialStatus === HttpStatusCode.OK_200
+    ? HttpStatusCode.CREATED_201
+    : specialStatus
 
-  // Wait torrent generation
-  if (specialStatus === HttpStatusCode.OK_200) {
-    let video: VideoDetails
-    do {
-      const resVideo = await getVideoWithToken(url, accessToken, res.body.video.uuid)
-      video = resVideo.body
+  expect(initStatus).to.equal(expectedInitStatus)
 
-      await wait(50)
-    } while (!video.files[0].torrentUrl)
+  return initializeSessionRes
+}
+
+async function prepareResumableUpload (options: {
+  url: string
+  token: string
+  attributes: VideoAttributes
+  size: number
+  mimetype: string
+}) {
+  const { url, token, attributes, size, mimetype } = options
+
+  const path = '/api/v1/videos/upload-resumable'
+
+  const req = request(url)
+              .post(path)
+              .set('Authorization', 'Bearer ' + token)
+              .set('X-Upload-Content-Type', mimetype)
+              .set('X-Upload-Content-Length', size.toString())
+
+  buildUploadReq(req, attributes)
+
+  if (attributes.fixture) {
+    req.field('filename', attributes.fixture)
   }
 
-  return res
+  return req
+}
+
+function sendResumableChunks (options: {
+  url: string
+  token: string
+  pathUploadId: string
+  videoFilePath: string
+  size: number
+  specialStatus?: HttpStatusCode
+  contentLength?: number
+  contentRangeBuilder?: (start: number, chunk: any) => string
+}) {
+  const { url, token, pathUploadId, videoFilePath, size, specialStatus, contentLength, contentRangeBuilder } = options
+
+  const expectedStatus = specialStatus || HttpStatusCode.OK_200
+
+  const path = '/api/v1/videos/upload-resumable'
+  let start = 0
+
+  const readable = createReadStream(videoFilePath, { highWaterMark: 8 * 1024 })
+  return new Promise<GotResponse>((resolve, reject) => {
+    readable.on('data', async function onData (chunk) {
+      readable.pause()
+
+      const headers = {
+        'Authorization': 'Bearer ' + token,
+        'Content-Type': 'application/octet-stream',
+        'Content-Range': contentRangeBuilder
+          ? contentRangeBuilder(start, chunk)
+          : `bytes ${start}-${start + chunk.length - 1}/${size}`,
+        'Content-Length': contentLength ? contentLength + '' : chunk.length + ''
+      }
+
+      const res = await got({
+        url,
+        method: 'put',
+        headers,
+        path: path + '?' + pathUploadId,
+        body: chunk,
+        responseType: 'json',
+        throwHttpErrors: false
+      })
+
+      start += chunk.length
+
+      if (res.statusCode === expectedStatus) {
+        return resolve(res)
+      }
+
+      if (res.statusCode !== HttpStatusCode.PERMANENT_REDIRECT_308) {
+        readable.off('data', onData)
+        return reject(new Error('Incorrect transient behaviour sending intermediary chunks'))
+      }
+
+      readable.resume()
+    })
+  })
 }
 
 function updateVideo (
@@ -749,11 +852,13 @@ export {
   getVideoWithToken,
   getVideosList,
   removeAllVideos,
+  checkUploadVideoParam,
   getVideosListPagination,
   getVideosListSort,
   removeVideo,
   getVideosListWithToken,
   uploadVideo,
+  sendResumableChunks,
   getVideosWithFilters,
   uploadRandomVideoOnServers,
   updateVideo,
@@ -767,5 +872,50 @@ export {
   getMyVideosWithFilter,
   uploadVideoAndGetId,
   getLocalIdByUUID,
-  getVideoIdFromUUID
+  getVideoIdFromUUID,
+  prepareResumableUpload
+}
+
+// ---------------------------------------------------------------------------
+
+function buildUploadReq (req: request.Test, attributes: VideoAttributes) {
+
+  for (const key of [ 'name', 'support', 'channelId', 'description', 'originallyPublishedAt' ]) {
+    if (attributes[key] !== undefined) {
+      req.field(key, attributes[key])
+    }
+  }
+
+  for (const key of [ 'nsfw', 'commentsEnabled', 'downloadEnabled', 'waitTranscoding' ]) {
+    if (attributes[key] !== undefined) {
+      req.field(key, JSON.stringify(attributes[key]))
+    }
+  }
+
+  for (const key of [ 'language', 'privacy', 'category', 'licence' ]) {
+    if (attributes[key] !== undefined) {
+      req.field(key, attributes[key].toString())
+    }
+  }
+
+  const tags = attributes.tags || []
+  for (let i = 0; i < tags.length; i++) {
+    req.field('tags[' + i + ']', attributes.tags[i])
+  }
+
+  for (const key of [ 'thumbnailfile', 'previewfile' ]) {
+    if (attributes[key] !== undefined) {
+      req.attach(key, buildAbsoluteFixturePath(attributes[key]))
+    }
+  }
+
+  if (attributes.scheduleUpdate) {
+    if (attributes.scheduleUpdate.updateAt) {
+      req.field('scheduleUpdate[updateAt]', attributes.scheduleUpdate.updateAt)
+    }
+
+    if (attributes.scheduleUpdate.privacy) {
+      req.field('scheduleUpdate[privacy]', attributes.scheduleUpdate.privacy)
+    }
+  }
 }