27 files changed, 367 insertions, 489 deletions

diff --git a/server/lib/activitypub/actor.ts b/server/lib/activitypub/actor.ts
index c708b38ba..712de7d0d 100644
--- a/server/lib/activitypub/actor.ts
+++ b/server/lib/activitypub/actor.ts
@@ -64,7 +64,11 @@ async function getOrCreateActorAndServerAndModel (actorUrl: string, recurseIfNee
     actor = await retryTransactionWrapper(saveActorAndServerAndModelIfNotExist, options)
   }
 
-  return refreshActorIfNeeded(actor)
+  const options = {
+    arguments: [ actor ],
+    errorMessage: 'Cannot refresh actor if needed with many retries.'
+  }
+  return retryTransactionWrapper(refreshActorIfNeeded, options)
 }
 
 function buildActorInstance (type: ActivityPubActorType, url: string, preferredUsername: string, uuid?: string) {
@@ -325,38 +329,43 @@ async function saveVideoChannel (actor: ActorModel, result: FetchRemoteActorResu
 async function refreshActorIfNeeded (actor: ActorModel) {
   if (!actor.isOutdated()) return actor
 
-  const actorUrl = await getUrlFromWebfinger(actor.preferredUsername, actor.getHost())
-  const result = await fetchRemoteActor(actorUrl)
-  if (result === undefined) {
-    logger.warn('Cannot fetch remote actor in refresh actor.')
-    return actor
-  }
-
-  return sequelizeTypescript.transaction(async t => {
-    updateInstanceWithAnother(actor, result.actor)
-
-    if (result.avatarName !== undefined) {
-      await updateActorAvatarInstance(actor, result.avatarName, t)
+  try {
+    const actorUrl = await getUrlFromWebfinger(actor.preferredUsername, actor.getHost())
+    const result = await fetchRemoteActor(actorUrl)
+    if (result === undefined) {
+      logger.warn('Cannot fetch remote actor in refresh actor.')
+      return actor
     }
 
-    // Force update
-    actor.setDataValue('updatedAt', new Date())
-    await actor.save({ transaction: t })
+    return sequelizeTypescript.transaction(async t => {
+      updateInstanceWithAnother(actor, result.actor)
 
-    if (actor.Account) {
-      await actor.save({ transaction: t })
+      if (result.avatarName !== undefined) {
+        await updateActorAvatarInstance(actor, result.avatarName, t)
+      }
 
-      actor.Account.set('name', result.name)
-      await actor.Account.save({ transaction: t })
-    } else if (actor.VideoChannel) {
+      // Force update
+      actor.setDataValue('updatedAt', new Date())
       await actor.save({ transaction: t })
 
-      actor.VideoChannel.set('name', result.name)
-      await actor.VideoChannel.save({ transaction: t })
-    }
+      if (actor.Account) {
+        await actor.save({ transaction: t })
+
+        actor.Account.set('name', result.name)
+        await actor.Account.save({ transaction: t })
+      } else if (actor.VideoChannel) {
+        await actor.save({ transaction: t })
+
+        actor.VideoChannel.set('name', result.name)
+        await actor.VideoChannel.save({ transaction: t })
+      }
 
+      return actor
+    })
+  } catch (err) {
+    logger.warn('Cannot refresh actor.', err)
     return actor
-  })
+  }
 }
 
 function normalizeActor (actor: any) {
diff --git a/server/lib/activitypub/fetch.ts b/server/lib/activitypub/fetch.ts
index 4fc97cc38..b1b370a1a 100644
--- a/server/lib/activitypub/fetch.ts
+++ b/server/lib/activitypub/fetch.ts
@@ -1,13 +1,12 @@
-import { Transaction } from 'sequelize'
 import { ActorModel } from '../../models/activitypub/actor'
-import { activitypubHttpJobScheduler, ActivityPubHttpPayload } from '../jobs/activitypub-http-job-scheduler'
+import { JobQueue } from '../job-queue'
 
-async function addFetchOutboxJob (actor: ActorModel, t: Transaction) {
-  const jobPayload: ActivityPubHttpPayload = {
+async function addFetchOutboxJob (actor: ActorModel) {
+  const payload = {
     uris: [ actor.outboxUrl ]
   }
 
-  return activitypubHttpJobScheduler.createJob(t, 'activitypubHttpFetcherHandler', jobPayload)
+  return JobQueue.Instance.createJob({ type: 'activitypub-http-fetcher', payload })
 }
 
 export {
diff --git a/server/lib/activitypub/process/process-accept.ts b/server/lib/activitypub/process/process-accept.ts
index 551f09ea7..7db2f8ff0 100644
--- a/server/lib/activitypub/process/process-accept.ts
+++ b/server/lib/activitypub/process/process-accept.ts
@@ -26,6 +26,6 @@ async function processAccept (actor: ActorModel, targetActor: ActorModel) {
   if (follow.state !== 'accepted') {
     follow.set('state', 'accepted')
     await follow.save()
-    await addFetchOutboxJob(targetActor, undefined)
+    await addFetchOutboxJob(targetActor)
   }
 }
diff --git a/server/lib/activitypub/process/process-follow.ts b/server/lib/activitypub/process/process-follow.ts
index 69f5c51b5..dc1d542b5 100644
--- a/server/lib/activitypub/process/process-follow.ts
+++ b/server/lib/activitypub/process/process-follow.ts
@@ -63,7 +63,7 @@ async function follow (actor: ActorModel, targetActorURL: string) {
     actorFollow.ActorFollowing = targetActor
 
     // Target sends to actor he accepted the follow request
-    return sendAccept(actorFollow, t)
+    return sendAccept(actorFollow)
   })
 
   logger.info('Actor %s is followed by actor %s.', targetActorURL, actor.url)
diff --git a/server/lib/activitypub/send/misc.ts b/server/lib/activitypub/send/misc.ts
index dc0d3de57..7a21f0c94 100644
--- a/server/lib/activitypub/send/misc.ts
+++ b/server/lib/activitypub/send/misc.ts
@@ -7,7 +7,7 @@ import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
 import { VideoModel } from '../../../models/video/video'
 import { VideoCommentModel } from '../../../models/video/video-comment'
 import { VideoShareModel } from '../../../models/video/video-share'
-import { activitypubHttpJobScheduler, ActivityPubHttpPayload } from '../../jobs/activitypub-http-job-scheduler'
+import { JobQueue } from '../../job-queue'
 
 async function forwardActivity (
   activity: Activity,
@@ -35,12 +35,11 @@ async function forwardActivity (
 
   logger.debug('Creating forwarding job.', { uris })
 
-  const jobPayload: ActivityPubHttpPayload = {
+  const payload = {
     uris,
     body: activity
   }
-
-  return activitypubHttpJobScheduler.createJob(t, 'activitypubHttpBroadcastHandler', jobPayload)
+  return JobQueue.Instance.createJob({ type: 'activitypub-http-broadcast', payload })
 }
 
 async function broadcastToFollowers (
@@ -51,44 +50,43 @@ async function broadcastToFollowers (
   actorsException: ActorModel[] = []
 ) {
   const uris = await computeFollowerUris(toActorFollowers, actorsException, t)
-  return broadcastTo(uris, data, byActor, t)
+  return broadcastTo(uris, data, byActor)
 }
 
 async function broadcastToActors (
   data: any,
   byActor: ActorModel,
   toActors: ActorModel[],
-  t: Transaction,
   actorsException: ActorModel[] = []
 ) {
   const uris = await computeUris(toActors, actorsException)
-  return broadcastTo(uris, data, byActor, t)
+  return broadcastTo(uris, data, byActor)
 }
 
-async function broadcastTo (uris: string[], data: any, byActor: ActorModel, t: Transaction) {
+async function broadcastTo (uris: string[], data: any, byActor: ActorModel) {
   if (uris.length === 0) return undefined
 
   logger.debug('Creating broadcast job.', { uris })
 
-  const jobPayload: ActivityPubHttpPayload = {
+  const payload = {
     uris,
     signatureActorId: byActor.id,
     body: data
   }
 
-  return activitypubHttpJobScheduler.createJob(t, 'activitypubHttpBroadcastHandler', jobPayload)
+  return JobQueue.Instance.createJob({ type: 'activitypub-http-broadcast', payload })
 }
 
-async function unicastTo (data: any, byActor: ActorModel, toActorUrl: string, t: Transaction) {
+async function unicastTo (data: any, byActor: ActorModel, toActorUrl: string) {
   logger.debug('Creating unicast job.', { uri: toActorUrl })
 
-  const jobPayload: ActivityPubHttpPayload = {
-    uris: [ toActorUrl ],
+  const payload = {
+    uri: toActorUrl,
     signatureActorId: byActor.id,
     body: data
   }
 
-  return activitypubHttpJobScheduler.createJob(t, 'activitypubHttpUnicastHandler', jobPayload)
+  return JobQueue.Instance.createJob({ type: 'activitypub-http-unicast', payload })
 }
 
 function getOriginVideoAudience (video: VideoModel, actorsInvolvedInVideo: ActorModel[]) {
diff --git a/server/lib/activitypub/send/send-accept.ts b/server/lib/activitypub/send/send-accept.ts
index 4eaa329d9..064fd88d2 100644
--- a/server/lib/activitypub/send/send-accept.ts
+++ b/server/lib/activitypub/send/send-accept.ts
@@ -1,4 +1,3 @@
-import { Transaction } from 'sequelize'
 import { ActivityAccept, ActivityFollow } from '../../../../shared/models/activitypub'
 import { ActorModel } from '../../../models/activitypub/actor'
 import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
@@ -6,7 +5,7 @@ import { getActorFollowAcceptActivityPubUrl, getActorFollowActivityPubUrl } from
 import { unicastTo } from './misc'
 import { followActivityData } from './send-follow'
 
-async function sendAccept (actorFollow: ActorFollowModel, t: Transaction) {
+async function sendAccept (actorFollow: ActorFollowModel) {
   const follower = actorFollow.ActorFollower
   const me = actorFollow.ActorFollowing
 
@@ -16,7 +15,7 @@ async function sendAccept (actorFollow: ActorFollowModel, t: Transaction) {
   const url = getActorFollowAcceptActivityPubUrl(actorFollow)
   const data = acceptActivityData(url, me, followData)
 
-  return unicastTo(data, me, follower.inboxUrl, t)
+  return unicastTo(data, me, follower.inboxUrl)
 }
 
 // ---------------------------------------------------------------------------
diff --git a/server/lib/activitypub/send/send-announce.ts b/server/lib/activitypub/send/send-announce.ts
index 578fbc630..93b5668d2 100644
--- a/server/lib/activitypub/send/send-announce.ts
+++ b/server/lib/activitypub/send/send-announce.ts
@@ -42,7 +42,7 @@ async function sendVideoAnnounceToOrigin (byActor: ActorModel, video: VideoModel
   const audience = getOriginVideoAudience(video, actorsInvolvedInVideo)
   const data = await createActivityData(url, byActor, announcedActivity, t, audience)
 
-  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl, t)
+  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl)
 }
 
 async function announceActivityData (
diff --git a/server/lib/activitypub/send/send-create.ts b/server/lib/activitypub/send/send-create.ts
index 9db663be1..b92615e9b 100644
--- a/server/lib/activitypub/send/send-create.ts
+++ b/server/lib/activitypub/send/send-create.ts
@@ -8,8 +8,14 @@ import { VideoAbuseModel } from '../../../models/video/video-abuse'
 import { VideoCommentModel } from '../../../models/video/video-comment'
 import { getVideoAbuseActivityPubUrl, getVideoDislikeActivityPubUrl, getVideoViewActivityPubUrl } from '../url'
 import {
-  audiencify, broadcastToActors, broadcastToFollowers, getActorsInvolvedInVideo, getAudience, getObjectFollowersAudience,
-  getOriginVideoAudience, getOriginVideoCommentAudience,
+  audiencify,
+  broadcastToActors,
+  broadcastToFollowers,
+  getActorsInvolvedInVideo,
+  getAudience,
+  getObjectFollowersAudience,
+  getOriginVideoAudience,
+  getOriginVideoCommentAudience,
   unicastTo
 } from './misc'
 
@@ -31,7 +37,7 @@ async function sendVideoAbuse (byActor: ActorModel, videoAbuse: VideoAbuseModel,
   const audience = { to: [ video.VideoChannel.Account.Actor.url ], cc: [] }
   const data = await createActivityData(url, byActor, videoAbuse.toActivityPubObject(), t, audience)
 
-  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl, t)
+  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl)
 }
 
 async function sendCreateVideoCommentToOrigin (comment: VideoCommentModel, t: Transaction) {
@@ -47,13 +53,13 @@ async function sendCreateVideoCommentToOrigin (comment: VideoCommentModel, t: Tr
 
   // This was a reply, send it to the parent actors
   const actorsException = [ byActor ]
-  await broadcastToActors(data, byActor, threadParentComments.map(c => c.Account.Actor), t, actorsException)
+  await broadcastToActors(data, byActor, threadParentComments.map(c => c.Account.Actor), actorsException)
 
   // Broadcast to our followers
   await broadcastToFollowers(data, byActor, [ byActor ], t)
 
   // Send to origin
-  return unicastTo(data, byActor, comment.Video.VideoChannel.Account.Actor.sharedInboxUrl, t)
+  return unicastTo(data, byActor, comment.Video.VideoChannel.Account.Actor.sharedInboxUrl)
 }
 
 async function sendCreateVideoCommentToVideoFollowers (comment: VideoCommentModel, t: Transaction) {
@@ -69,7 +75,7 @@ async function sendCreateVideoCommentToVideoFollowers (comment: VideoCommentMode
 
   // This was a reply, send it to the parent actors
   const actorsException = [ byActor ]
-  await broadcastToActors(data, byActor, threadParentComments.map(c => c.Account.Actor), t, actorsException)
+  await broadcastToActors(data, byActor, threadParentComments.map(c => c.Account.Actor), actorsException)
 
   // Broadcast to our followers
   await broadcastToFollowers(data, byActor, [ byActor ], t)
@@ -86,7 +92,7 @@ async function sendCreateViewToOrigin (byActor: ActorModel, video: VideoModel, t
   const audience = getOriginVideoAudience(video, actorsInvolvedInVideo)
   const data = await createActivityData(url, byActor, viewActivityData, t, audience)
 
-  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl, t)
+  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl)
 }
 
 async function sendCreateViewToVideoFollowers (byActor: ActorModel, video: VideoModel, t: Transaction) {
@@ -111,7 +117,7 @@ async function sendCreateDislikeToOrigin (byActor: ActorModel, video: VideoModel
   const audience = getOriginVideoAudience(video, actorsInvolvedInVideo)
   const data = await createActivityData(url, byActor, dislikeActivityData, t, audience)
 
-  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl, t)
+  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl)
 }
 
 async function sendCreateDislikeToVideoFollowers (byActor: ActorModel, video: VideoModel, t: Transaction) {
diff --git a/server/lib/activitypub/send/send-follow.ts b/server/lib/activitypub/send/send-follow.ts
index eac60e94f..4e9865af4 100644
--- a/server/lib/activitypub/send/send-follow.ts
+++ b/server/lib/activitypub/send/send-follow.ts
@@ -1,18 +1,17 @@
-import { Transaction } from 'sequelize'
 import { ActivityFollow } from '../../../../shared/models/activitypub'
 import { ActorModel } from '../../../models/activitypub/actor'
 import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
 import { getActorFollowActivityPubUrl } from '../url'
 import { unicastTo } from './misc'
 
-function sendFollow (actorFollow: ActorFollowModel, t: Transaction) {
+function sendFollow (actorFollow: ActorFollowModel) {
   const me = actorFollow.ActorFollower
   const following = actorFollow.ActorFollowing
 
   const url = getActorFollowActivityPubUrl(actorFollow)
   const data = followActivityData(url, me, following)
 
-  return unicastTo(data, me, following.inboxUrl, t)
+  return unicastTo(data, me, following.inboxUrl)
 }
 
 function followActivityData (url: string, byActor: ActorModel, targetActor: ActorModel): ActivityFollow {
diff --git a/server/lib/activitypub/send/send-like.ts b/server/lib/activitypub/send/send-like.ts
index 743646455..78ed1aaf2 100644
--- a/server/lib/activitypub/send/send-like.ts
+++ b/server/lib/activitypub/send/send-like.ts
@@ -20,7 +20,7 @@ async function sendLikeToOrigin (byActor: ActorModel, video: VideoModel, t: Tran
   const audience = getOriginVideoAudience(video, accountsInvolvedInVideo)
   const data = await likeActivityData(url, byActor, video, t, audience)
 
-  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl, t)
+  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl)
 }
 
 async function sendLikeToVideoFollowers (byActor: ActorModel, video: VideoModel, t: Transaction) {
diff --git a/server/lib/activitypub/send/send-undo.ts b/server/lib/activitypub/send/send-undo.ts
index 3a0597fba..4a08b5ca1 100644
--- a/server/lib/activitypub/send/send-undo.ts
+++ b/server/lib/activitypub/send/send-undo.ts
@@ -1,11 +1,5 @@
 import { Transaction } from 'sequelize'
-import {
-  ActivityAudience,
-  ActivityCreate,
-  ActivityFollow,
-  ActivityLike,
-  ActivityUndo
-} from '../../../../shared/models/activitypub'
+import { ActivityAudience, ActivityCreate, ActivityFollow, ActivityLike, ActivityUndo } from '../../../../shared/models/activitypub'
 import { ActorModel } from '../../../models/activitypub/actor'
 import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
 import { VideoModel } from '../../../models/video/video'
@@ -33,7 +27,7 @@ async function sendUndoFollow (actorFollow: ActorFollowModel, t: Transaction) {
   const object = followActivityData(followUrl, me, following)
   const data = await undoActivityData(undoUrl, me, object, t)
 
-  return unicastTo(data, me, following.inboxUrl, t)
+  return unicastTo(data, me, following.inboxUrl)
 }
 
 async function sendUndoLikeToOrigin (byActor: ActorModel, video: VideoModel, t: Transaction) {
@@ -45,7 +39,7 @@ async function sendUndoLikeToOrigin (byActor: ActorModel, video: VideoModel, t:
   const object = await likeActivityData(likeUrl, byActor, video, t)
   const data = await undoActivityData(undoUrl, byActor, object, t, audience)
 
-  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl, t)
+  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl)
 }
 
 async function sendUndoLikeToVideoFollowers (byActor: ActorModel, video: VideoModel, t: Transaction) {
@@ -72,7 +66,7 @@ async function sendUndoDislikeToOrigin (byActor: ActorModel, video: VideoModel,
 
   const data = await undoActivityData(undoUrl, byActor, object, t, audience)
 
-  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl, t)
+  return unicastTo(data, byActor, video.VideoChannel.Account.Actor.sharedInboxUrl)
 }
 
 async function sendUndoDislikeToVideoFollowers (byActor: ActorModel, video: VideoModel, t: Transaction) {
diff --git a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-broadcast-handler.ts b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
index 3f780e319..159856cda 100644
--- a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-broadcast-handler.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
@@ -1,10 +1,19 @@
+import * as kue from 'kue'
 import { logger } from '../../../helpers/logger'
 import { doRequest } from '../../../helpers/requests'
 import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
-import { ActivityPubHttpPayload, buildSignedRequestOptions, computeBody, maybeRetryRequestLater } from './activitypub-http-job-scheduler'
+import { buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils'
 
-async function process (payload: ActivityPubHttpPayload, jobId: number) {
-  logger.info('Processing ActivityPub broadcast in job %d.', jobId)
+export type ActivitypubHttpBroadcastPayload = {
+  uris: string[]
+  signatureActorId?: number
+  body: any
+}
+
+async function processActivityPubHttpBroadcast (job: kue.Job) {
+  logger.info('Processing ActivityPub broadcast in job %d.', job.id)
+
+  const payload = job.data as ActivitypubHttpBroadcastPayload
 
   const body = await computeBody(payload)
   const httpSignatureOptions = await buildSignedRequestOptions(payload)
@@ -26,28 +35,15 @@ async function process (payload: ActivityPubHttpPayload, jobId: number) {
       await doRequest(options)
       goodUrls.push(uri)
     } catch (err) {
-      const isRetryingLater = await maybeRetryRequestLater(err, payload, uri)
-      if (isRetryingLater === false) badUrls.push(uri)
+      badUrls.push(uri)
     }
   }
 
   return ActorFollowModel.updateActorFollowsScoreAndRemoveBadOnes(goodUrls, badUrls, undefined)
 }
 
-function onError (err: Error, jobId: number) {
-  logger.error('Error when broadcasting ActivityPub request in job %d.', jobId, err)
-  return Promise.resolve()
-}
-
-function onSuccess (jobId: number) {
-  logger.info('Job %d is a success.', jobId)
-  return Promise.resolve()
-}
-
 // ---------------------------------------------------------------------------
 
 export {
-  process,
-  onError,
-  onSuccess
+  processActivityPubHttpBroadcast
 }
diff --git a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts b/server/lib/job-queue/handlers/activitypub-http-fetcher.ts
index a7b5aabd0..062211c85 100644
--- a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-fetcher.ts
@@ -1,11 +1,18 @@
+import * as kue from 'kue'
 import { logger } from '../../../helpers/logger'
 import { doRequest } from '../../../helpers/requests'
 import { ACTIVITY_PUB } from '../../../initializers'
 import { processActivities } from '../../activitypub/process'
-import { ActivityPubHttpPayload } from './activitypub-http-job-scheduler'
+import { ActivitypubHttpBroadcastPayload } from './activitypub-http-broadcast'
 
-async function process (payload: ActivityPubHttpPayload, jobId: number) {
-  logger.info('Processing ActivityPub fetcher in job %d.', jobId)
+export type ActivitypubHttpFetcherPayload = {
+  uris: string[]
+}
+
+async function processActivityPubHttpFetcher (job: kue.Job) {
+  logger.info('Processing ActivityPub fetcher in job %d.', job.id)
+
+  const payload = job.data as ActivitypubHttpBroadcastPayload
 
   const options = {
     method: 'GET',
@@ -49,20 +56,8 @@ async function process (payload: ActivityPubHttpPayload, jobId: number) {
   }
 }
 
-function onError (err: Error, jobId: number) {
-  logger.error('Error when fetcher ActivityPub request in job %d.', jobId, err)
-  return Promise.resolve()
-}
-
-function onSuccess (jobId: number) {
-  logger.info('Job %d is a success.', jobId)
-  return Promise.resolve()
-}
-
 // ---------------------------------------------------------------------------
 
 export {
-  process,
-  onError,
-  onSuccess
+  processActivityPubHttpFetcher
 }
diff --git a/server/lib/job-queue/handlers/activitypub-http-unicast.ts b/server/lib/job-queue/handlers/activitypub-http-unicast.ts
new file mode 100644
index 000000000..9b4188c50
--- /dev/null
+++ b/server/lib/job-queue/handlers/activitypub-http-unicast.ts
@@ -0,0 +1,43 @@
+import * as kue from 'kue'
+import { logger } from '../../../helpers/logger'
+import { doRequest } from '../../../helpers/requests'
+import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
+import { buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils'
+
+export type ActivitypubHttpUnicastPayload = {
+  uri: string
+  signatureActorId?: number
+  body: any
+}
+
+async function processActivityPubHttpUnicast (job: kue.Job) {
+  logger.info('Processing ActivityPub unicast in job %d.', job.id)
+
+  const payload = job.data as ActivitypubHttpUnicastPayload
+  const uri = payload.uri
+
+  const body = await computeBody(payload)
+  const httpSignatureOptions = await buildSignedRequestOptions(payload)
+
+  const options = {
+    method: 'POST',
+    uri,
+    json: body,
+    httpSignature: httpSignatureOptions
+  }
+
+  try {
+    await doRequest(options)
+    ActorFollowModel.updateActorFollowsScoreAndRemoveBadOnes([ uri ], [], undefined)
+  } catch (err) {
+    ActorFollowModel.updateActorFollowsScoreAndRemoveBadOnes([], [ uri ], undefined)
+
+    throw err
+  }
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  processActivityPubHttpUnicast
+}
diff --git a/server/lib/job-queue/handlers/utils/activitypub-http-utils.ts b/server/lib/job-queue/handlers/utils/activitypub-http-utils.ts
new file mode 100644
index 000000000..c087371c6
--- /dev/null
+++ b/server/lib/job-queue/handlers/utils/activitypub-http-utils.ts
@@ -0,0 +1,39 @@
+import { buildSignedActivity } from '../../../../helpers/activitypub'
+import { getServerActor } from '../../../../helpers/utils'
+import { ActorModel } from '../../../../models/activitypub/actor'
+
+async function computeBody (payload: { body: any, signatureActorId?: number }) {
+  let body = payload.body
+
+  if (payload.signatureActorId) {
+    const actorSignature = await ActorModel.load(payload.signatureActorId)
+    if (!actorSignature) throw new Error('Unknown signature actor id.')
+    body = await buildSignedActivity(actorSignature, payload.body)
+  }
+
+  return body
+}
+
+async function buildSignedRequestOptions (payload: { signatureActorId?: number }) {
+  let actor: ActorModel
+  if (payload.signatureActorId) {
+    actor = await ActorModel.load(payload.signatureActorId)
+    if (!actor) throw new Error('Unknown signature actor id.')
+  } else {
+    // We need to sign the request, so use the server
+    actor = await getServerActor()
+  }
+
+  const keyId = actor.getWebfingerUrl()
+  return {
+    algorithm: 'rsa-sha256',
+    authorizationHeaderName: 'Signature',
+    keyId,
+    key: actor.privateKey
+  }
+}
+
+export {
+  computeBody,
+  buildSignedRequestOptions
+}
diff --git a/server/lib/jobs/transcoding-job-scheduler/video-file-optimizer-handler.ts b/server/lib/job-queue/handlers/video-file.ts
index f224a31b4..5294483bd 100644
--- a/server/lib/jobs/transcoding-job-scheduler/video-file-optimizer-handler.ts
+++ b/server/lib/job-queue/handlers/video-file.ts
@@ -1,38 +1,60 @@
-import * as Bluebird from 'bluebird'
+import * as kue from 'kue'
+import { VideoResolution } from '../../../../shared'
 import { VideoPrivacy } from '../../../../shared/models/videos'
 import { logger } from '../../../helpers/logger'
 import { computeResolutionsToTranscode } from '../../../helpers/utils'
 import { sequelizeTypescript } from '../../../initializers'
-import { JobModel } from '../../../models/job/job'
 import { VideoModel } from '../../../models/video/video'
 import { shareVideoByServerAndChannel } from '../../activitypub'
-import { sendCreateVideo } from '../../activitypub/send'
-import { JobScheduler } from '../job-scheduler'
-import { TranscodingJobPayload } from './transcoding-job-scheduler'
+import { sendCreateVideo, sendUpdateVideo } from '../../activitypub/send'
+import { JobQueue } from '../job-queue'
 
-async function process (data: TranscodingJobPayload, jobId: number) {
-  const video = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(data.videoUUID)
+export type VideoFilePayload = {
+  videoUUID: string
+  resolution?: VideoResolution
+}
+
+async function processVideoFile (job: kue.Job) {
+  const payload = job.data as VideoFilePayload
+  logger.info('Processing video file 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.', jobId, { videoUUID: video.uuid })
+    logger.info('Do not process job %d, video does not exist.', job.id, { videoUUID: video.uuid })
     return undefined
   }
 
-  await video.optimizeOriginalVideofile()
+  // Transcoding in other resolution
+  if (payload.resolution) {
+    await video.transcodeOriginalVideofile(payload.resolution)
+    await onVideoFileTranscoderSuccess(video)
+  } else {
+    await video.optimizeOriginalVideofile()
+    await onVideoFileOptimizerSuccess(video)
+  }
 
   return video
 }
 
-function onError (err: Error, jobId: number) {
-  logger.error('Error when optimized video file in job %d.', jobId, err)
-  return Promise.resolve()
+async function onVideoFileTranscoderSuccess (video: VideoModel) {
+  if (video === undefined) return undefined
+
+  // Maybe the video changed in database, refresh it
+  const videoDatabase = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(video.uuid)
+  // Video does not exist anymore
+  if (!videoDatabase) return undefined
+
+  if (video.privacy !== VideoPrivacy.PRIVATE) {
+    await sendUpdateVideo(video, undefined)
+  }
+
+  return undefined
 }
 
-async function onSuccess (jobId: number, video: VideoModel, jobScheduler: JobScheduler<TranscodingJobPayload, VideoModel>) {
+async function onVideoFileOptimizerSuccess (video: VideoModel) {
   if (video === undefined) return undefined
 
-  logger.info('Job %d is a success.', jobId)
-
   // Maybe the video changed in database, refresh it
   const videoDatabase = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(video.uuid)
   // Video does not exist anymore
@@ -56,7 +78,7 @@ async function onSuccess (jobId: number, video: VideoModel, jobScheduler: JobSch
   if (resolutionsEnabled.length !== 0) {
     try {
       await sequelizeTypescript.transaction(async t => {
-        const tasks: Bluebird<JobModel>[] = []
+        const tasks: Promise<any>[] = []
 
         for (const resolution of resolutionsEnabled) {
           const dataInput = {
@@ -64,7 +86,7 @@ async function onSuccess (jobId: number, video: VideoModel, jobScheduler: JobSch
             resolution
           }
 
-          const p = jobScheduler.createJob(t, 'videoFileTranscoder', dataInput)
+          const p = JobQueue.Instance.createJob({ type: 'video-file', payload: dataInput })
           tasks.push(p)
         }
 
@@ -84,7 +106,5 @@ async function onSuccess (jobId: number, video: VideoModel, jobScheduler: JobSch
 // ---------------------------------------------------------------------------
 
 export {
-  process,
-  onError,
-  onSuccess
+  processVideoFile
 }
diff --git a/server/lib/job-queue/index.ts b/server/lib/job-queue/index.ts
new file mode 100644
index 000000000..57231e649
--- /dev/null
+++ b/server/lib/job-queue/index.ts
@@ -0,0 +1 @@
+export * from './job-queue'
diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts
new file mode 100644
index 000000000..7a2b6c78d
--- /dev/null
+++ b/server/lib/job-queue/job-queue.ts
@@ -0,0 +1,124 @@
+import * as kue from 'kue'
+import { JobType, JobState } from '../../../shared/models'
+import { logger } from '../../helpers/logger'
+import { CONFIG, JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY } from '../../initializers'
+import { ActivitypubHttpBroadcastPayload, processActivityPubHttpBroadcast } from './handlers/activitypub-http-broadcast'
+import { ActivitypubHttpFetcherPayload, processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher'
+import { ActivitypubHttpUnicastPayload, processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast'
+import { processVideoFile, VideoFilePayload } from './handlers/video-file'
+
+type CreateJobArgument =
+  { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
+  { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } |
+  { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } |
+  { type: 'video-file', payload: VideoFilePayload }
+
+const handlers: { [ id in JobType ]: (job: kue.Job) => Promise<any>} = {
+  'activitypub-http-broadcast': processActivityPubHttpBroadcast,
+  'activitypub-http-unicast': processActivityPubHttpUnicast,
+  'activitypub-http-fetcher': processActivityPubHttpFetcher,
+  'video-file': processVideoFile
+}
+
+class JobQueue {
+
+  private static instance: JobQueue
+
+  private jobQueue: kue.Queue
+  private initialized = false
+
+  private constructor () {}
+
+  init () {
+    // Already initialized
+    if (this.initialized === true) return
+    this.initialized = true
+
+    this.jobQueue = kue.createQueue({
+      prefix: 'q-' + CONFIG.WEBSERVER.HOST,
+      redis: {
+        host: CONFIG.REDIS.HOSTNAME,
+        port: CONFIG.REDIS.PORT,
+        auth: CONFIG.REDIS.AUTH
+      }
+    })
+
+    this.jobQueue.on('error', err => {
+      logger.error('Error in job queue.', err)
+      process.exit(-1)
+    })
+    this.jobQueue.watchStuckJobs(5000)
+
+    for (const handlerName of Object.keys(handlers)) {
+      this.jobQueue.process(handlerName, JOB_CONCURRENCY[handlerName], async (job, done) => {
+        try {
+          const res = await handlers[ handlerName ](job)
+          return done(null, res)
+        } catch (err) {
+          return done(err)
+        }
+      })
+    }
+  }
+
+  createJob (obj: CreateJobArgument, priority = 'normal') {
+    return new Promise((res, rej) => {
+      this.jobQueue
+        .create(obj.type, obj.payload)
+        .priority(priority)
+        .attempts(JOB_ATTEMPTS[obj.type])
+        .backoff({ type: 'exponential' })
+        .save(err => {
+          if (err) return rej(err)
+
+          return res()
+        })
+    })
+  }
+
+  listForApi (state: JobState, start: number, count: number, sort: string) {
+    return new Promise<kue.Job[]>((res, rej) => {
+      kue.Job.rangeByState(state, start, count, sort, (err, jobs) => {
+        if (err) return rej(err)
+
+        return res(jobs)
+      })
+    })
+  }
+
+  count (state: JobState) {
+    return new Promise<number>((res, rej) => {
+      this.jobQueue[state + 'Count']((err, total) => {
+        if (err) return rej(err)
+
+        return res(total)
+      })
+    })
+  }
+
+  removeOldJobs () {
+    const now = new Date().getTime()
+    kue.Job.rangeByState('complete', 0, -1, 'asc', (err, jobs) => {
+      if (err) {
+        logger.error('Cannot get jobs when removing old jobs.', err)
+        return
+      }
+
+      for (const job of jobs) {
+        if (now - job.created_at > JOB_COMPLETED_LIFETIME) {
+          job.remove()
+        }
+      }
+    })
+  }
+
+  static get Instance () {
+    return this.instance || (this.instance = new this())
+  }
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  JobQueue
+}
diff --git a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts b/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts
deleted file mode 100644
index 4459152db..000000000
--- a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-job-scheduler.ts
+++ /dev/null
@@ -1,94 +0,0 @@
-import { JobCategory } from '../../../../shared'
-import { buildSignedActivity } from '../../../helpers/activitypub'
-import { logger } from '../../../helpers/logger'
-import { getServerActor } from '../../../helpers/utils'
-import { ACTIVITY_PUB } from '../../../initializers'
-import { ActorModel } from '../../../models/activitypub/actor'
-import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
-import { JobHandler, JobScheduler } from '../job-scheduler'
-
-import * as activitypubHttpBroadcastHandler from './activitypub-http-broadcast-handler'
-import * as activitypubHttpFetcherHandler from './activitypub-http-fetcher-handler'
-import * as activitypubHttpUnicastHandler from './activitypub-http-unicast-handler'
-
-type ActivityPubHttpPayload = {
-  uris: string[]
-  signatureActorId?: number
-  body?: any
-  attemptNumber?: number
-}
-
-const jobHandlers: { [ handlerName: string ]: JobHandler<ActivityPubHttpPayload, void> } = {
-  activitypubHttpBroadcastHandler,
-  activitypubHttpUnicastHandler,
-  activitypubHttpFetcherHandler
-}
-const jobCategory: JobCategory = 'activitypub-http'
-
-const activitypubHttpJobScheduler = new JobScheduler(jobCategory, jobHandlers)
-
-async function maybeRetryRequestLater (err: Error, payload: ActivityPubHttpPayload, uri: string) {
-  logger.warn('Cannot make request to %s.', uri, err)
-
-  let attemptNumber = payload.attemptNumber || 1
-  attemptNumber += 1
-
-  if (attemptNumber < ACTIVITY_PUB.MAX_HTTP_ATTEMPT) {
-    logger.debug('Retrying request to %s (attempt %d/%d).', uri, attemptNumber, ACTIVITY_PUB.MAX_HTTP_ATTEMPT, err)
-
-    const actor = await ActorFollowModel.loadByFollowerInbox(uri, undefined)
-    if (!actor) {
-      logger.debug('Actor %s is not a follower, do not retry the request.', uri)
-      return false
-    }
-
-    const newPayload = Object.assign(payload, {
-      uris: [ uri ],
-      attemptNumber
-    })
-    await activitypubHttpJobScheduler.createJob(undefined, 'activitypubHttpUnicastHandler', newPayload)
-
-    return true
-  }
-
-  return false
-}
-
-async function computeBody (payload: ActivityPubHttpPayload) {
-  let body = payload.body
-
-  if (payload.signatureActorId) {
-    const actorSignature = await ActorModel.load(payload.signatureActorId)
-    if (!actorSignature) throw new Error('Unknown signature actor id.')
-    body = await buildSignedActivity(actorSignature, payload.body)
-  }
-
-  return body
-}
-
-async function buildSignedRequestOptions (payload: ActivityPubHttpPayload) {
-  let actor: ActorModel
-  if (payload.signatureActorId) {
-    actor = await ActorModel.load(payload.signatureActorId)
-    if (!actor) throw new Error('Unknown signature actor id.')
-  } else {
-    // We need to sign the request, so use the server
-    actor = await getServerActor()
-  }
-
-  const keyId = actor.getWebfingerUrl()
-  return {
-    algorithm: 'rsa-sha256',
-    authorizationHeaderName: 'Signature',
-    keyId,
-    key: actor.privateKey
-  }
-}
-
-export {
-  ActivityPubHttpPayload,
-  activitypubHttpJobScheduler,
-  maybeRetryRequestLater,
-  computeBody,
-  buildSignedRequestOptions
-}
diff --git a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-unicast-handler.ts b/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-unicast-handler.ts
deleted file mode 100644
index 54a7504e8..000000000
--- a/server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-unicast-handler.ts
+++ /dev/null
@@ -1,50 +0,0 @@
-import { logger } from '../../../helpers/logger'
-import { doRequest } from '../../../helpers/requests'
-import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
-import { ActivityPubHttpPayload, buildSignedRequestOptions, computeBody, maybeRetryRequestLater } from './activitypub-http-job-scheduler'
-
-async function process (payload: ActivityPubHttpPayload, jobId: number) {
-  logger.info('Processing ActivityPub unicast in job %d.', jobId)
-
-  const uri = payload.uris[0]
-
-  const body = await computeBody(payload)
-  const httpSignatureOptions = await buildSignedRequestOptions(payload)
-
-  const options = {
-    method: 'POST',
-    uri,
-    json: body,
-    httpSignature: httpSignatureOptions
-  }
-
-  try {
-    await doRequest(options)
-    ActorFollowModel.updateActorFollowsScoreAndRemoveBadOnes([ uri ], [], undefined)
-  } catch (err) {
-    const isRetryingLater = await maybeRetryRequestLater(err, payload, uri)
-    if (isRetryingLater === false) {
-      ActorFollowModel.updateActorFollowsScoreAndRemoveBadOnes([], [ uri ], undefined)
-    }
-
-    throw err
-  }
-}
-
-function onError (err: Error, jobId: number) {
-  logger.error('Error when sending ActivityPub request in job %d.', jobId, err)
-  return Promise.resolve()
-}
-
-function onSuccess (jobId: number) {
-  logger.info('Job %d is a success.', jobId)
-  return Promise.resolve()
-}
-
-// ---------------------------------------------------------------------------
-
-export {
-  process,
-  onError,
-  onSuccess
-}
diff --git a/server/lib/jobs/activitypub-http-job-scheduler/index.ts b/server/lib/jobs/activitypub-http-job-scheduler/index.ts
deleted file mode 100644
index ad8f527b4..000000000
--- a/server/lib/jobs/activitypub-http-job-scheduler/index.ts
+++ /dev/null
@@ -1 +0,0 @@
-export * from './activitypub-http-job-scheduler'
diff --git a/server/lib/jobs/index.ts b/server/lib/jobs/index.ts
deleted file mode 100644
index 394264ec1..000000000
--- a/server/lib/jobs/index.ts
+++ /dev/null
@@ -1,2 +0,0 @@
-export * from './activitypub-http-job-scheduler'
-export * from './transcoding-job-scheduler'
diff --git a/server/lib/jobs/job-scheduler.ts b/server/lib/jobs/job-scheduler.ts
deleted file mode 100644
index 9d55880e6..000000000
--- a/server/lib/jobs/job-scheduler.ts
+++ /dev/null
@@ -1,144 +0,0 @@
-import { AsyncQueue, forever, queue } from 'async'
-import * as Sequelize from 'sequelize'
-import { JobCategory } from '../../../shared'
-import { logger } from '../../helpers/logger'
-import { JOB_STATES, JOBS_FETCH_LIMIT_PER_CYCLE, JOBS_FETCHING_INTERVAL } from '../../initializers'
-import { JobModel } from '../../models/job/job'
-
-export interface JobHandler<P, T> {
-  process (data: object, jobId: number): Promise<T>
-  onError (err: Error, jobId: number)
-  onSuccess (jobId: number, jobResult: T, jobScheduler: JobScheduler<P, T>): Promise<any>
-}
-type JobQueueCallback = (err: Error) => void
-
-class JobScheduler<P, T> {
-
-  constructor (
-    private jobCategory: JobCategory,
-    private jobHandlers: { [ id: string ]: JobHandler<P, T> }
-  ) {}
-
-  async activate () {
-    const limit = JOBS_FETCH_LIMIT_PER_CYCLE[this.jobCategory]
-
-    logger.info('Jobs scheduler %s activated.', this.jobCategory)
-
-    const jobsQueue = queue<JobModel, JobQueueCallback>(this.processJob.bind(this))
-
-    // Finish processing jobs from a previous start
-    const state = JOB_STATES.PROCESSING
-    try {
-      const jobs = await JobModel.listWithLimitByCategory(limit, state, this.jobCategory)
-
-      this.enqueueJobs(jobsQueue, jobs)
-    } catch (err) {
-      logger.error('Cannot list pending jobs.', err)
-    }
-
-    forever(
-      async next => {
-        if (jobsQueue.length() !== 0) {
-          // Finish processing the queue first
-          return setTimeout(next, JOBS_FETCHING_INTERVAL)
-        }
-
-        const state = JOB_STATES.PENDING
-        try {
-          const jobs = await JobModel.listWithLimitByCategory(limit, state, this.jobCategory)
-
-          this.enqueueJobs(jobsQueue, jobs)
-        } catch (err) {
-          logger.error('Cannot list pending jobs.', err)
-        }
-
-        // Optimization: we could use "drain" from queue object
-        return setTimeout(next, JOBS_FETCHING_INTERVAL)
-      },
-
-      err => logger.error('Error in job scheduler queue.', err)
-    )
-  }
-
-  createJob (transaction: Sequelize.Transaction, handlerName: string, handlerInputData: P) {
-    const createQuery = {
-      state: JOB_STATES.PENDING,
-      category: this.jobCategory,
-      handlerName,
-      handlerInputData
-    }
-
-    const options = { transaction }
-
-    return JobModel.create(createQuery, options)
-  }
-
-  private enqueueJobs (jobsQueue: AsyncQueue<JobModel>, jobs: JobModel[]) {
-    jobs.forEach(job => jobsQueue.push(job))
-  }
-
-  private async processJob (job: JobModel, callback: (err: Error) => void) {
-    const jobHandler = this.jobHandlers[job.handlerName]
-    if (jobHandler === undefined) {
-      const errorString = 'Unknown job handler ' + job.handlerName + ' for job ' + job.id
-      logger.error(errorString)
-
-      const error = new Error(errorString)
-      await this.onJobError(jobHandler, job, error)
-      return callback(error)
-    }
-
-    logger.info('Processing job %d with handler %s.', job.id, job.handlerName)
-
-    job.state = JOB_STATES.PROCESSING
-    await job.save()
-
-    try {
-      const result: T = await jobHandler.process(job.handlerInputData, job.id)
-      await this.onJobSuccess(jobHandler, job, result)
-    } catch (err) {
-      logger.error('Error in job handler %s.', job.handlerName, err)
-
-      try {
-        await this.onJobError(jobHandler, job, err)
-      } catch (innerErr) {
-        this.cannotSaveJobError(innerErr)
-        return callback(innerErr)
-      }
-    }
-
-    return callback(null)
-  }
-
-  private async onJobError (jobHandler: JobHandler<P, T>, job: JobModel, err: Error) {
-    job.state = JOB_STATES.ERROR
-
-    try {
-      await job.save()
-      if (jobHandler) await jobHandler.onError(err, job.id)
-    } catch (err) {
-      this.cannotSaveJobError(err)
-    }
-  }
-
-  private async onJobSuccess (jobHandler: JobHandler<P, T>, job: JobModel, jobResult: T) {
-    job.state = JOB_STATES.SUCCESS
-
-    try {
-      await job.save()
-      await jobHandler.onSuccess(job.id, jobResult, this)
-    } catch (err) {
-      this.cannotSaveJobError(err)
-    }
-  }
-
-  private cannotSaveJobError (err: Error) {
-    logger.error('Cannot save new job state.', err)
-  }
-}
-
-// ---------------------------------------------------------------------------
-
-export {
-  JobScheduler
-}
diff --git a/server/lib/jobs/transcoding-job-scheduler/index.ts b/server/lib/jobs/transcoding-job-scheduler/index.ts
deleted file mode 100644
index 73152a1be..000000000
--- a/server/lib/jobs/transcoding-job-scheduler/index.ts
+++ /dev/null
@@ -1 +0,0 @@
-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
deleted file mode 100644
index e5530a73c..000000000
--- a/server/lib/jobs/transcoding-job-scheduler/transcoding-job-scheduler.ts
+++ /dev/null
@@ -1,23 +0,0 @@
-import { JobCategory } from '../../../../shared'
-import { VideoModel } from '../../../models/video/video'
-import { JobHandler, JobScheduler } from '../job-scheduler'
-
-import * as videoFileOptimizer from './video-file-optimizer-handler'
-import * as videoFileTranscoder from './video-file-transcoder-handler'
-
-type TranscodingJobPayload = {
-  videoUUID: string
-  resolution?: number
-}
-const jobHandlers: { [ handlerName: string ]: JobHandler<TranscodingJobPayload, VideoModel> } = {
-  videoFileOptimizer,
-  videoFileTranscoder
-}
-const jobCategory: JobCategory = 'transcoding'
-
-const transcodingJobScheduler = new JobScheduler(jobCategory, jobHandlers)
-
-export {
-  TranscodingJobPayload,
-  transcodingJobScheduler
-}
diff --git a/server/lib/jobs/transcoding-job-scheduler/video-file-transcoder-handler.ts b/server/lib/jobs/transcoding-job-scheduler/video-file-transcoder-handler.ts
deleted file mode 100644
index 883d3eba8..000000000
--- a/server/lib/jobs/transcoding-job-scheduler/video-file-transcoder-handler.ts
+++ /dev/null
@@ -1,48 +0,0 @@
-import { VideoResolution } from '../../../../shared'
-import { VideoPrivacy } from '../../../../shared/models/videos'
-import { logger } from '../../../helpers/logger'
-import { VideoModel } from '../../../models/video/video'
-import { sendUpdateVideo } from '../../activitypub/send'
-
-async function process (data: { videoUUID: string, resolution: VideoResolution }, jobId: number) {
-  const video = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(data.videoUUID)
-  // No video, maybe deleted?
-  if (!video) {
-    logger.info('Do not process job %d, video does not exist.', jobId, { videoUUID: video.uuid })
-    return undefined
-  }
-
-  await video.transcodeOriginalVideofile(data.resolution)
-
-  return video
-}
-
-function onError (err: Error, jobId: number) {
-  logger.error('Error when transcoding video file in job %d.', jobId, err)
-  return Promise.resolve()
-}
-
-async function onSuccess (jobId: number, video: VideoModel) {
-  if (video === undefined) return undefined
-
-  logger.info('Job %d is a success.', jobId)
-
-  // Maybe the video changed in database, refresh it
-  const videoDatabase = await VideoModel.loadByUUIDAndPopulateAccountAndServerAndTags(video.uuid)
-  // Video does not exist anymore
-  if (!videoDatabase) return undefined
-
-  if (video.privacy !== VideoPrivacy.PRIVATE) {
-    await sendUpdateVideo(video, undefined)
-  }
-
-  return undefined
-}
-
-// ---------------------------------------------------------------------------
-
-export {
-  process,
-  onError,
-  onSuccess
-}
diff --git a/server/lib/schedulers/remove-old-jobs-scheduler.ts b/server/lib/schedulers/remove-old-jobs-scheduler.ts
new file mode 100644
index 000000000..add5677ac
--- /dev/null
+++ b/server/lib/schedulers/remove-old-jobs-scheduler.ts
@@ -0,0 +1,19 @@
+import { JobQueue } from '../job-queue'
+import { AbstractScheduler } from './abstract-scheduler'
+
+export class RemoveOldJobsScheduler extends AbstractScheduler {
+
+  private static instance: AbstractScheduler
+
+  private constructor () {
+    super()
+  }
+
+  async execute () {
+    JobQueue.Instance.removeOldJobs()
+  }
+
+  static get Instance () {
+    return this.instance || (this.instance = new this())
+  }
+}