Move to bullmq

3 files changed, 11 insertions, 10 deletions

diff --git a/server/controllers/api/jobs.ts b/server/controllers/api/jobs.ts
index c61b7362f..6a53e3083 100644
--- a/server/controllers/api/jobs.ts
+++ b/server/controllers/api/jobs.ts
@@ -1,3 +1,4 @@
+import { Job as BullJob } from 'bullmq'
 import express from 'express'
 import { HttpStatusCode, Job, JobState, JobType, ResultList, UserRight } from '@shared/models'
 import { isArray } from '../../helpers/custom-validators/misc'
@@ -25,7 +26,7 @@ jobsRouter.post('/pause',
 jobsRouter.post('/resume',
   authenticate,
   ensureUserHasRight(UserRight.MANAGE_JOBS),
-  asyncMiddleware(resumeJobQueue)
+  resumeJobQueue
 )
 
 jobsRouter.get('/:state?',
@@ -54,8 +55,8 @@ async function pauseJobQueue (req: express.Request, res: express.Response) {
   return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
 }
 
-async function resumeJobQueue (req: express.Request, res: express.Response) {
-  await JobQueue.Instance.resume()
+function resumeJobQueue (req: express.Request, res: express.Response) {
+  JobQueue.Instance.resume()
 
   return res.sendStatus(HttpStatusCode.NO_CONTENT_204)
 }
@@ -82,7 +83,7 @@ async function listJobs (req: express.Request, res: express.Response) {
   return res.json(result)
 }
 
-async function formatJob (job: any, state?: JobState): Promise<Job> {
+async function formatJob (job: BullJob, state?: JobState): Promise<Job> {
   const error = isArray(job.stacktrace) && job.stacktrace.length !== 0
     ? job.stacktrace[0]
     : null
@@ -90,9 +91,9 @@ async function formatJob (job: any, state?: JobState): Promise<Job> {
   return {
     id: job.id,
     state: state || await job.getState(),
-    type: job.queue.name as JobType,
+    type: job.queueName as JobType,
     data: job.data,
-    progress: await job.progress(),
+    progress: job.progress as number,
     priority: job.opts.priority,
     error,
     createdAt: new Date(job.timestamp),
diff --git a/server/controllers/api/videos/update.ts b/server/controllers/api/videos/update.ts
index 65a7321fd..1545a2232 100644
--- a/server/controllers/api/videos/update.ts
+++ b/server/controllers/api/videos/update.ts
@@ -199,7 +199,7 @@ async function updateTorrentsMetadataIfNeeded (video: MVideoFullLight, videoInfo
     const payload: ManageVideoTorrentPayload = { action: 'update-metadata', videoId: video.id, videoFileId: file.id }
 
     const job = await JobQueue.Instance.createJobWithPromise({ type: 'manage-video-torrent', payload })
-    await job.finished()
+    await JobQueue.Instance.waitJob(job)
   }
 
   const hls = video.getHLSPlaylist()
@@ -208,7 +208,7 @@ async function updateTorrentsMetadataIfNeeded (video: MVideoFullLight, videoInfo
     const payload: ManageVideoTorrentPayload = { action: 'update-metadata', streamingPlaylistId: hls.id, videoFileId: file.id }
 
     const job = await JobQueue.Instance.createJobWithPromise({ type: 'manage-video-torrent', payload })
-    await job.finished()
+    await JobQueue.Instance.waitJob(job)
   }
 
   // Refresh video since files have changed
diff --git a/server/controllers/api/videos/upload.ts b/server/controllers/api/videos/upload.ts
index 3ce66c9ca..4a9d7b619 100644
--- a/server/controllers/api/videos/upload.ts
+++ b/server/controllers/api/videos/upload.ts
@@ -17,6 +17,7 @@ import {
 import { VideoPathManager } from '@server/lib/video-path-manager'
 import { buildNextVideoState } from '@server/lib/video-state'
 import { openapiOperationDoc } from '@server/middlewares/doc'
+import { VideoSourceModel } from '@server/models/video/video-source'
 import { MVideoFile, MVideoFullLight } from '@server/types/models'
 import { getLowercaseExtension } from '@shared/core-utils'
 import { isAudioFile, uuidToShort } from '@shared/extra-utils'
@@ -44,7 +45,6 @@ import {
 import { ScheduleVideoUpdateModel } from '../../../models/video/schedule-video-update'
 import { VideoModel } from '../../../models/video/video'
 import { VideoFileModel } from '../../../models/video/video-file'
-import { VideoSourceModel } from '@server/models/video/video-source'
 
 const lTags = loggerTagsFactory('api', 'video')
 const auditLogger = auditLoggerFactory('videos')
@@ -270,7 +270,7 @@ async function createTorrentFederate (video: MVideoFullLight, videoFile: MVideoF
   const payload: ManageVideoTorrentPayload = { videoId: video.id, videoFileId: videoFile.id, action: 'create' }
 
   const job = await JobQueue.Instance.createJobWithPromise({ type: 'manage-video-torrent', payload })
-  await job.finished()
+  await JobQueue.Instance.waitJob(job)
 
   const refreshedVideo = await VideoModel.loadFull(video.id)
   if (!refreshedVideo) return