Add bull mq job scheduling and system health monitoring endpoint

package.json

@@ -27,7 +27,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",
@@ -42,6 +46,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 { ThrottlerModule } from '@nestjs/throttler';
 import { APP_GUARD, APP_INTERCEPTOR } from '@nestjs/core';
 import { AppController } from './app.controller';
@@ -245,6 +248,22 @@ async function createThrottlerStorage(configService: ConfigService): Promise<any
         };
       },
     }),
+    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

@@ -8,7 +8,7 @@ import { AuditActionType, AuditEntityType } from '../entities/audit-log.entity';
 
 describe('AuditTrailService - IP Security', () => {
   let service: AuditTrailService;
-  let repository: jest.Mocked<Repository<AuditLog>>;
+  let repository: any;
   let mockRequest: any;
 
   beforeEach(async () => {
@@ -153,6 +153,7 @@ describe('AuditTrailService - IP Security', () => {
 
       const serviceWithoutRequest =
         moduleWithoutRequest.get<AuditTrailService>(AuditTrailService);
+      const innerRepo = moduleWithoutRequest.get(getRepositoryToken(AuditLog)) as any;
 
       const auditInput = {
         actionType: AuditActionType.CLAIM_CREATED,
@@ -161,15 +162,15 @@ describe('AuditTrailService - IP Security', () => {
         description: 'Test without request',
       };
 
-      repository.create.mockReturnValue({
+      innerRepo.create.mockReturnValue({
         ...auditInput,
         ipAddress: undefined,
       });
-      repository.save.mockResolvedValue({ id: 'audit-4' });
+      innerRepo.save.mockResolvedValue({ id: 'audit-4' });
 
       await serviceWithoutRequest.log(auditInput);
 
-      expect(repository.create).toHaveBeenCalledWith(
+      expect(innerRepo.create).toHaveBeenCalledWith(
         expect.objectContaining({
           ipAddress: undefined,
         }),

src/claims/claim-resolution.service.spec.ts

@@ -1,7 +1,7 @@
 import { ClaimResolutionService } from "./claim-resolution.service";
 
 describe('Confidence Scoring', () => {
-  const service = new ClaimResolutionService(null as any);
+  const service = new ClaimResolutionService(null as any, null as any);
 
   it('returns high confidence for strong consensus', () => {
     const score = service.computeConfidenceScore({

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,11 +89,17 @@ 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,
-            source: createClaimDto.source,
-            metadata: createClaimDto.metadata,
+            source: createClaimDto.source ?? null,
+            metadata: createClaimDto.metadata ?? null,
             resolvedVerdict: null, // Will be computed later
             confidenceScore: null, // Will be computed later
             finalized: false,
@@ -172,16 +178,5 @@ export class ClaimsService {
 
         return updatedClaim;
     }
-     async findOne(id: string): Promise<Claim> {
-    const claim = await this.repo.findOne({
-      where: { id },
-    });
-
-    if (!claim) {
-      throw new NotFoundException(`Claim with id ${id} not found`);
-    }
-
-    return claim;
-  }
 }
 

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 {}