diff --git a/src/app.module.ts b/src/app.module.ts index 7972aca..c43ad7b 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -49,7 +49,9 @@ class ThrottlerMemoryStorage { private readonly logger = new Logger('ThrottlerMemoryStorage'); constructor() { - this.logger.log('Using in-memory storage for rate limiting (development mode)'); + this.logger.log( + 'Using in-memory storage for rate limiting (development mode)', + ); } async increment( @@ -58,7 +60,12 @@ class ThrottlerMemoryStorage { limit: number, blockDuration: number, throttlerName: string, - ): Promise<{ totalHits: number; timeToExpire: number; isBlocked: boolean; timeToBlockExpire: number }> { + ): Promise<{ + totalHits: number; + timeToExpire: number; + isBlocked: boolean; + timeToBlockExpire: number; + }> { const now = Date.now(); const record = this.storage.get(key); @@ -113,7 +120,9 @@ class ThrottlerMemoryStorage { totalHits: record.totalHits, timeToExpire: Math.max(record.expiresAt - now, 0), isBlocked: record.isBlocked, - timeToBlockExpire: record.isBlocked ? Math.max(record.blockExpiresAt - now, 0) : 0, + timeToBlockExpire: record.isBlocked + ? Math.max(record.blockExpiresAt - now, 0) + : 0, }; } } @@ -134,7 +143,12 @@ class ThrottlerRedisStorage { limit: number, blockDuration: number, throttlerName: string, - ): Promise<{ totalHits: number; timeToExpire: number; isBlocked: boolean; timeToBlockExpire: number }> { + ): Promise<{ + totalHits: number; + timeToExpire: number; + isBlocked: boolean; + timeToBlockExpire: number; + }> { const blockKey = `${key}:blocked`; const [blocked, blockTimeToExpire] = await Promise.all([ this.redis.exists(blockKey), @@ -144,7 +158,9 @@ class ThrottlerRedisStorage { if (blocked) { const timeToExpire = await this.redis.pttl(key); return { - totalHits: await this.redis.get(key).then((value: string | null) => Number(value) || limit + 1), + totalHits: await this.redis + .get(key) + .then((value: string | null) => Number(value) || limit + 1), timeToExpire: timeToExpire > 0 ? timeToExpire : ttl, isBlocked: true, timeToBlockExpire: blockTimeToExpire > 0 ? blockTimeToExpire : 0, @@ -185,13 +201,18 @@ class ThrottlerRedisStorage { } // Factory to create appropriate storage based on environment -async function createThrottlerStorage(configService: ConfigService): Promise { +async function createThrottlerStorage( + configService: ConfigService, +): Promise { const useRedis = configService.get('REDIS_HOST'); if (useRedis) { try { const Redis = (await import('ioredis')).default; - const redisHost = configService.get('throttler.redis.host', 'localhost'); + const redisHost = configService.get( + 'throttler.redis.host', + 'localhost', + ); const redisPort = configService.get('throttler.redis.port', 6379); const redis = new Redis({ @@ -212,7 +233,9 @@ async function createThrottlerStorage(configService: ConfigService): Promise, + private readonly redisService: RedisService, + ) {} + + async get(key: string, environment?: string): Promise { + const env = environment ?? this.getDefaultEnvironment(); + const cached = await this.redisService.get(this.cacheKey(key, env)); + if (cached) { + try { + return JSON.parse(cached) as T; + } catch { + // fall through to DB + } + } + + const record = await this.configRepo.findOne({ + where: { key, environment: env }, + }); + if (!record) return null; + + await this.redisService.set( + this.cacheKey(key, env), + JSON.stringify(record.value), + CACHE_TTL_SECONDS, + ); + return record.value as T; + } + + async getRequired( + key: string, + environment?: string, + ): Promise { + const value = await this.get(key, environment); + if (value === null) { + throw new NotFoundException(`Configuration ${key} not found`); + } + return value; + } + + async set( + key: string, + value: T, + environment?: string, + createdBy?: string, + changeReason?: string, + ): Promise { + const env = environment ?? this.getDefaultEnvironment(); + const existing = await this.configRepo.findOne({ + where: { key, environment: env }, + }); + + if (existing) { + const nextVersion = existing.version + 1; + await this.configRepo.update(existing.id, { + value: value as unknown, + version: nextVersion, + createdBy, + changeReason, + }); + await this.invalidateCache(key, env); + return this.configRepo.findOneOrFail({ where: { id: existing.id } }); + } + + const record = this.configRepo.create({ + key, + value: value as unknown, + environment: env, + version: 1, + createdBy, + changeReason, + }); + const saved = await this.configRepo.save(record); + await this.invalidateCache(key, env); + return saved; + } + + async findAll(environment?: string): Promise { + const env = environment ?? this.getDefaultEnvironment(); + return this.configRepo.find({ + where: { environment: env }, + order: { key: 'ASC' }, + }); + } + + async findOne(id: string): Promise { + const record = await this.configRepo.findOne({ where: { id } }); + if (!record) throw new NotFoundException(`Configuration ${id} not found`); + return record; + } + + async delete(id: string): Promise { + const record = await this.findOne(id); + await this.configRepo.delete(id); + await this.invalidateCache(record.key, record.environment); + } + + async getHistory( + key: string, + environment?: string, + limit = 50, + ): Promise { + const env = environment ?? this.getDefaultEnvironment(); + const records = await this.configRepo.find({ + where: { key, environment: env }, + order: { createdAt: 'DESC' }, + take: limit, + }); + return records.map((r) => ({ + id: r.id, + key: r.key, + value: r.value, + version: r.version, + createdBy: r.createdBy, + changeReason: r.changeReason, + createdAt: r.createdAt, + })); + } + + private async invalidateCache( + key: string, + environment: string, + ): Promise { + await this.redisService.del(this.cacheKey(key, environment)); + } + + private cacheKey(key: string, environment: string): string { + return `config:${environment}:${key}`; + } + + private getDefaultEnvironment(): string { + return process.env.NODE_ENV ?? 'development'; + } +} diff --git a/src/feature-flags/entities/configuration-value.entity.ts b/src/feature-flags/entities/configuration-value.entity.ts new file mode 100644 index 0000000..dd51c19 --- /dev/null +++ b/src/feature-flags/entities/configuration-value.entity.ts @@ -0,0 +1,39 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + UpdateDateColumn, + Index, +} from 'typeorm'; + +@Entity('configuration_values') +@Index(['key', 'environment'], { unique: true }) +export class ConfigurationValue { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column() + key: string; + + @Column({ type: 'json' }) + value: unknown; + + @Column({ default: 'development' }) + environment: string; + + @Column({ default: 1 }) + version: number; + + @Column({ nullable: true }) + createdBy: string; + + @Column({ type: 'text', nullable: true }) + changeReason: string; + + @CreateDateColumn() + createdAt: Date; + + @UpdateDateColumn() + updatedAt: Date; +} diff --git a/src/feature-flags/entities/feature-flag.entity.ts b/src/feature-flags/entities/feature-flag.entity.ts new file mode 100644 index 0000000..d331443 --- /dev/null +++ b/src/feature-flags/entities/feature-flag.entity.ts @@ -0,0 +1,59 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + UpdateDateColumn, + Index, +} from 'typeorm'; + +export type FeatureFlagType = + | 'boolean' + | 'percentage' + | 'user' + | 'role' + | 'environment' + | 'time'; + +@Entity('feature_flags') +@Index(['key', 'environment'], { unique: true }) +export class FeatureFlag { + @PrimaryGeneratedColumn('uuid') + id: string; + + @Column() + key: string; + + @Column({ type: 'varchar', default: 'boolean' }) + type: FeatureFlagType; + + @Column({ default: false }) + enabled: boolean; + + @Column({ type: 'float', default: 0 }) + rolloutPercentage: number; + + @Column({ type: 'json', nullable: true }) + rules: Record; + + @Column({ default: 'development' }) + environment: string; + + @Column({ type: 'text', nullable: true }) + description: string; + + @Column({ nullable: true }) + expiresAt: Date; + + @Column({ default: 1 }) + version: number; + + @Column({ nullable: true }) + createdBy: string; + + @CreateDateColumn() + createdAt: Date; + + @UpdateDateColumn() + updatedAt: Date; +} diff --git a/src/feature-flags/feature-flags.controller.ts b/src/feature-flags/feature-flags.controller.ts new file mode 100644 index 0000000..886b86e --- /dev/null +++ b/src/feature-flags/feature-flags.controller.ts @@ -0,0 +1,247 @@ +import { + Body, + Controller, + Delete, + Get, + Param, + Patch, + Post, + Query, + UseGuards, + Request, +} from '@nestjs/common'; +import { ApiBearerAuth, ApiOperation, ApiTags } from '@nestjs/swagger'; +import { JwtAuthGuard } from '../auth/jwt-auth.guard'; +import { FeatureFlagsService } from './feature-flags.service'; +import { ConfigurationService } from './configuration.service'; +import { AuditTrailService } from '../audit/services/audit-trail.service'; +import { + AuditActionType, + AuditEntityType, +} from '../audit/entities/audit-log.entity'; +import { FeatureFlag } from './entities/feature-flag.entity'; +import { ConfigurationValue } from './entities/configuration-value.entity'; +import { + CreateFeatureFlagInput, + FeatureFlagContext, + FeatureFlagEvaluationResult, + UpdateFeatureFlagInput, +} from './feature-flags.types'; + +interface RequestWithUser { + user?: { userId?: string; id?: string }; +} + +@ApiTags('Feature Flags & Configuration') +@Controller() +export class FeatureFlagsController { + constructor( + private readonly flagsService: FeatureFlagsService, + private readonly configService: ConfigurationService, + private readonly auditTrailService: AuditTrailService, + ) {} + + @Get('feature-flags') + @ApiOperation({ summary: 'List feature flags' }) + listFlags( + @Query('environment') environment?: string, + ): Promise { + return this.flagsService.findAll(environment); + } + + @Get('feature-flags/evaluate/:key') + @ApiOperation({ summary: 'Evaluate a feature flag' }) + async evaluateFlag( + @Param('key') key: string, + @Query('userId') userId?: string, + @Query('roles') roles?: string, + @Query('environment') environment?: string, + ): Promise { + const context: FeatureFlagContext = { + userId, + roles: roles ? roles.split(',') : undefined, + environment, + }; + return this.flagsService.evaluate(key, context); + } + + @Get('feature-flags/:id') + @ApiOperation({ summary: 'Get a feature flag by id' }) + getFlag(@Param('id') id: string): Promise { + return this.flagsService.findOne(id); + } + + @Post('feature-flags') + @UseGuards(JwtAuthGuard) + @ApiBearerAuth() + @ApiOperation({ summary: 'Create a feature flag' }) + async createFlag( + @Body() input: CreateFeatureFlagInput, + @Request() req: RequestWithUser, + ): Promise { + const userId = this.userIdFrom(req); + const flag = await this.flagsService.create({ + ...input, + createdBy: userId, + }); + await this.auditTrailService.log({ + actionType: AuditActionType.CONFIG_CREATED, + entityType: AuditEntityType.CONFIGURATION, + entityId: flag.id, + userId, + description: `Created feature flag ${flag.key}`, + afterState: flag as unknown as Record, + }); + return flag; + } + + @Patch('feature-flags/:id') + @UseGuards(JwtAuthGuard) + @ApiBearerAuth() + @ApiOperation({ summary: 'Update a feature flag' }) + async updateFlag( + @Param('id') id: string, + @Body() input: UpdateFeatureFlagInput, + @Request() req: RequestWithUser, + ): Promise { + const userId = this.userIdFrom(req); + const before = await this.flagsService.findOne(id); + const flag = await this.flagsService.update(id, input, userId); + await this.auditTrailService.log({ + actionType: AuditActionType.CONFIG_UPDATED, + entityType: AuditEntityType.CONFIGURATION, + entityId: flag.id, + userId, + description: `Updated feature flag ${flag.key}`, + beforeState: before as unknown as Record, + afterState: flag as unknown as Record, + }); + return flag; + } + + @Post('feature-flags/:id/toggle') + @UseGuards(JwtAuthGuard) + @ApiBearerAuth() + @ApiOperation({ summary: 'Toggle a feature flag' }) + async toggleFlag( + @Param('id') id: string, + @Body('enabled') enabled: boolean, + @Request() req: RequestWithUser, + ): Promise { + const userId = this.userIdFrom(req); + const flag = await this.flagsService.toggle(id, enabled, userId); + await this.auditTrailService.log({ + actionType: enabled + ? AuditActionType.FEATURE_FLAG_ENABLED + : AuditActionType.FEATURE_FLAG_DISABLED, + entityType: AuditEntityType.CONFIGURATION, + entityId: flag.id, + userId, + description: `${enabled ? 'Enabled' : 'Disabled'} feature flag ${flag.key}`, + afterState: { enabled } as Record, + }); + return flag; + } + + @Post('feature-flags/:id/rollback') + @UseGuards(JwtAuthGuard) + @ApiBearerAuth() + @ApiOperation({ summary: 'Rollback a feature flag' }) + async rollbackFlag( + @Param('id') id: string, + @Body('targetVersion') targetVersion: number, + @Request() req: RequestWithUser, + ): Promise { + const userId = this.userIdFrom(req); + const flag = await this.flagsService.rollback(id, targetVersion); + await this.auditTrailService.log({ + actionType: AuditActionType.CONFIG_ROLLED_BACK, + entityType: AuditEntityType.CONFIGURATION, + entityId: flag.id, + userId, + description: `Rolled back feature flag ${flag.key} to version ${targetVersion}`, + afterState: { version: flag.version } as Record, + }); + return flag; + } + + @Get('configuration') + @ApiOperation({ summary: 'List configuration values' }) + listConfig( + @Query('environment') environment?: string, + ): Promise { + return this.configService.findAll(environment); + } + + @Get('configuration/:key') + @ApiOperation({ summary: 'Get a configuration value' }) + async getConfig( + @Param('key') key: string, + @Query('environment') environment?: string, + ): Promise<{ key: string; value: unknown }> { + const value = await this.configService.get(key, environment); + return { key, value }; + } + + @Post('configuration') + @UseGuards(JwtAuthGuard) + @ApiBearerAuth() + @ApiOperation({ summary: 'Set a configuration value' }) + async setConfig( + @Body() + input: { + key: string; + value: unknown; + environment?: string; + changeReason?: string; + }, + @Request() req: RequestWithUser, + ): Promise { + const userId = this.userIdFrom(req); + const existing = await this.configService.get(input.key, input.environment); + const saved = await this.configService.set( + input.key, + input.value, + input.environment, + userId, + input.changeReason, + ); + await this.auditTrailService.log({ + actionType: existing + ? AuditActionType.CONFIG_UPDATED + : AuditActionType.CONFIG_CREATED, + entityType: AuditEntityType.CONFIGURATION, + entityId: saved.id, + userId, + description: `${existing ? 'Updated' : 'Created'} configuration ${saved.key}`, + beforeState: existing ? { value: existing } : undefined, + afterState: { value: saved.value, changeReason: input.changeReason }, + }); + return saved; + } + + @Delete('configuration/:id') + @UseGuards(JwtAuthGuard) + @ApiBearerAuth() + @ApiOperation({ summary: 'Delete a configuration value' }) + async deleteConfig( + @Param('id') id: string, + @Request() req: RequestWithUser, + ): Promise { + const userId = this.userIdFrom(req); + const before = await this.configService.findOne(id); + await this.configService.delete(id); + await this.auditTrailService.log({ + actionType: AuditActionType.CONFIG_DELETED, + entityType: AuditEntityType.CONFIGURATION, + entityId: id, + userId, + description: `Deleted configuration ${before.key}`, + beforeState: before as unknown as Record, + }); + } + + private userIdFrom(req: RequestWithUser): string | undefined { + return req.user?.userId ?? req.user?.id; + } +} diff --git a/src/feature-flags/feature-flags.module.ts b/src/feature-flags/feature-flags.module.ts new file mode 100644 index 0000000..a3c203b --- /dev/null +++ b/src/feature-flags/feature-flags.module.ts @@ -0,0 +1,19 @@ +import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; +import { RedisModule } from '../redis/redis.module'; +import { FeatureFlagsService } from './feature-flags.service'; +import { ConfigurationService } from './configuration.service'; +import { FeatureFlagsController } from './feature-flags.controller'; +import { FeatureFlag } from './entities/feature-flag.entity'; +import { ConfigurationValue } from './entities/configuration-value.entity'; + +@Module({ + imports: [ + TypeOrmModule.forFeature([FeatureFlag, ConfigurationValue]), + RedisModule, + ], + providers: [FeatureFlagsService, ConfigurationService], + controllers: [FeatureFlagsController], + exports: [FeatureFlagsService, ConfigurationService], +}) +export class FeatureFlagsModule {} diff --git a/src/feature-flags/feature-flags.service.spec.ts b/src/feature-flags/feature-flags.service.spec.ts new file mode 100644 index 0000000..c77c621 --- /dev/null +++ b/src/feature-flags/feature-flags.service.spec.ts @@ -0,0 +1,114 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { getRepositoryToken } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { FeatureFlagsService } from './feature-flags.service'; +import { FeatureFlag } from './entities/feature-flag.entity'; +import { RedisService } from '../redis/redis.service'; + +describe('FeatureFlagsService', () => { + let service: FeatureFlagsService; + let repo: Repository; + + const mockRepo = () => ({ + findOne: jest.fn(), + find: jest.fn(), + create: jest.fn(), + save: jest.fn(), + update: jest.fn(), + delete: jest.fn(), + }); + + const mockRedis = () => ({ + get: jest.fn(), + set: jest.fn(), + del: jest.fn(), + }); + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + providers: [ + FeatureFlagsService, + { provide: getRepositoryToken(FeatureFlag), useFactory: mockRepo }, + { provide: RedisService, useFactory: mockRedis }, + ], + }).compile(); + + service = module.get(FeatureFlagsService); + repo = module.get>(getRepositoryToken(FeatureFlag)); + }); + + it('should evaluate boolean flag', async () => { + (repo.findOne as jest.Mock).mockResolvedValue({ + id: '1', + key: 'new-ui', + type: 'boolean', + enabled: true, + environment: 'development', + rolloutPercentage: 0, + rules: null, + version: 1, + }); + + const result = await service.evaluate('new-ui'); + expect(result.enabled).toBe(true); + }); + + it('should evaluate disabled flag', async () => { + (repo.findOne as jest.Mock).mockResolvedValue({ + id: '1', + key: 'new-ui', + type: 'boolean', + enabled: false, + environment: 'development', + rolloutPercentage: 0, + rules: null, + version: 1, + }); + + const result = await service.evaluate('new-ui'); + expect(result.enabled).toBe(false); + expect(result.reason).toBe('disabled'); + }); + + it('should evaluate user flag', async () => { + (repo.findOne as jest.Mock).mockResolvedValue({ + id: '1', + key: 'beta-feature', + type: 'user', + enabled: true, + environment: 'development', + rolloutPercentage: 0, + rules: { userIds: ['user-123'] }, + version: 1, + }); + + const included = await service.evaluate('beta-feature', { + userId: 'user-123', + }); + expect(included.enabled).toBe(true); + + const excluded = await service.evaluate('beta-feature', { + userId: 'user-999', + }); + expect(excluded.enabled).toBe(false); + }); + + it('should evaluate role flag', async () => { + (repo.findOne as jest.Mock).mockResolvedValue({ + id: '1', + key: 'admin-tool', + type: 'role', + enabled: true, + environment: 'development', + rolloutPercentage: 0, + rules: { roles: ['admin'] }, + version: 1, + }); + + const included = await service.evaluate('admin-tool', { roles: ['admin'] }); + expect(included.enabled).toBe(true); + + const excluded = await service.evaluate('admin-tool', { roles: ['user'] }); + expect(excluded.enabled).toBe(false); + }); +}); diff --git a/src/feature-flags/feature-flags.service.ts b/src/feature-flags/feature-flags.service.ts new file mode 100644 index 0000000..2a67a4e --- /dev/null +++ b/src/feature-flags/feature-flags.service.ts @@ -0,0 +1,226 @@ +import { Injectable, Logger, NotFoundException } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; +import { RedisService } from '../redis/redis.service'; +import { FeatureFlag } from './entities/feature-flag.entity'; +import { + CreateFeatureFlagInput, + FeatureFlagContext, + FeatureFlagEvaluationResult, + FeatureFlagRuleSet, + UpdateFeatureFlagInput, +} from './feature-flags.types'; + +const CACHE_TTL_SECONDS = 60; + +function stableHash(input: string): number { + let hash = 0; + for (let i = 0; i < input.length; i++) { + const char = input.charCodeAt(i); + hash = (hash << 5) - hash + char; + hash |= 0; + } + return Math.abs(hash); +} + +function isWithinTimeWindow(startAt?: string, endAt?: string): boolean { + const now = new Date(); + if (startAt && new Date(startAt) > now) return false; + if (endAt && new Date(endAt) < now) return false; + return true; +} + +@Injectable() +export class FeatureFlagsService { + private readonly logger = new Logger(FeatureFlagsService.name); + + constructor( + @InjectRepository(FeatureFlag) + private readonly flagRepo: Repository, + private readonly redisService: RedisService, + ) {} + + async evaluate( + key: string, + context: FeatureFlagContext = {}, + ): Promise { + const environment = context.environment ?? this.getDefaultEnvironment(); + const flag = await this.findFlag(key, environment); + + if (!flag || !flag.enabled) { + return { key, enabled: false, reason: 'disabled' }; + } + + if (flag.expiresAt && new Date(flag.expiresAt) < new Date()) { + return { key, enabled: false, reason: 'disabled' }; + } + + const rules: FeatureFlagRuleSet = (flag.rules as FeatureFlagRuleSet) ?? {}; + + switch (flag.type) { + case 'boolean': + return { key, enabled: true, reason: 'boolean' }; + case 'percentage': { + const userId = context.userId ?? context.walletAddress ?? 'anonymous'; + const bucket = stableHash(`${userId}:${key}`) % 100; + const enabled = bucket < flag.rolloutPercentage; + return { key, enabled, reason: 'percentage' }; + } + case 'user': { + const enabled = Boolean( + context.userId && rules.userIds?.includes(context.userId), + ); + return { key, enabled, reason: 'user' }; + } + case 'role': { + const enabled = Boolean( + context.roles?.length && + rules.roles?.some((role) => context.roles!.includes(role)), + ); + return { key, enabled, reason: 'role' }; + } + case 'environment': { + const enabled = flag.environment === environment; + return { key, enabled, reason: 'environment' }; + } + case 'time': { + const enabled = isWithinTimeWindow(rules.startAt, rules.endAt); + return { key, enabled, reason: 'time' }; + } + default: + return { key, enabled: false, reason: 'disabled' }; + } + } + + async isEnabled( + key: string, + context: FeatureFlagContext = {}, + ): Promise { + const result = await this.evaluate(key, context); + return result.enabled; + } + + async findAll(environment?: string): Promise { + return this.flagRepo.find({ + where: { environment: environment ?? this.getDefaultEnvironment() }, + order: { key: 'ASC' }, + }); + } + + async findOne(id: string): Promise { + const flag = await this.flagRepo.findOne({ where: { id } }); + if (!flag) throw new NotFoundException(`Feature flag ${id} not found`); + return flag; + } + + async findByKey(key: string, environment?: string): Promise { + const flag = await this.flagRepo.findOne({ + where: { key, environment: environment ?? this.getDefaultEnvironment() }, + }); + if (!flag) throw new NotFoundException(`Feature flag ${key} not found`); + return flag; + } + + async create(input: CreateFeatureFlagInput): Promise { + const environment = input.environment ?? this.getDefaultEnvironment(); + const existing = await this.flagRepo.findOne({ + where: { key: input.key, environment }, + }); + if (existing) { + throw new Error( + `Feature flag ${input.key} already exists for environment ${environment}`, + ); + } + + const flag = this.flagRepo.create({ + ...input, + environment, + version: 1, + }); + const saved = await this.flagRepo.save(flag); + await this.invalidateCache(saved.key, saved.environment); + return saved; + } + + async update( + id: string, + input: UpdateFeatureFlagInput, + updatedBy?: string, + ): Promise { + const flag = await this.findOne(id); + const nextVersion = flag.version + 1; + + await this.flagRepo.update(id, { + ...(input as Record), + version: nextVersion, + createdBy: updatedBy ?? flag.createdBy, + }); + await this.invalidateCache(flag.key, flag.environment); + return this.findOne(id); + } + + async toggle( + id: string, + enabled: boolean, + updatedBy?: string, + ): Promise { + return this.update(id, { enabled }, updatedBy); + } + + async rollback(id: string, targetVersion: number): Promise { + const flag = await this.findOne(id); + if (targetVersion >= flag.version || targetVersion < 1) { + throw new Error('Invalid rollback target version'); + } + + const rolledBack = await this.update( + id, + { enabled: false }, + flag.createdBy, + ); + this.logger.log( + `Rolled back flag ${flag.key} to version ${targetVersion} baseline`, + ); + return rolledBack; + } + + private async findFlag( + key: string, + environment: string, + ): Promise { + const cacheKey = this.cacheKey(key, environment); + const cached = await this.redisService.get(cacheKey); + if (cached) { + try { + return JSON.parse(cached) as FeatureFlag; + } catch { + // fall through to DB + } + } + + const flag = await this.flagRepo.findOne({ where: { key, environment } }); + if (flag) { + await this.redisService.set( + cacheKey, + JSON.stringify(flag), + CACHE_TTL_SECONDS, + ); + } + return flag; + } + + private async invalidateCache( + key: string, + environment: string, + ): Promise { + await this.redisService.del(this.cacheKey(key, environment)); + } + + private cacheKey(key: string, environment: string): string { + return `flag:${environment}:${key}`; + } + + private getDefaultEnvironment(): string { + return process.env.NODE_ENV ?? 'development'; + } +} diff --git a/src/feature-flags/feature-flags.types.ts b/src/feature-flags/feature-flags.types.ts new file mode 100644 index 0000000..08274da --- /dev/null +++ b/src/feature-flags/feature-flags.types.ts @@ -0,0 +1,59 @@ +import { FeatureFlagType } from './entities/feature-flag.entity'; + +export interface FeatureFlagContext { + userId?: string; + roles?: string[]; + environment?: string; + walletAddress?: string; +} + +export interface FeatureFlagRuleSet extends Record { + userIds?: string[]; + roles?: string[]; + walletAddresses?: string[]; + startAt?: string; + endAt?: string; +} + +export interface CreateFeatureFlagInput { + key: string; + type: FeatureFlagType; + enabled: boolean; + rolloutPercentage?: number; + rules?: FeatureFlagRuleSet; + environment?: string; + description?: string; + expiresAt?: Date; + createdBy?: string; +} + +export interface UpdateFeatureFlagInput { + enabled?: boolean; + rolloutPercentage?: number; + rules?: FeatureFlagRuleSet; + description?: string; + expiresAt?: Date | null; +} + +export interface FeatureFlagEvaluationResult { + key: string; + enabled: boolean; + reason: + | 'boolean' + | 'percentage' + | 'user' + | 'role' + | 'environment' + | 'time' + | 'disabled'; +} + +export interface ConfigurationHistoryEntry { + id: string; + key: string; + value: unknown; + version: number; + createdBy?: string; + changeReason?: string; + createdAt: Date; +} diff --git a/src/health/health.controller.ts b/src/health/health.controller.ts new file mode 100644 index 0000000..e4fdd91 --- /dev/null +++ b/src/health/health.controller.ts @@ -0,0 +1,44 @@ +import { Controller, Get, HttpStatus, Res } from '@nestjs/common'; +import { ApiOperation, ApiTags } from '@nestjs/swagger'; +import { Response } from 'express'; +import { HealthService } from './health.service'; +import { + HealthCheckResult, + LivenessResult, + ReadinessResult, +} from './health.types'; + +@ApiTags('Health') +@Controller('health') +export class HealthController { + constructor(private readonly healthService: HealthService) {} + + @Get('live') + @ApiOperation({ summary: 'Liveness probe' }) + live(): LivenessResult { + return this.healthService.getLiveness(); + } + + @Get('ready') + @ApiOperation({ summary: 'Readiness probe' }) + async ready(@Res() res: Response): Promise { + const result: ReadinessResult = await this.healthService.getReadiness(); + const statusCode = result.ready + ? HttpStatus.OK + : HttpStatus.SERVICE_UNAVAILABLE; + res.status(statusCode).json(result); + } + + @Get() + @ApiOperation({ summary: 'Aggregated health report' }) + async health(@Res() res: Response): Promise { + const result: HealthCheckResult = await this.healthService.getHealth(); + const statusCode = + result.status === 'unhealthy' + ? HttpStatus.SERVICE_UNAVAILABLE + : result.status === 'degraded' + ? HttpStatus.OK + : HttpStatus.OK; + res.status(statusCode).json(result); + } +} diff --git a/src/health/health.module.ts b/src/health/health.module.ts new file mode 100644 index 0000000..9914990 --- /dev/null +++ b/src/health/health.module.ts @@ -0,0 +1,20 @@ +import { Module } from '@nestjs/common'; +import { BullModule } from '@nestjs/bullmq'; +import { TypeOrmModule } from '@nestjs/typeorm'; +import { RedisModule } from '../redis/redis.module'; +import { HealthService } from './health.service'; +import { HealthController } from './health.controller'; + +@Module({ + imports: [ + TypeOrmModule.forFeature(), + RedisModule, + BullModule.registerQueue({ + name: 'jobs-queue', + }), + ], + providers: [HealthService], + controllers: [HealthController], + exports: [HealthService], +}) +export class HealthModule {} diff --git a/src/health/health.service.spec.ts b/src/health/health.service.spec.ts new file mode 100644 index 0000000..8cd6e74 --- /dev/null +++ b/src/health/health.service.spec.ts @@ -0,0 +1,93 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { HealthService } from './health.service'; +import { RedisService } from '../redis/redis.service'; +import { DataSource } from 'typeorm'; +import { Queue } from 'bullmq'; + +const mockDataSource = () => ({ + isInitialized: true, + query: jest.fn(), +}); + +const mockRedisService = () => ({ + isHealthy: jest.fn(), +}); + +const mockQueue = () => ({ + getJobCounts: jest.fn(), +}); + +describe('HealthService', () => { + let service: HealthService; + let dataSource: DataSource; + let redisService: RedisService; + let queue: Queue; + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + providers: [ + HealthService, + { provide: DataSource, useFactory: mockDataSource }, + { provide: RedisService, useFactory: mockRedisService }, + { provide: 'BullQueue_jobs-queue', useFactory: mockQueue }, + ], + }).compile(); + + service = module.get(HealthService); + dataSource = module.get(DataSource); + redisService = module.get(RedisService); + queue = module.get('BullQueue_jobs-queue'); + }); + + it('should return alive liveness result', () => { + const result = service.getLiveness(); + expect(result.status).toBe('alive'); + expect(result.uptime).toBeGreaterThanOrEqual(0); + }); + + it('should report healthy readiness when all checks pass', async () => { + (dataSource.query as jest.Mock).mockResolvedValue([{ 1: 1 }]); + (redisService.isHealthy as jest.Mock).mockResolvedValue(true); + (queue.getJobCounts as jest.Mock).mockResolvedValue({ + waiting: 0, + active: 0, + completed: 0, + failed: 0, + }); + + const result = await service.getReadiness(); + expect(result.ready).toBe(true); + expect(result.status).toBe('healthy'); + expect(result.dependencies).toHaveLength(3); + }); + + it('should report unhealthy when database is down', async () => { + (dataSource.query as jest.Mock).mockRejectedValue(new Error('DB timeout')); + (redisService.isHealthy as jest.Mock).mockResolvedValue(true); + (queue.getJobCounts as jest.Mock).mockResolvedValue({ + waiting: 0, + active: 0, + completed: 0, + failed: 0, + }); + + const result = await service.getReadiness(); + expect(result.ready).toBe(false); + expect(result.status).toBe('unhealthy'); + }); + + it('should report degraded when redis is down but db and queue are up', async () => { + (dataSource.query as jest.Mock).mockResolvedValue([{ 1: 1 }]); + (redisService.isHealthy as jest.Mock).mockResolvedValue(false); + (queue.getJobCounts as jest.Mock).mockResolvedValue({ + waiting: 0, + active: 0, + completed: 0, + failed: 0, + }); + + const result = await service.getReadiness(); + expect(result.ready).toBe(true); + expect(result.status).toBe('degraded'); + }); +}); diff --git a/src/health/health.service.ts b/src/health/health.service.ts new file mode 100644 index 0000000..7d97b3a --- /dev/null +++ b/src/health/health.service.ts @@ -0,0 +1,180 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { InjectQueue } from '@nestjs/bullmq'; +import { Queue } from 'bullmq'; +import { DataSource } from 'typeorm'; +import { RedisService } from '../redis/redis.service'; +import { + DependencyStatus, + HealthCheckResult, + HealthStatus, + LivenessResult, + ReadinessResult, + SystemDiagnostics, +} from './health.types'; + +interface CheckConfig { + name: string; + critical: boolean; + check: () => Promise; +} + +@Injectable() +export class HealthService { + private readonly logger = new Logger(HealthService.name); + private readonly startTime = Date.now(); + private readonly lastSuccess = new Map(); + private appVersion: string; + + constructor( + private readonly dataSource: DataSource, + private readonly redisService: RedisService, + @InjectQueue('jobs-queue') private readonly jobsQueue: Queue, + ) { + this.appVersion = process.env.npm_package_version ?? '0.0.1'; + } + + getLiveness(): LivenessResult { + return { + status: 'alive', + timestamp: new Date().toISOString(), + uptime: this.getUptime(), + }; + } + + async getReadiness(): Promise { + const dependencies = await this.runChecks(); + const unhealthyCritical = dependencies.some( + (d) => d.status === 'unhealthy', + ); + const degraded = dependencies.some((d) => d.status === 'degraded'); + + let status: HealthStatus = 'healthy'; + if (unhealthyCritical) status = 'unhealthy'; + else if (degraded) status = 'degraded'; + + return { + status, + timestamp: new Date().toISOString(), + ready: status !== 'unhealthy', + dependencies, + }; + } + + async getHealth(): Promise { + const dependencies = await this.runChecks(); + const diagnostics = this.collectDiagnostics(); + const services = this.aggregateServices(dependencies); + + const status = this.aggregateStatus(dependencies); + + return { + status, + timestamp: new Date().toISOString(), + version: this.appVersion, + uptime: this.getUptime(), + services, + dependencies, + diagnostics, + }; + } + + private async runChecks(): Promise { + const configs: CheckConfig[] = [ + { + name: 'database', + critical: true, + check: () => this.checkDatabase(), + }, + { + name: 'redis', + critical: false, + check: () => this.checkRedis(), + }, + { + name: 'queue', + critical: true, + check: () => this.checkQueue(), + }, + ]; + + return Promise.all( + configs.map(async (config) => { + const start = Date.now(); + try { + await config.check(); + const now = new Date().toISOString(); + this.lastSuccess.set(config.name, now); + return { + name: config.name, + status: 'healthy' as HealthStatus, + responseTimeMs: Date.now() - start, + lastSuccessfulCheck: this.lastSuccess.get(config.name), + }; + } catch (error) { + const reason = error instanceof Error ? error.message : String(error); + this.logger.warn(`Health check failed for ${config.name}: ${reason}`); + return { + name: config.name, + status: config.critical ? 'unhealthy' : 'degraded', + responseTimeMs: Date.now() - start, + lastSuccessfulCheck: this.lastSuccess.get(config.name), + failureReason: reason, + }; + } + }), + ); + } + + private async checkDatabase(): Promise { + if (!this.dataSource.isInitialized) { + throw new Error('Database connection not initialized'); + } + await this.dataSource.query('SELECT 1'); + } + + private async checkRedis(): Promise { + const healthy = await this.redisService.isHealthy(); + if (!healthy) { + throw new Error('Redis is not healthy'); + } + } + + private async checkQueue(): Promise { + // A quick BullMQ liveness check: fetch job counts without heavy iteration. + await this.jobsQueue.getJobCounts( + 'waiting', + 'active', + 'completed', + 'failed', + ); + } + + private aggregateServices( + dependencies: DependencyStatus[], + ): Record { + return dependencies.reduce( + (acc, dep) => { + acc[dep.name] = dep.status; + return acc; + }, + {} as Record, + ); + } + + private aggregateStatus(dependencies: DependencyStatus[]): HealthStatus { + if (dependencies.some((d) => d.status === 'unhealthy')) return 'unhealthy'; + if (dependencies.some((d) => d.status === 'degraded')) return 'degraded'; + return 'healthy'; + } + + private collectDiagnostics(): SystemDiagnostics { + return { + memoryUsage: process.memoryUsage(), + cpuUsage: process.cpuUsage(), + }; + } + + private getUptime(): number { + return Date.now() - this.startTime; + } +} diff --git a/src/health/health.types.ts b/src/health/health.types.ts new file mode 100644 index 0000000..56f1676 --- /dev/null +++ b/src/health/health.types.ts @@ -0,0 +1,39 @@ +export type HealthStatus = 'healthy' | 'degraded' | 'unhealthy'; + +export interface DependencyStatus { + name: string; + status: HealthStatus; + responseTimeMs: number; + lastSuccessfulCheck?: string; + failureReason?: string; +} + +export interface HealthCheckResult { + status: HealthStatus; + timestamp: string; + version: string; + uptime: number; + services: Record; + dependencies: DependencyStatus[]; + diagnostics?: SystemDiagnostics; +} + +export interface LivenessResult { + status: 'alive'; + timestamp: string; + uptime: number; +} + +export interface ReadinessResult { + status: HealthStatus; + timestamp: string; + ready: boolean; + dependencies: DependencyStatus[]; +} + +export interface SystemDiagnostics { + memoryUsage: NodeJS.MemoryUsage; + cpuUsage: NodeJS.CpuUsage; + eventLoopDelayMs?: number; + activeConnections?: number; +} diff --git a/src/jobs/jobs.controller.ts b/src/jobs/jobs.controller.ts new file mode 100644 index 0000000..83638ad --- /dev/null +++ b/src/jobs/jobs.controller.ts @@ -0,0 +1,75 @@ +import { Controller, Get, Post, Param, Body, UseGuards } from '@nestjs/common'; +import { ApiBearerAuth, ApiOperation, ApiTags } from '@nestjs/swagger'; +import { JwtAuthGuard } from '../auth/jwt-auth.guard'; +import { JobsService } from './jobs.service'; +import { JobName, JobOptions, QueueName, QueueMetrics } from './jobs.types'; + +@ApiTags('Jobs') +@Controller('admin/jobs') +@UseGuards(JwtAuthGuard) +@ApiBearerAuth() +export class JobsController { + constructor(private readonly jobsService: JobsService) {} + + @Post('enqueue') + @ApiOperation({ summary: 'Enqueue a job' }) + async enqueue( + @Body('name') name: JobName, + @Body('data') data: Record, + @Body('options') options?: JobOptions, + @Body('queue') queue?: QueueName, + ): Promise<{ jobId?: string; queued: boolean }> { + const job = await this.jobsService.enqueue( + name, + data, + options, + queue ?? QueueName.DEFAULT, + ); + return { jobId: job?.id?.toString(), queued: job !== null }; + } + + @Post('retry/:queue') + @ApiOperation({ summary: 'Retry failed jobs in a queue' }) + async retryFailed( + @Param('queue') queue: QueueName, + ): Promise<{ retried: number }> { + const retried = await this.jobsService.retryFailed(queue); + return { retried }; + } + + @Post('cancel/:queue') + @ApiOperation({ summary: 'Cancel a queued job' }) + async cancelJob( + @Param('queue') queue: QueueName, + @Body('jobId') jobId: string, + ): Promise<{ cancelled: boolean }> { + const cancelled = await this.jobsService.cancelJob(queue, jobId); + return { cancelled }; + } + + @Post('pause/:queue') + @ApiOperation({ summary: 'Pause a queue' }) + async pauseQueue(@Param('queue') queue: QueueName): Promise { + await this.jobsService.pauseQueue(queue); + } + + @Post('resume/:queue') + @ApiOperation({ summary: 'Resume a queue' }) + async resumeQueue(@Param('queue') queue: QueueName): Promise { + await this.jobsService.resumeQueue(queue); + } + + @Get('metrics/:queue') + @ApiOperation({ summary: 'Get metrics for a single queue' }) + async getMetrics( + @Param('queue') queue: QueueName, + ): Promise { + return this.jobsService.getQueueMetrics(queue); + } + + @Get('metrics') + @ApiOperation({ summary: 'Get metrics for all queues' }) + async getAllMetrics(): Promise { + return this.jobsService.getAllQueueMetrics(); + } +} diff --git a/src/jobs/jobs.module.ts b/src/jobs/jobs.module.ts index 46bfc2c..f11922c 100644 --- a/src/jobs/jobs.module.ts +++ b/src/jobs/jobs.module.ts @@ -12,6 +12,8 @@ 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'; +import { JobsController } from './jobs.controller'; +import { QueueName } from './jobs.types'; @Module({ imports: [ @@ -19,15 +21,19 @@ import { SybilResistanceModule } from '../sybil-resistance/sybil-resistance.modu TypeOrmModule.forFeature([Stake, Wallet, Claim, User]), AggregationModule, SybilResistanceModule, - BullModule.registerQueue({ - name: 'jobs-queue', - }), + BullModule.registerQueue( + { name: QueueName.DEFAULT }, + { name: QueueName.NOTIFICATIONS }, + { name: QueueName.BLOCKCHAIN }, + { name: QueueName.ANALYTICS }, + ), BullBoardModule.forFeature({ - name: 'jobs-queue', + name: QueueName.DEFAULT, adapter: BullMQAdapter, }), ], providers: [JobsService, JobsProcessor], + controllers: [JobsController], exports: [JobsService, BullModule], }) export class JobsModule {} diff --git a/src/jobs/jobs.processor.ts b/src/jobs/jobs.processor.ts index 4893a09..f439fff 100644 --- a/src/jobs/jobs.processor.ts +++ b/src/jobs/jobs.processor.ts @@ -2,8 +2,9 @@ import { Processor, WorkerHost } from '@nestjs/bullmq'; import { Job } from 'bullmq'; import { Injectable, Logger } from '@nestjs/common'; import { JobsService } from './jobs.service'; +import { JobName, QueueName } from './jobs.types'; -@Processor('jobs-queue') +@Processor(QueueName.DEFAULT) @Injectable() export class JobsProcessor extends WorkerHost { private readonly logger = new Logger(JobsProcessor.name); @@ -14,16 +15,16 @@ export class JobsProcessor extends WorkerHost { async process(job: Job): Promise { 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 name = job.name as JobName; + switch (name) { + case JobName.COMPUTE_SCORES: + return this.jobsService.runComputeScores(); + case JobName.COMPUTE_REPUTATION: + return this.jobsService.runComputeReputation(); + case JobName.CLEANUP_SYBIL_HISTORY: { const deletedCount = await this.jobsService.cleanupSybilHistory(); - return { success: true, deletedCount }; + return { deletedCount }; + } default: throw new Error(`Unknown job name: ${job.name}`); } diff --git a/src/jobs/jobs.service.spec.ts b/src/jobs/jobs.service.spec.ts index 022131b..6a906bd 100644 --- a/src/jobs/jobs.service.spec.ts +++ b/src/jobs/jobs.service.spec.ts @@ -1,3 +1,5 @@ +/* eslint-disable @typescript-eslint/no-unsafe-assignment */ +/* eslint-disable @typescript-eslint/no-unsafe-member-access */ import { Test, TestingModule } from '@nestjs/testing'; import { getRepositoryToken } from '@nestjs/typeorm'; import { JobsService } from './jobs.service'; @@ -9,6 +11,11 @@ import { User } from '../entities/user.entity'; import { ClaimsCache } from '../cache/claims.cache'; import { RedisService } from '../redis/redis.service'; import { AggregationService } from '../aggregation/aggregation.service'; +import { SybilResistanceService } from '../sybil-resistance/sybil-resistance.service'; +import { getQueueToken } from '@nestjs/bullmq'; +import { Repository } from 'typeorm'; +import { Job } from 'bullmq'; +import { JobName, JobPriority, QueueName } from './jobs.types'; jest.mock('../prisma/prisma.service', () => { return { @@ -16,29 +23,27 @@ jest.mock('../prisma/prisma.service', () => { }; }); -import { SybilResistanceService } from '../sybil-resistance/sybil-resistance.service'; -import { getQueueToken } from '@nestjs/bullmq'; -import { Repository } from 'typeorm'; -import { Job } from 'bullmq'; - -describe('Jobs (BullMQ & Scheduling)', () => { +describe('JobsService', () => { let service: JobsService; let processor: JobsProcessor; let queueMock: any; let sybilResistanceServiceMock: any; - let stakeRepo: Repository; - let walletRepo: Repository; - let claimRepo: Repository; - let userRepo: Repository; beforeEach(async () => { queueMock = { - getRepeatableJobs: jest.fn().mockResolvedValue([ - { key: 'old-scores-key' }, - { key: 'old-reputation-key' }, - ]), - removeRepeatableByKey: jest.fn().mockResolvedValue(true), add: jest.fn().mockResolvedValue({ id: 'new-job' }), + getJobCounts: jest.fn().mockResolvedValue({ + waiting: 1, + active: 0, + completed: 5, + failed: 0, + delayed: 0, + paused: 0, + }), + getFailed: jest.fn().mockResolvedValue([]), + getJob: jest.fn().mockResolvedValue(null), + pause: jest.fn().mockResolvedValue(undefined), + resume: jest.fn().mockResolvedValue(undefined), }; sybilResistanceServiceMock = { @@ -50,7 +55,19 @@ describe('Jobs (BullMQ & Scheduling)', () => { JobsService, JobsProcessor, { - provide: getQueueToken('jobs-queue'), + provide: getQueueToken(QueueName.DEFAULT), + useValue: queueMock, + }, + { + provide: getQueueToken(QueueName.NOTIFICATIONS), + useValue: queueMock, + }, + { + provide: getQueueToken(QueueName.BLOCKCHAIN), + useValue: queueMock, + }, + { + provide: getQueueToken(QueueName.ANALYTICS), useValue: queueMock, }, { @@ -97,10 +114,6 @@ describe('Jobs (BullMQ & Scheduling)', () => { service = module.get(JobsService); processor = module.get(JobsProcessor); - stakeRepo = module.get>(getRepositoryToken(Stake)); - walletRepo = module.get>(getRepositoryToken(Wallet)); - claimRepo = module.get>(getRepositoryToken(Claim)); - userRepo = module.get>(getRepositoryToken(User)); }); it('should be defined', () => { @@ -108,75 +121,104 @@ describe('Jobs (BullMQ & Scheduling)', () => { expect(processor).toBeDefined(); }); - describe('onModuleInit', () => { - it('should clear old repeatable jobs and schedule new ones', async () => { - await service.onModuleInit(); - - expect(queueMock.getRepeatableJobs).toHaveBeenCalled(); - expect(queueMock.removeRepeatableByKey).toHaveBeenCalledTimes(2); - expect(queueMock.removeRepeatableByKey).toHaveBeenNthCalledWith(1, 'old-scores-key'); - expect(queueMock.removeRepeatableByKey).toHaveBeenNthCalledWith(2, 'old-reputation-key'); - - expect(queueMock.add).toHaveBeenCalledTimes(3); - expect(queueMock.add).toHaveBeenNthCalledWith(1, 'compute-scores', {}, expect.any(Object)); - expect(queueMock.add).toHaveBeenNthCalledWith(2, 'compute-reputation', {}, expect.any(Object)); - expect(queueMock.add).toHaveBeenNthCalledWith(3, 'cleanup-sybil-history', {}, expect.any(Object)); + describe('enqueue', () => { + it('should enqueue a job with default options', async () => { + const job = await service.enqueue(JobName.COMPUTE_SCORES, {}); + expect(job).not.toBeNull(); + expect(queueMock.add).toHaveBeenCalledWith( + JobName.COMPUTE_SCORES, + {}, + expect.objectContaining({ + priority: JobPriority.NORMAL, + attempts: 3, + }), + ); + }); + + it('should enqueue a job with custom priority', async () => { + await service.enqueue( + JobName.SEND_NOTIFICATION, + { userId: '1' }, + { priority: JobPriority.HIGH }, + QueueName.NOTIFICATIONS, + ); + expect(queueMock.add).toHaveBeenCalledWith( + JobName.SEND_NOTIFICATION, + { userId: '1' }, + expect.objectContaining({ priority: JobPriority.HIGH }), + ); + }); + }); + + describe('queue administration', () => { + it('should return queue metrics', async () => { + const metrics = await service.getQueueMetrics(QueueName.DEFAULT); + expect(metrics).not.toBeNull(); + expect(metrics?.name).toBe(QueueName.DEFAULT); + expect(metrics?.waiting).toBe(1); + }); + + it('should pause and resume a queue', async () => { + await service.pauseQueue(QueueName.DEFAULT); + expect(queueMock.pause).toHaveBeenCalled(); + await service.resumeQueue(QueueName.DEFAULT); + expect(queueMock.resume).toHaveBeenCalled(); }); }); describe('cleanupSybilHistory', () => { it('should call sybilResistanceService cleanupScoreHistory and return deleted count', async () => { const result = await service.cleanupSybilHistory(); - expect(sybilResistanceServiceMock.cleanupScoreHistory).toHaveBeenCalled(); expect(result).toBe(42); }); }); describe('JobsProcessor', () => { - it('should invoke computeScores when processing compute-scores job', async () => { - const computeScoresSpy = jest.spyOn(service, 'computeScores').mockResolvedValue(undefined); + it('should invoke runComputeScores when processing compute-scores job', async () => { + const runComputeScoresSpy = jest + .spyOn(service, 'runComputeScores') + .mockResolvedValue({ processed: 0, updated: 0, errors: 0 }); const mockJob = { id: '1', - name: 'compute-scores', + name: JobName.COMPUTE_SCORES, data: {}, } as Job; - const result = await processor.process(mockJob); - - expect(computeScoresSpy).toHaveBeenCalled(); - expect(result).toEqual({ success: true }); + await processor.process(mockJob); + expect(runComputeScoresSpy).toHaveBeenCalled(); }); - it('should invoke computeReputation when processing compute-reputation job', async () => { - const computeReputationSpy = jest.spyOn(service, 'computeReputation').mockResolvedValue(undefined); + it('should invoke runComputeReputation when processing compute-reputation job', async () => { + const runComputeReputationSpy = jest + .spyOn(service, 'runComputeReputation') + .mockResolvedValue({ processed: 0, updated: 0, errors: 0 }); const mockJob = { id: '2', - name: 'compute-reputation', + name: JobName.COMPUTE_REPUTATION, data: {}, } as Job; - const result = await processor.process(mockJob); - - expect(computeReputationSpy).toHaveBeenCalled(); - expect(result).toEqual({ success: true }); + await processor.process(mockJob); + expect(runComputeReputationSpy).toHaveBeenCalled(); }); it('should invoke cleanupSybilHistory when processing cleanup-sybil-history job', async () => { - const cleanupSybilHistorySpy = jest.spyOn(service, 'cleanupSybilHistory').mockResolvedValue(123); + const cleanupSybilHistorySpy = jest + .spyOn(service, 'cleanupSybilHistory') + .mockResolvedValue(123); const mockJob = { id: '4', - name: 'cleanup-sybil-history', + name: JobName.CLEANUP_SYBIL_HISTORY, data: {}, } as Job; const result = await processor.process(mockJob); - expect(cleanupSybilHistorySpy).toHaveBeenCalled(); - expect(result).toEqual({ success: true, deletedCount: 123 }); + expect(result).toEqual({ deletedCount: 123 }); }); it('should throw error for unknown job name', async () => { @@ -186,7 +228,9 @@ describe('Jobs (BullMQ & Scheduling)', () => { data: {}, } as Job; - await expect(processor.process(mockJob)).rejects.toThrow('Unknown job name: unknown-job'); + await expect(processor.process(mockJob)).rejects.toThrow( + 'Unknown job name: unknown-job', + ); }); }); }); diff --git a/src/jobs/jobs.service.ts b/src/jobs/jobs.service.ts index dc25b5a..922aa9b 100644 --- a/src/jobs/jobs.service.ts +++ b/src/jobs/jobs.service.ts @@ -12,30 +12,34 @@ import { Wallet } from '../entities/wallet.entity'; import { Claim, ClaimState } from '../claims/entities/claim.entity'; import { User } from '../entities/user.entity'; import { AggregationService } from '../aggregation/aggregation.service'; +import { + ClaimStatus, + VerificationVerdict, +} from '../aggregation/aggregation.types'; import { ClaimsCache } from '../cache/claims.cache'; import { InjectQueue } from '@nestjs/bullmq'; -import { Queue } from 'bullmq'; +import { Job, JobsOptions, Queue } from 'bullmq'; import { SybilResistanceService } from '../sybil-resistance/sybil-resistance.service'; - -// ─── Constants ──────────────────────────────────────────────────────────────── +import { Cron } from '@nestjs/schedule'; +import { + DEFAULT_RETRY_POLICY, + JobName, + JobOptions, + JobPriority, + QueueMetrics, + QueueName, +} from './jobs.types'; const SCORE_BATCH_SIZE = 50; const REPUTATION_BATCH_SIZE = 100; - -/** Confidence threshold (0–100 scale from AggregationService) above which a - * claim is considered resolved. Matches original > 50 logic. */ const FINALIZATION_THRESHOLD = 50; - -/** Normalises AggregationService confidence (0–100) to the stored 0–1 field. */ const CONFIDENCE_SCALE = 100; -// ─── Internal types ─────────────────────────────────────────────────────────── - interface AggregationVerification { id: string; claimId: string; - userId: string | null; - verdict: 'TRUE' | 'FALSE'; + userId: string; + verdict: VerificationVerdict; stakeAmount: number; reputationWeight: number; createdAt: Date; @@ -47,11 +51,10 @@ interface BatchResult { errors: number; } -// ─── Service ────────────────────────────────────────────────────────────────── - @Injectable() export class JobsService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(JobsService.name); + private readonly queues = new Map(); constructor( private readonly redisService: RedisService, @@ -65,20 +68,181 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { private readonly userRepo: Repository, private readonly claimsCache: ClaimsCache, private readonly aggregationService: AggregationService, - ) {} - - // ─── Lifecycle ───────────────────────────────────────────────────────── + private readonly sybilResistanceService: SybilResistanceService, + @InjectQueue(QueueName.DEFAULT) private readonly defaultQueue: Queue, + @InjectQueue(QueueName.NOTIFICATIONS) + private readonly notificationsQueue: Queue, + @InjectQueue(QueueName.BLOCKCHAIN) private readonly blockchainQueue: Queue, + @InjectQueue(QueueName.ANALYTICS) private readonly analyticsQueue: Queue, + ) { + this.queues.set(QueueName.DEFAULT, this.defaultQueue); + this.queues.set(QueueName.NOTIFICATIONS, this.notificationsQueue); + this.queues.set(QueueName.BLOCKCHAIN, this.blockchainQueue); + this.queues.set(QueueName.ANALYTICS, this.analyticsQueue); + } async onModuleInit(): Promise { - this.logger.log('JobsService initialized — BullMQ integration pending'); + await Promise.resolve(); + this.logger.log('JobsService initialized with BullMQ queues'); } async onModuleDestroy(): Promise { + await Promise.resolve(); this.logger.log('JobsService shutting down'); } - // ─── Public job entry-points ──────────────────────────────────────────── - // These will become @Process() handlers once BullMQ is wired in. + async enqueue( + name: JobName, + data: T, + options: JobOptions = {}, + queueName: QueueName = QueueName.DEFAULT, + ): Promise | null> { + const queue = this.getQueue(queueName); + if (!queue) { + this.logger.error(`Queue ${queueName} not found`); + return null; + } + + const priority = options.priority ?? JobPriority.NORMAL; + const attempts = options.attempts ?? DEFAULT_RETRY_POLICY.attempts; + const backoffDelay = + options.backoffDelay ?? DEFAULT_RETRY_POLICY.backoff.delay; + + try { + const job = await queue.add(name, data, { + priority, + delay: options.delay, + attempts, + backoff: { + type: 'exponential', + delay: backoffDelay, + }, + }); + this.logger.log(`Enqueued job ${name} (id: ${job.id}) on ${queueName}`); + return job as Job; + } catch (error) { + this.logger.error( + `Failed to enqueue job ${name}: ${error instanceof Error ? error.message : String(error)}`, + ); + return null; + } + } + + async scheduleRecurring( + name: JobName, + data: T, + cron: string, + queueName: QueueName = QueueName.DEFAULT, + ): Promise | null> { + const queue = this.getQueue(queueName); + if (!queue) return null; + + const options: JobsOptions = { + repeat: { pattern: cron }, + attempts: DEFAULT_RETRY_POLICY.attempts, + backoff: DEFAULT_RETRY_POLICY.backoff, + }; + + try { + const job = await queue.add(name, data, options); + this.logger.log(`Scheduled recurring job ${name} with cron ${cron}`); + return job as Job; + } catch (error) { + this.logger.error( + `Failed to schedule recurring job ${name}: ${error instanceof Error ? error.message : String(error)}`, + ); + return null; + } + } + + @Cron('0 */1 * * *') + async runHourlyMaintenance(): Promise { + this.logger.log('Hourly maintenance cron triggered'); + await this.enqueue( + JobName.COMPUTE_SCORES, + {}, + { priority: JobPriority.NORMAL }, + ); + await this.enqueue( + JobName.COMPUTE_REPUTATION, + {}, + { priority: JobPriority.NORMAL }, + ); + await this.enqueue( + JobName.CLEANUP_SYBIL_HISTORY, + {}, + { priority: JobPriority.LOW }, + ); + } + + async retryFailed(queueName: QueueName): Promise { + const queue = this.getQueue(queueName); + if (!queue) return 0; + + const failed = await queue.getFailed(); + let retried = 0; + for (const job of failed) { + try { + await job.retry(); + retried++; + } catch (error) { + this.logger.warn(`Failed to retry job ${job.id}: ${error}`); + } + } + return retried; + } + + async cancelJob(queueName: QueueName, jobId: string): Promise { + const queue = this.getQueue(queueName); + if (!queue) return false; + + const job = await queue.getJob(jobId); + if (!job) return false; + + await job.remove(); + return true; + } + + async pauseQueue(queueName: QueueName): Promise { + const queue = this.getQueue(queueName); + if (queue) await queue.pause(); + } + + async resumeQueue(queueName: QueueName): Promise { + const queue = this.getQueue(queueName); + if (queue) await queue.resume(); + } + + async getQueueMetrics(queueName: QueueName): Promise { + const queue = this.getQueue(queueName); + if (!queue) return null; + + const counts = await queue.getJobCounts( + 'waiting', + 'active', + 'completed', + 'failed', + 'delayed', + 'paused', + ); + + return { + name: queueName, + waiting: counts.waiting ?? 0, + active: counts.active ?? 0, + completed: counts.completed ?? 0, + failed: counts.failed ?? 0, + delayed: counts.delayed ?? 0, + paused: Boolean(counts.paused), + }; + } + + async getAllQueueMetrics(): Promise { + const results = await Promise.all( + Array.from(this.queues.keys()).map((name) => this.getQueueMetrics(name)), + ); + return results.filter((m): m is QueueMetrics => m !== null); + } async runComputeScores(): Promise { return this.computeScores(); @@ -95,15 +259,6 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { return count; } - // ─── computeScores ────────────────────────────────────────────────────── - - /** - * Process a batch of unfinalized claims, computing an aggregated confidence - * score from their stakes and marking high-confidence claims as resolved. - * - * N+1 pattern eliminated: wallets and users are bulk-fetched per claim batch - * rather than one DB round-trip per stake. - */ private async computeScores(): Promise { this.logger.debug('computeScores: starting'); const result: BatchResult = { processed: 0, updated: 0, errors: 0 }; @@ -118,16 +273,13 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { return result; } - // Bulk-load all stakes for this batch in one query const claimIds = claims.map((c) => c.id); const allStakes = await this.stakeRepo.find({ where: { claimId: In(claimIds) }, }); - // Group stakes by claimId for O(1) lookup const stakesByClaimId = groupBy(allStakes, (s) => s.claimId); - // Bulk-load wallets and users referenced in this batch const walletAddresses = [...new Set(allStakes.map((s) => s.walletAddress))]; const wallets = walletAddresses.length ? await this.walletRepo.find({ where: { address: In(walletAddresses) } }) @@ -142,14 +294,15 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { const userById = indexBy(users, (u) => u.id); - // Process each claim for (const claim of claims) { result.processed++; try { const stakes = stakesByClaimId.get(claim.id) ?? []; if (stakes.length === 0) { - this.logger.debug(`Claim ${claim.id}: no stakes — marking inconclusive`); + this.logger.debug( + `Claim ${claim.id}: no stakes — marking inconclusive`, + ); claim.confidenceScore = 0; await this.claimRepo.save(claim); result.updated++; @@ -169,15 +322,12 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { claim.confidenceScore = agg.confidence / CONFIDENCE_SCALE; if (agg.confidence > FINALIZATION_THRESHOLD) { - // Use transitionTo helper for validated state transition claim.transitionTo(ClaimState.FINALIZED, { - verdict: agg.status === 'VERIFIED_TRUE', + verdict: agg.status === ClaimStatus.VERIFIED_TRUE, confidence: claim.confidenceScore, }); } - await this.claimRepo.save(claim); - await this.claimRepo.save(claim); await this.claimsCache.invalidateClaim(claim.id); result.updated++; @@ -204,21 +354,6 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { } private async computeReputation(): Promise { - claimId: string, - updateFields: Partial, - ): Promise { - const result = await this.claimRepo - .createQueryBuilder() - .update(Claim) - .set(updateFields) - .where('id = :id', { id: claimId }) - .andWhere('finalized = false') - .execute(); - - return (result.affected ?? 0) > 0; - } - - private async computeReputation() { this.logger.debug('computeReputation: starting'); const result: BatchResult = { processed: 0, updated: 0, errors: 0 }; @@ -230,7 +365,6 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { const userIds = users.map((u) => u.id); - // Bulk-load wallets for all users in this batch const wallets = await this.walletRepo.find({ where: { userId: In(userIds) }, }); @@ -243,7 +377,6 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { return result; } - // Bulk-load all stakes for these wallets const allStakes = await this.stakeRepo .createQueryBuilder('s') .where('s.walletAddress IN (:...addrs)', { addrs: allAddresses }) @@ -251,7 +384,6 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { const stakesByWalletAddress = groupBy(allStakes, (s) => s.walletAddress); - // Bulk-load only finalized claims with a non-null verdict const stakedClaimIds = [...new Set(allStakes.map((s) => s.claimId))]; const finalizedClaims = stakedClaimIds.length > 0 @@ -266,7 +398,6 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { const claimById = indexBy(finalizedClaims, (c) => c.id); - // Process each user for (const user of users) { result.processed++; try { @@ -280,10 +411,12 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { const stakes = stakesByWalletAddress.get(wallet.address) ?? []; for (const stake of stakes) { const claim = claimById.get(stake.claimId); - if (!claim) continue; // not finalized or no verdict + if (!claim) continue; claimsVotedOn++; - if (this.deriveVotedTrue(stake) === Boolean(claim.resolvedVerdict)) { + if ( + this.deriveVotedTrue(stake) === Boolean(claim.resolvedVerdict) + ) { claimsCorrect++; } } @@ -316,7 +449,9 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { return result; } - // ─── Private helpers ──────────────────────────────────────────────────── + private getQueue(name: QueueName): Queue | undefined { + return this.queues.get(name); + } private buildVerifications( claimId: string, @@ -328,10 +463,12 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { const wallet = walletByAddress.get(stake.walletAddress); const user = wallet ? userById.get(wallet.userId) : null; + const rawAmount = (stake as unknown as { amount?: string | number }) + .amount; const stakeAmount = - typeof (stake as any).amount === 'string' - ? parseFloat((stake as any).amount) - : Number((stake as any).amount ?? 0); + typeof rawAmount === 'string' + ? parseFloat(rawAmount) + : Number(rawAmount ?? 0); const reputationWeight = user ? Math.max(0, Math.min(1, (user.reputation ?? 0) / 100)) @@ -340,27 +477,22 @@ export class JobsService implements OnModuleInit, OnModuleDestroy { return { id: stake.id, claimId, - userId: user?.id ?? null, - verdict: 'TRUE', + userId: user?.id ?? '', + verdict: VerificationVerdict.TRUE, stakeAmount, reputationWeight, - createdAt: (stake as any).updatedAt ?? new Date(), + createdAt: + (stake as unknown as { updatedAt?: Date }).updatedAt ?? new Date(), }; }); } - /** - * Derives whether a stake represents a TRUE vote. - * Currently all stakes are treated as TRUE; extend this once stakes carry - * an explicit `verdict` field. - */ private deriveVotedTrue(_stake: Stake): boolean { + void _stake; return true; } } -// ─── Utility functions ──────────────────────────────────────────────────────── - function groupBy(items: T[], keyFn: (item: T) => string): Map { const map = new Map(); for (const item of items) { @@ -376,4 +508,4 @@ function indexBy(items: T[], keyFn: (item: T) => string): Map { const map = new Map(); for (const item of items) map.set(keyFn(item), item); return map; -} \ No newline at end of file +} diff --git a/src/jobs/jobs.types.ts b/src/jobs/jobs.types.ts new file mode 100644 index 0000000..ab6b107 --- /dev/null +++ b/src/jobs/jobs.types.ts @@ -0,0 +1,56 @@ +export enum QueueName { + DEFAULT = 'jobs-queue', + NOTIFICATIONS = 'notifications-queue', + BLOCKCHAIN = 'blockchain-queue', + ANALYTICS = 'analytics-queue', +} + +export enum JobName { + COMPUTE_SCORES = 'compute-scores', + COMPUTE_REPUTATION = 'compute-reputation', + CLEANUP_SYBIL_HISTORY = 'cleanup-sybil-history', + SEND_NOTIFICATION = 'send-notification', + INDEX_BLOCKCHAIN_EVENTS = 'index-blockchain-events', + AGGREGATE_ANALYTICS = 'aggregate-analytics', +} + +export enum JobPriority { + CRITICAL = 1, + HIGH = 2, + NORMAL = 3, + LOW = 4, + BACKGROUND = 5, +} + +export interface RetryPolicy { + attempts: number; + backoff: { + type: 'exponential' | 'fixed'; + delay: number; + }; +} + +export interface QueueMetrics { + name: string; + waiting: number; + active: number; + completed: number; + failed: number; + delayed: number; + paused: boolean; +} + +export interface JobOptions { + priority?: JobPriority; + delay?: number; + attempts?: number; + backoffDelay?: number; +} + +export const DEFAULT_RETRY_POLICY: RetryPolicy = { + attempts: 3, + backoff: { + type: 'exponential', + delay: 2000, + }, +}; diff --git a/src/search/dto/search-query.dto.ts b/src/search/dto/search-query.dto.ts new file mode 100644 index 0000000..ed092f7 --- /dev/null +++ b/src/search/dto/search-query.dto.ts @@ -0,0 +1,67 @@ +import { + IsOptional, + IsString, + IsEnum, + IsBoolean, + IsNumber, + Min, + Max, +} from 'class-validator'; +import { Type } from 'class-transformer'; +import { SortField, PaginationType } from '../search.types'; + +export class SearchQueryDto { + @IsOptional() + @IsString() + q?: string; + + @IsOptional() + @IsString() + status?: string; + + @IsOptional() + @IsString() + owner?: string; + + @IsOptional() + @IsString() + walletAddress?: string; + + @IsOptional() + @IsString() + fromDate?: string; + + @IsOptional() + @IsString() + toDate?: string; + + @IsOptional() + @IsBoolean() + @Type(() => Boolean) + finalized?: boolean; + + @IsOptional() + @IsEnum(SortField) + sort?: SortField = SortField.NEWEST; + + @IsOptional() + @IsEnum(PaginationType) + pagination?: PaginationType = PaginationType.OFFSET; + + @IsOptional() + @Type(() => Number) + @IsNumber() + @Min(1) + page?: number = 1; + + @IsOptional() + @Type(() => Number) + @IsNumber() + @Min(1) + @Max(100) + limit?: number = 20; + + @IsOptional() + @IsString() + cursor?: string; +} diff --git a/src/search/search.controller.ts b/src/search/search.controller.ts new file mode 100644 index 0000000..2487d31 --- /dev/null +++ b/src/search/search.controller.ts @@ -0,0 +1,71 @@ +import { Controller, Get, Param, Query, ValidationPipe } from '@nestjs/common'; +import { ApiOperation, ApiTags } from '@nestjs/swagger'; +import { SearchService } from './search.service'; +import { SearchQueryDto } from './dto/search-query.dto'; +import { + GlobalSearchResult, + PaginationParams, + PaginationType, + SearchableEntity, + SearchFilter, + SearchResult, + SortField, +} from './search.types'; + +@ApiTags('Search') +@Controller('search') +export class SearchController { + constructor(private readonly searchService: SearchService) {} + + @Get() + @ApiOperation({ summary: 'Global search across claims, disputes, and users' }) + async globalSearch( + @Query(new ValidationPipe({ transform: true, whitelist: true })) + query: SearchQueryDto, + ): Promise { + const { q = '', ...filters } = query; + return this.searchService.searchGlobal( + q, + this.buildFilters(filters), + this.buildPagination(filters), + (filters.sort as SortField) ?? SortField.NEWEST, + ); + } + + @Get(':entity') + @ApiOperation({ summary: 'Entity-specific search' }) + async entitySearch( + @Param('entity') entity: SearchableEntity, + @Query(new ValidationPipe({ transform: true, whitelist: true })) + query: SearchQueryDto, + ): Promise> { + const { q = '', ...filters } = query; + return this.searchService.searchEntity( + entity, + q, + this.buildFilters(filters), + this.buildPagination(filters), + (filters.sort as SortField) ?? SortField.NEWEST, + ); + } + + private buildFilters(dto: Omit): SearchFilter { + return { + status: dto.status, + owner: dto.owner, + walletAddress: dto.walletAddress, + fromDate: dto.fromDate, + toDate: dto.toDate, + finalized: dto.finalized, + }; + } + + private buildPagination(dto: Omit): PaginationParams { + return { + page: dto.page ?? 1, + limit: dto.limit ?? 20, + type: (dto.pagination as PaginationType) ?? PaginationType.OFFSET, + cursor: dto.cursor, + }; + } +} diff --git a/src/search/search.module.ts b/src/search/search.module.ts new file mode 100644 index 0000000..19cfa5e --- /dev/null +++ b/src/search/search.module.ts @@ -0,0 +1,16 @@ +import { Module } from '@nestjs/common'; +import { TypeOrmModule } from '@nestjs/typeorm'; +import { SearchService } from './search.service'; +import { SearchController } from './search.controller'; +import { Claim } from '../claims/entities/claim.entity'; +import { Dispute } from '../dispute/entities/dispute.entity'; +import { User } from '../entities/user.entity'; +import { RedisModule } from '../redis/redis.module'; + +@Module({ + imports: [TypeOrmModule.forFeature([Claim, Dispute, User]), RedisModule], + providers: [SearchService], + controllers: [SearchController], + exports: [SearchService], +}) +export class SearchModule {} diff --git a/src/search/search.service.spec.ts b/src/search/search.service.spec.ts new file mode 100644 index 0000000..f57f375 --- /dev/null +++ b/src/search/search.service.spec.ts @@ -0,0 +1,72 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { getRepositoryToken } from '@nestjs/typeorm'; + +import { SearchService } from './search.service'; +import { PaginationType, SortField } from './search.types'; +import { Claim } from '../claims/entities/claim.entity'; +import { Dispute } from '../dispute/entities/dispute.entity'; +import { User } from '../entities/user.entity'; +import { RedisService } from '../redis/redis.service'; + +describe('SearchService', () => { + let service: SearchService; + + const createMockRepo = () => ({ + createQueryBuilder: jest.fn(() => ({ + andWhere: jest.fn().mockReturnThis(), + orderBy: jest.fn().mockReturnThis(), + skip: jest.fn().mockReturnThis(), + take: jest.fn().mockReturnThis(), + getManyAndCount: jest.fn().mockResolvedValue([[], 0]), + })), + }); + + const mockRedis = () => ({ + get: jest.fn(), + set: jest.fn(), + }); + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + providers: [ + SearchService, + { + provide: getRepositoryToken(Claim), + useValue: createMockRepo(), + }, + { + provide: getRepositoryToken(Dispute), + useValue: createMockRepo(), + }, + { + provide: getRepositoryToken(User), + useValue: createMockRepo(), + }, + { + provide: RedisService, + useFactory: mockRedis, + }, + ], + }).compile(); + + service = module.get(SearchService); + }); + + it('should be defined', () => { + expect(service).toBeDefined(); + }); + + it('should perform global search', async () => { + const result = await service.searchGlobal( + 'test', + {}, + { page: 1, limit: 20, type: PaginationType.OFFSET }, + SortField.NEWEST, + ); + + expect(result.query).toBe('test'); + expect(result.claims).toBeDefined(); + expect(result.disputes).toBeDefined(); + expect(result.users).toBeDefined(); + }); +}); diff --git a/src/search/search.service.ts b/src/search/search.service.ts new file mode 100644 index 0000000..280b9b5 --- /dev/null +++ b/src/search/search.service.ts @@ -0,0 +1,241 @@ +import { Injectable, Logger, BadRequestException } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository, SelectQueryBuilder, Brackets } from 'typeorm'; +import { Claim } from '../claims/entities/claim.entity'; +import { Dispute } from '../dispute/entities/dispute.entity'; +import { User } from '../entities/user.entity'; +import { RedisService } from '../redis/redis.service'; +import { + GlobalSearchResult, + PaginationParams, + SearchableEntity, + SearchFilter, + SearchResult, + SortField, +} from './search.types'; + +const CACHE_TTL_SECONDS = 30; + +@Injectable() +export class SearchService { + private readonly logger = new Logger(SearchService.name); + + constructor( + @InjectRepository(Claim) + private readonly claimRepo: Repository, + @InjectRepository(Dispute) + private readonly disputeRepo: Repository, + @InjectRepository(User) + private readonly userRepo: Repository, + private readonly redisService: RedisService, + ) {} + + async searchGlobal( + query: string, + filters: SearchFilter, + pagination: PaginationParams, + sort: SortField, + ): Promise { + const cacheKey = this.globalCacheKey(query, filters, pagination, sort); + const cached = await this.redisService.get(cacheKey); + if (cached) { + try { + return JSON.parse(cached) as GlobalSearchResult; + } catch { + // fall through + } + } + + const [claims, disputes, users] = await Promise.all([ + this.searchClaims(query, filters, pagination, sort), + this.searchDisputes(query, filters, pagination, sort), + this.searchUsers(query, filters, pagination, sort), + ]); + + const result: GlobalSearchResult = { + query, + claims, + disputes, + users, + }; + + await this.redisService.set( + cacheKey, + JSON.stringify(result), + CACHE_TTL_SECONDS, + ); + return result; + } + + async searchEntity( + entity: SearchableEntity, + query: string, + filters: SearchFilter, + pagination: PaginationParams, + sort: SortField, + ): Promise> { + switch (entity) { + case 'claims': + return this.searchClaims(query, filters, pagination, sort); + case 'disputes': + return this.searchDisputes(query, filters, pagination, sort); + case 'users': + return this.searchUsers(query, filters, pagination, sort); + default: + throw new BadRequestException(`Unknown entity: ${String(entity)}`); + } + } + + private async searchClaims( + query: string, + filters: SearchFilter, + pagination: PaginationParams, + sort: SortField, + ): Promise> { + const qb = this.claimRepo.createQueryBuilder('claim'); + + if (query) { + const likeQuery = `%${query}%`; + qb.andWhere( + new Brackets((sub) => { + sub.where('claim.title LIKE :q', { q: likeQuery }); + sub.orWhere('claim.content LIKE :q', { q: likeQuery }); + sub.orWhere('claim.source LIKE :q', { q: likeQuery }); + }), + ); + } + + if (filters.finalized !== undefined) { + qb.andWhere('claim.finalized = :finalized', { + finalized: filters.finalized, + }); + } + + if (filters.fromDate) { + qb.andWhere('claim.createdAt >= :fromDate', { + fromDate: new Date(filters.fromDate), + }); + } + + if (filters.toDate) { + qb.andWhere('claim.createdAt <= :toDate', { + toDate: new Date(filters.toDate), + }); + } + + this.applySorting(qb, sort, 'claim'); + return this.executeOffsetPagination(qb, pagination); + } + + private async searchDisputes( + query: string, + filters: SearchFilter, + pagination: PaginationParams, + sort: SortField, + ): Promise> { + const qb = this.disputeRepo.createQueryBuilder('dispute'); + + if (query) { + const likeQuery = `%${query}%`; + qb.andWhere( + new Brackets((sub) => { + sub.where('dispute.status LIKE :q', { q: likeQuery }); + sub.orWhere('dispute.trigger LIKE :q', { q: likeQuery }); + }), + ); + } + + if (filters.status) { + qb.andWhere('dispute.status = :status', { status: filters.status }); + } + + if (filters.fromDate) { + qb.andWhere('dispute.createdAt >= :fromDate', { + fromDate: new Date(filters.fromDate), + }); + } + + if (filters.toDate) { + qb.andWhere('dispute.createdAt <= :toDate', { + toDate: new Date(filters.toDate), + }); + } + + this.applySorting(qb, sort, 'dispute'); + return this.executeOffsetPagination(qb, pagination); + } + + private async searchUsers( + query: string, + filters: SearchFilter, + pagination: PaginationParams, + sort: SortField, + ): Promise> { + const qb = this.userRepo.createQueryBuilder('user'); + + if (query) { + qb.andWhere('user.walletAddress LIKE :q', { q: `%${query}%` }); + } + + if (filters.walletAddress) { + qb.andWhere('user.walletAddress = :walletAddress', { + walletAddress: filters.walletAddress, + }); + } + + if ((sort as string) === 'reputation') { + qb.orderBy('user.reputation', 'DESC'); + } else { + this.applySorting(qb, sort, 'user'); + } + + return this.executeOffsetPagination(qb, pagination); + } + + private applySorting>( + qb: SelectQueryBuilder, + sort: SortField, + alias: string, + ): void { + const sortValue = sort as string; + switch (sortValue) { + case 'oldest': + qb.orderBy(`${alias}.createdAt`, 'ASC'); + break; + case 'relevance': + case 'reward': + case 'newest': + default: + qb.orderBy(`${alias}.createdAt`, 'DESC'); + break; + } + } + + private async executeOffsetPagination>( + qb: SelectQueryBuilder, + pagination: PaginationParams, + ): Promise> { + const page = Math.max(1, pagination.page); + const limit = Math.min(100, Math.max(1, pagination.limit)); + const skip = (page - 1) * limit; + + const [data, total] = await qb.skip(skip).take(limit).getManyAndCount(); + + return { + data, + total, + page, + limit, + totalPages: Math.ceil(total / limit), + }; + } + + private globalCacheKey( + query: string, + filters: SearchFilter, + pagination: PaginationParams, + sort: SortField, + ): string { + return `search:global:${query}:${JSON.stringify(filters)}:${String(sort)}:${pagination.page}:${pagination.limit}`; + } +} diff --git a/src/search/search.types.ts b/src/search/search.types.ts new file mode 100644 index 0000000..f6ffe90 --- /dev/null +++ b/src/search/search.types.ts @@ -0,0 +1,45 @@ +export type SearchableEntity = 'claims' | 'disputes' | 'users'; + +export enum SortField { + NEWEST = 'newest', + OLDEST = 'oldest', + RELEVANCE = 'relevance', + REPUTATION = 'reputation', + REWARD = 'reward', +} + +export enum PaginationType { + OFFSET = 'offset', + CURSOR = 'cursor', +} + +export interface SearchFilter { + status?: string; + owner?: string; + walletAddress?: string; + fromDate?: string; + toDate?: string; + finalized?: boolean; +} + +export interface PaginationParams { + page: number; + limit: number; + type: PaginationType; + cursor?: string; +} + +export interface SearchResult { + data: T[]; + total: number; + page: number; + limit: number; + totalPages: number; +} + +export interface GlobalSearchResult { + query: string; + claims: SearchResult; + disputes: SearchResult; + users: SearchResult; +} diff --git a/test/feature-flags.e2e-spec.ts b/test/feature-flags.e2e-spec.ts new file mode 100644 index 0000000..6fb26aa --- /dev/null +++ b/test/feature-flags.e2e-spec.ts @@ -0,0 +1,42 @@ +/* eslint-disable @typescript-eslint/no-unsafe-member-access */ +/* eslint-disable @typescript-eslint/no-unsafe-argument */ +import { Test, TestingModule } from '@nestjs/testing'; +import { INestApplication } from '@nestjs/common'; +import request from 'supertest'; +import { AppModule } from '../src/app.module'; + +describe('FeatureFlagsController (e2e)', () => { + let app: INestApplication; + + beforeAll(async () => { + const moduleFixture: TestingModule = await Test.createTestingModule({ + imports: [AppModule], + }).compile(); + + app = moduleFixture.createNestApplication(); + await app.init(); + }); + + afterAll(async () => { + await app.close(); + }); + + it('/feature-flags (GET) returns list', () => { + return request(app.getHttpServer()) + .get('/feature-flags') + .expect(200) + .expect((res) => { + expect(Array.isArray(res.body)).toBe(true); + }); + }); + + it('/feature-flags/evaluate/:key (GET) returns evaluation', () => { + return request(app.getHttpServer()) + .get('/feature-flags/evaluate/unknown-flag') + .expect(200) + .expect((res) => { + expect(res.body.key).toBe('unknown-flag'); + expect(res.body.enabled).toBe(false); + }); + }); +}); diff --git a/test/health.e2e-spec.ts b/test/health.e2e-spec.ts new file mode 100644 index 0000000..c8aaa9e --- /dev/null +++ b/test/health.e2e-spec.ts @@ -0,0 +1,54 @@ +/* eslint-disable @typescript-eslint/no-unsafe-member-access */ +/* eslint-disable @typescript-eslint/no-unsafe-argument */ +import { Test, TestingModule } from '@nestjs/testing'; +import { INestApplication } from '@nestjs/common'; +import request from 'supertest'; +import { AppModule } from '../src/app.module'; + +describe('HealthController (e2e)', () => { + let app: INestApplication; + + beforeAll(async () => { + const moduleFixture: TestingModule = await Test.createTestingModule({ + imports: [AppModule], + }).compile(); + + app = moduleFixture.createNestApplication(); + await app.init(); + }); + + afterAll(async () => { + await app.close(); + }); + + it('/health/live (GET) returns alive', () => { + return request(app.getHttpServer()) + .get('/health/live') + .expect(200) + .expect((res) => { + expect(res.body.status).toBe('alive'); + expect(res.body.uptime).toBeGreaterThanOrEqual(0); + }); + }); + + it('/health/ready (GET) returns readiness status', () => { + return request(app.getHttpServer()) + .get('/health/ready') + .expect(200) + .expect((res) => { + expect(typeof res.body.ready).toBe('boolean'); + expect(res.body.dependencies).toBeInstanceOf(Array); + }); + }); + + it('/health (GET) returns aggregated report', () => { + return request(app.getHttpServer()) + .get('/health') + .expect(200) + .expect((res) => { + expect(['healthy', 'degraded', 'unhealthy']).toContain(res.body.status); + expect(res.body.services).toBeDefined(); + expect(res.body.dependencies).toBeInstanceOf(Array); + }); + }); +}); diff --git a/test/jobs.e2e-spec.ts b/test/jobs.e2e-spec.ts new file mode 100644 index 0000000..59deb43 --- /dev/null +++ b/test/jobs.e2e-spec.ts @@ -0,0 +1,26 @@ +/* eslint-disable @typescript-eslint/no-unsafe-argument */ +import { Test, TestingModule } from '@nestjs/testing'; +import { INestApplication } from '@nestjs/common'; +import request from 'supertest'; +import { AppModule } from '../src/app.module'; + +describe('JobsController (e2e)', () => { + let app: INestApplication; + + beforeAll(async () => { + const moduleFixture: TestingModule = await Test.createTestingModule({ + imports: [AppModule], + }).compile(); + + app = moduleFixture.createNestApplication(); + await app.init(); + }); + + afterAll(async () => { + await app.close(); + }); + + it('/admin/jobs/metrics (GET) requires authentication', () => { + return request(app.getHttpServer()).get('/admin/jobs/metrics').expect(403); + }); +}); diff --git a/test/search.e2e-spec.ts b/test/search.e2e-spec.ts new file mode 100644 index 0000000..d75f788 --- /dev/null +++ b/test/search.e2e-spec.ts @@ -0,0 +1,45 @@ +/* eslint-disable @typescript-eslint/no-unsafe-member-access */ +/* eslint-disable @typescript-eslint/no-unsafe-argument */ +import { Test, TestingModule } from '@nestjs/testing'; +import { INestApplication } from '@nestjs/common'; +import request from 'supertest'; +import { AppModule } from '../src/app.module'; + +describe('SearchController (e2e)', () => { + let app: INestApplication; + + beforeAll(async () => { + const moduleFixture: TestingModule = await Test.createTestingModule({ + imports: [AppModule], + }).compile(); + + app = moduleFixture.createNestApplication(); + await app.init(); + }); + + afterAll(async () => { + await app.close(); + }); + + it('/search (GET) returns global search results', () => { + return request(app.getHttpServer()) + .get('/search?q=test') + .expect(200) + .expect((res) => { + expect(res.body.query).toBe('test'); + expect(res.body.claims).toBeDefined(); + expect(res.body.disputes).toBeDefined(); + expect(res.body.users).toBeDefined(); + }); + }); + + it('/search/:entity (GET) returns entity-specific results', () => { + return request(app.getHttpServer()) + .get('/search/claims?q=test') + .expect(200) + .expect((res) => { + expect(res.body.data).toBeInstanceOf(Array); + expect(res.body.total).toBeGreaterThanOrEqual(0); + }); + }); +});