feat(email-plugin): Generate and send emails on the worker

packages/email-plugin/src/common.ts

@@ -0,0 +1,7 @@
+import { EmailPluginDevModeOptions, EmailPluginOptions } from './types';
+
+export function isDevModeOptions(
+    input: EmailPluginOptions | EmailPluginDevModeOptions,
+): input is EmailPluginDevModeOptions {
+    return (input as EmailPluginDevModeOptions).devMode === true;
+}

packages/email-plugin/src/constants.ts

@@ -0,0 +1 @@
+export const EMAIL_PLUGIN_OPTIONS = Symbol('EMAIL_PLUGIN_OPTIONS');

packages/email-plugin/src/email-worker.controller.ts

@@ -0,0 +1,67 @@
+import { Controller, Inject, OnModuleInit } from '@nestjs/common';
+import { MessagePattern } from '@nestjs/microservices';
+import { asyncObservable, InternalServerError } from '@vendure/core';
+import fs from 'fs-extra';
+import { Observable } from 'rxjs';
+
+import { isDevModeOptions } from './common';
+import { EMAIL_PLUGIN_OPTIONS } from './constants';
+import { EmailSender } from './email-sender';
+import { HandlebarsMjmlGenerator } from './handlebars-mjml-generator';
+import { TemplateLoader } from './template-loader';
+import { EmailPluginOptions, EmailTransportOptions, EmailWorkerMessage } from './types';
+
+@Controller()
+export class EmailWorkerController implements OnModuleInit {
+    private templateLoader: TemplateLoader;
+    private emailSender: EmailSender;
+    private generator: HandlebarsMjmlGenerator;
+    private transport: EmailTransportOptions;
+
+    constructor(@Inject(EMAIL_PLUGIN_OPTIONS) private options: EmailPluginOptions) {}
+
+    async onModuleInit() {
+        this.templateLoader = new TemplateLoader(this.options.templatePath);
+        this.emailSender = new EmailSender();
+        this.generator = new HandlebarsMjmlGenerator();
+        if (this.generator.onInit) {
+            await this.generator.onInit.call(this.generator, this.options);
+        }
+        if (isDevModeOptions(this.options)) {
+            this.transport = {
+                type: 'file',
+                raw: false,
+                outputPath: this.options.outputPath,
+            };
+        } else {
+            if (!this.options.transport) {
+                throw new InternalServerError(
+                    `When devMode is not set to true, the 'transport' property must be set.`,
+                );
+            }
+            this.transport = this.options.transport;
+        }
+        if (this.transport.type === 'file') {
+            // ensure the configured directory exists before
+            // we attempt to write files to it
+            const emailPath = this.transport.outputPath;
+            await fs.ensureDir(emailPath);
+        }
+    }
+
+    @MessagePattern(EmailWorkerMessage.pattern)
+    sendEmail(data: EmailWorkerMessage['data']): Observable<EmailWorkerMessage['response']> {
+        return asyncObservable(async () => {
+            const bodySource = await this.templateLoader.loadTemplate(data.type, data.templateFile);
+            const generated = await this.generator.generate(
+                data.from,
+                data.subject,
+                bodySource,
+                data.templateVars,
+            );
+            const emailDetails = { ...generated, recipient: data.recipient };
+            await this.emailSender.send(emailDetails, this.transport);
+            return true;
+        });
+    }
+}

packages/email-plugin/src/event-handler.ts

@@ -2,7 +2,7 @@ import { LanguageCode } from '@vendure/common/lib/generated-types';
 import { Type } from '@vendure/common/lib/shared-types';
 
 import { EmailEventListener, EmailTemplateConfig, SetTemplateVarsFn } from './event-listener';
-import { EventWithAsyncData, EventWithContext, LoadDataFn } from './types';
+import { EventWithAsyncData, EventWithContext, IntermediateEmailDetails, LoadDataFn } from './types';
 
 /**
  * @description
@@ -169,10 +169,7 @@ export class EmailEventHandler<T extends string = string, Event extends EventWit
     async handle(
         event: Event,
         globals: { [key: string]: any } = {},
-    ): Promise<
-        | { from: string; recipient: string; templateVars: any; subject: string; templateFile: string }
-        | undefined
-    > {
+    ): Promise<IntermediateEmailDetails | undefined> {
         for (const filterFn of this.filterFns) {
             if (!filterFn(event)) {
                 return;
@@ -202,6 +199,7 @@ export class EmailEventHandler<T extends string = string, Event extends EventWit
         const recipient = this.setRecipientFn(event);
         const templateVars = this.setTemplateVarsFn ? this.setTemplateVarsFn(event, globals) : {};
         return {
+            type: this.type,
             recipient,
             from: this.from,
             templateVars: { ...globals, ...templateVars },
@@ -227,7 +225,7 @@ export class EmailEventHandler<T extends string = string, Event extends EventWit
         if (this.configurations.length === 0) {
             return;
         }
-        const exactMatch = this.configurations.find(c => {
+        const exactMatch = this.configurations.find((c) => {
             return (
                 (c.channelCode === channelCode || c.channelCode === 'default') &&
                 c.languageCode === languageCode
@@ -237,7 +235,7 @@ export class EmailEventHandler<T extends string = string, Event extends EventWit
             return exactMatch;
         }
         const channelMatch = this.configurations.find(
-            c => c.channelCode === channelCode && c.languageCode === 'default',
+            (c) => c.channelCode === channelCode && c.languageCode === 'default',
         );
         if (channelMatch) {
             return channelMatch;

packages/email-plugin/src/plugin.ts

@@ -3,7 +3,8 @@ import { InjectConnection } from '@nestjs/typeorm';
 import {
     createProxyHandler,
     EventBus,
-    InternalServerError,
+    JobQueue,
+    JobQueueService,
     Logger,
     OnVendureBootstrap,
     OnVendureClose,
@@ -11,21 +12,22 @@ import {
     RuntimeVendureConfig,
     Type,
     VendurePlugin,
+    WorkerService,
 } from '@vendure/core';
-import fs from 'fs-extra';
 import { Connection } from 'typeorm';
 
+import { isDevModeOptions } from './common';
+import { EMAIL_PLUGIN_OPTIONS } from './constants';
 import { DevMailbox } from './dev-mailbox';
-import { EmailSender } from './email-sender';
+import { EmailWorkerController } from './email-worker.controller';
 import { EmailEventHandler, EmailEventHandlerWithAsyncData } from './event-handler';
-import { HandlebarsMjmlGenerator } from './handlebars-mjml-generator';
-import { TemplateLoader } from './template-loader';
 import {
     EmailPluginDevModeOptions,
     EmailPluginOptions,
-    EmailTransportOptions,
+    EmailWorkerMessage,
     EventWithAsyncData,
     EventWithContext,
+    IntermediateEmailDetails,
 } from './types';
 
 /**
@@ -138,21 +140,22 @@ import {
  */
 @VendurePlugin({
     imports: [PluginCommonModule],
-    configuration: config => EmailPlugin.configure(config),
+    providers: [{ provide: EMAIL_PLUGIN_OPTIONS, useFactory: () => EmailPlugin.options }],
+    workers: [EmailWorkerController],
+    configuration: (config) => EmailPlugin.configure(config),
 })
 export class EmailPlugin implements OnVendureBootstrap, OnVendureClose {
     private static options: EmailPluginOptions | EmailPluginDevModeOptions;
-    private transport: EmailTransportOptions;
-    private templateLoader: TemplateLoader;
-    private emailSender: EmailSender;
-    private generator: HandlebarsMjmlGenerator;
     private devMailbox: DevMailbox | undefined;
+    private jobQueue: JobQueue<IntermediateEmailDetails>;
 
     /** @internal */
     constructor(
         private eventBus: EventBus,
         @InjectConnection() private connection: Connection,
         private moduleRef: ModuleRef,
+        private workerService: WorkerService,
+        private jobQueueService: JobQueueService,
     ) {}
 
     /**
@@ -178,24 +181,6 @@ export class EmailPlugin implements OnVendureBootstrap, OnVendureClose {
     /** @internal */
     async onVendureBootstrap(): Promise<void> {
         const options = EmailPlugin.options;
-        if (isDevModeOptions(options)) {
-            this.transport = {
-                type: 'file',
-                raw: false,
-                outputPath: options.outputPath,
-            };
-        } else {
-            if (!options.transport) {
-                throw new InternalServerError(
-                    `When devMode is not set to true, the 'transport' property must be set.`,
-                );
-            }
-            this.transport = options.transport;
-        }
-
-        this.templateLoader = new TemplateLoader(options.templatePath);
-        this.emailSender = new EmailSender();
-        this.generator = new HandlebarsMjmlGenerator();
 
         if (isDevModeOptions(options) && options.mailboxPort !== undefined) {
             this.devMailbox = new DevMailbox();
@@ -204,9 +189,17 @@ export class EmailPlugin implements OnVendureBootstrap, OnVendureClose {
         }
 
         await this.setupEventSubscribers();
-        if (this.generator.onInit) {
-            await this.generator.onInit.call(this.generator, options);
-        }
+
+        this.jobQueue = this.jobQueueService.createQueue({
+            name: 'send-email',
+            concurrency: 5,
+            process: (job) => {
+                this.workerService.send(new EmailWorkerMessage(job.data)).subscribe({
+                    complete: () => job.complete(),
+                    error: (err) => job.fail(err),
+                });
+            },
+        });
     }
 
     /** @internal */
@@ -218,16 +211,10 @@ export class EmailPlugin implements OnVendureBootstrap, OnVendureClose {
 
     private async setupEventSubscribers() {
         for (const handler of EmailPlugin.options.handlers) {
-            this.eventBus.ofType(handler.event).subscribe(event => {
+            this.eventBus.ofType(handler.event).subscribe((event) => {
                 return this.handleEvent(handler, event);
             });
         }
-        if (this.transport.type === 'file') {
-            // ensure the configured directory exists before
-            // we attempt to write files to it
-            const emailPath = this.transport.outputPath;
-            await fs.ensureDir(emailPath);
-        }
     }
 
     private async handleEvent(
@@ -241,30 +228,16 @@ export class EmailPlugin implements OnVendureBootstrap, OnVendureClose {
                 (event as EventWithAsyncData<EventWithContext, any>).data = await handler._loadDataFn({
                     event,
                     connection: this.connection,
-                    inject: t => this.moduleRef.get(t, { strict: false }),
+                    inject: (t) => this.moduleRef.get(t, { strict: false }),
                 });
             }
             const result = await handler.handle(event as any, EmailPlugin.options.globalTemplateVars);
             if (!result) {
                 return;
             }
-            const bodySource = await this.templateLoader.loadTemplate(type, result.templateFile);
-            const generated = await this.generator.generate(
-                result.from,
-                result.subject,
-                bodySource,
-                result.templateVars,
-            );
-            const emailDetails = { ...generated, recipient: result.recipient };
-            await this.emailSender.send(emailDetails, this.transport);
+            await this.jobQueue.add(result);
         } catch (e) {
             Logger.error(e.message, 'EmailPlugin', e.stack);
         }
     }
 }
-
-function isDevModeOptions(
-    input: EmailPluginOptions | EmailPluginDevModeOptions,
-): input is EmailPluginDevModeOptions {
-    return (input as EmailPluginDevModeOptions).devMode === true;
-}

packages/email-plugin/src/types.ts

@@ -1,5 +1,5 @@
 import { Omit } from '@vendure/common/lib/omit';
-import { RequestContext, Type, VendureEvent } from '@vendure/core';
+import { RequestContext, Type, VendureEvent, WorkerMessage } from '@vendure/core';
 import { Connection } from 'typeorm';
 
 import { EmailEventHandler } from './event-handler';
@@ -272,3 +272,16 @@ export type LoadDataFn<Event extends EventWithContext, R> = (context: {
     connection: Connection;
     inject: <T>(type: Type<T>) => T;
 }) => Promise<R>;
+
+export type IntermediateEmailDetails = {
+    type: string;
+    from: string;
+    recipient: string;
+    templateVars: any;
+    subject: string;
+    templateFile: string;
+};
+
+export class EmailWorkerMessage extends WorkerMessage<IntermediateEmailDetails, boolean> {
+    static readonly pattern = 'send-email';
+}