server/initializers/migrations/0230-kue-to-bull.ts

   1 import * as Sequelize from 'sequelize'
   2 import { createClient } from 'redis'
   3 import { CONFIG } from '../constants'
   4 import { JobQueue } from '../../lib/job-queue'
   5 import { Redis } from '../../lib/redis'
   6 import { initDatabaseModels } from '../database'
   7
   8 async function up (utils: {
   9   transaction: Sequelize.Transaction
  10   queryInterface: Sequelize.QueryInterface
  11   sequelize: Sequelize.Sequelize
  12 }): Promise<any> {
  13   await initDatabaseModels(false)
  14
  15   return new Promise((res, rej) => {
  16     const client = createClient(Redis.getRedisClient())
  17
  18     const jobsPrefix = 'q-' + CONFIG.WEBSERVER.HOST
  19
  20     client.sort(jobsPrefix + ':jobs:inactive', 'by', 'alpha', 'ASC', (err, jobStrings) => {
  21       if (err) return rej(err)
  22
  23       const jobPromises = jobStrings
  24         .map(s => s.split('|'))
  25         .map(([ , jobId ]) => {
  26           return new Promise((res, rej) => {
  27             client.hgetall(jobsPrefix + ':job:' + jobId, (err, job) => {
  28               if (err) return rej(err)
  29
  30               try {
  31                 const parsedData = JSON.parse(job.data)
  32
  33                 return res({ type: job.type, payload: parsedData })
  34               } catch (err) {
  35                 console.error('Cannot parse data %s.', job.data)
  36                 return res(undefined)
  37               }
  38             })
  39           })
  40         })
  41
  42       JobQueue.Instance.init()
  43               .then(() => Promise.all(jobPromises))
  44               .then((jobs: any) => {
  45                 const createJobPromises = jobs
  46                   .filter(job => job !== null)
  47                   .map(job => JobQueue.Instance.createJob(job))
  48
  49                 return Promise.all(createJobPromises)
  50               })
  51               .then(() => res())
  52     })
  53   })
  54 }
  55
  56 function down (options) {
  57   throw new Error('Not implemented.')
  58 }
  59
  60 export { up, down }