diff --git a/apps/api/src/app/inbox/usecases/bulk-update-preferences/bulk-update-preferences.spec.ts b/apps/api/src/app/inbox/usecases/bulk-update-preferences/bulk-update-preferences.spec.ts index 752bf9f3c76..ce48fa718d6 100644 --- a/apps/api/src/app/inbox/usecases/bulk-update-preferences/bulk-update-preferences.spec.ts +++ b/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((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', diff --git a/apps/api/src/app/inbox/usecases/bulk-update-preferences/bulk-update-preferences.usecase.ts b/apps/api/src/app/inbox/usecases/bulk-update-preferences/bulk-update-preferences.usecase.ts index 5e001bce177..9bcaf989d7e 100644 --- a/apps/api/src/app/inbox/usecases/bulk-update-preferences/bulk-update-preferences.usecase.ts +++ b/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; } diff --git a/apps/api/src/app/inbox/usecases/update-preferences/update-preferences.spec.ts b/apps/api/src/app/inbox/usecases/update-preferences/update-preferences.spec.ts index e184e6e0b38..a9f278b455b 100644 --- a/apps/api/src/app/inbox/usecases/update-preferences/update-preferences.spec.ts +++ b/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 () => { diff --git a/apps/api/src/app/inbox/usecases/update-preferences/update-preferences.usecase.ts b/apps/api/src/app/inbox/usecases/update-preferences/update-preferences.usecase.ts index 20121a1d9e2..59ac92ec6c4 100644 --- a/apps/api/src/app/inbox/usecases/update-preferences/update-preferences.usecase.ts +++ b/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 { diff --git a/apps/api/src/app/subscribers-v2/subscribers.controller.ts b/apps/api/src/app/subscribers-v2/subscribers.controller.ts index 9535ea75fad..5b8f06079a2 100644 --- a/apps/api/src/app/subscribers-v2/subscribers.controller.ts +++ b/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 { + await assertPreferencesUpdateEnabled(this.featureFlagsService, user.organizationId, user.environmentId); + const preferences = body.preferences.map((preference) => ({ workflowId: preference.workflowId, ...preference.channels, diff --git a/apps/api/src/app/subscribers/utils/assert-preferences-update-enabled.spec.ts b/apps/api/src/app/subscribers/utils/assert-preferences-update-enabled.spec.ts new file mode 100644 index 00000000000..1ac57c0e51e --- /dev/null +++ b/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'); + } + }); +}); diff --git a/apps/api/src/app/subscribers/utils/assert-preferences-update-enabled.ts b/apps/api/src/app/subscribers/utils/assert-preferences-update-enabled.ts new file mode 100644 index 00000000000..4a9e618b56a --- /dev/null +++ b/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 { + 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'); + } +} diff --git a/libs/application-generic/src/usecases/get-subscriber-template-preference/get-subscriber-template-preference.command.ts b/libs/application-generic/src/usecases/get-subscriber-template-preference/get-subscriber-template-preference.command.ts index fb199517b0e..2f5e7c4e942 100644 --- a/libs/application-generic/src/usecases/get-subscriber-template-preference/get-subscriber-template-preference.command.ts +++ b/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() diff --git a/libs/application-generic/src/usecases/get-subscriber-template-preference/get-subscriber-template-preference.usecase.ts b/libs/application-generic/src/usecases/get-subscriber-template-preference/get-subscriber-template-preference.usecase.ts index 0b7cdd0a9ce..6770037c2cb 100644 --- a/libs/application-generic/src/usecases/get-subscriber-template-preference/get-subscriber-template-preference.usecase.ts +++ b/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 { + if (!command.template) { + throw new BadRequestException('Template is required'); + } + const subscriber: Pick | null = command.subscriber ?? (await this.getSubscriber(command)); const initialChannels = await this.getChannels(command);