@iin-mdc/koa-utils/initializers/bg-processing-initializer.js

"use strict"

const EVERY_30_MINUTES = 60 * 1000 * 30

class BgProcessingInitializer {
  constructor(logger, applicationContext, Queue, createRedisClient) {
    this.logger             = logger
    this.jobDefinitions     = {}
    this.jobsCallback       = async () => {}
    this.applicationContext = applicationContext
    this.Queue              = Queue
    this.createRedisClient  = createRedisClient
    this.queues             = {}
    this.redisClients       = {
      __default: []
    }
    this.cleanupTimeout     = null
  }

  async start() {
    await this._initializeJobs()
    await this._initializeProcessors()
    this._startCleanup()
  }

  async stop() {
    this._stopCleanup()
    await this._closeQueues()
    await this._disconnectRedisClients()
  }

  async initializeQueues(jobDefinitions) {
    this.jobDefinitions = jobDefinitions

    let Queue = this.Queue
    let jobNames = Object.keys(this.jobDefinitions)
    for (let name of jobNames) {
      this.queues[name] = new Queue(name, this._queueOptions)
    }
  }

  async initializeJobs(jobsCallback) {
    this.jobsCallback = jobsCallback
  }

  async runCleanup() {
    await Promise.all(
      Object.values(this.queues).map((queue) => {
        if (queue.clean) {
          return queue.clean(100).catch(e => this.logger.error(e))
        }

        return Promise.reslove()
      })
    )
  }

  async fetchStatus() {
    let statuses = await Promise.all(
      Object.values(this.queues).map((queue) => {
        return queue.getJobCounts().then((counts) => [queue.name, counts])
      })
    )

    return statuses.reduce((acc, status) => {
      acc[status[0]] = status[1]
      return acc
    }, {})
  }

  async _initializeJobs() {
    await this.jobsCallback.call(this, this.queues)
  }

  async _initializeProcessors() {
    let jobNames = Object.keys(this.jobDefinitions)
    for (let name of jobNames) {
      let queue       = this.queues[name]
      let processor   = this.jobDefinitions[name]
      let concurrency = 1
      if (Array.isArray(processor)) {
        concurrency = processor[0]
        processor   = processor[1]
      }
      this._assignEventHandlers(queue)
      this._assignProcessor(queue, processor, concurrency)
    }
  }

  _assignProcessor(queue, process, concurency) {
    queue.process("*", concurency, async (job) => await process(this.applicationContext, job))
  }

  _prettyPrintJob(job) {
    return `Job[${job.id}] { ${Object.keys(job.data).map(k => `${k}: ${job.data[k]}`).join(", ")} }`
  }

  _assignEventHandlers(queue) {
    let logger = this.logger.child(queue.name)
    queue
      .on("error",   (error)      => logger.error("Queue error", error))
      .on("paused",  ()           => logger.info("Queue paused"))
      .on("resumed", ()           => logger.info("Queue resumed"))
      .on("drained", ()           => logger.debug("Queue drained"))
      .on("cleaned", (jobIds, type) => jobIds.length && logger.debug(`Jobs(${type}) cleaned: ${jobIds}`))
      //waiting is internal event and cause warns
      //.on("waiting",   (jobId)         => logger.debug(`Job[${jobId}] waiting`))
      .on("active",    (job)           => logger.debug(`${this._prettyPrintJob(job)} started`))
      .on("stalled",   (job)           => logger.error(`${this._prettyPrintJob(job)} stalled`))
      .on("progress",  (job, progress) => logger.debug(`${this._prettyPrintJob(job)} progress ${progress * 100}`))
      .on("completed", (job)           => logger.info( `${this._prettyPrintJob(job)} completed`))
      .on("failed",    (job, err)      => logger.error(`${this._prettyPrintJob(job)} failed`, err))
      .on("removed",   (job)           => logger.debug(`${this._prettyPrintJob(job)} removed`))
  }

  async _closeQueues() {
    await Promise.all(
      Object.values(this.queues).map((queue) => queue.close())
    )
  }

  async _disconnectRedisClients() {
    let disconnects = []


    if (this.redisClients.client)     disconnects.push(this._disconnectRedisClient(this.redisClients.client))
    if (this.redisClients.subscriber) disconnects.push(this._disconnectRedisClient(this.redisClients.subscriber))
    this.redisClients.__default.forEach((client) => {
      disconnects.push( this._disconnectRedisClient(client) )
    })

    await Promise.all(disconnects)
  }

  _disconnectRedisClient(client) {
    return client
      .quit()
      .catch((e) => this.logger.debug(e))
  }

  _startCleanup() {
    this.cleanupTimeout = setTimeout(async () => {
      await this.runCleanup()
      this._startCleanup()
    }, EVERY_30_MINUTES)
  }

  _stopCleanup() {
    clearTimeout(this.cleanupTimeout)
  }

  get _queueOptions() {
    return {
      createClient: this._createClient.bind(this)
    }
  }

  _createClient(type) {
    if (type === "client" || type === "subscriber") {
      if (!this.redisClients[type]) {
        this.redisClients[type] = this.createRedisClient()
      }
      return this.redisClients[type]
    } else {
      let client = this.createRedisClient()
      this.redisClients.__default.push(client)
      return client
    }
  }
}

module.exports = BgProcessingInitializer