v1.8.1 (#1100)

Author: thisismayuresh

apps/api/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@impler/api",
-  "version": "1.8.0",
+  "version": "1.8.1",
   "author": "implerhq",
   "license": "MIT",
   "private": true,

apps/api/src/app/activity/usecases/upload-history/upload-history.usecase.ts

@@ -9,15 +9,15 @@ export class UploadHistory {
 
   async execute({ _projectId, date, limit, name, page }: UploadHistoryCommand): Promise<PaginationResult> {
     const uploadResult = await this.uploadRepository.getList(_projectId, name, date, page, limit);
-    uploadResult.uploads = uploadResult.uploads.map((upload) => {
-      upload.name = upload._template.name;
-      delete upload._template;
 
-      return upload;
-    });
+    const transformedRecords = uploadResult.uploads.map((record) => ({
+      ...record,
+      name: record._template?.name,
+      _template: undefined,
+    }));
 
     return {
-      data: uploadResult.uploads,
+      data: transformedRecords,
       limit,
       page,
       totalPages: Math.ceil(uploadResult.totalRecords / limit),

apps/api/src/app/auto-import-jobs-schedular/usecase/auto-import-jobs-schedular.ts

@@ -48,6 +48,7 @@ export class AutoImportJobsSchedular {
 
       const interval = parser.parseExpression(cronExpression.trim(), {
         currentDate: currentNextRun,
+        tz: 'Asia/Kolkata',
       });
 
       const nextRun = interval.next().toDate();

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

@@ -73,17 +73,19 @@ export class CreateUserJob {
         return dayjs().add(5, 'minutes').toDate();
       }
 
+      const interval = parser.parseExpression(cronExpression.trim(), {
+        tz: 'Asia/Kolkata',
+      });
+      const nextCronTime = dayjs(interval.next().toDate());
       const now = dayjs();
-      const fiveMinutesFromNow = now.add(5, 'minutes');
 
-      const interval = parser.parseExpression(cronExpression.trim());
-      const nextCronTime = dayjs(interval.next().toDate());
+      if (nextCronTime.isBefore(now.add(1, 'minute'))) {
+        const nextOccurrence = dayjs(interval.next().toDate());
 
-      if (nextCronTime.isAfter(fiveMinutesFromNow)) {
-        return fiveMinutesFromNow.toDate();
-      } else {
-        return nextCronTime.toDate();
+        return nextOccurrence.toDate();
       }
+
+      return nextCronTime.toDate();
     } catch (error) {
       return dayjs().add(5, 'minutes').toDate();
     }

apps/queue-manager/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@impler/queue-manager",
-  "version": "1.8.0",
+  "version": "1.8.1",
   "author": "implerhq",
   "license": "MIT",
   "private": true,

apps/queue-manager/src/bootstrap.ts

@@ -10,7 +10,7 @@ import {
   SendWebhookDataConsumer,
   EndImportConsumer,
   SendBubbleDataConsumer,
-  GetImportJobDataConsumer,
+  SendAutoImportJobDataConsumer,
   SendImportJobDataConsumer,
   SendFailedWebhookDataConsumer,
 } from './consumers';
@@ -47,7 +47,7 @@ export async function bootstrap() {
   const sendBubbleDataConsumer = new SendBubbleDataConsumer();
   const sendWebhookdataConsumer = new SendWebhookDataConsumer();
   const sendFailedWebhookDataConsumer = new SendFailedWebhookDataConsumer();
-  const getImportJobbDataConsumer = new GetImportJobDataConsumer();
+  const autoImportJobbDataConsumer = new SendAutoImportJobDataConsumer();
   const sendImportJobDataConsumer = new SendImportJobDataConsumer();
 
   // add queues to channel
@@ -84,7 +84,7 @@ export async function bootstrap() {
       }),
       channel.consume(
         QueuesEnum.GET_IMPORT_JOB_DATA,
-        getImportJobbDataConsumer.message.bind(getImportJobbDataConsumer),
+        autoImportJobbDataConsumer.message.bind(autoImportJobbDataConsumer),
         {
           noAck: true,
         }

…onsumers/get-import-job-data.consumer.ts …ers/get-auto-import-job-data.consumer.tsapps/queue-manager/src/consumers/get-import-job-data.consumer.ts renamed to apps/queue-manager/src/consumers/get-auto-import-job-data.consumer.ts apps/queue-manager/src/consumers/get-import-job-data.consumer.ts renamed to apps/queue-manager/src/consumers/get-auto-import-job-data.consumer.ts

@@ -1,67 +1,69 @@
-import { RSSXMLService } from '@impler/services';
+import { PaymentAPIService, RSSXMLService } from '@impler/services';
 import {
-  ImportJobHistoryStatusEnum,
   SendImportJobCachedData,
   ColumnTypesEnum,
   ITemplateSchemaItem,
   ColumnDelimiterEnum,
+  UserJobImportStatusEnum,
 } from '@impler/shared';
 
-import dayjs from 'dayjs';
 import { SendImportJobDataConsumer } from './send-import-job-data.consumer';
-import { CommonRepository, JobMappingRepository, ColumnRepository } from '@impler/dal';
+import { JobMappingRepository, ColumnRepository, UserJobEntity } from '@impler/dal';
+
+interface IValidationResult {
+  hasInvalidRecords: boolean;
+  totalRecords: number;
+  validRecords: number;
+  invalidRecords: number;
+  validData: Record<string, unknown>[];
+  invalidData: Record<string, unknown>[];
+}
 
-export class GetImportJobDataConsumer extends SendImportJobDataConsumer {
-  private commonRepository: CommonRepository = new CommonRepository();
+export class SendAutoImportJobDataConsumer extends SendImportJobDataConsumer {
   private jobMappingRepository: JobMappingRepository = new JobMappingRepository();
   private columnRepo: ColumnRepository = new ColumnRepository();
   private rssXmlService: RSSXMLService = new RSSXMLService();
+  private paymentAPIService: PaymentAPIService = new PaymentAPIService();
 
   async message(message: { content: string }) {
     const data = JSON.parse(message.content) as { _jobId: string };
-    const importJobHistoryId = this.commonRepository.generateMongoId().toString();
     const validationResult = await this.getJobImportedData(data._jobId);
 
-    // Create history entry
-    await this.importJobHistoryRepository.create({
-      _id: importJobHistoryId,
-      _jobId: data._jobId,
-      status: ImportJobHistoryStatusEnum.PROCESSING,
-    });
-
     const userJobInfo = await this.userJobRepository.getUserJobWithTemplate(data._jobId);
     const webhookDestination = await this.webhookDestinationRepository.findOne({
       _templateId: userJobInfo._templateId,
     });
 
     if (webhookDestination?.callbackUrl) {
       if (validationResult.validRecords > 0) {
-        await this.sendDataImportData(data._jobId, validationResult.validData, 1, undefined, false, userJobInfo.endsOn);
+        await this.sendDataImportData(
+          data._jobId,
+          validationResult.validData,
+          1,
+          undefined,
+          false,
+          userJobInfo.endsOn,
+          validationResult
+        );
       }
+
       if (validationResult.invalidRecords > 0) {
         await this.sendDataImportData(
           data._jobId,
           validationResult.invalidData,
           1,
           undefined,
           true,
-          userJobInfo.endsOn
+          userJobInfo.endsOn,
+          validationResult
         );
       }
     }
 
     return;
   }
 
-  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>[];
-  }> {
+  async getJobImportedData(_jobId: string): Promise<IValidationResult & { importedData: Record<string, unknown>[] }> {
     try {
       const userJob = await this.userJobRepository.findOne({ _id: _jobId });
       if (!userJob) {
@@ -217,7 +219,8 @@ export class GetImportJobDataConsumer extends SendImportJobDataConsumer {
     page = 1,
     initialCachedData?: SendImportJobCachedData,
     areInvalidRecords?: boolean,
-    endsOn?: Date
+    endsOn?: Date,
+    validationResult?: IValidationResult
   ) {
     try {
       let cachedData = null;
@@ -268,11 +271,14 @@ export class GetImportJobDataConsumer extends SendImportJobDataConsumer {
             nextPageNumber,
             { ...cachedData, page: nextPageNumber },
             areInvalidRecords,
-            endsOn
+            endsOn,
+            validationResult
           );
         } else {
-          if (endsOn && dayjs(endsOn).isSame(dayjs(), 'day')) {
-            await this.finalizeUpload(_jobId);
+          if (validationResult) {
+            // Get userJobInfo for finalization
+            const userJobInfo = await this.userJobRepository.getUserJobWithTemplate(_jobId);
+            await this.finalizeAutoImportJob(_jobId, validationResult, userJobInfo);
           }
         }
       }
@@ -281,6 +287,48 @@ export class GetImportJobDataConsumer extends SendImportJobDataConsumer {
     }
   }
 
+  private async finalizeAutoImportJob(
+    _jobId: string,
+    validationResult: IValidationResult,
+    userJobInfo: UserJobEntity
+  ): Promise<void> {
+    try {
+      await this.templateRepository.update(
+        { _id: userJobInfo._templateId },
+        {
+          $set: {
+            isInvalidRecords: validationResult.hasInvalidRecords,
+            invalidRecords: validationResult.invalidRecords,
+            validRecords: validationResult.validRecords,
+            totalRecords: validationResult.totalRecords,
+          },
+        }
+      );
+
+      await this.paymentAPIService.createEvent(
+        {
+          uploadId: _jobId,
+          totalRecords: validationResult.totalRecords,
+          validRecords: validationResult.validRecords,
+          invalidRecords: validationResult.invalidRecords,
+        },
+        userJobInfo.externalUserId
+      );
+
+      await this.userJobRepository.update(
+        { _id: _jobId },
+        {
+          $set: {
+            status: UserJobImportStatusEnum.RUNNING, //UserJobImportStatusEnum.COMPLETED,
+            totalRecords: validationResult.totalRecords,
+            validRecords: validationResult.validRecords,
+            invalidRecords: validationResult.invalidRecords,
+          },
+        }
+      );
+    } catch (error) {}
+  }
+
   private validateRecordUsingColumnSchema(
     record: Record<string, unknown>,
     columns: ITemplateSchemaItem[]

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

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

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

@@ -5,8 +5,9 @@ import {
   WebhookDestinationRepository,
   WebhookLogEntity,
   UserJobRepository,
+  WebhookLogRepository,
 } from '@impler/dal';
-import { StorageService } from '@impler/services';
+import { PaymentAPIService, StorageService } from '@impler/services';
 import {
   SendImportJobData,
   SendImportJobCachedData,
@@ -20,17 +21,19 @@ import {
 
 import { publishToQueue } from '../bootstrap';
 import { BaseConsumer } from './base.consumer';
-import { getStorageServiceClass } from '../helpers/services.helper';
+import { getPaymentApiServiceClass, getStorageServiceClass } from '../helpers/services.helper';
 
 const MIN_LIMIT = 0;
 const DEFAULT_PAGE = 1;
 
 export class SendImportJobDataConsumer extends BaseConsumer {
   private columnRepository: ColumnRepository = new ColumnRepository();
   public userJobRepository: UserJobRepository = new UserJobRepository();
-  private templateRepository: TemplateRepository = new TemplateRepository();
+  public templateRepository: TemplateRepository = new TemplateRepository();
   public importJobHistoryRepository: ImportJobHistoryRepository = new ImportJobHistoryRepository();
   public webhookDestinationRepository: WebhookDestinationRepository = new WebhookDestinationRepository();
+  public webhookLogRepository: WebhookLogRepository = new WebhookLogRepository();
+  public paymentApiService: PaymentAPIService = getPaymentApiServiceClass();
   public storageService: StorageService = getStorageServiceClass();
 
   async message(message: { content: string }) {

apps/queue-manager/src/helpers/services.helper.ts

@@ -1,8 +1,16 @@
 import { StorageTypeEnum } from '@impler/shared';
-import { AzureStorageService, EmailService, S3StorageService, SESEmailService, StorageService } from '@impler/services';
+import {
+  AzureStorageService,
+  EmailService,
+  S3StorageService,
+  SESEmailService,
+  StorageService,
+  PaymentAPIService,
+} from '@impler/services';
 
 let storageService: StorageService;
 let emailService: EmailService;
+let paymentApiService: PaymentAPIService;
 
 // Implementing singleton pattern for storage service
 export function getStorageServiceClass() {
@@ -19,3 +27,10 @@ export function getEmailServiceClass() {
 
   return emailService;
 }
+
+export function getPaymentApiServiceClass() {
+  if (paymentApiService) return paymentApiService;
+  paymentApiService = new PaymentAPIService();
+
+  return paymentApiService;
+}