server/lib/job-queue/job-queue.ts

   1 import * as kue from 'kue'
   2 import { JobState, JobType } from '../../../shared/models'
   3 import { logger } from '../../helpers/logger'
   4 import { CONFIG, JOB_ATTEMPTS, JOB_COMPLETED_LIFETIME, JOB_CONCURRENCY, JOB_REQUEST_TTL } from '../../initializers'
   5 import { Redis } from '../redis'
   6 import { ActivitypubHttpBroadcastPayload, processActivityPubHttpBroadcast } from './handlers/activitypub-http-broadcast'
   7 import { ActivitypubHttpFetcherPayload, processActivityPubHttpFetcher } from './handlers/activitypub-http-fetcher'
   8 import { ActivitypubHttpUnicastPayload, processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast'
   9 import { EmailPayload, processEmail } from './handlers/email'
  10 import { processVideoFile, processVideoFileImport, VideoFilePayload, VideoFileImportPayload } from './handlers/video-file'
  11 import { ActivitypubFollowPayload, processActivityPubFollow } from './handlers/activitypub-follow'
  12
  13 type CreateJobArgument =
  14   { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } |
  15   { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } |
  16   { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } |
  17   { type: 'activitypub-follow', payload: ActivitypubFollowPayload } |
  18   { type: 'video-file-import', payload: VideoFileImportPayload } |
  19   { type: 'video-file', payload: VideoFilePayload } |
  20   { type: 'email', payload: EmailPayload }
  21
  22 const handlers: { [ id in JobType ]: (job: kue.Job) => Promise<any>} = {
  23   'activitypub-http-broadcast': processActivityPubHttpBroadcast,
  24   'activitypub-http-unicast': processActivityPubHttpUnicast,
  25   'activitypub-http-fetcher': processActivityPubHttpFetcher,
  26   'activitypub-follow': processActivityPubFollow,
  27   'video-file-import': processVideoFileImport,
  28   'video-file': processVideoFile,
  29   'email': processEmail
  30 }
  31
  32 const jobsWithTLL: JobType[] = [
  33   'activitypub-http-broadcast',
  34   'activitypub-http-unicast',
  35   'activitypub-http-fetcher',
  36   'activitypub-follow'
  37 ]
  38
  39 class JobQueue {
  40
  41   private static instance: JobQueue
  42
  43   private jobQueue: kue.Queue
  44   private initialized = false
  45   private jobRedisPrefix: string
  46
  47   private constructor () {}
  48
  49   async init () {
  50     // Already initialized
  51     if (this.initialized === true) return
  52     this.initialized = true
  53
  54     this.jobRedisPrefix = 'q-' + CONFIG.WEBSERVER.HOST
  55
  56     this.jobQueue = kue.createQueue({
  57       prefix: this.jobRedisPrefix,
  58       redis: {
  59         host: CONFIG.REDIS.HOSTNAME,
  60         port: CONFIG.REDIS.PORT,
  61         auth: CONFIG.REDIS.AUTH,
  62         db: CONFIG.REDIS.DB
  63       }
  64     })
  65
  66     this.jobQueue.setMaxListeners(20)
  67
  68     this.jobQueue.on('error', err => {
  69       logger.error('Error in job queue.', { err })
  70       process.exit(-1)
  71     })
  72     this.jobQueue.watchStuckJobs(5000)
  73
  74     await this.reactiveStuckJobs()
  75
  76     for (const handlerName of Object.keys(handlers)) {
  77       this.jobQueue.process(handlerName, JOB_CONCURRENCY[handlerName], async (job, done) => {
  78         try {
  79           const res = await handlers[ handlerName ](job)
  80           return done(null, res)
  81         } catch (err) {
  82           logger.error('Cannot execute job %d.', job.id, { err })
  83           return done(err)
  84         }
  85       })
  86     }
  87   }
  88
  89   createJob (obj: CreateJobArgument, priority = 'normal') {
  90     return new Promise((res, rej) => {
  91       let job = this.jobQueue
  92         .create(obj.type, obj.payload)
  93         .priority(priority)
  94         .attempts(JOB_ATTEMPTS[obj.type])
  95         .backoff({ delay: 60 * 1000, type: 'exponential' })
  96
  97       if (jobsWithTLL.indexOf(obj.type) !== -1) {
  98         job = job.ttl(JOB_REQUEST_TTL)
  99       }
 100
 101       return job.save(err => {
 102         if (err) return rej(err)
 103
 104         return res()
 105       })
 106     })
 107   }
 108
 109   async listForApi (state: JobState, start: number, count: number, sort: 'ASC' | 'DESC'): Promise<kue.Job[]> {
 110     const jobStrings = await Redis.Instance.listJobs(this.jobRedisPrefix, state, 'alpha', sort, start, count)
 111
 112     const jobPromises = jobStrings
 113       .map(s => s.split('|'))
 114       .map(([ , jobId ]) => this.getJob(parseInt(jobId, 10)))
 115
 116     return Promise.all(jobPromises)
 117   }
 118
 119   count (state: JobState) {
 120     return new Promise<number>((res, rej) => {
 121       this.jobQueue[state + 'Count']((err, total) => {
 122         if (err) return rej(err)
 123
 124         return res(total)
 125       })
 126     })
 127   }
 128
 129   removeOldJobs () {
 130     const now = new Date().getTime()
 131     kue.Job.rangeByState('complete', 0, -1, 'asc', (err, jobs) => {
 132       if (err) {
 133         logger.error('Cannot get jobs when removing old jobs.', { err })
 134         return
 135       }
 136
 137       for (const job of jobs) {
 138         if (now - job.created_at > JOB_COMPLETED_LIFETIME) {
 139           job.remove()
 140         }
 141       }
 142     })
 143   }
 144
 145   private reactiveStuckJobs () {
 146     const promises: Promise<any>[] = []
 147
 148     this.jobQueue.active((err, ids) => {
 149       if (err) throw err
 150
 151       for (const id of ids) {
 152         kue.Job.get(id, (err, job) => {
 153           if (err) throw err
 154
 155           const p = new Promise((res, rej) => {
 156             job.inactive(err => {
 157               if (err) return rej(err)
 158               return res()
 159             })
 160           })
 161
 162           promises.push(p)
 163         })
 164       }
 165     })
 166
 167     return Promise.all(promises)
 168   }
 169
 170   private getJob (id: number) {
 171     return new Promise<kue.Job>((res, rej) => {
 172       kue.Job.get(id, (err, job) => {
 173         if (err) return rej(err)
 174
 175         return res(job)
 176       })
 177     })
 178   }
 179
 180   static get Instance () {
 181     return this.instance || (this.instance = new this())
 182   }
 183 }
 184
 185 // ---------------------------------------------------------------------------
 186
 187 export {
 188   JobQueue
 189 }