Merging Next Into ---> Main (#1073)

thisismayuresh · web-flow · commit 8e71dbb95303 · 2025-09-11T16:09:43.000+05:30
diff --git a/apps/queue-manager/src/consumers/get-import-job-data.consumer.ts b/apps/queue-manager/src/consumers/get-import-job-data.consumer.ts
@@ -1,17 +1,25 @@
 import { RSSXMLService } from '@impler/services';
-import { ImportJobHistoryStatusEnum, SendImportJobCachedData } from '@impler/shared';
-import { JobMappingRepository, CommonRepository } from '@impler/dal';
+import {
+  ImportJobHistoryStatusEnum,
+  SendImportJobCachedData,
+  ColumnTypesEnum,
+  ITemplateSchemaItem,
+  ColumnDelimiterEnum,
+} from '@impler/shared';
+
 import { SendImportJobDataConsumer } from './send-import-job-data.consumer';
+import { CommonRepository, JobMappingRepository, ColumnRepository } from '@impler/dal';
 
 export class GetImportJobDataConsumer extends SendImportJobDataConsumer {
   private commonRepository: CommonRepository = new CommonRepository();
   private jobMappingRepository: JobMappingRepository = new JobMappingRepository();
+  private columnRepo: ColumnRepository = new ColumnRepository();
   private rssXmlService: RSSXMLService = new RSSXMLService();
 
   async message(message: { content: string }) {
     const data = JSON.parse(message.content) as { _jobId: string };
     const importJobHistoryId = this.commonRepository.generateMongoId().toString();
-    const importedData = await this.getJobImportedData(data._jobId);
+    const validationResult = await this.getJobImportedData(data._jobId);
 
     // Create history entry
     await this.importJobHistoryRepository.create({
@@ -26,13 +34,26 @@ export class GetImportJobDataConsumer extends SendImportJobDataConsumer {
     });
 
     if (webhookDestination?.callbackUrl) {
-      await this.sendDataImportData(data._jobId, importedData);
+      if (validationResult.validRecords > 0) {
+        await this.sendDataImportData(data._jobId, validationResult.validData, 1, undefined, false);
+      }
+      if (validationResult.invalidRecords > 0) {
+        await this.sendDataImportData(data._jobId, validationResult.invalidData, 1, undefined, true);
+      }
     }
 
     return;
   }
 
-  async getJobImportedData(_jobId: string) {
+  async getJobImportedData(_jobId: string): Promise<{
+    importedData: Record<string, unknown>[];
+    hasInvalidRecords: boolean;
+    totalRecords: number;
+    validRecords: number;
+    invalidRecords: number;
+    validData: Record<string, unknown>[];
+    invalidData: Record<string, unknown>[];
+  }> {
     try {
       const userJob = await this.userJobRepository.findOne({ _id: _jobId });
       if (!userJob) {
@@ -55,17 +76,140 @@ export class GetImportJobDataConsumer extends SendImportJobDataConsumer {
       const batchResult = await this.rssXmlService.getBatchXMLKeyValuesByPaths(parsedXMLData.xmlData, mappings);
       const mappedData = await this.rssXmlService.mappingFunction(mappings, batchResult);
 
-      return mappedData;
+      const validationResult = await this.validateData(_jobId, mappedData);
+
+      // Send data to webhook with validation status
+      this.sendDataImportData(_jobId, mappedData, 1, undefined, validationResult.hasInvalidRecords);
+
+      if (validationResult.hasInvalidRecords) {
+        await this.userJobRepository.update({ _id: _jobId }, { $set: { isInvalidRecords: true } });
+      }
+
+      return {
+        importedData: mappedData,
+        hasInvalidRecords: validationResult.hasInvalidRecords,
+        totalRecords: validationResult.totalRecords,
+        validRecords: validationResult.validRecords,
+        invalidRecords: validationResult.invalidRecords,
+        validData: validationResult.validData,
+        invalidData: validationResult.invalidData,
+      };
+    } catch (error) {}
+  }
+
+  private async validateData(
+    _jobId: string,
+    mappedData: Record<string, unknown>[]
+  ): Promise<{
+    hasInvalidRecords: boolean;
+    totalRecords: number;
+    validRecords: number;
+    invalidRecords: number;
+    validData: Record<string, unknown>[];
+    invalidData: Record<string, unknown>[];
+  }> {
+    try {
+      const userJob = await this.userJobRepository.findOne({ _id: _jobId });
+      if (!userJob) {
+        throw new Error(`Job not found for _jobId: ${_jobId}`);
+      }
+
+      // Get template columns (schema)
+      const columns = await this.columnRepo.find({ _templateId: userJob._templateId });
+      if (!columns || columns.length === 0) {
+        return {
+          hasInvalidRecords: false,
+          totalRecords: 0,
+          validRecords: 0,
+          invalidRecords: 0,
+          validData: [],
+          invalidData: [],
+        };
+      }
+
+      const multiSelectColumnHeadings: Record<string, string> = {};
+
+      (columns as unknown as ITemplateSchemaItem[]).forEach((column) => {
+        if (column.type === ColumnTypesEnum.SELECT && column.allowMultiSelect)
+          multiSelectColumnHeadings[column.key] = column.delimiter || ColumnDelimiterEnum.COMMA;
+      });
+
+      let totalRecords = 0;
+      let validRecords = 0;
+      let invalidRecords = 0;
+      const validData: Record<string, unknown>[] = [];
+      const invalidData: Record<string, unknown>[] = [];
+
+      for (const recordData of mappedData) {
+        // Format record for multi-select handling
+        const checkRecord: Record<string, unknown> = this.formatRecord({
+          record: { record: recordData },
+          multiSelectColumnHeadings,
+        });
+
+        const validationResult = this.validateRecordUsingColumnSchema(
+          checkRecord,
+          columns as unknown as ITemplateSchemaItem[]
+        );
+
+        totalRecords++;
+
+        if (validationResult.isValid) {
+          validRecords++;
+          validData.push(recordData);
+        } else {
+          invalidRecords++;
+          invalidData.push(recordData);
+        }
+      }
+
+      const hasInvalidRecords = invalidRecords > 0;
+
+      return {
+        hasInvalidRecords,
+        totalRecords,
+        validRecords,
+        invalidRecords,
+        validData,
+        invalidData,
+      };
     } catch (error) {
-      throw error;
+      return {
+        hasInvalidRecords: false,
+        totalRecords: 0,
+        validRecords: 0,
+        invalidRecords: 0,
+        validData: [],
+        invalidData: [],
+      };
     }
   }
 
+  private formatRecord({
+    record,
+    multiSelectColumnHeadings,
+  }: {
+    record: { record: Record<string, unknown> };
+    multiSelectColumnHeadings?: Record<string, string>;
+  }) {
+    return Object.keys(multiSelectColumnHeadings || {}).reduce(
+      (acc, heading) => {
+        if (typeof record.record[heading] === 'string') {
+          acc[heading] = (record.record[heading] as string)?.split(multiSelectColumnHeadings[heading]);
+        }
+
+        return acc;
+      },
+      { ...record.record }
+    );
+  }
+
   private async sendDataImportData(
     _jobId: string,
-    allDataJson: any[],
+    allDataJson: Record<string, any>[],
     page = 1,
-    initialCachedData?: SendImportJobCachedData
+    initialCachedData?: SendImportJobCachedData,
+    areInvalidRecords?: boolean
   ) {
     try {
       let cachedData = null;
@@ -84,6 +228,7 @@ export class GetImportJobDataConsumer extends SendImportJobDataConsumer {
           recordFormat: cachedData.recordFormat,
           chunkFormat: cachedData.chunkFormat,
           ...cachedData,
+          isInvalidRecords: areInvalidRecords,
         });
 
         const headers =
@@ -109,15 +254,129 @@ export class GetImportJobDataConsumer extends SendImportJobDataConsumer {
         });
 
         if (nextPageNumber) {
-          // Recursively call for next page with updated page number
           await this.sendDataImportData(_jobId, allDataJson, nextPageNumber, { ...cachedData, page: nextPageNumber });
         } else {
-          // Processing is done
           await this.finalizeUpload(_jobId);
         }
       }
     } catch (error) {
       throw error;
     }
   }
+
+  private validateRecordUsingColumnSchema(
+    record: Record<string, unknown>,
+    columns: ITemplateSchemaItem[]
+  ): { isValid: boolean } {
+    enum ValidationTypesEnum {
+      RANGE = 'range',
+      LENGTH = 'length',
+      UNIQUE_WITH = 'unique_with',
+      DIGITS = 'digits',
+    }
+    let isValid = true;
+
+    for (const column of columns) {
+      const value = record[column.key];
+
+      if (value === undefined) {
+        isValid = false;
+        continue;
+      }
+
+      if (column.isRequired && (value === null || value === '' || !value)) {
+        isValid = false;
+        continue;
+      }
+
+      if (value !== null && value !== '') {
+        switch (column.type) {
+          case ColumnTypesEnum.NUMBER:
+            if (isNaN(Number(value))) {
+              isValid = false;
+            }
+            break;
+          case ColumnTypesEnum.DOUBLE:
+            if (isNaN(Number(value)) || !Number.isFinite(Number(value))) {
+              isValid = false;
+            }
+            break;
+          case ColumnTypesEnum.EMAIL:
+            const emailRegex = /^[^\s@]+@[^\s@]+\.[^\s@]+$/;
+            if (!emailRegex.test(String(value))) {
+              isValid = false;
+            }
+            break;
+          case ColumnTypesEnum.DATE:
+            if (isNaN(Date.parse(String(value)))) {
+              isValid = false;
+            }
+            break;
+          case ColumnTypesEnum.REGEX:
+            if (column.regex && !new RegExp(column.regex).test(String(value))) {
+              isValid = false;
+            }
+            break;
+          case ColumnTypesEnum.SELECT:
+            if (column.selectValues && !column.selectValues.includes(String(value))) {
+              isValid = false;
+            }
+            break;
+          case ColumnTypesEnum.IMAGE:
+            const imageUrlRegex = /^https?:\/\/.+\.(jpg|jpeg|png|gif|bmp|webp|svg)$/i;
+            if (!imageUrlRegex.test(String(value))) {
+              isValid = false;
+            }
+            break;
+        }
+      }
+
+      if (
+        value !== undefined &&
+        value !== null &&
+        value !== '' &&
+        column.validations &&
+        column.validations.length > 0
+      ) {
+        for (const validation of column.validations) {
+          switch (validation.validate) {
+            case ValidationTypesEnum.RANGE:
+              const numValue = Number(value);
+              if (!isNaN(numValue)) {
+                if (validation.min !== undefined && numValue < validation.min) {
+                  isValid = false;
+                }
+                if (validation.max !== undefined && numValue > validation.max) {
+                  isValid = false;
+                }
+              }
+              break;
+            case ValidationTypesEnum.LENGTH:
+              const strValue = String(value);
+              if (validation.min !== undefined && strValue.length < validation.min) {
+                isValid = false;
+              }
+              if (validation.max !== undefined && strValue.length > validation.max) {
+                isValid = false;
+              }
+              break;
+            case ValidationTypesEnum.DIGITS:
+              const digitStr = String(value).replace(/[^0-9]/g, '');
+              if (validation.min !== undefined && digitStr.length < validation.min) {
+                isValid = false;
+              }
+              if (validation.max !== undefined && digitStr.length > validation.max) {
+                isValid = false;
+              }
+              break;
+            case ValidationTypesEnum.UNIQUE_WITH:
+              break;
+          }
+          if (!isValid) break;
+        }
+      }
+    }
+
+    return { isValid };
+  }
 }
diff --git a/apps/queue-manager/src/consumers/send-import-job-data.consumer.ts b/apps/queue-manager/src/consumers/send-import-job-data.consumer.ts
@@ -104,7 +104,15 @@ export class SendImportJobDataConsumer extends BaseConsumer {
     recordFormat,
     extra = '',
     multiSelectHeadings,
-  }: SendImportJobCachedData & { data: any[]; uploadId: string }): { sendData: Record<string, unknown>; page: number } {
+    isInvalidRecords,
+  }: SendImportJobCachedData & {
+    data: any[];
+    uploadId: string;
+    isInvalidRecords?: boolean;
+  }): {
+    sendData: Record<string, unknown>;
+    page: number;
+  } {
     const defaultValuesObj = JSON.parse(defaultValues);
     let slicedData = data.slice(
       Math.max((page - DEFAULT_PAGE) * chunkSize, MIN_LIMIT),
@@ -132,6 +140,7 @@ export class SendImportJobDataConsumer extends BaseConsumer {
       chunkSize: slicedData.length,
       extra: extra ? JSON.parse(extra) : '',
       totalPages: this.getTotalPages(data.length, chunkSize),
+      ...(isInvalidRecords !== undefined && { isInvalidRecords }),
     };
 
     return {
@@ -182,6 +191,7 @@ export class SendImportJobDataConsumer extends BaseConsumer {
       defaultValues: JSON.stringify(defaultValueObj),
       recordFormat: userJob.customRecordFormat,
       chunkFormat: userJob.customChunkFormat,
+      isInvalidRecords: userJob.isInvalidRecords,
     };
   }
 
diff --git a/libs/dal/src/repositories/user-job/user-job.entity.ts b/libs/dal/src/repositories/user-job/user-job.entity.ts
@@ -26,4 +26,6 @@ export class UserJobEntity {
   customChunkFormat: string;
 
   customSchema: string;
+
+  isInvalidRecords?: boolean;
 }
diff --git a/libs/dal/src/repositories/user-job/user-job.schema.ts b/libs/dal/src/repositories/user-job/user-job.schema.ts
@@ -46,6 +46,10 @@ const userJobSchema = new Schema(
     customSchema: {
       type: Schema.Types.String,
     },
+    isInvalidRecords: {
+      type: Schema.Types.Boolean,
+      default: false,
+    },
   },
   { ...schemaOptions }
 );
diff --git a/libs/services/src/rss-xml/rssxml.service.ts b/libs/services/src/rss-xml/rssxml.service.ts
diff --git a/libs/shared/src/types/upload/upload.types.ts b/libs/shared/src/types/upload/upload.types.ts

Original file line number	Diff line number	Diff line change
`@@ -26,4 +26,6 @@ export class UserJobEntity {`
`26`	`26`	`customChunkFormat: string;`
`27`	`27`
`28`	`28`	`customSchema: string;`
	`29`	`+`
	`30`	`+ isInvalidRecords?: boolean;`
`29`	`31`	`}`