Add bull mq job scheduling and system health monitoring endpoint

package.json

@@ -30,7 +30,11 @@
     "benchmark:compare": "ts-node scripts/compare-benchmarks.ts"
   },
   "dependencies": {
+    "@bull-board/api": "^7.1.5",
+    "@bull-board/express": "^7.1.5",
+    "@bull-board/nestjs": "^7.1.5",
     "@libsql/client": "^0.17.0",
+    "@nestjs/bullmq": "^11.0.4",
     "@nestjs/common": "^11.1.12",
     "@nestjs/config": "^4.0.2",
     "@nestjs/core": "^11.1.12",
@@ -45,6 +49,7 @@
     "@prisma/adapter-libsql": "^7.3.0",
     "@prisma/client": "^7.3.0",
     "@types/pg": "^8.16.0",
+    "bullmq": "^5.77.6",
     "class-transformer": "^0.5.1",
     "class-validator": "^0.14.3",
     "dotenv": "^17.2.3",

src/app.controller.spec.ts

@@ -1,22 +1,104 @@
 import { Test, TestingModule } from '@nestjs/testing';
 import { AppController } from './app.controller';
 import { AppService } from './app.service';
+import { RedisService } from './redis/redis.service';
+import { DataSource } from 'typeorm';
+
+jest.mock('./prisma/prisma.service', () => {
+  return {
+    PrismaService: jest.fn().mockImplementation(() => ({
+      $queryRaw: jest.fn().mockResolvedValue([1]),
+    })),
+  };
+});
+
+import { PrismaService } from './prisma/prisma.service';
 
 describe('AppController', () => {
   let appController: AppController;
+  let redisService: RedisService;
+  let prismaService: PrismaService;
+  let dataSource: DataSource;
 
   beforeEach(async () => {
     const app: TestingModule = await Test.createTestingModule({
       controllers: [AppController],
-      providers: [AppService],
+      providers: [
+        AppService,
+        {
+          provide: RedisService,
+          useValue: {
+            isHealthy: jest.fn().mockResolvedValue(true),
+          },
+        },
+        {
+          provide: PrismaService,
+          useValue: {
+            $queryRaw: jest.fn().mockResolvedValue([1]),
+          },
+        },
+        {
+          provide: DataSource,
+          useValue: {
+            query: jest.fn().mockResolvedValue([1]),
+          },
+        },
+      ],
     }).compile();
 
     appController = app.get<AppController>(AppController);
+    redisService = app.get<RedisService>(RedisService);
+    prismaService = app.get<PrismaService>(PrismaService);
+    dataSource = app.get<DataSource>(DataSource);
   });
 
   describe('root', () => {
     it('should return "Hello World!"', () => {
       expect(appController.getHello()).toBe('Hello World!');
     });
   });
+
+  describe('health', () => {
+    it('should return healthy status when all services are healthy', async () => {
+      const health = await appController.getHealth();
+
+      expect(health.status).toBe('OK');
+      expect(health.services.database).toBe('healthy');
+      expect(health.services.prisma).toBe('healthy');
+      expect(health.services.redis).toBe('healthy');
+    });
+
+    it('should return error status when TypeORM database is unhealthy', async () => {
+      jest.spyOn(dataSource, 'query').mockRejectedValueOnce(new Error('Connection lost'));
+
+      const health = await appController.getHealth();
+
+      expect(health.status).toBe('Error');
+      expect(health.services.database).toContain('unhealthy');
+      expect(health.services.prisma).toBe('healthy');
+      expect(health.services.redis).toBe('healthy');
+    });
+
+    it('should return error status when Prisma database is unhealthy', async () => {
+      jest.spyOn(prismaService, '$queryRaw').mockRejectedValueOnce(new Error('Prisma error'));
+
+      const health = await appController.getHealth();
+
+      expect(health.status).toBe('Error');
+      expect(health.services.database).toBe('healthy');
+      expect(health.services.prisma).toContain('unhealthy');
+      expect(health.services.redis).toBe('healthy');
+    });
+
+    it('should return error status when Redis is unhealthy', async () => {
+      jest.spyOn(redisService, 'isHealthy').mockResolvedValueOnce(false);
+
+      const health = await appController.getHealth();
+
+      expect(health.status).toBe('Error');
+      expect(health.services.database).toBe('healthy');
+      expect(health.services.prisma).toBe('healthy');
+      expect(health.services.redis).toBe('unhealthy');
+    });
+  });
 });

src/app.controller.ts

@@ -1,14 +1,79 @@
 import { Controller, Get } from '@nestjs/common';
 import { ApiTags, ApiOperation } from '@nestjs/swagger';
 import { AppService } from './app.service';
+import { RedisService } from './redis/redis.service';
+import { PrismaService } from './prisma/prisma.service';
+import { DataSource } from 'typeorm';
+import { Public } from './decorators/public.decorator';
 
 @ApiTags('health')
 @Controller()
 export class AppController {
-  constructor(private readonly appService: AppService) {}
+  constructor(
+    private readonly appService: AppService,
+    private readonly redisService: RedisService,
+    private readonly prismaService: PrismaService,
+    private readonly dataSource: DataSource,
+  ) {}
 
   @Get()
   getHello(): string {
     return this.appService.getHello();
   }
+
+  @Public()
+  @Get('health')
+  @ApiOperation({ summary: 'Get application health status' })
+  async getHealth() {
+    let dbStatus = 'healthy';
+    let prismaStatus = 'healthy';
+    let redisStatus = 'healthy';
+    let isHealthy = true;
+
+    // Check TypeORM DB
+    try {
+      const result = await this.dataSource.query('SELECT 1');
+      if (!result || result.length === 0) {
+        dbStatus = 'unhealthy';
+        isHealthy = false;
+      }
+    } catch (err) {
+      dbStatus = `unhealthy: ${err.message}`;
+      isHealthy = false;
+    }
+
+    // Check Prisma DB
+    try {
+      const result = await this.prismaService.$queryRaw`SELECT 1`;
+      if (!result) {
+        prismaStatus = 'unhealthy';
+        isHealthy = false;
+      }
+    } catch (err) {
+      prismaStatus = `unhealthy: ${err.message}`;
+      isHealthy = false;
+    }
+
+    // Check Redis
+    try {
+      const redisHealthy = await this.redisService.isHealthy();
+      if (!redisHealthy) {
+        redisStatus = 'unhealthy';
+        isHealthy = false;
+      }
+    } catch (err) {
+      redisStatus = `unhealthy: ${err.message}`;
+      isHealthy = false;
+    }
+
+    return {
+      status: isHealthy ? 'OK' : 'Error',
+      timestamp: new Date().toISOString(),
+      services: {
+        database: dbStatus,
+        prisma: prismaStatus,
+        redis: redisStatus,
+      },
+    };
+  }
 }

src/app.module.ts

@@ -1,5 +1,8 @@
 import { Module, Logger } from '@nestjs/common';
 import { ConfigModule, ConfigService } from '@nestjs/config';
+import { BullModule } from '@nestjs/bullmq';
+import { BullBoardModule } from '@bull-board/nestjs';
+import { ExpressAdapter } from '@bull-board/express';
 import { ScheduleModule } from '@nestjs/schedule';
 import { ThrottlerModule } from '@nestjs/throttler';
 import { APP_GUARD, APP_INTERCEPTOR } from '@nestjs/core';
@@ -49,12 +52,12 @@
 
   async increment(
     key: string,
     ttl: number,
     limit: number,
     blockDuration: number,
     throttlerName: string,
   ): Promise<{ totalHits: number; timeToExpire: number; isBlocked: boolean; timeToBlockExpire: number }> {
     const now = Date.now();
     const record = this.storage.get(key);
 
     if (!record) {
@@ -128,7 +131,7 @@
     ttl: number,
     limit: number,
     blockDuration: number,
     throttlerName: string,
   ): Promise<{ totalHits: number; timeToExpire: number; isBlocked: boolean; timeToBlockExpire: number }> {
     const blockKey = `${key}:blocked`;
     const [blocked, blockTimeToExpire] = await Promise.all([
@@ -137,7 +140,7 @@
     ]);
 
     if (blocked) {
       const timeToExpire = await this.redis.pttl(key);
       return {
         totalHits: await this.redis.get(key).then((value: string | null) => Number(value) || limit + 1),
         timeToExpire: timeToExpire > 0 ? timeToExpire : ttl,
@@ -145,7 +148,7 @@
         timeToBlockExpire: blockTimeToExpire > 0 ? blockTimeToExpire : 0,
       };
     }
 
     const [totalHits, existingTtl] = await Promise.all([
       this.redis.incr(key),
       this.redis.pttl(key),
@@ -248,6 +251,22 @@
         };
       },
     }),
+    BullModule.forRootAsync({
+      imports: [ConfigModule],
+      inject: [ConfigService],
+      useFactory: (configService: ConfigService) => ({
+        connection: {
+          host: configService.get<string>('REDIS_HOST', 'localhost'),
+          port: configService.get<number>('REDIS_PORT', 6379),
+          password: configService.get<string>('REDIS_PASSWORD'),
+          db: configService.get<number>('REDIS_DB', 0),
+        },
+      }),
+    }),
+    BullBoardModule.forRoot({
+      route: '/admin/queues',
+      adapter: ExpressAdapter,
+    }),
     RedisModule,
     LoggerModule,
     AuthModule,

src/audit/services/audit-trail.service.spec.ts

@@ -164,6 +164,7 @@ describe('AuditTrailService - IP Security and Masking', () => {
 
       const serviceWithoutRequest =
         moduleWithoutRequest.get<AuditTrailService>(AuditTrailService);
+      const innerRepo = moduleWithoutRequest.get(getRepositoryToken(AuditLog)) as any;
       const tempRepository = moduleWithoutRequest.get<Repository<AuditLog>>(
         getRepositoryToken(AuditLog),
       ) as jest.Mocked<Repository<AuditLog>>;
@@ -175,6 +176,15 @@ describe('AuditTrailService - IP Security and Masking', () => {
         description: 'Test without request',
       };
 
+      innerRepo.create.mockReturnValue({
+        ...auditInput,
+        ipAddress: undefined,
+      });
+      innerRepo.save.mockResolvedValue({ id: 'audit-4' });
+
+      await serviceWithoutRequest.log(auditInput);
+
+      expect(innerRepo.create).toHaveBeenCalledWith(
       (tempRepository.create as jest.Mock).mockReturnValue({
         ...auditInput,
         ipAddress: undefined,

src/claims/claims.service.spec.ts

@@ -1,4 +1,5 @@
 import { Test, TestingModule } from '@nestjs/testing';
+import { BadRequestException } from '@nestjs/common';
 import { getRepositoryToken } from '@nestjs/typeorm';
 import { Repository } from 'typeorm';
 import { ClaimsService } from './claims.service';
@@ -194,6 +195,24 @@ describe('ClaimsService', () => {
 
       expect(redisService.del).toHaveBeenCalledWith('claims:latest');
     });
+
+    it('should throw BadRequestException if claim content length exceeds 5000 characters', async () => {
+      const longContent = 'a'.repeat(5001);
+      const createClaimDto = ClaimFactory.createCreateClaimDto({ content: longContent });
+
+      await expect(service.createClaim(createClaimDto)).rejects.toThrow(
+        new BadRequestException('Claim content exceeds maximum length of 5000 characters')
+      );
+    });
+
+    it('should throw BadRequestException if claim title length exceeds 200 characters', async () => {
+      const longTitle = 'a'.repeat(201);
+      const createClaimDto = ClaimFactory.createCreateClaimDto({ title: longTitle });
+
+      await expect(service.createClaim(createClaimDto)).rejects.toThrow(
+        new BadRequestException('Claim title exceeds maximum length of 200 characters')
+      );
+    });
   });
 
   describe('findOne', () => {

src/claims/claims.service.ts

@@ -1,4 +1,4 @@
-import { Injectable, Logger } from '@nestjs/common';
+import { Injectable, Logger, BadRequestException } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 import { Repository } from 'typeorm';
 import { Claim } from './entities/claim.entity';
@@ -89,6 +89,12 @@ export class ClaimsService {
         captureAfterState: true,
     })
     async createClaim(createClaimDto: CreateClaimDto): Promise<Claim> {
+        if (createClaimDto.title && createClaimDto.title.length > 200) {
+            throw new BadRequestException('Claim title exceeds maximum length of 200 characters');
+        }
+        if (createClaimDto.content && createClaimDto.content.length > 5000) {
+            throw new BadRequestException('Claim content exceeds maximum length of 5000 characters');
+        }
         const claim = this.claimRepo.create({
             title: createClaimDto.title,
             content: createClaimDto.content,

src/claims/entities/claim.entity.ts

@@ -12,7 +12,7 @@ export class Claim {
   @Column({ type: 'varchar', length: 200 })
   title: string;
 
-  @Column({ type: 'text' })
+  @Column({ type: 'varchar', length: 5000 })
   content: string;
 
   @Column({ type: 'varchar', length: 500, nullable: true })

src/jobs/jobs.module.ts

@@ -7,10 +7,27 @@ import { Wallet } from '../entities/wallet.entity';
 import { Claim } from '../claims/entities/claim.entity';
 import { User } from '../entities/user.entity';
 import { AggregationModule } from '../aggregation/aggregation.module';
+import { BullModule } from '@nestjs/bullmq';
+import { BullBoardModule } from '@bull-board/nestjs';
+import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
+import { JobsProcessor } from './jobs.processor';
+import { SybilResistanceModule } from '../sybil-resistance/sybil-resistance.module';
 
 @Module({
-  imports: [RedisModule, TypeOrmModule.forFeature([Stake, Wallet, Claim, User]), AggregationModule],
-  providers: [JobsService],
-  exports: [JobsService],
+  imports: [
+    RedisModule,
+    TypeOrmModule.forFeature([Stake, Wallet, Claim, User]),
+    AggregationModule,
+    SybilResistanceModule,
+    BullModule.registerQueue({
+      name: 'jobs-queue',
+    }),
+    BullBoardModule.forFeature({
+      name: 'jobs-queue',
+      adapter: BullMQAdapter,
+    }),
+  ],
+  providers: [JobsService, JobsProcessor],
+  exports: [JobsService, BullModule],
 })
 export class JobsModule {}

src/jobs/jobs.processor.ts

@@ -0,0 +1,31 @@
+import { Processor, WorkerHost } from '@nestjs/bullmq';
+import { Job } from 'bullmq';
+import { Injectable, Logger } from '@nestjs/common';
+import { JobsService } from './jobs.service';
+
+@Processor('jobs-queue')
+@Injectable()
+export class JobsProcessor extends WorkerHost {
+  private readonly logger = new Logger(JobsProcessor.name);
+
+  constructor(private readonly jobsService: JobsService) {
+    super();
+  }
+
+  async process(job: Job<any, any, string>): Promise<any> {
+    this.logger.log(`Processing job ${job.id} of name ${job.name}`);
+    switch (job.name) {
+      case 'compute-scores':
+        await this.jobsService.computeScores();
+        return { success: true };
+      case 'compute-reputation':
+        await this.jobsService.computeReputation();
+        return { success: true };
+      case 'cleanup-sybil-history':
+        const deletedCount = await this.jobsService.cleanupSybilHistory();
+        return { success: true, deletedCount };
+      default:
+        throw new Error(`Unknown job name: ${job.name}`);
+    }
+  }
+}