fix: processed by moved one property up & fix analytics empty

Author: berenteb

api-generated/src/api.ts

@@ -37,10 +37,10 @@ export interface AnalyticsData {
     'id': string;
     /**
      * 
-     * @type {SendRequestJobData}
+     * @type {SingleSendRequestDto}
      * @memberof AnalyticsData
      */
-    'data': SendRequestJobData;
+    'data': SingleSendRequestDto;
     /**
      * 
      * @type {string}
@@ -59,6 +59,12 @@ export interface AnalyticsData {
      * @memberof AnalyticsData
      */
     'queue': string;
+    /**
+     * 
+     * @type {string}
+     * @memberof AnalyticsData
+     */
+    'processedBy': string;
 }
 
 export const AnalyticsDataStatusEnum = {
@@ -139,55 +145,6 @@ export interface ResponseDto {
      */
     'message': string;
 }
-/**
- * 
- * @export
- * @interface SendRequestJobData
- */
-export interface SendRequestJobData {
-    /**
-     * 
-     * @type {string}
-     * @memberof SendRequestJobData
-     */
-    'from': string;
-    /**
-     * 
-     * @type {string}
-     * @memberof SendRequestJobData
-     */
-    'to': string;
-    /**
-     * 
-     * @type {string}
-     * @memberof SendRequestJobData
-     */
-    'subject': string;
-    /**
-     * 
-     * @type {string}
-     * @memberof SendRequestJobData
-     */
-    'html': string;
-    /**
-     * 
-     * @type {string}
-     * @memberof SendRequestJobData
-     */
-    'replyTo': string;
-    /**
-     * 
-     * @type {string}
-     * @memberof SendRequestJobData
-     */
-    'directQueue': string;
-    /**
-     * 
-     * @type {string}
-     * @memberof SendRequestJobData
-     */
-    'processedBy': string;
-}
 /**
  * 
  * @export
@@ -229,7 +186,7 @@ export interface SingleSendRequestDto {
      * @type {string}
      * @memberof SingleSendRequestDto
      */
-    'directQueue': string;
+    'queue': string;
 }
 /**
  * 
@@ -684,6 +641,9 @@ export const GatewayApiAxiosParamCreator = function (configuration?: Configurati
             const localVarHeaderParameter = {} as any;
             const localVarQueryParameter = {} as any;
 
+            // authentication Api-Key required
+            await setApiKeyToObject(localVarHeaderParameter, "Authorization", configuration)
+
 
 
             localVarHeaderParameter['Content-Type'] = 'application/json';

apps/consumer/src/app/consumer.service.ts

@@ -1,4 +1,4 @@
-import { SendRequestJobData } from '@kir-mail/types';
+import { SingleSendRequestDto } from '@kir-mail/types';
 import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common';
 import { MailerService } from '@nestjs-modules/mailer';
 import { Job, MetricsTime, Worker } from 'bullmq';
@@ -21,7 +21,7 @@ export class ConsumerService implements OnModuleDestroy {
 
   constructor(private readonly mailerService: MailerService) {
     for (const queueId of QUEUE_IDS) {
-      this.logger.log(`Creating direct queue: ${queueId}`);
+      this.logger.log(`Creating queue: ${queueId}`);
       this.workers.push(
         new Worker(queueId, this.process.bind(this), {
           connection: {
@@ -47,10 +47,9 @@ export class ConsumerService implements OnModuleDestroy {
     }
   }
 
-  async process(job: Job<SendRequestJobData>) {
+  async process(job: Job<SingleSendRequestDto>) {
     this.logger.log(`Processing job #${job.id} 🔄`);
-    await job.updateData({ ...job.data, processedBy: CONSUMER_NAME });
-
+    await new Promise((resolve) => setTimeout(resolve, 10000));
     try {
       if (!DISABLE_EMAILS)
         await this.mailerService.sendMail({
@@ -64,7 +63,6 @@ export class ConsumerService implements OnModuleDestroy {
       this.logger.error(`Job ${job.id} failed with error: ${error} 🚨`);
       throw error;
     }
-    await job.updateData({ ...job.data, processedBy: CONSUMER_NAME });
     this.logger.log(`Job #${job.id} processed ✅`);
   }
 }

apps/dashboard/src/app/components/analytics-data-table.tsx

@@ -29,7 +29,7 @@ export function AnalyticsDataTable({ data }: AnalyticsDataTableProps) {
           accessorKey: 'data.subject',
           header: 'Tárgy',
         },
-        { accessorKey: 'data.processedBy', header: 'Szolgáltatás' },
+        { accessorKey: 'processedBy', header: 'Szolgáltatás' },
         { accessorKey: 'queue', header: 'Üzenetsor' },
         { accessorKey: 'status', header: 'Státusz', cell: ({ row }) => <StatusBadge status={row.original.status} /> },
       ]}

apps/gateway/src/app/gateway.service.ts

@@ -1,4 +1,4 @@
-import { SendRequestJobData, SingleSendRequestDto } from '@kir-mail/types';
+import { SingleSendRequestDto } from '@kir-mail/types';
 import { Injectable, InternalServerErrorException, Logger, NotFoundException } from '@nestjs/common';
 import { Job, JobType, Queue } from 'bullmq';
 
@@ -10,7 +10,7 @@ const DEFAULT_QUEUE = 'send';
 @Injectable()
 export class GatewayService {
   private readonly logger = new Logger(GatewayService.name);
-  private readonly directQueues: Queue[] = [];
+  private readonly queues: Queue[] = [];
   constructor() {
     this.createQueue(DEFAULT_QUEUE);
     QUEUE_IDS.forEach((queueId) => this.createQueue(queueId));
@@ -31,7 +31,7 @@ export class GatewayService {
         },
       },
     });
-    this.directQueues.push(queue);
+    this.queues.push(queue);
   }
 
   async sendMessage(request: SingleSendRequestDto) {
@@ -62,11 +62,11 @@ export class GatewayService {
     ];
 
     const jobs: AnalyticsData[] = [];
-
     for (const type of types) {
-      let rawJobs: Job<SendRequestJobData>[] = [];
-      for (const queue of this.directQueues) {
-        rawJobs = await queue.getJobs(type);
+      const rawJobs: Job<SingleSendRequestDto>[] = [];
+      for (const queue of this.queues) {
+        const jobsForQueues = await queue.getJobs(type);
+        rawJobs.push(...jobsForQueues);
       }
       jobs.push(...this.mapJobsToDto(rawJobs, type));
     }
@@ -82,28 +82,29 @@ export class GatewayService {
     return {
       items: splitJobs,
       timestamps: timestamps,
-      availableQueues: this.directQueues.map((queue) => queue.name),
+      availableQueues: this.queues.map((queue) => queue.name),
     };
   }
 
   private getQueueForRequest(request: SingleSendRequestDto): Queue | undefined {
-    if (request.directQueue) {
-      return this.directQueues.find((queue) => queue.name === request.directQueue);
+    if (request.queue) {
+      return this.queues.find((queue) => queue.name === request.queue);
     }
     return this.getDefaultQueue();
   }
 
   private getDefaultQueue(): Queue {
-    return this.directQueues.find((queue) => queue.name === DEFAULT_QUEUE);
+    return this.queues.find((queue) => queue.name === DEFAULT_QUEUE);
   }
 
-  private mapJobsToDto(jobs: Job<SendRequestJobData>[], status: JobType): AnalyticsData[] {
+  private mapJobsToDto(jobs: Job<SingleSendRequestDto>[], status: JobType): AnalyticsData[] {
     return jobs.map((job) => ({
       id: job.id,
       data: job.data,
       status: status,
       timestamp: job.timestamp,
       queue: job.queueName,
+      processedBy: job.processedBy,
     }));
   }
 

apps/gateway/src/types/response.type.ts

@@ -1,4 +1,4 @@
-import { SendRequestJobData } from '@kir-mail/types';
+import { SingleSendRequestDto } from '@kir-mail/types';
 import { ApiProperty } from '@nestjs/swagger';
 import { JobType } from 'bullmq';
 
@@ -14,8 +14,8 @@ export class AnalyticsData {
   @ApiProperty({ example: '1' })
   id: string;
 
-  @ApiProperty({ type: SendRequestJobData })
-  data: SendRequestJobData;
+  @ApiProperty({ type: SingleSendRequestDto })
+  data: SingleSendRequestDto;
 
   @ApiProperty({
     example: 'completed',
@@ -39,6 +39,9 @@ export class AnalyticsData {
 
   @ApiProperty({ example: 'send' })
   queue: string;
+
+  @ApiProperty({ example: 'consumer' })
+  processedBy: string;
 }
 
 export class TimestampsDto {

types/src/lib/types.ts

@@ -31,13 +31,6 @@ export class SingleSendRequestDto {
   queue?: string;
 }
 
-export class SendRequestJobData extends SingleSendRequestDto {
-  @ApiProperty({ example: 'consumer' })
-  @IsString()
-  @IsOptional()
-  processedBy?: string;
-}
-
 export class UserDto {
   @ApiProperty({ example: 'Test User' })
   @IsString()