clean up

apps/data-service/src/services/queue.service.ts

@@ -11,26 +11,53 @@ export interface JobData {
 }
 
 export class QueueService {
-  private logger = new Logger('queue-manager');
-  private queue: Queue;
-  private worker: Worker;
-  private queueEvents: QueueEvents;
+  private logger = new Logger('queue-service');
+  private queue!: Queue;
+  private worker!: Worker;
+  private queueEvents!: QueueEvents;
+  private isInitialized = false;
 
   constructor() {
+    // Don't initialize in constructor to allow for proper async initialization
+  }
+
+  async initialize() {
+    if (this.isInitialized) {
+      this.logger.warn('Queue service already initialized');
+      return;
+    }
+
+    this.logger.info('Initializing queue service...');
+
     const connection = {
       host: process.env.DRAGONFLY_HOST || 'localhost',
       port: parseInt(process.env.DRAGONFLY_PORT || '6379'),
     };
 
-    this.queue = new Queue('data-service-queue', { connection });
-    this.worker = new Worker('data-service-queue', this.processJob.bind(this), {
-      connection,
-      concurrency: 10
-    });
-    this.queueEvents = new QueueEvents('data-service-queue', { connection });
+    this.logger.info('Connecting to Redis/Dragonfly', connection);
 
-    this.setupEventListeners();
-    this.setupScheduledTasks();
+    try {
+      this.queue = new Queue('data-service-queue', { connection });
+      this.worker = new Worker('data-service-queue', this.processJob.bind(this), {
+        connection,
+        concurrency: 10
+      });
+      this.queueEvents = new QueueEvents('data-service-queue', { connection });
+
+      // Test connection
+      await this.queue.waitUntilReady();
+      await this.worker.waitUntilReady();
+      await this.queueEvents.waitUntilReady();
+
+      this.setupEventListeners();
+      this.setupScheduledTasks();
+
+      this.isInitialized = true;
+      this.logger.info('Queue service initialized successfully');
+    } catch (error) {
+      this.logger.error('Failed to initialize queue service', { error });
+      throw error;
+    }
   }
 
   private async processJob(job: any) {
@@ -111,8 +138,10 @@ export class QueueService {
 
     this.logger.info('Scheduled tasks configured');
   }
-
   async addJob(jobData: JobData, options?: any) {
+    if (!this.isInitialized) {
+      throw new Error('Queue service not initialized. Call initialize() first.');
+    }
     return this.queue.add(jobData.type, jobData, {
       priority: jobData.priority || 0,
       removeOnComplete: 10,
@@ -122,6 +151,9 @@ export class QueueService {
   }
 
   async addRecurringJob(jobData: JobData, cronPattern: string) {
+    if (!this.isInitialized) {
+      throw new Error('Queue service not initialized. Call initialize() first.');
+    }
     return this.queue.add(
       `recurring-${jobData.type}`,
       jobData,
@@ -135,6 +167,9 @@ export class QueueService {
   }
 
   async getJobStats() {
+    if (!this.isInitialized) {
+      throw new Error('Queue service not initialized. Call initialize() first.');
+    }
     const [waiting, active, completed, failed, delayed] = await Promise.all([
       this.queue.getWaiting(),
       this.queue.getActive(),
@@ -150,8 +185,10 @@ export class QueueService {
       failed: failed.length,
       delayed: delayed.length
     };
-  }
-  async getQueueStatus() {
+  }  async getQueueStatus() {
+    if (!this.isInitialized) {
+      throw new Error('Queue service not initialized. Call initialize() first.');
+    }
     const stats = await this.getJobStats();
     return {
       ...stats,
@@ -165,6 +202,9 @@ export class QueueService {
   }
 
   getWorkerCount() {
+    if (!this.isInitialized) {
+      return 0;
+    }
     return this.worker.opts.concurrency || 1;
   }
 
@@ -174,12 +214,16 @@ export class QueueService {
       { name: 'market-data-provider', type: 'market-data', operations: ['live-data', 'historical-data'] }
     ];
   }
-
   async shutdown() {
-    this.logger.info('Shutting down queue manager');
+    if (!this.isInitialized) {
+      this.logger.warn('Queue service not initialized, nothing to shutdown');
+      return;
+    }
+    this.logger.info('Shutting down queue service');
     await this.worker.close();
     await this.queue.close();
     await this.queueEvents.close();
+    this.isInitialized = false;
   }
 }