Merging Next into ---> Main

.gitignore

@@ -1,6 +1,6 @@
 node_modules
 dist
 build
-
+.vscode/
 .env
-.nx
+.nx

apps/api/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@impler/api",
-  "version": "1.7.0",
+  "version": "1.8.0",
   "author": "implerhq",
   "license": "MIT",
   "private": true,
@@ -23,6 +23,7 @@
   },
   "dependencies": {
     "@amplitude/analytics-node": "^1.3.6",
+    "@faker-js/faker": "^8.4.1",
     "@impler/client": "workspace:^",
     "@impler/dal": "workspace:^",
     "@impler/services": "workspace:^",

apps/api/src/app/activity/activity.controller.ts

@@ -1,8 +1,8 @@
 import { ApiTags, ApiOperation, ApiSecurity } from '@nestjs/swagger';
-import { Controller, Param, UseGuards, Get, Query } from '@nestjs/common';
+import { Controller, Param, UseGuards, Get, Query, Post } from '@nestjs/common';
 import { ValidateMongoId } from '@shared/validations/valid-mongo-id.validation';
 
-import { UploadSummary, UploadHistory } from './usecases';
+import { UploadSummary, UploadHistory, RetryUpload, WebhookLogs } from './usecases';
 import { ACCESS_KEY_NAME, Defaults } from '@impler/shared';
 import { JwtAuthGuard } from '@shared/framework/auth.gaurd';
 import { isDateString } from '@shared/helpers/common.helper';
@@ -14,7 +14,9 @@ import { isDateString } from '@shared/helpers/common.helper';
 export class ActivityController {
   constructor(
     private uploadSummary: UploadSummary,
-    private uploadHistory: UploadHistory
+    private uploadHistory: UploadHistory,
+    private retryUpload: RetryUpload,
+    private webhookLogs: WebhookLogs
   ) {}
 
   @Get(':projectId/summary')
@@ -52,4 +54,35 @@ export class ActivityController {
       limit,
     });
   }
+
+  @Post(':uploadId/retry')
+  @ApiOperation({
+    summary: 'Retry webhook data for a specific upload',
+  })
+  async retryUploadRoute(@Param('uploadId', ValidateMongoId) uploadId: string) {
+    return this.retryUpload.execute(uploadId);
+  }
+
+  @Get('upload/:uploadId/webhook-logs')
+  @ApiOperation({
+    summary: 'Get webhook logs for a specific upload',
+  })
+  async getWebhookLogsRoute(
+    @Param('uploadId', ValidateMongoId) uploadId: string,
+    @Query('page') page = Defaults.ONE,
+    @Query('limit') limit = Defaults.PAGE_LIMIT,
+    @Query('isRetry') isRetry?: boolean
+  ) {
+    if (isNaN(page)) page = Defaults.ONE;
+    else page = Number(page);
+    if (isNaN(limit)) limit = Defaults.PAGE_LIMIT;
+    else limit = Number(limit);
+
+    return this.webhookLogs.execute({
+      uploadId,
+      page,
+      limit,
+      isRetry,
+    });
+  }
 }

apps/api/src/app/activity/usecases/index.ts

@@ -1,10 +1,14 @@
 import { UploadSummary } from './upload-summary/upload-summary.usecase';
 import { UploadHistory } from './upload-history/upload-history.usecase';
+import { RetryUpload } from './retry-upload/retry-upload.usecase';
+import { WebhookLogs } from './webhook-logs/webhook-logs.usecase';
+import { QueueService } from '@shared/services/queue.service';
 
-export const USE_CASES = [
-  UploadSummary,
-  UploadHistory,
-  //
-];
+export * from './upload-summary/upload-summary.usecase';
+export * from './upload-history/upload-history.usecase';
+export * from './retry-upload/retry-upload.usecase';
+export * from './webhook-logs/webhook-logs.usecase';
 
-export { UploadSummary, UploadHistory };
+export const USE_CASES = [UploadSummary, UploadHistory, RetryUpload, WebhookLogs, QueueService];
+
+export { UploadSummary, UploadHistory, RetryUpload, WebhookLogs, QueueService };

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

@@ -0,0 +1,45 @@
+import { BadRequestException, Injectable } from '@nestjs/common';
+import { UploadRepository, TemplateRepository, WebhookDestinationRepository } from '@impler/dal';
+import { QueueService } from '@shared/services/queue.service';
+import { QueuesEnum, DestinationsEnum } from '@impler/shared';
+
+@Injectable()
+export class RetryUpload {
+  constructor(
+    private uploadRepository: UploadRepository,
+    private templateRepository: TemplateRepository,
+    private webhookDestinationRepository: WebhookDestinationRepository,
+    private queueService: QueueService
+  ) {}
+
+  async execute(uploadId: string) {
+    const upload = await this.uploadRepository.findById(uploadId);
+
+    if (!upload) {
+      throw new Error('Upload not found');
+    }
+
+    const template = await this.templateRepository.findOne({ _id: upload._templateId });
+
+    if (!template || template.destination !== DestinationsEnum.WEBHOOK) {
+      throw new BadRequestException('Template does not have a webhook destination');
+    }
+
+    const webhookDestination = await this.webhookDestinationRepository.findOne({ _templateId: upload._templateId });
+
+    if (!webhookDestination) {
+      throw new BadRequestException('Webhook destination not found for template');
+    }
+
+    this.queueService.publishToQueue(QueuesEnum.SEND_WEBHOOK_DATA, {
+      uploadId: upload._id,
+      isRetry: true,
+    });
+
+    return {
+      success: true,
+      message: 'Retry request queued successfully',
+      upload: upload,
+    };
+  }
+}

apps/api/src/app/activity/usecases/webhook-logs/webhook-logs.command.ts

@@ -0,0 +1,23 @@
+import { IsDefined, IsMongoId, IsOptional, IsNumber, Min } from 'class-validator';
+import { Transform } from 'class-transformer';
+
+export class WebhookLogsCommand {
+  @IsDefined()
+  @IsMongoId()
+  uploadId: string;
+
+  @IsOptional()
+  @Transform(({ value }) => Number(value))
+  @IsNumber()
+  @Min(1)
+  page?: number = 1;
+
+  @IsOptional()
+  @Transform(({ value }) => Number(value))
+  @IsNumber()
+  @Min(1)
+  limit?: number = 10;
+
+  @IsOptional()
+  isRetry?: boolean;
+}

apps/api/src/app/activity/usecases/webhook-logs/webhook-logs.usecase.ts

@@ -0,0 +1,39 @@
+import { Injectable } from '@nestjs/common';
+import { WebhookLogRepository } from '@impler/dal';
+import { PaginationResult } from '@impler/shared';
+import { WebhookLogsCommand } from './webhook-logs.command';
+
+@Injectable()
+export class WebhookLogs {
+  constructor(private webhookLogRepository: WebhookLogRepository) {}
+
+  async execute({ uploadId, page = 1, limit = 10, isRetry }: WebhookLogsCommand): Promise<PaginationResult> {
+    const skip = (page - 1) * limit;
+
+    const matchQuery: any = { _uploadId: uploadId };
+    if (isRetry !== undefined) {
+      matchQuery.isRetry = isRetry;
+    }
+
+    const thirtyDaysAgo = new Date();
+    thirtyDaysAgo.setDate(thirtyDaysAgo.getDate() - 30);
+    matchQuery.callDate = { $gte: thirtyDaysAgo };
+
+    const [logs, totalCount] = await Promise.all([
+      this.webhookLogRepository.find(matchQuery, '', {
+        sort: { createdAt: -1 },
+        skip,
+        limit,
+      }),
+      this.webhookLogRepository.count(matchQuery),
+    ]);
+
+    return {
+      data: logs,
+      limit,
+      page,
+      totalPages: Math.ceil(totalCount / limit),
+      totalRecords: totalCount,
+    };
+  }
+}

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

@@ -3,10 +3,10 @@ import { Injectable } from '@nestjs/common';
 import * as dayjs from 'dayjs';
 import * as parser from 'cron-parser';
 import { Cron } from '@nestjs/schedule';
-import { ScheduleUserJob, UpdateUserJob } from 'app/import-jobs/usecase';
+import { UpdateUserJob } from 'app/import-jobs/usecase';
 import { UserJobImportStatusEnum } from '@impler/shared';
+import { UserJobTriggerService } from 'app/import-jobs/usecase';
 import { CRON_SCHEDULE } from '@shared/constants';
-// import { CRON_SCHEDULE } from '@shared/constants';
 const parseCronExpression = require('@impler/shared/src/utils/cronstrue');
 
 @Injectable()
@@ -15,11 +15,12 @@ export class AutoImportJobsSchedular {
     private readonly userJobRepository: UserJobRepository,
     private readonly webhookDestinationRepository: WebhookDestinationRepository,
     private readonly updateUserJob: UpdateUserJob,
-    private readonly scheduleUserJob: ScheduleUserJob
+    private readonly userJobTriggerService: UserJobTriggerService
   ) {}
 
   @Cron(CRON_SCHEDULE.AUTO_IMPORT_DEFAULT_CRON_TIME)
   async handleCronSchedular() {
+    console.log('Crone Running');
     await this.fetchAndExecuteScheduledJobs();
   }
 
@@ -28,6 +29,7 @@ export class AutoImportJobsSchedular {
     const userJobs = await this.userJobRepository.find({});
 
     for (const userJob of userJobs) {
+      console.log('Should run the Cron Job ?', await this.shouldCroneRun({ userJob }), userJob._id);
       if (await this.shouldCroneRun({ userJob })) {
         try {
           const interval = parser.parseExpression(userJob.cron);
@@ -39,7 +41,8 @@ export class AutoImportJobsSchedular {
 
             await this.updateUserJob.execute(userJob._id, userJob);
 
-            await this.scheduleUserJob.execute(userJob._id, userJob.cron);
+            // await this.scheduleUserJob.execute(userJob._id, userJob.cron);
+            await this.userJobTriggerService.execute(userJob._id);
           }
         } catch (error) {}
       }
@@ -89,10 +92,16 @@ export class AutoImportJobsSchedular {
   async shouldCroneRun({ userJob }: { userJob: UserJobEntity }): Promise<boolean> {
     const now = dayjs();
 
+    if (userJob.status === UserJobImportStatusEnum.PAUSED) {
+      return false;
+    }
+
     if (
       (userJob.cron && userJob.status === UserJobImportStatusEnum.SCHEDULING) ||
       userJob.status === UserJobImportStatusEnum.RUNNING ||
-      (userJob.status === 'Completed' && (await this.fetchDestination(userJob._templateId)) && !userJob.endsOn) ||
+      (userJob.status === UserJobImportStatusEnum.COMPLETED &&
+        (await this.fetchDestination(userJob._templateId)) &&
+        !userJob.endsOn) ||
       !dayjs(userJob.endsOn).isSame(now, 'd')
     ) {
       return true;

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

@@ -1,4 +1,4 @@
-import { ScheduleUserJob, UpdateUserJob, UserJobTriggerService } from 'app/import-jobs/usecase';
+import { UpdateUserJob, UserJobTriggerService } from 'app/import-jobs/usecase';
 import { AutoImportJobsSchedular } from './auto-import-jobs-schedular';
 import { QueueService } from '@shared/services/queue.service';
 
@@ -7,7 +7,6 @@ export const USE_CASES = [
   UpdateUserJob,
   UserJobTriggerService,
   UserJobTriggerService,
-  ScheduleUserJob,
   QueueService,
   //
 ];

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

@@ -10,7 +10,6 @@ import { UserJobDelete } from './userjob-usecase/userjob-delete.usecase';
 import { UserJobTerminate } from './userjob-usecase/userjob-terminate.usecase';
 import { UserJobTriggerService } from './userjob-usecase/userjob-trigger.usecase';
 import { UserJobResume } from './userjob-usecase/userjob.resume.usecsae';
-import { ScheduleUserJob } from './schedule-user-job/schedule-user-job';
 
 export const USECASES = [
   CreateUserJob,
@@ -25,7 +24,6 @@ export const USECASES = [
   UserJobDelete,
   UserJobTerminate,
   UserJobTriggerService,
-  ScheduleUserJob,
   //
 ];
 
@@ -40,5 +38,4 @@ export {
   UserJobDelete,
   UserJobTerminate,
   UserJobTriggerService,
-  ScheduleUserJob,
 };

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

@@ -1,16 +1,10 @@
 import { Injectable } from '@nestjs/common';
-import { SchedulerRegistry } from '@nestjs/schedule';
-import { NameService } from '@impler/services';
 import { UserJobEntity, UserJobRepository } from '@impler/dal';
 import { DocumentNotFoundException } from '@shared/exceptions/document-not-found.exception';
 
 @Injectable()
 export class UserJobDelete {
-  constructor(
-    private readonly nameService: NameService,
-    private readonly schedulerRegistry: SchedulerRegistry,
-    private readonly userJobRepository: UserJobRepository
-  ) {}
+  constructor(private readonly userJobRepository: UserJobRepository) {}
 
   async execute({ externalUserId, _jobId }: { externalUserId: string; _jobId: string }): Promise<UserJobEntity> {
     const userJobToDelete = await this.userJobRepository.findOne({ _id: _jobId, externalUserId });
@@ -24,11 +18,9 @@ export class UserJobDelete {
     }
 
     try {
-      this.schedulerRegistry.deleteCronJob(this.nameService.getCronName(_jobId));
+      await this.userJobRepository.delete({ _id: _jobId });
     } catch (error) {}
 
-    await this.userJobRepository.delete({ _id: _jobId });
-
     return userJobToDelete;
   }
 }