diff --git a/package-lock.json b/package-lock.json index 48c7f93..0b82c87 100644 --- a/package-lock.json +++ b/package-lock.json @@ -5746,7 +5746,6 @@ "version": "2.3.3", "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz", "integrity": "sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw==", - "dev": true, "hasInstallScript": true, "license": "MIT", "optional": true, diff --git a/package.json b/package.json index a0e2ba3..7859505 100644 --- a/package.json +++ b/package.json @@ -9,11 +9,11 @@ "node": ">=20.0.0" }, "scripts": { - "build": "node_modules/.bin/nest build", + "build": "nest build", "format": "prettier --write \"src/**/*.ts\" \"test/**/*.ts\"", - "start": "node_modules/.bin/nest start", - "start:dev": "node_modules/.bin/nest start --watch", - "start:debug": "node_modules/.bin/nest start --debug --watch", + "start": "nest start", + "start:dev": "nest start --watch", + "start:debug": "nest start --debug --watch", "start:prod": "node dist/main.js", "lint": "eslint \"{src,apps,libs,test}/**/*.ts\"", "typecheck": "tsc --noEmit", diff --git a/src/app.module.ts b/src/app.module.ts index 5d1e27b..8501ec6 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -39,6 +39,7 @@ import { AuditModule } from './modules/audit/audit.module'; import { AiModule } from './modules/ai/ai.module'; import { HealthModule } from './modules/health/health.module'; import { MetricsModule } from './modules/metrics/metrics.module'; +import { AdminModule } from './modules/admin/admin.module'; import { RequestMetricsMiddleware } from './modules/metrics/metrics.middleware'; import { DeadLetterModule } from './modules/dead-letter/dead-letter.module'; import { AgentTraceInterceptor } from './common/interceptors/agent-trace.interceptor'; @@ -119,6 +120,7 @@ import { AgentTraceInterceptor } from './common/interceptors/agent-trace.interce HealthModule, MetricsModule, DeadLetterModule, + AdminModule, ], providers: [ { provide: APP_GUARD, useClass: JwtAuthGuard }, diff --git a/src/modules/admin/admin.module.ts b/src/modules/admin/admin.module.ts new file mode 100644 index 0000000..e33692f --- /dev/null +++ b/src/modules/admin/admin.module.ts @@ -0,0 +1,16 @@ +import { Module } from '@nestjs/common'; +import { QueueModule } from '../../queues/queue.module'; +import { DlqController } from './dlq.controller'; +import { DlqService } from './dlq.service'; + +/** + * Administrative module providing secured operator controls over + * background queues, dead-letter processing, and forensic error recovery. + */ +@Module({ + imports: [QueueModule], + controllers: [DlqController], + providers: [DlqService], + exports: [DlqService], +}) +export class AdminModule {} diff --git a/src/modules/admin/dlq.controller.spec.ts b/src/modules/admin/dlq.controller.spec.ts new file mode 100644 index 0000000..c3469e7 --- /dev/null +++ b/src/modules/admin/dlq.controller.spec.ts @@ -0,0 +1,159 @@ +import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { Reflector } from '@nestjs/core'; +import { UserRole } from '@prisma/client'; +import { DlqController } from './dlq.controller'; +import { DlqService } from './dlq.service'; +import { ROLES_KEY } from '../../common/decorators/roles.decorator'; +import { Queues } from '../../queues/queues.constants'; + +describe('DlqController', () => { + let controller: DlqController; + let service: Record>; + + beforeEach(() => { + service = { + listFailedJobs: vi.fn().mockResolvedValue({ + items: [], + total: 0, + page: 1, + limit: 20, + }), + getQueueStats: vi.fn().mockResolvedValue([ + { + queue: Queues.DeadLetter, + failed: 0, + active: 0, + waiting: 0, + delayed: 0, + completed: 0, + paused: 0, + }, + ]), + getJobDetails: vi.fn().mockResolvedValue({ + id: 'job-1', + queue: Queues.DeadLetter, + name: 'test', + data: {}, + opts: {}, + attemptsMade: 1, + timestamp: 123456, + }), + retryJob: vi.fn().mockResolvedValue({ + jobId: 'job-1', + queue: Queues.DeadLetter, + retried: true, + message: 'Job retried', + }), + retryAllFailedJobs: vi.fn().mockResolvedValue({ + retriedCount: 5, + queues: [Queues.DeadLetter], + }), + removeJob: vi.fn().mockResolvedValue({ + jobId: 'job-1', + queue: Queues.DeadLetter, + removed: true, + }), + purgeQueue: vi.fn().mockResolvedValue({ + purgedCount: 10, + removedJobIds: ['job-1', 'job-2'], + queues: [Queues.DeadLetter], + }), + }; + + controller = new DlqController(service as unknown as DlqService); + }); + + describe('RBAC Roles Guard Configuration', () => { + it('has @Roles(UserRole.OWNER, UserRole.ADMIN) defined at the class level', () => { + const reflector = new Reflector(); + const roles = reflector.get(ROLES_KEY, DlqController); + + expect(roles).toBeDefined(); + expect(roles).toContain(UserRole.OWNER); + expect(roles).toContain(UserRole.ADMIN); + expect(roles).toHaveLength(2); + }); + }); + + describe('Endpoints', () => { + it('listFailedJobs delegates query to dlqService', async () => { + const query = { page: 2, limit: 10, queue: Queues.Webhooks }; + const res = await controller.listFailedJobs(query); + + expect(service.listFailedJobs).toHaveBeenCalledWith(query); + expect(res.page).toBe(1); + }); + + it('getQueueStats delegates to dlqService', async () => { + const res = await controller.getQueueStats(); + + expect(service.getQueueStats).toHaveBeenCalledOnce(); + expect(res).toHaveLength(1); + }); + + it('getJobDetails delegates queue and id to dlqService', async () => { + await controller.getJobDetails(Queues.Webhooks, 'wh-job-1'); + + expect(service.getJobDetails).toHaveBeenCalledWith(Queues.Webhooks, 'wh-job-1'); + }); + + it('getDlqJobDetails defaults to DeadLetter queue', async () => { + await controller.getDlqJobDetails('dlq-job-1'); + + expect(service.getJobDetails).toHaveBeenCalledWith(Queues.DeadLetter, 'dlq-job-1'); + }); + + it('retryJob delegates queue and id to dlqService', async () => { + await controller.retryJob(Queues.Webhooks, 'wh-job-1'); + + expect(service.retryJob).toHaveBeenCalledWith(Queues.Webhooks, 'wh-job-1'); + }); + + it('retryDlqJob delegates to dlqService with default queue', async () => { + await controller.retryDlqJob('dlq-job-1'); + + expect(service.retryJob).toHaveBeenCalledWith(Queues.DeadLetter, 'dlq-job-1'); + }); + + it('retryAllJobs delegates to dlqService with optional queue', async () => { + await controller.retryAllJobs(Queues.Webhooks); + + expect(service.retryAllFailedJobs).toHaveBeenCalledWith(Queues.Webhooks); + }); + + it('retryQueueAllJobs delegates to dlqService', async () => { + await controller.retryQueueAllJobs(Queues.Transactions); + + expect(service.retryAllFailedJobs).toHaveBeenCalledWith(Queues.Transactions); + }); + + it('removeJob delegates queue and id to dlqService', async () => { + await controller.removeJob(Queues.Transactions, 'tx-job-1'); + + expect(service.removeJob).toHaveBeenCalledWith(Queues.Transactions, 'tx-job-1'); + }); + + it('removeDlqJob defaults to DeadLetter queue', async () => { + await controller.removeDlqJob('dlq-job-1'); + + expect(service.removeJob).toHaveBeenCalledWith(Queues.DeadLetter, 'dlq-job-1'); + }); + + it('purgeQueue delegates query to dlqService', async () => { + const purgeDto = { queue: Queues.Reports, gracePeriodMs: 1000, limit: 50 }; + await controller.purgeQueue(purgeDto); + + expect(service.purgeQueue).toHaveBeenCalledWith(purgeDto); + }); + + it('purgeSpecificQueue merges path parameter into purgeDto', async () => { + const purgeDto = { gracePeriodMs: 0, limit: 100 }; + await controller.purgeSpecificQueue(Queues.RiskAnalysis, purgeDto); + + expect(service.purgeQueue).toHaveBeenCalledWith({ + ...purgeDto, + queue: Queues.RiskAnalysis, + }); + }); + }); +}); diff --git a/src/modules/admin/dlq.controller.ts b/src/modules/admin/dlq.controller.ts new file mode 100644 index 0000000..7f999ea --- /dev/null +++ b/src/modules/admin/dlq.controller.ts @@ -0,0 +1,150 @@ +import { + Controller, + Get, + Post, + Delete, + Param, + Query, + HttpCode, + HttpStatus, +} from '@nestjs/common'; +import { ApiOperation, ApiTags, ApiResponse } from '@nestjs/swagger'; +import { UserRole } from '@prisma/client'; +import { Roles } from '../../common/decorators/roles.decorator'; +import { ZodValidationPipe } from '../../common/pipes/zod-validation.pipe'; +import { DlqService } from './dlq.service'; +import { + ListDlqJobsQuery, + listDlqJobsQuerySchema, + PurgeDlqDto, + purgeDlqSchema, + DlqJobDetails, + QueueJobCounts, +} from './dto/dlq.dto'; +import { Queues } from '../../queues/queues.constants'; + +/** + * Administrative Dead-Letter Queue (DLQ) controller. + * Restricted strictly to system administrators (OWNER and ADMIN roles). + */ +@ApiTags('admin-dlq') +@Controller('admin/dlq') +@Roles(UserRole.OWNER, UserRole.ADMIN) +export class DlqController { + constructor(private readonly dlqService: DlqService) {} + + @Get() + @ApiOperation({ summary: 'List failed jobs across queues or for a specific queue' }) + @ApiResponse({ status: 200, description: 'List of failed / dead-lettered jobs' }) + async listFailedJobs( + @Query(new ZodValidationPipe(listDlqJobsQuerySchema)) query: ListDlqJobsQuery, + ) { + return this.dlqService.listFailedJobs(query); + } + + @Get('stats') + @ApiOperation({ summary: 'Get queue job counts and DLQ health stats' }) + @ApiResponse({ status: 200, description: 'Summary counts across all BullMQ queues' }) + async getQueueStats(): Promise { + return this.dlqService.getQueueStats(); + } + + @Post('retry-all') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Retry all failed jobs across all queues or a specified queue' }) + @ApiResponse({ status: 200, description: 'Results of batch retry operation' }) + async retryAllJobs(@Query('queue') queue?: string) { + return this.dlqService.retryAllFailedJobs(queue); + } + + @Delete('purge') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Purge failed jobs across all queues or a specified queue' }) + @ApiResponse({ status: 200, description: 'Results of purge operation' }) + async purgeQueue( + @Query(new ZodValidationPipe(purgeDlqSchema)) query: PurgeDlqDto, + ) { + return this.dlqService.purgeQueue(query); + } + + @Post(':queue/retry-all') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Retry all failed jobs in a specific queue' }) + async retryQueueAllJobs(@Param('queue') queue: string) { + return this.dlqService.retryAllFailedJobs(queue); + } + + @Delete(':queue/purge') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Purge failed jobs in a specific queue' }) + async purgeSpecificQueue( + @Param('queue') queue: string, + @Query(new ZodValidationPipe(purgeDlqSchema)) query: PurgeDlqDto, + ) { + return this.dlqService.purgeQueue({ ...query, queue }); + } + + @Get(':queue/:id') + @ApiOperation({ summary: 'Inspect a specific failed job and error payload in a named queue' }) + @ApiResponse({ status: 200, description: 'Job inspection details' }) + async getJobDetails( + @Param('queue') queue: string, + @Param('id') id: string, + ): Promise { + return this.dlqService.getJobDetails(queue, id); + } + + @Get(':id') + @ApiOperation({ summary: 'Inspect a specific failed job in the default Dead-Letter Queue' }) + @ApiResponse({ status: 200, description: 'Job inspection details' }) + async getDlqJobDetails( + @Param('id') id: string, + @Query('queue') queue?: string, + ): Promise { + return this.dlqService.getJobDetails(queue ?? Queues.DeadLetter, id); + } + + @Post(':queue/:id/retry') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Retry a specific failed job in a named queue' }) + @ApiResponse({ status: 200, description: 'Job retry confirmation' }) + async retryJob( + @Param('queue') queue: string, + @Param('id') id: string, + ) { + return this.dlqService.retryJob(queue, id); + } + + @Post(':id/retry') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Retry a specific failed job in the default Dead-Letter Queue' }) + @ApiResponse({ status: 200, description: 'Job retry confirmation' }) + async retryDlqJob( + @Param('id') id: string, + @Query('queue') queue?: string, + ) { + return this.dlqService.retryJob(queue ?? Queues.DeadLetter, id); + } + + @Delete(':queue/:id') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Delete/remove a specific failed job from a named queue' }) + @ApiResponse({ status: 200, description: 'Job removal confirmation' }) + async removeJob( + @Param('queue') queue: string, + @Param('id') id: string, + ) { + return this.dlqService.removeJob(queue, id); + } + + @Delete(':id') + @HttpCode(HttpStatus.OK) + @ApiOperation({ summary: 'Delete/remove a specific failed job from the default Dead-Letter Queue' }) + @ApiResponse({ status: 200, description: 'Job removal confirmation' }) + async removeDlqJob( + @Param('id') id: string, + @Query('queue') queue?: string, + ) { + return this.dlqService.removeJob(queue ?? Queues.DeadLetter, id); + } +} diff --git a/src/modules/admin/dlq.service.spec.ts b/src/modules/admin/dlq.service.spec.ts new file mode 100644 index 0000000..7451af9 --- /dev/null +++ b/src/modules/admin/dlq.service.spec.ts @@ -0,0 +1,191 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import { NotFoundException, BadRequestException } from '@nestjs/common'; +import { DlqService } from './dlq.service'; +import { Queues } from '../../queues/queues.constants'; + +describe('DlqService', () => { + let service: DlqService; + let mockQueue: Record>; + let mockJob: Record | unknown>; + + beforeEach(() => { + mockJob = { + id: 'job-100', + name: 'send-notification', + data: { userId: 'u-1', message: 'Hello' }, + opts: { attempts: 3 }, + failedReason: 'SMTP connection refused', + stacktrace: ['Error: SMTP connection refused at SmtpClient.connect'], + attemptsMade: 3, + timestamp: 1725050000000, + processedOn: 1725050001000, + finishedOn: 1725050003000, + returnvalue: null, + getState: vi.fn().mockResolvedValue('failed'), + retry: vi.fn().mockResolvedValue(undefined), + remove: vi.fn().mockResolvedValue(undefined), + }; + + mockQueue = { + getFailed: vi.fn().mockResolvedValue([mockJob]), + getFailedCount: vi.fn().mockResolvedValue(1), + getJob: vi.fn().mockResolvedValue(mockJob), + clean: vi.fn().mockResolvedValue(['job-100']), + getJobCounts: vi.fn().mockResolvedValue({ + failed: 1, + active: 0, + waiting: 2, + delayed: 0, + completed: 10, + paused: 0, + }), + close: vi.fn().mockResolvedValue(undefined), + }; + + service = new DlqService(); + // Override getOrCreateQueue to return our mock + vi.spyOn(service, 'getOrCreateQueue').mockReturnValue(mockQueue as never); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + describe('listFailedJobs', () => { + it('returns paginated failed jobs for a specific queue', async () => { + const result = await service.listFailedJobs({ + queue: Queues.Notifications, + page: 1, + limit: 10, + }); + + expect(result.queue).toBe(Queues.Notifications); + expect(result.total).toBe(1); + expect(result.items).toHaveLength(1); + expect(result.items[0]).toEqual( + expect.objectContaining({ + id: 'job-100', + name: 'send-notification', + queue: Queues.Notifications, + failedReason: 'SMTP connection refused', + attemptsMade: 3, + }), + ); + }); + + it('aggregates failed jobs across all queues when no specific queue is provided', async () => { + const result = await service.listFailedJobs({ + page: 1, + limit: 20, + }); + + expect(result.total).toBeGreaterThanOrEqual(1); + expect(result.items.length).toBeGreaterThanOrEqual(1); + expect(result.items[0].id).toBe('job-100'); + }); + }); + + describe('getJobDetails', () => { + it('returns complete inspection details for an existing job', async () => { + const details = await service.getJobDetails(Queues.Notifications, 'job-100'); + + expect(details.id).toBe('job-100'); + expect(details.queue).toBe(Queues.Notifications); + expect(details.data).toEqual({ userId: 'u-1', message: 'Hello' }); + expect(details.failedReason).toBe('SMTP connection refused'); + expect(details.stacktrace).toEqual(['Error: SMTP connection refused at SmtpClient.connect']); + }); + + it('throws NotFoundException when the job does not exist in the queue', async () => { + mockQueue.getJob.mockResolvedValue(null); + + await expect( + service.getJobDetails(Queues.Notifications, 'non-existent-job'), + ).rejects.toThrow(NotFoundException); + }); + }); + + describe('retryJob', () => { + it('retries a failed job and returns success confirmation', async () => { + const result = await service.retryJob(Queues.Notifications, 'job-100'); + + expect(result.retried).toBe(true); + expect(result.jobId).toBe('job-100'); + expect(result.queue).toBe(Queues.Notifications); + expect((mockJob.retry as ReturnType)).toHaveBeenCalledOnce(); + }); + + it('throws NotFoundException if job is not found', async () => { + mockQueue.getJob.mockResolvedValue(null); + + await expect( + service.retryJob(Queues.Notifications, 'missing-job'), + ).rejects.toThrow(NotFoundException); + }); + + it('throws BadRequestException if job is not in failed state', async () => { + (mockJob.getState as ReturnType).mockResolvedValue('active'); + + await expect( + service.retryJob(Queues.Notifications, 'job-100'), + ).rejects.toThrow(BadRequestException); + }); + }); + + describe('retryAllFailedJobs', () => { + it('retries all failed jobs for a target queue', async () => { + const result = await service.retryAllFailedJobs(Queues.Notifications); + + expect(result.retriedCount).toBe(1); + expect(result.queues).toContain(Queues.Notifications); + expect((mockJob.retry as ReturnType)).toHaveBeenCalled(); + }); + }); + + describe('removeJob', () => { + it('removes a job from the queue', async () => { + const result = await service.removeJob(Queues.Notifications, 'job-100'); + + expect(result.removed).toBe(true); + expect(result.jobId).toBe('job-100'); + expect((mockJob.remove as ReturnType)).toHaveBeenCalledOnce(); + }); + + it('throws NotFoundException if job to remove does not exist', async () => { + mockQueue.getJob.mockResolvedValue(null); + + await expect( + service.removeJob(Queues.Notifications, 'non-existent'), + ).rejects.toThrow(NotFoundException); + }); + }); + + describe('purgeQueue', () => { + it('purges failed jobs using queue.clean', async () => { + const result = await service.purgeQueue({ + queue: Queues.Notifications, + gracePeriodMs: 5000, + limit: 500, + }); + + expect(mockQueue.clean).toHaveBeenCalledWith(5000, 500, 'failed'); + expect(result.purgedCount).toBe(1); + expect(result.removedJobIds).toEqual(['job-100']); + }); + }); + + describe('getQueueStats', () => { + it('returns job statistics across queues', async () => { + const stats = await service.getQueueStats(); + + expect(stats.length).toBeGreaterThan(0); + expect(stats[0]).toEqual( + expect.objectContaining({ + failed: 1, + waiting: 2, + completed: 10, + }), + ); + }); + }); +}); diff --git a/src/modules/admin/dlq.service.ts b/src/modules/admin/dlq.service.ts new file mode 100644 index 0000000..37490e0 --- /dev/null +++ b/src/modules/admin/dlq.service.ts @@ -0,0 +1,341 @@ +import { + Injectable, + Logger, + NotFoundException, + BadRequestException, + OnModuleDestroy, +} from '@nestjs/common'; +import { Queue, Job } from 'bullmq'; +import { redisConfig } from '../../config/redis.config'; +import { Queues } from '../../queues/queues.constants'; +import { + DlqJobDetails, + ListDlqJobsQuery, + PurgeDlqDto, + QueueJobCounts, +} from './dto/dlq.dto'; + +@Injectable() +export class DlqService implements OnModuleDestroy { + private readonly logger = new Logger(DlqService.name); + private readonly queueHandles: Map = new Map(); + private readonly knownQueueNames: Set = new Set(Object.values(Queues)); + + /** + * Returns all known queue names (both standard and dynamically accessed). + */ + public getKnownQueueNames(): string[] { + return Array.from(new Set([...this.knownQueueNames, ...this.queueHandles.keys()])); + } + + /** + * Retrieves or lazily creates a BullMQ Queue instance by name. + */ + public getOrCreateQueue(queueName: string): Queue { + this.knownQueueNames.add(queueName); + let queue = this.queueHandles.get(queueName); + if (!queue) { + const { host, port, password, db } = redisConfig(); + queue = new Queue(queueName, { + connection: { + host, + port, + password: password || undefined, + db, + }, + }); + this.queueHandles.set(queueName, queue); + } + return queue; + } + + /** + * Lists failed jobs with pagination across one or all registered queues. + */ + async listFailedJobs(query: ListDlqJobsQuery): Promise<{ + items: DlqJobDetails[]; + total: number; + page: number; + limit: number; + queue?: string; + }> { + const page = query.page ?? 1; + const limit = query.limit ?? 20; + + if (query.queue) { + const queue = this.getOrCreateQueue(query.queue); + const start = query.start !== undefined ? query.start : (page - 1) * limit; + const end = query.end !== undefined ? query.end : start + limit - 1; + + const [jobs, total] = await Promise.all([ + queue.getFailed(start, end), + queue.getFailedCount(), + ]); + + return { + items: jobs.map((job) => this.formatJobDetails(job, query.queue!)), + total, + page, + limit, + queue: query.queue, + }; + } + + // Query across all known queues + const allQueueNames = this.getKnownQueueNames(); + const queueResults = await Promise.all( + allQueueNames.map(async (qName) => { + const queue = this.getOrCreateQueue(qName); + try { + const count = await queue.getFailedCount(); + return { qName, count, queue }; + } catch { + return { qName, count: 0, queue }; + } + }), + ); + + const total = queueResults.reduce((acc, curr) => acc + curr.count, 0); + const start = query.start !== undefined ? query.start : (page - 1) * limit; + const end = query.end !== undefined ? query.end : start + limit - 1; + + // Collect failed jobs from queues that have failures + const collectedJobs: DlqJobDetails[] = []; + for (const { qName, queue, count } of queueResults) { + if (count > 0) { + try { + const jobs = await queue.getFailed(0, 100); + for (const job of jobs) { + collectedJobs.push(this.formatJobDetails(job, qName)); + } + } catch (err) { + this.logger.warn(`Failed to retrieve failed jobs from ${qName}: ${(err as Error).message}`); + } + } + } + + // Sort by timestamp descending + collectedJobs.sort((a, b) => (b.finishedOn ?? b.timestamp) - (a.finishedOn ?? a.timestamp)); + const paginatedItems = collectedJobs.slice(start, end + 1); + + return { + items: paginatedItems, + total, + page, + limit, + }; + } + + /** + * Retrieves full details and failure payload for a specific job. + */ + async getJobDetails(queueName: string, jobId: string): Promise { + const queue = this.getOrCreateQueue(queueName); + const job = await queue.getJob(jobId); + + if (!job) { + throw new NotFoundException(`Job '${jobId}' not found in queue '${queueName}'`); + } + + return this.formatJobDetails(job, queueName); + } + + /** + * Retries a specific failed job. + */ + async retryJob( + queueName: string, + jobId: string, + ): Promise<{ jobId: string; queue: string; retried: boolean; message: string }> { + const queue = this.getOrCreateQueue(queueName); + const job = await queue.getJob(jobId); + + if (!job) { + throw new NotFoundException(`Job '${jobId}' not found in queue '${queueName}'`); + } + + const state = await job.getState(); + if (state !== 'failed') { + throw new BadRequestException( + `Job '${jobId}' in queue '${queueName}' is in '${state}' state and cannot be retried. Only 'failed' jobs can be retried.`, + ); + } + + await job.retry(); + this.logger.log(`Job '${jobId}' in queue '${queueName}' retried by admin.`); + + return { + jobId, + queue: queueName, + retried: true, + message: `Job '${jobId}' successfully moved from failed back to waiting queue.`, + }; + } + + /** + * Retries all failed jobs in a specific queue or across all queues. + */ + async retryAllFailedJobs( + queueName?: string, + ): Promise<{ retriedCount: number; queues: string[] }> { + const targetQueues = queueName ? [queueName] : this.getKnownQueueNames(); + let retriedCount = 0; + const processedQueues: string[] = []; + + for (const qName of targetQueues) { + try { + const queue = this.getOrCreateQueue(qName); + const failedJobs = await queue.getFailed(); + for (const job of failedJobs) { + await job.retry(); + retriedCount++; + } + processedQueues.push(qName); + } catch (err) { + this.logger.warn(`Failed to retry jobs for queue ${qName}: ${(err as Error).message}`); + } + } + + this.logger.log(`Retried ${retriedCount} failed jobs across queues: ${processedQueues.join(', ')}`); + + return { + retriedCount, + queues: processedQueues, + }; + } + + /** + * Removes / deletes a specific failed or dead-letter job. + */ + async removeJob( + queueName: string, + jobId: string, + ): Promise<{ jobId: string; queue: string; removed: boolean }> { + const queue = this.getOrCreateQueue(queueName); + const job = await queue.getJob(jobId); + + if (!job) { + throw new NotFoundException(`Job '${jobId}' not found in queue '${queueName}'`); + } + + await job.remove(); + this.logger.log(`Job '${jobId}' removed from queue '${queueName}'.`); + + return { + jobId, + queue: queueName, + removed: true, + }; + } + + /** + * Purges failed jobs in a queue or across all queues. + */ + async purgeQueue( + dto: PurgeDlqDto, + ): Promise<{ purgedCount: number; removedJobIds: string[]; queues: string[] }> { + const targetQueues = dto.queue ? [dto.queue] : this.getKnownQueueNames(); + const gracePeriodMs = dto.gracePeriodMs ?? 0; + const limit = dto.limit ?? 1000; + + let totalPurged = 0; + const allRemovedIds: string[] = []; + const processedQueues: string[] = []; + + for (const qName of targetQueues) { + try { + const queue = this.getOrCreateQueue(qName); + const removedIds = await queue.clean(gracePeriodMs, limit, 'failed'); + totalPurged += removedIds.length; + allRemovedIds.push(...removedIds); + processedQueues.push(qName); + } catch (err) { + this.logger.warn(`Failed to purge queue ${qName}: ${(err as Error).message}`); + } + } + + this.logger.log(`Purged ${totalPurged} failed jobs from queues: ${processedQueues.join(', ')}`); + + return { + purgedCount: totalPurged, + removedJobIds: allRemovedIds, + queues: processedQueues, + }; + } + + /** + * Returns job counts for all registered queues. + */ + async getQueueStats(): Promise { + const allQueueNames = this.getKnownQueueNames(); + + return Promise.all( + allQueueNames.map(async (qName) => { + const queue = this.getOrCreateQueue(qName); + try { + const counts = await queue.getJobCounts( + 'failed', + 'active', + 'waiting', + 'delayed', + 'completed', + 'paused', + ); + + return { + queue: qName, + failed: counts.failed ?? 0, + active: counts.active ?? 0, + waiting: counts.waiting ?? 0, + delayed: counts.delayed ?? 0, + completed: counts.completed ?? 0, + paused: counts.paused ?? 0, + }; + } catch (err) { + this.logger.warn(`Failed to get job counts for queue ${qName}: ${(err as Error).message}`); + return { + queue: qName, + failed: 0, + active: 0, + waiting: 0, + delayed: 0, + completed: 0, + paused: 0, + }; + } + }), + ); + } + + /** + * Helper to format a BullMQ Job into a clean DlqJobDetails DTO. + */ + private formatJobDetails(job: Job, queueName: string): DlqJobDetails { + return { + id: String(job.id), + name: job.name, + queue: queueName, + data: job.data, + opts: (job.opts as Record) ?? {}, + failedReason: job.failedReason, + stacktrace: job.stacktrace ?? [], + attemptsMade: job.attemptsMade, + timestamp: job.timestamp, + processedOn: job.processedOn, + finishedOn: job.finishedOn, + returnvalue: job.returnvalue, + }; + } + + async onModuleDestroy(): Promise { + await Promise.all( + Array.from(this.queueHandles.values()).map(async (queue) => { + try { + await queue.close(); + } catch (err) { + this.logger.warn(`Error closing queue: ${(err as Error).message}`); + } + }), + ); + } +} diff --git a/src/modules/admin/dto/dlq.dto.ts b/src/modules/admin/dto/dlq.dto.ts new file mode 100644 index 0000000..05837da --- /dev/null +++ b/src/modules/admin/dto/dlq.dto.ts @@ -0,0 +1,64 @@ +import { z } from 'zod'; + +/** Query parameters for listing failed / dead-letter jobs. */ +export const listDlqJobsQuerySchema = z.object({ + /** Target queue name filter. If omitted, queries across all registered queues. */ + queue: z.string().optional(), + /** Page number for pagination (1-indexed). */ + page: z.coerce.number().int().positive().default(1), + /** Number of items per page. */ + limit: z.coerce.number().int().positive().max(100).default(20), + /** Zero-based start index (overrides page/limit if provided). */ + start: z.coerce.number().int().nonnegative().optional(), + /** Zero-based end index (overrides page/limit if provided). */ + end: z.coerce.number().int().nonnegative().optional(), +}); + +export type ListDlqJobsQuery = z.infer; + +/** Payload for retrying failed jobs. */ +export const retryJobDtoSchema = z.object({ + /** Queue name if not supplied in path parameter. */ + queue: z.string().optional(), +}); + +export type RetryJobDto = z.infer; + +/** Payload for purging obsolete failed jobs. */ +export const purgeDlqSchema = z.object({ + /** Queue name to purge. If omitted, purges across all queues. */ + queue: z.string().optional(), + /** Grace period in milliseconds. Jobs failed more recently than this are kept. Defaults to 0 (purge all). */ + gracePeriodMs: z.coerce.number().int().nonnegative().default(0), + /** Maximum number of jobs to purge in this invocation. Defaults to 1000. */ + limit: z.coerce.number().int().positive().max(10000).default(1000), +}); + +export type PurgeDlqDto = z.infer; + +/** Detailed representation of a failed / DLQ job. */ +export interface DlqJobDetails { + id: string; + name: string; + queue: string; + data: unknown; + opts: Record; + failedReason?: string; + stacktrace?: string[]; + attemptsMade: number; + timestamp: number; + processedOn?: number; + finishedOn?: number; + returnvalue?: unknown; +} + +/** Queue summary statistics. */ +export interface QueueJobCounts { + queue: string; + failed: number; + active: number; + waiting: number; + delayed: number; + completed: number; + paused: number; +} diff --git a/src/queues/dlq.processor.spec.ts b/src/queues/dlq.processor.spec.ts new file mode 100644 index 0000000..c0b1976 --- /dev/null +++ b/src/queues/dlq.processor.spec.ts @@ -0,0 +1,125 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import { Job, Queue } from 'bullmq'; +import { DlqProcessor } from './dlq.processor'; +import { DlqJobData, Queues } from './queues.constants'; + +describe('DlqProcessor', () => { + let processor: DlqProcessor; + let mockPrisma: Record; + let mockDomainEventCreate: ReturnType; + + beforeEach(() => { + mockDomainEventCreate = vi.fn().mockResolvedValue({ id: 'event-1' }); + mockPrisma = { + domainEvent: { + create: mockDomainEventCreate, + }, + }; + processor = new DlqProcessor(mockPrisma as never); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + describe('process', () => { + it('processes a dead-letter job, logs details, and records an audit domain event', async () => { + const mockJobData: DlqJobData = { + originalQueue: Queues.Webhooks, + originalJobId: 'job-123', + originalJobName: 'deliver-webhook', + payload: { webhookId: 'wh-1', event: 'payment.completed' }, + failedReason: 'HTTP 500: Internal Server Error', + stacktrace: ['Error: HTTP 500 at fetch'], + attemptsMade: 5, + failedAt: '2026-08-30T21:00:00.000Z', + }; + + const mockJob = { + id: 'dlq-job-1', + data: mockJobData, + } as unknown as Job; + + const result = await processor.process(mockJob); + + expect(result.handled).toBe(true); + expect(result.deadLetteredAt).toBe('2026-08-30T21:00:00.000Z'); + expect(mockDomainEventCreate).toHaveBeenCalledWith({ + data: { + name: 'job.dead_lettered', + aggregateType: 'DEAD_LETTER_QUEUE', + aggregateId: 'job-123', + payload: { + originalQueue: Queues.Webhooks, + originalJobName: 'deliver-webhook', + failedReason: 'HTTP 500: Internal Server Error', + attemptsMade: 5, + failedAt: '2026-08-30T21:00:00.000Z', + }, + }, + }); + }); + + it('gracefully handles missing database client without throwing errors', async () => { + const processorNoDb = new DlqProcessor(undefined); + const mockJobData: DlqJobData = { + originalQueue: Queues.Transactions, + originalJobId: 'tx-456', + payload: { transactionId: 'tx-456' }, + failedReason: 'Insufficient funds', + attemptsMade: 3, + failedAt: '2026-08-30T21:05:00.000Z', + }; + + const mockJob = { + id: 'dlq-job-2', + data: mockJobData, + } as unknown as Job; + + const result = await processorNoDb.process(mockJob); + expect(result.handled).toBe(true); + }); + }); + + describe('moveToDeadLetter static helper', () => { + it('constructs correct DlqJobData and adds job to the DLQ queue', async () => { + const mockDlqQueue = { + add: vi.fn().mockResolvedValue({ id: 'dlq-added-1' }), + } as unknown as Queue; + + const originalFailedJob = { + id: 'orig-job-999', + name: 'sync-stellar-balance', + data: { walletId: 'wallet-1', address: 'GABC123' }, + attemptsMade: 3, + stacktrace: ['Error: Horizon connection timeout'], + timestamp: 1725050000000, + processedOn: 1725050001000, + finishedOn: 1725050005000, + } as unknown as Job; + + const error = new Error('Horizon connection timeout'); + + const result = await DlqProcessor.moveToDeadLetter( + mockDlqQueue, + originalFailedJob, + error, + Queues.StellarSync, + ); + + expect(mockDlqQueue.add).toHaveBeenCalledWith( + expect.stringContaining('dlq:stellar-sync:orig-job-999'), + expect.objectContaining({ + originalQueue: Queues.StellarSync, + originalJobId: 'orig-job-999', + originalJobName: 'sync-stellar-balance', + payload: { walletId: 'wallet-1', address: 'GABC123' }, + failedReason: 'Horizon connection timeout', + attemptsMade: 3, + }), + expect.any(Object), + ); + expect(result).toEqual({ id: 'dlq-added-1' }); + }); + }); +}); diff --git a/src/queues/dlq.processor.ts b/src/queues/dlq.processor.ts new file mode 100644 index 0000000..c6fd495 --- /dev/null +++ b/src/queues/dlq.processor.ts @@ -0,0 +1,126 @@ +import { Processor, WorkerHost } from '@nestjs/bullmq'; +import { Inject, Injectable, Logger, Optional } from '@nestjs/common'; +import { Job, Queue } from 'bullmq'; +import { Queues, DlqJobData } from './queues.constants'; +import { PrismaService } from '../database/prisma.service'; + +/** + * BullMQ worker processor for the Dead-Letter Queue (DLQ). + * + * Jobs that exhaust maximum retry attempts across transaction execution, + * webhook delivery, risk analysis, and other background workers are routed here + * for logging, alerting, forensic auditing, and administrative triage. + */ +@Injectable() +@Processor(Queues.DeadLetter) +export class DlqProcessor extends WorkerHost { + private readonly logger = new Logger(DlqProcessor.name); + + constructor(@Optional() @Inject(PrismaService) private readonly prisma?: PrismaService) { + super(); + } + + /** + * Processes a dead-lettered job. + * Logs full error diagnostics, failure reasons, and stack traces. + */ + async process(job: Job): Promise<{ handled: boolean; deadLetteredAt: string }> { + const { + originalQueue, + originalJobId, + originalJobName, + payload, + failedReason, + attemptsMade, + failedAt, + } = job.data; + + this.logger.error( + `[DLQ-JOB-RECEIVED] Job ${originalJobId ?? job.id} from queue '${originalQueue}' (${originalJobName ?? 'unknown'}) permanently failed after ${attemptsMade} attempts. Reason: ${failedReason ?? 'Unknown'}`, + ); + + this.logger.debug( + `[DLQ-JOB-DETAILS] Payload: ${JSON.stringify(payload)} | FailedAt: ${failedAt}`, + ); + + await this.recordDeadLetterAudit(job.data); + + return { + handled: true, + deadLetteredAt: failedAt || new Date().toISOString(), + }; + } + + /** + * Helper method to route a failed job from any queue into the DLQ. + * + * @param dlqQueue BullMQ Queue instance for the dead-letter queue + * @param failedJob Original failed job + * @param error Error that caused the failure + * @param originalQueue Name of the queue where failure occurred + */ + static async moveToDeadLetter( + dlqQueue: Queue, + failedJob: Job, + error: Error | string, + originalQueue: string, + ): Promise> { + const failedReason = typeof error === 'string' ? error : error.message; + const stacktrace = typeof error === 'object' && error.stack ? [error.stack] : failedJob.stacktrace; + + const dlqData: DlqJobData = { + originalQueue, + originalJobId: failedJob.id, + originalJobName: failedJob.name, + payload: failedJob.data, + failedReason, + stacktrace: stacktrace ?? [], + attemptsMade: failedJob.attemptsMade, + failedAt: new Date().toISOString(), + metadata: { + timestamp: failedJob.timestamp, + processedOn: failedJob.processedOn, + finishedOn: failedJob.finishedOn, + }, + }; + + return dlqQueue.add(`dlq:${originalQueue}:${failedJob.id ?? Date.now()}`, dlqData, { + removeOnComplete: { count: 5000 }, + removeOnFail: { age: 7 * 24 * 3600 }, + }); + } + + /** + * Records a domain event or audit log for dead-lettered jobs if database is available. + */ + private async recordDeadLetterAudit(data: DlqJobData): Promise { + if (!this.prisma) return; + + try { + const client = this.prisma.workerClient ?? this.prisma; + const prismaAny = client as unknown as Record; + const domainEvents = prismaAny['domainEvent'] as + | { create?: (args: unknown) => Promise } + | undefined; + + if (domainEvents?.create) { + await domainEvents.create({ + data: { + name: 'job.dead_lettered', + aggregateType: 'DEAD_LETTER_QUEUE', + aggregateId: data.originalJobId ?? null, + payload: { + originalQueue: data.originalQueue, + originalJobName: data.originalJobName, + failedReason: data.failedReason, + attemptsMade: data.attemptsMade, + failedAt: data.failedAt, + }, + }, + }); + } + } catch (err) { + this.logger.warn(`Failed to record DLQ audit event: ${(err as Error).message}`); + } + } +} diff --git a/src/queues/index.ts b/src/queues/index.ts index 904279c..2679a62 100644 --- a/src/queues/index.ts +++ b/src/queues/index.ts @@ -1,2 +1,4 @@ export * from './queue.module'; export * from './queues.constants'; +export * from './dlq.processor'; + diff --git a/src/queues/queue.module.ts b/src/queues/queue.module.ts index 148bc84..3f1f807 100644 --- a/src/queues/queue.module.ts +++ b/src/queues/queue.module.ts @@ -1,4 +1,7 @@ import { Module } from '@nestjs/common'; +import { BullModule } from '@nestjs/bullmq'; +import { Queues } from './queues.constants'; +import { DlqProcessor } from './dlq.processor'; /** * Central queue plumbing. The BullMQ connections and named queues are provisioned @@ -9,8 +12,18 @@ import { Module } from '@nestjs/common'; * below are the public surface every worker uses. */ @Module({ - providers: [], - exports: [], + imports: [ + BullModule.registerQueue({ + name: Queues.DeadLetter, + defaultJobOptions: { + attempts: 1, + removeOnComplete: { count: 5_000 }, + removeOnFail: { age: 7 * 24 * 3_600 }, + }, + }), + ], + providers: [DlqProcessor], + exports: [BullModule, DlqProcessor], }) export class QueueModule {} @@ -21,6 +34,9 @@ export const QueueTokens = { StellarSync: Symbol.for('queue:stellar-sync'), Analytics: Symbol.for('queue:analytics'), Reports: Symbol.for('queue:reports'), + Transactions: Symbol.for('queue:transactions'), + RiskAnalysis: Symbol.for('queue:risk-analysis'), + DeadLetter: Symbol.for('queue:dead-letter'), } as const; /** Default job options for every queue: bounded retries with backoff. */ @@ -30,3 +46,4 @@ export const DEFAULT_JOB_OPTIONS = { removeOnComplete: { count: 1_000 }, removeOnFail: { age: 24 * 3_600 }, } as const; + diff --git a/src/queues/queues.constants.ts b/src/queues/queues.constants.ts index 3b6439e..ae4a133 100644 --- a/src/queues/queues.constants.ts +++ b/src/queues/queues.constants.ts @@ -17,6 +17,34 @@ export const Queues = { OutboxEvents: 'outbox-events', /** Stellar fee-bump submission retries. */ StellarFeeBump: 'stellar-fee-bump', + /** Transaction execution pipeline. */ + Transactions: 'transactions', + /** Asynchronous risk evaluation and scoring. */ + RiskAnalysis: 'risk-analysis', + /** Dead-letter queue for terminal job failures across all workers. */ + DeadLetter: 'dead-letter', } as const; export type QueueName = (typeof Queues)[keyof typeof Queues]; + +/** Standard payload stored when a job is dead-lettered. */ +export interface DlqJobData { + /** Original queue the job originated from. */ + originalQueue: string; + /** Original BullMQ job ID. */ + originalJobId?: string; + /** Original job name. */ + originalJobName?: string; + /** Payload of the original failed job. */ + payload: unknown; + /** Terminal failure reason or error message. */ + failedReason?: string; + /** Stack trace if available. */ + stacktrace?: string[]; + /** Total retry attempts made before dead-lettering. */ + attemptsMade: number; + /** ISO timestamp when job was dead-lettered. */ + failedAt: string; + /** Additional metadata (organizationId, transactionId, etc.). */ + metadata?: Record; +}