Me/gf/wave 3

backend/src/app.module.ts

@@ -8,6 +8,7 @@ import { SentryModule } from './sentry/sentry.module';
 import { RedisModule } from './common/redis/redis.module';
 import { RateLimitModule } from './common/rate-limit/rate-limit.module';
 import { UserProfileModule } from './user-profile/user-profile.module';
+import { EventIngestionModule } from './event-ingestion/event-ingestion.module';
 
 @Module({
   imports: [
@@ -20,6 +21,7 @@ import { UserProfileModule } from './user-profile/user-profile.module';
     WebhookModule,
     MonitoringModule,
     StellarModule,
+    EventIngestionModule,
   ],
 })
 export class AppModule {}

backend/src/event-ingestion/dto/event-ingestion.dto.ts

@@ -0,0 +1,29 @@
+import { IsString, IsOptional, IsNumber } from 'class-validator';
+import { ApiProperty, ApiPropertyOptional } from '@nestjs/swagger';
+
+export class StartPollingDto {
+  @ApiPropertyOptional({ description: 'Contract ID to monitor' })
+  @IsString()
+  @IsOptional()
+  contractId?: string;
+}
+
+export class IngestLedgerDto {
+  @ApiProperty({ description: 'Contract ID to ingest events for' })
+  @IsString()
+  contractId: string;
+
+  @ApiProperty({ description: 'Ledger sequence to ingest' })
+  @IsNumber()
+  ledger: number;
+}
+
+export class HandleReorgDto {
+  @ApiProperty({ description: 'Contract ID affected by reorg' })
+  @IsString()
+  contractId: string;
+
+  @ApiProperty({ description: 'Ledger sequence to reprocess from' })
+  @IsNumber()
+  fromLedger: number;
+}

backend/src/event-ingestion/event-ingestion.controller.spec.ts

@@ -0,0 +1,65 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { EventIngestionController } from './event-ingestion.controller';
+import { EventIngestionService } from './event-ingestion.service';
+import { LedgerCursorService } from './ledger-cursor.service';
+import { EventProcessorService } from './event-processor.service';
+import { EscrowService } from '../escrow/escrow.service';
+
+describe('EventIngestionController', () => {
+  let controller: EventIngestionController;
+  let service: EventIngestionService;
+
+  const mockEscrowService = {
+    create: jest.fn(),
+    findById: jest.fn(),
+    release: jest.fn(),
+    raiseDispute: jest.fn(),
+  };
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      controllers: [EventIngestionController],
+      providers: [
+        EventIngestionService,
+        LedgerCursorService,
+        EventProcessorService,
+        { provide: EscrowService, useValue: mockEscrowService },
+      ],
+    }).compile();
+
+    controller = module.get<EventIngestionController>(EventIngestionController);
+    service = module.get<EventIngestionService>(EventIngestionService);
+  });
+
+  it('should be defined', () => {
+    expect(controller).toBeDefined();
+  });
+
+  describe('startPolling', () => {
+    it('should start polling', async () => {
+      const startSpy = jest.spyOn(service, 'startPolling').mockResolvedValue();
+      await controller.startPolling({ contractId: 'test-contract' });
+      expect(startSpy).toHaveBeenCalledWith('test-contract');
+    });
+  });
+
+  describe('stopPolling', () => {
+    it('should stop polling', async () => {
+      const stopSpy = jest.spyOn(service, 'stopPolling');
+      await controller.stopPolling();
+      expect(stopSpy).toHaveBeenCalled();
+    });
+  });
+
+  describe('getStatus', () => {
+    it('should return status', async () => {
+      const statusSpy = jest.spyOn(service, 'getStatus').mockResolvedValue({
+        isRunning: false,
+        failedEvents: 0,
+      });
+      const result = await controller.getStatus();
+      expect(result).toHaveProperty('isRunning');
+      expect(statusSpy).toHaveBeenCalled();
+    });
+  });
+});

backend/src/event-ingestion/event-ingestion.controller.ts

@@ -0,0 +1,62 @@
+import { Controller, Post, Get, Body, Param, HttpCode, HttpStatus } from '@nestjs/common';
+import { ApiTags, ApiOperation, ApiResponse } from '@nestjs/swagger';
+import { EventIngestionService } from './event-ingestion.service';
+import { StartPollingDto, IngestLedgerDto, HandleReorgDto } from './dto/event-ingestion.dto';
+
+@ApiTags('Event Ingestion')
+@Controller('event-ingestion')
+export class EventIngestionController {
+  constructor(private readonly eventIngestionService: EventIngestionService) {}
+
+  @Post('start')
+  @HttpCode(HttpStatus.OK)
+  @ApiOperation({ summary: 'Start polling for Soroban events' })
+  @ApiResponse({ status: 200, description: 'Polling started successfully' })
+  async startPolling(@Body() dto: StartPollingDto) {
+    await this.eventIngestionService.startPolling(dto.contractId);
+    return { message: 'Polling started', contractId: dto.contractId };
+  }
+
+  @Post('stop')
+  @HttpCode(HttpStatus.OK)
+  @ApiOperation({ summary: 'Stop polling for Soroban events' })
+  @ApiResponse({ status: 200, description: 'Polling stopped successfully' })
+  async stopPolling() {
+    this.eventIngestionService.stopPolling();
+    return { message: 'Polling stopped' };
+  }
+
+  @Post('ingest')
+  @HttpCode(HttpStatus.OK)
+  @ApiOperation({ summary: 'Ingest events from a specific ledger' })
+  @ApiResponse({ status: 200, description: 'Events ingested successfully' })
+  async ingestLedger(@Body() dto: IngestLedgerDto) {
+    const results = await this.eventIngestionService.ingestSingleLedger(dto.contractId, dto.ledger);
+    return { processed: results.length, results };
+  }
+
+  @Post('reorg')
+  @HttpCode(HttpStatus.OK)
+  @ApiOperation({ summary: 'Handle chain reorg by reprocessing from a ledger' })
+  @ApiResponse({ status: 200, description: 'Reorg handled successfully' })
+  async handleReorg(@Body() dto: HandleReorgDto) {
+    await this.eventIngestionService.handleReorg(dto.contractId, dto.fromLedger);
+    return { message: 'Reorg handled', contractId: dto.contractId, fromLedger: dto.fromLedger };
+  }
+
+  @Get('status')
+  @ApiOperation({ summary: 'Get event ingestion status' })
+  @ApiResponse({ status: 200, description: 'Status retrieved successfully' })
+  async getStatus() {
+    return this.eventIngestionService.getStatus();
+  }
+
+  @Post('retry-failed')
+  @HttpCode(HttpStatus.OK)
+  @ApiOperation({ summary: 'Retry all failed events' })
+  @ApiResponse({ status: 200, description: 'Failed events retried' })
+  async retryFailedEvents() {
+    const results = await this.eventIngestionService.retryFailedEvents();
+    return { retried: results.length, results };
+  }
+}

backend/src/event-ingestion/event-ingestion.module.ts

@@ -0,0 +1,15 @@
+import { Module } from '@nestjs/common';
+import { EventIngestionService } from './event-ingestion.service';
+import { LedgerCursorService } from './ledger-cursor.service';
+import { EventProcessorService } from './event-processor.service';
+import { EventIngestionController } from './event-ingestion.controller';
+import { StellarModule } from '../stellar/stellar.module';
+import { EscrowModule } from '../escrow/escrow.module';
+
+@Module({
+  imports: [StellarModule, EscrowModule],
+  controllers: [EventIngestionController],
+  providers: [EventIngestionService, LedgerCursorService, EventProcessorService],
+  exports: [EventIngestionService],
+})
+export class EventIngestionModule {}

backend/src/event-ingestion/event-ingestion.service.spec.ts

@@ -0,0 +1,67 @@
+import { Test, TestingModule } from '@nestjs/testing';
+import { EventIngestionService } from './event-ingestion.service';
+import { LedgerCursorService } from './ledger-cursor.service';
+import { EventProcessorService } from './event-processor.service';
+import { EscrowService } from '../escrow/escrow.service';
+
+describe('EventIngestionService', () => {
+  let service: EventIngestionService;
+  let ledgerCursorService: LedgerCursorService;
+  let eventProcessorService: EventProcessorService;
+
+  const mockEscrowService = {
+    create: jest.fn(),
+    findById: jest.fn(),
+    release: jest.fn(),
+    raiseDispute: jest.fn(),
+  };
+
+  beforeEach(async () => {
+    const module: TestingModule = await Test.createTestingModule({
+      providers: [
+        EventIngestionService,
+        LedgerCursorService,
+        EventProcessorService,
+        { provide: EscrowService, useValue: mockEscrowService },
+      ],
+    }).compile();
+
+    service = module.get<EventIngestionService>(EventIngestionService);
+    ledgerCursorService = module.get<LedgerCursorService>(LedgerCursorService);
+    eventProcessorService = module.get<EventProcessorService>(EventProcessorService);
+  });
+
+  it('should be defined', () => {
+    expect(service).toBeDefined();
+  });
+
+  describe('getStatus', () => {
+    it('should return current status', async () => {
+      const status = await service.getStatus();
+      expect(status).toHaveProperty('isRunning');
+      expect(status).toHaveProperty('failedEvents');
+    });
+  });
+
+  describe('startPolling and stopPolling', () => {
+    it('should start and stop polling', async () => {
+      await service.startPolling('test-contract');
+      expect(service['isRunning']).toBe(true);
+
+      service.stopPolling();
+      expect(service['isRunning']).toBe(false);
+    });
+  });
+
+  describe('handleReorg', () => {
+    it('should handle reorg by clearing events and resetting cursor', async () => {
+      const clearSpy = jest.spyOn(eventProcessorService, 'clearEventsBeforeLedger');
+      const updateSpy = jest.spyOn(ledgerCursorService, 'updateCursor');
+
+      await service.handleReorg('test-contract', 100);
+
+      expect(clearSpy).toHaveBeenCalledWith(100);
+      expect(updateSpy).toHaveBeenCalledWith('test-contract', 99, 'reorg-100', '');
+    });
+  });
+});