[pull] next from novuhq:next

apps/api/src/app/inbox/usecases/bulk-update-preferences/bulk-update-preferences.spec.ts

@@ -379,6 +379,7 @@ describe('BulkUpdatePreferences', () => {
   });
 
   it('should update multiple workflow preferences in parallel', async () => {
+    const environment = { _id: 'env-1' } as any;
     const command = BulkUpdatePreferencesCommand.create({
       environmentId: 'env-1',
       organizationId: 'org-1',
@@ -399,7 +400,7 @@ describe('BulkUpdatePreferences', () => {
 
     subscriberRepositoryMock.findBySubscriberId.resolves(mockedSubscriber);
     notificationTemplateRepositoryMock.findForBulkPreferences.resolves([mockedWorkflow1, mockedWorkflow2]);
-    environmentRepositoryMock.findOne.resolves({ _id: 'env-1' } as any);
+    environmentRepositoryMock.findOne.resolves(environment);
 
     updatePreferencesUsecaseMock.execute.onFirstCall().resolves(mockedInboxPreference1);
     updatePreferencesUsecaseMock.execute.onSecondCall().resolves(mockedInboxPreference2);
@@ -418,6 +419,9 @@ describe('BulkUpdatePreferences', () => {
       in_app: true,
       email: false,
     });
+    expect(firstCallArgs.workflow).to.equal(mockedWorkflow1);
+    expect(firstCallArgs.subscriber).to.equal(mockedSubscriber);
+    expect(firstCallArgs.environment).to.equal(environment);
 
     const secondCallArgs = updatePreferencesUsecaseMock.execute.secondCall.args[0];
     expect(secondCallArgs).to.include({
@@ -433,6 +437,59 @@ describe('BulkUpdatePreferences', () => {
     expect(result).to.deep.equal([mockedInboxPreference1, mockedInboxPreference2]);
   });
 
+  it('should update at most five workflow preferences concurrently', async () => {
+    const workflows = Array.from({ length: 6 }, (_, index) => ({
+      ...mockedWorkflow1,
+      _id: index.toString(16).padStart(24, '0'),
+      triggers: [{ identifier: `test-trigger-${index}` }],
+    }));
+    const command = BulkUpdatePreferencesCommand.create({
+      environmentId: 'env-1',
+      organizationId: 'org-1',
+      subscriberId: 'test-mockSubscriber',
+      preferences: workflows.map((workflow) => ({
+        workflowId: workflow._id,
+        in_app: true,
+      })),
+    });
+    let activeUpdates = 0;
+    let maxActiveUpdates = 0;
+    let releaseFirstBatch: () => void = () => {};
+    const firstBatchGate = new Promise<void>((resolve) => {
+      releaseFirstBatch = resolve;
+    });
+
+    subscriberRepositoryMock.findBySubscriberId.resolves(mockedSubscriber);
+    notificationTemplateRepositoryMock.findForBulkPreferences.resolves(workflows);
+    environmentRepositoryMock.findOne.resolves({ _id: 'env-1' } as any);
+    updatePreferencesUsecaseMock.execute.callsFake(async () => {
+      activeUpdates += 1;
+      maxActiveUpdates = Math.max(maxActiveUpdates, activeUpdates);
+
+      if (updatePreferencesUsecaseMock.execute.callCount <= 5) {
+        await firstBatchGate;
+      }
+
+      activeUpdates -= 1;
+
+      return mockedInboxPreference1;
+    });
+
+    const execution = bulkUpdatePreferences.execute(command);
+
+    while (updatePreferencesUsecaseMock.execute.callCount < 5) {
+      await new Promise((resolve) => setImmediate(resolve));
+    }
+
+    expect(updatePreferencesUsecaseMock.execute.callCount).to.equal(5);
+    releaseFirstBatch();
+    const result = await execution;
+
+    expect(updatePreferencesUsecaseMock.execute.callCount).to.equal(6);
+    expect(maxActiveUpdates).to.equal(5);
+    expect(result).to.have.length(6);
+  });
+
   it('should support lookup by workflow identifier', async () => {
     const command = BulkUpdatePreferencesCommand.create({
       environmentId: 'env-1',

apps/api/src/app/inbox/usecases/bulk-update-preferences/bulk-update-preferences.usecase.ts

@@ -17,6 +17,7 @@ import { UpdatePreferences } from '../update-preferences/update-preferences.usec
 import { BulkUpdatePreferencesCommand } from './bulk-update-preferences.command';
 
 const MAX_BULK_LIMIT = 100;
+const UPDATE_BATCH_SIZE = 5;
 
 @Injectable()
 export class BulkUpdatePreferences {
@@ -101,44 +102,54 @@ export class BulkUpdatePreferences {
       _id: command.environmentId,
     });
 
-    const updatePromises = Array.from(workflowPreferencesMap.entries()).map(
-      async ([workflowId, { preference, workflow }]) => {
-        const isUpdatingSubscriptionPreference =
-          preference.subscriptionIdentifier &&
-          (typeof preference.enabled !== 'undefined' || typeof preference.condition !== 'undefined');
-
-        return this.updatePreferencesUsecase.execute(
-          UpdatePreferencesCommand.create({
-            organizationId: command.organizationId,
-            subscriberId: command.subscriberId,
-            environmentId: command.environmentId,
-            contextKeys,
-            level: PreferenceLevelEnum.TEMPLATE,
-            subscriptionIdentifier: preference.subscriptionIdentifier,
-            ...(isUpdatingSubscriptionPreference && {
-              all: {
-                ...(typeof preference.enabled !== 'undefined' && { enabled: preference.enabled }),
-                ...(typeof preference.condition !== 'undefined' && { condition: preference.condition }),
+    const workflowPreferenceEntries = Array.from(workflowPreferencesMap.entries());
+    const updatedPreferences: InboxPreference[] = [];
+
+    for (let batchStart = 0; batchStart < workflowPreferenceEntries.length; batchStart += UPDATE_BATCH_SIZE) {
+      const batch = workflowPreferenceEntries.slice(batchStart, batchStart + UPDATE_BATCH_SIZE);
+      const batchResults = await Promise.all(
+        batch.map(async ([workflowId, { preference, workflow }]) => {
+          const isUpdatingSubscriptionPreference =
+            preference.subscriptionIdentifier &&
+            (typeof preference.enabled !== 'undefined' || typeof preference.condition !== 'undefined');
+
+          return this.updatePreferencesUsecase.execute(
+            UpdatePreferencesCommand.create(
+              {
+                organizationId: command.organizationId,
+                subscriberId: command.subscriberId,
+                environmentId: command.environmentId,
+                contextKeys,
+                level: PreferenceLevelEnum.TEMPLATE,
+                subscriptionIdentifier: preference.subscriptionIdentifier,
+                ...(isUpdatingSubscriptionPreference && {
+                  all: {
+                    ...(typeof preference.enabled !== 'undefined' && { enabled: preference.enabled }),
+                    ...(typeof preference.condition !== 'undefined' && { condition: preference.condition }),
+                  },
+                }),
+                chat: preference.chat,
+                email: preference.email,
+                in_app: preference.in_app,
+                push: preference.push,
+                sms: preference.sms,
+                tool: preference.tool,
+                workflowIdOrIdentifier: workflowId,
+                includeInactiveChannels: false,
               },
-            }),
-            chat: preference.chat,
-            email: preference.email,
-            in_app: preference.in_app,
-            push: preference.push,
-            sms: preference.sms,
-            tool: preference.tool,
-            workflowIdOrIdentifier: workflowId,
-            workflow,
-            includeInactiveChannels: false,
-            subscriber,
-            // biome-ignore lint/style/noNonNullAssertion: environment is always found
-            environment: environment!,
-          })
-        );
-      }
-    );
-
-    const updatedPreferences = await Promise.all(updatePromises);
+              {
+                workflow,
+                subscriber,
+                // biome-ignore lint/style/noNonNullAssertion: environment is always found
+                environment: environment!,
+              }
+            )
+          );
+        })
+      );
+
+      updatedPreferences.push(...batchResults);
+    }
 
     return updatedPreferences;
   }

apps/api/src/app/inbox/usecases/update-preferences/update-preferences.spec.ts

@@ -154,6 +154,7 @@ describe('UpdatePreferences', () => {
     getWorkflowByIdsUsecase.execute.resolves(mockedWorkflow);
 
     const result = await updatePreferences.execute(command);
+    const templatePreferenceCommand = getSubscriberTemplatePreferenceUsecase.execute.firstCall.args[0];
 
     expect(result).to.deep.equal({
       level: command.level,
@@ -168,6 +169,8 @@ describe('UpdatePreferences', () => {
         severity: mockedWorkflow.severity,
       },
     });
+    expect(templatePreferenceCommand.template).to.equal(mockedWorkflow);
+    expect(templatePreferenceCommand.subscriber).to.equal(mockedSubscriber);
   });
 
   it('should throw NotFoundException when the subscriptionIdentifier is not owned by the authenticated subscriber', async () => {

apps/api/src/app/inbox/usecases/update-preferences/update-preferences.usecase.ts

@@ -261,15 +261,19 @@ export class UpdatePreferences {
 
     if (command.level === PreferenceLevelEnum.TEMPLATE && command.workflowIdOrIdentifier && workflow) {
       const { preference } = await this.getSubscriberTemplatePreferenceUsecase.execute(
-        GetSubscriberTemplatePreferenceCommand.create({
-          organizationId: command.organizationId,
-          subscriberId: command.subscriberId,
-          environmentId: command.environmentId,
-          template: workflow,
-          subscriber,
-          includeInactiveChannels: command.includeInactiveChannels,
-          contextKeys: command.contextKeys,
-        } as GetSubscriberTemplatePreferenceCommand)
+        GetSubscriberTemplatePreferenceCommand.create(
+          {
+            organizationId: command.organizationId,
+            subscriberId: command.subscriberId,
+            environmentId: command.environmentId,
+            includeInactiveChannels: command.includeInactiveChannels,
+            contextKeys: command.contextKeys,
+          } as GetSubscriberTemplatePreferenceCommand,
+          {
+            template: workflow,
+            subscriber,
+          }
+        )
       );
 
       return {

apps/api/src/app/subscribers-v2/subscribers.controller.ts

@@ -65,6 +65,7 @@ import {
   GetSubscriberGlobalPreferenceCommand,
 } from '../subscribers/usecases/get-subscriber-global-preference';
 import { assertGetPreferencesEnabled } from '../subscribers/utils/assert-get-preferences-enabled';
+import { assertPreferencesUpdateEnabled } from '../subscribers/utils/assert-preferences-update-enabled';
 import { ListSubscriberSubscriptionsQueryDto } from '../topics-v2/dtos/list-subscriber-subscriptions-query.dto';
 import { ListTopicSubscriptionsResponseDto } from '../topics-v2/dtos/list-topic-subscriptions-response.dto';
 import { ListSubscriberSubscriptionsCommand } from '../topics-v2/usecases/list-subscriber-subscriptions/list-subscriber-subscriptions.command';
@@ -371,6 +372,8 @@ export class SubscribersController {
     @Param('subscriberId') subscriberId: string,
     @Body() body: BulkUpdateSubscriberPreferencesDto
   ): Promise<GetPreferencesResponseDto[]> {
+    await assertPreferencesUpdateEnabled(this.featureFlagsService, user.organizationId, user.environmentId);
+
     const preferences = body.preferences.map((preference) => ({
       workflowId: preference.workflowId,
       ...preference.channels,

apps/api/src/app/subscribers/utils/assert-preferences-update-enabled.spec.ts

@@ -0,0 +1,42 @@
+import { ServiceUnavailableException } from '@nestjs/common';
+import { FeatureFlagsService } from '@novu/application-generic';
+import { FeatureFlagsKeysEnum } from '@novu/shared';
+import { expect } from 'chai';
+import sinon from 'sinon';
+import { assertPreferencesUpdateEnabled } from './assert-preferences-update-enabled';
+
+describe('assertPreferencesUpdateEnabled', () => {
+  afterEach(() => {
+    sinon.restore();
+  });
+
+  it('should allow preference updates when the organization killswitch is disabled', async () => {
+    const featureFlagsService = sinon.createStubInstance(FeatureFlagsService);
+    featureFlagsService.getFlag.resolves(false);
+
+    await assertPreferencesUpdateEnabled(featureFlagsService as any, 'org-1', 'env-1');
+
+    expect(
+      featureFlagsService.getFlag.calledOnceWithExactly({
+        key: FeatureFlagsKeysEnum.IS_ORG_KILLSWITCH_FLAG_ENABLED,
+        defaultValue: false,
+        organization: { _id: 'org-1' },
+        environment: { _id: 'env-1' },
+        component: 'preferences',
+      })
+    ).to.be.true;
+  });
+
+  it('should reject preference updates when the organization killswitch is enabled', async () => {
+    const featureFlagsService = sinon.createStubInstance(FeatureFlagsService);
+    featureFlagsService.getFlag.resolves(true);
+
+    try {
+      await assertPreferencesUpdateEnabled(featureFlagsService as any, 'org-1', 'env-1');
+      expect.fail('Should throw an exception');
+    } catch (error) {
+      expect(error).to.be.instanceOf(ServiceUnavailableException);
+      expect(error.message).to.equal('Service temporarily unavailable for this organization');
+    }
+  });
+});

apps/api/src/app/subscribers/utils/assert-preferences-update-enabled.ts

@@ -0,0 +1,21 @@
+import { ServiceUnavailableException } from '@nestjs/common';
+import { FeatureFlagsService } from '@novu/application-generic';
+import { FeatureFlagsKeysEnum } from '@novu/shared';
+
+export async function assertPreferencesUpdateEnabled(
+  featureFlagsService: FeatureFlagsService,
+  organizationId: string,
+  environmentId: string
+): Promise<void> {
+  const isPreferencesDisabled = await featureFlagsService.getFlag({
+    key: FeatureFlagsKeysEnum.IS_ORG_KILLSWITCH_FLAG_ENABLED,
+    defaultValue: false,
+    organization: { _id: organizationId },
+    environment: { _id: environmentId },
+    component: 'preferences',
+  });
+
+  if (isPreferencesDisabled) {
+    throw new ServiceUnavailableException('Service temporarily unavailable for this organization');
+  }
+}

libs/application-generic/src/usecases/get-subscriber-template-preference/get-subscriber-template-preference.command.ts

@@ -1,11 +1,9 @@
 import { NotificationTemplateEntity, SubscriberEntity } from '@novu/dal';
 import { ITenantDefine } from '@novu/shared';
-import { IsBoolean, IsDefined, IsNotEmpty, IsOptional } from 'class-validator';
+import { IsBoolean, IsDefined, IsOptional } from 'class-validator';
 import { EnvironmentWithSubscriber } from '../../commands';
 
 export class GetSubscriberTemplatePreferenceCommand extends EnvironmentWithSubscriber {
-  @IsNotEmpty()
-  @IsDefined()
   template: NotificationTemplateEntity;
 
   @IsOptional()

libs/application-generic/src/usecases/get-subscriber-template-preference/get-subscriber-template-preference.usecase.ts

@@ -42,6 +42,10 @@ export class GetSubscriberTemplatePreference {
 
   @InstrumentUsecase()
   async execute(command: GetSubscriberTemplatePreferenceCommand): Promise<ISubscriberPreferenceResponse> {
+    if (!command.template) {
+      throw new BadRequestException('Template is required');
+    }
+
     const subscriber: Pick<SubscriberEntity, '_id'> | null = command.subscriber ?? (await this.getSubscriber(command));
 
     const initialChannels = await this.getChannels(command);