chore: apply feedback to job services

Author: k1eu

apps/api/.env.example

@@ -8,6 +8,7 @@ EMAIL_ADAPTER="mailhog"
 DATABASE_URL="postgres://postgres:boilerplate@localhost:5432/boilerplate"
 DATABASE_TEST_URL="postgres://postgres:boilerplate@localhost:5432/boilerplate_test"
 REDIS_URL="redis://localhost:6379"
+BULLBOARD_PASSWORD=admin123
 
 # MAILS
 SMTP_HOST=

apps/api/src/auth/auth-email.consumer.ts

@@ -2,49 +2,56 @@ import { Processor, WorkerHost } from "@nestjs/bullmq";
 import { AuthService } from "./auth.service";
 import { Job } from "bullmq";
 import {
-  QUEUE_EMAIL,
-  QueueEmailJobData,
-  QueueEmailJobPayloads,
+  EMAIL_QUEUE,
+  EmailQueueJobData,
+  EmailQueueJobPayloads,
 } from "./auth.queue";
 
-@Processor(QUEUE_EMAIL.name)
+type QueueEmailJobType = EmailQueueJobData["type"];
+type QueueEmailJob<T extends QueueEmailJobType = QueueEmailJobType> = Job<
+  Extract<EmailQueueJobData, { type: T }>,
+  unknown,
+  string
+>;
+
+@Processor(EMAIL_QUEUE.name)
 export class AuthEmailConsumer extends WorkerHost {
   constructor(private readonly authService: AuthService) {
     super();
   }
 
-  async process(
-    job: Job<QueueEmailJobData, unknown, string>,
-  ): Promise<unknown> {
-    switch (job.name) {
-      case QUEUE_EMAIL.actions.SEND_WELCOME_EMAIL:
-        return this.sendWelcomeEmail(
-          job as Job<QueueEmailJobPayloads["SEND_WELCOME_EMAIL"]>,
-        );
-      case QUEUE_EMAIL.actions.SEND_RESET_PASSWORD_EMAIL:
-        return this.sendResetPasswordEmail(
-          job as Job<QueueEmailJobPayloads["SEND_RESET_PASSWORD_EMAIL"]>,
-        );
-      default:
-        throw new Error(`Unknown job name: ${job.name}`);
+  async process(job: QueueEmailJob): Promise<unknown> {
+    if (this.isJobOfType(job, EMAIL_QUEUE.actions.SEND_WELCOME_EMAIL)) {
+      return this.sendWelcomeEmail(job.data);
+    }
+
+    if (this.isJobOfType(job, EMAIL_QUEUE.actions.SEND_RESET_PASSWORD_EMAIL)) {
+      return this.sendResetPasswordEmail(job.data);
     }
+
+    throw new Error(`Unknown job name: ${job.name}`);
   }
 
   async sendResetPasswordEmail(
-    job: Job<QueueEmailJobPayloads["SEND_RESET_PASSWORD_EMAIL"]>,
+    payload: EmailQueueJobPayloads["SEND_RESET_PASSWORD_EMAIL"],
   ) {
-    const { url, name, email } = job.data;
+    const { url, name, email } = payload;
     return this.authService.sendResetPassordEmail(email, { url, name });
   }
 
-  async sendWelcomeEmail(
-    job: Job<QueueEmailJobPayloads["SEND_WELCOME_EMAIL"]>,
-  ) {
-    const { url, name, email } = job.data;
+  async sendWelcomeEmail(payload: EmailQueueJobPayloads["SEND_WELCOME_EMAIL"]) {
+    const { url, name, email } = payload;
     return this.authService.sendWelcomeVerifyEmail(email, {
       url,
       name,
       email,
     });
   }
+
+  private isJobOfType<T extends QueueEmailJobType>(
+    job: QueueEmailJob,
+    type: T,
+  ): job is QueueEmailJob<T> {
+    return job.data.type === type;
+  }
 }

apps/api/src/auth/auth-email.producer.ts

@@ -0,0 +1,44 @@
+import { InjectQueue } from "@nestjs/bullmq";
+import { Injectable } from "@nestjs/common";
+import { Queue } from "bullmq";
+import { EMAIL_QUEUE, EmailQueueJobPayloads } from "./auth.queue";
+
+@Injectable()
+export class AuthEmailService {
+  private readonly jobSettings = { age: 3600 };
+
+  constructor(@InjectQueue(EMAIL_QUEUE.name) private readonly queue: Queue) {}
+
+  public async sendWelcomeMessageEmailAsync(
+    jobId: string,
+    payload: Omit<EmailQueueJobPayloads["SEND_WELCOME_EMAIL"], "type">,
+  ) {
+    const opts = { jobId, removeOnComplete: this.jobSettings };
+    const payloadWithType = {
+      ...payload,
+      type: EMAIL_QUEUE.actions.SEND_WELCOME_EMAIL,
+    };
+    await this.queue.add(
+      EMAIL_QUEUE.actions.SEND_WELCOME_EMAIL,
+      payloadWithType,
+      opts,
+    );
+  }
+
+  public async sendResetPasswordEmailAsync(
+    jobId: string,
+    payload: Omit<EmailQueueJobPayloads["SEND_RESET_PASSWORD_EMAIL"], "type">,
+  ) {
+    const opts = { jobId, removeOnComplete: this.jobSettings };
+    const payloadWithType = {
+      ...payload,
+      type: EMAIL_QUEUE.actions.SEND_RESET_PASSWORD_EMAIL,
+    };
+
+    await this.queue.add(
+      EMAIL_QUEUE.actions.SEND_RESET_PASSWORD_EMAIL,
+      payloadWithType,
+      opts,
+    );
+  }
+}

apps/api/src/auth/auth-email.service.ts

@@ -2,9 +2,9 @@ import { Module } from "@nestjs/common";
 import { ConfigModule } from "@nestjs/config";
 import { AuthService } from "./auth.service";
 import { BullModule } from "@nestjs/bullmq";
-import { QUEUE_EMAIL } from "./auth.queue";
+import { EMAIL_QUEUE } from "./auth.queue";
 import { AuthEmailConsumer } from "./auth-email.consumer";
-import { AuthEmailProducer } from "./auth-email.producer";
+import { AuthEmailService } from "./auth-email.service";
 import { EmailModule } from "src/common/emails/emails.module";
 import { BullBoardModule } from "@bull-board/nestjs";
 import { BullMQAdapter } from "@bull-board/api/bullMQAdapter";
@@ -13,15 +13,15 @@ import { BullMQAdapter } from "@bull-board/api/bullMQAdapter";
   imports: [
     ConfigModule,
     BullModule.registerQueue({
-      name: QUEUE_EMAIL.name,
+      name: EMAIL_QUEUE.name,
     }),
     BullBoardModule.forFeature({
-      name: QUEUE_EMAIL.name,
+      name: EMAIL_QUEUE.name,
       adapter: BullMQAdapter,
     }),
     EmailModule,
   ],
-  providers: [AuthService, AuthEmailConsumer, AuthEmailProducer],
-  exports: [AuthService, AuthEmailProducer],
+  providers: [AuthService, AuthEmailConsumer, AuthEmailService],
+  exports: [AuthService, AuthEmailService],
 })
 export class AuthModule {}

apps/api/src/auth/auth.module.ts

@@ -1,24 +1,26 @@
-export const QUEUE_EMAIL = {
+export const EMAIL_QUEUE = {
   name: "email-queue",
   actions: {
-    SEND_WELCOME_EMAIL: "send-welcome-email" as const,
-    SEND_RESET_PASSWORD_EMAIL: "send-reset-password-email" as const,
+    SEND_WELCOME_EMAIL: "SEND_WELCOME_EMAIL" as const,
+    SEND_RESET_PASSWORD_EMAIL: "SEND_RESET_PASSWORD_EMAIL" as const,
   },
 };
 
-export type QueueEmailJobPayloads = {
+export type EmailQueueJobPayloads = {
   SEND_WELCOME_EMAIL: {
+    type: "SEND_WELCOME_EMAIL";
     url: string;
     name: string;
     email: string;
   };
   SEND_RESET_PASSWORD_EMAIL: {
+    type: "SEND_RESET_PASSWORD_EMAIL";
     url: string;
     name: string;
     email: string;
   };
 };
 
-export type QueueEmailJobData =
-  | QueueEmailJobPayloads["SEND_WELCOME_EMAIL"]
-  | QueueEmailJobPayloads["SEND_RESET_PASSWORD_EMAIL"];
+export type EmailQueueJobData =
+  | EmailQueueJobPayloads["SEND_WELCOME_EMAIL"]
+  | EmailQueueJobPayloads["SEND_RESET_PASSWORD_EMAIL"];

apps/api/src/auth/auth.queue.ts

@@ -6,21 +6,21 @@ import {
   RESET_PASSWORD_EMAIL,
   WELCOME_VERIFY_EMAIL,
 } from "src/common/emails/email.consts";
-import { AuthEmailProducer } from "./auth-email.producer";
+import { AuthEmailService } from "./auth-email.service";
 
 @Injectable()
 export class AuthService {
   constructor(
     private readonly emailService: EmailService,
-    private readonly authEmailProducer: AuthEmailProducer,
+    private readonly authEmailService: AuthEmailService,
   ) {}
 
   public async onWelcomeEmail(
     to: string,
     data: { email: string; name: string; url: string },
   ) {
     const jobId = `welcome-email-${data.email}-${Date.now()}`;
-    await this.authEmailProducer.newUserWelcomeMessageSent(jobId, {
+    await this.authEmailService.sendWelcomeMessageEmailAsync(jobId, {
       email: data.email,
       name: data.name,
       url: data.url,
@@ -32,7 +32,7 @@ export class AuthService {
     data: { email: string; name: string; url: string },
   ) {
     const jobId = `reset-password-${data.email}-${Date.now()}`;
-    await this.authEmailProducer.resetPasswordMessageSent(jobId, {
+    await this.authEmailService.sendResetPasswordEmailAsync(jobId, {
       email: data.email,
       name: data.name,
       url: data.url,

apps/api/src/auth/auth.service.ts

@@ -17,12 +17,16 @@ import basicAuth from "express-basic-auth";
         },
       }),
     }),
-    BullBoardModule.forRoot({
-      route: "/queues",
-      adapter: ExpressAdapter,
-      middleware: basicAuth({
-        challenge: true,
-        users: { admin: "passwordhere" },
+    BullBoardModule.forRootAsync({
+      imports: [ConfigModule],
+      inject: [ConfigService],
+      useFactory: async (configService: ConfigService) => ({
+        route: "/queues",
+        adapter: ExpressAdapter,
+        middleware: basicAuth({
+          challenge: true,
+          users: { admin: configService.get<string>("BULLBOARD_PASSWORD")! },
+        }),
       }),
     }),
   ],

apps/api/src/queue/queue.module.ts

@@ -15,7 +15,7 @@ import {
 import { truncateTables } from "test/helpers/test-helpers";
 import { EmailService } from "src/common/emails/emails.service";
 import { Email } from "src/common/emails/email.interface";
-import { QUEUE_USER_ALERT } from "../users.queue";
+import { USER_ALERT_QUEUE } from "../users.queue";
 import {
   createQueueTestHarness,
   QueueTestHarness,
@@ -45,7 +45,7 @@ describe("UsersController (e2e)", () => {
     app = testApp;
     db = testDb;
     userFactory = createUserFactory(db);
-    queueHarness = await createQueueTestHarness(app, QUEUE_USER_ALERT.name);
+    queueHarness = await createQueueTestHarness(app, USER_ALERT_QUEUE.name);
   });
 
   afterAll(async () => {

apps/api/src/users/__tests__/user.controller.e2e-spec.ts

@@ -2,33 +2,40 @@ import { Processor, WorkerHost } from "@nestjs/bullmq";
 import { UsersService } from "./users.service";
 import { Job } from "bullmq";
 import {
-  QUEUE_USER_ALERT,
-  QueueUserAlertJobData,
-  QueueUserAlertJobPayloads,
+  USER_ALERT_QUEUE,
+  UserAlertQueueJobData,
+  UserAlertQueueJobPayloads,
 } from "./users.queue";
 
-@Processor(QUEUE_USER_ALERT.name)
+type QueueUserAlertJobType = UserAlertQueueJobData["type"];
+type QueueUserAlertJob<
+  T extends QueueUserAlertJobType = QueueUserAlertJobType,
+> = Job<Extract<UserAlertQueueJobData, { type: T }>, unknown, string>;
+
+@Processor(USER_ALERT_QUEUE.name)
 export class UsersAlertConsumer extends WorkerHost {
   constructor(private readonly usersService: UsersService) {
     super();
   }
 
-  async process(job: Job<QueueUserAlertJobData, unknown, string>): Promise<unknown> {
-    switch (job.name) {
-      case QUEUE_USER_ALERT.actions.SEND_ALERT_EMAIL:
-        return this.sendAlertEmail(
-          job as Job<QueueUserAlertJobPayloads["SEND_ALERT_EMAIL"]>,
-        );
-      default:
-        throw new Error(`Unknown job name: ${job.name}`);
+  async process(job: QueueUserAlertJob): Promise<unknown> {
+    if (this.isJobOfType(job, USER_ALERT_QUEUE.actions.SEND_ALERT_EMAIL)) {
+      return this.sendAlertEmail(job.data);
     }
+
+    // @ts-expect-error should not reach here
+    throw new Error(`Unknown job name: ${job.name}`);
   }
 
-  async sendAlertEmail(
-    job: Job<QueueUserAlertJobPayloads["SEND_ALERT_EMAIL"]>,
-  ) {
-    console.log('processing job', job.id, job.data);
-    const { email } = job.data;
+  async sendAlertEmail(payload: UserAlertQueueJobPayloads["SEND_ALERT_EMAIL"]) {
+    const { email } = payload;
     return this.usersService.sendAlertEmail(email);
   }
+
+  private isJobOfType<T extends QueueUserAlertJobType>(
+    job: QueueUserAlertJob,
+    type: T,
+  ): job is QueueUserAlertJob<T> {
+    return job.data.type === type;
+  }
 }