19 files changed, 404 insertions, 48 deletions

diff --git a/server/lib/activitypub/index.ts b/server/lib/activitypub/index.ts
new file mode 100644
index 000000000..740800606
--- /dev/null
+++ b/server/lib/activitypub/index.ts
@@ -0,0 +1,3 @@
+export * from './process-create'
+export * from './process-flag'
+export * from './process-update'
diff --git a/server/lib/activitypub/process-create.ts b/server/lib/activitypub/process-create.ts
new file mode 100644
index 000000000..114ff1848
--- /dev/null
+++ b/server/lib/activitypub/process-create.ts
@@ -0,0 +1,104 @@
+import {
+  ActivityCreate,
+  VideoTorrentObject,
+  VideoChannelObject
+} from '../../../shared'
+import { database as db } from '../../initializers'
+import { logger, retryTransactionWrapper } from '../../helpers'
+
+function processCreateActivity (activity: ActivityCreate) {
+  const activityObject = activity.object
+  const activityType = activityObject.type
+
+  if (activityType === 'Video') {
+    return processCreateVideo(activityObject as VideoTorrentObject)
+  } else if (activityType === 'VideoChannel') {
+    return processCreateVideoChannel(activityObject as VideoChannelObject)
+  }
+
+  logger.warn('Unknown activity object type %s when creating activity.', activityType, { activity: activity.id })
+  return Promise.resolve()
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  processCreateActivity
+}
+
+// ---------------------------------------------------------------------------
+
+function processCreateVideo (video: VideoTorrentObject) {
+  const options = {
+    arguments: [ video ],
+    errorMessage: 'Cannot insert the remote video with many retries.'
+  }
+
+  return retryTransactionWrapper(addRemoteVideo, options)
+}
+
+async function addRemoteVideo (videoToCreateData: VideoTorrentObject) {
+  logger.debug('Adding remote video %s.', videoToCreateData.url)
+
+  await db.sequelize.transaction(async t => {
+    const sequelizeOptions = {
+      transaction: t
+    }
+
+    const videoFromDatabase = await db.Video.loadByUUID(videoToCreateData.uuid)
+    if (videoFromDatabase) throw new Error('UUID already exists.')
+
+    const videoChannel = await db.VideoChannel.loadByHostAndUUID(fromPod.host, videoToCreateData.channelUUID, t)
+    if (!videoChannel) throw new Error('Video channel ' + videoToCreateData.channelUUID + ' not found.')
+
+    const tags = videoToCreateData.tags
+    const tagInstances = await db.Tag.findOrCreateTags(tags, t)
+
+    const videoData = {
+      name: videoToCreateData.name,
+      uuid: videoToCreateData.uuid,
+      category: videoToCreateData.category,
+      licence: videoToCreateData.licence,
+      language: videoToCreateData.language,
+      nsfw: videoToCreateData.nsfw,
+      description: videoToCreateData.truncatedDescription,
+      channelId: videoChannel.id,
+      duration: videoToCreateData.duration,
+      createdAt: videoToCreateData.createdAt,
+      // FIXME: updatedAt does not seems to be considered by Sequelize
+      updatedAt: videoToCreateData.updatedAt,
+      views: videoToCreateData.views,
+      likes: videoToCreateData.likes,
+      dislikes: videoToCreateData.dislikes,
+      remote: true,
+      privacy: videoToCreateData.privacy
+    }
+
+    const video = db.Video.build(videoData)
+    await db.Video.generateThumbnailFromData(video, videoToCreateData.thumbnailData)
+    const videoCreated = await video.save(sequelizeOptions)
+
+    const tasks = []
+    for (const fileData of videoToCreateData.files) {
+      const videoFileInstance = db.VideoFile.build({
+        extname: fileData.extname,
+        infoHash: fileData.infoHash,
+        resolution: fileData.resolution,
+        size: fileData.size,
+        videoId: videoCreated.id
+      })
+
+      tasks.push(videoFileInstance.save(sequelizeOptions))
+    }
+
+    await Promise.all(tasks)
+
+    await videoCreated.setTags(tagInstances, sequelizeOptions)
+  })
+
+  logger.info('Remote video with uuid %s inserted.', videoToCreateData.uuid)
+}
+
+function processCreateVideoChannel (videoChannel: VideoChannelObject) {
+
+}
diff --git a/server/lib/activitypub/process-flag.ts b/server/lib/activitypub/process-flag.ts
new file mode 100644
index 000000000..6fa862ee9
--- /dev/null
+++ b/server/lib/activitypub/process-flag.ts
@@ -0,0 +1,17 @@
+import {
+  ActivityCreate,
+  VideoTorrentObject,
+  VideoChannelObject
+} from '../../../shared'
+
+function processFlagActivity (activity: ActivityCreate) {
+  // empty
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  processFlagActivity
+}
+
+// ---------------------------------------------------------------------------
diff --git a/server/lib/activitypub/process-update.ts b/server/lib/activitypub/process-update.ts
new file mode 100644
index 000000000..187c7be7c
--- /dev/null
+++ b/server/lib/activitypub/process-update.ts
@@ -0,0 +1,29 @@
+import {
+  ActivityCreate,
+  VideoTorrentObject,
+  VideoChannelObject
+} from '../../../shared'
+
+function processUpdateActivity (activity: ActivityCreate) {
+  if (activity.object.type === 'Video') {
+    return processUpdateVideo(activity.object)
+  } else if (activity.object.type === 'VideoChannel') {
+    return processUpdateVideoChannel(activity.object)
+  }
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  processUpdateActivity
+}
+
+// ---------------------------------------------------------------------------
+
+function processUpdateVideo (video: VideoTorrentObject) {
+
+}
+
+function processUpdateVideoChannel (videoChannel: VideoChannelObject) {
+
+}
diff --git a/server/lib/activitypub/send-request.ts b/server/lib/activitypub/send-request.ts
new file mode 100644
index 000000000..6a31c226d
--- /dev/null
+++ b/server/lib/activitypub/send-request.ts
@@ -0,0 +1,129 @@
+import * as Sequelize from 'sequelize'
+
+import {
+  AccountInstance,
+  VideoInstance,
+  VideoChannelInstance
+} from '../../models'
+import { httpRequestJobScheduler } from '../jobs'
+import { signObject, activityPubContextify } from '../../helpers'
+import { Activity } from '../../../shared'
+
+function sendCreateVideoChannel (videoChannel: VideoChannelInstance, t: Sequelize.Transaction) {
+  const videoChannelObject = videoChannel.toActivityPubObject()
+  const data = createActivityData(videoChannel.url, videoChannel.Account, videoChannelObject)
+
+  return broadcastToFollowers(data, t)
+}
+
+function sendUpdateVideoChannel (videoChannel: VideoChannelInstance, t: Sequelize.Transaction) {
+  const videoChannelObject = videoChannel.toActivityPubObject()
+  const data = updateActivityData(videoChannel.url, videoChannel.Account, videoChannelObject)
+
+  return broadcastToFollowers(data, t)
+}
+
+function sendDeleteVideoChannel (videoChannel: VideoChannelInstance, t: Sequelize.Transaction) {
+  const videoChannelObject = videoChannel.toActivityPubObject()
+  const data = deleteActivityData(videoChannel.url, videoChannel.Account, videoChannelObject)
+
+  return broadcastToFollowers(data, t)
+}
+
+function sendAddVideo (video: VideoInstance, t: Sequelize.Transaction) {
+  const videoObject = video.toActivityPubObject()
+  const data = addActivityData(video.url, video.VideoChannel.Account, video.VideoChannel.url, videoObject)
+
+  return broadcastToFollowers(data, t)
+}
+
+function sendUpdateVideo (video: VideoInstance, t: Sequelize.Transaction) {
+  const videoObject = video.toActivityPubObject()
+  const data = updateActivityData(video.url, video.VideoChannel.Account, videoObject)
+
+  return broadcastToFollowers(data, t)
+}
+
+function sendDeleteVideo (video: VideoInstance, t: Sequelize.Transaction) {
+  const videoObject = video.toActivityPubObject()
+  const data = deleteActivityData(video.url, video.VideoChannel.Account, videoObject)
+
+  return broadcastToFollowers(data, t)
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+
+}
+
+// ---------------------------------------------------------------------------
+
+function broadcastToFollowers (data: any, t: Sequelize.Transaction) {
+  return httpRequestJobScheduler.createJob(t, 'http-request', 'httpRequestBroadcastHandler', data)
+}
+
+function buildSignedActivity (byAccount: AccountInstance, data: Object) {
+  const activity = activityPubContextify(data)
+
+  return signObject(byAccount, activity) as Promise<Activity>
+}
+
+async function getPublicActivityTo (account: AccountInstance) {
+  const inboxUrls = await account.getFollowerSharedInboxUrls()
+
+  return inboxUrls.concat('https://www.w3.org/ns/activitystreams#Public')
+}
+
+async function createActivityData (url: string, byAccount: AccountInstance, object: any) {
+  const to = await getPublicActivityTo(byAccount)
+  const base = {
+    type: 'Create',
+    id: url,
+    actor: byAccount.url,
+    to,
+    object
+  }
+
+  return buildSignedActivity(byAccount, base)
+}
+
+async function updateActivityData (url: string, byAccount: AccountInstance, object: any) {
+  const to = await getPublicActivityTo(byAccount)
+  const base = {
+    type: 'Update',
+    id: url,
+    actor: byAccount.url,
+    to,
+    object
+  }
+
+  return buildSignedActivity(byAccount, base)
+}
+
+async function deleteActivityData (url: string, byAccount: AccountInstance, object: any) {
+  const to = await getPublicActivityTo(byAccount)
+  const base = {
+    type: 'Update',
+    id: url,
+    actor: byAccount.url,
+    to,
+    object
+  }
+
+  return buildSignedActivity(byAccount, base)
+}
+
+async function addActivityData (url: string, byAccount: AccountInstance, target: string, object: any) {
+  const to = await getPublicActivityTo(byAccount)
+  const base = {
+    type: 'Add',
+    id: url,
+    actor: byAccount.url,
+    to,
+    object,
+    target
+  }
+
+  return buildSignedActivity(byAccount, base)
+}
diff --git a/server/lib/index.ts b/server/lib/index.ts
index d1534b085..bfb415ad2 100644
--- a/server/lib/index.ts
+++ b/server/lib/index.ts
@@ -1,3 +1,4 @@
+export * from './activitypub'
 export * from './cache'
 export * from './jobs'
 export * from './request'
diff --git a/server/lib/jobs/handlers/index.ts b/server/lib/jobs/handlers/index.ts
deleted file mode 100644
index cef1f89a9..000000000
--- a/server/lib/jobs/handlers/index.ts
+++ /dev/null
@@ -1,17 +0,0 @@
-import * as videoFileOptimizer from './video-file-optimizer'
-import * as videoFileTranscoder from './video-file-transcoder'
-
-export interface JobHandler<T> {
-  process (data: object, jobId: number): T
-  onError (err: Error, jobId: number)
-  onSuccess (jobId: number, jobResult: T)
-}
-
-const jobHandlers: { [ handlerName: string ]: JobHandler<any> } = {
-  videoFileOptimizer,
-  videoFileTranscoder
-}
-
-export {
-  jobHandlers
-}
diff --git a/server/lib/jobs/http-request-job-scheduler/http-request-broadcast-handler.ts b/server/lib/jobs/http-request-job-scheduler/http-request-broadcast-handler.ts
new file mode 100644
index 000000000..6b6946d02
--- /dev/null
+++ b/server/lib/jobs/http-request-job-scheduler/http-request-broadcast-handler.ts
@@ -0,0 +1,25 @@
+import * as Bluebird from 'bluebird'
+
+import { database as db } from '../../../initializers/database'
+import { logger } from '../../../helpers'
+
+async function process (data: { videoUUID: string }, jobId: number) {
+
+}
+
+function onError (err: Error, jobId: number) {
+  logger.error('Error when optimized video file in job %d.', jobId, err)
+  return Promise.resolve()
+}
+
+async function onSuccess (jobId: number) {
+
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  process,
+  onError,
+  onSuccess
+}
diff --git a/server/lib/jobs/http-request-job-scheduler/http-request-job-scheduler.ts b/server/lib/jobs/http-request-job-scheduler/http-request-job-scheduler.ts
new file mode 100644
index 000000000..42cb9139c
--- /dev/null
+++ b/server/lib/jobs/http-request-job-scheduler/http-request-job-scheduler.ts
@@ -0,0 +1,17 @@
+import { JobScheduler, JobHandler } from '../job-scheduler'
+
+import * as httpRequestBroadcastHandler from './http-request-broadcast-handler'
+import * as httpRequestUnicastHandler from './http-request-unicast-handler'
+import { JobCategory } from '../../../../shared'
+
+const jobHandlers: { [ handlerName: string ]: JobHandler<any> } = {
+  httpRequestBroadcastHandler,
+  httpRequestUnicastHandler
+}
+const jobCategory: JobCategory = 'http-request'
+
+const httpRequestJobScheduler = new JobScheduler(jobCategory, jobHandlers)
+
+export {
+  httpRequestJobScheduler
+}
diff --git a/server/lib/jobs/http-request-job-scheduler/http-request-unicast-handler.ts b/server/lib/jobs/http-request-job-scheduler/http-request-unicast-handler.ts
new file mode 100644
index 000000000..6b6946d02
--- /dev/null
+++ b/server/lib/jobs/http-request-job-scheduler/http-request-unicast-handler.ts
@@ -0,0 +1,25 @@
+import * as Bluebird from 'bluebird'
+
+import { database as db } from '../../../initializers/database'
+import { logger } from '../../../helpers'
+
+async function process (data: { videoUUID: string }, jobId: number) {
+
+}
+
+function onError (err: Error, jobId: number) {
+  logger.error('Error when optimized video file in job %d.', jobId, err)
+  return Promise.resolve()
+}
+
+async function onSuccess (jobId: number) {
+
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  process,
+  onError,
+  onSuccess
+}
diff --git a/server/lib/jobs/http-request-job-scheduler/index.ts b/server/lib/jobs/http-request-job-scheduler/index.ts
new file mode 100644
index 000000000..4d2573296
--- /dev/null
+++ b/server/lib/jobs/http-request-job-scheduler/index.ts
@@ -0,0 +1 @@
+export * from './http-request-job-scheduler'
diff --git a/server/lib/jobs/index.ts b/server/lib/jobs/index.ts
index b18a3d845..a92743707 100644
--- a/server/lib/jobs/index.ts
+++ b/server/lib/jobs/index.ts
@@ -1 +1,2 @@
-export * from './job-scheduler'
+export * from './http-request-job-scheduler'
+export * from './transcoding-job-scheduler'
diff --git a/server/lib/jobs/job-scheduler.ts b/server/lib/jobs/job-scheduler.ts
index 61d483268..89a4bca88 100644
--- a/server/lib/jobs/job-scheduler.ts
+++ b/server/lib/jobs/job-scheduler.ts
@@ -1,39 +1,41 @@
 import { AsyncQueue, forever, queue } from 'async'
 import * as Sequelize from 'sequelize'
 
-import { database as db } from '../../initializers/database'
 import {
+  database as db,
   JOBS_FETCHING_INTERVAL,
   JOBS_FETCH_LIMIT_PER_CYCLE,
   JOB_STATES
 } from '../../initializers'
 import { logger } from '../../helpers'
 import { JobInstance } from '../../models'
-import { JobHandler, jobHandlers } from './handlers'
+import { JobCategory } from '../../../shared'
 
+export interface JobHandler<T> {
+  process (data: object, jobId: number): T
+  onError (err: Error, jobId: number)
+  onSuccess (jobId: number, jobResult: T)
+}
 type JobQueueCallback = (err: Error) => void
 
-class JobScheduler {
-
-  private static instance: JobScheduler
+class JobScheduler<T> {
 
-  private constructor () { }
-
-  static get Instance () {
-    return this.instance || (this.instance = new this())
-  }
+  constructor (
+    private jobCategory: JobCategory,
+    private jobHandlers: { [ id: string ]: JobHandler<T> }
+  ) {}
 
   async activate () {
-    const limit = JOBS_FETCH_LIMIT_PER_CYCLE
+    const limit = JOBS_FETCH_LIMIT_PER_CYCLE[this.jobCategory]
 
-    logger.info('Jobs scheduler activated.')
+    logger.info('Jobs scheduler %s activated.', this.jobCategory)
 
     const jobsQueue = queue<JobInstance, JobQueueCallback>(this.processJob.bind(this))
 
     // Finish processing jobs from a previous start
     const state = JOB_STATES.PROCESSING
     try {
-      const jobs = await db.Job.listWithLimit(limit, state)
+      const jobs = await db.Job.listWithLimitByCategory(limit, state, this.jobCategory)
 
       this.enqueueJobs(jobsQueue, jobs)
     } catch (err) {
@@ -49,7 +51,7 @@ class JobScheduler {
 
         const state = JOB_STATES.PENDING
         try {
-          const jobs = await db.Job.listWithLimit(limit, state)
+          const jobs = await db.Job.listWithLimitByCategory(limit, state, this.jobCategory)
 
           this.enqueueJobs(jobsQueue, jobs)
         } catch (err) {
@@ -64,9 +66,10 @@ class JobScheduler {
     )
   }
 
-  createJob (transaction: Sequelize.Transaction, handlerName: string, handlerInputData: object) {
+  createJob (transaction: Sequelize.Transaction, category: JobCategory, handlerName: string, handlerInputData: object) {
     const createQuery = {
       state: JOB_STATES.PENDING,
+      category,
       handlerName,
       handlerInputData
     }
@@ -80,7 +83,7 @@ class JobScheduler {
   }
 
   private async processJob (job: JobInstance, callback: (err: Error) => void) {
-    const jobHandler = jobHandlers[job.handlerName]
+    const jobHandler = this.jobHandlers[job.handlerName]
     if (jobHandler === undefined) {
       logger.error('Unknown job handler for job %s.', job.handlerName)
       return callback(null)
diff --git a/server/lib/jobs/transcoding-job-scheduler/index.ts b/server/lib/jobs/transcoding-job-scheduler/index.ts
new file mode 100644
index 000000000..73152a1be
--- /dev/null
+++ b/server/lib/jobs/transcoding-job-scheduler/index.ts
@@ -0,0 +1 @@
+export * from './transcoding-job-scheduler'
diff --git a/server/lib/jobs/transcoding-job-scheduler/transcoding-job-scheduler.ts b/server/lib/jobs/transcoding-job-scheduler/transcoding-job-scheduler.ts
new file mode 100644
index 000000000..d7c614fb8
--- /dev/null
+++ b/server/lib/jobs/transcoding-job-scheduler/transcoding-job-scheduler.ts
@@ -0,0 +1,17 @@
+import { JobScheduler, JobHandler } from '../job-scheduler'
+
+import * as videoFileOptimizer from './video-file-optimizer-handler'
+import * as videoFileTranscoder from './video-file-transcoder-handler'
+import { JobCategory } from '../../../../shared'
+
+const jobHandlers: { [ handlerName: string ]: JobHandler<any> } = {
+  videoFileOptimizer,
+  videoFileTranscoder
+}
+const jobCategory: JobCategory = 'transcoding'
+
+const transcodingJobScheduler = new JobScheduler(jobCategory, jobHandlers)
+
+export {
+  transcodingJobScheduler
+}
diff --git a/server/lib/jobs/handlers/video-file-optimizer.ts b/server/lib/jobs/transcoding-job-scheduler/video-file-optimizer-handler.ts
index ccded4721..ccded4721 100644
--- a/server/lib/jobs/handlers/video-file-optimizer.ts
+++ b/server/lib/jobs/transcoding-job-scheduler/video-file-optimizer-handler.ts
diff --git a/server/lib/jobs/handlers/video-file-transcoder.ts b/server/lib/jobs/transcoding-job-scheduler/video-file-transcoder-handler.ts
index 853645510..853645510 100644
--- a/server/lib/jobs/handlers/video-file-transcoder.ts
+++ b/server/lib/jobs/transcoding-job-scheduler/video-file-transcoder-handler.ts
diff --git a/server/lib/user.ts b/server/lib/user.ts
index a92f4777b..57c653e55 100644
--- a/server/lib/user.ts
+++ b/server/lib/user.ts
@@ -1,9 +1,9 @@
 import { database as db } from '../initializers'
 import { UserInstance } from '../models'
-import { addVideoAuthorToFriends } from './friends'
+import { addVideoAccountToFriends } from './friends'
 import { createVideoChannel } from './video-channel'
 
-async function createUserAuthorAndChannel (user: UserInstance, validateUser = true) {
+async function createUserAccountAndChannel (user: UserInstance, validateUser = true) {
   const res = await db.sequelize.transaction(async t => {
     const userOptions = {
       transaction: t,
@@ -11,25 +11,25 @@ async function createUserAuthorAndChannel (user: UserInstance, validateUser = tr
     }
 
     const userCreated = await user.save(userOptions)
-    const authorInstance = db.Author.build({
+    const accountInstance = db.Account.build({
       name: userCreated.username,
       podId: null, // It is our pod
       userId: userCreated.id
     })
 
-    const authorCreated = await authorInstance.save({ transaction: t })
+    const accountCreated = await accountInstance.save({ transaction: t })
 
-    const remoteVideoAuthor = authorCreated.toAddRemoteJSON()
+    const remoteVideoAccount = accountCreated.toAddRemoteJSON()
 
     // Now we'll add the video channel's meta data to our friends
-    const author = await addVideoAuthorToFriends(remoteVideoAuthor, t)
+    const account = await addVideoAccountToFriends(remoteVideoAccount, t)
 
     const videoChannelInfo = {
       name: `Default ${userCreated.username} channel`
     }
-    const videoChannel = await createVideoChannel(videoChannelInfo, authorCreated, t)
+    const videoChannel = await createVideoChannel(videoChannelInfo, accountCreated, t)
 
-    return { author, videoChannel }
+    return { account, videoChannel }
   })
 
   return res
@@ -38,5 +38,5 @@ async function createUserAuthorAndChannel (user: UserInstance, validateUser = tr
 // ---------------------------------------------------------------------------
 
 export {
-  createUserAuthorAndChannel
+  createUserAccountAndChannel
 }
diff --git a/server/lib/video-channel.ts b/server/lib/video-channel.ts
index 678ffe643..a6dd4d061 100644
--- a/server/lib/video-channel.ts
+++ b/server/lib/video-channel.ts
@@ -3,15 +3,15 @@ import * as Sequelize from 'sequelize'
 import { addVideoChannelToFriends } from './friends'
 import { database as db } from '../initializers'
 import { logger } from '../helpers'
-import { AuthorInstance } from '../models'
+import { AccountInstance } from '../models'
 import { VideoChannelCreate } from '../../shared/models'
 
-async function createVideoChannel (videoChannelInfo: VideoChannelCreate, author: AuthorInstance, t: Sequelize.Transaction) {
+async function createVideoChannel (videoChannelInfo: VideoChannelCreate, account: AccountInstance, t: Sequelize.Transaction) {
   const videoChannelData = {
     name: videoChannelInfo.name,
     description: videoChannelInfo.description,
     remote: false,
-    authorId: author.id
+    authorId: account.id
   }
 
   const videoChannel = db.VideoChannel.build(videoChannelData)
@@ -19,8 +19,8 @@ async function createVideoChannel (videoChannelInfo: VideoChannelCreate, author:
 
   const videoChannelCreated = await videoChannel.save(options)
 
-  // Do not forget to add Author information to the created video channel
-  videoChannelCreated.Author = author
+  // Do not forget to add Account information to the created video channel
+  videoChannelCreated.Account = account
 
   const remoteVideoChannel = videoChannelCreated.toAddRemoteJSON()