fixed data-service apperently

apps/data-service/src/index.ts

@@ -10,7 +10,7 @@ import { initializeServiceConfig } from '@stock-bot/config';
 import { getLogger, setLoggerConfig, shutdownLoggers } from '@stock-bot/logger';
 import { createAndConnectMongoDBClient, MongoDBClient } from '@stock-bot/mongodb-client';
 import { createAndConnectPostgreSQLClient, PostgreSQLClient } from '@stock-bot/postgres-client';
-import { getQueue, initializeQueueSystem, shutdownAllQueues } from '@stock-bot/queue';
+import { QueueManager, handlerRegistry, type QueueManagerConfig } from '@stock-bot/queue';
 import { Shutdown } from '@stock-bot/shutdown';
 // Local imports
 import { exchangeRoutes, healthRoutes, queueRoutes } from './routes';
@@ -50,7 +50,7 @@ const PORT = serviceConfig.port;
 let server: ReturnType<typeof Bun.serve> | null = null;
 let postgresClient: PostgreSQLClient | null = null;
 let mongoClient: MongoDBClient | null = null;
-// Queue system will be initialized globally
+let queueManager: QueueManager | null = null;
 
 // Initialize shutdown manager
 const shutdown = Shutdown.getInstance({ timeout: 15000 });
@@ -100,14 +100,37 @@ async function initializeServices() {
 
     // Initialize queue system
     logger.info('Initializing queue system...');
-    await initializeQueueSystem({
+    const queueManagerConfig: QueueManagerConfig = {
       redis: queueConfig.redis,
-      defaultJobOptions: queueConfig.defaultJobOptions,
-      workers: 5,
-      concurrency: 20,
-    });
+      defaultQueueOptions: {
+        defaultJobOptions: queueConfig.defaultJobOptions,
+        workers: 5,
+        concurrency: 20,
+        enableMetrics: true,
+        enableDLQ: true,
+      },
+      enableScheduledJobs: true,
+    };
+    
+    queueManager = QueueManager.getInstance(queueManagerConfig);
+    await queueManager.initialize();
     logger.info('Queue system initialized');
 
+    // Initialize providers (register handlers and scheduled jobs)
+    logger.info('Initializing data providers...');
+    const { initializeExchangeSyncProvider } = await import('./providers/exchange-sync.provider');
+    const { initializeIBProvider } = await import('./providers/ib.provider');
+    const { initializeProxyProvider } = await import('./providers/proxy.provider');
+    const { initializeQMProvider } = await import('./providers/qm.provider');
+    const { initializeWebShareProvider } = await import('./providers/webshare.provider');
+    
+    initializeExchangeSyncProvider();
+    initializeIBProvider();
+    initializeProxyProvider();
+    initializeQMProvider();
+    initializeWebShareProvider();
+    logger.info('Data providers initialized');
+
     logger.info('All services initialized successfully');
   } catch (error) {
     logger.error('Failed to initialize services', { error });
@@ -144,7 +167,9 @@ shutdown.onShutdown(async () => {
 shutdown.onShutdown(async () => {
   logger.info('Shutting down queue system...');
   try {
-    await shutdownAllQueues();
+    if (queueManager) {
+      await queueManager.shutdown();
+    }
     logger.info('Queue system shut down');
   } catch (error) {
     logger.error('Error shutting down queue system', { error });
@@ -183,5 +208,5 @@ startServer().catch(error => {
 
 logger.info('Data service startup initiated');
 
-// Export queue functions for providers
-export { getQueue };
+// Export queue manager for providers
+export { queueManager };

apps/data-service/src/providers/exchange-sync.provider.ts

@@ -3,15 +3,15 @@
  */
 import { getLogger } from '@stock-bot/logger';
 import type { MasterExchange } from '@stock-bot/mongodb-client';
-import type { ProviderConfigWithSchedule } from '@stock-bot/queue';
-import { providerRegistry } from '@stock-bot/queue';
+import type { HandlerConfigWithSchedule } from '@stock-bot/queue';
+import { handlerRegistry } from '@stock-bot/queue';
 
 const logger = getLogger('exchange-sync');
 
 export function initializeExchangeSyncProvider() {
   logger.info('Registering exchange sync provider...');
 
-  const exchangeSyncConfig: ProviderConfigWithSchedule = {
+  const exchangeSyncConfig: HandlerConfigWithSchedule = {
     name: 'exchange-sync',
 
     operations: {
@@ -40,7 +40,7 @@ export function initializeExchangeSyncProvider() {
     ],
   };
 
-  providerRegistry.registerWithSchedule(exchangeSyncConfig);
+  handlerRegistry.registerWithSchedule(exchangeSyncConfig);
   logger.info('Exchange sync provider registered successfully');
 }
 

apps/data-service/src/providers/ib.provider.ts

@@ -2,8 +2,8 @@
  * Interactive Brokers Provider for new queue system
  */
 import { getLogger } from '@stock-bot/logger';
-import type { ProviderConfigWithSchedule } from '@stock-bot/queue';
-import { providerRegistry } from '@stock-bot/queue';
+import type { HandlerConfigWithSchedule } from '@stock-bot/queue';
+import { handlerRegistry } from '@stock-bot/queue';
 
 const logger = getLogger('ib-provider');
 
@@ -11,7 +11,7 @@ const logger = getLogger('ib-provider');
 export function initializeIBProvider() {
   logger.info('Registering IB provider with scheduled jobs...');
 
-  const ibProviderConfig: ProviderConfigWithSchedule = {
+  const ibProviderConfig: HandlerConfigWithSchedule = {
     name: 'ib',
     operations: {
       'fetch-session': async _payload => {
@@ -77,6 +77,6 @@ export function initializeIBProvider() {
     ],
   };
 
-  providerRegistry.registerWithSchedule(ibProviderConfig);
+  handlerRegistry.registerWithSchedule(ibProviderConfig);
   logger.info('IB provider registered successfully with scheduled jobs');
 }

apps/data-service/src/providers/proxy.provider.ts

@@ -3,8 +3,8 @@
  */
 import { ProxyInfo } from '@stock-bot/http';
 import { getLogger } from '@stock-bot/logger';
-import type { ProviderConfigWithSchedule } from '@stock-bot/queue';
-import { providerRegistry } from '@stock-bot/queue';
+import type { HandlerConfigWithSchedule } from '@stock-bot/queue';
+import { handlerRegistry } from '@stock-bot/queue';
 
 const logger = getLogger('proxy-provider');
 
@@ -12,7 +12,7 @@ const logger = getLogger('proxy-provider');
 export function initializeProxyProvider() {
   logger.info('Registering proxy provider with scheduled jobs...');
 
-  const proxyProviderConfig: ProviderConfigWithSchedule = {
+  const proxyProviderConfig: HandlerConfigWithSchedule = {
     name: 'proxy',
 
     operations: {
@@ -20,7 +20,7 @@ export function initializeProxyProvider() {
         // Fetch proxies from all configured sources
         logger.info('Processing fetch proxies from sources request');
         const { fetchProxiesFromSources } = await import('./proxy.tasks');
-        const { processItems, queueManager } = await import('../index');
+        const { processItems } = await import('@stock-bot/queue');
 
         // Fetch all proxies from sources
         const proxies = await fetchProxiesFromSources();
@@ -32,8 +32,8 @@ export function initializeProxyProvider() {
         }
 
         // Batch process the proxies through check-proxy operation
-        const batchResult = await processItems(proxies, queueManager, {
-          provider: 'proxy',
+        const batchResult = await processItems(proxies, 'proxy', {
+          handler: 'proxy',
           operation: 'check-proxy',
           totalDelayHours: 0.083, // 5 minutes (5/60 hours)
           batchSize: 50, // Process 50 proxies per batch
@@ -83,6 +83,6 @@ export function initializeProxyProvider() {
     ],
   };
 
-  providerRegistry.registerWithSchedule(proxyProviderConfig);
+  handlerRegistry.registerWithSchedule(proxyProviderConfig);
   logger.info('Proxy provider registered successfully with scheduled jobs');
 }

apps/data-service/src/providers/proxy.tasks.ts

@@ -310,9 +310,12 @@ async function updateProxyInCache(proxy: ProxyInfo, isWorking: boolean): Promise
 // Individual task functions
 export async function queueProxyFetch(): Promise<string> {
   const { queueManager } = await import('../index');
-  const job = await queueManager.add('proxy-fetch', {
-    type: 'proxy-fetch',
-    provider: 'proxy-service',
+  if (!queueManager) {
+    throw new Error('Queue manager not initialized');
+  }
+  const queue = queueManager.getQueue('proxy');
+  const job = await queue.add('proxy-fetch', {
+    handler: 'proxy',
     operation: 'fetch-and-check',
     payload: {},
     priority: 5,
@@ -325,9 +328,12 @@ export async function queueProxyFetch(): Promise<string> {
 
 export async function queueProxyCheck(proxies: ProxyInfo[]): Promise<string> {
   const { queueManager } = await import('../index');
-  const job = await queueManager.add('proxy-check', {
-    type: 'proxy-check',
-    provider: 'proxy-service',
+  if (!queueManager) {
+    throw new Error('Queue manager not initialized');
+  }
+  const queue = queueManager.getQueue('proxy');
+  const job = await queue.add('proxy-check', {
+    handler: 'proxy',
     operation: 'check-specific',
     payload: { proxies },
     priority: 3,

apps/data-service/src/providers/qm.provider.ts

@@ -1,5 +1,5 @@
 import { getLogger } from '@stock-bot/logger';
-import { providerRegistry, type ProviderConfigWithSchedule } from '@stock-bot/queue';
+import { handlerRegistry, type HandlerConfigWithSchedule } from '@stock-bot/queue';
 import type { SymbolSpiderJob } from './qm.tasks';
 
 const logger = getLogger('qm-provider');
@@ -8,7 +8,7 @@ const logger = getLogger('qm-provider');
 export function initializeQMProvider() {
   logger.info('Registering IB provider with scheduled jobs...');
 
-  const qmProviderConfig: ProviderConfigWithSchedule = {
+  const qmProviderConfig: HandlerConfigWithSchedule = {
     name: 'qm',
     operations: {
       'create-sessions': async () => {
@@ -77,6 +77,6 @@ export function initializeQMProvider() {
     ],
   };
 
-  providerRegistry.registerWithSchedule(qmProviderConfig);
+  handlerRegistry.registerWithSchedule(qmProviderConfig);
   logger.info('IB provider registered successfully with scheduled jobs');
 }

apps/data-service/src/providers/qm.tasks.ts

@@ -172,6 +172,10 @@ async function createAlphabetJobs(
 ): Promise<{ success: boolean; symbolsFound: number; jobsCreated: number }> {
   try {
     const { queueManager } = await import('../index');
+    if (!queueManager) {
+      throw new Error('Queue manager not initialized');
+    }
+    const queue = queueManager.getQueue('qm');
     let jobsCreated = 0;
 
     // Create jobs for A-Z
@@ -185,10 +189,10 @@ async function createAlphabetJobs(
         maxDepth,
       };
 
-      await queueManager.add(
-        'qm',
+      await queue.add(
+        'spider-symbol-search',
         {
-          provider: 'qm',
+          handler: 'qm',
           operation: 'spider-symbol-search',
           payload: job,
         },
@@ -239,6 +243,10 @@ async function searchAndSpawnJobs(
     // If we have 50+ symbols and haven't reached max depth, spawn sub-jobs
     if (symbolCount >= 50 && depth < maxDepth) {
       const { queueManager } = await import('../index');
+      if (!queueManager) {
+        throw new Error('Queue manager not initialized');
+      }
+      const queue = queueManager.getQueue('qm');
 
       logger.info(`Spawning sub-jobs for prefix "${prefix}" (${symbolCount} >= 50 symbols)`);
 
@@ -254,10 +262,10 @@ async function searchAndSpawnJobs(
           maxDepth,
         };
 
-        await queueManager.add(
-          'qm',
+        await queue.add(
+          'spider-symbol-search',
           {
-            provider: 'qm',
+            handler: 'qm',
             operation: 'spider-symbol-search',
             payload: job,
           },

apps/data-service/src/providers/webshare.provider.ts

@@ -2,8 +2,8 @@
  * WebShare Provider for proxy management
  */
 import { getLogger } from '@stock-bot/logger';
-import type { ProviderConfigWithSchedule } from '@stock-bot/queue';
-import { providerRegistry } from '@stock-bot/queue';
+import type { HandlerConfigWithSchedule } from '@stock-bot/queue';
+import { handlerRegistry } from '@stock-bot/queue';
 
 const logger = getLogger('webshare-provider');
 
@@ -26,7 +26,7 @@ export function getProxy(): string | null {
 export function initializeWebShareProvider() {
   logger.info('Registering WebShare provider with scheduled jobs...');
 
-  const webShareProviderConfig: ProviderConfigWithSchedule = {
+  const webShareProviderConfig: HandlerConfigWithSchedule = {
     name: 'webshare',
 
     operations: {
@@ -83,7 +83,7 @@ export function initializeWebShareProvider() {
   };
 
   // Register the provider
-  providerRegistry.registerWithSchedule(webShareProviderConfig);
+  handlerRegistry.registerWithSchedule(webShareProviderConfig);
 
   logger.info('WebShare provider registered successfully');
 }

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

@@ -1,5 +1,6 @@
 import { Hono } from 'hono';
 import { getLogger } from '@stock-bot/logger';
+import { QueueManager } from '@stock-bot/queue';
 
 const logger = getLogger('queue-routes');
 const queue = new Hono();
@@ -7,16 +8,13 @@ const queue = new Hono();
 // Queue status endpoint
 queue.get('/status', async c => {
   try {
-    // TODO: Implement queue management
+    const queueManager = QueueManager.getInstance();
+    const globalStats = await queueManager.getGlobalStats();
+    
     return c.json({
       status: 'success',
-      data: {
-        active: 0,
-        waiting: 0,
-        completed: 0,
-        failed: 0
-      },
-      message: 'Queue management will be implemented'
+      data: globalStats,
+      message: 'Queue status retrieved successfully'
     });
   } catch (error) {
     logger.error('Failed to get queue status', { error });