Ensure that our queue services get validated object data #4843

djabarovgeorge · 2023-11-14T09:46:10Z

What change does this PR introduce?

At the moment we do not validate the data we pass queue services for example SubscriberProcessQueueService.

Suggestions:
What we could do is create a middleware function 'add' in the child class, for example, SubscriberProcessQueueService with command and after then pass it to the parent 'add'.

Why was this change needed?

We could provide an invalid object to the queue services by mistake.

Other information and DOD's (Screenshots)

The remaining things that still need to be done:

Refactor the job data object to be defined, at the moment we allow to pass undefined data. IMO it was made in order to create metric cron jobs that do not pass any data, however, this option adds complexity to the code and could lead to a failure point where someone won't pass the data object. Instead, we could add minimal data on the metric identifier.
Make a discovery in order to make the base queue service move generic by getting the type from its child classes, this way we won't need to duplicate the methods in the services. in order to do so we will need to remove the mapping of the bull mq in the queue service and move it to the bull mq service.
Refactor so that all of the workers will execute the creation of the Command. was not sure if it should be done in this PR or not, because it could be a breaking change.
Align all of DTO's params to use align internal conventions, meaning if it is novu internal id it should have a prefix of _ at the moment we sometimes use it and sometimes not. for example userId _userId. This will be a breaking change that will require backward combability support so I'd prefer to do it in separate dedicated PR.

linear · 2023-11-14T09:46:13Z

NV-3012 Ensure that our queue services get validated object data

What?

At the moment we do not validate the data we pass queue services for example SubscriberProcessQueueService.

Suggestions:
What we could do is create a middleware function 'add' in the child class, for example, SubscriberProcessQueueService with command and after then pass it to the parent 'add'.

Why? (Context)

We could provide an invalid object by mistake.

Definition of Done

djabarovgeorge · 2023-11-14T09:49:32Z

apps/api/src/app/events/usecases/parse-event-request/parse-event-request.usecase.ts

-    if ((command as ParseEventRequestMulticastCommand).to?.length > 0) {
-      jobData = jobData as ParseEventRequestMulticastCommand;
-      jobData.to = (command as ParseEventRequestMulticastCommand).to;


Done by me this month, it is not needed as it will be destructured above in ...command,.
More than that i regret that i created two types of one usecase command, Apologies in advance, i will revisit it in order to think of a better way.

djabarovgeorge · 2023-11-14T10:40:03Z

apps/api/src/app/inbound-parse/services/inbound-parse.queue.service.ts

+    return async ({ data }: { data: IInboundParseDataDto }) => {
      Logger.verbose({ data }, 'Processing the inbound parsed email', LOG_CONTEXT);
      await this.emailParseUsecase.execute(InboundEmailParseCommand.create({ ...data }));


I wonder if we should do the execution of Command create in this PR or not. it could be a breaking change.

djabarovgeorge · 2023-11-14T10:42:24Z

apps/api/src/app/inbound-parse/usecases/inbound-email-parse/inbound-email-parse.command.ts

-export interface IHeaders {
-  'content-type': string;
-  from: string;
-  to: string;
-  subject: string;
-  'message-id': string;
-  date: string;
-  'mime-version': string;
-}
-
-export interface IFrom {
-  address: string;
-  name: string;
-}
-
-export interface ITo {
-  address: string;
-  name: string;
-}
-
-export interface ITlsOptions {
-  name: string;
-  standardName: string;
-  version: string;
-}
-
-export interface IMailFrom {
-  address: string;
-  args: boolean;
-}
-
-export interface IRcptTo {
-  address: string;
-  args: boolean;
-}
-
-export interface IEnvelope {
-  mailFrom: IMailFrom;
-  rcptTo: IRcptTo[];
-}
-
-export interface IConnection {
-  id: string;
-  remoteAddress: string;
-  remotePort: number;
-  clientHostname: string;
-  openingCommand: string;
-  hostNameAppearsAs: string;
-  xClient: any;
-  xForward: any;
-  transmissionType: string;
-  tlsOptions: ITlsOptions;
-  envelope: IEnvelope;
-  transaction: number;
-  mailPath: string;
-}
-
-export interface IEnvelopeFrom {
-  address: string;
-  args: boolean;
-}
-
-export interface IEnvelopeTo {
-  address: string;
-  args: boolean;
-}


Extracted to application-generic so we could reuse it.

djabarovgeorge · 2023-11-14T10:42:41Z

apps/api/src/app/widgets/usecases/mark-all-messages-as/mark-all-messages-as.usecase.ts

-    const countQuery = isUnreadCountChanged ? { read: false } : { seen: false };
-
-    const count = await this.messageRepository.getCount(
-      command.environmentId,
-      subscriber._id,
-      ChannelTypeEnum.IN_APP,
-      countQuery
-    );


Was not used.

djabarovgeorge · 2023-11-14T10:44:13Z

apps/inbound-mail/src/server/inbound-mail.service.spec.ts

-      await inboundMailService.inboundParseQueueService.add(jobId, jobData, _organizationId);
+      await inboundMailService.inboundParseQueueService.add({
+        name: jobId,
+        data: jobData as unknown as IInboundParseDataDto,


Make for the test, IInboundParseDataDto contained a lot of attributes that needed to be added. for now, added casting as those attributes were not tested.

djabarovgeorge · 2023-11-14T10:47:29Z

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

@@ -45,7 +41,7 @@ export class StandardWorker extends StandardWorkerService implements INovuWorker

    this.initWorker(this.getWorkerProcessor(), this.getWorkerOptions());

-    this.worker.on('failed', async (job: Job<IJobData, void, string>, error: Error): Promise<void> => {
+    this.worker.on('failed', async (job: Job<IStandardDataDto, void, string>, error: Error): Promise<void> => {


we will need either merge IJobData and IStandardDataDto, or connect them, as they share the same interface.

djabarovgeorge · 2023-11-14T10:48:30Z

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

+      if (!message) {
+        throw new Error('Job data is missing required fields' + JSON.stringify(data));
+      }


Should not happen. if it does we need to throw it as no message exists.

djabarovgeorge · 2023-11-14T10:49:27Z

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


    if (!environmentId || !jobId || !organizationId || !userId) {
-      const message = job.payload.message;
+      const message = data.payload?.message;


This object is structure data.payload?.message deprecated, i could not locate any place in the code that passes this data.

djabarovgeorge · 2023-11-14T10:52:43Z

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

@@ -44,7 +45,7 @@ export class WorkflowWorker extends WorkflowWorkerService implements INovuWorker

            storage.run(new Store(PinoLogger.root), () => {
              _this.triggerEventUsecase
-                .execute(data)
+                .execute(data as TriggerEventCommand)


We should create a command here, but before and solve this issue of complex command interface https://github.com/novuhq/novu/pull/4843/files#r1392301028

djabarovgeorge · 2023-11-14T10:54:04Z

apps/ws/src/socket/services/web-socket.worker.ts

@@ -37,14 +39,16 @@ export class WebSocketWorker extends WebSocketsWorkerService implements INovuWor
          'WS Service',
          function () {
            const transaction = nr.getTransaction();
+            const { data: jobData } = job;
+            const data = jobData as IWebSocketDataDto;


I am not sure what is the type of job above therefore i created this mapping so we could at least get the data object right here.

djabarovgeorge · 2023-11-14T10:56:08Z

packages/application-generic/src/dtos/index.ts

Created those DTO's of our queue-worker communication for us to use.

djabarovgeorge · 2023-11-14T10:59:40Z

packages/application-generic/src/dtos/process-subscriber-job.dto.ts

+export interface IProcessSubscriberJobDto extends IJobParams {
+  data?: IProcessSubscriberDataDto;
+}
+
+export interface IProcessSubscriberBulkJobDto extends IBulkJobParams {
+  data: IProcessSubscriberDataDto;
+}


This is part of the complexity i was talking about in the issue description, due to the metrics add execution allowing undefined data objects we need to create two separate interfaces because we do not want to allow undefined in the data as much as possible e.x the bulk DTO.

djabarovgeorge · 2023-11-14T11:02:24Z

packages/application-generic/src/services/queues/queue-base.service.ts

+    await this.instance.add(
+      name,
+      jobData,
+      {
+        removeOnComplete: true,
+        removeOnFail: true,
+        ...options,
+      },
+      groupId


I removed the dependency of the add method.

djabarovgeorge · 2023-11-14T11:03:20Z

packages/application-generic/src/services/queues/queue-base.service.ts

-    groupId?: string,
-    options: JobsOptions = {}
-  ) {
+  public async addMinimalJob(params: IJobParams) {


Refactored to get object instead of params list so we could be explicit about the params.

djabarovgeorge · 2023-11-14T11:03:26Z

packages/application-generic/src/services/queues/queue-base.service.ts

-    groupId?: string,
-    options: JobsOptions = {}
-  ) {
+  public async add(params: IJobParams) {


Refactored to get object instead of params list so we could be explicit about the params.

djabarovgeorge · 2023-11-14T11:04:18Z

packages/application-generic/src/services/queues/queue-base.service.ts

+export interface IJobParams {
+  name: string;
+  data?: any;
+  groupId?: string;
+  options?: JobsOptions;
+}
+
+export interface IBulkJobParams {
+  name: string;
+  data: any;
+  groupId?: string;
+  options?: BulkJobOptions;
+}


This is part of the complexity i was talking about in the issue description, due to the metrics add execution allowing undefined data objects we need to create two separate interfaces because we do not want to allow undefined in the data as much as possible e.x the bulk DTO.

We need to find a better place for those interfaces.

djabarovgeorge · 2023-11-14T11:04:29Z

packages/application-generic/src/services/queues/queue-base.service.ts

-    ]
-  ) {
+
+  public async addBulk(data: IBulkJobParams[]) {


Refactored to get object instead of params list so we could be explicit about the params.

djabarovgeorge · 2023-11-14T11:04:51Z

packages/application-generic/src/services/queues/standard/standard-queue.service.spec.ts

@@ -58,14 +59,17 @@ describe('Standard Queue service', () => {
      const _environmentId = 'standard-environment-id';
      const _organizationId = 'standard-organization-id';
      const _userId = 'standard-user-id';
-      const jobData = {
-        _id: jobId,
-        test: 'standard-job-data',


invalid param

djabarovgeorge · 2023-11-14T11:05:58Z

packages/application-generic/src/services/queues/standard/standard-queue.service.ts

+  public async add(data: unknown) {
+    // in order to implement we need to know what should be the data dto first
+    throw new Error('Not implemented');
+  }
+
+  public async addBulk(data: unknown[]) {
+    // in order to implement we need to know what should be the data dto first
+    throw new Error('Not implemented');
+  }


We do not use those methods at the moment, therefore i can not tell what should be the DTO here, i wonder if we should keep the exception or allow any here.

djabarovgeorge · 2023-11-14T11:11:29Z

packages/application-generic/src/services/queues/queue-base.service.ts

    await this.instance.addBulk(data);
  }
 }
+
+export interface IJobParams {
+  name: string;


Changed the id to name as it is the perm that is used in pull mq so i wanted to align with it. Same for IBulkJobParams.

…ur-queue-services-get-validated-object-data # Conflicts: # packages/application-generic/src/custom-providers/index.ts # packages/application-generic/src/services/queues/index.ts # packages/application-generic/src/usecases/add-job/add-job.usecase.ts

…alidated-object-data

feat: typed the queue add method and worker data object

437d432

github-actions bot added @novu/api @novu/ws @novu/worker @novu/inbound-mail labels Nov 14, 2023

djabarovgeorge commented Nov 14, 2023

View reviewed changes

fix: update method input

85e6e61

djabarovgeorge commented Nov 14, 2023

View reviewed changes

djabarovgeorge added 3 commits November 26, 2023 16:11

fix: after next merge

a019d3b

fix: spec function input

670f154

djabarovgeorge and others added 2 commits November 26, 2023 20:03

fix: spec function input

1025f18

Merge branch 'next' into nv-3012-ensure-that-our-queue-services-get-v…

7a16677

…alidated-object-data

djabarovgeorge closed this Dec 10, 2023

djabarovgeorge deleted the nv-3012-ensure-that-our-queue-services-get-validated-object-data branch December 10, 2023 10:59

djabarovgeorge mentioned this pull request Dec 10, 2023

feat: add queue and workers dto typed #4963

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Ensure that our queue services get validated object data #4843

Ensure that our queue services get validated object data #4843

djabarovgeorge commented Nov 14, 2023 •

edited

Loading

linear bot commented Nov 14, 2023

What?

Why? (Context)

Definition of Done

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023 •

edited

Loading

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023 •

edited

Loading

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

djabarovgeorge Nov 14, 2023

Ensure that our queue services get validated object data #4843

Ensure that our queue services get validated object data #4843

Conversation

djabarovgeorge commented Nov 14, 2023 • edited Loading

What change does this PR introduce?

Why was this change needed?

Other information and DOD's (Screenshots)

linear bot commented Nov 14, 2023

What?

Why? (Context)

Definition of Done

Choose a reason for hiding this comment

djabarovgeorge Nov 14, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

djabarovgeorge Nov 14, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

djabarovgeorge commented Nov 14, 2023 •

edited

Loading

djabarovgeorge Nov 14, 2023 •

edited

Loading

djabarovgeorge Nov 14, 2023 •

edited

Loading