wikijs/server/core/queue.js

const path = require('path')
const Bull = require('bull')
const Promise = require('bluebird')
const _ = require('lodash')

/* global WIKI */

module.exports = {
  job: {},
  init() {
    _.forOwn(WIKI.data.jobs, (queueParams, queueName) => {
      this.job[queueName] = new Bull(queueName, {
        prefix: `queue`,
        redis: WIKI.config.redis
      })
      if (queueParams.concurrency > 0) {
        this.job[queueName].process(queueParams.concurrency, path.join(WIKI.SERVERPATH, `jobs/${_.kebabCase(queueName)}.js`))
      } else {
        this.job[queueName].process(path.join(WIKI.SERVERPATH, `jobs/${_.kebabCase(queueName)}.js`))
      }
    })
    return this
  },
  start() {
    _.forOwn(WIKI.data.jobs, (queueParams, queueName) => {
      if (queueParams.onInit) {
        this.job[queueName].add({}, {
          removeOnComplete: true
        })
      }
      if (queueParams.cron) {
        this.job[queueName].add({}, {
          repeat: { cron: queueParams.cron },
          removeOnComplete: true
        })
      }
    })
  },
  async quit() {
    for (const queueName in this.job) {
      await this.job[queueName].close()
    }
  },
  async clean() {
    return Promise.each(_.keys(WIKI.data.jobs), queueName => {
      return new Promise((resolve, reject) => {
        let keyStream = WIKI.redis.scanStream({
          match: `queue:${queueName}:*`
        })
        keyStream.on('data', resultKeys => {
          if (resultKeys.length > 0) {
            WIKI.redis.del(resultKeys)
          }
        })
        keyStream.on('end', resolve)
      })
    }).then(() => {
      WIKI.logger.info('Purging old queue jobs: [ OK ]')
    }).return(true).catch(err => {
      WIKI.logger.error(err)
    })
  }
}