overleaf-cep/services/web/app/src/infrastructure/QueueWorkers.js

const Features = require('./Features')
const Queues = require('./Queues')
const UserOnboardingEmailManager = require('../Features/User/UserOnboardingEmailManager')
const UserPostRegistrationAnalyticsManager = require('../Features/User/UserPostRegistrationAnalyticsManager')
const FeaturesUpdater = require('../Features/Subscription/FeaturesUpdater')
const InstitutionsManager = require('../Features/Institutions/InstitutionsManager')
const {
  addOptionalCleanupHandlerBeforeStoppingTraffic,
  addRequiredCleanupHandlerBeforeDrainingConnections,
} = require('./GracefulShutdown')
const EmailHandler = require('../Features/Email/EmailHandler')
const logger = require('@overleaf/logger')
const OError = require('@overleaf/o-error')
const Modules = require('./Modules')

function start() {
  if (!Features.hasFeature('saas')) {
    return
  }

  const scheduledJobsQueue = Queues.getQueue('scheduled-jobs')
  scheduledJobsQueue.process(async job => {
    const { queueName, name, data, options } = job.data
    const queue = Queues.getQueue(queueName)
    if (name) {
      await queue.add(name, data || {}, options || {})
    } else {
      await queue.add(data || {}, options || {})
    }
  })
  registerCleanup(scheduledJobsQueue)

  const onboardingEmailsQueue = Queues.getQueue('emails-onboarding')
  onboardingEmailsQueue.process(async job => {
    const { userId } = job.data
    await UserOnboardingEmailManager.sendOnboardingEmail(userId)
  })
  registerCleanup(onboardingEmailsQueue)

  const postRegistrationAnalyticsQueue = Queues.getQueue(
    'post-registration-analytics'
  )
  postRegistrationAnalyticsQueue.process(async job => {
    const { userId } = job.data
    await UserPostRegistrationAnalyticsManager.postRegistrationAnalytics(userId)
  })
  registerCleanup(postRegistrationAnalyticsQueue)

  const refreshFeaturesQueue = Queues.getQueue('refresh-features')
  refreshFeaturesQueue.process(async job => {
    const { userId, reason } = job.data
    await FeaturesUpdater.promises.refreshFeatures(userId, reason)
  })
  registerCleanup(refreshFeaturesQueue)

  const deferredEmailsQueue = Queues.getQueue('deferred-emails')
  deferredEmailsQueue.process(async job => {
    const { emailType, opts } = job.data
    try {
      await EmailHandler.promises.sendEmail(emailType, opts)
    } catch (e) {
      const error = OError.tag(e, 'failed to send deferred email')
      logger.warn(error)
      throw error
    }
  })
  registerCleanup(deferredEmailsQueue)

  const confirmInstitutionDomainQueue = Queues.getQueue(
    'confirm-institution-domain'
  )
  confirmInstitutionDomainQueue.process(async job => {
    const { hostname } = job.data
    try {
      await InstitutionsManager.promises.affiliateUsers(hostname)
    } catch (e) {
      const error = OError.tag(e, 'failed to confirm university domain')
      logger.warn(error)
      throw error
    }
  })
  registerCleanup(confirmInstitutionDomainQueue)

  const groupSSOReminderQueue = Queues.getQueue('group-sso-reminder')
  groupSSOReminderQueue.process(async job => {
    const { userId, subscriptionId } = job.data
    try {
      await Modules.promises.hooks.fire(
        'sendGroupSSOReminder',
        userId,
        subscriptionId
      )
    } catch (e) {
      const error = OError.tag(
        e,
        'failed to send scheduled Group SSO account linking reminder'
      )
      logger.warn(error)
      throw error
    }
  })
  registerCleanup(groupSSOReminderQueue)
}

function registerCleanup(queue) {
  const label = `bull queue ${queue.name}`

  // Stop accepting new jobs.
  addOptionalCleanupHandlerBeforeStoppingTraffic(label, async () => {
    const justThisWorker = true
    await queue.pause(justThisWorker)
  })

  // Wait for all jobs to process before shutting down connections.
  addRequiredCleanupHandlerBeforeDrainingConnections(label, async () => {
    await queue.close()
  })

  // Disconnect from redis is scheduled in queue setup.
}

module.exports = { start }