Feat: Stuck background job

api/.env.template

@@ -58,6 +58,8 @@ LISTING_OPEN_DATE_NOTIFICATION_CRON_STRING=0 9 * * *
 LOTTERY_PUBLISH_PROCESSING_CRON_STRING=58 23 * * *
 # controls the repetition of the lottery cron job
 LOTTERY_PROCESSING_CRON_STRING=0 * * * *
+# controls the repetition of the stuck background jobs recovery job
+BACKGROUND_JOBS_RECOVERY_CRON_STRING=0/30 * * * *
 # how many days till lottery data expires
 LOTTERY_DAYS_TILL_EXPIRY=45
 # controls the repetition of the msq retire cron job (should occur after LISTING_PROCESSING_CRON_STRING)

api/src/modules/background-jobs.module.ts

@@ -1,13 +1,14 @@
-import { Module } from '@nestjs/common';
+import { Logger, Module } from '@nestjs/common';
 import { PrismaModule } from './prisma.module';
 import { S3Module } from './s3.module';
 import { PermissionModule } from './permission.module';
 import { BackgroundJobsController } from '../controllers/background-jobs.controller';
 import { BackgroundJobsService } from '../services/background-jobs.service';
+import { CronJobModule } from './cron-job.module';
 
 @Module({
-  imports: [PrismaModule, S3Module, PermissionModule],
-  providers: [BackgroundJobsService],
+  imports: [PrismaModule, S3Module, PermissionModule, CronJobModule],
+  providers: [BackgroundJobsService, Logger],
   controllers: [BackgroundJobsController],
   exports: [BackgroundJobsService],
 })

api/src/services/background-jobs.service.ts

@@ -1,6 +1,8 @@
 import {
   ConflictException,
+  Inject,
   Injectable,
+  Logger,
   NotFoundException,
 } from '@nestjs/common';
 import { PrismaService } from './prisma.service';
@@ -13,15 +15,29 @@ import { BackgroundJobCreate } from '../dtos/background-jobs/background-job-crea
 import { PermissionService } from './permission.service';
 import { permissionActions } from '../enums/permissions/permission-actions-enum';
 import { SuccessDTO } from '../dtos/shared/success.dto';
+import { CronJobService } from './cron-job.service';
 
+const BACKGROUND_JOBS_RECOVERY_JOB_NAME = 'BACKGROUND_JOBS_RECOVERY_CRON_JOB';
+export const BACKGROUND_JOB_STALE_TIME_IN_MINUTES = 30;
 @Injectable()
 export class BackgroundJobsService {
   constructor(
     private readonly prismaService: PrismaService,
     private readonly s3Service: S3Service,
+    private readonly cronJobService: CronJobService,
     private readonly permissionService: PermissionService,
+    @Inject(Logger)
+    private logger = new Logger(BackgroundJobsService.name),
   ) {}
 
+  onModuleInit() {
+    this.cronJobService.startCronJob(
+      BACKGROUND_JOBS_RECOVERY_JOB_NAME,
+      process.env.BACKGROUND_JOBS_RECOVERY_CRON_STRING,
+      this.recoverStuckJobCronJob.bind(this),
+    );
+  }
+
   /**
    * Creates an instance of a background job runner for a listing
    * @param dto - background job creation DTO
@@ -144,4 +160,61 @@ export class BackgroundJobsService {
       success: !!activeJob,
     };
   }
+
+  async recoverStuckJobCronJob(): Promise<SuccessDTO> {
+    const logName = 'recoverStuckJobCron';
+    this.logger.warn(`${logName} job running`);
+    const currentTime = new Date();
+
+    this.cronJobService.markCronJobAsStarted(BACKGROUND_JOBS_RECOVERY_JOB_NAME);
+
+    const runningJobs = await this.prismaService.backgroundJob.findMany({
+      select: {
+        id: true,
+        updatedAt: true,
+      },
+      where: {
+        status: BackgroundJobStatusEnum.processing,
+      },
+    });
+
+    if (!runningJobs.length) {
+      this.logger.warn('No jobs are currently running');
+      return {
+        success: true,
+      };
+    }
+
+    const markedForDeletion = [];
+    runningJobs.forEach((job) => {
+      if (
+        (currentTime.getTime() - job.updatedAt.getTime()) / 60000 >
+        BACKGROUND_JOB_STALE_TIME_IN_MINUTES
+      ) {
+        markedForDeletion.push(job.id);
+      }
+    });
+
+    this.logger.warn(
+      `${markedForDeletion.length} running jobs have been marked for deletion`,
+    );
+
+    const deletedJobs = await this.prismaService.backgroundJob.deleteMany({
+      where: {
+        id: {
+          in: markedForDeletion,
+        },
+      },
+    });
+
+    if (deletedJobs.count !== markedForDeletion.length) {
+      this.logger.error(
+        `Failed to delete all the marked jobs (${deletedJobs.count}/${markedForDeletion.length} jobs deleted)`,
+      );
+    }
+
+    return {
+      success: true,
+    };
+  }
 }

api/test/unit/services/background-jobs.service.spec.ts

@@ -2,16 +2,22 @@ import { Test, TestingModule } from '@nestjs/testing';
 import {
   ConflictException,
   ForbiddenException,
+  Logger,
   NotFoundException,
 } from '@nestjs/common';
+import { SchedulerRegistry } from '@nestjs/schedule';
 import { BackgroundJobStatusEnum } from '@prisma/client';
 import { randomUUID } from 'crypto';
-import { BackgroundJobsService } from '../../../src/services/background-jobs.service';
+import {
+  BACKGROUND_JOB_STALE_TIME_IN_MINUTES,
+  BackgroundJobsService,
+} from '../../../src/services/background-jobs.service';
 import { PrismaService } from '../../../src/services/prisma.service';
 import { S3Service } from '../../../src/services/s3.service';
 import { PermissionService } from '../../../src/services/permission.service';
 import { BackgroundJob } from '../../../src/dtos/background-jobs/background-job.dto';
 import { UserRoleEnum } from '../../../src/enums/permissions/user-role-enum';
+import { CronJobService } from '../../../src/services/cron-job.service';
 
 const listingId = randomUUID();
 const jobId = randomUUID();
@@ -48,6 +54,9 @@ describe('Background Jobs Service Tests', () => {
       providers: [
         BackgroundJobsService,
         PrismaService,
+        Logger,
+        SchedulerRegistry,
+        CronJobService,
         { provide: S3Service, useValue: {} },
         {
           provide: PermissionService,
@@ -196,4 +205,53 @@ describe('Background Jobs Service Tests', () => {
       });
     });
   });
+
+  describe('recoverStuckJobCronJob', () => {
+    it(`should delete only jobs running fro more than ${BACKGROUND_JOB_STALE_TIME_IN_MINUTES} minutes`, async () => {
+      const runningValidJobId = randomUUID();
+      const runningStaleJobId = randomUUID();
+
+      prisma.cronJob.findFirst = jest
+        .fn()
+        .mockResolvedValue({ id: randomUUID() });
+      prisma.cronJob.update = jest.fn().mockResolvedValue(true);
+      prisma.backgroundJob.findMany = jest.fn().mockReturnValue([
+        {
+          id: runningValidJobId,
+          updatedAt: new Date(),
+        },
+        {
+          id: runningStaleJobId,
+          updatedAt: new Date(
+            new Date().getTime() -
+              (BACKGROUND_JOB_STALE_TIME_IN_MINUTES + 1) * 60000,
+          ),
+        },
+      ]);
+      prisma.backgroundJob.deleteMany = jest.fn().mockResolvedValue({
+        count: 1,
+      });
+
+      await service.recoverStuckJobCronJob();
+
+      expect(prisma.backgroundJob.findMany).toHaveBeenCalledTimes(1);
+      expect(prisma.backgroundJob.findMany).toHaveBeenCalledWith({
+        select: {
+          id: true,
+          updatedAt: true,
+        },
+        where: {
+          status: BackgroundJobStatusEnum.processing,
+        },
+      });
+      expect(prisma.backgroundJob.deleteMany).toHaveBeenCalledTimes(1);
+      expect(prisma.backgroundJob.deleteMany).toHaveBeenCalledWith({
+        where: {
+          id: {
+            in: [runningStaleJobId],
+          },
+        },
+      });
+    });
+  });
 });