dtp-base-v1/app/workers/newsletter.js

// newsletter.js
// Copyright (C) 2022 DTP Technologies, LLC
// License: Apache-2.0

'use strict';

const path = require('path');
require('dotenv').config({ path: path.resolve(__dirname, '..', '..', '.env') });

const mongoose = require('mongoose');

const { SiteWorker, SiteLog } = require(path.join(__dirname, '..', '..', 'lib', 'site-lib'));

module.pkg = require(path.resolve(__dirname, '..', '..', 'package.json'));
module.config = {
  component: { name: 'Newsletter Worker', slug: 'newsletter' },
  root: path.resolve(__dirname, '..', '..'),
};

class NewsletterWorker extends SiteWorker {

  constructor (dtp) {
    super(dtp, dtp.config.component);
    this.newsletters = this.newsletters || { };
  }

  async start ( ) {
    await super.start();
    
    const { jobQueue: jobQueueService } = this.dtp.services;
    this.jobQueue = await jobQueueService.getJobQueue('newsletter', {
      attempts: 3,
    });
    this.jobQueue.process('transmit', this.transmitNewsletter.bind(this));
    this.jobQueue.process('email-send', this.sendNewsletterEmail.bind(this));
  }

  async stop ( ) {
    if (this.jobQueue) {
      this.log.info('stopping newsletter job queue');
      await this.jobQueue.close();
      delete this.jobQueue;
    }
    await super.stop();
  }

  async loadNewsletter (newsletterId) {
    const { newsletter: newsletterService } = this.dtp.services;
    let newsletter = this.newsletters[newsletterId];
    if (!newsletter) {
      newsletter = await newsletterService.getById(newsletterId);
      this.newsletters[newsletterId] = newsletter;
    }
    return newsletter;
  }

  async transmitNewsletter (job) {
    const User = mongoose.model('User');
    const NewsletterRecipient = mongoose.model('NewsletterRecipient');
    this.log.info('newsletter email job received', { data: job.data });
    try {
      /*
       * Transmit first to all local user accounts with verified email who've
       * opted in for receiving marketing email.
       */
      await User
        .find({
          'flags.isEmailVerified': true,
          'optIn.marketing': true,
        })
        .select('email displayName username username_lc')
        .lean()
        .cursor()
        .eachAsync(async (user) => {
          try {
            const jobData = {
              newsletterId: job.data.newsletterId,
              recipient: user.email,
              recipientName: user.displayName || user.username,
            };
            const jobOptions = { attempts: 3 };
            await this.jobQueue.add('email-send', jobData, jobOptions);
          } catch (error) {
            this.log.error('failed to create newsletter email job', { error });
          }
        }, { parallel: 4 });

      /*
       * Transmit to all newsletter recipients on file who've joined through the
       * widget on the site w/o signing up for an account.
       */
      await NewsletterRecipient
        .find({ 'flags.isVerified': true, 'flags.isOptIn': true, 'flags.isRejected': false })
        .lean()
        .cursor()
        .eachAsync(async (recipient) => {
          try {
            const jobData = {
              newsletterId: job.data.newsletterId,
              recipient: recipient.address,
            };
            const jobOptions = { attempts: 3 };
            await this.jobQueue.add('email-send', jobData, jobOptions);
          } catch (error) {
            this.log.error('failed to create newsletter email job', { error });
          }
        }, { parallel: 4 });
    } catch (error) {
      this.log.error('failed to send newsletter', { newsletterId: job.data.newsletterId, error });
      throw error;
    }
  }

  async sendNewsletterEmail (job) {
    const { email: emailService } = this.dtp.services;
    const { newsletterId, recipient } = job.data;

    try {
      let newsletter = await this.loadNewsletter(newsletterId);
      if (!newsletter) {
        throw new Error('newsletter not found');
      }
  
      const result = await emailService.send({
        from: process.env.DTP_EMAIL_SMTP_FROM || `noreply@${this.dtp.config.site.domainKey}`,
        to: recipient,
        subject: newsletter.title,
        html: newsletter.content.html,
        text: newsletter.content.text,
      });
  
      job.log(`newsletter email sent: ${result}`);
      this.log.info('newsletter email sent', { recipient, result });
    } catch (error) {
      this.log.error('failed to send newsletter email', { newsletterId, recipient, error });
      throw error; // throw error to Bull so it can report in job reports
    }
  }
}  

(async ( ) => {
  try {
    module.log = new SiteLog(module, module.config.component);

    module.worker = new NewsletterWorker(module);
    await module.worker.start();

    module.log.info(`${module.pkg.name} v${module.pkg.version} ${module.config.component.name} started`);
  } catch (error) {
    module.log.error('failed to start Newsletter worker', { error });
    process.exit(-1);
  }
})();