feat(api-service): Organization switch flag

.cspell.json

@@ -311,6 +311,7 @@
     "Keymap",
     "keyrings",
     "keysize",
+    "KILLSWITCH",
     "kitterma",
     "Kitterman",
     "klass",

apps/api/src/app/auth/services/passport/apikey.strategy.ts

@@ -1,4 +1,4 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, ServiceUnavailableException } from '@nestjs/common';
 import { PassportStrategy } from '@nestjs/passport';
 import { FeatureFlagsService, HttpRequestHeaderKeysEnum } from '@novu/application-generic';
 import { ApiAuthSchemeEnum, FeatureFlagsKeysEnum, UserSessionData } from '@novu/shared';
@@ -52,6 +52,8 @@ export class ApiKeyStrategy extends PassportStrategy(HeaderAPIKeyStrategy) {
     if (isLruCacheEnabled) {
       const cached = apiKeyUserCache.get(hashedApiKey);
       if (cached) {
+        await this.checkKillSwitch(cached);
+
         return cached;
       }
 
@@ -63,11 +65,15 @@ export class ApiKeyStrategy extends PassportStrategy(HeaderAPIKeyStrategy) {
 
     const fetchPromise = this.authService
       .getUserByApiKey(apiKey)
-      .then((user) => {
+      .then(async (user) => {
         if (user && isLruCacheEnabled) {
           apiKeyUserCache.set(hashedApiKey, user);
         }
 
+        if (user) {
+          await this.checkKillSwitch(user);
+        }
+
         return user;
       })
       .finally(() => {
@@ -82,4 +88,18 @@ export class ApiKeyStrategy extends PassportStrategy(HeaderAPIKeyStrategy) {
 
     return fetchPromise;
   }
+
+  private async checkKillSwitch(user: UserSessionData): Promise<void> {
+    const isKillSwitchEnabled = await this.featureFlagsService.getFlag({
+      key: FeatureFlagsKeysEnum.IS_ORG_KILLSWITCH_FLAG_ENABLED,
+      defaultValue: false,
+      organization: { _id: user.organizationId },
+      environment: { _id: user.environmentId },
+      component: 'api',
+    });
+
+    if (isKillSwitchEnabled) {
+      throw new ServiceUnavailableException('Service temporarily unavailable for this organization');
+    }
+  }
 }

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

@@ -1,10 +1,11 @@
-import { Body, Controller, Delete, Param, Post, Req, Scope } from '@nestjs/common';
+import { Body, Controller, Delete, Param, Post, Req, Scope, ServiceUnavailableException } from '@nestjs/common';
 import { ApiExcludeEndpoint, ApiOperation, ApiTags } from '@nestjs/swagger';
-import { RequirePermissions, ResourceCategory } from '@novu/application-generic';
+import { FeatureFlagsService, RequirePermissions, ResourceCategory } from '@novu/application-generic';
 import {
   AddressingTypeEnum,
   ApiRateLimitCategoryEnum,
   ApiRateLimitCostEnum,
+  FeatureFlagsKeysEnum,
   PermissionsEnum,
   ResourceEnum,
   TriggerRequestCategoryEnum,
@@ -64,9 +65,24 @@ export class EventsController {
     private triggerEventToAll: TriggerEventToAll,
     private sendTestEmail: SendTestEmail,
     private parseEventRequest: ParseEventRequest,
-    private processBulkTriggerUsecase: ProcessBulkTrigger
+    private processBulkTriggerUsecase: ProcessBulkTrigger,
+    private featureFlagsService: FeatureFlagsService
   ) {}
 
+  private async checkKillSwitch(user: UserSessionData): Promise<void> {
+    const isKillSwitchEnabled = await this.featureFlagsService.getFlag({
+      key: FeatureFlagsKeysEnum.IS_ORG_KILLSWITCH_FLAG_ENABLED,
+      defaultValue: false,
+      organization: { _id: user.organizationId },
+      environment: { _id: user.environmentId },
+      component: 'trigger',
+    });
+
+    if (isKillSwitchEnabled) {
+      throw new ServiceUnavailableException('Service temporarily unavailable for this organization');
+    }
+  }
+
   @KeylessAccessible()
   @ExternalApiAccessible()
   @Post('/trigger')
@@ -91,6 +107,8 @@ export class EventsController {
     @Req() req: RequestWithReqId,
     @Body() body: TriggerEventRequestDto
   ): Promise<TriggerEventResponseDto> {
+    await this.checkKillSwitch(user);
+
     const result = await this.parseEventRequest.execute(
       ParseEventRequestMulticastCommand.create({
         userId: user._id,
@@ -140,6 +158,8 @@ export class EventsController {
     @Body() body: BulkTriggerEventDto,
     @Req() req: RequestWithReqId
   ): Promise<TriggerEventResponseDto[]> {
+    await this.checkKillSwitch(user);
+
     return this.processBulkTriggerUsecase.execute(
       ProcessBulkTriggerCommand.create({
         userId: user._id,
@@ -178,6 +198,8 @@ export class EventsController {
     @Body() body: TriggerEventToAllRequestDto,
     @Req() req: RequestWithReqId
   ): Promise<TriggerEventResponseDto> {
+    await this.checkKillSwitch(user);
+
     const transactionId = body.transactionId || uuidv4();
 
     return this.triggerEventToAll.execute(

apps/worker/src/app/workflow/services/standard.worker.spec.ts

@@ -142,6 +142,7 @@ describe('Standard Worker', () => {
       WorkflowInMemoryProviderService
     );
     const organizationRepository = moduleRef.get<CommunityOrganizationRepository>(CommunityOrganizationRepository);
+    const featureFlagsService = moduleRef.get<FeatureFlagsService>(FeatureFlagsService);
 
     standardWorker = new StandardWorker(
       handleLastFailedJob,
@@ -150,7 +151,8 @@ describe('Standard Worker', () => {
       webhookFilterBackoffStrategy,
       workflowInMemoryProviderService,
       organizationRepository,
-      jobRepository
+      jobRepository,
+      featureFlagsService
     );
   });
 

apps/worker/src/app/workflow/services/standard.worker.ts

@@ -1,6 +1,7 @@
 import { forwardRef, Inject, Injectable, Logger } from '@nestjs/common';
 import {
   BullMqService,
+  FeatureFlagsService,
   getStandardWorkerOptions,
   IStandardDataDto,
   Job,
@@ -12,7 +13,7 @@ import {
   WorkflowInMemoryProviderService,
 } from '@novu/application-generic';
 import { CommunityOrganizationRepository, JobRepository } from '@novu/dal';
-import { JobStatusEnum, ObservabilityBackgroundTransactionEnum } from '@novu/shared';
+import { FeatureFlagsKeysEnum, JobStatusEnum, ObservabilityBackgroundTransactionEnum } from '@novu/shared';
 import {
   HandleLastFailedJob,
   HandleLastFailedJobCommand,
@@ -38,7 +39,8 @@ export class StandardWorker extends StandardWorkerService {
     @Inject(forwardRef(() => WorkflowInMemoryProviderService))
     public workflowInMemoryProviderService: WorkflowInMemoryProviderService,
     private organizationRepository: CommunityOrganizationRepository,
-    private jobRepository: JobRepository
+    private jobRepository: JobRepository,
+    private featureFlagsService: FeatureFlagsService
   ) {
     super(new BullMqService(workflowInMemoryProviderService));
 
@@ -93,10 +95,29 @@ export class StandardWorker extends StandardWorkerService {
     };
   }
 
+  private async isKillSwitchEnabled(data: IStandardDataDto): Promise<boolean> {
+    return this.featureFlagsService.getFlag({
+      key: FeatureFlagsKeysEnum.IS_ORG_KILLSWITCH_FLAG_ENABLED,
+      defaultValue: false,
+      organization: { _id: data._organizationId },
+      environment: { _id: data._environmentId },
+      component: 'worker',
+    });
+  }
+
   private getWorkerProcessor() {
     return async ({ data }: { data: IStandardDataDto }) => {
+      const isKillSwitchEnabled = await this.isKillSwitchEnabled(data);
+
+      if (isKillSwitchEnabled) {
+        Logger.log(`Kill switch enabled for organizationId ${data._organizationId}. Skipping job.`, LOG_CONTEXT);
+
+        return;
+      }
+
       if (data.skipProcessing) {
         Logger.log(`Skipping job ${data._id} - skipProcessing flag is set,`, LOG_CONTEXT);
+
         return;
       }
 

apps/worker/src/app/workflow/services/subscriber-process.worker.ts

@@ -1,6 +1,7 @@
 import { Injectable, Logger } from '@nestjs/common';
 import {
   BullMqService,
+  FeatureFlagsService,
   getSubscriberProcessWorkerOptions,
   IProcessSubscriberDataDto,
   PinoLogger,
@@ -11,7 +12,7 @@ import {
   WorkflowInMemoryProviderService,
 } from '@novu/application-generic';
 import { CommunityOrganizationRepository } from '@novu/dal';
-import { ObservabilityBackgroundTransactionEnum } from '@novu/shared';
+import { FeatureFlagsKeysEnum, ObservabilityBackgroundTransactionEnum } from '@novu/shared';
 import { SubscriberJobBound } from '../usecases/subscriber-job-bound/subscriber-job-bound.usecase';
 
 const nr = require('newrelic');
@@ -23,15 +24,34 @@ export class SubscriberProcessWorker extends SubscriberProcessWorkerService {
   constructor(
     private subscriberJobBoundUsecase: SubscriberJobBound,
     public workflowInMemoryProviderService: WorkflowInMemoryProviderService,
-    private organizationRepository: CommunityOrganizationRepository
+    private organizationRepository: CommunityOrganizationRepository,
+    private featureFlagsService: FeatureFlagsService
   ) {
     super(new BullMqService(workflowInMemoryProviderService));
 
     this.initWorker(this.getWorkerProcessor(), this.getWorkerOpts());
   }
 
+  private async isKillSwitchEnabled(data: IProcessSubscriberDataDto): Promise<boolean> {
+    return this.featureFlagsService.getFlag({
+      key: FeatureFlagsKeysEnum.IS_ORG_KILLSWITCH_FLAG_ENABLED,
+      defaultValue: false,
+      organization: { _id: data.organizationId },
+      environment: { _id: data.environmentId },
+      component: 'worker',
+    });
+  }
+
   public getWorkerProcessor() {
     return async ({ data }: { data: IProcessSubscriberDataDto }) => {
+      const isKillSwitchEnabled = await this.isKillSwitchEnabled(data);
+
+      if (isKillSwitchEnabled) {
+        Logger.log(`Kill switch enabled for organizationId ${data.organizationId}. Skipping job.`, LOG_CONTEXT);
+
+        return;
+      }
+
       const organizationExists = await this.organizationExist(data);
 
       if (!organizationExists) {

apps/worker/src/app/workflow/services/workflow.worker.ts

@@ -1,6 +1,7 @@
 import { Injectable } from '@nestjs/common';
 import {
   BullMqService,
+  FeatureFlagsService,
   getWorkflowWorkerOptions,
   IWorkflowDataDto,
   PinoLogger,
@@ -13,7 +14,7 @@ import {
   WorkflowWorkerService,
 } from '@novu/application-generic';
 import { CommunityOrganizationRepository } from '@novu/dal';
-import { ObservabilityBackgroundTransactionEnum } from '@novu/shared';
+import { FeatureFlagsKeysEnum, ObservabilityBackgroundTransactionEnum } from '@novu/shared';
 
 const nr = require('newrelic');
 
@@ -23,7 +24,8 @@ export class WorkflowWorker extends WorkflowWorkerService {
     private triggerEventUsecase: TriggerEvent,
     public workflowInMemoryProviderService: WorkflowInMemoryProviderService,
     private organizationRepository: CommunityOrganizationRepository,
-    private logger: PinoLogger
+    private logger: PinoLogger,
+    private featureFlagsService: FeatureFlagsService
   ) {
     super(new BullMqService(workflowInMemoryProviderService));
     this.logger.setContext(this.constructor.name);
@@ -34,8 +36,26 @@ export class WorkflowWorker extends WorkflowWorkerService {
     return getWorkflowWorkerOptions();
   }
 
+  private async isKillSwitchEnabled(data: IWorkflowDataDto): Promise<boolean> {
+    return this.featureFlagsService.getFlag({
+      key: FeatureFlagsKeysEnum.IS_ORG_KILLSWITCH_FLAG_ENABLED,
+      defaultValue: false,
+      organization: { _id: data.organizationId },
+      environment: { _id: data.environmentId },
+      component: 'worker',
+    });
+  }
+
   private getWorkerProcessor(): WorkerProcessor {
     return async ({ data }: { data: IWorkflowDataDto }) => {
+      const isKillSwitchEnabled = await this.isKillSwitchEnabled(data);
+
+      if (isKillSwitchEnabled) {
+        this.logger.warn(`Kill switch enabled for organizationId ${data.organizationId}. Skipping job.`);
+
+        return;
+      }
+
       const organizationExists = await this.organizationExist(data);
 
       if (!organizationExists) {

packages/shared/src/types/feature-flags.ts

@@ -87,6 +87,7 @@ export enum FeatureFlagsKeysEnum {
   IS_BILLING_USAGE_CLICKHOUSE_ENABLED = 'IS_BILLING_USAGE_CLICKHOUSE_ENABLED',
   IS_BILLING_USAGE_CLICKHOUSE_SHADOW_ENABLED = 'IS_BILLING_USAGE_CLICKHOUSE_SHADOW_ENABLED',
   IS_BILLING_USAGE_DETAILED_DIAGNOSTICS_ENABLED = 'IS_BILLING_USAGE_DETAILED_DIAGNOSTICS_ENABLED',
+  IS_ORG_KILLSWITCH_FLAG_ENABLED = 'IS_ORG_KILLSWITCH_FLAG_ENABLED',
 
   // String flags
   CF_SCHEDULER_MODE = 'CF_SCHEDULER_MODE', // Values: "off" | "shadow" | "live" | "complete"