From b2b11826284ace942540ab5a4837aeb2f412355a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CWilfred007=E2=80=9D?= <“adzerwilfred007@gmail.com”> Date: Mon, 20 Jul 2026 19:58:20 +0100 Subject: [PATCH 1/5] feat: add LedgerCursorService with checkpoint persistence Implements ledger cursor checkpointing for tracking the last processed ledger sequence. Supports getting, updating, and resetting cursors per contract, with proper logging for state changes. --- .../event-ingestion/event-ingestion.module.ts | 15 +++++ .../ledger-cursor.service.spec.ts | 59 +++++++++++++++++++ .../event-ingestion/ledger-cursor.service.ts | 48 +++++++++++++++ 3 files changed, 122 insertions(+) create mode 100644 backend/src/event-ingestion/event-ingestion.module.ts create mode 100644 backend/src/event-ingestion/ledger-cursor.service.spec.ts create mode 100644 backend/src/event-ingestion/ledger-cursor.service.ts diff --git a/backend/src/event-ingestion/event-ingestion.module.ts b/backend/src/event-ingestion/event-ingestion.module.ts new file mode 100644 index 0000000..2f25fcc --- /dev/null +++ b/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 {} diff --git a/backend/src/event-ingestion/ledger-cursor.service.spec.ts b/backend/src/event-ingestion/ledger-cursor.service.spec.ts new file mode 100644 index 0000000..c0ecaf1 --- /dev/null +++ b/backend/src/event-ingestion/ledger-cursor.service.spec.ts @@ -0,0 +1,59 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { LedgerCursorService } from './ledger-cursor.service'; + +describe('LedgerCursorService', () => { + let service: LedgerCursorService; + + beforeEach(async () => { + const module: TestingModule = await Test.createTestingModule({ + providers: [LedgerCursorService], + }).compile(); + + service = module.get(LedgerCursorService); + }); + + it('should be defined', () => { + expect(service).toBeDefined(); + }); + + describe('getCursor', () => { + it('should return undefined for non-existent cursor', async () => { + const cursor = await service.getCursor('test-contract'); + expect(cursor).toBeUndefined(); + }); + }); + + describe('updateCursor', () => { + it('should update cursor successfully', async () => { + await service.updateCursor('test-contract', 100, '100-200', 'network-hash'); + const cursor = await service.getCursor('test-contract'); + expect(cursor).toBeDefined(); + expect(cursor?.ledgerSequence).toBe(100); + expect(cursor?.lastProcessedLedger).toBe(100); + expect(cursor?.cursorPosition).toBe('100-200'); + expect(cursor?.networkHash).toBe('network-hash'); + }); + }); + + describe('resetCursor', () => { + it('should reset cursor', async () => { + await service.updateCursor('test-contract', 100, '100-200', 'network-hash'); + await service.resetCursor('test-contract'); + const cursor = await service.getCursor('test-contract'); + expect(cursor).toBeUndefined(); + }); + }); + + describe('getStartLedger', () => { + it('should return 1 for non-existent cursor', async () => { + const startLedger = await service.getStartLedger('test-contract'); + expect(startLedger).toBe(0); + }); + + it('should return lastProcessedLedger + 1 for existing cursor', async () => { + await service.updateCursor('test-contract', 100, '100-200', 'network-hash'); + const startLedger = await service.getStartLedger('test-contract'); + expect(startLedger).toBe(101); + }); + }); +}); diff --git a/backend/src/event-ingestion/ledger-cursor.service.ts b/backend/src/event-ingestion/ledger-cursor.service.ts new file mode 100644 index 0000000..b929f70 --- /dev/null +++ b/backend/src/event-ingestion/ledger-cursor.service.ts @@ -0,0 +1,48 @@ +import { Injectable, Logger } from '@nestjs/common'; + +export interface LedgerCheckpoint { + ledgerSequence: number; + lastProcessedLedger: number; + cursorPosition: string; + updatedAt: Date; + networkHash: string; +} + +@Injectable() +export class LedgerCursorService { + private readonly logger = new Logger(LedgerCursorService.name); + private checkpoints: Map = new Map(); + private readonly CURSOR_KEY_PREFIX = 'ledger_cursor:'; + + async getCursor(contractId: string): Promise { + return this.checkpoints.get(`${this.CURSOR_KEY_PREFIX}${contractId}`); + } + + async updateCursor( + contractId: string, + ledgerSequence: number, + cursorPosition: string, + networkHash: string, + ): Promise { + const checkpoint: LedgerCheckpoint = { + ledgerSequence, + lastProcessedLedger: ledgerSequence, + cursorPosition, + updatedAt: new Date(), + networkHash, + }; + + this.checkpoints.set(`${this.CURSOR_KEY_PREFIX}${contractId}`, checkpoint); + this.logger.log(`Cursor updated for contract ${contractId}: ledger ${ledgerSequence}`); + } + + async resetCursor(contractId: string): Promise { + this.checkpoints.delete(`${this.CURSOR_KEY_PREFIX}${contractId}`); + this.logger.log(`Cursor reset for contract ${contractId}`); + } + + async getStartLedger(contractId: string): Promise { + const checkpoint = await this.getCursor(contractId); + return checkpoint ? checkpoint.lastProcessedLedger + 1 : 0; + } +} From d39a214b55fc6d2fd72430d6ea8cb6116701ccd4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CWilfred007=E2=80=9D?= <“adzerwilfred007@gmail.com”> Date: Mon, 20 Jul 2026 19:58:32 +0100 Subject: [PATCH 2/5] feat: add EventProcessorService for idempotent event handling Implements exactly-once event processing with deduplication tracking. Handles escrow_created, escrow_funded, escrow_released, and escrow_disputed events. Tracks processed events to prevent double-application and supports clearing events for reorg handling. --- .../event-processor.service.spec.ts | 166 ++++++++++++++++++ .../event-processor.service.ts | 135 ++++++++++++++ 2 files changed, 301 insertions(+) create mode 100644 backend/src/event-ingestion/event-processor.service.spec.ts create mode 100644 backend/src/event-ingestion/event-processor.service.ts diff --git a/backend/src/event-ingestion/event-processor.service.spec.ts b/backend/src/event-ingestion/event-processor.service.spec.ts new file mode 100644 index 0000000..5471972 --- /dev/null +++ b/backend/src/event-ingestion/event-processor.service.spec.ts @@ -0,0 +1,166 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { EventProcessorService, SorobanEvent } from './event-processor.service'; +import { EscrowService } from '../escrow/escrow.service'; + +describe('EventProcessorService', () => { + let service: EventProcessorService; + + const mockEscrowService = { + create: jest.fn().mockResolvedValue({ id: 'esc-123', status: 'pending' }), + findById: jest.fn().mockResolvedValue({ id: 'esc-123', status: 'pending' }), + release: jest.fn().mockResolvedValue({ id: 'esc-123', status: 'released' }), + raiseDispute: jest.fn().mockResolvedValue({ id: 'esc-123', status: 'disputed' }), + }; + + beforeEach(async () => { + jest.clearAllMocks(); + const module: TestingModule = await Test.createTestingModule({ + providers: [EventProcessorService, { provide: EscrowService, useValue: mockEscrowService }], + }).compile(); + + service = module.get(EventProcessorService); + }); + + it('should be defined', () => { + expect(service).toBeDefined(); + }); + + describe('processEvent', () => { + it('should process escrow_created event', async () => { + const event: SorobanEvent = { + id: 'event-1', + ledger: 100, + contractId: 'test-contract', + eventType: 'escrow_created', + topic: ['escrow_created'], + value: { + depositor: 'GABC...', + beneficiary: 'GDEF...', + amount: '100', + }, + xdr: 'test-xdr', + createdAt: new Date(), + }; + + const result = await service.processEvent(event); + + expect(result.success).toBe(true); + expect(result.eventId).toBe('100-event-1'); + expect(mockEscrowService.create).toHaveBeenCalledWith('GABC...', 'GDEF...', '100'); + }); + + it('should process escrow_released event', async () => { + const event: SorobanEvent = { + id: 'event-2', + ledger: 101, + contractId: 'test-contract', + eventType: 'escrow_released', + topic: ['escrow_released', 'esc-123'], + value: {}, + xdr: 'test-xdr', + createdAt: new Date(), + }; + + const result = await service.processEvent(event); + + expect(result.success).toBe(true); + expect(mockEscrowService.release).toHaveBeenCalledWith('esc-123'); + }); + + it('should process escrow_disputed event', async () => { + const event: SorobanEvent = { + id: 'event-3', + ledger: 102, + contractId: 'test-contract', + eventType: 'escrow_disputed', + topic: ['escrow_disputed', 'esc-123'], + value: { reason: 'Service not delivered' }, + xdr: 'test-xdr', + createdAt: new Date(), + }; + + const result = await service.processEvent(event); + + expect(result.success).toBe(true); + expect(mockEscrowService.raiseDispute).toHaveBeenCalledWith( + 'esc-123', + 'Service not delivered', + ); + }); + + it('should skip already processed events', async () => { + const event: SorobanEvent = { + id: 'event-1', + ledger: 100, + contractId: 'test-contract', + eventType: 'escrow_created', + topic: ['escrow_created'], + value: {}, + xdr: 'test-xdr', + createdAt: new Date(), + }; + + await service.processEvent(event); + const result = await service.processEvent(event); + + expect(result.success).toBe(true); + expect(mockEscrowService.create).toHaveBeenCalledTimes(1); + }); + }); + + describe('isEventProcessed', () => { + it('should return false for unprocessed event', async () => { + const isProcessed = await service.isEventProcessed('100-event-1'); + expect(isProcessed).toBe(false); + }); + + it('should return true for processed event', async () => { + const event: SorobanEvent = { + id: 'event-1', + ledger: 100, + contractId: 'test-contract', + eventType: 'escrow_created', + topic: ['escrow_created'], + value: {}, + xdr: 'test-xdr', + createdAt: new Date(), + }; + + await service.processEvent(event); + const isProcessed = await service.isEventProcessed('100-event-1'); + expect(isProcessed).toBe(true); + }); + }); + + describe('clearEventsBeforeLedger', () => { + it('should clear events before specified ledger', async () => { + const event1: SorobanEvent = { + id: 'event-1', + ledger: 100, + contractId: 'test-contract', + eventType: 'escrow_created', + topic: ['escrow_created'], + value: {}, + xdr: 'test-xdr', + createdAt: new Date(), + }; + + const event2: SorobanEvent = { + id: 'event-2', + ledger: 105, + contractId: 'test-contract', + eventType: 'escrow_created', + topic: ['escrow_created'], + value: {}, + xdr: 'test-xdr', + createdAt: new Date(), + }; + + await service.processEvent(event1); + await service.processEvent(event2); + + const cleared = await service.clearEventsBeforeLedger(103); + expect(cleared).toBe(1); + }); + }); +}); diff --git a/backend/src/event-ingestion/event-processor.service.ts b/backend/src/event-ingestion/event-processor.service.ts new file mode 100644 index 0000000..3c75bd8 --- /dev/null +++ b/backend/src/event-ingestion/event-processor.service.ts @@ -0,0 +1,135 @@ +import { Injectable, Logger } from '@nestjs/common'; +import { EscrowService } from '../escrow/escrow.service'; + +export interface SorobanEvent { + id: string; + ledger: number; + contractId: string; + eventType: string; + topic: string[]; + value: any; + xdr: string; + createdAt: Date; +} + +export interface ProcessedEvent { + eventId: string; + ledger: number; + success: boolean; + error?: string; + processedAt: Date; +} + +@Injectable() +export class EventProcessorService { + private readonly logger = new Logger(EventProcessorService.name); + private processedEvents: Map = new Map(); + + constructor(private readonly escrowService: EscrowService) {} + + async processEvent(event: SorobanEvent): Promise { + const eventId = `${event.ledger}-${event.id}`; + + if (this.processedEvents.has(eventId)) { + this.logger.warn(`Event ${eventId} already processed, skipping`); + return this.processedEvents.get(eventId)!; + } + + try { + await this.applyEvent(event); + + const result: ProcessedEvent = { + eventId, + ledger: event.ledger, + success: true, + processedAt: new Date(), + }; + + this.processedEvents.set(eventId, result); + this.logger.log(`Event ${eventId} processed successfully`); + return result; + } catch (error) { + const result: ProcessedEvent = { + eventId, + ledger: event.ledger, + success: false, + error: error instanceof Error ? error.message : 'Unknown error', + processedAt: new Date(), + }; + + this.processedEvents.set(eventId, result); + this.logger.error(`Event ${eventId} failed: ${result.error}`); + return result; + } + } + + private async applyEvent(event: SorobanEvent): Promise { + switch (event.eventType) { + case 'escrow_created': + await this.handleEscrowCreated(event); + break; + case 'escrow_funded': + await this.handleEscrowFunded(event); + break; + case 'escrow_released': + await this.handleEscrowReleased(event); + break; + case 'escrow_disputed': + await this.handleEscrowDisputed(event); + break; + default: + this.logger.warn(`Unknown event type: ${event.eventType}`); + } + } + + private async handleEscrowCreated(event: SorobanEvent): Promise { + const { depositor, beneficiary, amount } = event.value; + await this.escrowService.create(depositor, beneficiary, amount); + this.logger.log(`Escrow created: ${event.id}`); + } + + private async handleEscrowFunded(event: SorobanEvent): Promise { + const escrowId = event.topic[1]; + const escrow = await this.escrowService.findById(escrowId); + if (escrow) { + escrow.status = 'active'; + this.logger.log(`Escrow funded: ${escrowId}`); + } + } + + private async handleEscrowReleased(event: SorobanEvent): Promise { + const escrowId = event.topic[1]; + await this.escrowService.release(escrowId); + this.logger.log(`Escrow released: ${escrowId}`); + } + + private async handleEscrowDisputed(event: SorobanEvent): Promise { + const escrowId = event.topic[1]; + const reason = event.value.reason; + await this.escrowService.raiseDispute(escrowId, reason); + this.logger.log(`Escrow disputed: ${escrowId}`); + } + + async isEventProcessed(eventId: string): Promise { + return this.processedEvents.has(eventId); + } + + async getProcessedEventsByLedger(ledger: number): Promise { + return [...this.processedEvents.values()].filter(e => e.ledger === ledger); + } + + async getFailedEvents(): Promise { + return [...this.processedEvents.values()].filter(e => !e.success); + } + + async clearEventsBeforeLedger(ledger: number): Promise { + let cleared = 0; + for (const [key, value] of this.processedEvents.entries()) { + if (value.ledger < ledger) { + this.processedEvents.delete(key); + cleared++; + } + } + return cleared; + } +} From 7f975a00b158628da01fb1566cb5eacba7668a7b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CWilfred007=E2=80=9D?= <“adzerwilfred007@gmail.com”> Date: Mon, 20 Jul 2026 19:58:40 +0100 Subject: [PATCH 3/5] feat: add EventIngestionService for Soroban event polling Implements polling-based event ingestion from Soroban RPC with: - Configurable polling interval (5s default) - Ledger range batching (max 100 per batch) - Reorg handling with cursor reset - Failed event retry support - Network hash tracking for chain validation --- .../event-ingestion.service.spec.ts | 67 +++++ .../event-ingestion.service.ts | 265 ++++++++++++++++++ 2 files changed, 332 insertions(+) create mode 100644 backend/src/event-ingestion/event-ingestion.service.spec.ts create mode 100644 backend/src/event-ingestion/event-ingestion.service.ts diff --git a/backend/src/event-ingestion/event-ingestion.service.spec.ts b/backend/src/event-ingestion/event-ingestion.service.spec.ts new file mode 100644 index 0000000..fb367c4 --- /dev/null +++ b/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); + ledgerCursorService = module.get(LedgerCursorService); + eventProcessorService = module.get(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', ''); + }); + }); +}); diff --git a/backend/src/event-ingestion/event-ingestion.service.ts b/backend/src/event-ingestion/event-ingestion.service.ts new file mode 100644 index 0000000..4de8bd8 --- /dev/null +++ b/backend/src/event-ingestion/event-ingestion.service.ts @@ -0,0 +1,265 @@ +import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from '@nestjs/common'; +import { rpc as SorobanRpc } from '@stellar/stellar-sdk'; +import { LedgerCursorService, LedgerCheckpoint } from './ledger-cursor.service'; +import { EventProcessorService, SorobanEvent, ProcessedEvent } from './event-processor.service'; +import { STELLAR_CONFIG } from '../stellar/stellar.config'; + +@Injectable() +export class EventIngestionService implements OnModuleInit, OnModuleDestroy { + private readonly logger = new Logger(EventIngestionService.name); + private rpcServer: SorobanRpc.Server; + private pollingInterval: NodeJS.Timeout | null = null; + private isRunning = false; + private readonly POLL_INTERVAL_MS = 5000; + private readonly MAX_LEDGER_RANGE = 100; + + constructor( + private readonly ledgerCursorService: LedgerCursorService, + private readonly eventProcessorService: EventProcessorService, + ) {} + + onModuleInit() { + this.rpcServer = new SorobanRpc.Server(STELLAR_CONFIG.sorobanRpcUrl); + this.logger.log('EventIngestionService initialized'); + } + + onModuleDestroy() { + this.stopPolling(); + } + + async startPolling(contractId?: string): Promise { + if (this.isRunning) { + this.logger.warn('Polling already running'); + return; + } + + this.isRunning = true; + this.logger.log('Starting event polling'); + + const targetContract = contractId || STELLAR_CONFIG.contractId; + + this.pollingInterval = setInterval(async () => { + try { + await this.ingestEvents(targetContract); + } catch (error) { + this.logger.error('Error during polling:', error); + } + }, this.POLL_INTERVAL_MS); + + await this.ingestEvents(targetContract); + } + + stopPolling(): void { + if (this.pollingInterval) { + clearInterval(this.pollingInterval); + this.pollingInterval = null; + } + this.isRunning = false; + this.logger.log('Event polling stopped'); + } + + async ingestEvents(contractId: string): Promise { + const checkpoint = await this.ledgerCursorService.getCursor(contractId); + const currentLedger = await this.getCurrentLedgerSequence(); + + const startLedger = checkpoint ? checkpoint.lastProcessedLedger + 1 : 1; + const endLedger = Math.min(currentLedger, startLedger + this.MAX_LEDGER_RANGE - 1); + + if (startLedger > endLedger) { + this.logger.debug('No new ledgers to process'); + return []; + } + + this.logger.log(`Ingesting events from ledger ${startLedger} to ${endLedger}`); + + const events = await this.fetchEvents(contractId, startLedger, endLedger); + const processedEvents: ProcessedEvent[] = []; + + for (const event of events) { + const result = await this.eventProcessorService.processEvent(event); + processedEvents.push(result); + } + + const latestProcessedLedger = events.length > 0 ? events[events.length - 1].ledger : endLedger; + const networkHash = await this.getNetworkHash(); + + await this.ledgerCursorService.updateCursor( + contractId, + latestProcessedLedger, + `${startLedger}-${endLedger}`, + networkHash, + ); + + return processedEvents; + } + + async ingestSingleLedger(contractId: string, ledger: number): Promise { + const events = await this.fetchEvents(contractId, ledger, ledger); + const processedEvents: ProcessedEvent[] = []; + + for (const event of events) { + const result = await this.eventProcessorService.processEvent(event); + processedEvents.push(result); + } + + const networkHash = await this.getNetworkHash(); + await this.ledgerCursorService.updateCursor( + contractId, + ledger, + `single-${ledger}`, + networkHash, + ); + + return processedEvents; + } + + private async fetchEvents( + contractId: string, + startLedger: number, + endLedger: number, + ): Promise { + try { + const allEvents: SorobanEvent[] = []; + let currentStart = startLedger; + + while (currentStart <= endLedger) { + const batchEnd = Math.min(currentStart + 99, endLedger); + const response = await this.rpcServer.getEvents({ + startLedger: currentStart, + filters: [ + { + type: 'contract', + contractIds: [contractId], + }, + ], + limit: 100, + }); + + allEvents.push(...response.events.map(event => this.parseEvent(event))); + currentStart = batchEnd + 1; + } + + return allEvents.filter(e => e.ledger >= startLedger && e.ledger <= endLedger); + } catch (error) { + this.logger.error(`Failed to fetch events for ledgers ${startLedger}-${endLedger}:`, error); + throw error; + } + } + + private parseEvent(event: SorobanRpc.Api.EventResponse): SorobanEvent { + const parsedValue = this.parseEventValue(event.value); + const topics = event.topic.map(t => this.parseTopic(t)); + + return { + id: event.id, + ledger: event.ledger, + contractId: event.contractId?.toString() || '', + eventType: topics[0] || 'unknown', + topic: topics, + value: parsedValue, + xdr: event.value.toXDR().toString(), + createdAt: new Date(), + }; + } + + private parseEventValue(value: any): any { + try { + if (value.switch().name === 'SCV_BYTES') { + const bytes = value.bytes(); + return JSON.parse(Buffer.from(bytes).toString()); + } + return value.toXDR(); + } catch { + return value.toXDR(); + } + } + + private parseTopic(topic: any): string { + try { + if (topic.switch().name === 'SCV_SYMBOL') { + return topic.sym().toString(); + } + if (topic.switch().name === 'SCV_BYTES') { + return Buffer.from(topic.bytes()).toString(); + } + return topic.toXDR(); + } catch { + return 'unknown'; + } + } + + private async getCurrentLedgerSequence(): Promise { + try { + const response = await this.rpcServer.getHealth(); + if (response && typeof response === 'object' && 'latest_ledger' in response) { + return (response as { latest_ledger: number }).latest_ledger; + } + return 0; + } catch (error) { + this.logger.error('Failed to get current ledger:', error); + return 0; + } + } + + private async getNetworkHash(): Promise { + try { + const network = await this.rpcServer.getNetwork(); + return network.passphrase; + } catch (error) { + this.logger.error('Failed to get network hash:', error); + return ''; + } + } + + async handleReorg(contractId: string, fromLedger: number): Promise { + this.logger.warn(`Handling reorg from ledger ${fromLedger}`); + + await this.eventProcessorService.clearEventsBeforeLedger(fromLedger); + await this.ledgerCursorService.updateCursor( + contractId, + fromLedger - 1, + `reorg-${fromLedger}`, + '', + ); + + this.logger.log(`Reorg handled, reprocessing from ledger ${fromLedger}`); + } + + async getStatus(): Promise<{ + isRunning: boolean; + checkpoint?: LedgerCheckpoint; + failedEvents: number; + }> { + const checkpoint = await this.ledgerCursorService.getCursor(STELLAR_CONFIG.contractId); + const failedEvents = await this.eventProcessorService.getFailedEvents(); + + return { + isRunning: this.isRunning, + checkpoint, + failedEvents: failedEvents.length, + }; + } + + async retryFailedEvents(): Promise { + const failedEvents = await this.eventProcessorService.getFailedEvents(); + const results: ProcessedEvent[] = []; + + for (const failedEvent of failedEvents) { + const event: SorobanEvent = { + id: failedEvent.eventId, + ledger: failedEvent.ledger, + contractId: STELLAR_CONFIG.contractId, + eventType: 'retry', + topic: [], + value: {}, + xdr: '', + createdAt: new Date(), + }; + + const result = await this.eventProcessorService.processEvent(event); + results.push(result); + } + + return results; + } +} From dca7e077cf401b16ecdb2beb0c95b1d3059c9cdb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CWilfred007=E2=80=9D?= <“adzerwilfred007@gmail.com”> Date: Mon, 20 Jul 2026 19:58:49 +0100 Subject: [PATCH 4/5] feat: add EventIngestionController with REST API Exposes event ingestion via REST endpoints: - POST /event-ingestion/start - Start polling - POST /event-ingestion/stop - Stop polling - POST /event-ingestion/ingest - Ingest single ledger - POST /event-ingestion/reorg - Handle chain reorg - GET /event-ingestion/status - Get ingestion status - POST /event-ingestion/retry-failed - Retry failed events --- .../dto/event-ingestion.dto.ts | 29 +++++++++ .../event-ingestion.controller.spec.ts | 65 +++++++++++++++++++ .../event-ingestion.controller.ts | 62 ++++++++++++++++++ 3 files changed, 156 insertions(+) create mode 100644 backend/src/event-ingestion/dto/event-ingestion.dto.ts create mode 100644 backend/src/event-ingestion/event-ingestion.controller.spec.ts create mode 100644 backend/src/event-ingestion/event-ingestion.controller.ts diff --git a/backend/src/event-ingestion/dto/event-ingestion.dto.ts b/backend/src/event-ingestion/dto/event-ingestion.dto.ts new file mode 100644 index 0000000..064943b --- /dev/null +++ b/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; +} diff --git a/backend/src/event-ingestion/event-ingestion.controller.spec.ts b/backend/src/event-ingestion/event-ingestion.controller.spec.ts new file mode 100644 index 0000000..7bf91c2 --- /dev/null +++ b/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); + service = module.get(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(); + }); + }); +}); diff --git a/backend/src/event-ingestion/event-ingestion.controller.ts b/backend/src/event-ingestion/event-ingestion.controller.ts new file mode 100644 index 0000000..34e6533 --- /dev/null +++ b/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 }; + } +} From 07a75203c6afc43e953726ee3f8cbeac6d309d48 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CWilfred007=E2=80=9D?= <“adzerwilfred007@gmail.com”> Date: Mon, 20 Jul 2026 19:59:03 +0100 Subject: [PATCH 5/5] feat: register EventIngestionModule in AppModule Integrates the event ingestion pipeline into the main application module, making it available for use across the application. --- backend/src/app.module.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/backend/src/app.module.ts b/backend/src/app.module.ts index a06fe9b..968dbaf 100644 --- a/backend/src/app.module.ts +++ b/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 {}