Use bullmq job dependency

15 files changed, 87 insertions, 22 deletions

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 })
+  )
 }
 
 // ---------------------------------------------------------------------------