import { describe, it, expect, beforeEach, afterEach } from 'vitest'; import { Queue, Worker, Job } from 'bullmq'; import { JobPriority } from '../core/src/types/priority.types'; import { isPeakHour, shouldDeferJob, calculatePeakDelay } from '../core/src/utils/peak-hours'; import { getRedisConnection, waitForJobs, waitFor } from './setup'; /** * E2E Integration Tests for Queue Package * * These tests verify: * - Job enqueuing and processing * - Priority handling * - Retry behavior * - Peak-hour deferral * - Bulk operations */ describe('Queue Integration - Job Processing', () => { let queue: Queue; let worker: Worker; const processedJobs: Array<{ id: string; data: any }> = []; beforeEach(async () => { processedJobs.length = 0; queue = new Queue('test-queue', { connection: getRedisConnection(), }); worker = new Worker( 'test-queue', async (job: Job) => { processedJobs.push({ id: job.id!, data: job.data }); return { success: true }; }, { connection: getRedisConnection(), }, ); }); afterEach(async () => { await worker.close(); await queue.close(); }); it('should enqueue and process a simple job', async () => { const jobData = { message: 'Hello, Queue!' }; const job = await queue.add('test-job', jobData); expect(job.id).toBeDefined(); await waitForJobs(queue, 'completed', 1); expect(processedJobs).toHaveLength(1); expect(processedJobs[0].data).toMatchObject(jobData); }); it('should process multiple jobs in order', async () => { const jobs = [ { message: 'Job 1' }, { message: 'Job 2' }, { message: 'Job 3' }, ]; for (const data of jobs) { await queue.add('test-job', data); } await waitForJobs(queue, 'completed', 3); expect(processedJobs).toHaveLength(3); expect(processedJobs.map((j) => j.data.message)).toEqual(['Job 1', 'Job 2', 'Job 3']); }); it('should handle job completion with return value', async () => { const job = await queue.add('test-job', { value: 42 }); await waitForJobs(queue, 'completed', 1); const completedJob = await queue.getJob(job.id!); expect(completedJob?.returnvalue).toEqual({ success: true }); }); }); describe('Queue Integration - Priority Handling', () => { let queue: Queue; let worker: Worker; const processedOrder: number[] = []; beforeEach(async () => { processedOrder.length = 0; queue = new Queue('priority-queue', { connection: getRedisConnection(), }); worker = new Worker( 'priority-queue', async (job: Job) => { processedOrder.push(job.data.order); await new Promise((resolve) => setTimeout(resolve, 50)); return { processed: true }; }, { connection: getRedisConnection(), concurrency: 1, // Process one at a time to verify order }, ); }); afterEach(async () => { await worker.close(); await queue.close(); }); it('should process jobs in priority order', async () => { // Add jobs with different priorities (lower number = higher priority) await queue.add('job', { order: 3 }, { priority: JobPriority.LOW }); await queue.add('job', { order: 1 }, { priority: JobPriority.URGENT }); await queue.add('job', { order: 2 }, { priority: JobPriority.HIGH }); await queue.add('job', { order: 4 }, { priority: JobPriority.NORMAL }); await waitForJobs(queue, 'completed', 4); // Should process in priority order: URGENT, HIGH, NORMAL, LOW expect(processedOrder).toEqual([1, 2, 4, 3]); }); it('should handle mixed priority jobs correctly', async () => { await queue.add('job', { order: 5 }, { priority: JobPriority.BATCH }); await queue.add('job', { order: 2 }, { priority: JobPriority.HIGH }); await queue.add('job', { order: 3 }, { priority: JobPriority.NORMAL }); await queue.add('job', { order: 1 }, { priority: JobPriority.URGENT }); await queue.add('job', { order: 4 }, { priority: JobPriority.LOW }); await waitForJobs(queue, 'completed', 5); expect(processedOrder).toEqual([1, 2, 3, 4, 5]); }); }); describe('Queue Integration - Retry Behavior', () => { let queue: Queue; let worker: Worker; let attemptCounts = new Map(); beforeEach(async () => { attemptCounts.clear(); queue = new Queue('retry-queue', { connection: getRedisConnection(), }); worker = new Worker( 'retry-queue', async (job: Job) => { const currentAttempts = attemptCounts.get(job.id!) ?? 0; attemptCounts.set(job.id!, currentAttempts + 1); if (job.data.shouldFail && currentAttempts < job.data.failUntilAttempt) { throw new Error('Intentional failure for retry test'); } return { attempts: currentAttempts + 1 }; }, { connection: getRedisConnection(), }, ); }); afterEach(async () => { await worker.close(); await queue.close(); }); it('should retry failed jobs with exponential backoff', async () => { const job = await queue.add( 'retry-job', { shouldFail: true, failUntilAttempt: 2 }, { attempts: 3, backoff: { type: 'exponential', delay: 100, }, }, ); await waitForJobs(queue, 'completed', 1, 10000); const completedJob = await queue.getJob(job.id!); expect(completedJob?.attemptsMade).toBe(3); expect(attemptCounts.get(job.id!)).toBe(3); }); it('should fail job after max attempts', async () => { const job = await queue.add( 'retry-job', { shouldFail: true, failUntilAttempt: 10 }, // Never succeed { attempts: 3, backoff: { type: 'fixed', delay: 50, }, }, ); await waitForJobs(queue, 'failed', 1, 10000); const failedJob = await queue.getJob(job.id!); expect(failedJob?.attemptsMade).toBe(3); expect(failedJob?.failedReason).toContain('Intentional failure'); }); it('should handle jobs that succeed on first retry', async () => { const job = await queue.add( 'retry-job', { shouldFail: true, failUntilAttempt: 1 }, { attempts: 3, backoff: { type: 'fixed', delay: 50, }, }, ); await waitForJobs(queue, 'completed', 1, 5000); const completedJob = await queue.getJob(job.id!); expect(completedJob?.attemptsMade).toBe(2); expect(attemptCounts.get(job.id!)).toBe(2); }); }); describe('Queue Integration - Peak Hour Deferral', () => { let queue: Queue; beforeEach(async () => { queue = new Queue('peak-queue', { connection: getRedisConnection(), }); }); afterEach(async () => { await queue.close(); }); it('should detect peak hours correctly', () => { // Test peak hour (6pm UTC = 18:00) const peakTime = new Date('2025-01-15T18:30:00Z'); // Wednesday expect(isPeakHour(peakTime)).toBe(true); // Test non-peak hour (2am UTC) const nonPeakTime = new Date('2025-01-15T02:00:00Z'); expect(isPeakHour(nonPeakTime)).toBe(false); // Test weekend (not peak even during peak hours) const weekend = new Date('2025-01-18T18:30:00Z'); // Saturday expect(isPeakHour(weekend)).toBe(false); }); it('should defer low-priority jobs during peak hours', () => { const peakTime = new Date('2025-01-15T18:30:00Z'); // Low priority should defer const lowPriorityResult = shouldDeferJob(JobPriority.LOW, peakTime); expect(lowPriorityResult.shouldDefer).toBe(true); expect(lowPriorityResult.delay).toBeGreaterThan(0); // Normal priority should defer const normalPriorityResult = shouldDeferJob(JobPriority.NORMAL, peakTime); expect(normalPriorityResult.shouldDefer).toBe(true); expect(normalPriorityResult.delay).toBeGreaterThan(0); }); it('should not defer high-priority jobs during peak hours', () => { const peakTime = new Date('2025-01-15T18:30:00Z'); // High priority should bypass const highPriorityResult = shouldDeferJob(JobPriority.HIGH, peakTime); expect(highPriorityResult.shouldDefer).toBe(false); expect(highPriorityResult.delay).toBe(0); // Urgent priority should bypass const urgentPriorityResult = shouldDeferJob(JobPriority.URGENT, peakTime); expect(urgentPriorityResult.shouldDefer).toBe(false); expect(urgentPriorityResult.delay).toBe(0); }); it('should calculate correct delay until peak ends', () => { // 6pm UTC, peak ends at 10pm UTC (4 hours) const peakTime = new Date('2025-01-15T18:00:00Z'); const delay = calculatePeakDelay(peakTime); // Should be approximately 4 hours (14400000 ms) expect(delay).toBeGreaterThan(14000000); expect(delay).toBeLessThan(14500000); }); it('should apply delay when adding job during peak hours', async () => { const peakTime = new Date('2025-01-15T18:30:00Z'); const { delay } = shouldDeferJob(JobPriority.NORMAL, peakTime); const job = await queue.add( 'deferred-job', { data: 'test' }, { priority: JobPriority.NORMAL, delay, }, ); // Job should be in delayed state const state = await job.getState(); expect(state).toBe('delayed'); const counts = await queue.getJobCounts(); expect(counts.delayed).toBe(1); expect(counts.waiting).toBe(0); }); }); describe('Queue Integration - Bulk Operations', () => { let queue: Queue; let worker: Worker; const processedJobs: string[] = []; beforeEach(async () => { processedJobs.length = 0; queue = new Queue('bulk-queue', { connection: getRedisConnection(), }); worker = new Worker( 'bulk-queue', async (job: Job) => { processedJobs.push(job.id!); return { processed: true }; }, { connection: getRedisConnection(), concurrency: 5, }, ); }); afterEach(async () => { await worker.close(); await queue.close(); }); it('should add bulk jobs efficiently', async () => { const bulkJobs = Array.from({ length: 100 }, (_, i) => ({ name: 'bulk-job', data: { index: i }, opts: {}, })); const startTime = Date.now(); const jobs = await queue.addBulk(bulkJobs); const duration = Date.now() - startTime; expect(jobs).toHaveLength(100); expect(duration).toBeLessThan(1000); // Should be fast await waitForJobs(queue, 'completed', 100, 30000); expect(processedJobs).toHaveLength(100); }); it('should handle bulk jobs with different priorities', async () => { const bulkJobs = [ ...Array.from({ length: 10 }, (_, i) => ({ name: 'low', data: { priority: 'low', index: i }, opts: { priority: JobPriority.LOW }, })), ...Array.from({ length: 10 }, (_, i) => ({ name: 'high', data: { priority: 'high', index: i }, opts: { priority: JobPriority.HIGH }, })), ...Array.from({ length: 10 }, (_, i) => ({ name: 'normal', data: { priority: 'normal', index: i }, opts: { priority: JobPriority.NORMAL }, })), ]; const jobs = await queue.addBulk(bulkJobs); expect(jobs).toHaveLength(30); await waitForJobs(queue, 'completed', 30, 15000); expect(processedJobs).toHaveLength(30); }); it('should handle bulk job failures gracefully', async () => { const bulkJobs = Array.from({ length: 20 }, (_, i) => ({ name: 'bulk-job', data: { index: i, shouldFail: i % 5 === 0 }, // Every 5th job fails opts: { attempts: 1, }, })); await queue.addBulk(bulkJobs); // Update worker to handle failures await worker.close(); worker = new Worker( 'bulk-queue', async (job: Job) => { processedJobs.push(job.id!); if (job.data.shouldFail) { throw new Error('Intentional bulk failure'); } return { processed: true }; }, { connection: getRedisConnection(), concurrency: 5, }, ); await waitFor( async () => { const counts = await queue.getJobCounts(); return counts.completed + counts.failed === 20; }, { timeout: 15000 }, ); const counts = await queue.getJobCounts(); expect(counts.completed).toBe(16); expect(counts.failed).toBe(4); }); }); describe('Queue Integration - Advanced Scenarios', () => { let queue: Queue; let worker: Worker; beforeEach(async () => { queue = new Queue('advanced-queue', { connection: getRedisConnection(), }); }); afterEach(async () => { if (worker) { await worker.close(); } await queue.close(); }); it('should handle job cancellation', async () => { const job = await queue.add('cancellable-job', { data: 'test' }, { delay: 5000 }); // Job should be delayed expect(await job.getState()).toBe('delayed'); // Remove the job await job.remove(); // Job should no longer exist const removedJob = await queue.getJob(job.id!); expect(removedJob).toBeNull(); }); it('should track job progress', async () => { const progressUpdates: number[] = []; worker = new Worker( 'advanced-queue', async (job: Job) => { for (let i = 0; i <= 100; i += 25) { await job.updateProgress(i); await new Promise((resolve) => setTimeout(resolve, 50)); } return { complete: true }; }, { connection: getRedisConnection(), }, ); const job = await queue.add('progress-job', { data: 'test' }); // Monitor progress const checkProgress = async () => { const currentJob = await queue.getJob(job.id!); if (currentJob && typeof currentJob.progress === 'number') { progressUpdates.push(currentJob.progress); } }; const interval = setInterval(checkProgress, 100); await waitForJobs(queue, 'completed', 1, 5000); clearInterval(interval); expect(progressUpdates.length).toBeGreaterThan(0); expect(progressUpdates).toContain(100); }); it('should handle concurrent job processing', async () => { const processedJobIds = new Set(); worker = new Worker( 'advanced-queue', async (job: Job) => { processedJobIds.add(job.id!); await new Promise((resolve) => setTimeout(resolve, 100)); return { id: job.id }; }, { connection: getRedisConnection(), concurrency: 10, // Process 10 jobs concurrently }, ); // Add 50 jobs const jobs = await Promise.all( Array.from({ length: 50 }, (_, i) => queue.add('concurrent-job', { index: i })), ); await waitForJobs(queue, 'completed', 50, 15000); expect(processedJobIds.size).toBe(50); expect(jobs.every((job) => processedJobIds.has(job.id!))).toBe(true); }); it('should respect rate limiting', async () => { const processTimestamps: number[] = []; queue = new Queue('rate-limited-queue', { connection: getRedisConnection(), limiter: { max: 5, // Max 5 jobs duration: 1000, // Per second }, }); worker = new Worker( 'rate-limited-queue', async (job: Job) => { processTimestamps.push(Date.now()); return { processed: true }; }, { connection: getRedisConnection(), }, ); // Add 15 jobs (should take at least 2 seconds with rate limit) await Promise.all(Array.from({ length: 15 }, (_, i) => queue.add('rate-job', { index: i }))); const startTime = Date.now(); await waitForJobs(queue, 'completed', 15, 10000); const duration = Date.now() - startTime; // Should take at least 2 seconds due to rate limiting expect(duration).toBeGreaterThan(2000); // Verify jobs were processed in batches expect(processTimestamps).toHaveLength(15); }); });