Add create-import-video-file-job command

package.json

@@ -40,6 +40,7 @@
     "start": "node dist/server",
     "update-host": "node ./dist/scripts/update-host.js",
     "create-transcoding-job": "node ./dist/scripts/create-transcoding-job.js",
+    "create-import-video-file-job": "node ./dist/scripts/create-import-video-file-job.js",
     "test": "scripty",
     "help": "scripty",
     "generate-api-doc": "scripty",

scripts/create-import-video-file-job.ts

@@ -0,0 +1,39 @@
+import * as program from 'commander'
+import { resolve } from 'path'
+import { VideoModel } from '../server/models/video/video'
+import { initDatabaseModels } from '../server/initializers'
+import { JobQueue } from '../server/lib/job-queue'
+
+program
+  .option('-v, --video [videoUUID]', 'Video UUID')
+  .option('-i, --import [videoFile]', 'Video file')
+  .description('Import a video file to replace an already uploaded file or to add a new resolution')
+  .parse(process.argv)
+
+if (program['video'] === undefined || program['import'] === undefined) {
+  console.error('All parameters are mandatory.')
+  process.exit(-1)
+}
+
+run()
+  .then(() => process.exit(0))
+  .catch(err => {
+    console.error(err)
+    process.exit(-1)
+  })
+
+async function run () {
+  await initDatabaseModels(true)
+
+  const video = await VideoModel.loadByUUID(program['video'])
+  if (!video) throw new Error('Video not found.')
+
+  const dataInput = {
+    videoUUID: video.uuid,
+    filePath: resolve(program['import'])
+  }
+
+  await JobQueue.Instance.init()
+  await JobQueue.Instance.createJob({ type: 'video-file-import', payload: dataInput })
+  console.log('Import job for video %s created.', video.uuid)
+}

server/helpers/core-utils.ts

@@ -6,7 +6,7 @@
 import * as bcrypt from 'bcrypt'
 import * as createTorrent from 'create-torrent'
 import { pseudoRandomBytes } from 'crypto'
-import { readdir, readFile, rename, stat, Stats, unlink, writeFile } from 'fs'
+import { copyFile, readdir, readFile, rename, stat, Stats, unlink, writeFile } from 'fs'
 import * as mkdirp from 'mkdirp'
 import { isAbsolute, join } from 'path'
 import * as pem from 'pem'
@@ -136,6 +136,7 @@ function promisify2WithVoid<T, U> (func: (arg1: T, arg2: U, cb: (err: any) => vo
   }
 }
 
+const copyFilePromise = promisify2WithVoid<string, string>(copyFile)
 const readFileBufferPromise = promisify1<string, Buffer>(readFile)
 const unlinkPromise = promisify1WithVoid<string>(unlink)
 const renamePromise = promisify2WithVoid<string, string>(rename)
@@ -167,6 +168,7 @@ export {
   promisify0,
   promisify1,
 
+  copyFilePromise,
   readdirPromise,
   readFileBufferPromise,
   unlinkPromise,

server/initializers/constants.ts

@@ -74,6 +74,7 @@ const JOB_ATTEMPTS: { [ id in JobType ]: number } = {
   'activitypub-http-unicast': 5,
   'activitypub-http-fetcher': 5,
   'activitypub-follow': 5,
+  'video-file-import': 1,
   'video-file': 1,
   'email': 5
 }
@@ -82,6 +83,7 @@ const JOB_CONCURRENCY: { [ id in JobType ]: number } = {
   'activitypub-http-unicast': 5,
   'activitypub-http-fetcher': 1,
   'activitypub-follow': 3,
+  'video-file-import': 1,
   'video-file': 1,
   'email': 5
 }

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

@@ -16,6 +16,28 @@ export type VideoFilePayload = {
   isPortraitMode?: boolean
 }
 
+export type VideoImportPayload = {
+  videoUUID: string,
+  filePath: string
+}
+
+async function processVideoImport (job: kue.Job) {
+  const payload = job.data as VideoImportPayload
+  logger.info('Processing video import in job %d.', job.id)
+
+  const video = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(payload.videoUUID)
+  // No video, maybe deleted?
+  if (!video) {
+    logger.info('Do not process job %d, video does not exist.', job.id, { videoUUID: video.uuid })
+    return undefined
+  }
+
+  await video.importVideoFile(payload.filePath)
+
+  await onVideoFileTranscoderOrImportSuccess(video)
+  return video
+}
+
 async function processVideoFile (job: kue.Job) {
   const payload = job.data as VideoFilePayload
   logger.info('Processing video file in job %d.', job.id)
@@ -30,7 +52,7 @@ async function processVideoFile (job: kue.Job) {
   // Transcoding in other resolution
   if (payload.resolution) {
     await video.transcodeOriginalVideofile(payload.resolution, payload.isPortraitMode)
-    await onVideoFileTranscoderSuccess(video)
+    await onVideoFileTranscoderOrImportSuccess(video)
   } else {
     await video.optimizeOriginalVideofile()
     await onVideoFileOptimizerSuccess(video, payload.isNewVideo)
@@ -39,7 +61,7 @@ async function processVideoFile (job: kue.Job) {
   return video
 }
 
-async function onVideoFileTranscoderSuccess (video: VideoModel) {
+async function onVideoFileTranscoderOrImportSuccess (video: VideoModel) {
   if (video === undefined) return undefined
 
   // Maybe the video changed in database, refresh it
@@ -109,5 +131,6 @@ async function onVideoFileOptimizerSuccess (video: VideoModel, isNewVideo: boole
 // ---------------------------------------------------------------------------
 
 export {
-  processVideoFile
+  processVideoFile,
+  processVideoImport
 }

server/lib/job-queue/job-queue.ts

@@ -7,7 +7,7 @@ import { ActivitypubHttpBroadcastPayload, processActivityPubHttpBroadcast } from
 import { ActivitypubHttpFetcherPayload, processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher'
 import { ActivitypubHttpUnicastPayload, processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast'
 import { EmailPayload, processEmail } from './handlers/email'
-import { processVideoFile, VideoFilePayload } from './handlers/video-file'
+import { processVideoFile, processVideoImport, VideoFilePayload, VideoImportPayload } from './handlers/video-file'
 import { ActivitypubFollowPayload, processActivityPubFollow } from './handlers/activitypub-follow'
 
 type CreateJobArgument =
@@ -15,6 +15,7 @@ type CreateJobArgument =
   { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } |
   { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } |
   { type: 'activitypub-follow', payload: ActivitypubFollowPayload } |
+  { type: 'video-file-import', payload: VideoImportPayload } |
   { type: 'video-file', payload: VideoFilePayload } |
   { type: 'email', payload: EmailPayload }
 
@@ -23,6 +24,7 @@ const handlers: { [ id in JobType ]: (job: kue.Job) => Promise<any>} = {
   'activitypub-http-unicast': processActivityPubHttpUnicast,
   'activitypub-http-fetcher': processActivityPubHttpFetcher,
   'activitypub-follow': processActivityPubFollow,
+  'video-file-import': processVideoImport,
   'video-file': processVideoFile,
   'email': processEmail
 }

server/models/video/video.ts

@@ -2,7 +2,7 @@ import * as Bluebird from 'bluebird'
 import { map, maxBy } from 'lodash'
 import * as magnetUtil from 'magnet-uri'
 import * as parseTorrent from 'parse-torrent'
-import { join } from 'path'
+import { join, extname } from 'path'
 import * as Sequelize from 'sequelize'
 import {
   AllowNull,
@@ -32,6 +32,7 @@ import { VideoFilter } from '../../../shared/models/videos/video-query.type'
 import {
   createTorrentPromise,
   peertubeTruncate,
+  copyFilePromise,
   renamePromise,
   statPromise,
   unlinkPromise,
@@ -1315,6 +1316,38 @@ export class VideoModel extends Model<VideoModel> {
     this.VideoFiles.push(newVideoFile)
   }
 
+  async importVideoFile (inputFilePath: string) {
+    let updatedVideoFile = new VideoFileModel({
+      resolution: (await getVideoFileResolution(inputFilePath)).videoFileResolution,
+      extname: extname(inputFilePath),
+      size: (await statPromise(inputFilePath)).size,
+      videoId: this.id
+    })
+
+    const outputPath = this.getVideoFilePath(updatedVideoFile)
+    await copyFilePromise(inputFilePath, outputPath)
+
+    const currentVideoFile = this.VideoFiles.find(videoFile => videoFile.resolution === updatedVideoFile.resolution)
+    const isNewVideoFile = !currentVideoFile
+
+    if (!isNewVideoFile) {
+      if (currentVideoFile.extname !== updatedVideoFile.extname) {
+        await this.removeFile(currentVideoFile)
+        currentVideoFile.set('extname', updatedVideoFile.extname)
+      }
+      currentVideoFile.set('size', updatedVideoFile.size)
+      updatedVideoFile = currentVideoFile
+    }
+
+    await this.createTorrentAndSetInfoHash(updatedVideoFile)
+
+    await updatedVideoFile.save()
+
+    if (isNewVideoFile) {
+      this.VideoFiles.push(updatedVideoFile)
+    }
+  }
+
   getOriginalFileResolution () {
     const originalFilePath = this.getVideoFilePath(this.getOriginalFile())
 

server/tests/cli/create-import-video-file-job.ts

@@ -0,0 +1,111 @@
+/* tslint:disable:no-unused-expression */
+
+import 'mocha'
+import * as chai from 'chai'
+import { VideoDetails, VideoFile } from '../../../shared/models/videos'
+const expect = chai.expect
+
+import {
+  execCLI,
+  flushTests,
+  getEnvCli,
+  getVideosList,
+  killallServers,
+  parseTorrentVideo,
+  runServer,
+  ServerInfo,
+  setAccessTokensToServers,
+  uploadVideo,
+  wait,
+  getVideo, flushAndRunMultipleServers, doubleFollow
+} from '../utils'
+
+function assertVideoProperties (video: VideoFile, resolution: number, extname: string) {
+  expect(video).to.have.nested.property('resolution.id', resolution)
+  expect(video).to.have.property('magnetUri').that.includes(`.${extname}`)
+  expect(video).to.have.property('torrentUrl').that.includes(`-${resolution}.torrent`)
+  expect(video).to.have.property('fileUrl').that.includes(`.${extname}`)
+  expect(video).to.have.property('size').that.is.above(0)
+}
+
+describe('Test create import video jobs', function () {
+  this.timeout(60000)
+
+  let servers: ServerInfo[] = []
+  let video1UUID: string
+  let video2UUID: string
+
+  before(async function () {
+    this.timeout(90000)
+    await flushTests()
+
+    // Run server 2 to have transcoding enabled
+    servers = await flushAndRunMultipleServers(2)
+    await setAccessTokensToServers(servers)
+
+    await doubleFollow(servers[0], servers[1])
+
+    // Upload two videos for our needs
+    const res1 = await uploadVideo(servers[0].url, servers[0].accessToken, { name: 'video1' })
+    video1UUID = res1.body.video.uuid
+    const res2 = await uploadVideo(servers[1].url, servers[1].accessToken, { name: 'video2' })
+    video2UUID = res2.body.video.uuid
+
+    await wait(40000)
+  })
+
+  it('Should run a import job on video 1 with a lower resolution', async function () {
+    const env = getEnvCli(servers[0])
+    await execCLI(`${env} npm run create-import-video-file-job -- -v ${video1UUID} -i server/tests/api/fixtures/video_short-480.webm`)
+
+    await wait(30000)
+
+    for (const server of servers) {
+      const { data: videos } = (await getVideosList(server.url)).body
+      expect(videos).to.have.lengthOf(2)
+
+      let infoHashes: { [ id: number ]: string } = {}
+
+      const video = videos.find(({ uuid }) => uuid === video1UUID)
+      const videoDetail: VideoDetails = (await getVideo(server.url, video.uuid)).body
+
+      expect(videoDetail.files).to.have.lengthOf(2)
+      const [originalVideo, transcodedVideo] = videoDetail.files
+      assertVideoProperties(originalVideo, 720, 'webm')
+      assertVideoProperties(transcodedVideo, 480, 'webm')
+    }
+  })
+
+  it('Should run a import job on video 2 with the same resolution', async function () {
+    const env = getEnvCli(servers[1])
+    await execCLI(`${env} npm run create-import-video-file-job -- -v ${video2UUID} -i server/tests/api/fixtures/video_short.ogv`)
+
+    await wait(30000)
+
+    for (const server of servers.reverse()) {
+      const { data: videos } = (await getVideosList(server.url)).body
+      expect(videos).to.have.lengthOf(2)
+
+      let infoHashes: { [ id: number ]: string }
+
+      const video = videos.find(({ uuid }) => uuid === video2UUID)
+      const videoDetail: VideoDetails = (await getVideo(server.url, video.uuid)).body
+
+      expect(videoDetail.files).to.have.lengthOf(4)
+      const [originalVideo, transcodedVideo420, transcodedVideo320, transcodedVideo240] = videoDetail.files
+      assertVideoProperties(originalVideo, 720, 'ogv')
+      assertVideoProperties(transcodedVideo420, 480, 'mp4')
+      assertVideoProperties(transcodedVideo320, 360, 'mp4')
+      assertVideoProperties(transcodedVideo240, 240, 'mp4')
+    }
+  })
+
+  after(async function () {
+    killallServers(servers)
+
+    // Keep the logs if the test failed
+    if (this['ok']) {
+      await flushTests()
+    }
+  })
+})

server/tests/cli/index.ts

@@ -1,4 +1,5 @@
 // Order of the tests we want to execute
 import './create-transcoding-job'
+import './create-import-video-file-job'
 import './reset-password'
 import './update-host'

shared/models/server/job.model.ts

@@ -4,6 +4,7 @@ export type JobType = 'activitypub-http-unicast' |
   'activitypub-http-broadcast' |
   'activitypub-http-fetcher' |
   'activitypub-follow' |
+  'video-file-import' |
   'video-file' |
   'email'