29 files changed, 206 insertions, 77 deletions

diff --git a/server/lib/activitypub/actors/get.ts b/server/lib/activitypub/actors/get.ts
index d2b651082..e73b7d707 100644
--- a/server/lib/activitypub/actors/get.ts
+++ b/server/lib/activitypub/actors/get.ts
@@ -110,7 +110,7 @@ async function loadActorFromDB (actorUrl: string, fetchType: ActorLoadByUrlType)
 async function scheduleOutboxFetchIfNeeded (actor: MActor, created: boolean, refreshed: boolean, updateCollections: boolean) {
   if ((created === true || refreshed === true) && updateCollections === true) {
     const payload = { uri: actor.outboxUrl, type: 'activity' as 'activity' }
-    await JobQueue.Instance.createJobWithPromise({ type: 'activitypub-http-fetcher', payload })
+    await JobQueue.Instance.createJob({ type: 'activitypub-http-fetcher', payload })
   }
 }
 
@@ -118,6 +118,6 @@ async function schedulePlaylistFetchIfNeeded (actor: MActorAccountId, created: b
   // We created a new account: fetch the playlists
   if (created === true && actor.Account && accountPlaylistsUrl) {
     const payload = { uri: accountPlaylistsUrl, type: 'account-playlists' as 'account-playlists' }
-    await JobQueue.Instance.createJobWithPromise({ type: 'activitypub-http-fetcher', payload })
+    await JobQueue.Instance.createJob({ type: 'activitypub-http-fetcher', payload })
   }
 }
diff --git a/server/lib/activitypub/follow.ts b/server/lib/activitypub/follow.ts
index 741b54df5..f6e2a48fd 100644
--- a/server/lib/activitypub/follow.ts
+++ b/server/lib/activitypub/follow.ts
@@ -27,7 +27,7 @@ async function autoFollowBackIfNeeded (actorFollow: MActorFollowActors, transact
       isAutoFollow: true
     }
 
-    JobQueue.Instance.createJob({ type: 'activitypub-follow', payload })
+    JobQueue.Instance.createJobAsync({ type: 'activitypub-follow', payload })
   }
 }
 
diff --git a/server/lib/activitypub/outbox.ts b/server/lib/activitypub/outbox.ts
index ecdc33a77..5eef76871 100644
--- a/server/lib/activitypub/outbox.ts
+++ b/server/lib/activitypub/outbox.ts
@@ -16,7 +16,7 @@ async function addFetchOutboxJob (actor: Pick<ActorModel, 'id' | 'outboxUrl'>) {
     type: 'activity' as 'activity'
   }
 
-  return JobQueue.Instance.createJob({ type: 'activitypub-http-fetcher', payload })
+  return JobQueue.Instance.createJobAsync({ type: 'activitypub-http-fetcher', payload })
 }
 
 export {
diff --git a/server/lib/activitypub/playlists/refresh.ts b/server/lib/activitypub/playlists/refresh.ts
index 493e8c7ec..33260ea02 100644
--- a/server/lib/activitypub/playlists/refresh.ts
+++ b/server/lib/activitypub/playlists/refresh.ts
@@ -9,7 +9,7 @@ import { fetchRemoteVideoPlaylist } from './shared'
 function scheduleRefreshIfNeeded (playlist: MVideoPlaylist) {
   if (!playlist.isOutdated()) return
 
-  JobQueue.Instance.createJob({ type: 'activitypub-refresher', payload: { type: 'video-playlist', url: playlist.url } })
+  JobQueue.Instance.createJobAsync({ type: 'activitypub-refresher', payload: { type: 'video-playlist', url: playlist.url } })
 }
 
 async function refreshVideoPlaylistIfNeeded (videoPlaylist: MVideoPlaylistOwner): Promise<MVideoPlaylistOwner> {
diff --git a/server/lib/activitypub/send/shared/send-utils.ts b/server/lib/activitypub/send/shared/send-utils.ts
index fcec63991..2bc1ef8f5 100644
--- a/server/lib/activitypub/send/shared/send-utils.ts
+++ b/server/lib/activitypub/send/shared/send-utils.ts
@@ -120,7 +120,7 @@ async function forwardActivity (
     body: activity,
     contextType: null
   }
-  return afterCommitIfTransaction(t, () => JobQueue.Instance.createJob({ type: 'activitypub-http-broadcast', payload }))
+  return afterCommitIfTransaction(t, () => JobQueue.Instance.createJobAsync({ type: 'activitypub-http-broadcast', payload }))
 }
 
 // ---------------------------------------------------------------------------
@@ -205,7 +205,7 @@ function broadcastTo (options: {
       contextType
     }
 
-    JobQueue.Instance.createJob({
+    JobQueue.Instance.createJobAsync({
       type: parallelizable
         ? 'activitypub-http-broadcast-parallel'
         : 'activitypub-http-broadcast',
@@ -222,7 +222,7 @@ function broadcastTo (options: {
       contextType
     }
 
-    JobQueue.Instance.createJob({ type: 'activitypub-http-unicast', payload })
+    JobQueue.Instance.createJobAsync({ type: 'activitypub-http-unicast', payload })
   }
 }
 
@@ -243,7 +243,7 @@ function unicastTo (options: {
     contextType
   }
 
-  JobQueue.Instance.createJob({ type: 'activitypub-http-unicast', payload })
+  JobQueue.Instance.createJobAsync({ type: 'activitypub-http-unicast', payload })
 }
 
 // ---------------------------------------------------------------------------
diff --git a/server/lib/activitypub/videos/get.ts b/server/lib/activitypub/videos/get.ts
index b74df132c..14ba55034 100644
--- a/server/lib/activitypub/videos/get.ts
+++ b/server/lib/activitypub/videos/get.ts
@@ -107,7 +107,7 @@ async function scheduleRefresh (video: MVideoThumbnail, fetchType: VideoLoadByUr
     return refreshVideoIfNeeded(refreshOptions)
   }
 
-  await JobQueue.Instance.createJobWithPromise({
+  await JobQueue.Instance.createJob({
     type: 'activitypub-refresher',
     payload: { type: 'video', url: video.url }
   })
diff --git a/server/lib/activitypub/videos/shared/video-sync-attributes.ts b/server/lib/activitypub/videos/shared/video-sync-attributes.ts
index 8cf0c87a6..8ed1b6447 100644
--- a/server/lib/activitypub/videos/shared/video-sync-attributes.ts
+++ b/server/lib/activitypub/videos/shared/video-sync-attributes.ts
@@ -74,7 +74,7 @@ async function getRatesCount (type: 'like' | 'dislike', video: MVideo, fetchedVi
 }
 
 function createJob (payload: ActivitypubHttpFetcherPayload) {
-  return JobQueue.Instance.createJobWithPromise({ type: 'activitypub-http-fetcher', payload })
+  return JobQueue.Instance.createJob({ type: 'activitypub-http-fetcher', payload })
 }
 
 function syncShares (video: MVideo, fetchedVideo: VideoObject, isSync: boolean) {
diff --git a/server/lib/emailer.ts b/server/lib/emailer.ts
index bd1089530..9e546de7f 100644
--- a/server/lib/emailer.ts
+++ b/server/lib/emailer.ts
@@ -66,7 +66,7 @@ class Emailer {
       }
     }
 
-    return JobQueue.Instance.createJob({ type: 'email', payload: emailPayload })
+    return JobQueue.Instance.createJobAsync({ type: 'email', payload: emailPayload })
   }
 
   addPasswordCreateEmailJob (username: string, to: string, createPasswordUrl: string) {
@@ -80,7 +80,7 @@ class Emailer {
       }
     }
 
-    return JobQueue.Instance.createJob({ type: 'email', payload: emailPayload })
+    return JobQueue.Instance.createJobAsync({ type: 'email', payload: emailPayload })
   }
 
   addVerifyEmailJob (username: string, to: string, verifyEmailUrl: string) {
@@ -94,7 +94,7 @@ class Emailer {
       }
     }
 
-    return JobQueue.Instance.createJob({ type: 'email', payload: emailPayload })
+    return JobQueue.Instance.createJobAsync({ type: 'email', payload: emailPayload })
   }
 
   addUserBlockJob (user: MUser, blocked: boolean, reason?: string) {
@@ -108,7 +108,7 @@ class Emailer {
       text: `Your account ${user.username} on ${CONFIG.INSTANCE.NAME} has been ${blockedWord}${reasonString}.`
     }
 
-    return JobQueue.Instance.createJob({ type: 'email', payload: emailPayload })
+    return JobQueue.Instance.createJobAsync({ type: 'email', payload: emailPayload })
   }
 
   addContactFormJob (fromEmail: string, fromName: string, subject: string, body: string) {
@@ -127,7 +127,7 @@ class Emailer {
       }
     }
 
-    return JobQueue.Instance.createJob({ type: 'email', payload: emailPayload })
+    return JobQueue.Instance.createJobAsync({ type: 'email', payload: emailPayload })
   }
 
   async sendMail (options: EmailPayload) {
diff --git a/server/lib/job-queue/handlers/activitypub-follow.ts b/server/lib/job-queue/handlers/activitypub-follow.ts
index 944da5be1..a68c32ba0 100644
--- a/server/lib/job-queue/handlers/activitypub-follow.ts
+++ b/server/lib/job-queue/handlers/activitypub-follow.ts
@@ -17,7 +17,7 @@ async function processActivityPubFollow (job: Job) {
   const payload = job.data as ActivitypubFollowPayload
   const host = payload.host
 
-  logger.info('Processing ActivityPub follow in job %d.', job.id)
+  logger.info('Processing ActivityPub follow in job %s.', job.id)
 
   let targetActor: MActorFull
   if (!host || host === WEBSERVER.HOST) {
diff --git a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
index 354c608fb..13eff5211 100644
--- a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
@@ -8,7 +8,7 @@ import { doRequest } from '../../../helpers/requests'
 import { BROADCAST_CONCURRENCY } from '../../../initializers/constants'
 
 async function processActivityPubHttpBroadcast (job: Job) {
-  logger.info('Processing ActivityPub broadcast in job %d.', job.id)
+  logger.info('Processing ActivityPub broadcast in job %s.', job.id)
 
   const payload = job.data as ActivitypubHttpBroadcastPayload
 
diff --git a/server/lib/job-queue/handlers/activitypub-http-fetcher.ts b/server/lib/job-queue/handlers/activitypub-http-fetcher.ts
index e0b841887..b6cb3c4a6 100644
--- a/server/lib/job-queue/handlers/activitypub-http-fetcher.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-fetcher.ts
@@ -12,7 +12,7 @@ import { addVideoShares } from '../../activitypub/share'
 import { addVideoComments } from '../../activitypub/video-comments'
 
 async function processActivityPubHttpFetcher (job: Job) {
-  logger.info('Processing ActivityPub fetcher in job %d.', job.id)
+  logger.info('Processing ActivityPub fetcher in job %s.', job.id)
 
   const payload = job.data as ActivitypubHttpFetcherPayload
 
diff --git a/server/lib/job-queue/handlers/activitypub-http-unicast.ts b/server/lib/job-queue/handlers/activitypub-http-unicast.ts
index 837a597a5..9e4e84002 100644
--- a/server/lib/job-queue/handlers/activitypub-http-unicast.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-unicast.ts
@@ -6,7 +6,7 @@ import { doRequest } from '../../../helpers/requests'
 import { ActorFollowHealthCache } from '../../actor-follow-health-cache'
 
 async function processActivityPubHttpUnicast (job: Job) {
-  logger.info('Processing ActivityPub unicast in job %d.', job.id)
+  logger.info('Processing ActivityPub unicast in job %s.', job.id)
 
   const payload = job.data as ActivitypubHttpUnicastPayload
   const uri = payload.uri
diff --git a/server/lib/job-queue/handlers/activitypub-refresher.ts b/server/lib/job-queue/handlers/activitypub-refresher.ts
index 600f858a0..307e771ff 100644
--- a/server/lib/job-queue/handlers/activitypub-refresher.ts
+++ b/server/lib/job-queue/handlers/activitypub-refresher.ts
@@ -11,7 +11,7 @@ import { refreshActorIfNeeded } from '../../activitypub/actors'
 async function refreshAPObject (job: Job) {
   const payload = job.data as RefreshPayload
 
-  logger.info('Processing AP refresher in job %d for %s.', job.id, payload.url)
+  logger.info('Processing AP refresher in job %s for %s.', job.id, payload.url)
 
   if (payload.type === 'video') return refreshVideo(payload.url)
   if (payload.type === 'video-playlist') return refreshVideoPlaylist(payload.url)
diff --git a/server/lib/job-queue/handlers/actor-keys.ts b/server/lib/job-queue/handlers/actor-keys.ts
index 4a5bad9fb..27a2d431b 100644
--- a/server/lib/job-queue/handlers/actor-keys.ts
+++ b/server/lib/job-queue/handlers/actor-keys.ts
@@ -6,7 +6,7 @@ import { logger } from '../../../helpers/logger'
 
 async function processActorKeys (job: Job) {
   const payload = job.data as ActorKeysPayload
-  logger.info('Processing actor keys in job %d.', job.id)
+  logger.info('Processing actor keys in job %s.', job.id)
 
   const actor = await ActorModel.load(payload.actorId)
 
diff --git a/server/lib/job-queue/handlers/email.ts b/server/lib/job-queue/handlers/email.ts
index b5b9475b1..567bcc076 100644
--- a/server/lib/job-queue/handlers/email.ts
+++ b/server/lib/job-queue/handlers/email.ts
@@ -5,7 +5,7 @@ import { Emailer } from '../../emailer'
 
 async function processEmail (job: Job) {
   const payload = job.data as EmailPayload
-  logger.info('Processing email in job %d.', job.id)
+  logger.info('Processing email in job %s.', job.id)
 
   return Emailer.Instance.sendMail(payload)
 }
diff --git a/server/lib/job-queue/handlers/federate-video.ts b/server/lib/job-queue/handlers/federate-video.ts
new file mode 100644
index 000000000..6aac36741
--- /dev/null
+++ b/server/lib/job-queue/handlers/federate-video.ts
@@ -0,0 +1,28 @@
+import { Job } from 'bullmq'
+import { retryTransactionWrapper } from '@server/helpers/database-utils'
+import { sequelizeTypescript } from '@server/initializers/database'
+import { federateVideoIfNeeded } from '@server/lib/activitypub/videos'
+import { VideoModel } from '@server/models/video/video'
+import { FederateVideoPayload } from '@shared/models'
+import { logger } from '../../../helpers/logger'
+
+function processFederateVideo (job: Job) {
+  const payload = job.data as FederateVideoPayload
+
+  logger.info('Processing video federation in job %s.', job.id)
+
+  return retryTransactionWrapper(() => {
+    return sequelizeTypescript.transaction(async t => {
+      const video = await VideoModel.loadFull(payload.videoUUID, t)
+      if (!video) return
+
+      return federateVideoIfNeeded(video, payload.isNewVideo, t)
+    })
+  })
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  processFederateVideo
+}
diff --git a/server/lib/job-queue/handlers/manage-video-torrent.ts b/server/lib/job-queue/handlers/manage-video-torrent.ts
index 4505ca79e..03aa414c9 100644
--- a/server/lib/job-queue/handlers/manage-video-torrent.ts
+++ b/server/lib/job-queue/handlers/manage-video-torrent.ts
@@ -8,7 +8,7 @@ import { logger } from '../../../helpers/logger'
 
 async function processManageVideoTorrent (job: Job) {
   const payload = job.data as ManageVideoTorrentPayload
-  logger.info('Processing torrent in job %d.', job.id)
+  logger.info('Processing torrent in job %s.', job.id)
 
   if (payload.action === 'create') return doCreateAction(payload)
   if (payload.action === 'update-metadata') return doUpdateMetadataAction(payload)
diff --git a/server/lib/job-queue/handlers/move-to-object-storage.ts b/server/lib/job-queue/handlers/move-to-object-storage.ts
index d608fd865..25bdebeea 100644
--- a/server/lib/job-queue/handlers/move-to-object-storage.ts
+++ b/server/lib/job-queue/handlers/move-to-object-storage.ts
@@ -17,7 +17,7 @@ const lTagsBase = loggerTagsFactory('move-object-storage')
 
 export async function processMoveToObjectStorage (job: Job) {
   const payload = job.data as MoveObjectStoragePayload
-  logger.info('Moving video %s in job %d.', payload.videoUUID, job.id)
+  logger.info('Moving video %s in job %s.', payload.videoUUID, job.id)
 
   const video = await VideoModel.loadWithFiles(payload.videoUUID)
   // No video, maybe deleted?
@@ -43,7 +43,7 @@ export async function processMoveToObjectStorage (job: Job) {
 
     const pendingMove = await VideoJobInfoModel.decrease(video.uuid, 'pendingMove')
     if (pendingMove === 0) {
-      logger.info('Running cleanup after moving files to object storage (video %s in job %d)', video.uuid, job.id, lTags)
+      logger.info('Running cleanup after moving files to object storage (video %s in job %s)', video.uuid, job.id, lTags)
 
       await doAfterLastJob({ video, previousVideoState: payload.previousVideoState, isNewVideo: payload.isNewVideo })
     }
diff --git a/server/lib/job-queue/handlers/notify.ts b/server/lib/job-queue/handlers/notify.ts
new file mode 100644
index 000000000..83605396c
--- /dev/null
+++ b/server/lib/job-queue/handlers/notify.ts
@@ -0,0 +1,27 @@
+import { Job } from 'bullmq'
+import { Notifier } from '@server/lib/notifier'
+import { VideoModel } from '@server/models/video/video'
+import { NotifyPayload } from '@shared/models'
+import { logger } from '../../../helpers/logger'
+
+async function processNotify (job: Job) {
+  const payload = job.data as NotifyPayload
+  logger.info('Processing %s notification in job %s.', payload.action, job.id)
+
+  if (payload.action === 'new-video') return doNotifyNewVideo(payload)
+}
+
+// ---------------------------------------------------------------------------
+
+export {
+  processNotify
+}
+
+// ---------------------------------------------------------------------------
+
+async function doNotifyNewVideo (payload: NotifyPayload & { action: 'new-video' }) {
+  const refreshedVideo = await VideoModel.loadFull(payload.videoUUID)
+  if (!refreshedVideo) return
+
+  Notifier.Instance.notifyOnNewVideoIfNeeded(refreshedVideo)
+}
diff --git a/server/lib/job-queue/handlers/video-file-import.ts b/server/lib/job-queue/handlers/video-file-import.ts
index 40c44cf52..d950f6407 100644
--- a/server/lib/job-queue/handlers/video-file-import.ts
+++ b/server/lib/job-queue/handlers/video-file-import.ts
@@ -4,7 +4,7 @@ import { createTorrentAndSetInfoHash } from '@server/helpers/webtorrent'
 import { CONFIG } from '@server/initializers/config'
 import { federateVideoIfNeeded } from '@server/lib/activitypub/videos'
 import { generateWebTorrentVideoFilename } from '@server/lib/paths'
-import { addMoveToObjectStorageJob } from '@server/lib/video'
+import { buildMoveToObjectStorageJob } from '@server/lib/video'
 import { VideoPathManager } from '@server/lib/video-path-manager'
 import { VideoModel } from '@server/models/video/video'
 import { VideoFileModel } from '@server/models/video/video-file'
@@ -13,10 +13,11 @@ import { getLowercaseExtension } from '@shared/core-utils'
 import { VideoFileImportPayload, VideoStorage } from '@shared/models'
 import { getVideoStreamFPS, getVideoStreamDimensionsInfo } from '../../../helpers/ffmpeg'
 import { logger } from '../../../helpers/logger'
+import { JobQueue } from '../job-queue'
 
 async function processVideoFileImport (job: Job) {
   const payload = job.data as VideoFileImportPayload
-  logger.info('Processing video file import in job %d.', job.id)
+  logger.info('Processing video file import in job %s.', job.id)
 
   const video = await VideoModel.loadFull(payload.videoUUID)
   // No video, maybe deleted?
@@ -28,7 +29,7 @@ async function processVideoFileImport (job: Job) {
   await updateVideoFile(video, payload.filePath)
 
   if (CONFIG.OBJECT_STORAGE.ENABLED) {
-    await addMoveToObjectStorageJob({ video, previousVideoState: video.state })
+    await JobQueue.Instance.createJob(await buildMoveToObjectStorageJob({ video, previousVideoState: video.state }))
   } else {
     await federateVideoIfNeeded(video, false)
   }
diff --git a/server/lib/job-queue/handlers/video-import.ts b/server/lib/job-queue/handlers/video-import.ts
index e5cd35865..f4629159c 100644
--- a/server/lib/job-queue/handlers/video-import.ts
+++ b/server/lib/job-queue/handlers/video-import.ts
@@ -8,7 +8,7 @@ import { generateWebTorrentVideoFilename } from '@server/lib/paths'
 import { Hooks } from '@server/lib/plugins/hooks'
 import { ServerConfigManager } from '@server/lib/server-config-manager'
 import { isAbleToUploadVideo } from '@server/lib/user'
-import { addMoveToObjectStorageJob, addOptimizeOrMergeAudioJob } from '@server/lib/video'
+import { buildOptimizeOrMergeAudioJob, buildMoveToObjectStorageJob } from '@server/lib/video'
 import { VideoPathManager } from '@server/lib/video-path-manager'
 import { buildNextVideoState } from '@server/lib/video-state'
 import { ThumbnailModel } from '@server/models/video/thumbnail'
@@ -39,6 +39,7 @@ import { MThumbnail } from '../../../types/models/video/thumbnail'
 import { federateVideoIfNeeded } from '../../activitypub/videos'
 import { Notifier } from '../../notifier'
 import { generateVideoMiniature } from '../../thumbnail'
+import { JobQueue } from '../job-queue'
 
 async function processVideoImport (job: Job) {
   const payload = job.data as VideoImportPayload
@@ -65,7 +66,7 @@ export {
 // ---------------------------------------------------------------------------
 
 async function processTorrentImport (job: Job, videoImport: MVideoImportDefault, payload: VideoImportTorrentPayload) {
-  logger.info('Processing torrent video import in job %d.', job.id)
+  logger.info('Processing torrent video import in job %s.', job.id)
 
   const options = { type: payload.type, videoImportId: payload.videoImportId }
 
@@ -77,7 +78,7 @@ async function processTorrentImport (job: Job, videoImport: MVideoImportDefault,
 }
 
 async function processYoutubeDLImport (job: Job, videoImport: MVideoImportDefault, payload: VideoImportYoutubeDLPayload) {
-  logger.info('Processing youtubeDL video import in job %d.', job.id)
+  logger.info('Processing youtubeDL video import in job %s.', job.id)
 
   const options = { type: payload.type, videoImportId: videoImport.id }
 
@@ -259,12 +260,16 @@ async function processFile (downloader: () => Promise<string>, videoImport: MVid
     }
 
     if (video.state === VideoState.TO_MOVE_TO_EXTERNAL_STORAGE) {
-      return addMoveToObjectStorageJob({ video: videoImportUpdated.Video, previousVideoState: VideoState.TO_IMPORT })
+      await JobQueue.Instance.createJob(
+        await buildMoveToObjectStorageJob({ video: videoImportUpdated.Video, previousVideoState: VideoState.TO_IMPORT })
+      )
     }
 
     // Create transcoding jobs?
     if (video.state === VideoState.TO_TRANSCODE) {
-      await addOptimizeOrMergeAudioJob({ video: videoImportUpdated.Video, videoFile, user: videoImport.User })
+      await JobQueue.Instance.createJob(
+        await buildOptimizeOrMergeAudioJob({ video: videoImportUpdated.Video, videoFile, user: videoImport.User })
+      )
     }
 
   } catch (err) {
diff --git a/server/lib/job-queue/handlers/video-redundancy.ts b/server/lib/job-queue/handlers/video-redundancy.ts
index 75ab2cd02..bac99fdb7 100644
--- a/server/lib/job-queue/handlers/video-redundancy.ts
+++ b/server/lib/job-queue/handlers/video-redundancy.ts
@@ -5,7 +5,7 @@ import { logger } from '../../../helpers/logger'
 
 async function processVideoRedundancy (job: Job) {
   const payload = job.data as VideoRedundancyPayload
-  logger.info('Processing video redundancy in job %d.', job.id)
+  logger.info('Processing video redundancy in job %s.', job.id)
 
   return VideosRedundancyScheduler.Instance.createManualRedundancy(payload.videoId)
 }
diff --git a/server/lib/job-queue/handlers/video-studio-edition.ts b/server/lib/job-queue/handlers/video-studio-edition.ts
index 078243538..23f9a34cc 100644
--- a/server/lib/job-queue/handlers/video-studio-edition.ts
+++ b/server/lib/job-queue/handlers/video-studio-edition.ts
@@ -8,7 +8,7 @@ import { federateVideoIfNeeded } from '@server/lib/activitypub/videos'
 import { generateWebTorrentVideoFilename } from '@server/lib/paths'
 import { VideoTranscodingProfilesManager } from '@server/lib/transcoding/default-transcoding-profiles'
 import { isAbleToUploadVideo } from '@server/lib/user'
-import { addOptimizeOrMergeAudioJob } from '@server/lib/video'
+import { buildOptimizeOrMergeAudioJob } from '@server/lib/video'
 import { removeHLSPlaylist, removeWebTorrentFile } from '@server/lib/video-file'
 import { VideoPathManager } from '@server/lib/video-path-manager'
 import { approximateIntroOutroAdditionalSize } from '@server/lib/video-studio'
@@ -36,6 +36,7 @@ import {
   VideoStudioTaskWatermarkPayload
 } from '@shared/models'
 import { logger, loggerTagsFactory } from '../../../helpers/logger'
+import { JobQueue } from '../job-queue'
 
 const lTagsBase = loggerTagsFactory('video-edition')
 
@@ -43,7 +44,7 @@ async function processVideoStudioEdition (job: Job) {
   const payload = job.data as VideoStudioEditionPayload
   const lTags = lTagsBase(payload.videoUUID)
 
-  logger.info('Process video studio edition of %s in job %d.', payload.videoUUID, job.id, lTags)
+  logger.info('Process video studio edition of %s in job %s.', payload.videoUUID, job.id, lTags)
 
   const video = await VideoModel.loadFull(payload.videoUUID)
 
@@ -100,7 +101,10 @@ async function processVideoStudioEdition (job: Job) {
   await federateVideoIfNeeded(video, false, undefined)
 
   const user = await UserModel.loadByVideoId(video.id)
-  await addOptimizeOrMergeAudioJob({ video, videoFile: newFile, user, isNewVideo: false })
+
+  await JobQueue.Instance.createJob(
+    await buildOptimizeOrMergeAudioJob({ video, videoFile: newFile, user, isNewVideo: false })
+  )
 }
 
 // ---------------------------------------------------------------------------
diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts
index 0cf5d53ce..50d732beb 100644
--- a/server/lib/job-queue/job-queue.ts
+++ b/server/lib/job-queue/job-queue.ts
@@ -1,4 +1,6 @@
 import {
+  FlowJob,
+  FlowProducer,
   Job,
   JobsOptions,
   Queue,
@@ -13,7 +15,7 @@ import {
 import { jobStates } from '@server/helpers/custom-validators/jobs'
 import { CONFIG } from '@server/initializers/config'
 import { processVideoRedundancy } from '@server/lib/job-queue/handlers/video-redundancy'
-import { timeoutPromise } from '@shared/core-utils'
+import { pick, timeoutPromise } from '@shared/core-utils'
 import {
   ActivitypubFollowPayload,
   ActivitypubHttpBroadcastPayload,
@@ -22,10 +24,12 @@ import {
   ActorKeysPayload,
   DeleteResumableUploadMetaFilePayload,
   EmailPayload,
+  FederateVideoPayload,
   JobState,
   JobType,
   ManageVideoTorrentPayload,
   MoveObjectStoragePayload,
+  NotifyPayload,
   RefreshPayload,
   VideoFileImportPayload,
   VideoImportPayload,
@@ -45,8 +49,10 @@ import { processActivityPubHttpUnicast } from './handlers/activitypub-http-unica
 import { refreshAPObject } from './handlers/activitypub-refresher'
 import { processActorKeys } from './handlers/actor-keys'
 import { processEmail } from './handlers/email'
+import { processFederateVideo } from './handlers/federate-video'
 import { processManageVideoTorrent } from './handlers/manage-video-torrent'
 import { onMoveToObjectStorageFailure, processMoveToObjectStorage } from './handlers/move-to-object-storage'
+import { processNotify } from './handlers/notify'
 import { processVideoFileImport } from './handlers/video-file-import'
 import { processVideoImport } from './handlers/video-import'
 import { processVideoLiveEnding } from './handlers/video-live-ending'
@@ -54,7 +60,7 @@ import { processVideoStudioEdition } from './handlers/video-studio-edition'
 import { processVideoTranscoding } from './handlers/video-transcoding'
 import { processVideosViewsStats } from './handlers/video-views-stats'
 
-type CreateJobArgument =
+export type CreateJobArgument =
   { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
   { type: 'activitypub-http-broadcast-parallel', payload: ActivitypubHttpBroadcastPayload } |
   { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } |
@@ -73,7 +79,9 @@ type CreateJobArgument =
   { type: 'delete-resumable-upload-meta-file', payload: DeleteResumableUploadMetaFilePayload } |
   { type: 'video-studio-edition', payload: VideoStudioEditionPayload } |
   { type: 'manage-video-torrent', payload: ManageVideoTorrentPayload } |
-  { type: 'move-to-object-storage', payload: MoveObjectStoragePayload }
+  { type: 'notify', payload: NotifyPayload } |
+  { type: 'move-to-object-storage', payload: MoveObjectStoragePayload } |
+  { type: 'federate-video', payload: FederateVideoPayload }
 
 export type CreateJobOptions = {
   delay?: number
@@ -98,7 +106,9 @@ const handlers: { [id in JobType]: (job: Job) => Promise<any> } = {
   'video-redundancy': processVideoRedundancy,
   'move-to-object-storage': processMoveToObjectStorage,
   'manage-video-torrent': processManageVideoTorrent,
-  'video-studio-edition': processVideoStudioEdition
+  'notify': processNotify,
+  'video-studio-edition': processVideoStudioEdition,
+  'federate-video': processFederateVideo
 }
 
 const errorHandlers: { [id in JobType]?: (job: Job, err: any) => Promise<any> } = {
@@ -123,7 +133,9 @@ const jobTypes: JobType[] = [
   'video-live-ending',
   'move-to-object-storage',
   'manage-video-torrent',
-  'video-studio-edition'
+  'video-studio-edition',
+  'notify',
+  'federate-video'
 ]
 
 const silentFailure = new Set<JobType>([ 'activitypub-http-unicast' ])
@@ -137,6 +149,8 @@ class JobQueue {
   private queueSchedulers: { [id in JobType]?: QueueScheduler } = {}
   private queueEvents: { [id in JobType]?: QueueEvents } = {}
 
+  private flowProducer: FlowProducer
+
   private initialized = false
   private jobRedisPrefix: string
 
@@ -157,6 +171,11 @@ class JobQueue {
       this.buildQueueEvent(handlerName, produceOnly)
     }
 
+    this.flowProducer = new FlowProducer({
+      connection: this.getRedisConnection(),
+      prefix: this.jobRedisPrefix
+    })
+
     this.addRepeatableJobs()
   }
 
@@ -243,6 +262,8 @@ class JobQueue {
     }
   }
 
+  // ---------------------------------------------------------------------------
+
   async terminate () {
     const promises = Object.keys(this.workers)
       .map(handlerName => {
@@ -278,28 +299,56 @@ class JobQueue {
     }
   }
 
-  createJob (obj: CreateJobArgument, options: CreateJobOptions = {}): void {
-    this.createJobWithPromise(obj, options)
-        .catch(err => logger.error('Cannot create job.', { err, obj }))
+  // ---------------------------------------------------------------------------
+
+  createJobAsync (options: CreateJobArgument & CreateJobOptions): void {
+    this.createJob(options)
+        .catch(err => logger.error('Cannot create job.', { err, options }))
   }
 
-  async createJobWithPromise (obj: CreateJobArgument, options: CreateJobOptions = {}) {
-    const queue: Queue = this.queues[obj.type]
+  async createJob (options: CreateJobArgument & CreateJobOptions) {
+    const queue: Queue = this.queues[options.type]
     if (queue === undefined) {
-      logger.error('Unknown queue %s: cannot create job.', obj.type)
+      logger.error('Unknown queue %s: cannot create job.', options.type)
       return
     }
 
-    const jobArgs: JobsOptions = {
+    const jobOptions = this.buildJobOptions(options.type as JobType, pick(options, [ 'priority', 'delay' ]))
+
+    return queue.add('job', options.payload, jobOptions)
+  }
+
+  async createSequentialJobFlow (...jobs: ((CreateJobArgument & CreateJobOptions) | undefined)[]) {
+    let lastJob: FlowJob
+
+    for (const job of jobs) {
+      if (!job) continue
+
+      lastJob = {
+        name: 'job',
+        data: job.payload,
+        queueName: job.type,
+        opts: this.buildJobOptions(job.type as JobType, pick(job, [ 'priority', 'delay' ])),
+        children: lastJob
+          ? [ lastJob ]
+          : []
+      }
+    }
+
+    return this.flowProducer.add(lastJob)
+  }
+
+  private buildJobOptions (type: JobType, options: CreateJobOptions = {}): JobsOptions {
+    return {
       backoff: { delay: 60 * 1000, type: 'exponential' },
-      attempts: JOB_ATTEMPTS[obj.type],
+      attempts: JOB_ATTEMPTS[type],
       priority: options.priority,
       delay: options.delay
     }
-
-    return queue.add('job', obj.payload, jobArgs)
   }
 
+  // ---------------------------------------------------------------------------
+
   async listForApi (options: {
     state?: JobState
     start: number
@@ -367,6 +416,8 @@ class JobQueue {
     return Promise.all(promises)
   }
 
+  // ---------------------------------------------------------------------------
+
   async removeOldJobs () {
     for (const key of Object.keys(this.queues)) {
       const queue: Queue = this.queues[key]
diff --git a/server/lib/live/live-manager.ts b/server/lib/live/live-manager.ts
index 1410889a2..aadd8e308 100644
--- a/server/lib/live/live-manager.ts
+++ b/server/lib/live/live-manager.ts
@@ -408,7 +408,7 @@ class LiveManager {
         await liveSession.save()
       }
 
-      JobQueue.Instance.createJob({
+      JobQueue.Instance.createJobAsync({
         type: 'video-live-ending',
         payload: {
           videoId: fullVideo.id,
@@ -421,8 +421,12 @@ class LiveManager {
           streamingPlaylistId: fullVideo.getHLSPlaylist()?.id,
 
           publishedAt: fullVideo.publishedAt.toISOString()
-        }
-      }, { delay: cleanupNow ? 0 : VIDEO_LIVE.CLEANUP_DELAY })
+        },
+
+        delay: cleanupNow
+          ? 0
+          : VIDEO_LIVE.CLEANUP_DELAY
+      })
 
       fullVideo.state = live.permanentLive
         ? VideoState.WAITING_FOR_LIVE
diff --git a/server/lib/notifier/notifier.ts b/server/lib/notifier/notifier.ts
index d1c4c0215..66cfc31c4 100644
--- a/server/lib/notifier/notifier.ts
+++ b/server/lib/notifier/notifier.ts
@@ -242,7 +242,7 @@ class Notifier {
 
     for (const to of toEmails) {
       const payload = await object.createEmail(to)
-      JobQueue.Instance.createJob({ type: 'email', payload })
+      JobQueue.Instance.createJobAsync({ type: 'email', payload })
     }
   }
 
diff --git a/server/lib/schedulers/auto-follow-index-instances.ts b/server/lib/schedulers/auto-follow-index-instances.ts
index d9f9c2de3..956ece749 100644
--- a/server/lib/schedulers/auto-follow-index-instances.ts
+++ b/server/lib/schedulers/auto-follow-index-instances.ts
@@ -59,7 +59,7 @@ export class AutoFollowIndexInstances extends AbstractScheduler {
             isAutoFollow: true
           }
 
-          JobQueue.Instance.createJob({ type: 'activitypub-follow', payload })
+          JobQueue.Instance.createJobAsync({ type: 'activitypub-follow', payload })
         }
       }
 
diff --git a/server/lib/video-state.ts b/server/lib/video-state.ts
index b5d8353b7..9ebbd7679 100644
--- a/server/lib/video-state.ts
+++ b/server/lib/video-state.ts
@@ -1,4 +1,5 @@
 import { Transaction } from 'sequelize'
+import { retryTransactionWrapper } from '@server/helpers/database-utils'
 import { logger } from '@server/helpers/logger'
 import { CONFIG } from '@server/initializers/config'
 import { sequelizeTypescript } from '@server/initializers/database'
@@ -7,9 +8,9 @@ import { VideoJobInfoModel } from '@server/models/video/video-job-info'
 import { MVideo, MVideoFullLight, MVideoUUID } from '@server/types/models'
 import { VideoState } from '@shared/models'
 import { federateVideoIfNeeded } from './activitypub/videos'
+import { JobQueue } from './job-queue'
 import { Notifier } from './notifier'
-import { addMoveToObjectStorageJob } from './video'
-import { retryTransactionWrapper } from '@server/helpers/database-utils'
+import { buildMoveToObjectStorageJob } from './video'
 
 function buildNextVideoState (currentState?: VideoState) {
   if (currentState === VideoState.PUBLISHED) {
@@ -86,7 +87,7 @@ async function moveToExternalStorageState (options: {
   logger.info('Creating external storage move job for video %s.', video.uuid, { tags: [ video.uuid ] })
 
   try {
-    await addMoveToObjectStorageJob({ video, previousVideoState, isNewVideo })
+    await JobQueue.Instance.createJob(await buildMoveToObjectStorageJob({ video, previousVideoState, isNewVideo }))
 
     return true
   } catch (err) {
diff --git a/server/lib/video.ts b/server/lib/video.ts
index b843b11bc..f7d7aa186 100644
--- a/server/lib/video.ts
+++ b/server/lib/video.ts
@@ -1,5 +1,7 @@
 import { UploadFiles } from 'express'
+import memoizee from 'memoizee'
 import { Transaction } from 'sequelize/types'
+import { CONFIG } from '@server/initializers/config'
 import { DEFAULT_AUDIO_RESOLUTION, JOB_PRIORITY, MEMOIZE_LENGTH, MEMOIZE_TTL } from '@server/initializers/constants'
 import { TagModel } from '@server/models/video/tag'
 import { VideoModel } from '@server/models/video/video'
@@ -9,8 +11,6 @@ import { MThumbnail, MUserId, MVideoFile, MVideoTag, MVideoThumbnail, MVideoUUID
 import { ThumbnailType, VideoCreate, VideoPrivacy, VideoState, VideoTranscodingPayload } from '@shared/models'
 import { CreateJobOptions, JobQueue } from './job-queue/job-queue'
 import { updateVideoMiniatureFromExisting } from './thumbnail'
-import { CONFIG } from '@server/initializers/config'
-import memoizee from 'memoizee'
 
 function buildLocalVideoFromReq (videoInfo: VideoCreate, channelId: number): FilteredModelAttributes<VideoModel> {
   return {
@@ -86,7 +86,7 @@ async function setVideoTags (options: {
 
 // ---------------------------------------------------------------------------
 
-async function addOptimizeOrMergeAudioJob (options: {
+async function buildOptimizeOrMergeAudioJob (options: {
   video: MVideoUUID
   videoFile: MVideoFile
   user: MUserId
@@ -94,10 +94,10 @@ async function addOptimizeOrMergeAudioJob (options: {
 }) {
   const { video, videoFile, user, isNewVideo } = options
 
-  let dataInput: VideoTranscodingPayload
+  let payload: VideoTranscodingPayload
 
   if (videoFile.isAudio()) {
-    dataInput = {
+    payload = {
       type: 'merge-audio-to-webtorrent',
       resolution: DEFAULT_AUDIO_RESOLUTION,
       videoUUID: video.uuid,
@@ -105,24 +105,26 @@ async function addOptimizeOrMergeAudioJob (options: {
       isNewVideo
     }
   } else {
-    dataInput = {
+    payload = {
       type: 'optimize-to-webtorrent',
       videoUUID: video.uuid,
       isNewVideo
     }
   }
 
-  const jobOptions = {
-    priority: await getTranscodingJobPriority(user)
-  }
+  await VideoJobInfoModel.increaseOrCreate(payload.videoUUID, 'pendingTranscode')
 
-  return addTranscodingJob(dataInput, jobOptions)
+  return {
+    type: 'video-transcoding' as 'video-transcoding',
+    priority: await getTranscodingJobPriority(user),
+    payload
+  }
 }
 
 async function addTranscodingJob (payload: VideoTranscodingPayload, options: CreateJobOptions = {}) {
   await VideoJobInfoModel.increaseOrCreate(payload.videoUUID, 'pendingTranscode')
 
-  return JobQueue.Instance.createJobWithPromise({ type: 'video-transcoding', payload }, options)
+  return JobQueue.Instance.createJob({ type: 'video-transcoding', payload, ...options })
 }
 
 async function getTranscodingJobPriority (user: MUserId) {
@@ -136,7 +138,7 @@ async function getTranscodingJobPriority (user: MUserId) {
 
 // ---------------------------------------------------------------------------
 
-async function addMoveToObjectStorageJob (options: {
+async function buildMoveToObjectStorageJob (options: {
   video: MVideoUUID
   previousVideoState: VideoState
   isNewVideo?: boolean // Default true
@@ -145,8 +147,14 @@ async function addMoveToObjectStorageJob (options: {
 
   await VideoJobInfoModel.increaseOrCreate(video.uuid, 'pendingMove')
 
-  const dataInput = { videoUUID: video.uuid, isNewVideo, previousVideoState }
-  return JobQueue.Instance.createJobWithPromise({ type: 'move-to-object-storage', payload: dataInput })
+  return {
+    type: 'move-to-object-storage' as 'move-to-object-storage',
+    payload: {
+      videoUUID: video.uuid,
+      isNewVideo,
+      previousVideoState
+    }
+  }
 }
 
 // ---------------------------------------------------------------------------
@@ -173,9 +181,9 @@ export {
   buildLocalVideoFromReq,
   buildVideoThumbnailsFromReq,
   setVideoTags,
-  addOptimizeOrMergeAudioJob,
+  buildOptimizeOrMergeAudioJob,
   addTranscodingJob,
-  addMoveToObjectStorageJob,
+  buildMoveToObjectStorageJob,
   getTranscodingJobPriority,
   getCachedVideoDuration
 }