refactor(jobs): create job component, remove queues when deleting HE from manuscript

packages/component-invite/package.json

@@ -26,7 +26,8 @@
   "peerDependencies": {
     "@pubsweet/component-send-email": "0.2.4",
     "@pubsweet/logger": "^0.0.1",
-    "pubsweet-server": "^10.0.0"
+    "pubsweet-server": "^10.0.0",
+    "pubsweet-component-jobs": "0.0.1"
   },
   "devDependencies": {
     "apidoc": "^0.17.6",

packages/component-invite/src/routes/collectionsInvitations/delete.js

@@ -10,6 +10,8 @@ const {
   deleteFilesS3,
 } = require('pubsweet-component-mts-package/src/PackageManager')
 
+const Job = require('pubsweet-component-jobs')
+
 const { last, get, chain, difference } = require('lodash')
 
 const s3Config = get(config, 'pubsweet-component-aws-s3', {})
@@ -132,6 +134,11 @@ module.exports = models => async (req, res) => {
         })),
       ]
 
+      fragment.invitations.forEach(inv => {
+        Job.cancelQueue(`removal-${inv.userId}-${inv.id}`)
+        Job.cancelQueue(`reminders-${inv.userId}-${inv.id}`)
+      })
+
       fragment.invitations = []
       fragment.recommendations = []
       fragment.revision && delete fragment.revision

packages/component-invite/src/routes/fragmentsInvitations/decline.js

@@ -6,7 +6,7 @@ const {
 } = require('pubsweet-component-helper-service')
 
 const notifications = require('./emails/notifications')
-const { cancelReviewerJobs } = require('./jobs/cancelReviewerJobs')
+const Job = require('pubsweet-component-jobs')
 
 module.exports = models => async (req, res) => {
   const { collectionId, invitationId, fragmentId } = req.params
@@ -68,7 +68,8 @@ module.exports = models => async (req, res) => {
       error: 'Unauthorized.',
     })
 
-  await cancelReviewerJobs(user.id)
+  await Job.cancelQueue(`reminders-${user.id}-${invitation.id}`)
+  await Job.cancelQueue(`removal-${user.id}-${invitation.id}`)
 
   invitation.respondedOn = Date.now()
   invitation.hasAnswer = true

packages/component-invite/src/routes/fragmentsInvitations/delete.js

@@ -6,7 +6,7 @@ const {
 } = require('pubsweet-component-helper-service')
 
 const notifications = require('./emails/notifications')
-const { cancelReviewerJobs } = require('./jobs/cancelReviewerJobs')
+const Job = require('pubsweet-component-jobs')
 
 module.exports = models => async (req, res) => {
   const { collectionId, invitationId, fragmentId } = req.params
@@ -80,7 +80,8 @@ module.exports = models => async (req, res) => {
       UserModel: models.User,
     })
 
-    await cancelReviewerJobs(user.id)
+    await Job.cancelQueue(`reminders-${user.id}-${invitation.id}`)
+    await Job.cancelQueue(`removal-${user.id}-${invitation.id}`)
 
     return res.status(200).json({ fragment })
   } catch (e) {

packages/component-invite/src/routes/fragmentsInvitations/emails/invitations.js

@@ -11,7 +11,7 @@ const { services, Fragment } = require('pubsweet-component-helper-service')
 
 const { getEmailCopy } = require('./emailCopy')
 const { scheduleReminderJob } = require('../jobs/reminders')
-const { scheduleRemovalJob } = require('../jobs/removals')
+const { scheduleRemovalJob } = require('../jobs/removal')
 
 const daysList = [
   config.get('reminders.reviewer.first'),
@@ -123,8 +123,9 @@ module.exports = {
       scheduleReminderJob({
         days,
         email,
-        userId: invitedUser.id,
         timeUnit,
+        userId: invitedUser.id,
+        invitationId: invitation.id,
         subject: `${subjectBaseText} reminder`,
         titleText: `the manuscript titled "${title}" by ${authorName}`,
         expectedDate: services.getExpectedDate({ timestamp, daysExpected: 0 }),

packages/component-invite/src/routes/fragmentsInvitations/jobs/cancelReviewerJobs.js

-const logger = require('@pubsweet/logger')
-
-const { jobs: { connectToJobQueue } } = require('pubsweet-server/src')
-
-const cancelReviewerJobs = async userId => {
-  const removalQueue = `reviewer-removal-${userId}`
-  const remindersQueue = `reviewer-reminders-${userId}`
-
-  const jobQueue = await connectToJobQueue()
-
-  await jobQueue.unsubscribe(remindersQueue)
-  await jobQueue.unsubscribe(removalQueue)
-
-  logger.info(`Canceled queue for reviewer ${userId}`)
-}
-
-module.exports = { cancelReviewerJobs }

packages/component-invite/src/routes/fragmentsInvitations/jobs/reminders.js

 const moment = require('moment')
 const { cloneDeep } = require('lodash')
-const logger = require('@pubsweet/logger')
+const Job = require('pubsweet-component-jobs')
 
-const { jobs: { connectToJobQueue } } = require('pubsweet-server/src')
 const { getEmailCopy } = require('../emails/emailCopy')
 const Email = require('@pubsweet/component-email-templating')
 
@@ -13,15 +12,14 @@ const scheduleReminderJob = async ({
   subject,
   timeUnit,
   titleText,
+  invitationId,
   expectedDate,
 }) => {
   const executionDate = moment()
     .add(days, timeUnit)
     .toISOString()
 
-  const queue = `reviewer-reminders-${userId}`
-
-  const jobQueue = await connectToJobQueue()
+  const queue = `reminders-${userId}-${invitationId}`
 
   const { paragraph, ...bodyProps } = getEmailCopy({
     emailType: 'reviewer-resend-invitation',
@@ -32,25 +30,14 @@ const scheduleReminderJob = async ({
   email.bodyProps = bodyProps
   email.content.subject = subject
 
-  // Add job to the queue
-  await jobQueue.publishAfter(
-    queue,
-    {
-      days,
-      timeUnit,
-      executionDate,
-      emailProps: cloneDeep(email),
-    },
-    {},
+  const params = {
+    days,
+    timeUnit,
     executionDate,
-  )
-
-  // Subscribe to the job queue with an async handler
-  await jobQueue.subscribe(queue, jobHandler)
+    emailProps: cloneDeep(email),
+  }
 
-  await jobQueue.onComplete(queue, job => {
-    logger.info(job.data.response.value)
-  })
+  await Job.schedule({ queue, params, executionDate, jobHandler })
 }
 
 const jobHandler = async job => {

packages/component-invite/src/routes/fragmentsInvitations/jobs/removals.js → packages/component-invite/src/routes/fragmentsInvitations/jobs/removal.js

 const moment = require('moment')
-const logger = require('@pubsweet/logger')
+
 const { Team, Collection } = require('pubsweet-component-helper-service')
 const {
   Team: TeamModel,
@@ -8,7 +8,7 @@ const {
   Collection: CollectionModel,
 } = require('pubsweet-server')
 
-const { jobs: { connectToJobQueue } } = require('pubsweet-server/src')
+const Job = require('pubsweet-component-jobs')
 
 const scheduleRemovalJob = async ({
   days,
@@ -22,31 +22,18 @@ const scheduleRemovalJob = async ({
     .add(days, timeUnit)
     .toISOString()
 
-  const queue = `reviewer-removal-${userId}`
-
-  const jobQueue = await connectToJobQueue()
-
-  // Add job to the queue
-  await jobQueue.publishAfter(
-    queue,
-    {
-      days,
-      timeUnit,
-      invitation,
-      fragmentId,
-      collectionId,
-      executionDate,
-    },
-    {},
-    executionDate,
-  )
+  const queue = `removal-${userId}-${invitation.id}`
 
-  // Subscribe to the job queue with an async handler
-  await jobQueue.subscribe(queue, jobHandler)
+  const params = {
+    days,
+    timeUnit,
+    invitation,
+    fragmentId,
+    collectionId,
+    executionDate,
+  }
 
-  await jobQueue.onComplete(queue, job => {
-    logger.info(job.data.response.value)
-  })
+  await Job.schedule({ queue, executionDate, jobHandler, params })
 }
 
 const jobHandler = async job => {
@@ -98,7 +85,7 @@ const jobHandler = async job => {
   await fragment.save()
 
   return `Job ${
-    job.id
+    job.name
   }: the ${days} ${timeUnit} removal has been executed at ${executionDate} for user ${
     user.id
   }`

packages/component-invite/src/routes/fragmentsInvitations/patch.js

@@ -6,7 +6,7 @@ const {
   authsome: authsomeHelper,
 } = require('pubsweet-component-helper-service')
 
-const { cancelReviewerJobs } = require('./jobs/cancelReviewerJobs')
+const Job = require('pubsweet-component-jobs')
 const notifications = require('./emails/notifications')
 
 module.exports = models => async (req, res) => {
@@ -49,7 +49,8 @@ module.exports = models => async (req, res) => {
         error: 'Unauthorized.',
       })
 
-    await cancelReviewerJobs(user.id)
+    await Job.cancelQueue(`reminders-${user.id}-${invitation.id}`)
+    await Job.cancelQueue(`removal-${user.id}-${invitation.id}`)
 
     const collectionHelper = new Collection({ collection })
     const baseUrl = services.getBaseUrl(req)

packages/component-invite/src/tests/fragmentsInvitations/decline.test.js

@@ -10,15 +10,9 @@ const cloneDeep = require('lodash/cloneDeep')
 jest.mock('@pubsweet/component-send-email', () => ({
   send: jest.fn(),
 }))
-jest.mock('pubsweet-server/src', () => ({
-  jobs: {
-    connectToJobQueue: () => ({
-      unsubscribe: jest.fn(),
-      publishAfter: jest.fn(),
-      subscribe: jest.fn(),
-      onComplete: jest.fn(),
-    }),
-  },
+jest.mock('pubsweet-component-jobs', () => ({
+  schedule: jest.fn(),
+  cancelQueue: jest.fn(),
 }))
 
 const reqBody = {

packages/component-invite/src/tests/fragmentsInvitations/delete.test.js

@@ -9,15 +9,9 @@ const { Model, fixtures } = fixturesService
 jest.mock('@pubsweet/component-send-email', () => ({
   send: jest.fn(),
 }))
-jest.mock('pubsweet-server/src', () => ({
-  jobs: {
-    connectToJobQueue: () => ({
-      unsubscribe: jest.fn(),
-      publishAfter: jest.fn(),
-      subscribe: jest.fn(),
-      onComplete: jest.fn(),
-    }),
-  },
+jest.mock('pubsweet-component-jobs', () => ({
+  schedule: jest.fn(),
+  cancelQueue: jest.fn(),
 }))
 
 const path = '../routes/fragmentsInvitations/delete'

packages/component-invite/src/tests/fragmentsInvitations/patch.test.js

@@ -10,15 +10,9 @@ jest.mock('@pubsweet/component-send-email', () => ({
   send: jest.fn(),
 }))
 
-jest.mock('pubsweet-server/src', () => ({
-  jobs: {
-    connectToJobQueue: () => ({
-      unsubscribe: jest.fn(),
-      publishAfter: jest.fn(),
-      subscribe: jest.fn(),
-      onComplete: jest.fn(),
-    }),
-  },
+jest.mock('pubsweet-component-jobs', () => ({
+  schedule: jest.fn(),
+  cancelQueue: jest.fn(),
 }))
 
 const reqBody = {

packages/component-invite/src/tests/fragmentsInvitations/post.test.js

@@ -11,15 +11,9 @@ const { Model, fixtures } = fixturesService
 jest.mock('@pubsweet/component-send-email', () => ({
   send: jest.fn(),
 }))
-jest.mock('pubsweet-server/src', () => ({
-  jobs: {
-    connectToJobQueue: () => ({
-      unsubscribe: jest.fn(),
-      publishAfter: jest.fn(),
-      subscribe: jest.fn(),
-      onComplete: jest.fn(),
-    }),
-  },
+jest.mock('pubsweet-component-jobs', () => ({
+  schedule: jest.fn(),
+  cancelQueue: jest.fn(),
 }))
 
 const chance = new Chance()

packages/component-jobs/index.js

+module.exports = require('./src/Job')

packages/component-jobs/package.json

+{
+  "name": "pubsweet-component-jobs",
+  "version": "0.0.1",
+  "description": "publishing and scheduling jobs",
+  "license": "MIT",
+  "author": "Collaborative Knowledge Foundation",
+  "files": [
+    "src"
+  ],
+  "main": "index.js",
+  "dependencies": {
+  },
+  "peerDependencies": {
+  },
+  "jest": {
+    "verbose": true,
+    "testRegex": "/tests/.*.test.js$"
+  },
+  "scripts": {
+    "test": "jest"
+  },
+  "publishConfig": {
+    "access": "public"
+  }
+}

packages/component-jobs/src/Job.js

+const { jobs: { connectToJobQueue } } = require('pubsweet-server/src')
+const logger = require('@pubsweet/logger')
+
+module.exports = {
+  schedule: async ({ queue, jobHandler, executionDate, params }) => {
+    const jobQueue = await connectToJobQueue()
+
+    // Add job to the queue
+    await jobQueue.publishAfter(queue, params, {}, executionDate)
+
+    // Subscribe to the job queue with an async handler
+    await jobQueue.subscribe(queue, jobHandler)
+
+    await jobQueue.onComplete(queue, job => {
+      logger.info(job.data.response.value)
+    })
+  },
+  cancelQueue: async name => {
+    const jobQueue = await connectToJobQueue()
+
+    await jobQueue.unsubscribe(name)
+    logger.info(`Successfully unsubscribed from queue: ${name}`)
+  },
+}