refactor(app-gen,worker): move add job to worker

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

@@ -4,7 +4,6 @@ import { Injectable, Logger } from '@nestjs/common';
 import { ObservabilityBackgroundTransactionEnum } from '@novu/shared';
 import {
  getSubscriberProcessWorkerOptions,
- SubscriberJobBound,
  SubscriberProcessWorkerService,
  PinoLogger,
  storage,
@@ -15,6 +14,8 @@ import {
  IProcessSubscriberDataDto,
 } from '@novu/application-generic';
 
+import { SubscriberJobBound } from '../usecases/subscriber-job-bound/subscriber-job-bound.usecase';
+
 const LOG_CONTEXT = 'SubscriberProcessWorker';
 
 @Injectable()

packages/application-generic/src/usecases/add-job/add-delay-job.usecase.ts → apps/worker/src/app/workflow/usecases/add-job/add-delay-job.usecase.ts

@@ -1,22 +1,17 @@
-import { forwardRef, Inject, Injectable, Logger } from '@nestjs/common';
+import { forwardRef, Inject, Injectable } from '@nestjs/common';
 
 import { JobRepository, JobStatusEnum } from '@novu/dal';
+import { DelayTypeEnum, ExecutionDetailsSourceEnum, ExecutionDetailsStatusEnum, StepTypeEnum } from '@novu/shared';
 import {
- DelayTypeEnum,
- ExecutionDetailsSourceEnum,
- ExecutionDetailsStatusEnum,
- StepTypeEnum,
-} from '@novu/shared';
-
-import { ApiException } from '../../utils/exceptions';
-import { AddJobCommand } from './add-job.command';
-import { CalculateDelayService } from '../../services';
-import { InstrumentUsecase } from '../../instrumentation';
-import { DetailEnum } from '../create-execution-details';
-import {
+ ApiException,
+ CalculateDelayService,
+ DetailEnum,
  ExecutionLogRoute,
  ExecutionLogRouteCommand,
-} from '../execution-log-route';
+ InstrumentUsecase,
+} from '@novu/application-generic';
+
+import { AddJobCommand } from './add-job.command';
 
 @Injectable()
 export class AddDelayJob {
@@ -56,11 +51,7 @@ export class AddDelayJob {
  : undefined,
  });
 
- await this.jobRepository.updateStatus(
- command.environmentId,
- data._id,
- JobStatusEnum.DELAYED
- );
+ await this.jobRepository.updateStatus(command.environmentId, data._id, JobStatusEnum.DELAYED);
  } catch (error: any) {
  await this.executionLogRoute.execute(
  ExecutionLogRouteCommand.create({
@@ -74,11 +65,7 @@ export class AddDelayJob {
  })
  );
 
- await this.jobRepository.updateStatus(
- command.environmentId,
- data._id,
- JobStatusEnum.CANCELED
- );
+ await this.jobRepository.updateStatus(command.environmentId, data._id, JobStatusEnum.CANCELED);
 
  throw error;
  }

packages/application-generic/src/usecases/add-job/add-job.command.ts → apps/worker/src/app/workflow/usecases/add-job/add-job.command.ts

@@ -1,7 +1,6 @@
 import { IsDefined } from 'class-validator';
 import { JobEntity } from '@novu/dal';
-
-import { EnvironmentWithUserCommand } from '../../commands';
+import { EnvironmentWithUserCommand } from '@novu/application-generic';
 
 export class AddJobCommand extends EnvironmentWithUserCommand {
  @IsDefined()

packages/application-generic/src/usecases/add-job/add-job.usecase.ts → apps/worker/src/app/workflow/usecases/add-job/add-job.usecase.ts

@@ -12,31 +12,25 @@ import { AddDelayJob } from './add-delay-job.usecase';
 import { MergeOrCreateDigestCommand } from './merge-or-create-digest.command';
 import { MergeOrCreateDigest } from './merge-or-create-digest.usecase';
 import { AddJobCommand } from './add-job.command';
+import { validateDigest } from './validation';
+import { ModuleRef } from '@nestjs/core';
 import {
+ CalculateDelayService,
  ConditionsFilter,
  ConditionsFilterCommand,
  DetailEnum,
-} from '../../usecases';
-import {
- CalculateDelayService,
- JobsOptions,
- StandardQueueService,
-} from '../../services';
-import { LogDecorator } from '../../logging';
-import { InstrumentUsecase } from '../../instrumentation';
-import { validateDigest } from './validation';
-import {
  ExecutionLogRoute,
  ExecutionLogRouteCommand,
-} from '../execution-log-route';
-import { ModuleRef } from '@nestjs/core';
-import {
- ExecuteOutput,
  IChimeraDigestResponse,
+ IFilterVariables,
+ InstrumentUsecase,
  IUseCaseInterfaceInline,
+ JobsOptions,
+ LogDecorator,
  requireInject,
-} from '../../utils/require-inject';
-import { IFilterVariables } from '../../utils/filter-processing-details';
+ StandardQueueService,
+ ExecuteOutput,
+} from '@novu/application-generic';
 
 export enum BackoffStrategiesEnum {
  WEBHOOK_FILTER_BACKOFF = 'webhookFilterBackoff',
@@ -73,26 +67,19 @@ export class AddJob {
  Logger.debug(`Job contents for job ${job._id}`, job, LOG_CONTEXT);
 
  if (!job) {
- Logger.warn(
- `Job ${job._id} was null in both the input and search`,
- LOG_CONTEXT
- );
+ Logger.warn(`Job was null in both the input and search`, LOG_CONTEXT);
 
  return;
  }
 
- Logger.log(
- `Scheduling New Job ${job._id} of type: ${job.type}`,
-  LOG_CONTEXT
- );
+ Logger.log(`Scheduling New Job ${job._id} of type: ${job.type}`, LOG_CONTEXT);
+
+ let digestAmount: number | undefined;
+ let delayAmount: number | undefined = undefined;
 
  let filtered = false;
  let filterVariables: IFilterVariables | undefined;
- if (
- [StepTypeEnum.DELAY, StepTypeEnum.DIGEST].includes(
- job.type as StepTypeEnum
- )
- ) {
+ if ([StepTypeEnum.DELAY, StepTypeEnum.DIGEST].includes(job.type as StepTypeEnum)) {
  const shouldRun = await this.conditionsFilter.filter(
  ConditionsFilterCommand.create({
  filters: job.step.filters || [],
@@ -106,105 +93,89 @@ export class AddJob {
 
  filterVariables = shouldRun.variables;
  filtered = !shouldRun.passed;
- }
-
- let digestAmount: number | undefined;
- let digestCreationResult: DigestCreationResultEnum | undefined;
- if (job.type === StepTypeEnum.DIGEST) {
- const resonateResponse = await this.resonateUsecase.execute<
- AddJobCommand & { variables: IFilterVariables },
- ExecuteOutput<IChimeraDigestResponse>
- >({
- ...command,
- variables: filterVariables,
- });
 
- validateDigest(job);
-
- digestAmount = this.calculateDelayService.calculateDelay({
- stepMetadata: job.digest,
- payload: job.payload,
- overrides: job.overrides,
- // TODO: Remove fallback after other digest types are implemented.
- chimeraResponse: resonateResponse
- ? { type: DigestTypeEnum.REGULAR, ...resonateResponse.outputs }
- : undefined,
- });
+ let digestCreationResult: DigestCreationResultEnum | undefined;
+ if (job.type === StepTypeEnum.DIGEST) {
+ const resonateResponse = await this.resonateUsecase.execute<
+ AddJobCommand & { variables: IFilterVariables },
+ ExecuteOutput<IChimeraDigestResponse>
+ >({
+ ...command,
+ variables: filterVariables,
+ });
 
- Logger.debug(`Digest step amount is: ${digestAmount}`, LOG_CONTEXT);
+  validateDigest(job);
 
- digestCreationResult = await this.mergeOrCreateDigestUsecase.execute(
- MergeOrCreateDigestCommand.create({
- job,
- filtered,
- chimeraData: resonateResponse?.outputs,
- })
- );
+ digestAmount = this.calculateDelayService.calculateDelay({
+ stepMetadata: job.digest,
+ payload: job.payload,
+ overrides: job.overrides,
+ chimeraResponse: this.fallbackToRegularDigest(resonateResponse?.outputs),
+ });
 
- if (digestCreationResult === DigestCreationResultEnum.MERGED) {
- Logger.log('Digest was merged, queueing next job', LOG_CONTEXT);
+ Logger.debug(`Digest step amount is: ${digestAmount}`, LOG_CONTEXT);
 
- return;
- }
+ digestCreationResult = await this.mergeOrCreateDigestUsecase.execute(
+ MergeOrCreateDigestCommand.create({
+ job,
+ filtered,
+ chimeraData: resonateResponse?.outputs,
+ })
+ );
 
- if (digestCreationResult === DigestCreationResultEnum.SKIPPED) {
- const nextJobToSchedule = await this.jobRepository.findOne({
- _environmentId: command.environmentId,
- _parentId: job._id,
- });
+ if (digestCreationResult === DigestCreationResultEnum.MERGED) {
+ Logger.log('Digest was merged, queueing next job', LOG_CONTEXT);
 
- if (!nextJobToSchedule) {
  return;
  }
 
- await this.execute({
- userId: job._userId,
- environmentId: job._environmentId,
- organizationId: command.organizationId,
- jobId: nextJobToSchedule._id,
- job: nextJobToSchedule,
- });
+ if (digestCreationResult === DigestCreationResultEnum.SKIPPED) {
+ const nextJobToSchedule = await this.jobRepository.findOne({
+ _environmentId: command.environmentId,
+ _parentId: job._id,
+ });
 
- return;
- }
- }
+  if (!nextJobToSchedule) {
+  return;
+  }
 
- let delayAmount: number | undefined = undefined;
+ await this.execute({
+ userId: job._userId,
+ environmentId: job._environmentId,
+ organizationId: command.organizationId,
+ jobId: nextJobToSchedule._id,
+ job: nextJobToSchedule,
+ });
 
- if (job.type === StepTypeEnum.DELAY) {
- const resonateResponse = await this.resonateUsecase.execute<
- AddJobCommand & { variables: IFilterVariables },
- ExecuteOutput<IChimeraDigestResponse>
- >({
- ...command,
- variables: filterVariables,
- });
+ return;
+ }
+ }
 
- command.chimeraResponse = resonateResponse;
- delayAmount = await this.addDelayJob.execute(command);
+ if (job.type === StepTypeEnum.DELAY) {
+ const resonateResponse = await this.resonateUsecase.execute<
+ AddJobCommand & { variables: IFilterVariables },
+ ExecuteOutput<IChimeraDigestResponse>
+ >({
+ ...command,
+ variables: filterVariables,
+ });
 
- Logger.debug(`Delay step Amount is: ${delayAmount}`, LOG_CONTEXT);
+ command.chimeraResponse = resonateResponse;
+ delayAmount = await this.addDelayJob.execute(command);
 
- if (delayAmount === undefined) {
- Logger.warn(
- `Delay Amount does not exist on a delay job ${job._id}`,
- LOG_CONTEXT
- );
+ Logger.debug(`Delay step Amount is: ${delayAmount}`, LOG_CONTEXT);
 
- return;
+ if (delayAmount === undefined) {
+ Logger.warn(`Delay Amount does not exist on a delay job ${job._id}`, LOG_CONTEXT);
+
+ return;
+ }
  }
  }
 
  if (digestAmount === undefined && delayAmount === undefined) {
- Logger.verbose(
- `Updating status to queued for job ${job._id}`,
- LOG_CONTEXT
- );
- await this.jobRepository.updateStatus(
- command.environmentId,
- job._id,
- JobStatusEnum.QUEUED
- );
+ Logger.verbose(`Updating status to queued for job ${job._id}`, LOG_CONTEXT);
+ await this.jobRepository.updateStatus(command.environmentId, job._id, JobStatusEnum.QUEUED);
  }
 
  await this.executionLogRoute.execute(
@@ -218,18 +189,30 @@ export class AddJob {
  })
  );
 
- const delay = filtered ? 0 : digestAmount ?? delayAmount;
+ const delay = (filtered ? 0 : digestAmount ?? delayAmount) ?? 0;
 
  if ((digestAmount || delayAmount) && filtered) {
- Logger.verbose(
- `Delay for job ${job._id} will be 0 because job was filtered`,
- LOG_CONTEXT
- );
+ Logger.verbose(`Delay for job ${job._id} will be 0 because job was filtered`, LOG_CONTEXT);
  }
 
  await this.queueJob(job, delay);
  }
 
+ /*
+ * Fallback to regular digest type.
+ * This is a temporary solution until other digest types are implemented.
+ */
+ private fallbackToRegularDigest(outputs: IChimeraDigestResponse | undefined): IChimeraDigestResponse | undefined {
+ let resonateResponseOutput: IChimeraDigestResponse | undefined = undefined;
+
+ if (outputs) {
+ const { type, ...resonateResponseOutputsOmitType } = outputs;
+ resonateResponseOutput = { type: DigestTypeEnum.REGULAR, ...resonateResponseOutputsOmitType };
+ }
+
+ return resonateResponseOutput;
+ }
+
  public async queueJob(job: JobEntity, delay: number) {
  Logger.verbose(`Adding Job ${job._id} to Queue`, LOG_CONTEXT);
  const stepContainsWebhookFilter = this.stepContainsFilter(job, 'webhook');
@@ -250,11 +233,7 @@ export class AddJob {
  _userId: job._userId,
  };
 
- Logger.verbose(
- jobData,
- 'Going to add a minimal job in Standard Queue',
- LOG_CONTEXT
- );
+ Logger.verbose(jobData, 'Going to add a minimal job in Standard Queue', LOG_CONTEXT);
 
  await this.standardQueueService.add({
  name: job._id,
@@ -276,10 +255,7 @@ export class AddJob {
  await this.executionLogRoute.execute(
  ExecutionLogRouteCommand.create({
  ...ExecutionLogRouteCommand.getDetailsFromJob(job),
- detail:
- job.type === StepTypeEnum.DELAY
- ? DetailEnum.STEP_DELAYED
- : DetailEnum.STEP_DIGESTED,
+ detail: job.type === StepTypeEnum.DELAY ? DetailEnum.STEP_DELAYED : DetailEnum.STEP_DIGESTED,
  source: ExecutionDetailsSourceEnum.INTERNAL,
  status: ExecutionDetailsStatusEnum.PENDING,
  isTest: false,

packages/application-generic/src/usecases/add-job/merge-or-create-digest.command.ts → apps/worker/src/app/workflow/usecases/add-job/merge-or-create-digest.command.ts

@@ -1,8 +1,7 @@
 import { IsDefined, IsOptional } from 'class-validator';
-import { JobEntity } from '@novu/dal';
 
-import { BaseCommand } from '../../commands/base.command';
-import { IChimeraDigestResponse } from '../../utils/require-inject';
+import { JobEntity } from '@novu/dal';
+import { BaseCommand, IChimeraDigestResponse } from '@novu/application-generic';
 
 export class MergeOrCreateDigestCommand extends BaseCommand {
  @IsDefined()