feat: move logic for calling webhook to separate queue

apps/api/src/common/constants/notifications.ts

@@ -23,6 +23,7 @@ export const ChannelType = {
 export const QueueAction = {
   SEND: 'send',
   DELIVERY_STATUS: 'delivery-status',
+  WEBHOOK: 'webhook',
 };
 
 export const ProviderDeliveryStatus = {

apps/api/src/jobs/consumers/notifications/mailgun-notifications.job.consumer.ts

@@ -10,6 +10,7 @@ import { MessagesSendResult } from 'mailgun.js';
 import { Inject, Injectable, forwardRef } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { WebhookService } from 'src/modules/webhook/webhook.service';
+import { NotificationQueueProducer } from 'src/jobs/producers/notifications/notifications.job.producer';
 
 @Injectable()
 export class MailgunNotificationConsumer extends NotificationConsumer {
@@ -19,10 +20,18 @@ export class MailgunNotificationConsumer extends NotificationConsumer {
     private readonly mailgunService: MailgunService,
     @Inject(forwardRef(() => NotificationsService))
     notificationsService: NotificationsService,
+    @Inject(forwardRef(() => NotificationQueueProducer))
+    notificationsQueueService: NotificationQueueProducer,
     webhookService: WebhookService,
     configService: ConfigService,
   ) {
-    super(notificationRepository, notificationsService, webhookService, configService);
+    super(
+      notificationRepository,
+      notificationsService,
+      notificationsQueueService,
+      webhookService,
+      configService,
+    );
   }
 
   async processMailgunNotificationQueue(id: number): Promise<void> {

apps/api/src/jobs/consumers/notifications/notification.consumer.ts

@@ -4,11 +4,13 @@ import { Repository } from 'typeorm';
 import { Notification } from 'src/modules/notifications/entities/notification.entity';
 import {
   DeliveryStatus,
+  QueueAction,
   SkipProviderConfirmationChannels,
 } from 'src/common/constants/notifications';
 import { NotificationsService } from 'src/modules/notifications/notifications.service';
 import { ConfigService } from '@nestjs/config';
 import { WebhookService } from 'src/modules/webhook/webhook.service';
+import { NotificationQueueProducer } from 'src/jobs/producers/notifications/notifications.job.producer';
 
 @Injectable()
 export abstract class NotificationConsumer {
@@ -19,6 +21,7 @@ export abstract class NotificationConsumer {
     @InjectRepository(Notification)
     protected readonly notificationRepository: Repository<Notification>,
     protected readonly notificationsService: NotificationsService,
+    private readonly notificationQueueService: NotificationQueueProducer,
     protected readonly webhookService: WebhookService,
     private readonly configService: ConfigService,
   ) {
@@ -43,7 +46,11 @@ export abstract class NotificationConsumer {
           `Channel type: ${notification.channelType} is included in skip queue. Provider confirmation skipped for notification id ${notification.id}`,
         );
         notification.deliveryStatus = DeliveryStatus.SUCCESS;
-        await this.webhookService.triggerWebhook(notification);
+        await this.notificationRepository.save(notification);
+        await this.notificationQueueService.addNotificationToQueue(
+          QueueAction.WEBHOOK,
+          notification,
+        );
       } else {
         this.logger.debug(
           `Notification id ${notification.id} is awaiting confirmation from provider`,
@@ -115,7 +122,11 @@ export abstract class NotificationConsumer {
       }
 
       if (notification.deliveryStatus === DeliveryStatus.SUCCESS) {
-        await this.webhookService.triggerWebhook(notification);
+        await this.notificationRepository.save(notification);
+        await this.notificationQueueService.addNotificationToQueue(
+          QueueAction.WEBHOOK,
+          notification,
+        );
       }
     } catch (error) {
       if (notification.retryCount < this.maxRetryCount) {
@@ -126,7 +137,11 @@ export abstract class NotificationConsumer {
           `Notification with ID ${notification.id} has attempted max allowed retries (provider confirmation), setting delivery status to ${DeliveryStatus.FAILED}`,
         );
         notification.deliveryStatus = DeliveryStatus.FAILED;
-        await this.webhookService.triggerWebhook(notification);
+        await this.notificationRepository.save(notification);
+        await this.notificationQueueService.addNotificationToQueue(
+          QueueAction.WEBHOOK,
+          notification,
+        );
       }
 
       this.logger.error(

apps/api/src/jobs/consumers/notifications/pushSns-notifications.job.consumer.ts

@@ -7,6 +7,7 @@ import { Inject, Injectable, forwardRef } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { PushSnsData, PushSnsService } from 'src/modules/providers/push-sns/push-sns.service';
 import { WebhookService } from 'src/modules/webhook/webhook.service';
+import { NotificationQueueProducer } from 'src/jobs/producers/notifications/notifications.job.producer';
 
 @Injectable()
 export class PushSnsNotificationConsumer extends NotificationConsumer {
@@ -16,10 +17,18 @@ export class PushSnsNotificationConsumer extends NotificationConsumer {
     private readonly pushSnsService: PushSnsService,
     @Inject(forwardRef(() => NotificationsService))
     notificationsService: NotificationsService,
-    configService: ConfigService,
+    @Inject(forwardRef(() => NotificationQueueProducer))
+    notificationsQueueService: NotificationQueueProducer,
     webhookService: WebhookService,
+    configService: ConfigService,
   ) {
-    super(notificationRepository, notificationsService, webhookService, configService);
+    super(
+      notificationRepository,
+      notificationsService,
+      notificationsQueueService,
+      webhookService,
+      configService,
+    );
   }
 
   async processPushSnsNotificationQueue(id: number): Promise<void> {

apps/api/src/jobs/consumers/notifications/smsKapsystem-notifications.job.consumer.ts

@@ -10,6 +10,7 @@ import { Notification } from 'src/modules/notifications/entities/notification.en
 import { Inject, Injectable, forwardRef } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { WebhookService } from 'src/modules/webhook/webhook.service';
+import { NotificationQueueProducer } from 'src/jobs/producers/notifications/notifications.job.producer';
 
 @Injectable()
 export class SmsKapsystemNotificationsConsumer extends NotificationConsumer {
@@ -19,10 +20,18 @@ export class SmsKapsystemNotificationsConsumer extends NotificationConsumer {
     private readonly kapsystemService: SmsKapsystemService,
     @Inject(forwardRef(() => NotificationsService))
     notificationsService: NotificationsService,
+    @Inject(forwardRef(() => NotificationQueueProducer))
+    notificationsQueueService: NotificationQueueProducer,
     webhookService: WebhookService,
     configService: ConfigService,
   ) {
-    super(notificationRepository, notificationsService, webhookService, configService);
+    super(
+      notificationRepository,
+      notificationsService,
+      notificationsQueueService,
+      webhookService,
+      configService,
+    );
   }
 
   async processSmsKapsystemNotificationQueue(id: number): Promise<void> {

apps/api/src/jobs/consumers/notifications/smsPlivo-notifications.job.consumer.ts

@@ -12,6 +12,7 @@ import { Inject, Injectable, forwardRef } from '@nestjs/common';
 import { DeliveryStatus, ProviderDeliveryStatus } from 'src/common/constants/notifications';
 import { ConfigService } from '@nestjs/config';
 import { WebhookService } from 'src/modules/webhook/webhook.service';
+import { NotificationQueueProducer } from 'src/jobs/producers/notifications/notifications.job.producer';
 
 @Injectable()
 export class SmsPlivoNotificationsConsumer extends NotificationConsumer {
@@ -21,10 +22,18 @@ export class SmsPlivoNotificationsConsumer extends NotificationConsumer {
     private readonly smsPlivoService: SmsPlivoService,
     @Inject(forwardRef(() => NotificationsService))
     notificationsService: NotificationsService,
+    @Inject(forwardRef(() => NotificationQueueProducer))
+    notificationsQueueService: NotificationQueueProducer,
     webhookService: WebhookService,
     configService: ConfigService,
   ) {
-    super(notificationRepository, notificationsService, webhookService, configService);
+    super(
+      notificationRepository,
+      notificationsService,
+      notificationsQueueService,
+      webhookService,
+      configService,
+    );
   }
 
   async processSmsPlivoNotificationQueue(id: number): Promise<void> {

apps/api/src/jobs/consumers/notifications/smsTwilio-notifications.job.consumer.ts

@@ -12,6 +12,7 @@ import { Inject, Injectable, forwardRef } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { DeliveryStatus, ProviderDeliveryStatus } from 'src/common/constants/notifications';
 import { WebhookService } from 'src/modules/webhook/webhook.service';
+import { NotificationQueueProducer } from 'src/jobs/producers/notifications/notifications.job.producer';
 
 @Injectable()
 export class SmsTwilioNotificationsConsumer extends NotificationConsumer {
@@ -21,10 +22,18 @@ export class SmsTwilioNotificationsConsumer extends NotificationConsumer {
     private readonly smsTwilioService: SmsTwilioService,
     @Inject(forwardRef(() => NotificationsService))
     notificationsService: NotificationsService,
+    @Inject(forwardRef(() => NotificationQueueProducer))
+    notificationsQueueService: NotificationQueueProducer,
     webhookService: WebhookService,
     configService: ConfigService,
   ) {
-    super(notificationRepository, notificationsService, webhookService, configService);
+    super(
+      notificationRepository,
+      notificationsService,
+      notificationsQueueService,
+      webhookService,
+      configService,
+    );
   }
 
   async processSmsTwilioNotificationQueue(id: number): Promise<void> {

apps/api/src/jobs/consumers/notifications/smtp-notifications.job.consumer.ts

@@ -8,6 +8,7 @@ import { NotificationConsumer } from './notification.consumer';
 import { Inject, Injectable, forwardRef } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { WebhookService } from 'src/modules/webhook/webhook.service';
+import { NotificationQueueProducer } from 'src/jobs/producers/notifications/notifications.job.producer';
 
 @Injectable()
 export class SmtpNotificationConsumer extends NotificationConsumer {
@@ -17,10 +18,18 @@ export class SmtpNotificationConsumer extends NotificationConsumer {
     private readonly smtpService: SmtpService,
     @Inject(forwardRef(() => NotificationsService))
     notificationsService: NotificationsService,
+    @Inject(forwardRef(() => NotificationQueueProducer))
+    notificationsQueueService: NotificationQueueProducer,
     webhookService: WebhookService,
     configService: ConfigService,
   ) {
-    super(notificationRepository, notificationsService, webhookService, configService);
+    super(
+      notificationRepository,
+      notificationsService,
+      notificationsQueueService,
+      webhookService,
+      configService,
+    );
   }
 
   async processSmtpNotificationQueue(id: number): Promise<void> {

apps/api/src/jobs/consumers/notifications/vcTwilio-notifications.job.consumer.ts

@@ -12,6 +12,7 @@ import { Inject, Injectable, forwardRef } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { DeliveryStatus, ProviderDeliveryStatus } from 'src/common/constants/notifications';
 import { WebhookService } from 'src/modules/webhook/webhook.service';
+import { NotificationQueueProducer } from 'src/jobs/producers/notifications/notifications.job.producer';
 
 @Injectable()
 export class VcTwilioNotificationsConsumer extends NotificationConsumer {
@@ -21,10 +22,18 @@ export class VcTwilioNotificationsConsumer extends NotificationConsumer {
     private readonly vcTwilioService: VcTwilioService,
     @Inject(forwardRef(() => NotificationsService))
     notificationsService: NotificationsService,
+    @Inject(forwardRef(() => NotificationQueueProducer))
+    notificationsQueueService: NotificationQueueProducer,
     webhookService: WebhookService,
     configService: ConfigService,
   ) {
-    super(notificationRepository, notificationsService, webhookService, configService);
+    super(
+      notificationRepository,
+      notificationsService,
+      notificationsQueueService,
+      webhookService,
+      configService,
+    );
   }
 
   async processVcTwilioNotificationQueue(id: number): Promise<void> {

apps/api/src/jobs/consumers/notifications/wa360dialog-notifications.job.consumer.ts

@@ -10,6 +10,7 @@ import { NotificationConsumer } from './notification.consumer';
 import { Inject, Injectable, forwardRef } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { WebhookService } from 'src/modules/webhook/webhook.service';
+import { NotificationQueueProducer } from 'src/jobs/producers/notifications/notifications.job.producer';
 
 @Injectable()
 export class Wa360dialogNotificationsConsumer extends NotificationConsumer {
@@ -19,10 +20,18 @@ export class Wa360dialogNotificationsConsumer extends NotificationConsumer {
     private readonly wa360dialogService: Wa360dialogService,
     @Inject(forwardRef(() => NotificationsService))
     notificationsService: NotificationsService,
+    @Inject(forwardRef(() => NotificationQueueProducer))
+    notificationsQueueService: NotificationQueueProducer,
     webhookService: WebhookService,
     configService: ConfigService,
   ) {
-    super(notificationRepository, notificationsService, webhookService, configService);
+    super(
+      notificationRepository,
+      notificationsService,
+      notificationsQueueService,
+      webhookService,
+      configService,
+    );
   }
 
   async processWa360dialogNotificationQueue(id: number): Promise<void> {

apps/api/src/jobs/consumers/notifications/waTwilio-notifications.job.consumer.ts

@@ -12,6 +12,7 @@ import { Inject, Injectable, forwardRef } from '@nestjs/common';
 import { DeliveryStatus, ProviderDeliveryStatus } from 'src/common/constants/notifications';
 import { ConfigService } from '@nestjs/config';
 import { WebhookService } from 'src/modules/webhook/webhook.service';
+import { NotificationQueueProducer } from 'src/jobs/producers/notifications/notifications.job.producer';
 
 @Injectable()
 export class WaTwilioNotificationsConsumer extends NotificationConsumer {
@@ -21,10 +22,18 @@ export class WaTwilioNotificationsConsumer extends NotificationConsumer {
     private readonly waTwilioService: WaTwilioService,
     @Inject(forwardRef(() => NotificationsService))
     notificationsService: NotificationsService,
+    @Inject(forwardRef(() => NotificationQueueProducer))
+    notificationsQueueService: NotificationQueueProducer,
     webhookService: WebhookService,
     configService: ConfigService,
   ) {
-    super(notificationRepository, notificationsService, webhookService, configService);
+    super(
+      notificationRepository,
+      notificationsService,
+      notificationsQueueService,
+      webhookService,
+      configService,
+    );
   }
 
   async processWaTwilioNotificationQueue(id: number): Promise<void> {

apps/api/src/jobs/consumers/notifications/waTwilioBusiness-notifications.job.consumer.ts

@@ -12,6 +12,7 @@ import { Inject, Injectable, forwardRef } from '@nestjs/common';
 import { ConfigService } from '@nestjs/config';
 import { ProviderDeliveryStatus, DeliveryStatus } from 'src/common/constants/notifications';
 import { WebhookService } from 'src/modules/webhook/webhook.service';
+import { NotificationQueueProducer } from 'src/jobs/producers/notifications/notifications.job.producer';
 
 @Injectable()
 export class WaTwilioBusinessNotificationsConsumer extends NotificationConsumer {
@@ -21,10 +22,18 @@ export class WaTwilioBusinessNotificationsConsumer extends NotificationConsumer
     private readonly waTwilioBusinessService: WaTwilioBusinessService,
     @Inject(forwardRef(() => NotificationsService))
     notificationsService: NotificationsService,
+    @Inject(forwardRef(() => NotificationQueueProducer))
+    notificationsQueueService: NotificationQueueProducer,
     webhookService: WebhookService,
     configService: ConfigService,
   ) {
-    super(notificationRepository, notificationsService, webhookService, configService);
+    super(
+      notificationRepository,
+      notificationsService,
+      notificationsQueueService,
+      webhookService,
+      configService,
+    );
   }
 
   async processWaTwilioBusinessNotificationQueue(id: number): Promise<void> {

apps/api/src/modules/notifications/queues/queue.service.ts

@@ -13,6 +13,7 @@ import { VcTwilioNotificationsConsumer } from 'src/jobs/consumers/notifications/
 import { Wa360dialogNotificationsConsumer } from 'src/jobs/consumers/notifications/wa360dialog-notifications.job.consumer';
 import { WaTwilioNotificationsConsumer } from 'src/jobs/consumers/notifications/waTwilio-notifications.job.consumer';
 import { WaTwilioBusinessNotificationsConsumer } from 'src/jobs/consumers/notifications/waTwilioBusiness-notifications.job.consumer';
+import { WebhookService } from 'src/modules/webhook/webhook.service';
 
 @Injectable()
 export class QueueService {
@@ -36,6 +37,7 @@ export class QueueService {
     private readonly smsKapsystemNotificationConsumer: SmsKapsystemNotificationsConsumer,
     private readonly pushSnsNotificationConsumer: PushSnsNotificationConsumer,
     private readonly vcTwilioNotificationsConsumer: VcTwilioNotificationsConsumer,
+    protected readonly webhookService: WebhookService,
   ) {
     this.redisConfig = {
       host: this.configService.get<string>('REDIS_HOST'),
@@ -168,6 +170,19 @@ export class QueueService {
             job.data.id,
           );
           break;
+        // WEBHOOK
+        case `${QueueAction.WEBHOOK}-${ChannelType.SMTP}`:
+        case `${QueueAction.WEBHOOK}-${ChannelType.MAILGUN}`:
+        case `${QueueAction.WEBHOOK}-${ChannelType.WA_360_DAILOG}`:
+        case `${QueueAction.WEBHOOK}-${ChannelType.WA_TWILIO}`:
+        case `${QueueAction.WEBHOOK}-${ChannelType.SMS_TWILIO}`:
+        case `${QueueAction.WEBHOOK}-${ChannelType.SMS_PLIVO}`:
+        case `${QueueAction.WEBHOOK}-${ChannelType.WA_TWILIO_BUSINESS}`:
+        case `${QueueAction.WEBHOOK}-${ChannelType.SMS_KAPSYSTEM}`:
+        case `${QueueAction.WEBHOOK}-${ChannelType.PUSH_SNS}`:
+        case `${QueueAction.WEBHOOK}-${ChannelType.VC_TWILIO}`:
+          await this.webhookService.triggerWebhook(job.data.id);
+          break;
         default:
           this.logger.error(
             `Unsupported action-providerType combination: ${action}-${providerType}`,