Merge branch 'next' into feat/password-validation

Signed-off-by: Bhavik Chavda <[email protected]>

Author: chavda-bhavik

apps/api/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@impler/api",
-  "version": "0.27.1",
+  "version": "0.28.0",
   "author": "implerhq",
   "license": "MIT",
   "private": true,
@@ -48,6 +48,7 @@
     "compression": "^1.7.4",
     "cookie-parser": "^1.4.6",
     "cron": "^3.1.9",
+    "cron-parser": "^4.9.0",
     "date-fns": "^2.30.0",
     "dayjs": "^1.11.11",
     "dotenv": "^16.0.2",

apps/api/src/app.module.ts

@@ -16,6 +16,8 @@ import { ActivityModule } from './app/activity/activity.module';
 import { UserModule } from './app/user/user.module';
 import { ImportJobsModule } from 'app/import-jobs/import-jobs.module';
 import { TeamModule } from 'app/team/team.module';
+import { AutoImportJobsSchedularModule } from 'app/auto-import-jobs-schedular/auto-import-jobs-schedular.module';
+import { FailedWebhookRetryModule } from 'app/failed-webhook-request-retry/failed-webhook-retry.module';
 
 const modules: Array<Type | DynamicModule | Promise<DynamicModule> | ForwardReference> = [
   ProjectModule,
@@ -33,6 +35,8 @@ const modules: Array<Type | DynamicModule | Promise<DynamicModule> | ForwardRefe
   ActivityModule,
   ImportJobsModule,
   TeamModule,
+  AutoImportJobsSchedularModule,
+  FailedWebhookRetryModule,
 ];
 
 const providers = [Logger];

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

@@ -157,16 +157,13 @@ export class AuthController {
     @UserSession() user: IJwtPayload,
     @Res({ passthrough: true }) res: Response
   ) {
-    const projectWithEnvironment = await this.onboardUser.execute(
-      {
-        _userId: user._id,
-        projectName: body.projectName,
-        role: body.role,
-        companySize: body.companySize,
-        source: body.source,
-      },
-      user.email
-    );
+    const projectWithEnvironment = await this.onboardUser.execute({
+      _userId: user._id,
+      projectName: body.projectName,
+      role: body.role,
+      companySize: body.companySize,
+      source: body.source,
+    });
 
     const userApiKey = projectWithEnvironment.environment.apiKeys.find(
       (apiKey) => apiKey._userId.toString() === user._id

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

@@ -14,7 +14,13 @@ import { RequestForgotPasswordCommand } from './request-forgot-password/request-
 import { CreateProject } from 'app/project/usecases';
 import { SaveSampleFile, UpdateImageColumns } from '@shared/usecases';
 import { CreateEnvironment, GenerateUniqueApiKey } from 'app/environment/usecases';
-import { CreateTemplate, UpdateCustomization, UpdateTemplateColumns } from 'app/template/usecases';
+import {
+  AddColumn,
+  CreateTemplate,
+  GetImportFileSchema,
+  UpdateCustomization,
+  UpdateTemplateColumns,
+} from 'app/template/usecases';
 
 export const USE_CASES = [
   Verify,
@@ -24,6 +30,8 @@ export const USE_CASES = [
   RegisterUser,
   ResetPassword,
   CreateProject,
+  GetImportFileSchema,
+  AddColumn,
   SaveSampleFile,
   CreateTemplate,
   CreateEnvironment,

apps/api/src/app/auth/usecases/onboard-user/onboard-user.usecase.ts

@@ -16,14 +16,13 @@ export class OnboardUser {
     private paymentAPIService: PaymentAPIService
   ) {}
 
-  async execute(command: OnboardUserCommand, email: string) {
+  async execute(command: OnboardUserCommand) {
     const createdProject = await this.createProject.execute(
       CreateProjectCommand.create({
         _userId: command._userId,
         name: command.projectName,
         onboarding: true,
-      }),
-      email
+      })
     );
 
     await this.userRepository.update(

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

@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+import { SharedModule } from '@shared/shared.module';
+import { ScheduleModule } from '@nestjs/schedule';
+import { USE_CASES } from './usecase';
+
+@Module({
+  imports: [ScheduleModule.forRoot(), SharedModule],
+  providers: [...USE_CASES],
+})
+export class AutoImportJobsSchedularModule {}

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

@@ -0,0 +1,102 @@
+import { UserJobEntity, UserJobRepository, WebhookDestinationRepository } from '@impler/dal';
+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 { UserJobImportStatusEnum } from '@impler/shared';
+import { CRON_SCHEDULE } from '@shared/constants';
+const parseCronExpression = require('@impler/shared/src/utils/cronstrue');
+
+@Injectable()
+export class AutoImportJobsSchedular {
+  constructor(
+    private readonly userJobRepository: UserJobRepository,
+    private readonly webhookDestinationRepository: WebhookDestinationRepository,
+    private readonly updateUserJob: UpdateUserJob,
+    private readonly scheduleUserJob: ScheduleUserJob
+  ) {}
+
+  @Cron(CRON_SCHEDULE.DEFAULT_CRON_TIME)
+  async handleCronSchedular() {
+    await this.fetchAndExecuteScheduledJobs();
+  }
+
+  private async fetchAndExecuteScheduledJobs() {
+    const now = dayjs();
+    const userJobs = await this.userJobRepository.find({});
+
+    for (const userJob of userJobs) {
+      if (await this.shouldCroneRun({ userJob })) {
+        try {
+          const interval = parser.parseExpression(userJob.cron);
+
+          const nextScheduledTime = dayjs(interval.next().toDate().toDateString());
+
+          if (this.isJobDueToday(userJob.cron, now)) {
+            await this.scheduleUpdateNextRun(userJob._id, nextScheduledTime, dayjs(userJob.endsOn));
+
+            await this.updateUserJob.execute(userJob._id, userJob);
+
+            await this.scheduleUserJob.execute(userJob._id, userJob.cron);
+          }
+        } catch (error) {}
+      }
+    }
+  }
+
+  async scheduleUpdateNextRun(userJobId: string, nextRunSchedule: dayjs.Dayjs, endsOn: dayjs.Dayjs) {
+    const nextRunValue = dayjs(nextRunSchedule).isAfter(endsOn) && endsOn ? undefined : nextRunSchedule;
+
+    await this.userJobRepository.update({ _id: userJobId }, { $set: { nextRun: nextRunValue } });
+  }
+
+  async parseCronExpression(userCronExpression: string) {
+    if (!userCronExpression || userCronExpression.trim() === '') {
+      throw new Error('Cron expression is empty');
+    }
+
+    return parseCronExpression.toString(userCronExpression);
+  }
+
+  private isJobDueToday(cronExpression: string, currentDate: dayjs.Dayjs): boolean {
+    try {
+      const interval = parser.parseExpression(cronExpression);
+
+      const nextScheduledTime = dayjs(interval.next().toDate());
+
+      return nextScheduledTime.isSame(currentDate, 'd');
+    } catch (error) {
+      return false;
+    }
+  }
+
+  async fetchDestination(templateId: string) {
+    const webhookDestination = await this.webhookDestinationRepository.findOne({
+      _templateId: templateId,
+    });
+
+    if (!webhookDestination) {
+      return false;
+    }
+
+    const { _id, callbackUrl } = webhookDestination;
+
+    return { _id, callbackUrl };
+  }
+
+  async shouldCroneRun({ userJob }: { userJob: UserJobEntity }): Promise<boolean> {
+    const now = dayjs();
+
+    if (
+      (userJob.cron && userJob.status === UserJobImportStatusEnum.SCHEDULING) ||
+      userJob.status === UserJobImportStatusEnum.RUNNING ||
+      (userJob.status === 'Completed' && (await this.fetchDestination(userJob._templateId)) && !userJob.endsOn) ||
+      !dayjs(userJob.endsOn).isSame(now, 'd')
+    ) {
+      return true;
+    }
+
+    return false;
+  }
+}

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

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

apps/api/src/app/failed-webhook-request-retry/failed-webhook-retry.module.ts

@@ -0,0 +1,10 @@
+import { Module } from '@nestjs/common';
+import { SharedModule } from '@shared/shared.module';
+import { ScheduleModule } from '@nestjs/schedule';
+import { USE_CASES } from './usecase';
+
+@Module({
+  imports: [ScheduleModule.forRoot(), SharedModule],
+  providers: [...USE_CASES],
+})
+export class FailedWebhookRetryModule {}

apps/api/src/app/failed-webhook-request-retry/usecase/failed-webhook-request-retry.usecase.ts

@@ -0,0 +1,39 @@
+import { Injectable } from '@nestjs/common';
+import { Cron, CronExpression } from '@nestjs/schedule';
+
+import { QueuesEnum } from '@impler/shared';
+import { QueueService } from '@shared/services/queue.service';
+import { FailedWebhookRetryRequestsEntity, FailedWebhookRetryRequestsRepository } from '@impler/dal';
+
+@Injectable()
+export class FailedWebhookRetry {
+  constructor(
+    private failedWebhookRetryRequestsRepository: FailedWebhookRetryRequestsRepository = new FailedWebhookRetryRequestsRepository(),
+    private queueService: QueueService
+  ) {}
+
+  @Cron(CronExpression.EVERY_5_MINUTES)
+  async processWebhookRetries() {
+    try {
+      const failedWebhooks: FailedWebhookRetryRequestsEntity[] = await this.failedWebhookRetryRequestsRepository.find({
+        nextRequestTime: { $lt: new Date() },
+      });
+
+      if (!failedWebhooks.length) {
+        return;
+      }
+
+      await Promise.allSettled(failedWebhooks.map((wbh) => this.processWebhook(wbh)));
+    } catch (error) {
+      throw error;
+    }
+  }
+
+  private async processWebhook(webhook: FailedWebhookRetryRequestsEntity) {
+    try {
+      this.queueService.publishToQueue(QueuesEnum.SEND_FAILED_WEBHOOK_DATA, webhook._id as string);
+    } catch (error) {
+      throw error;
+    }
+  }
+}