auto import extra considerations (#850)

apps/api/src/app/import-jobs/dtos/create-userjob.dto.ts

@@ -12,4 +12,8 @@ export class CreateUserJobDto {
   @IsString()
   @IsOptional()
   extra?: string;
+
+  @IsString()
+  @IsOptional()
+  authHeaderValue?: string;
 }

apps/api/src/app/import-jobs/import-jobs.controller.ts

@@ -2,20 +2,20 @@ import { ApiTags, ApiSecurity, ApiOperation } from '@nestjs/swagger';
 import { Body, Controller, Delete, Get, Param, ParseArrayPipe, Post, Put, UseGuards } from '@nestjs/common';
 import {
   CreateUserJob,
-  GetColumnSchemaMapping,
-  CreateJobMapping,
   UpdateUserJob,
+  CreateJobMapping,
+  GetColumnSchemaMapping,
   GetUserJob,
   UserJobPause,
+  UserJobDelete,
   UserJobResume,
   UserJobTerminate,
-  UserJobDelete,
 } from './usecase';
 import { ACCESS_KEY_NAME } from '@impler/shared';
 import { JwtAuthGuard } from '@shared/framework/auth.gaurd';
 import { UpdateJobDto, CreateUserJobDto, UpdateJobMappingDto } from './dtos';
 
-@ApiTags('Import-Jobs')
+@ApiTags('Import Jobs')
 @Controller('/import-jobs')
 @UseGuards(JwtAuthGuard)
 @ApiSecurity(ACCESS_KEY_NAME)
@@ -35,12 +35,13 @@ export class ImportJobsController {
   @Post(':templateId')
   @ApiOperation({ summary: 'Create User-Job' })
   @ApiSecurity(ACCESS_KEY_NAME)
-  async createUserJobRoute(@Param('templateId') templateId: string, @Body() createUserJobData: CreateUserJobDto) {
+  async createUserJobRoute(@Param('templateId') templateId: string, @Body() jobData: CreateUserJobDto) {
     return this.createUserJob.execute({
       _templateId: templateId,
-      url: createUserJobData.url,
-      extra: createUserJobData.extra,
-      externalUserId: createUserJobData.externalUserId,
+      url: jobData.url,
+      extra: jobData.extra,
+      externalUserId: jobData.externalUserId,
+      authHeaderValue: jobData.authHeaderValue,
     });
   }
 

apps/api/src/app/import-jobs/usecase/create-userjob/create-userjob.command.ts

@@ -1,6 +1,7 @@
 export class CreateUserJobCommand {
   url: string;
+  extra?: string;
   _templateId: string;
   externalUserId?: string;
-  extra?: string;
+  authHeaderValue?: string;
 }

apps/api/src/app/import-jobs/usecase/create-userjob/create-userjob.usecase.ts

@@ -12,7 +12,13 @@ export class CreateUserJob {
     private readonly userJobRepository: UserJobRepository
   ) {}
 
-  async execute({ _templateId, url, externalUserId, extra }: CreateUserJobCommand): Promise<UserJobEntity> {
+  async execute({
+    url,
+    extra,
+    _templateId,
+    externalUserId,
+    authHeaderValue,
+  }: CreateUserJobCommand): Promise<UserJobEntity> {
     const mimeType = await this.rssService.getMimeType(url);
     if (mimeType === FileMimeTypesEnum.XML || mimeType === FileMimeTypesEnum.TEXTXML) {
       const { rssKeyHeading } = await this.rssService.parseRssFeed(url);
@@ -23,9 +29,10 @@ export class CreateUserJob {
 
       return await this.userJobRepository.create({
         url,
+        extra,
+        authHeaderValue,
         headings: rssKeyHeading,
         _templateId: _templateId,
-        extra,
         externalUserId: externalUserId || (formattedExtra as unknown as Record<string, any>)?.externalUserId,
       });
     } else {

apps/queue-manager/src/consumers/import-job-data.consumer.ts → apps/queue-manager/src/consumers/get-import-job-data.consumer.ts

@@ -28,11 +28,12 @@ export class GetImportJobDataConsumer extends BaseConsumer {
     const data = JSON.parse(message.content) as { _jobId: string };
     const importJobHistoryId = this.commonRepository.generateMongoId().toString();
     const importedData = await this.getJobImportedData(data._jobId);
+    const allDataFilePath = this.fileNameService.getAllJsonDataFilePath(importJobHistoryId);
     await this.convertRecordsToJsonFile(importJobHistoryId, importedData);
     await this.importJobHistoryRepository.create({
       _id: importJobHistoryId,
       _jobId: data._jobId,
-      allDataFilePath: this.fileNameService.getAllJsonDataFilePath(importJobHistoryId),
+      allDataFilePath,
       status: ImportJobHistoryStatusEnum.PROCESSING,
     });
     const userJobInfo = await this.userJobRepository.getUserJobWithTemplate(data._jobId);
@@ -42,7 +43,7 @@ export class GetImportJobDataConsumer extends BaseConsumer {
     });
 
     if (webhookDestination?.callbackUrl) {
-      publishToQueue(QueuesEnum.SEND_IMPORT_JOB_DATA, { importJobHistoryId });
+      publishToQueue(QueuesEnum.SEND_IMPORT_JOB_DATA, { _jobId: data._jobId, allDataFilePath });
     }
 
     return;

apps/queue-manager/src/consumers/index.ts

@@ -1,5 +1,5 @@
 export * from './send-webhook-data.consumer';
 export * from './end-import.consumer';
 export * from './send-bubble-data.consumer';
-export * from './import-job-data.consumer';
+export * from './get-import-job-data.consumer';
 export * from './send-import-job-data.consumer';

apps/queue-manager/src/consumers/send-import-job-data.consumer.ts

@@ -1,23 +1,20 @@
 import {
-  TemplateRepository,
-  WebhookDestinationRepository,
   ColumnRepository,
+  TemplateRepository,
   ImportJobHistoryRepository,
-  UserJobEntity,
+  WebhookDestinationRepository,
   WebhookLogEntity,
-  UploadRepository,
-  WebhookLogRepository,
   UserJobRepository,
 } from '@impler/dal';
 import { StorageService } from '@impler/services';
 import {
-  ColumnTypesEnum,
   SendImportJobData,
   SendImportJobCachedData,
   replaceVariablesInObject,
-  FileEncodingsEnum,
   QueuesEnum,
+  ColumnTypesEnum,
   UploadStatusEnum,
+  FileEncodingsEnum,
   ColumnDelimiterEnum,
 } from '@impler/shared';
 
@@ -30,17 +27,15 @@ const DEFAULT_PAGE = 1;
 
 export class SendImportJobDataConsumer extends BaseConsumer {
   private columnRepository: ColumnRepository = new ColumnRepository();
-  private uploadRepository: UploadRepository = new UploadRepository();
   private userJobRepository: UserJobRepository = new UserJobRepository();
   private templateRepository: TemplateRepository = new TemplateRepository();
-  private webhookLogRepository: WebhookLogRepository = new WebhookLogRepository();
   private importJobHistoryRepository: ImportJobHistoryRepository = new ImportJobHistoryRepository();
   private webhookDestinationRepository: WebhookDestinationRepository = new WebhookDestinationRepository();
   private storageService: StorageService = getStorageServiceClass();
 
   async message(message: { content: string }) {
     const data = JSON.parse(message.content) as SendImportJobData;
-    const cachedData = data.cache || (await this.getInitialCachedData(data.importJobHistoryId));
+    const cachedData = data.cache || (await this.getInitialCachedData(data._jobId, data.allDataFilePath));
     let allDataJson: null | any[] = null;
 
     if (cachedData && cachedData.callbackUrl) {
@@ -53,7 +48,7 @@ export class SendImportJobDataConsumer extends BaseConsumer {
       }
       const { sendData, page } = this.buildSendData({
         data: allDataJson,
-        uploadId: data.importJobHistoryId,
+        uploadId: data._jobId,
         chunkSize: cachedData.chunkSize,
         recordFormat: cachedData.recordFormat,
         chunkFormat: cachedData.chunkFormat,
@@ -67,14 +62,14 @@ export class SendImportJobDataConsumer extends BaseConsumer {
 
       const response = await this.makeApiCall({
         data: sendData,
-        uploadId: data.importJobHistoryId,
+        uploadId: data._jobId,
         page,
         method: 'POST',
         url: cachedData.callbackUrl,
         headers,
       });
 
-      this.makeResponseEntry(response);
+      await this.makeResponseEntry(response);
 
       const nextPageNumber = this.getNextPageNumber({
         totalRecords: allDataJson.length,
@@ -85,15 +80,16 @@ export class SendImportJobDataConsumer extends BaseConsumer {
       if (nextPageNumber) {
         // Make next call
         publishToQueue(QueuesEnum.SEND_IMPORT_JOB_DATA, {
-          importJobHistoryId: data.importJobHistoryId,
+          _jobId: data._jobId,
+          allDataFilePath: data.allDataFilePath,
           cache: {
             ...cachedData,
             page: nextPageNumber,
           },
-        });
+        } as SendImportJobData);
       } else {
         // Processing is done
-        this.finalizeUpload(data.importJobHistoryId);
+        this.finalizeUpload(data._jobId);
       }
     }
   }
@@ -115,10 +111,10 @@ export class SendImportJobDataConsumer extends BaseConsumer {
       Math.min(page * chunkSize, data.length)
     );
 
-    if (multiSelectHeadings && Object.keys(multiSelectHeadings).length > 0) {
+    if (Array.isArray(multiSelectHeadings) && multiSelectHeadings.length > 0) {
       slicedData = slicedData.map((obj) => {
-        Object.keys(multiSelectHeadings).forEach((heading) => {
-          obj.record[heading] = obj.record[heading] ? obj.record[heading].split(multiSelectHeadings[heading]) : [];
+        multiSelectHeadings.forEach((heading) => {
+          obj[heading] = obj[heading] ? (Array.isArray(obj[heading]) ? obj[heading] : obj[heading].split(',')) : [];
         });
 
         return obj;
@@ -144,21 +140,16 @@ export class SendImportJobDataConsumer extends BaseConsumer {
     };
   }
 
-  private async getInitialCachedData(_importJobHistoryId: string): Promise<SendImportJobCachedData> {
-    const importJobHistory = await this.importJobHistoryRepository.getHistoryWithJob(_importJobHistoryId, [
-      '_templateId',
-    ]);
-    const userJobEmail = await this.userJobRepository.getUserEmailFromJobId(importJobHistory._jobId);
+  private async getInitialCachedData(_jobId: string, allDataFilePath: string): Promise<SendImportJobCachedData> {
+    const userJob = await this.userJobRepository.findById(_jobId);
+    const userJobEmail = await this.userJobRepository.getUserEmailFromJobId(_jobId);
 
     const columns = await this.columnRepository.find({
-      _templateId: (importJobHistory._jobId as unknown as UserJobEntity)._templateId,
+      _templateId: userJob._templateId,
     });
-    const templateData = await this.templateRepository.findById(
-      (importJobHistory._jobId as unknown as UserJobEntity)._templateId,
-      'name _projectId code'
-    );
+    const templateData = await this.templateRepository.findById(userJob._templateId, 'name _projectId code');
     const webhookDestination = await this.webhookDestinationRepository.findOne({
-      _templateId: (importJobHistory._jobId as unknown as UserJobEntity)._templateId,
+      _templateId: userJob._templateId,
     });
 
     if (!webhookDestination || !webhookDestination.callbackUrl) {
@@ -175,33 +166,29 @@ export class SendImportJobDataConsumer extends BaseConsumer {
       });
     }
 
-    this.importJobHistoryRepository.create({
-      _jobId: importJobHistory._id,
-    });
-
     return {
+      page: 1,
+      allDataFilePath,
       email: userJobEmail,
-      _templateId: (importJobHistory._jobId as unknown as UserJobEntity)._templateId,
+      multiSelectHeadings,
+      extra: userJob.extra,
+      name: templateData.name,
+      _templateId: userJob._templateId,
+      authHeaderValue: userJob.authHeaderValue,
       callbackUrl: webhookDestination?.callbackUrl,
       chunkSize: webhookDestination?.chunkSize,
-      name: templateData.name,
-      page: 1,
-      extra: (importJobHistory._jobId as unknown as UserJobEntity).extra,
       authHeaderName: webhookDestination.authHeaderName,
-      authHeaderValue: '',
-      allDataFilePath: importJobHistory.allDataFilePath,
       defaultValues: JSON.stringify(defaultValueObj),
-      recordFormat: (importJobHistory._jobId as unknown as UserJobEntity).customRecordFormat,
-      chunkFormat: (importJobHistory._jobId as unknown as UserJobEntity).customChunkFormat,
-      multiSelectHeadings,
+      recordFormat: userJob.customRecordFormat,
+      chunkFormat: userJob.customChunkFormat,
     };
   }
 
   private async makeResponseEntry(data: Partial<WebhookLogEntity>) {
-    return await this.webhookLogRepository.create(data);
+    return this.importJobHistoryRepository.create(data);
   }
 
-  private async finalizeUpload(importJobHistoryId: string) {
-    return await this.uploadRepository.update({ _id: importJobHistoryId }, { status: UploadStatusEnum.COMPLETED });
+  private async finalizeUpload(_jobId: string) {
+    return await this.userJobRepository.update({ _id: _jobId }, { status: UploadStatusEnum.COMPLETED });
   }
 }

apps/widget/src/hooks/AutoImportPhase1/useAutoImportPhase1.tsx

@@ -2,12 +2,12 @@ import { useMutation } from '@tanstack/react-query';
 import { useForm, SubmitHandler } from 'react-hook-form';
 
 import { notifier } from '@util';
+import { IAutoImportValues } from '@types';
+import { useAppState } from '@store/app.context';
 import { useAPIState } from '@store/api.context';
 import { useJobsInfo } from '@store/jobinfo.context';
 import { useImplerState } from '@store/impler.context';
 import { IUserJob, IErrorObject } from '@impler/shared';
-import { IAutoImportValues } from '@types';
-import { useAppState } from '@store/app.context';
 
 interface IUseAutoImportPhase1Props {
   goNext: () => void;

libs/dal/src/repositories/user-job/user-job.entity.ts

@@ -15,6 +15,8 @@ export class UserJobEntity {
 
   status: string;
 
+  authHeaderValue: string;
+
   customRecordFormat: string;
 
   customChunkFormat: string;

libs/dal/src/repositories/user-job/user-job.schema.ts

@@ -23,6 +23,9 @@ const userJobSchema = new Schema(
     externalUserId: {
       type: Schema.Types.String,
     },
+    authHeaderValue: {
+      type: Schema.Types.String,
+    },
     status: {
       type: Schema.Types.String,
     },

libs/shared/src/entities/UserJob/Userjob.interface.ts

@@ -1,7 +1,7 @@
 export interface IUserJob {
   _id: string;
   url: string;
-  _templateId: string;
-  headings: string[];
   cron: string;
+  headings: string[];
+  _templateId: string;
 }

libs/shared/src/types/upload/upload.types.ts

@@ -99,7 +99,8 @@ export type SendWebhookData = {
 };
 
 export type SendImportJobData = {
-  importJobHistoryId: string;
+  _jobId: string;
+  allDataFilePath: string;
   cache?: SendImportJobCachedData;
 };