lib/src/services/job-queue.js

"use strict";
/**
 * Asynchronous Job Queue for Webhook Processing
 *
 * GitHub webhooks timeout after 10 seconds. Long-running PR reviews
 * can exceed this. This queue acknowledges webhooks immediately and
 * processes them asynchronously.
 *
 * TODO: For production with horizontal scaling, replace with Redis-backed queue (Bull/BullMQ)
 */
var __importDefault = (this && this.__importDefault) || function (mod) {
    return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.jobQueue = void 0;
const pino_1 = __importDefault(require("pino"));
const logger = (0, pino_1.default)({ level: process.env.LOG_LEVEL || 'info' });
const defaultConfig = {
    maxConcurrent: 3,
    pollIntervalMs: 1000,
    maxAttempts: 3
};
class JobQueue {
    jobs = new Map();
    handlers = new Map();
    processing = new Set();
    config;
    isRunning = false;
    intervalId;
    constructor(config = {}) {
        this.config = { ...defaultConfig, ...config };
    }
    /**
     * Register a job handler for a specific job type
     */
    registerHandler(type, handler) {
        this.handlers.set(type, handler);
        logger.info({ type }, 'Registered job handler');
    }
    /**
     * Add a job to the queue
     */
    async add(type, data, maxAttempts) {
        const id = `${type}_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
        const job = {
            id,
            type,
            data,
            attempts: 0,
            maxAttempts: maxAttempts || this.config.maxAttempts,
            createdAt: Date.now(),
            status: 'pending'
        };
        this.jobs.set(id, job);
        logger.info({ jobId: id, type }, 'Job added to queue');
        // Start processing if not already running
        if (!this.isRunning) {
            this.start();
        }
        return id;
    }
    /**
     * Get job status
     */
    getStatus(jobId) {
        return this.jobs.get(jobId);
    }
    /**
     * Get queue metrics
     */
    getMetrics() {
        const jobs = Array.from(this.jobs.values());
        return {
            total: jobs.length,
            pending: jobs.filter(j => j.status === 'pending').length,
            processing: jobs.filter(j => j.status === 'processing').length,
            completed: jobs.filter(j => j.status === 'completed').length,
            failed: jobs.filter(j => j.status === 'failed').length
        };
    }
    /**
     * Start the job processor
     */
    start() {
        if (this.isRunning)
            return;
        this.isRunning = true;
        logger.info('Job queue started');
        this.intervalId = setInterval(() => {
            this.processJobs();
        }, this.config.pollIntervalMs);
    }
    /**
     * Stop the job processor
     */
    stop() {
        this.isRunning = false;
        if (this.intervalId) {
            clearInterval(this.intervalId);
            this.intervalId = undefined;
        }
        logger.info('Job queue stopped');
    }
    /**
     * Process pending jobs
     */
    async processJobs() {
        if (this.processing.size >= this.config.maxConcurrent) {
            return; // At capacity
        }
        const pendingJobs = Array.from(this.jobs.values())
            .filter(j => j.status === 'pending')
            .sort((a, b) => a.createdAt - b.createdAt); // FIFO
        for (const job of pendingJobs) {
            if (this.processing.size >= this.config.maxConcurrent) {
                break;
            }
            this.processJob(job);
        }
    }
    /**
     * Process a single job
     */
    async processJob(job) {
        const handler = this.handlers.get(job.type);
        if (!handler) {
            logger.error({ jobId: job.id, type: job.type }, 'No handler registered for job type');
            job.status = 'failed';
            job.error = 'No handler registered';
            return;
        }
        this.processing.add(job.id);
        job.status = 'processing';
        job.attempts++;
        logger.info({ jobId: job.id, type: job.type, attempt: job.attempts }, 'Processing job');
        try {
            await handler(job);
            job.status = 'completed';
            logger.info({ jobId: job.id }, 'Job completed successfully');
        }
        catch (error) {
            logger.error({ jobId: job.id, error: error.message }, 'Job failed');
            if (job.attempts < job.maxAttempts) {
                job.status = 'pending'; // Retry
                job.error = error.message;
                logger.info({ jobId: job.id, attemptsLeft: job.maxAttempts - job.attempts }, 'Job scheduled for retry');
            }
            else {
                job.status = 'failed';
                job.error = error.message;
                logger.error({ jobId: job.id }, 'Job failed permanently after max attempts');
            }
        }
        finally {
            this.processing.delete(job.id);
        }
    }
    /**
     * Cleanup old completed jobs
     */
    cleanup(maxAgeMs = 24 * 60 * 60 * 1000) {
        const now = Date.now();
        let cleaned = 0;
        for (const [id, job] of this.jobs.entries()) {
            if ((job.status === 'completed' || job.status === 'failed') &&
                now - job.createdAt > maxAgeMs) {
                this.jobs.delete(id);
                cleaned++;
            }
        }
        if (cleaned > 0) {
            logger.info({ cleaned }, 'Cleaned up old jobs');
        }
    }
}
// Singleton instance
exports.jobQueue = new JobQueue();
// Auto-cleanup every hour
setInterval(() => {
    exports.jobQueue.cleanup();
}, 60 * 60 * 1000);
//# sourceMappingURL=job-queue.js.map