Merge pull request #749 from RaymondAbiola/Event_Emitter

[Backend] — Event Emitter Has No Dead Letter Queue

Author: Devsol-01

backend/package.json

@@ -27,6 +27,7 @@
     "@keyv/redis": "^5.1.6",
     "@nestjs-modules/mailer": "^2.0.2",
     "@nestjs/axios": "^4.0.1",
+    "@nestjs/bull": "^11.0.2",
     "@nestjs/cache-manager": "^3.1.0",
     "@nestjs/common": "^11.0.1",
     "@nestjs/config": "^4.0.3",
@@ -44,6 +45,7 @@
     "archiver": "^7.0.1",
     "axios": "^1.13.5",
     "bcrypt": "^6.0.0",
+    "bull": "^4.16.5",
     "cache-manager": "^7.2.8",
     "cache-manager-redis-store": "^3.0.1",
     "cacheable": "^2.3.4",
@@ -81,6 +83,7 @@
     "@swc/core": "^1.10.7",
     "@types/archiver": "^6.0.3",
     "@types/bcrypt": "^6.0.0",
+    "@types/bull": "^3.15.9",
     "@types/express": "^5.0.0",
     "@types/jest": "^29.5.14",
     "@types/multer": "^2.0.0",

backend/src/app.module.ts

@@ -9,6 +9,7 @@ import { RequestLoggingInterceptor } from './common/interceptors/request-logging
 import { GracefulShutdownInterceptor } from './common/interceptors/graceful-shutdown.interceptor';
 import { TieredThrottlerGuard } from './common/guards/tiered-throttler.guard';
 import { CommonModule } from './common/common.module';
+import { EventBusModule } from './common/event-bus/event-bus.module';
 import { EventEmitterModule } from '@nestjs/event-emitter';
 import { LoggerModule } from 'nestjs-pino';
 import * as Joi from 'joi';
@@ -219,6 +220,7 @@ const envValidationSchema = Joi.object({
     PostmanModule,
     PerformanceModule,
     CommonModule,
+    EventBusModule.forRoot(),
     ThrottlerModule.forRoot([
       {
         name: 'default',

backend/src/common/event-bus/dto/replay-events.dto.ts

@@ -0,0 +1,16 @@
+import { ApiProperty } from '@nestjs/swagger';
+import { ArrayMinSize, ArrayMaxSize, IsArray, IsUUID } from 'class-validator';
+
+export class ReplayEventsDto {
+  @ApiProperty({
+    description: 'IDs of failed events to replay',
+    type: [String],
+    minItems: 1,
+    maxItems: 100,
+  })
+  @IsArray()
+  @ArrayMinSize(1)
+  @ArrayMaxSize(100)
+  @IsUUID('4', { each: true })
+  ids: string[];
+}

backend/src/common/event-bus/entities/failed-event.entity.ts

@@ -0,0 +1,75 @@
+import {
+  Entity,
+  PrimaryGeneratedColumn,
+  Column,
+  CreateDateColumn,
+  UpdateDateColumn,
+  Index,
+} from 'typeorm';
+
+export enum FailedEventStatus {
+  PENDING_RETRY = 'pending_retry',
+  DEAD = 'dead',
+  REPLAYED = 'replayed',
+  DISCARDED = 'discarded',
+}
+
+/**
+ * General-purpose Dead Letter Queue record for events whose
+ * EventEmitter2 listeners failed after all retry attempts were exhausted.
+ *
+ * Distinct from `DeadLetterEvent`, which is scoped to raw Soroban indexer events.
+ */
+@Entity('failed_events')
+@Index('idx_failed_events_status_created', ['status', 'createdAt'])
+@Index('idx_failed_events_event_name', ['eventName'])
+export class FailedEvent {
+  @PrimaryGeneratedColumn('uuid')
+  id: string;
+
+  @Column({ type: 'varchar', length: 255 })
+  eventName: string;
+
+  /** Original event payload, serialised. */
+  @Column({ type: 'jsonb' })
+  payload: unknown;
+
+  @Column({ type: 'text', nullable: true })
+  errorMessage: string | null;
+
+  @Column({ type: 'text', nullable: true })
+  errorStack: string | null;
+
+  @Column({ type: 'int', default: 0 })
+  attempts: number;
+
+  @Column({ type: 'int', default: 3 })
+  maxAttempts: number;
+
+  @Column({
+    type: 'enum',
+    enum: FailedEventStatus,
+    default: FailedEventStatus.PENDING_RETRY,
+  })
+  status: FailedEventStatus;
+
+  /** Optional source identifier (service, module) that emitted the event. */
+  @Column({ type: 'varchar', length: 255, nullable: true })
+  source: string | null;
+
+  /** Optional correlation id for distributed tracing. */
+  @Column({ type: 'varchar', length: 255, nullable: true })
+  correlationId: string | null;
+
+  @Column({ type: 'timestamp with time zone', nullable: true })
+  lastAttemptAt: Date | null;
+
+  @Column({ type: 'timestamp with time zone', nullable: true })
+  nextRetryAt: Date | null;
+
+  @CreateDateColumn({ type: 'timestamp with time zone' })
+  createdAt: Date;
+
+  @UpdateDateColumn({ type: 'timestamp with time zone' })
+  updatedAt: Date;
+}

backend/src/common/event-bus/event-bus.constants.ts

@@ -0,0 +1,11 @@
+export const EVENT_DLQ_QUEUE = 'event-dlq';
+
+export const EVENT_DLQ_RETRY_JOB = 'retry-failed-event';
+
+/**
+ * Default retry policy used by ResilientEventBus when a caller does
+ * not provide explicit options.
+ */
+export const DEFAULT_MAX_ATTEMPTS = 3;
+export const DEFAULT_BASE_BACKOFF_MS = 1000;
+export const MAX_BACKOFF_MS = 30 * 1000;

backend/src/common/event-bus/event-bus.module.ts

@@ -0,0 +1,68 @@
+import { DynamicModule, Global, Logger, Module } from '@nestjs/common';
+import { ConfigModule, ConfigService } from '@nestjs/config';
+import { TypeOrmModule } from '@nestjs/typeorm';
+import { BullModule } from '@nestjs/bull';
+import { FailedEvent } from './entities/failed-event.entity';
+import { EventDlqService } from './event-dlq.service';
+import { ResilientEventBus } from './resilient-event-bus.service';
+import { EventDlqController } from './event-dlq.controller';
+import { EventDlqProcessor } from './event-dlq.processor';
+import { EVENT_DLQ_QUEUE } from './event-bus.constants';
+
+/**
+ * Wires the Resilient Event Bus + DLQ infrastructure.
+ *
+ * Bull-backed async retries are activated only when REDIS_URL is configured.
+ * Without Redis, failed events are still recorded to the DB and can be
+ * replayed manually via the admin endpoints — they are simply marked DEAD
+ * immediately rather than retried in the background.
+ */
+@Global()
+@Module({})
+export class EventBusModule {
+  private static readonly logger = new Logger(EventBusModule.name);
+
+  static forRoot(): DynamicModule {
+    const redisUrl = process.env.REDIS_URL;
+
+    const bullImports = redisUrl
+      ? [
+          BullModule.forRootAsync({
+            imports: [ConfigModule],
+            inject: [ConfigService],
+            useFactory: (config: ConfigService) => {
+              const url = config.get<string>('REDIS_URL');
+              return {
+                redis: url,
+                defaultJobOptions: {
+                  attempts: 3,
+                  backoff: { type: 'exponential', delay: 1000 },
+                  removeOnComplete: true,
+                  removeOnFail: false,
+                },
+              };
+            },
+          }),
+          BullModule.registerQueue({ name: EVENT_DLQ_QUEUE }),
+        ]
+      : [];
+
+    if (!redisUrl) {
+      this.logger.warn(
+        'REDIS_URL not configured — Event DLQ will run without async Bull retries. Failed events will be persisted and available for manual replay only.',
+      );
+    }
+
+    const providers = redisUrl
+      ? [ResilientEventBus, EventDlqService, EventDlqProcessor]
+      : [ResilientEventBus, EventDlqService];
+
+    return {
+      module: EventBusModule,
+      imports: [TypeOrmModule.forFeature([FailedEvent]), ...bullImports],
+      controllers: [EventDlqController],
+      providers,
+      exports: [ResilientEventBus, EventDlqService],
+    };
+  }
+}

backend/src/common/event-bus/event-dlq.controller.ts

@@ -0,0 +1,94 @@
+import {
+  Body,
+  Controller,
+  Delete,
+  Get,
+  NotFoundException,
+  Param,
+  ParseUUIDPipe,
+  Post,
+  Query,
+  UseGuards,
+} from '@nestjs/common';
+import {
+  ApiBearerAuth,
+  ApiOperation,
+  ApiQuery,
+  ApiResponse,
+  ApiTags,
+} from '@nestjs/swagger';
+import { JwtAuthGuard } from '../../auth/guards/jwt-auth.guard';
+import { RolesGuard } from '../guards/roles.guard';
+import { Roles } from '../decorators/roles.decorator';
+import { Role } from '../enums/role.enum';
+import { EventDlqService } from './event-dlq.service';
+import { FailedEventStatus } from './entities/failed-event.entity';
+import { ReplayEventsDto } from './dto/replay-events.dto';
+
+@ApiTags('admin: event-dlq')
+@ApiBearerAuth()
+@Controller('admin/event-dlq')
+@UseGuards(JwtAuthGuard, RolesGuard)
+@Roles(Role.ADMIN)
+export class EventDlqController {
+  constructor(private readonly dlqService: EventDlqService) {}
+
+  @Get('stats')
+  @ApiOperation({ summary: 'DLQ summary stats for monitoring dashboards' })
+  @ApiResponse({ status: 200, description: 'Counts by status & event name' })
+  async getStats() {
+    return this.dlqService.getStats();
+  }
+
+  @Get()
+  @ApiOperation({ summary: 'List failed events' })
+  @ApiQuery({ name: 'status', enum: FailedEventStatus, required: false })
+  @ApiQuery({ name: 'eventName', required: false })
+  @ApiQuery({ name: 'limit', required: false, type: Number })
+  @ApiQuery({ name: 'offset', required: false, type: Number })
+  async list(
+    @Query('status') status?: FailedEventStatus,
+    @Query('eventName') eventName?: string,
+    @Query('limit') limit?: string,
+    @Query('offset') offset?: string,
+  ) {
+    return this.dlqService.list({
+      status,
+      eventName,
+      limit: limit ? parseInt(limit, 10) : undefined,
+      offset: offset ? parseInt(offset, 10) : undefined,
+    });
+  }
+
+  @Get(':id')
+  @ApiOperation({ summary: 'Get a failed event by id' })
+  async getOne(@Param('id', new ParseUUIDPipe()) id: string) {
+    const record = await this.dlqService.getOne(id);
+    if (!record) throw new NotFoundException('Failed event not found');
+    return record;
+  }
+
+  @Post(':id/replay')
+  @ApiOperation({ summary: 'Replay a single failed event' })
+  async replayOne(@Param('id', new ParseUUIDPipe()) id: string) {
+    const result = await this.dlqService.replay(id);
+    if (!result.success && result.error === 'Failed event not found') {
+      throw new NotFoundException(result.error);
+    }
+    return result;
+  }
+
+  @Post('replay')
+  @ApiOperation({ summary: 'Replay multiple failed events' })
+  async replayMany(@Body() dto: ReplayEventsDto) {
+    return this.dlqService.replayMany(dto.ids);
+  }
+
+  @Delete(':id')
+  @ApiOperation({ summary: 'Discard a failed event without replaying' })
+  async discard(@Param('id', new ParseUUIDPipe()) id: string) {
+    const ok = await this.dlqService.discard(id);
+    if (!ok) throw new NotFoundException('Failed event not found');
+    return { discarded: true };
+  }
+}

backend/src/common/event-bus/event-dlq.processor.ts

@@ -0,0 +1,25 @@
+import { Process, Processor } from '@nestjs/bull';
+import { Logger } from '@nestjs/common';
+import { Job } from 'bull';
+import { EventDlqService } from './event-dlq.service';
+import { EVENT_DLQ_QUEUE, EVENT_DLQ_RETRY_JOB } from './event-bus.constants';
+
+interface RetryJobData {
+  failedEventId: string;
+}
+
+@Processor(EVENT_DLQ_QUEUE)
+export class EventDlqProcessor {
+  private readonly logger = new Logger(EventDlqProcessor.name);
+
+  constructor(private readonly dlqService: EventDlqService) {}
+
+  @Process(EVENT_DLQ_RETRY_JOB)
+  async handleRetry(job: Job<RetryJobData>): Promise<void> {
+    const { failedEventId } = job.data;
+    this.logger.debug(
+      `Processing retry job ${job.id} for failed event ${failedEventId} (Bull attempt ${job.attemptsMade + 1})`,
+    );
+    await this.dlqService.runRetryJob(failedEventId);
+  }
+}