added env back and fixed up queue service

apps/data-service/src/index.ts

@@ -4,7 +4,7 @@
 import { getLogger } from '@stock-bot/logger';
 import { loadEnvVariables } from '@stock-bot/config';
 import { Hono } from 'hono';
-import { onShutdown, setShutdownTimeout } from '@stock-bot/shutdown';
+import { Shutdown } from '@stock-bot/shutdown';
 import { queueManager } from './services/queue.service';
 import { initializeBatchCache } from './utils/batch-helpers';
 import { initializeProxyCache } from './providers/proxy.tasks';
@@ -24,6 +24,9 @@ const logger = getLogger('data-service');
 const PORT = parseInt(process.env.DATA_SERVICE_PORT || '3002');
 let server: any = null;
 
+// Initialize shutdown manager with 15 second timeout
+const shutdown = Shutdown.getInstance({ timeout: 15000 });
+
 // Register all routes
 app.route('', healthRoutes);
 app.route('', queueRoutes);
@@ -70,26 +73,34 @@ async function startServer() {
   logger.info(`Data Service started on port ${PORT}`);
 }
 
-// Setup shutdown handling
-setShutdownTimeout(15000);
-
-// Register cleanup for HTTP server
-onShutdown(async () => {
+// Register shutdown handlers
+shutdown.onShutdown(async () => {
   if (server) {
     logger.info('Stopping HTTP server...');
-    server.stop();
+    try {
+      server.stop();
+      logger.info('HTTP server stopped successfully');
+    } catch (error) {
+      logger.error('Error stopping HTTP server', { error });
+    }
   }
 });
 
-// Register cleanup for queue manager
-onShutdown(async () => {
+shutdown.onShutdown(async () => {
   logger.info('Shutting down queue manager...');
-  await queueManager.shutdown();
+  try {
+    await queueManager.shutdown();
+    logger.info('Queue manager shut down successfully');
+  } catch (error) {
+    logger.error('Error shutting down queue manager', { error });
+    throw error; // Re-throw to mark shutdown as failed
+  }
 });
 
+// Start the application
 startServer().catch(error => {
   logger.error('Failed to start server', { error });
   process.exit(1);
 });
 
-logger.info('Shutdown handlers registered');
+logger.info('Data service startup initiated with graceful shutdown handlers');

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

@@ -34,7 +34,8 @@ queueRoutes.post('/api/queue/job', async (c) => {
 // Provider registry endpoints
 queueRoutes.get('/api/providers', async (c) => {
   try {
-    const providers = queueManager.getRegisteredProviders();
+    const { providerRegistry } = await import('../services/provider-registry.service');
+    const providers = providerRegistry.getProviders();
     return c.json({ status: 'success', providers });
   } catch (error) {
     logger.error('Failed to get providers', { error });
@@ -45,7 +46,8 @@ queueRoutes.get('/api/providers', async (c) => {
 // Add new endpoint to see scheduled jobs
 queueRoutes.get('/api/scheduled-jobs', async (c) => {
   try {
-    const jobs = queueManager.getScheduledJobsInfo();
+    const { providerRegistry } = await import('../services/provider-registry.service');
+    const jobs = providerRegistry.getAllScheduledJobs();
     return c.json({ 
       status: 'success', 
       count: jobs.length,
@@ -56,3 +58,14 @@ queueRoutes.get('/api/scheduled-jobs', async (c) => {
     return c.json({ status: 'error', message: 'Failed to get scheduled jobs' }, 500);
   }
 });
+
+queueRoutes.post('/api/queue/drain', async (c) => {
+  try {
+    await queueManager.drainQueue();
+    const status = await queueManager.getQueueStatus();
+    return c.json({ status: 'success', message: 'Queue drained', queueStatus: status });
+  } catch (error) {
+    logger.error('Failed to drain queue', { error });
+    return c.json({ status: 'error', message: 'Failed to drain queue' }, 500);
+  }
+});

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

@@ -292,7 +292,6 @@ export class QueueService {
       delayed: delayed.length
     };
   }
-
   async drainQueue() {
     if (this.isInitialized) {
       await this.queue.drain();
@@ -309,24 +308,71 @@ export class QueueService {
       workers: this.workers.length,
       concurrency: this.getTotalConcurrency()
     };
-  }
-  async shutdown() {
+  }  async shutdown() {
     if (!this.isInitialized) {
       this.logger.warn('Queue service not initialized, nothing to shutdown');
       return;
     }
     
-    this.logger.info('Shutting down queue service');
+    this.logger.info('Shutting down queue service gracefully...');
     
-    // Close all workers
-    await Promise.all(this.workers.map((worker, index) => {
-      this.logger.debug(`Closing worker ${index + 1}`);
-      return worker.close();
-    }));
-    
-    await this.queue.close();
-    await this.queueEvents.close();
-    this.logger.info('Queue service shutdown complete');
+    try {
+      // Step 1: Stop accepting new jobs and wait for current jobs to finish
+      this.logger.debug('Closing workers gracefully...');
+      const workerClosePromises = this.workers.map(async (worker, index) => {
+        this.logger.debug(`Closing worker ${index + 1}/${this.workers.length}`);
+        try {
+          // Wait for current jobs to finish, then close
+          await Promise.race([
+            worker.close(),
+            new Promise((_, reject) => 
+              setTimeout(() => reject(new Error(`Worker ${index + 1} close timeout`)), 5000)
+            )
+          ]);
+          this.logger.debug(`Worker ${index + 1} closed successfully`);
+        } catch (error) {
+          this.logger.error(`Failed to close worker ${index + 1}`, { error });
+          // Force close if graceful close fails
+          await worker.close(true);
+        }
+      });
+
+      await Promise.allSettled(workerClosePromises);
+      this.logger.debug('All workers closed');
+
+      // Step 2: Close queue and events with timeout protection
+      this.logger.debug('Closing queue and events...');
+      await Promise.allSettled([
+        Promise.race([
+          this.queue.close(),
+          new Promise((_, reject) => 
+            setTimeout(() => reject(new Error('Queue close timeout')), 3000)
+          )
+        ]).catch(error => this.logger.error('Queue close error', { error })),
+        
+        Promise.race([
+          this.queueEvents.close(),
+          new Promise((_, reject) => 
+            setTimeout(() => reject(new Error('QueueEvents close timeout')), 3000)
+          )
+        ]).catch(error => this.logger.error('QueueEvents close error', { error }))
+      ]);
+
+      this.logger.info('Queue service shutdown completed successfully');
+    } catch (error) {
+      this.logger.error('Error during queue service shutdown', { error });
+      // Force close everything as last resort
+      try {
+        await Promise.allSettled([
+          ...this.workers.map(worker => worker.close(true)),
+          this.queue.close(),
+          this.queueEvents.close()
+        ]);
+      } catch (forceCloseError) {
+        this.logger.error('Force close also failed', { error: forceCloseError });
+      }
+      throw error;
+    }
   }
 }