feat(queue): improve jobs

package.json

@@ -289,10 +289,7 @@
       "#src/validators/update.validator"
     ],
     "jobs": [
-      "#src/jobs/userconfirm.job",
-      "#src/jobs/useremail.job",
-      "#src/jobs/useremailpassword.job",
-      "#src/jobs/userpassword.job"
+      "#src/jobs/mail.job"
     ]
   }
 }

src/helpers/queue.ts

@@ -3,12 +3,17 @@ import { Service } from '@athenna/ioc'
 import { Database, type DatabaseImpl } from '@athenna/database'
 
 class VanillaQueue {
+  public connection: string
   private queueName = 'default'
   private queues: Record<string, any[]> = {
     default: [],
     deadletter: []
   }
 
+  public constructor(connection: string) {
+    this.connection = connection
+  }
+
   public async truncate() {
     Object.keys(this.queues).forEach(key => (this.queues[key] = []))
   }
@@ -179,11 +184,11 @@ class DatabaseQueue {
 
 @Service({ alias: 'App/Helpers/Queue', type: 'singleton' })
 export class QueueImpl {
-  public driver: any = new VanillaQueue()
+  public driver: any = new VanillaQueue('vanilla')
 
   public connection(name: string) {
-    if (name === 'vanilla') {
-      this.driver = new VanillaQueue()
+    if (name === 'vanilla' || name === 'default') {
+      this.driver = new VanillaQueue('vanilla')
     }
 
     if (name === 'database') {

src/jobs/base.job.ts

@@ -0,0 +1,19 @@
+import { Queue } from '#src/providers/facades/queue'
+
+export class BaseJob {
+  public static connection() {
+    return 'default'
+  }
+
+  public static queue() {
+    const connection = this.connection()
+
+    return Config.get(`queue.connections.${connection}.queue`)
+  }
+
+  public queue() {
+    const Job = this.constructor as typeof BaseJob
+
+    return Queue.connection(Job.connection()).queue(Job.queue())
+  }
+}

src/jobs/mail.job.ts

@@ -0,0 +1,31 @@
+import { Mail } from '@athenna/mail'
+import type { User } from '#src/models/user'
+import { BaseJob } from '#src/jobs/base.job'
+
+type Item = {
+  view: string
+  subject: string
+  user: User
+  email: string
+  password: string
+  token: string
+}
+
+export class MailJob extends BaseJob {
+  public static queue() {
+    return 'mail'
+  }
+
+  public async handle(item: Item) {
+    await Mail.from('noreply@athenna.io')
+      .to(item.user.email)
+      .subject(item.subject)
+      .view(item.view, {
+        user: item.user,
+        email: item.email,
+        password: item.password,
+        token: item.token
+      })
+      .send()
+  }
+}

src/providers/queueworker.provider.ts

@@ -1,7 +1,6 @@
 import { Log } from '@athenna/logger'
 import { ServiceProvider } from '@athenna/ioc'
 import { Exec, Module } from '@athenna/common'
-import { Queue } from '#src/providers/facades/queue'
 
 export default class QueueWorkerProvider extends ServiceProvider {
   public intervals = []
@@ -17,7 +16,7 @@ export default class QueueWorkerProvider extends ServiceProvider {
       const job = this.container.transient(Job, alias).use(alias)
 
       const interval = setInterval(async () => {
-        const queue = Queue.queue(queueName)
+        const queue = job.queue()
 
         if (queue.isEmpty()) {
           return

src/services/auth.service.ts

@@ -5,9 +5,9 @@ import { Uuid } from '@athenna/common'
 import { Service } from '@athenna/ioc'
 import { Config } from '@athenna/config'
 import { User } from '#src/models/user'
-import { Queue } from '#src/providers/facades/queue'
 import { UnauthorizedException } from '@athenna/http'
 import type { UserService } from '#src/services/user.service'
+import { Queue } from '#src/providers/facades/queue'
 
 @Service()
 export class AuthService {
@@ -47,7 +47,11 @@ export class AuthService {
 
     const user = await this.userService.create(data)
 
-    await Queue.queue('user:confirm').add(user)
+    await Queue.connection('vanilla').queue('mail').add({
+      user,
+      view: 'mail/confirm',
+      subject: 'Athenna Account Confirmation'
+    })
 
     return user
   }

src/services/user.service.ts

@@ -5,8 +5,8 @@ import { Role } from '#src/models/role'
 import { RoleUser } from '#src/models/roleuser'
 import { RoleEnum } from '#src/enums/role.enum'
 import { NotFoundException } from '@athenna/http'
-import { Queue } from '#src/providers/facades/queue'
 import { Json, type PaginationOptions } from '@athenna/common'
+import { Queue } from '#src/providers/facades/queue'
 
 @Service()
 export class UserService {
@@ -62,25 +62,38 @@ export class UserService {
 
     switch (`${isEmailEqual}:${isPasswordEqual}`) {
       case 'false:true':
-        await Queue.queue('user:email').add({ user, token, email: data.email })
+        await Queue.connection('vanilla').queue('mail').add({
+          user,
+          token,
+          email: data.email,
+          view: 'mail/change-email',
+          subject: 'Athenna Email Change'
+        })
+
         break
       case 'true:false':
+        // TODO create a password_resets table to save the password
         data.password = await bcrypt.hash(data.password, 10)
 
-        await Queue.queue('user:password').add({
+        await Queue.connection('vanilla').queue('mail').add({
           user,
           token,
-          password: data.password
+          password: data.password,
+          view: 'mail/change-password',
+          subject: 'Athenna Email Change'
         })
         break
       case 'false:false':
+        // TODO create a password_resets table to save the password
         data.password = await bcrypt.hash(data.password, 10)
 
-        await Queue.queue('user:email:password').add({
+        await Queue.connection('vanilla').queue('mail').add({
           user,
           token,
           email: data.email,
-          password: data.password
+          password: data.password,
+          view: 'mail/change-email-password',
+          subject: 'Athenna Email & Password Change'
         })
     }
 

tests/e2e/auth.controller.test.ts

@@ -127,7 +127,7 @@ export default class AuthControllerTest extends BaseE2ETest {
       }
     })
 
-    const queue = Queue.queue('user:confirm')
+    const queue = Queue.queue('mail')
 
     assert.deepEqual(await queue.length(), 1)
     assert.isTrue(await User.exists({ email: 'test@athenna.io' }))

tests/e2e/user.controller.test.ts

@@ -155,7 +155,7 @@ export default class UserControllerTest extends BaseE2ETest {
 
     await user.refresh()
 
-    const queue = Queue.queue('user:email')
+    const queue = Queue.queue('mail')
 
     assert.deepEqual(await queue.length(), 1)
     assert.deepEqual(user.name, 'Customer Updated')
@@ -177,7 +177,7 @@ export default class UserControllerTest extends BaseE2ETest {
 
     await user.refresh()
 
-    const queue = Queue.queue('user:password')
+    const queue = Queue.queue('mail')
 
     assert.deepEqual(await queue.length(), 1)
     assert.deepEqual(user.name, 'Customer Updated')
@@ -204,7 +204,7 @@ export default class UserControllerTest extends BaseE2ETest {
 
     await user.refresh()
 
-    const queue = Queue.queue('user:email:password')
+    const queue = Queue.queue('mail')
 
     assert.deepEqual(await queue.length(), 1)
     assert.deepEqual(user.name, 'Customer Updated')