queue.service.ts

/**
 * Queue Service Implementation using BullMQ and Redis
 */

import { Queue, Worker, Job } from 'bullmq';
import { IQueueService } from '../interfaces/services';
import { logger } from '../utils/logger';
import { config } from '../config';
import IORedis from 'ioredis';

/**
 * Queue Service using BullMQ
 */
export class QueueService implements IQueueService {
  private queues: Map<string, Queue>;
  private workers: Map<string, Worker>;
  private connection: IORedis;

  constructor() {
    this.queues = new Map();
    this.workers = new Map();
    
    // Create Redis connection
    this.connection = new IORedis({
      host: config.redis.host,
      port: config.redis.port,
      password: config.redis.password,
      maxRetriesPerRequest: null,
    });

    logger.info('Queue service initialized', {
      host: config.redis.host,
      port: config.redis.port,
    });
  }

  /**
   * Add a job to the queue
   */
  async addJob(
    jobName: string,
    data: Record<string, unknown>,
    options?: Record<string, unknown>
  ): Promise<string> {
    try {
      const queue = this.getOrCreateQueue(jobName);
      
      const job = await queue.add(jobName, data, {
        attempts: config.processing.maxRetries,
        backoff: {
          type: 'exponential',
          delay: config.processing.retryDelayBase,
        },
        removeOnComplete: {
          count: 100, // Keep last 100 completed jobs
        },
        removeOnFail: {
          count: 500, // Keep last 500 failed jobs
        },
        ...options,
      });

      logger.debug('Job added to queue', {
        jobId: job.id,
        jobName,
        data,
      });

      return job.id as string;
    } catch (error) {
      logger.error('Failed to add job to queue', { error, jobName, data });
      throw new Error(`Failed to add job to queue: ${jobName}`);
    }
  }

  /**
   * Process jobs from the queue
   */
  async processJobs(
    jobName: string,
    processor: (job: Job) => Promise<void>
  ): Promise<void> {
    try {
      if (this.workers.has(jobName)) {
        logger.warn('Worker already exists for job', { jobName });
        return;
      }

      const worker = new Worker(
        jobName,
        async (job: Job) => {
          logger.info('Processing job', {
            jobId: job.id,
            jobName: job.name,
            data: job.data,
          });

          try {
            await processor(job);
            logger.info('Job completed successfully', {
              jobId: job.id,
              jobName: job.name,
            });
          } catch (error) {
            logger.error('Job processing failed', {
              jobId: job.id,
              jobName: job.name,
              error,
            });
            throw error;
          }
        },
        {
          connection: this.connection,
          concurrency: config.processing.workerConcurrency,
          settings: {
            backoffStrategy: (attemptsMade: number) => {
              // Exponential backoff with max delay cap
              const delay = Math.min(
                config.processing.retryDelayBase * Math.pow(2, attemptsMade - 1),
                config.processing.retryDelayMax
              );
              return delay;
            },
          },
        }
      );

      // Handle worker events
      worker.on('completed', (job) => {
        logger.debug('Worker completed job', {
          jobId: job.id,
          jobName: job.name,
        });
      });

      worker.on('failed', (job, err) => {
        logger.error('Worker failed job', {
          jobId: job?.id,
          jobName: job?.name,
          error: err,
        });
      });

      this.workers.set(jobName, worker);
      logger.info('Worker created and started', { jobName });
    } catch (error) {
      logger.error('Failed to create worker', { error, jobName });
      throw new Error(`Failed to create worker for: ${jobName}`);
    }
  }

  /**
   * Get job status
   */
  async getJobStatus(jobId: string): Promise<string> {
    try {
      // Try to find the job in all queues
      for (const [queueName, queue] of this.queues) {
        try {
          const job = await queue.getJob(jobId);
          if (job) {
            const state = await job.getState();
            logger.debug('Job status retrieved', {
              jobId,
              queueName,
              state,
            });
            return state;
          }
        } catch (error) {
          // Continue searching in other queues
          continue;
        }
      }

      logger.warn('Job not found in any queue', { jobId });
      return 'not_found';
    } catch (error) {
      logger.error('Failed to get job status', { error, jobId });
      throw new Error('Failed to get job status');
    }
  }

  /**
   * Remove a job from the queue
   */
  async removeJob(jobId: string): Promise<void> {
    try {
      // Try to remove from all queues
      for (const [queueName, queue] of this.queues) {
        try {
          const job = await queue.getJob(jobId);
          if (job) {
            await job.remove();
            logger.info('Job removed from queue', {
              jobId,
              queueName,
            });
            return;
          }
        } catch (error) {
          // Continue searching in other queues
          continue;
        }
      }

      logger.warn('Job not found for removal', { jobId });
    } catch (error) {
      logger.error('Failed to remove job', { error, jobId });
      throw new Error('Failed to remove job');
    }
  }

  /**
   * Get or create a queue for a job type
   */
  private getOrCreateQueue(jobName: string): Queue {
    if (!this.queues.has(jobName)) {
      const queue = new Queue(jobName, {
        connection: this.connection,
        defaultJobOptions: {
          removeOnComplete: {
            count: 100,
          },
          removeOnFail: {
            count: 500,
          },
        },
      });

      this.queues.set(jobName, queue);
      logger.debug('Created new queue', { jobName });
    }

    return this.queues.get(jobName)!;
  }

  /**
   * Close all queues and workers
   */
  async close(): Promise<void> {
    try {
      // Close all workers
      for (const [name, worker] of this.workers) {
        await worker.close();
        logger.debug('Worker closed', { name });
      }

      // Close all queues
      for (const [name, queue] of this.queues) {
        await queue.close();
        logger.debug('Queue closed', { name });
      }

      // Close Redis connection
      await this.connection.quit();

      logger.info('Queue service closed');
    } catch (error) {
      logger.error('Error closing queue service', { error });
      throw error;
    }
  }

  /**
   * Get queue metrics
   */
  async getQueueMetrics(jobName: string): Promise<{
    waiting: number;
    active: number;
    completed: number;
    failed: number;
  }> {
    try {
      const queue = this.getOrCreateQueue(jobName);
      
      const [waiting, active, completed, failed] = await Promise.all([
        queue.getWaitingCount(),
        queue.getActiveCount(),
        queue.getCompletedCount(),
        queue.getFailedCount(),
      ]);

      return { waiting, active, completed, failed };
    } catch (error) {
      logger.error('Failed to get queue metrics', { error, jobName });
      throw new Error('Failed to get queue metrics');
    }
  }

  /**
   * Ping Redis to check connectivity
   */
  async ping(): Promise<boolean> {
    try {
      const result = await this.connection.ping();
      return result === 'PONG';
    } catch (error) {
      logger.error('Redis ping failed', { error });
      return false;
    }
  }
}