Merge pull request #1181 from Musab1258/feat/1142-correlation-id-background-jobs

feat(backend): propagate correlationId to background jobs

Author: Devsol-01

backend/src/modules/admin/admin-audit-logs.controller.ts

@@ -28,6 +28,7 @@ import { JwtAuthGuard } from '../../auth/guards/jwt-auth.guard';
 import { RolesGuard } from '../../common/guards/roles.guard';
 import { Roles } from '../../common/decorators/roles.decorator';
 import { Role } from '../../common/enums/role.enum';
+import { CorrelationId } from '../../common/decorators/correlation-id.decorator';
 import { JobQueueService } from '../job-queue/job-queue.service';
 import { QUEUE_NAMES } from '../job-queue/job-queue.constants';
 import { Job } from 'bullmq';
@@ -158,19 +159,24 @@ export class AdminAuditLogsController {
   async queueExport(
     @Body() dto: AuditLogExportDto,
     @Query('format') format: 'csv' | 'json' = 'csv',
+    @CorrelationId() correlationId?: string,
   ) {
-    const job = await this.jobQueueService.addAuditLogExportJob({
-      filters: {
-        actor: dto.actor,
-        action: dto.action,
-        resourceType: dto.resourceType,
-        resourceId: dto.resourceId,
-        fromDate: dto.fromDate,
-        toDate: dto.toDate,
+    const job = await this.jobQueueService.addAuditLogExportJob(
+      {
+        filters: {
+          actor: dto.actor,
+          action: dto.action,
+          resourceType: dto.resourceType,
+          resourceId: dto.resourceId,
+          fromDate: dto.fromDate,
+          toDate: dto.toDate,
+        },
+        format,
+        requestedBy: 'admin',
       },
-      format,
-      requestedBy: 'admin',
-    });
+      undefined,
+      correlationId,
+    );
 
     return {
       jobId: job.id,

backend/src/modules/disputes/disputes.controller.ts

@@ -228,8 +228,9 @@ export class DisputesController {
       }),
     )
     file: any,
+    @CorrelationId() correlationId?: string,
   ): Promise<DisputeEvidence> {
-    return this.disputesService.uploadEvidence(id, file, dto);
+    return this.disputesService.uploadEvidence(id, file, dto, correlationId);
   }
 
   @Get(':id/evidence')

backend/src/modules/disputes/disputes.service.ts

@@ -396,6 +396,7 @@ export class DisputesService {
     disputeId: string,
     file: any,
     dto: UploadEvidenceDto,
+    correlationId?: string,
   ): Promise<DisputeEvidence> {
     // Verify dispute exists
     const dispute = await this.findOne(disputeId);
@@ -449,6 +450,8 @@ export class DisputesService {
     const savedEvidence = await this.evidenceRepository.save(evidence);
 
     // Enqueue background processing job
+    const job = await this.jobQueueService.addEvidenceProcessingJob(
+      {
     let job;
     try {
       job = await this.jobQueueService.addEvidenceProcessingJob({
@@ -458,6 +461,10 @@ export class DisputesService {
         mimeType: file.mimetype,
         originalFilename: file.originalname,
         uploadedBy: dto.uploadedBy,
+      },
+      undefined,
+      correlationId,
+    );
       });
     } catch (err) {
       await this.quotaService.release(reservation.reservationId, {

backend/src/modules/job-queue/job-queue.service.spec.ts

@@ -86,13 +86,23 @@ describe('JobQueueService', () => {
       expect(result.id).toBe('job-1');
     });
 
+    it('should propagate correlationId in job data', async () => {
     it('should return existing job on duplicate enqueue', async () => {
       const data = {
         userId: 'user-1',
         type: 'sweep_completed',
         title: 'Sweep Done',
         message: 'Swept 100 XLM',
       };
+      const correlationId = 'corr-notif-123';
+
+      await service.addNotificationJob(data, undefined, correlationId);
+
+      expect(notificationQueue.add).toHaveBeenCalledWith(
+        'send-notification',
+        expect.objectContaining({ correlationId }),
+        undefined,
+      );
       const duplicateError = new Error('Job already exists');
       duplicateError.name = 'JobAlreadyExistsError';
       const existingJob = { id: 'existing-job', data };
@@ -138,6 +148,24 @@ describe('JobQueueService', () => {
       );
       expect(result.id).toBe('job-1');
     });
+
+    it('should propagate correlationId in job data', async () => {
+      const data = {
+        to: 'user@test.com',
+        subject: 'Welcome',
+        template: 'welcome',
+        context: { name: 'Alice' },
+      };
+      const correlationId = 'corr-email-456';
+
+      await service.addEmailJob(data, undefined, correlationId);
+
+      expect(emailQueue.add).toHaveBeenCalledWith(
+        'send-email',
+        expect.objectContaining({ correlationId }),
+        undefined,
+      );
+    });
   });
 
   describe('addBlockchainJob', () => {
@@ -157,6 +185,24 @@ describe('JobQueueService', () => {
         { jobId: 'blockchain-evt-123' },
       );
     });
+
+    it('should propagate correlationId in job data', async () => {
+      const data = {
+        eventId: 'evt-456',
+        contractId: 'CDEF456',
+        eventType: 'withdrawal',
+        rawEvent: { ledger: 200 },
+      };
+      const correlationId = 'corr-bc-789';
+
+      await service.addBlockchainJob(data, undefined, correlationId);
+
+      expect(blockchainQueue.add).toHaveBeenCalledWith(
+        'process-blockchain-event',
+        expect.objectContaining({ correlationId }),
+        expect.objectContaining({ jobId: 'blockchain-evt-456' }),
+      );
+    });
   });
 
   describe('addReportJob', () => {
@@ -177,6 +223,23 @@ describe('JobQueueService', () => {
         }),
       );
     });
+
+    it('should propagate correlationId in job data', async () => {
+      const data = {
+        reportType: 'annual-summary',
+        userId: 'user-2',
+        params: { year: 2025 },
+      };
+      const correlationId = 'corr-report-abc';
+
+      await service.addReportJob(data, undefined, correlationId);
+
+      expect(reportQueue.add).toHaveBeenCalledWith(
+        'generate-report',
+        expect.objectContaining({ correlationId }),
+        undefined,
+      );
+    });
   });
 
   describe('addAvatarProcessingJob', () => {
@@ -200,6 +263,25 @@ describe('JobQueueService', () => {
         }),
       );
     });
+
+    it('should propagate correlationId in job data', async () => {
+      const data = {
+        uploadId: 'upload-2',
+        userId: 'user-3',
+        storagePath: 'avatars/user-3/raw.png',
+        mimeType: 'image/jpeg',
+        originalFilename: 'photo.jpg',
+      };
+      const correlationId = 'corr-avatar-def';
+
+      await service.addAvatarProcessingJob(data, undefined, correlationId);
+
+      expect(avatarQueue.add).toHaveBeenCalledWith(
+        'process-avatar',
+        expect.objectContaining({ correlationId }),
+        expect.objectContaining({ jobId: 'avatar-upload-2' }),
+      );
+    });
   });
 
   describe('addAuditLogExportJob', () => {
@@ -220,6 +302,45 @@ describe('JobQueueService', () => {
         }),
       );
     });
+
+    it('should propagate correlationId in job data', async () => {
+      const data = {
+        filters: { actor: 'admin-2' },
+        format: 'json' as const,
+        requestedBy: 'admin-2',
+      };
+      const correlationId = 'corr-export-ghi';
+
+      await service.addAuditLogExportJob(data, undefined, correlationId);
+
+      expect(auditLogExportQueue.add).toHaveBeenCalledWith(
+        'export-audit-logs',
+        expect.objectContaining({ correlationId }),
+        expect.objectContaining({ attempts: 3 }),
+      );
+    });
+  });
+
+  describe('addEvidenceProcessingJob', () => {
+    it('should propagate correlationId in job data', async () => {
+      const data = {
+        evidenceId: 'evid-1',
+        disputeId: 'disp-1',
+        storagePath: 'evidence/file.pdf',
+        mimeType: 'application/pdf',
+        originalFilename: 'file.pdf',
+        uploadedBy: 'user-1',
+      };
+      const correlationId = 'corr-evid-jkl';
+
+      await service.addEvidenceProcessingJob(data, undefined, correlationId);
+
+      expect(disputeEvidenceQueue.add).toHaveBeenCalledWith(
+        'process-dispute-evidence',
+        expect.objectContaining({ correlationId }),
+        expect.objectContaining({ jobId: 'evidence-evid-1' }),
+      );
+    });
   });
 
   describe('getQueueStatus', () => {