Migrate to bull

8 files changed, 95 insertions, 122 deletions

diff --git a/server/lib/job-queue/handlers/activitypub-follow.ts b/server/lib/job-queue/handlers/activitypub-follow.ts
index 286e343f2..2c1b4f49d 100644
--- a/server/lib/job-queue/handlers/activitypub-follow.ts
+++ b/server/lib/job-queue/handlers/activitypub-follow.ts
@@ -1,4 +1,4 @@
-import * as kue from 'kue'
+import * as Bull from 'bull'
 import { logger } from '../../../helpers/logger'
 import { getServerActor } from '../../../helpers/utils'
 import { REMOTE_SCHEME, sequelizeTypescript, SERVER_ACTOR_NAME } from '../../../initializers'
@@ -14,7 +14,7 @@ export type ActivitypubFollowPayload = {
   host: string
 }
 
-async function processActivityPubFollow (job: kue.Job) {
+async function processActivityPubFollow (job: Bull.Job) {
   const payload = job.data as ActivitypubFollowPayload
   const host = payload.host
 
diff --git a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
index d8b8ec222..03a9e12a4 100644
--- a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
@@ -1,4 +1,4 @@
-import * as kue from 'kue'
+import * as Bull from 'bull'
 import * as Bluebird from 'bluebird'
 import { logger } from '../../../helpers/logger'
 import { doRequest } from '../../../helpers/requests'
@@ -12,7 +12,7 @@ export type ActivitypubHttpBroadcastPayload = {
   body: any
 }
 
-async function processActivityPubHttpBroadcast (job: kue.Job) {
+async function processActivityPubHttpBroadcast (job: Bull.Job) {
   logger.info('Processing ActivityPub broadcast in job %d.', 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 10c0e606f..f21da087e 100644
--- a/server/lib/job-queue/handlers/activitypub-http-fetcher.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-fetcher.ts
@@ -1,4 +1,4 @@
-import * as kue from 'kue'
+import * as Bull from 'bull'
 import { logger } from '../../../helpers/logger'
 import { processActivities } from '../../activitypub/process'
 import { ActivitypubHttpBroadcastPayload } from './activitypub-http-broadcast'
@@ -9,7 +9,7 @@ export type ActivitypubHttpFetcherPayload = {
   uris: string[]
 }
 
-async function processActivityPubHttpFetcher (job: kue.Job) {
+async function processActivityPubHttpFetcher (job: Bull.Job) {
   logger.info('Processing ActivityPub fetcher in job %d.', job.id)
 
   const payload = job.data as ActivitypubHttpBroadcastPayload
diff --git a/server/lib/job-queue/handlers/activitypub-http-unicast.ts b/server/lib/job-queue/handlers/activitypub-http-unicast.ts
index 173f3bb52..c90d735f6 100644
--- a/server/lib/job-queue/handlers/activitypub-http-unicast.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-unicast.ts
@@ -1,4 +1,4 @@
-import * as kue from 'kue'
+import * as Bull from 'bull'
 import { logger } from '../../../helpers/logger'
 import { doRequest } from '../../../helpers/requests'
 import { ActorFollowModel } from '../../../models/activitypub/actor-follow'
@@ -11,7 +11,7 @@ export type ActivitypubHttpUnicastPayload = {
   body: any
 }
 
-async function processActivityPubHttpUnicast (job: kue.Job) {
+async function processActivityPubHttpUnicast (job: Bull.Job) {
   logger.info('Processing ActivityPub unicast in job %d.', job.id)
 
   const payload = job.data as ActivitypubHttpUnicastPayload
diff --git a/server/lib/job-queue/handlers/email.ts b/server/lib/job-queue/handlers/email.ts
index 9d7686116..73d98ae54 100644
--- a/server/lib/job-queue/handlers/email.ts
+++ b/server/lib/job-queue/handlers/email.ts
@@ -1,4 +1,4 @@
-import * as kue from 'kue'
+import * as Bull from 'bull'
 import { logger } from '../../../helpers/logger'
 import { Emailer } from '../../emailer'
 
@@ -8,7 +8,7 @@ export type EmailPayload = {
   text: string
 }
 
-async function processEmail (job: kue.Job) {
+async function processEmail (job: Bull.Job) {
   const payload = job.data as EmailPayload
   logger.info('Processing email in job %d.', job.id)
 
diff --git a/server/lib/job-queue/handlers/video-file.ts b/server/lib/job-queue/handlers/video-file.ts
index fc40527c7..bd68dd78b 100644
--- a/server/lib/job-queue/handlers/video-file.ts
+++ b/server/lib/job-queue/handlers/video-file.ts
@@ -1,4 +1,4 @@
-import * as kue from 'kue'
+import * as Bull from 'bull'
 import { VideoResolution, VideoState } from '../../../../shared'
 import { logger } from '../../../helpers/logger'
 import { computeResolutionsToTranscode } from '../../../helpers/utils'
@@ -7,6 +7,7 @@ import { JobQueue } from '../job-queue'
 import { federateVideoIfNeeded } from '../../activitypub'
 import { retryTransactionWrapper } from '../../../helpers/database-utils'
 import { sequelizeTypescript } from '../../../initializers'
+import * as Bluebird from 'bluebird'
 
 export type VideoFilePayload = {
   videoUUID: string
@@ -20,7 +21,7 @@ export type VideoFileImportPayload = {
   filePath: string
 }
 
-async function processVideoFileImport (job: kue.Job) {
+async function processVideoFileImport (job: Bull.Job) {
   const payload = job.data as VideoFileImportPayload
   logger.info('Processing video file import in job %d.', job.id)
 
@@ -37,7 +38,7 @@ async function processVideoFileImport (job: kue.Job) {
   return video
 }
 
-async function processVideoFile (job: kue.Job) {
+async function processVideoFile (job: Bull.Job) {
   const payload = job.data as VideoFilePayload
   logger.info('Processing video file in job %d.', job.id)
 
@@ -109,7 +110,7 @@ async function onVideoFileOptimizerSuccess (video: VideoModel, isNewVideo: boole
     )
 
     if (resolutionsEnabled.length !== 0) {
-      const tasks: Promise<any>[] = []
+      const tasks: Bluebird<any>[] = []
 
       for (const resolution of resolutionsEnabled) {
         const dataInput = {
diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts
index 695fe0eea..77aaa7fa8 100644
--- a/server/lib/job-queue/job-queue.ts
+++ b/server/lib/job-queue/job-queue.ts
@@ -1,13 +1,12 @@
-import * as kue from 'kue'
+import * as Bull from 'bull'
 import { JobState, JobType } from '../../../shared/models'
 import { logger } from '../../helpers/logger'
 import { CONFIG, JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_REQUEST_TTL } from '../../initializers'
-import { Redis } from '../redis'
 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 { EmailPayload, processEmail } from './handlers/email'
-import { processVideoFile, processVideoFileImport, VideoFilePayload, VideoFileImportPayload } from './handlers/video-file'
+import { processVideoFile, processVideoFileImport, VideoFileImportPayload, VideoFilePayload } from './handlers/video-file'
 import { ActivitypubFollowPayload, processActivityPubFollow } from './handlers/activitypub-follow'
 
 type CreateJobArgument =
@@ -19,7 +18,7 @@ type CreateJobArgument =
   { type: 'video-file', payload: VideoFilePayload } |
   { type: 'email', payload: EmailPayload }
 
-const handlers: { [ id in JobType ]: (job: kue.Job) => Promise<any>} = {
+const handlers: { [ id in JobType ]: (job: Bull.Job) => Promise<any>} = {
   'activitypub-http-broadcast': processActivityPubHttpBroadcast,
   'activitypub-http-unicast': processActivityPubHttpUnicast,
   'activitypub-http-fetcher': processActivityPubHttpFetcher,
@@ -29,18 +28,28 @@ const handlers: { [ id in JobType ]: (job: kue.Job) => Promise<any>} = {
   'email': processEmail
 }
 
-const jobsWithTLL: JobType[] = [
+const jobsWithRequestTimeout: { [ id in JobType ]?: boolean } = {
+  'activitypub-http-broadcast': true,
+  'activitypub-http-unicast': true,
+  'activitypub-http-fetcher': true,
+  'activitypub-follow': true
+}
+
+const jobTypes: JobType[] = [
+  'activitypub-follow',
   'activitypub-http-broadcast',
-  'activitypub-http-unicast',
   'activitypub-http-fetcher',
-  'activitypub-follow'
+  'activitypub-http-unicast',
+  'email',
+  'video-file',
+  'video-file-import'
 ]
 
 class JobQueue {
 
   private static instance: JobQueue
 
-  private jobQueue: kue.Queue
+  private queues: { [ id in JobType ]?: Bull.Queue } = {}
   private initialized = false
   private jobRedisPrefix: string
 
@@ -51,9 +60,8 @@ class JobQueue {
     if (this.initialized === true) return
     this.initialized = true
 
-    this.jobRedisPrefix = 'q-' + CONFIG.WEBSERVER.HOST
-
-    this.jobQueue = kue.createQueue({
+    this.jobRedisPrefix = 'bull-' + CONFIG.WEBSERVER.HOST
+    const queueOptions = {
       prefix: this.jobRedisPrefix,
       redis: {
         host: CONFIG.REDIS.HOSTNAME,
@@ -61,120 +69,94 @@ class JobQueue {
         auth: CONFIG.REDIS.AUTH,
         db: CONFIG.REDIS.DB
       }
-    })
-
-    this.jobQueue.setMaxListeners(20)
+    }
 
-    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)) {
+      const queue = new Bull(handlerName, queueOptions)
+      const handler = handlers[handlerName]
 
-    await this.reactiveStuckJobs()
+      queue.process(JOB_CONCURRENCY[handlerName], handler)
+        .catch(err => logger.error('Cannot execute job queue %s.', handlerName, { err }))
 
-    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) {
-          logger.error('Cannot execute job %d.', job.id, { err })
-          return done(err)
-        }
+      queue.on('error', err => {
+        logger.error('Error in job queue %s.', handlerName, { err })
+        process.exit(-1)
       })
+
+      this.queues[handlerName] = queue
     }
   }
 
-  createJob (obj: CreateJobArgument, priority = 'normal') {
-    return new Promise((res, rej) => {
-      let job = this.jobQueue
-        .create(obj.type, obj.payload)
-        .priority(priority)
-        .attempts(JOB_ATTEMPTS[obj.type])
-        .backoff({ delay: 60 * 1000, type: 'exponential' })
+  createJob (obj: CreateJobArgument) {
+    const queue = this.queues[obj.type]
+    if (queue === undefined) {
+      logger.error('Unknown queue %s: cannot create job.', obj.type)
+      return
+    }
 
-      if (jobsWithTLL.indexOf(obj.type) !== -1) {
-        job = job.ttl(JOB_REQUEST_TTL)
-      }
+    const jobArgs: Bull.JobOptions = {
+      backoff: { delay: 60 * 1000, type: 'exponential' },
+      attempts: JOB_ATTEMPTS[obj.type]
+    }
 
-      return job.save(err => {
-        if (err) return rej(err)
+    if (jobsWithRequestTimeout[obj.type] === true) {
+      jobArgs.timeout = JOB_REQUEST_TTL
+    }
 
-        return res()
-      })
-    })
+    return queue.add(obj.payload, jobArgs)
   }
 
-  async listForApi (state: JobState, start: number, count: number, sort: 'ASC' | 'DESC'): Promise<kue.Job[]> {
-    const jobStrings = await Redis.Instance.listJobs(this.jobRedisPrefix, state, 'alpha', sort, start, count)
+  async listForApi (state: JobState, start: number, count: number, asc?: boolean): Promise<Bull.Job[]> {
+    let results: Bull.Job[] = []
 
-    const jobPromises = jobStrings
-      .map(s => s.split('|'))
-      .map(([ , jobId ]) => this.getJob(parseInt(jobId, 10)))
+    // TODO: optimize
+    for (const jobType of jobTypes) {
+      const queue = this.queues[ jobType ]
+      if (queue === undefined) {
+        logger.error('Unknown queue %s to list jobs.', jobType)
+        continue
+      }
 
-    return Promise.all(jobPromises)
-  }
+      // FIXME: Bull queue typings does not have getJobs method
+      const jobs = await (queue as any).getJobs(state, 0, start + count, asc)
+      results = results.concat(jobs)
+    }
 
-  count (state: JobState) {
-    return new Promise<number>((res, rej) => {
-      this.jobQueue[state + 'Count']((err, total) => {
-        if (err) return rej(err)
+    results.sort((j1: any, j2: any) => {
+      if (j1.timestamp < j2.timestamp) return -1
+      else if (j1.timestamp === j2.timestamp) return 0
 
-        return res(total)
-      })
+      return 1
     })
-  }
 
-  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
-      }
+    if (asc === false) results.reverse()
 
-      for (const job of jobs) {
-        if (now - job.created_at > JOB_COMPLETED_LIFETIME) {
-          job.remove()
-        }
-      }
-    })
+    return results.slice(start, start + count)
   }
 
-  private reactiveStuckJobs () {
-    const promises: Promise<any>[] = []
-
-    this.jobQueue.active((err, ids) => {
-      if (err) throw err
+  async count (state: JobState): Promise<number> {
+    let total = 0
 
-      for (const id of ids) {
-        kue.Job.get(id, (err, job) => {
-          if (err) throw err
+    for (const type of jobTypes) {
+      const queue = this.queues[ type ]
+      if (queue === undefined) {
+        logger.error('Unknown queue %s to count jobs.', type)
+        continue
+      }
 
-          const p = new Promise((res, rej) => {
-            job.inactive(err => {
-              if (err) return rej(err)
-              return res()
-            })
-          })
+      const counts = await queue.getJobCounts()
 
-          promises.push(p)
-        })
-      }
-    })
+      total += counts[ state ]
+    }
 
-    return Promise.all(promises)
+    return total
   }
 
-  private getJob (id: number) {
-    return new Promise<kue.Job>((res, rej) => {
-      kue.Job.get(id, (err, job) => {
-        if (err) return rej(err)
-
-        return res(job)
-      })
-    })
+  removeOldJobs () {
+    for (const key of Object.keys(this.queues)) {
+      const queue = this.queues[key]
+      queue.clean(JOB_COMPLETED_LIFETIME, 'completed')
+    }
   }
 
   static get Instance () {
diff --git a/server/lib/redis.ts b/server/lib/redis.ts
index 5bd55109c..78b28986a 100644
--- a/server/lib/redis.ts
+++ b/server/lib/redis.ts
@@ -78,16 +78,6 @@ class Redis {
     return this.setObject(this.buildCachedRouteKey(req), cached, lifetime)
   }
 
-  listJobs (jobsPrefix: string, state: string, mode: 'alpha', order: 'ASC' | 'DESC', offset: number, count: number) {
-    return new Promise<string[]>((res, rej) => {
-      this.client.sort(jobsPrefix + ':jobs:' + state, 'by', mode, order, 'LIMIT', offset.toString(), count.toString(), (err, values) => {
-        if (err) return rej(err)
-
-        return res(values)
-      })
-    })
-  }
-
   generateResetPasswordKey (userId: number) {
     return 'reset-password-' + userId
   }