feat(api,worker): skip sending messages outside of the subscribers schedule fixes NV-6618

apps/api/src/app/inbox/usecases/index.ts

@@ -1,4 +1,5 @@
 import {
+  GetSubscriberSchedule,
   GetSubscriberTemplatePreference,
   GetWorkflowByIdsUseCase,
   MessageInteractionService,
@@ -11,7 +12,6 @@ import { GenerateUniqueApiKey } from '../../environments-v1/usecases/generate-un
 import { ParseEventRequest } from '../../events/usecases/parse-event-request';
 import { VerifyPayload } from '../../events/usecases/verify-payload';
 import { GetSubscriberGlobalPreference } from '../../subscribers/usecases/get-subscriber-global-preference';
-import { GetSubscriberSchedule } from '../../subscribers/usecases/get-subscriber-schedule';
 import { BulkUpdatePreferences } from './bulk-update-preferences/bulk-update-preferences.usecase';
 import { DeleteAllNotifications } from './delete-all-notifications/delete-all-notifications.usecase';
 import { DeleteManyNotifications } from './delete-many-notifications/delete-many-notifications.usecase';

apps/api/src/app/inbox/usecases/session/session.spec.ts

@@ -3,6 +3,7 @@ import {
   AnalyticsService,
   CreateOrUpdateSubscriberUseCase,
   FeatureFlagsService,
+  GetSubscriberSchedule,
   PinoLogger,
   SelectIntegration,
   UpsertControlValuesUseCase,
@@ -24,7 +25,6 @@ import { AuthService } from '../../../auth/services/auth.service';
 import { GenerateUniqueApiKey } from '../../../environments-v1/usecases/generate-unique-api-key/generate-unique-api-key.usecase';
 import { CreateNovuIntegrations } from '../../../integrations/usecases/create-novu-integrations/create-novu-integrations.usecase';
 import { GetOrganizationSettings } from '../../../organization/usecases/get-organization-settings/get-organization-settings.usecase';
-import { GetSubscriberSchedule } from '../../../subscribers/usecases/get-subscriber-schedule/get-subscriber-schedule.usecase';
 import { SubscriberSessionResponseDto } from '../../dtos/subscriber-session-response.dto';
 import { AnalyticsEventsEnum } from '../../utils';
 import * as encryption from '../../utils/encryption';

apps/api/src/app/inbox/usecases/session/session.usecase.ts

@@ -11,6 +11,8 @@ import {
   CreateOrUpdateSubscriberUseCase,
   encryptApiKey,
   FeatureFlagsService,
+  GetSubscriberSchedule,
+  GetSubscriberScheduleCommand,
   generateTimestampHex,
   LogDecorator,
   PinoLogger,
@@ -59,10 +61,6 @@ import { GetOrganizationSettingsCommand } from '../../../organization/usecases/g
 import { GetOrganizationSettings } from '../../../organization/usecases/get-organization-settings/get-organization-settings.usecase';
 import { ScheduleDto } from '../../../shared/dtos/schedule';
 import { isHmacValid } from '../../../shared/helpers/is-valid-hmac';
-import {
-  GetSubscriberSchedule,
-  GetSubscriberScheduleCommand,
-} from '../../../subscribers/usecases/get-subscriber-schedule';
 import { SubscriberDto, SubscriberSessionRequestDto } from '../../dtos/subscriber-session-request.dto';
 import { SubscriberSessionResponseDto } from '../../dtos/subscriber-session-response.dto';
 import { AnalyticsEventsEnum } from '../../utils';

apps/worker/src/.env.test

@@ -92,3 +92,4 @@ IS_TRACE_LOGS_ENABLED=true
 IS_STEP_RUN_LOGS_WRITE_ENABLED=true
 IS_WORKFLOW_RUN_LOGS_WRITE_ENABLED=true
 IS_NOTIFICATION_SEVERITY_ENABLED=true
+IS_SUBSCRIBERS_SCHEDULE_ENABLED=true

apps/worker/src/app/workflow/usecases/run-job/run-job.usecase.ts

@@ -3,6 +3,9 @@ import {
   CreateExecutionDetails,
   CreateExecutionDetailsCommand,
   DetailEnum,
+  FeatureFlagsService,
+  GetSubscriberSchedule,
+  GetSubscriberScheduleCommand,
   getJobDigest,
   Instrument,
   InstrumentUsecase,
@@ -12,8 +15,20 @@ import {
   WorkflowRunService,
   WorkflowRunStatusEnum,
 } from '@novu/application-generic';
-import { JobEntity, JobRepository, JobStatusEnum, NotificationRepository } from '@novu/dal';
-import { ExecutionDetailsSourceEnum, ExecutionDetailsStatusEnum, StepTypeEnum } from '@novu/shared';
+import {
+  JobEntity,
+  JobRepository,
+  JobStatusEnum,
+  NotificationEntity,
+  NotificationRepository,
+  SubscriberRepository,
+} from '@novu/dal';
+import {
+  ExecutionDetailsSourceEnum,
+  ExecutionDetailsStatusEnum,
+  FeatureFlagsKeysEnum,
+  StepTypeEnum,
+} from '@novu/shared';
 import { setUser } from '@sentry/node';
 import { EXCEPTION_MESSAGE_ON_WEBHOOK_FILTER, PlatformException, shouldHaltOnStepFailure } from '../../../shared/utils';
 import { AddJob } from '../add-job';
@@ -23,6 +38,7 @@ import { SendMessageStatus } from '../send-message/send-message-type.usecase';
 import { SetJobAsFailedCommand } from '../update-job-status/set-job-as.command';
 import { SetJobAsFailed } from '../update-job-status/set-job-as-failed.usecase';
 import { RunJobCommand } from './run-job.command';
+import { isWithinSchedule } from './schedule-validator';
 
 const nr = require('newrelic');
 
@@ -41,7 +57,10 @@ export class RunJob {
     private stepRunRepository: StepRunRepository,
     private workflowRunService: WorkflowRunService,
     private createExecutionDetails: CreateExecutionDetails,
-    private logger: PinoLogger
+    private getSubscriberSchedule: GetSubscriberSchedule,
+    private logger: PinoLogger,
+    private subscriberRepository: SubscriberRepository,
+    private featureFlagsService: FeatureFlagsService
   ) {
     this.logger.setContext(this.constructor.name);
   }
@@ -90,7 +109,7 @@ export class RunJob {
     });
 
     let shouldQueueNextJob = true;
-    let error: any;
+    let error: Error | undefined;
 
     try {
       await this.jobRepository.updateStatus(job._environmentId, job._id, JobStatusEnum.RUNNING);
@@ -106,6 +125,63 @@ export class RunJob {
         throw new PlatformException(`Notification with id ${job._notificationId} not found`);
       }
 
+      const isSubscribersScheduleEnabled = await this.featureFlagsService.getFlag({
+        key: FeatureFlagsKeysEnum.IS_SUBSCRIBERS_SCHEDULE_ENABLED,
+        defaultValue: false,
+        organization: { _id: job._organizationId },
+        environment: { _id: job._environmentId },
+      });
+
+      if (isSubscribersScheduleEnabled && !this.shouldSkipScheduleCheck(job, notification)) {
+        const schedule = await this.getSubscriberSchedule.execute(
+          GetSubscriberScheduleCommand.create({
+            environmentId: job._environmentId,
+            organizationId: job._organizationId,
+            _subscriberId: job._subscriberId,
+          })
+        );
+
+        const subscriber = await this.subscriberRepository.findOne(
+          {
+            _id: job._subscriberId,
+            _environmentId: job._environmentId,
+            _organizationId: job._organizationId,
+          },
+          'timezone',
+          { readPreference: 'secondaryPreferred' }
+        );
+
+        if (!isWithinSchedule(schedule, new Date(), subscriber?.timezone)) {
+          this.logger.info(
+            {
+              jobId: job._id,
+              subscriberId: job.subscriberId,
+              stepType: job.type,
+            },
+            "The step was skipped as it fell outside the subscriber's schedule"
+          );
+
+          await this.jobRepository.updateStatus(job._environmentId, job._id, JobStatusEnum.CANCELED);
+
+          await this.stepRunRepository.create(job, {
+            status: JobStatusEnum.CANCELED,
+          });
+
+          await this.createExecutionDetails.execute(
+            CreateExecutionDetailsCommand.create({
+              ...CreateExecutionDetailsCommand.getDetailsFromJob(job),
+              detail: DetailEnum.SKIPPED_STEP_OUTSIDE_OF_THE_SCHEDULE,
+              source: ExecutionDetailsSourceEnum.INTERNAL,
+              status: ExecutionDetailsStatusEnum.SUCCESS,
+              isTest: false,
+              isRetry: false,
+            })
+          );
+
+          return;
+        }
+      }
+
       if (this.isUnsnoozeJob(job)) {
         await this.processUnsnoozeJob.execute(
           ProcessUnsnoozeJobCommand.create({
@@ -191,15 +267,15 @@ export class RunJob {
           status: JobStatusEnum.CANCELED,
         });
       }
-    } catch (caughtError: any) {
-      error = caughtError;
+    } catch (caughtError: unknown) {
+      error = caughtError as Error;
       await this.stepRunRepository.create(job, {
         status: JobStatusEnum.FAILED,
         errorCode: 'execution_error',
-        errorMessage: caughtError.message,
+        errorMessage: error.message,
       });
 
-      if (shouldHaltOnStepFailure(job) && !this.shouldBackoff(caughtError)) {
+      if (shouldHaltOnStepFailure(job) && !this.shouldBackoff(error)) {
         await this.jobRepository.cancelPendingJobs({
           transactionId: job.transactionId,
           _environmentId: job._environmentId,
@@ -208,7 +284,7 @@ export class RunJob {
         });
       }
 
-      if (shouldHaltOnStepFailure(job) || this.shouldBackoff(caughtError)) {
+      if (shouldHaltOnStepFailure(job) || this.shouldBackoff(error)) {
         shouldQueueNextJob = false;
       }
       throw caughtError;
@@ -316,7 +392,7 @@ export class RunJob {
             subscriberId: nextJob._subscriberId,
           });
         }
-      } catch (error: any) {
+      } catch (error: unknown) {
         if (!nextJob) {
           // Fallback: update workflow run status if nextJob is unexpectedly missing
           // (should not occur due to prior nextJob check in loop)
@@ -336,10 +412,10 @@ export class RunJob {
             organizationId: nextJob._organizationId,
             userId: nextJob._userId,
           }),
-          error
+          error as Error
         );
 
-        if (shouldHaltOnStepFailure(nextJob) && !this.shouldBackoff(error)) {
+        if (shouldHaltOnStepFailure(nextJob) && !this.shouldBackoff(error as Error)) {
           // Update workflow run status based on step runs when halting on step failure
           await this.workflowRunService.updateDeliveryLifecycle({
             notificationId: nextJob._notificationId,
@@ -356,7 +432,7 @@ export class RunJob {
           });
         }
 
-        if (shouldHaltOnStepFailure(nextJob) || this.shouldBackoff(error)) {
+        if (shouldHaltOnStepFailure(nextJob) || this.shouldBackoff(error as Error)) {
           return;
         }
 
@@ -369,7 +445,7 @@ export class RunJob {
     }
   }
 
-  private assignLogger(job) {
+  private assignLogger(job: JobEntity) {
     try {
       if (this.logger) {
         this.logger.assign({
@@ -442,4 +518,20 @@ export class RunJob {
   public shouldBackoff(error: Error): boolean {
     return error?.message?.includes(EXCEPTION_MESSAGE_ON_WEBHOOK_FILTER);
   }
+
+  private shouldSkipScheduleCheck(job: JobEntity, notification: NotificationEntity): boolean {
+    // always deliver in-app messages or critical messages
+    // let trigger,digest and delay finish their execution
+    if (
+      job.type === StepTypeEnum.TRIGGER ||
+      job.type === StepTypeEnum.IN_APP ||
+      job.type === StepTypeEnum.DELAY ||
+      job.type === StepTypeEnum.DIGEST ||
+      notification.critical
+    ) {
+      return true;
+    }
+
+    return false;
+  }
 }