[Updater] Use BullMQ to schedule student updates

updater/sis-updater-scheduler/src/scheduler.js

@@ -6,7 +6,6 @@ const { chunk } = require('lodash')
 const {
   NATS_GROUP,
   SIS_UPDATER_SCHEDULE_CHANNEL,
-  SIS_MISC_SCHEDULE_CHANNEL,
   CHUNK_SIZE,
   isDev,
   DEV_SCHEDULE_COUNT,
@@ -19,6 +18,7 @@ const {
 } = require('./config')
 const { knexConnection } = require('./db/connection')
 const { startPrePurge, startPurge } = require('./purge')
+const { queue } = require('./queue')
 const { logger } = require('./utils/logger')
 const { redisClient } = require('./utils/redis')
 const { stan, opts } = require('./utils/stan')
@@ -213,12 +213,8 @@ const scheduleByStudentNumbers = async studentNumbers => {
   logger.info('Scheduling by student numbers')
   const { knex } = knexConnection
   const personsToUpdate = await knex('persons').column('id', 'student_number').whereIn('student_number', studentNumbers)
-
-  await eachLimit(
-    chunk(personsToUpdate, CHUNK_SIZE),
-    10,
-    async s => await createJobs(s, 'students', SIS_MISC_SCHEDULE_CHANNEL)
-  )
+  const personChunks = chunk(personsToUpdate, CHUNK_SIZE)
+  await queue.addBulk(personChunks.map(personsToUpdate => ({ name: 'students', data: personsToUpdate })))
 }
 
 const scheduleByCourseCodes = async courseCodes => {

updater/sis-updater-worker/src/index.js

@@ -1,7 +1,6 @@
 const {
   SIS_UPDATER_SCHEDULE_CHANNEL,
   SIS_PURGE_CHANNEL,
-  SIS_MISC_SCHEDULE_CHANNEL,
   NATS_GROUP,
   REDIS_TOTAL_META_KEY,
   REDIS_TOTAL_STUDENTS_KEY,
@@ -12,7 +11,7 @@ const {
 const { dbConnections } = require('./db/connection')
 const { postUpdate } = require('./postUpdate')
 const { update } = require('./updater')
-const { purge, prePurge, purgeByStudentNumber } = require('./updater/purge')
+const { purge, prePurge } = require('./updater/purge')
 const { loadMapsOnDemand } = require('./updater/shared')
 const { logger } = require('./utils/logger')
 const { redisClient } = require('./utils/redis')
@@ -111,13 +110,6 @@ const purgeMsgHandler = async purgeMsg => {
   if (purgeMsg.action === 'PREPURGE_START') await prePurge(purgeMsg)
 }
 
-const miscMsgHandler = async miscMessage => {
-  const studentNumbers = miscMessage.entityIds.map(s => s.student_number)
-  const msgInUpdateFormat = { ...miscMessage, entityIds: miscMessage.entityIds.map(s => s.id) }
-  await purgeByStudentNumber(studentNumbers)
-  await updateMsgHandler(msgInUpdateFormat)
-}
-
 stan.on('error', error => {
   logger.error({ message: 'NATS connection failed', meta: error })
   if (!process.env.CI) process.exit(1)
@@ -150,10 +142,4 @@ dbConnections.on('connect', async () => {
 
   const infoChannel = stan.subscribe('SIS_INFO_CHANNEL', NATS_GROUP, opts)
   infoChannel.on('message', handleMessage(handleInfoMessage))
-
-  const miscChannel = stan.subscribe(SIS_MISC_SCHEDULE_CHANNEL, NATS_GROUP, opts)
-  miscChannel.on('message', handleMessage(miscMsgHandler))
-  miscChannel.on('error', error => {
-    logger.error({ message: 'Misc channel error', meta: error.stack })
-  })
 })

updater/sis-updater-worker/src/processor.js

@@ -1 +1,24 @@
-module.exports = async job => {}
+const { postUpdate } = require('./postUpdate')
+const { update } = require('./updater')
+const { purgeByStudentNumber } = require('./updater/purge')
+
+const updateMsgHandler = async updateMsg => {
+  // TODO: Remove the following line after postUpdate is implemented correctly (worker.js calculates the processing time more accurately)
+  const startTime = new Date()
+  await update(updateMsg)
+  await postUpdate(updateMsg, startTime)
+}
+
+module.exports = async job => {
+  switch (job.name) {
+    case 'students': {
+      const studentNumbers = job.data.map(student => student.student_number)
+      const msgInUpdateFormat = { type: job.name, entityIds: job.data.map(student => student.id) }
+      await purgeByStudentNumber(studentNumbers)
+      await updateMsgHandler(msgInUpdateFormat)
+      break
+    }
+    default:
+      throw new Error(`Unknown job type: ${job.name}`)
+  }
+}