Refactor the broadcast method to utilize the new queue

apps/api/src/app/events/events.controller.ts

@@ -8,7 +8,7 @@ import {
   TriggerRecipientSubscriber,
   TriggerTenantContext,
 } from '@novu/shared';
-import { MapTriggerRecipients, SendTestEmail, SendTestEmailCommand } from '@novu/application-generic';
+import { SendTestEmail, SendTestEmailCommand } from '@novu/application-generic';
 
 import {
   BulkTriggerEventDto,
@@ -35,7 +35,6 @@ import { DataBooleanDto } from '../shared/dtos/data-wrapper-dto';
 @ApiTags('Events')
 export class EventsController {
   constructor(
-    private mapTriggerRecipients: MapTriggerRecipients,
     private cancelDelayedUsecase: CancelDelayed,
     private triggerEventToAll: TriggerEventToAll,
     private sendTestEmail: SendTestEmail,
@@ -119,7 +118,7 @@ export class EventsController {
     @Body() body: TriggerEventToAllRequestDto
   ): Promise<TriggerEventResponseDto> {
     const transactionId = body.transactionId || uuidv4();
-    const mappedActor = body.actor ? this.mapActor(body.actor) : null;
+    const mappedActor = this.mapActor(body.actor);
     const mappedTenant = body.tenant ? this.mapTenant(body.tenant) : null;
 
     return this.triggerEventToAll.execute(
@@ -187,7 +186,11 @@ export class EventsController {
   private mapActor(actor?: TriggerRecipientSubscriber | null): ISubscribersDefine | null {
     if (!actor) return null;
 
-    return this.mapTriggerRecipients.mapSubscriber(actor);
+    if (typeof actor === 'string') {
+      return { subscriberId: actor };
+    }
+
+    return actor;
   }
 
   private mapTenant(tenant?: TriggerTenantContext | null): ITenantDefine | null {

apps/api/src/app/events/events.module.ts

@@ -2,25 +2,12 @@ import { Module } from '@nestjs/common';
 import { TerminusModule } from '@nestjs/terminus';
 
 import {
-  AddJob,
-  AddDelayJob,
-  MergeOrCreateDigest,
   CreateExecutionDetails,
-  CreateNotificationJobs,
-  DigestFilterSteps,
-  DigestFilterStepsBackoff,
-  DigestFilterStepsRegular,
-  DigestFilterStepsTimed,
   EventsDistributedLockService,
   GetNovuProviderCredentials,
-  ProcessSubscriber,
-  ProcessTenant,
-  QueuesModule,
   StorageHelperService,
   SendTestEmail,
-  StoreSubscriberJobs,
-  TriggerEvent,
-  MapTriggerRecipients,
+  BaseApiQueuesModule,
 } from '@novu/application-generic';
 
 import { EventsController } from './events.controller';
@@ -39,24 +26,11 @@ import { LayoutsModule } from '../layouts/layouts.module';
 import { TenantModule } from '../tenant/tenant.module';
 
 const PROVIDERS = [
-  AddJob,
-  AddDelayJob,
-  MergeOrCreateDigest,
   CreateExecutionDetails,
-  CreateNotificationJobs,
-  DigestFilterSteps,
-  DigestFilterStepsBackoff,
-  DigestFilterStepsRegular,
-  DigestFilterStepsTimed,
   GetNovuProviderCredentials,
   StorageHelperService,
   EventsDistributedLockService,
-  ProcessSubscriber,
-  ProcessTenant,
   SendTestEmail,
-  StoreSubscriberJobs,
-  TriggerEvent,
-  MapTriggerRecipients,
 ];
 
 @Module({
@@ -73,7 +47,7 @@ const PROVIDERS = [
     TopicsModule,
     LayoutsModule,
     TenantModule,
-    QueuesModule,
+    BaseApiQueuesModule,
   ],
   controllers: [EventsController],
   providers: [...PROVIDERS, ...USE_CASES],

apps/api/src/app/events/usecases/parse-event-request/parse-event-request.command.ts

@@ -1,5 +1,5 @@
 import { IsDefined, IsString, IsOptional, ValidateNested } from 'class-validator';
-import { TriggerRecipients, TriggerRecipientSubscriber, TriggerTenantContext } from '@novu/shared';
+import { AddressingTypeEnum, TriggerRecipients, TriggerRecipientSubscriber, TriggerTenantContext } from '@novu/shared';
 
 import { EnvironmentWithUserCommand } from '../../../shared/commands/project.command';
 
@@ -14,8 +14,8 @@ export class ParseEventRequestCommand extends EnvironmentWithUserCommand {
   @IsDefined()
   overrides: Record<string, Record<string, unknown>>;
 
-  @IsDefined()
-  to: TriggerRecipients;
+  @IsOptional()
+  to?: TriggerRecipients;
 
   @IsString()
   @IsOptional()
@@ -28,4 +28,7 @@ export class ParseEventRequestCommand extends EnvironmentWithUserCommand {
   @IsOptional()
   @ValidateNested()
   tenant?: TriggerTenantContext | null;
+
+  @IsOptional()
+  addressingType?: AddressingTypeEnum;
 }

apps/api/src/app/events/usecases/process-bulk-trigger/process-bulk-trigger.usecase.ts

@@ -1,6 +1,5 @@
 import { Injectable } from '@nestjs/common';
 import { TriggerEventStatusEnum } from '@novu/shared';
-import { MapTriggerRecipients } from '@novu/application-generic';
 
 import { ProcessBulkTriggerCommand } from './process-bulk-trigger.command';
 
@@ -10,7 +9,7 @@ import { ParseEventRequest } from '../parse-event-request/parse-event-request.us
 
 @Injectable()
 export class ProcessBulkTrigger {
-  constructor(private parseEventRequest: ParseEventRequest, private mapTriggerRecipients: MapTriggerRecipients) {}
+  constructor(private parseEventRequest: ParseEventRequest) {}
 
   async execute(command: ProcessBulkTriggerCommand) {
     const results: TriggerEventResponseDto[] = [];

apps/api/src/app/events/usecases/trigger-event-to-all/trigger-event-to-all.usecase.ts

@@ -1,62 +1,34 @@
 import { Injectable } from '@nestjs/common';
-import * as _ from 'lodash';
-import { SubscriberEntity, SubscriberRepository } from '@novu/dal';
-import { TriggerEvent, TriggerEventCommand } from '@novu/application-generic';
-import { TriggerEventStatusEnum } from '@novu/shared';
+import { SubscriberRepository } from '@novu/dal';
+import { AddressingTypeEnum, TriggerEventStatusEnum } from '@novu/shared';
 
 import { TriggerEventToAllCommand } from './trigger-event-to-all.command';
+import { ParseEventRequest, ParseEventRequestCommand } from '../parse-event-request';
 
 @Injectable()
 export class TriggerEventToAll {
-  constructor(private triggerEvent: TriggerEvent, private subscriberRepository: SubscriberRepository) {}
+  constructor(private subscriberRepository: SubscriberRepository, private parseEventRequest: ParseEventRequest) {}
 
   public async execute(command: TriggerEventToAllCommand) {
-    const batchSize = 500;
-    let list: SubscriberEntity[] = [];
-
-    for await (const subscriber of this.subscriberRepository.findBatch(
-      {
-        _environmentId: command.environmentId,
-        _organizationId: command.organizationId,
-      },
-      'subscriberId',
-      {},
-      batchSize
-    )) {
-      list.push(subscriber);
-      if (list.length === batchSize) {
-        await this.trigger(command, list);
-        list = [];
-      }
-    }
-
-    if (list.length > 0) {
-      await this.trigger(command, list);
-    }
-
-    return {
-      acknowledged: true,
-      status: TriggerEventStatusEnum.PROCESSED,
-      transactionId: command.transactionId,
-    };
-  }
-
-  private async trigger(command: TriggerEventToAllCommand, list: SubscriberEntity[]) {
-    await this.triggerEvent.execute(
-      TriggerEventCommand.create({
+    await this.parseEventRequest.execute(
+      ParseEventRequestCommand.create({
         userId: command.userId,
         environmentId: command.environmentId,
         organizationId: command.organizationId,
         identifier: command.identifier,
-        payload: command.payload,
-        to: list.map((item) => ({
-          subscriberId: item.subscriberId,
-        })),
+        payload: command.payload || {},
+        addressingType: AddressingTypeEnum.BROADCAST,
         transactionId: command.transactionId,
-        overrides: command.overrides,
+        overrides: command.overrides || {},
         actor: command.actor,
         tenant: command.tenant,
       })
     );
+
+    return {
+      acknowledged: true,
+      status: TriggerEventStatusEnum.PROCESSED,
+      transactionId: command.transactionId,
+    };
   }
 }

apps/api/src/app/health/health.controller.ts

@@ -17,7 +17,6 @@ export class HealthController {
     private healthCheckService: HealthCheckService,
     private cacheHealthIndicator: CacheServiceHealthIndicator,
     private dalHealthIndicator: DalServiceHealthIndicator,
-    private standardQueueHealthIndicator: StandardQueueServiceHealthIndicator,
     private workflowQueueHealthIndicator: WorkflowQueueServiceHealthIndicator
   ) {}
 
@@ -26,7 +25,6 @@ export class HealthController {
   healthCheck(): Promise<HealthCheckResult> {
     const checks: HealthIndicatorFunction[] = [
       async () => this.dalHealthIndicator.isHealthy(),
-      async () => this.standardQueueHealthIndicator.isHealthy(),
       async () => this.workflowQueueHealthIndicator.isHealthy(),
       async () => {
         return {

apps/api/src/app/health/health.module.ts

@@ -1,12 +1,13 @@
 import { Module } from '@nestjs/common';
 import { TerminusModule } from '@nestjs/terminus';
 
-import { HealthController } from './health.controller';
+import { BaseApiQueuesModule } from '@novu/application-generic';
 
+import { HealthController } from './health.controller';
 import { SharedModule } from '../shared/shared.module';
 
 @Module({
-  imports: [SharedModule, TerminusModule],
+  imports: [SharedModule, TerminusModule, BaseApiQueuesModule],
   controllers: [HealthController],
   providers: [],
 })

apps/api/src/app/inbound-parse/inbound-parse.module.ts

@@ -1,22 +1,21 @@
 import { MiddlewareConsumer, Module, NestModule } from '@nestjs/common';
-import { CompileTemplate, QueuesModule } from '@novu/application-generic';
+import { BaseApiQueuesModule, CompileTemplate } from '@novu/application-generic';
 
 import { USE_CASES } from './usecases';
 import { InboundParseController } from './inbound-parse.controller';
 import { InboundParseQueueService } from './services/inbound-parse.queue.service';
 import { GetMxRecord } from './usecases/get-mx-record/get-mx-record.usecase';
-import { InboundEmailParse } from './usecases/inbound-email-parse/inbound-email-parse.usecase';
 
 import { SharedModule } from '../shared/shared.module';
 import { AuthModule } from '../auth/auth.module';
 
 const PROVIDERS = [InboundParseQueueService, GetMxRecord, CompileTemplate];
 
 @Module({
-  imports: [SharedModule, AuthModule, QueuesModule],
+  imports: [SharedModule, AuthModule, BaseApiQueuesModule],
   controllers: [InboundParseController],
   providers: [...PROVIDERS, ...USE_CASES],
-  exports: [...USE_CASES, QueuesModule],
+  exports: [...USE_CASES],
 })
 export class InboundParseModule implements NestModule {
   configure(consumer: MiddlewareConsumer): MiddlewareConsumer | void {}

apps/api/src/app/shared/shared.module.ts

@@ -100,9 +100,8 @@ const PROVIDERS = [
         version: packageJson.version,
       })
     ),
-    QueuesModule,
   ],
   providers: [...PROVIDERS],
-  exports: [...PROVIDERS, LoggerModule, QueuesModule],
+  exports: [...PROVIDERS, LoggerModule],
 })
 export class SharedModule {}

apps/api/src/app/widgets/widgets.module.ts

@@ -1,5 +1,7 @@
 import { Module } from '@nestjs/common';
 
+import { BaseApiQueuesModule } from '@novu/application-generic';
+
 import { USE_CASES } from './usecases';
 import { WidgetsController } from './widgets.controller';
 import { SharedModule } from '../shared/shared.module';
@@ -8,7 +10,7 @@ import { SubscribersModule } from '../subscribers/subscribers.module';
 import { IntegrationModule } from '../integrations/integrations.module';
 
 @Module({
-  imports: [SharedModule, SubscribersModule, AuthModule, IntegrationModule],
+  imports: [SharedModule, SubscribersModule, AuthModule, IntegrationModule, BaseApiQueuesModule],
   providers: [...USE_CASES],
   exports: [...USE_CASES],
   controllers: [WidgetsController],

libs/shared/src/types/events/index.ts

@@ -60,3 +60,8 @@ export interface ITopic {
 }
 
 export type TriggerRecipientTopics = ITopic[];
+
+export enum AddressingTypeEnum {
+  BROADCAST = 'broadcast',
+  MULTICAST = 'multicast',
+}

packages/application-generic/src/modules/index.ts

@@ -1 +1,2 @@
 export { QueuesModule } from './queues.module';
+export { BaseApiQueuesModule } from './queues.module';

packages/application-generic/src/modules/queues.module.ts

@@ -67,3 +67,19 @@ const PROVIDERS: Provider[] = [
   exports: [...PROVIDERS],
 })
 export class QueuesModule {}
+
+const APP_PROVIDERS: Provider[] = [
+  InboundParseQueue,
+  InboundParseWorker,
+  InboundParseQueueServiceHealthIndicator,
+  WebSocketsQueueService,
+  WebSocketsQueueServiceHealthIndicator,
+  WorkflowQueueService,
+  WorkflowQueueServiceHealthIndicator,
+];
+
+@Module({
+  providers: [...APP_PROVIDERS],
+  exports: [...APP_PROVIDERS],
+})
+export class BaseApiQueuesModule {}

packages/application-generic/src/usecases/trigger-event/trigger-event.command.ts

@@ -3,8 +3,13 @@ import {
   IsString,
   IsOptional,
   ValidateNested,
+  IsEnum,
 } from 'class-validator';
-import { ISubscribersDefine, ITenantDefine } from '@novu/shared';
+import {
+  AddressingTypeEnum,
+  ISubscribersDefine,
+  ITenantDefine,
+} from '@novu/shared';
 
 import { EnvironmentWithUserCommand } from '../../commands';
 
@@ -19,8 +24,8 @@ export class TriggerEventCommand extends EnvironmentWithUserCommand {
   @IsDefined()
   overrides: Record<string, Record<string, unknown>>;
 
-  @IsDefined()
-  to: ISubscribersDefine[];
+  @IsOptional()
+  to?: ISubscribersDefine[];
 
   @IsString()
   @IsDefined()
@@ -33,4 +38,7 @@ export class TriggerEventCommand extends EnvironmentWithUserCommand {
   @IsOptional()
   @ValidateNested()
   tenant?: ITenantDefine | null;
+
+  @IsOptional()
+  addressingType?: AddressingTypeEnum;
 }