/** * 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) */ import pino from 'pino' const logger = pino({ level: process.env.LOG_LEVEL || 'info' }) export interface Job { id: string type: string data: T attempts: number maxAttempts: number createdAt: number status: 'pending' | 'processing' | 'completed' | 'failed' error?: string } type JobHandler = (job: Job) => Promise interface JobQueueConfig { maxConcurrent: number pollIntervalMs: number maxAttempts: number } const defaultConfig: JobQueueConfig = { maxConcurrent: 3, pollIntervalMs: 1000, maxAttempts: 3 } class JobQueue { private jobs: Map = new Map() private handlers: Map> = new Map() private processing: Set = new Set() private config: JobQueueConfig private isRunning = false private intervalId?: NodeJS.Timeout constructor(config: Partial = {}) { this.config = { ...defaultConfig, ...config } } /** * Register a job handler for a specific job type */ registerHandler(type: string, handler: JobHandler): void { this.handlers.set(type, handler) logger.info({ type }, 'Registered job handler') } /** * Add a job to the queue */ async add(type: string, data: T, maxAttempts?: number): Promise { const id = `${type}_${Date.now()}_${Math.random().toString(36).substr(2, 9)}` const job: 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: string): Job | undefined { 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(): void { 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(): void { this.isRunning = false if (this.intervalId) { clearInterval(this.intervalId) this.intervalId = undefined } logger.info('Job queue stopped') } /** * Process pending jobs */ private async processJobs(): Promise { 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 */ private async processJob(job: Job): Promise { 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: any) { 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: number = 24 * 60 * 60 * 1000): void { 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 export const jobQueue = new JobQueue() // Auto-cleanup every hour setInterval(() => { jobQueue.cleanup() }, 60 * 60 * 1000)