feat: merged next

Author: bhavik

apps/api/package.json

@@ -67,6 +67,7 @@
     "passport-jwt": "^4.0.1",
     "passport-oauth2": "^1.6.1",
     "rimraf": "^3.0.2",
+    "sax": "^1.4.1",
     "socket.io": "^4.7.2",
     "source-map-support": "^0.5.21",
     "uuid": "^9.0.0",
@@ -92,6 +93,7 @@
     "@types/passport-github2": "^1.2.5",
     "@types/passport-jwt": "^3.0.8",
     "@types/passport-oauth2": "^1.4.11",
+    "@types/sax": "^1.2.7",
     "@types/uuid": "^9.0.2",
     "chai": "^4.3.7",
     "mocha": "^10.1.0",

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

@@ -2,10 +2,10 @@ import { UserJobEntity, UserJobRepository, WebhookDestinationRepository } from '
 import { Injectable } from '@nestjs/common';
 import * as dayjs from 'dayjs';
 import * as parser from 'cron-parser';
-import { Cron } from '@nestjs/schedule';
+import { Cron, CronExpression } from '@nestjs/schedule';
 import { ScheduleUserJob, UpdateUserJob } from 'app/import-jobs/usecase';
 import { UserJobImportStatusEnum } from '@impler/shared';
-import { CRON_SCHEDULE } from '@shared/constants';
+// import { CRON_SCHEDULE } from '@shared/constants';
 const parseCronExpression = require('@impler/shared/src/utils/cronstrue');
 
 @Injectable()
@@ -17,7 +17,7 @@ export class AutoImportJobsSchedular {
     private readonly scheduleUserJob: ScheduleUserJob
   ) {}
 
-  @Cron(CRON_SCHEDULE.DEFAULT_CRON_TIME)
+  @Cron(CronExpression.EVERY_10_SECONDS)
   async handleCronSchedular() {
     await this.fetchAndExecuteScheduledJobs();
   }

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

@@ -1,6 +1,10 @@
 import { IsNotEmpty, IsOptional, IsString } from 'class-validator';
 
 export class CreateUserJobDto {
+  @IsString()
+  @IsNotEmpty()
+  webSocketSessionId: string;
+
   @IsString()
   @IsNotEmpty()
   url: string;

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

@@ -37,6 +37,7 @@ export class ImportJobsController {
   @ApiSecurity(ACCESS_KEY_NAME)
   async createUserJobRoute(@Param('templateId') templateId: string, @Body() jobData: CreateUserJobDto) {
     return this.createUserJob.execute({
+      webSocketSessionId: jobData.webSocketSessionId,
       _templateId: templateId,
       url: jobData.url,
       extra: jobData.extra,

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

@@ -1,4 +1,5 @@
 export class CreateUserJobCommand {
+  webSocketSessionId: string;
   url: string;
   extra?: string;
   _templateId: string;

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

@@ -1,40 +1,60 @@
-import { BadRequestException, Injectable } from '@nestjs/common';
-import { isValidXMLMimeType } from '@shared/helpers/common.helper';
-import { APIMessages } from '@shared/constants';
+import { Injectable } from '@nestjs/common';
 import { UserJobEntity, UserJobRepository } from '@impler/dal';
-import { RSSService } from '@shared/services';
 import { CreateUserJobCommand } from './create-userjob.command';
+import { APIMessages } from '@shared/constants';
+import { BadRequestException } from '@nestjs/common';
+import { RSSXMLService } from '@impler/services';
+import { isValidXMLMimeType } from '@shared/helpers/common.helper';
+import { WebSocketService } from '@shared/services';
 
 @Injectable()
 export class CreateUserJob {
   constructor(
-    private readonly rssService: RSSService,
+    private readonly webSocketService: WebSocketService,
     private readonly userJobRepository: UserJobRepository
   ) {}
 
   async execute({
+    webSocketSessionId,
     url,
     extra,
     _templateId,
     externalUserId,
     authHeaderValue,
   }: CreateUserJobCommand): Promise<UserJobEntity> {
-    const mimeType = await this.rssService.getMimeType(url);
+    const rssService = new RSSXMLService(url);
+
+    const mimeType = await rssService.getMimeType(url);
+
     if (isValidXMLMimeType(mimeType)) {
-      const { rssKeyHeading } = await this.rssService.parseRssFeed(url);
-      let formattedExtra = extra || '{}';
       try {
-        formattedExtra = JSON.parse(extra);
-      } catch (_) {}
-
-      return await this.userJobRepository.create({
-        url,
-        extra,
-        authHeaderValue,
-        headings: rssKeyHeading,
-        _templateId: _templateId,
-        externalUserId: externalUserId || (formattedExtra as unknown as Record<string, any>)?.externalUserId,
-      });
+        const abortController = new AbortController();
+
+        this.webSocketService.registerSessionAbort(webSocketSessionId, abortController);
+
+        const rssXmlParsedDataKeys = await rssService.parseXMLAndExtractData({
+          xmlUrl: url,
+          sessionId: webSocketSessionId,
+          sendProgress: this.webSocketService.sendProgress,
+          sendError: this.webSocketService.sendError,
+          sendCompletion: this.webSocketService.sendCompletion,
+          abortSignal: abortController.signal,
+        });
+
+        let formattedExtra = extra || '{}';
+        try {
+          formattedExtra = JSON.parse(extra);
+        } catch (_) {}
+
+        return await this.userJobRepository.create({
+          url,
+          extra,
+          authHeaderValue,
+          headings: rssXmlParsedDataKeys?.keys || [],
+          _templateId: _templateId,
+          externalUserId: externalUserId || (formattedExtra as unknown as Record<string, any>)?.externalUserId,
+        });
+      } catch (error) {}
     } else {
       throw new BadRequestException(APIMessages.INVALID_RSS_URL);
     }

apps/api/src/app/import-jobs/usecase/index.ts

@@ -3,7 +3,7 @@ import { GetColumnSchemaMapping } from './get-columnschema-mapping/get-columnsch
 import { CreateJobMapping } from './create-jobmapping/create-jobmapping.usecase';
 import { UpdateUserJob } from './update-userjob/update-userjob.usecase';
 import { GetUserJob } from './get-userjob/get-userjob.usecase';
-import { RSSService } from '@shared/services';
+import { RSSXMLService } from '@impler/services';
 import { QueueService } from '@shared/services/queue.service';
 import { UserJobPause } from './userjob-usecase/userjob-pause.usecase';
 import { UserJobDelete } from './userjob-usecase/userjob-delete.usecase';
@@ -18,7 +18,7 @@ export const USECASES = [
   CreateJobMapping,
   UpdateUserJob,
   GetUserJob,
-  RSSService,
+  RSSXMLService,
   QueueService,
   UserJobResume,
   UserJobPause,

apps/api/src/app/shared/services/index.ts

@@ -1,2 +1,2 @@
 export * from './file';
-export * from './rss.service';
+export * from './websocket-service';

apps/api/src/app/shared/services/rss.service.ts

@@ -0,0 +1 @@
+export * from './websocket.service';