Working proof of concept for queue

src/managers/queueManager.ts

@@ -4,6 +4,8 @@ import config from '../utilities/config'
 import { Server as SocketServer } from 'socket.io'
 import { TSocket } from '../utilities/types'
 import { queueLogger } from '../utilities/logger'
+import fs from 'fs'
+import path from 'path'
 
 class QueueManager {
   private connection!: IORedis
@@ -26,7 +28,10 @@ class QueueManager {
       process.exit(1)
     }
 
-    this.queue = new Queue('jobs')
+    this.queue = new Queue('jobs', {
+      connection: this.connection
+    })
+
     this.worker = new Worker('jobs', this.processJob.bind(this), {
       connection: this.connection,
       concurrency: 10000
@@ -44,46 +49,51 @@ class QueueManager {
   }
 
   private async processJob(job: Job) {
-    const { jobClass, params, socketId } = job.data
+    console.log('Processing job:', job.data.jobName)
+    const { jobName, params, socketId } = job.data
 
-    console.log('Processing job:', job)
     try {
-      const JobClass = await import(`../jobs/${jobClass}`)
+      const jobsDir = path.join(process.cwd(), 'src', 'jobs')
+      const extension = config.ENV === 'development' ? '.ts' : '.js'
+      const jobPath = path.join(jobsDir, `${jobName}${extension}`)
 
-      if (!JobClass) {
+      if (!fs.existsSync(jobPath)) {
-        queueLogger.warn(`Job class not found: ${jobClass}`)
+        queueLogger.warn(`Job file not found: ${jobPath}`)
         return
       }
 
-      console.log('Job class:', JobClass)
+      const JobModule = await import(jobPath)
+      const JobClass = JobModule.default
 
-      console.log('Job class:', JobClass.name)
+      if (!JobClass || typeof JobClass !== 'function') {
+        queueLogger.warn(`Invalid job class in file: ${jobPath}`)
+        return
+      }
 
-      const job = new JobClass(params)
+      const jobInstance = new JobClass(params)
 
       if (socketId && this.io) {
         const socket = this.io.sockets.sockets.get(socketId)
         if (socket) {
-          await job.execute(this.io, socket)
+          await jobInstance.execute(this.io, socket)
         } else {
           queueLogger.warn(`Socket not found for job: ${socketId}`)
-          await job.execute(this.io)
+          await jobInstance.execute(this.io)
         }
       } else {
-        await job.execute(this.io)
+        await jobInstance.execute(this.io)
       }
     } catch (error: any) {
-      queueLogger.error(`Error processing job: ${error.message}`)
+      queueLogger.error(`Error processing job ${jobName}: ${error.message}`)
     }
   }
 
-  public async addToQueue(jobClass: any, params: any, socket?: TSocket) {
+  public async newJob(jobName: string, params: any, socket?: TSocket) {
     const jobData = {
-      jobClass: jobClass.name,
+      jobName,
       params,
       socketId: socket?.id
     }
-
     await this.queue.add('job', jobData)
   }
 }

src/socketEvents/character/connect.ts

@@ -3,7 +3,6 @@ import { TSocket, ExtendedCharacter } from '../../utilities/types'
 import CharacterRepository from '../../repositories/characterRepository'
 import CharacterManager from '../../managers/characterManager'
 import QueueManager from '../../managers/queueManager'
-import SomeJob from '../../jobs/characterLeaveZone'
 
 type SocketResponseT = {
   character_id: number
@@ -16,7 +15,7 @@ export default function (io: Server, socket: TSocket) {
       const character = await CharacterRepository.getByUserAndId(socket?.user?.id as number, data.character_id)
       if (!character) return
       socket.characterId = character.id
-      await QueueManager.addToQueue(SomeJob, { someParam: 'value' }, socket)
+      await QueueManager.newJob('SomeJob', { someParam: 'value' }, socket)
       CharacterManager.initCharacter(character as ExtendedCharacter)
       socket.emit('character:connect', character)
     } catch (error: any) {

src/utilities/config.ts

@@ -3,7 +3,7 @@ import dotenv from 'dotenv'
 dotenv.config()
 
 class config {
-  static ENV: string = process.env.ENV || 'prod'
+  static ENV: string = process.env.ENV || 'development'
   static REDIS_URL: string = process.env.REDIS_URL || 'redis://@127.0.0.1:6379/4'
   static HOST: string = process.env.HOST || '0.0.0.0'
   static PORT: number = process.env.PORT ? parseInt(process.env.PORT) : 6969