moved handlers out of queue will be reused with event-bus

apps/data-ingestion/package.json

@@ -15,6 +15,7 @@
     "@stock-bot/cache": "*",
     "@stock-bot/config": "*",
     "@stock-bot/di": "*",
+    "@stock-bot/handlers": "*",
     "@stock-bot/logger": "*",
     "@stock-bot/mongodb": "*",
     "@stock-bot/postgres": "*",

apps/data-ingestion/src/handlers/qm/qm.handler.ts

@@ -1,48 +1,110 @@
-import { getLogger } from '@stock-bot/logger';
 import {
-  createJobHandler,
+  BaseHandler,
+  ScheduledHandler,
+  Handler,
+  Operation,
+  QueueSchedule,
   handlerRegistry,
+  createJobHandler,
+  type ExecutionContext,
   type HandlerConfigWithSchedule
-} from '@stock-bot/queue';
-import type { ServiceContainer } from '@stock-bot/connection-factory';
+} from '@stock-bot/handlers';
+import type { ServiceContainer } from '@stock-bot/di';
 import type { SymbolSpiderJob } from './shared/types';
 
-const handlerLogger = getLogger('qm-handler');
+@Handler('qm')
+export class QMHandler extends ScheduledHandler {
+  constructor(container: ServiceContainer) {
+    super(container);
+  }
+
+  async execute(operation: string, input: unknown, context: ExecutionContext): Promise<unknown> {
+    switch (operation) {
+      case 'create-sessions':
+        return await this.createSessions(input, context);
+      case 'search-symbols':
+        return await this.searchSymbols(input, context);
+      case 'spider-symbol-search':
+        return await this.spiderSymbolSearch(input as SymbolSpiderJob, context);
+      default:
+        throw new Error(`Unknown operation: ${operation}`);
+    }
+  }
+
+  @Operation('create-sessions')
+  @QueueSchedule('0 */15 * * *', { 
+    priority: 7, 
+    immediately: true, 
+    description: 'Create and maintain QM sessions' 
+  })
+  async createSessions(input: unknown, context: ExecutionContext): Promise<unknown> {
+    const { createSessions } = await import('./operations/session.operations');
+    await createSessions(context.serviceContainer);
+    return { success: true, message: 'QM sessions created successfully' };
+  }
+
+  @Operation('search-symbols')
+  async searchSymbols(input: unknown, context: ExecutionContext): Promise<unknown> {
+    const { fetchSymbols } = await import('./operations/symbols.operations');
+    const symbols = await fetchSymbols(context.serviceContainer);
+
+    if (symbols && symbols.length > 0) {
+      return {
+        success: true,
+        message: 'QM symbol search completed successfully',
+        count: symbols.length,
+        symbols: symbols.slice(0, 10), // Return first 10 symbols as sample
+      };
+    } else {
+      return {
+        success: false,
+        message: 'No symbols found',
+        count: 0,
+      };
+    }
+  }
+
+  @Operation('spider-symbol-search')
+  @QueueSchedule('0 0 * * 0', { 
+    priority: 10, 
+    immediately: true, 
+    description: 'Comprehensive symbol search using QM API' 
+  })
+  async spiderSymbolSearch(payload: SymbolSpiderJob, context: ExecutionContext): Promise<unknown> {
+    const { spiderSymbolSearch } = await import('./operations/spider.operations');
+    return await spiderSymbolSearch(payload, context.serviceContainer);
+  }
+}
 
 // Initialize and register the QM provider
 export function initializeQMProvider(container: ServiceContainer) {
-  handlerLogger.debug('Registering QM provider with scheduled jobs...');
-
+  // Create handler instance
+  const handler = new QMHandler(container);
+  
+  // Register with legacy format for now
   const qmProviderConfig: HandlerConfigWithSchedule = {
     name: 'qm',
     operations: {
-      'create-sessions': createJobHandler(async () => {
-        const { createSessions } = await import('./operations/session.operations');
-        await createSessions(container);
-        return { success: true, message: 'QM sessions created successfully' };
+      'create-sessions': createJobHandler(async (payload) => {
+        return await handler.execute('create-sessions', payload, {
+          type: 'queue',
+          serviceContainer: container,
+          metadata: { source: 'queue', timestamp: Date.now() }
+        });
       }),
-      'search-symbols': createJobHandler(async () => {
-        const { fetchSymbols } = await import('./operations/symbols.operations');
-        const symbols = await fetchSymbols(container);
-
-        if (symbols && symbols.length > 0) {
-          return {
-            success: true,
-            message: 'QM symbol search completed successfully',
-            count: symbols.length,
-            symbols: symbols.slice(0, 10), // Return first 10 symbols as sample
-          };
-        } else {
-          return {
-            success: false,
-            message: 'No symbols found',
-            count: 0,
-          };
-        }
+      'search-symbols': createJobHandler(async (payload) => {
+        return await handler.execute('search-symbols', payload, {
+          type: 'queue',
+          serviceContainer: container,
+          metadata: { source: 'queue', timestamp: Date.now() }
+        });
       }),
       'spider-symbol-search': createJobHandler(async (payload: SymbolSpiderJob) => {
-        const { spiderSymbolSearch } = await import('./operations/spider.operations');
-        return await spiderSymbolSearch(payload, container);
+        return await handler.execute('spider-symbol-search', payload, {
+          type: 'queue',
+          serviceContainer: container,
+          metadata: { source: 'queue', timestamp: Date.now() }
+        });
       }),
     },
 
@@ -52,7 +114,7 @@ export function initializeQMProvider(container: ServiceContainer) {
         operation: 'create-sessions',
         cronPattern: '0 */15 * * *', // Every 15 minutes
         priority: 7,
-        immediately: true, // Don't run on startup to avoid blocking
+        immediately: true,
         description: 'Create and maintain QM sessions',
       },
       {
@@ -66,12 +128,12 @@ export function initializeQMProvider(container: ServiceContainer) {
         },
         cronPattern: '0 0 * * 0', // Every Sunday at midnight
         priority: 10,
-        immediately: true, // Don't run on startup - this is a heavy operation
+        immediately: true,
         description: 'Comprehensive symbol search using QM API',
       },
     ],
   };
 
   handlerRegistry.registerWithSchedule(qmProviderConfig);
-  handlerLogger.debug('QM provider registered successfully with scheduled jobs');
+  handler.logger.debug('QM provider registered successfully with scheduled jobs');
 }

apps/data-ingestion/tsconfig.json

@@ -5,6 +5,7 @@
     { "path": "../../libs/core/config" },
     { "path": "../../libs/core/logger" },
     { "path": "../../libs/core/di" },
+    { "path": "../../libs/core/handlers" },
     { "path": "../../libs/data/cache" },
     { "path": "../../libs/data/mongodb" },
     { "path": "../../libs/data/postgres" },