From 481415d4a1daddec9cc3772f45e94b243a92fc1b Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 20 Jun 2026 11:58:41 +0000 Subject: [PATCH 1/2] Persist Stellar cursor in Postgres --- prisma/schema.prisma | 27 +++++ src/stellar/stellar-event.service.ts | 145 +++++++++++++++++++++++---- 2 files changed, 154 insertions(+), 18 deletions(-) diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 9024871..ab8b81d 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -447,3 +447,30 @@ model SmartContract { @@index([campaignId]) @@map("contracts") } + +/// EventCursor model for durable event listener cursors +model EventCursor { + id String @id @default(uuid()) + key String @unique + cursor String + networkPassphrase String + horizonUrl String + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@map("event_cursors") +} + +/// ProcessedContractEvent model for contract-event idempotency guards +model ProcessedContractEvent { + id String @id @default(uuid()) + contractId String + eventType String + txHash String + pagingToken String? + createdAt DateTime @default(now()) + + @@unique([txHash, eventType]) + @@index([contractId]) + @@map("processed_contract_events") +} diff --git a/src/stellar/stellar-event.service.ts b/src/stellar/stellar-event.service.ts index d663d67..a557e0f 100644 --- a/src/stellar/stellar-event.service.ts +++ b/src/stellar/stellar-event.service.ts @@ -1,13 +1,13 @@ -import { Injectable, Inject, OnApplicationBootstrap, Logger } from '@nestjs/common'; +import { Injectable, OnApplicationBootstrap, Logger } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import type { Queue } from 'bull'; import { InjectQueue } from '@nestjs/bull'; -import { CACHE_MANAGER } from '@nestjs/cache-manager'; -import type { Cache } from 'cache-manager'; import { Horizon, xdr, scValToNative, StrKey } from '@stellar/stellar-sdk'; import { PrismaService } from '../prisma/prisma.service'; import { QUEUE_CONTRACT_EVENTS } from '../queue/queue.constants'; +const STELLAR_CURSOR_KEY = 'stellar:event_listener:cursor'; + /** * Listens for Stellar blockchain events (payments, contract events) * and triggers donation processing workflows. @@ -18,6 +18,7 @@ export class StellarEventService implements OnApplicationBootstrap { private readonly logger = new Logger(StellarEventService.name); private readonly horizonUrl: string; private readonly horizonServer: Horizon.Server; + private readonly networkPassphrase: string; private streamCloseFn?: () => void; private lastCursor = 'now'; private isConnecting = false; @@ -28,29 +29,20 @@ export class StellarEventService implements OnApplicationBootstrap { private readonly prisma: PrismaService, @InjectQueue(QUEUE_CONTRACT_EVENTS) private readonly contractEventsQueue: Queue, - @Inject(CACHE_MANAGER) private readonly cacheManager: Cache, ) { this.horizonUrl = this.config.get('STELLAR_HORIZON_URL') || 'https://horizon-testnet.stellar.org'; + this.networkPassphrase = + this.config.get('STELLAR_NETWORK_PASSPHRASE') || + 'Test SDF Network ; September 2015'; this.horizonServer = new Horizon.Server(this.horizonUrl); } async onApplicationBootstrap() { this.logger.log('Starting Stellar Event Listener Service...'); - // Load last cursor from cache - const savedCursor = await this.cacheManager.get( - 'stellar:event_listener:cursor', - ); - if (savedCursor) { - this.lastCursor = savedCursor; - this.logger.log( - `Loaded last processed transaction cursor: ${this.lastCursor}`, - ); - } else { - this.logger.log('No saved cursor found. Starting from "now"'); - } + this.lastCursor = await this.loadStartupCursor(); // Catch up on any missed events and start the stream await this.catchUpAndStartStream(); @@ -186,7 +178,7 @@ export class StellarEventService implements OnApplicationBootstrap { `Found contract event [${eventType}] from contract ID ${event.contractId} in tx ${tx.hash}`, ); - await this.contractEventsQueue.add('process-event', { + await this.enqueueContractEvent({ contractId: event.contractId, eventType, topics: event.topics, @@ -211,7 +203,124 @@ export class StellarEventService implements OnApplicationBootstrap { private async saveCursor(cursor: string) { this.lastCursor = cursor; - await this.cacheManager.set('stellar:event_listener:cursor', cursor); + await this.prisma.eventCursor.upsert({ + where: { key: STELLAR_CURSOR_KEY }, + create: { + key: STELLAR_CURSOR_KEY, + cursor, + horizonUrl: this.horizonUrl, + networkPassphrase: this.networkPassphrase, + }, + update: { + cursor, + horizonUrl: this.horizonUrl, + networkPassphrase: this.networkPassphrase, + }, + }); + } + + private async loadStartupCursor(): Promise { + const savedCursor = await this.prisma.eventCursor.findUnique({ + where: { key: STELLAR_CURSOR_KEY }, + }); + + if (!savedCursor) { + this.logger.log('No saved cursor found in Postgres. Rolling forward to "now".'); + await this.saveCursor('now'); + return 'now'; + } + + const networkMatches = + savedCursor.networkPassphrase === this.networkPassphrase && + savedCursor.horizonUrl === this.horizonUrl; + if (!networkMatches) { + this.logger.warn( + `Saved cursor network mismatch (saved: ${savedCursor.horizonUrl}). Rolling forward to "now".`, + ); + await this.saveCursor('now'); + return 'now'; + } + + if (savedCursor.cursor === 'now') { + return 'now'; + } + + const cursorIsValid = await this.isCursorValid(savedCursor.cursor); + if (!cursorIsValid) { + this.logger.warn( + `Saved cursor ${savedCursor.cursor} is invalid for current network. Rolling forward to "now".`, + ); + await this.saveCursor('now'); + return 'now'; + } + + this.logger.log( + `Loaded last processed transaction cursor from Postgres: ${savedCursor.cursor}`, + ); + return savedCursor.cursor; + } + + private async isCursorValid(cursor: string): Promise { + try { + await this.horizonServer.transactions().cursor(cursor).limit(1).call(); + return true; + } catch (err) { + this.logger.warn(`Failed validating saved cursor ${cursor}: ${err.message}`); + return false; + } + } + + private async enqueueContractEvent(payload: { + contractId: string; + eventType: string; + topics: unknown[]; + value: unknown; + ledger: number | string; + txHash: string; + pagingToken: string; + createdAt: string; + }) { + const existing = await this.prisma.processedContractEvent.findUnique({ + where: { + txHash_eventType: { + txHash: payload.txHash, + eventType: payload.eventType, + }, + }, + select: { id: true }, + }); + if (existing) { + this.logger.debug( + `Skipping duplicate contract event [${payload.eventType}] for tx ${payload.txHash}`, + ); + return; + } + + await this.contractEventsQueue.add('process-event', payload, { + jobId: `${payload.txHash}:${payload.eventType}`, + }); + + try { + await this.prisma.processedContractEvent.create({ + data: { + contractId: payload.contractId, + eventType: payload.eventType, + txHash: payload.txHash, + pagingToken: payload.pagingToken, + }, + }); + } catch (err) { + if (this.isUniqueConstraintError(err)) { + return; + } + throw err; + } + } + + private isUniqueConstraintError(err: unknown): boolean { + if (!err || typeof err !== 'object') return false; + if (!('code' in err)) return false; + return String(err.code) === 'P2002'; } private parseEvents(resultMetaXdr: string): any[] { From 0fbf25cbdba8750e9c3c0b696749cfd2164791aa Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 20 Jun 2026 11:59:28 +0000 Subject: [PATCH 2/2] Add tests for Stellar cursor and idempotency guard --- src/stellar/stellar-event.service.spec.ts | 100 ++++++++++++++++++++++ 1 file changed, 100 insertions(+) create mode 100644 src/stellar/stellar-event.service.spec.ts diff --git a/src/stellar/stellar-event.service.spec.ts b/src/stellar/stellar-event.service.spec.ts new file mode 100644 index 0000000..867082e --- /dev/null +++ b/src/stellar/stellar-event.service.spec.ts @@ -0,0 +1,100 @@ +import { StellarEventService } from './stellar-event.service'; + +describe('StellarEventService', () => { + const createService = () => { + const config = { + get: jest.fn((key: string, fallback?: string) => { + if (key === 'STELLAR_HORIZON_URL') return 'https://horizon-testnet.stellar.org'; + if (key === 'STELLAR_NETWORK_PASSPHRASE') { + return 'Test SDF Network ; September 2015'; + } + return fallback; + }), + }; + + const prisma = { + eventCursor: { + findUnique: jest.fn(), + upsert: jest.fn(), + }, + processedContractEvent: { + findUnique: jest.fn(), + create: jest.fn(), + }, + smartContract: { + findMany: jest.fn(), + }, + }; + + const queue = { + add: jest.fn(), + }; + + const service = new StellarEventService( + config as never, + prisma as never, + queue as never, + ); + + (service as any).horizonServer = { + transactions: () => ({ + cursor: () => ({ + limit: () => ({ + call: jest.fn().mockResolvedValue({ records: [] }), + }), + }), + }), + }; + + return { service, prisma, queue }; + }; + + it('rolls forward to now when cursor is missing', async () => { + const { service, prisma } = createService(); + prisma.eventCursor.findUnique.mockResolvedValue(null); + + const cursor = await (service as any).loadStartupCursor(); + + expect(cursor).toBe('now'); + expect(prisma.eventCursor.upsert).toHaveBeenCalledWith( + expect.objectContaining({ + create: expect.objectContaining({ cursor: 'now' }), + update: expect.objectContaining({ cursor: 'now' }), + }), + ); + }); + + it('uses saved cursor when network matches and cursor validates', async () => { + const { service, prisma } = createService(); + prisma.eventCursor.findUnique.mockResolvedValue({ + key: 'stellar:event_listener:cursor', + cursor: '12345', + horizonUrl: 'https://horizon-testnet.stellar.org', + networkPassphrase: 'Test SDF Network ; September 2015', + }); + + const cursor = await (service as any).loadStartupCursor(); + + expect(cursor).toBe('12345'); + expect(prisma.eventCursor.upsert).not.toHaveBeenCalled(); + }); + + it('skips queueing duplicate contract events by txHash and eventType', async () => { + const { service, prisma, queue } = createService(); + prisma.processedContractEvent.findUnique.mockResolvedValue({ id: 'seen' }); + + await (service as any).enqueueContractEvent({ + contractId: 'C123', + eventType: 'DonationReceived', + topics: [], + value: {}, + ledger: 1, + txHash: 'tx1', + pagingToken: 'pt1', + createdAt: new Date().toISOString(), + }); + + expect(queue.add).not.toHaveBeenCalled(); + expect(prisma.processedContractEvent.create).not.toHaveBeenCalled(); + }); +});