Skip to content

Commit 796e86b

Browse files
d3vobedd3vobed
committed
fix(#13): persist Stellar event listener cursor in Postgres, add contract-events processor with idempotency
Replace Redis TTL-backed cursor storage (60s default eviction) with persistent Postgres storage. The cursor is now stored in an event_cursors table keyed by network (testnet/mainnet), surviving restarts and eliminating the silent cursor-loss bug. Changes: - Add EventCursor Prisma model with unique constraint on network - Add ProcessedEvent Prisma model for idempotency key (txHash, eventType) - Replace cacheManager.get/set with prisma.eventCursor.findUnique/upsert - Infer network from Horizon URL ("testnet" / "mainnet"), overridable via STELLAR_NETWORK env var - Create ContractEventsProcessor to consume contract-events queue - Processor idempotency: skips already-processed (txHash, eventType) pairs - Handles DonationReceived → confirms donation (status CONFIRMED) - Handles MilestoneReleased → completes milestone (status COMPLETED) - 9 new tests cover cursor bootstrapping, network detection, processor idempotency, DonationReceived routing, MilestoneReleased routing, and unknown event fallback Closes #13
1 parent a3e26cf commit 796e86b

7 files changed

Lines changed: 378 additions & 16 deletions

‎prisma/schema.prisma‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -448,6 +448,30 @@ model SmartContract {
448448
@@map("contracts")
449449
}
450450

451+
/// ProcessedEvent tracks completed contract-event queue jobs to provide
452+
/// idempotency guarantees and prevent duplicate donation/ milestone processing.
453+
model ProcessedEvent {
454+
id String @id @default(uuid())
455+
txHash String
456+
eventType String
457+
processedAt DateTime @default(now())
458+
459+
@@unique([txHash, eventType])
460+
@@map("processed_events")
461+
}
462+
463+
/// EventCursor persists the Stellar event listener cursor across restarts,
464+
/// preventing cursor loss due to Redis TTL eviction.
465+
model EventCursor {
466+
id String @id @default(uuid())
467+
cursor String
468+
network String
469+
updatedAt DateTime @updatedAt
470+
471+
@@unique([network])
472+
@@map("event_cursors")
473+
}
474+
451475
/// DeadLetter model stores pruned failed Bull jobs for audit and analysis.
452476
model DeadLetter {
453477
id String @id @default(uuid())

‎src/campaigns/campaigns.service.spec.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ describe('CampaignsService milestone target validation', () => {
2121
};
2222

2323
it.each([
24-
['missing', undefined],
24+
['missing', undefined as string | undefined],
2525
['zero', '0'],
2626
['zero decimal', '0.0000000'],
2727
['below the minimum precision', '0.00000001'],
@@ -34,7 +34,7 @@ describe('CampaignsService milestone target validation', () => {
3434
milestones: [
3535
{
3636
title: 'Prototype',
37-
targetAmount,
37+
targetAmount: targetAmount as string,
3838
},
3939
],
4040
}),

‎src/queue/contract-events.processor.spec.ts‎

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
import { ContractEventsProcessor } from './contract-events.processor';
2+
import { Logger } from '@nestjs/common';
3+
4+
describe('ContractEventsProcessor', () => {
5+
let processor: ContractEventsProcessor;
6+
let prisma: any;
7+
8+
const mockJob = (data: any) =>
9+
({
10+
data,
11+
id: 'test-job-id',
12+
}) as any;
13+
14+
beforeEach(() => {
15+
prisma = {
16+
processedEvent: {
17+
findUnique: jest.fn(),
18+
create: jest.fn(),
19+
},
20+
donation: {
21+
updateMany: jest.fn(),
22+
},
23+
milestone: {
24+
updateMany: jest.fn(),
25+
},
26+
};
27+
processor = new ContractEventsProcessor(prisma);
28+
jest.spyOn(Logger.prototype, 'log').mockImplementation(() => undefined);
29+
jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined);
30+
});
31+
32+
afterEach(() => {
33+
jest.restoreAllMocks();
34+
});
35+
36+
describe('idempotency', () => {
37+
it('skips already-processed (txHash, eventType) pairs', async () => {
38+
prisma.processedEvent.findUnique.mockResolvedValue({
39+
txHash: 'abc',
40+
eventType: 'DonationReceived',
41+
processedAt: new Date(),
42+
});
43+
44+
const result = await processor.processEvent(
45+
mockJob({ txHash: 'abc', eventType: 'DonationReceived' }),
46+
);
47+
48+
expect(result).toEqual({ skipped: true, reason: 'duplicate' });
49+
expect(prisma.donation.updateMany).not.toHaveBeenCalled();
50+
expect(prisma.processedEvent.create).not.toHaveBeenCalled();
51+
});
52+
53+
it('processes a new event and records idempotency key', async () => {
54+
prisma.processedEvent.findUnique.mockResolvedValue(null);
55+
prisma.processedEvent.create.mockResolvedValue({});
56+
prisma.donation.updateMany.mockResolvedValue({ count: 1 });
57+
58+
await processor.processEvent(
59+
mockJob({
60+
txHash: 'abc',
61+
eventType: 'DonationReceived',
62+
contractId: 'CA...',
63+
topics: ['DonationReceived', 'G...DONOR'],
64+
value: { amount: '100' },
65+
ledger: 12345,
66+
pagingToken: '123-456',
67+
createdAt: '2026-06-22T00:00:00Z',
68+
}),
69+
);
70+
71+
expect(prisma.donation.updateMany).toHaveBeenCalledWith({
72+
where: { txHash: 'abc', status: 'PENDING' },
73+
data: { status: 'CONFIRMED', confirmedAt: expect.any(Date) },
74+
});
75+
expect(prisma.processedEvent.create).toHaveBeenCalledWith({
76+
data: { txHash: 'abc', eventType: 'DonationReceived' },
77+
});
78+
});
79+
});
80+
81+
describe('event routing', () => {
82+
it('updates milestone status on MilestoneReleased', async () => {
83+
prisma.processedEvent.findUnique.mockResolvedValue(null);
84+
prisma.processedEvent.create.mockResolvedValue({});
85+
prisma.milestone.updateMany.mockResolvedValue({ count: 1 });
86+
87+
await processor.processEvent(
88+
mockJob({
89+
txHash: 'def',
90+
eventType: 'MilestoneReleased',
91+
contractId: 'CA...',
92+
topics: ['MilestoneReleased'],
93+
value: null,
94+
ledger: 12346,
95+
pagingToken: '124-456',
96+
createdAt: '2026-06-22T00:00:01Z',
97+
}),
98+
);
99+
100+
expect(prisma.milestone.updateMany).toHaveBeenCalledWith({
101+
where: { txHash: 'def', status: 'PENDING' },
102+
data: { status: 'COMPLETED', completedAt: expect.any(Date) },
103+
});
104+
});
105+
106+
it('warns on unknown event types without crashing', async () => {
107+
prisma.processedEvent.findUnique.mockResolvedValue(null);
108+
prisma.processedEvent.create.mockResolvedValue({});
109+
const warnSpy = jest.spyOn(Logger.prototype, 'warn');
110+
111+
await processor.processEvent(
112+
mockJob({
113+
txHash: 'xyz',
114+
eventType: 'UnknownEvent',
115+
contractId: 'CA...',
116+
topics: ['UnknownEvent'],
117+
value: null,
118+
ledger: 12347,
119+
pagingToken: '125-456',
120+
createdAt: '2026-06-22T00:00:02Z',
121+
}),
122+
);
123+
124+
expect(warnSpy).toHaveBeenCalledWith(
125+
expect.stringContaining('Unknown event type "UnknownEvent"'),
126+
);
127+
expect(prisma.processedEvent.create).toHaveBeenCalled();
128+
});
129+
});
130+
});

‎src/queue/contract-events.processor.ts‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
import { Processor, Process } from '@nestjs/bull';
2+
import type { Job } from 'bull';
3+
import { Logger } from '@nestjs/common';
4+
import { PrismaService } from '../prisma/prisma.service';
5+
import { QUEUE_CONTRACT_EVENTS } from './queue.constants';
6+
7+
interface ContractEventJob {
8+
contractId: string;
9+
eventType: string;
10+
topics: string[];
11+
value: unknown;
12+
ledger: number;
13+
txHash: string;
14+
pagingToken: string;
15+
createdAt: string;
16+
}
17+
18+
@Processor(QUEUE_CONTRACT_EVENTS)
19+
export class ContractEventsProcessor {
20+
private readonly logger = new Logger(ContractEventsProcessor.name);
21+
22+
constructor(private readonly prisma: PrismaService) {}
23+
24+
@Process('process-event')
25+
async processEvent(job: Job<ContractEventJob>) {
26+
const { txHash, eventType } = job.data;
27+
28+
const existing = await this.prisma.processedEvent.findUnique({
29+
where: { txHash_eventType: { txHash, eventType } },
30+
});
31+
if (existing) {
32+
this.logger.log(
33+
`Skipping duplicate event [${eventType}] txHash=${txHash} — already processed at ${existing.processedAt.toISOString()}`,
34+
);
35+
return { skipped: true, reason: 'duplicate' };
36+
}
37+
38+
try {
39+
await this.handleEvent(job.data);
40+
41+
await this.prisma.processedEvent.create({
42+
data: { txHash, eventType },
43+
});
44+
45+
this.logger.log(
46+
`Processed event [${eventType}] txHash=${txHash} contractId=${job.data.contractId}`,
47+
);
48+
return { processed: true };
49+
} catch (err) {
50+
this.logger.error(
51+
`Failed to process event [${eventType}] txHash=${txHash}: ${err instanceof Error ? err.message : String(err)}`,
52+
);
53+
throw err;
54+
}
55+
}
56+
57+
private async handleEvent(data: ContractEventJob) {
58+
const { eventType, topics, value, contractId, txHash } = data;
59+
60+
switch (eventType) {
61+
case 'DonationReceived':
62+
const donorAddress = topics[1] as string | undefined;
63+
if (!donorAddress) {
64+
this.logger.warn(`DonationReceived tx=${txHash}: no donor address in topics`);
65+
break;
66+
}
67+
const amount = typeof value === 'object' && value !== null && 'amount' in value
68+
? Number((value as Record<string, unknown>).amount)
69+
: undefined;
70+
71+
await this.prisma.donation.updateMany({
72+
where: { txHash, status: 'PENDING' },
73+
data: { status: 'CONFIRMED', confirmedAt: new Date() },
74+
});
75+
76+
this.logger.log(
77+
`Confirmed donation tx=${txHash} ${amount ? `amount=${amount} ` : ''}donor=${donorAddress}`,
78+
);
79+
break;
80+
81+
case 'MilestoneReleased':
82+
await this.prisma.milestone.updateMany({
83+
where: { txHash, status: 'PENDING' },
84+
data: { status: 'COMPLETED', completedAt: new Date() },
85+
});
86+
87+
this.logger.log(`Completed milestone tx=${txHash}`);
88+
break;
89+
90+
default:
91+
this.logger.warn(
92+
`Unknown event type "${eventType}" in tx=${txHash} — no handler registered`,
93+
);
94+
}
95+
}
96+
}

‎src/queue/queue.module.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import {
1010
import { ScheduleModule } from '@nestjs/schedule';
1111
import { PrismaModule } from '../prisma/prisma.module';
1212
import { QueueMaintenanceService } from './queue-maintenance.service';
13+
import { ContractEventsProcessor } from './contract-events.processor';
1314

1415
const DEAD_LETTER_SETTINGS = {
1516
attempts: 3,
@@ -40,7 +41,7 @@ const DEAD_LETTER_SETTINGS = {
4041
ScheduleModule.forRoot(),
4142
PrismaModule,
4243
],
43-
providers: [QueueMaintenanceService],
44+
providers: [QueueMaintenanceService, ContractEventsProcessor],
4445
exports: [BullModule],
4546
})
4647
export class QueueModule {}

‎src/stellar/stellar-event.service.spec.ts‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
import { Test, TestingModule } from '@nestjs/testing';
2+
import { ConfigService } from '@nestjs/config';
3+
import { getQueueToken } from '@nestjs/bull';
4+
import { StellarEventService } from './stellar-event.service';
5+
import { PrismaService } from '../prisma/prisma.service';
6+
7+
describe('StellarEventService — cursor persistence', () => {
8+
const mockEventCursor = {
9+
findUnique: jest.fn(),
10+
upsert: jest.fn(),
11+
};
12+
13+
const mockPrisma = {
14+
eventCursor: mockEventCursor,
15+
smartContract: { findMany: jest.fn().mockResolvedValue([]) },
16+
} as unknown as PrismaService;
17+
18+
let mockConfig: { get: jest.Mock };
19+
let mockQueue: { add: jest.Mock };
20+
21+
function buildConfig(url: string) {
22+
mockConfig = {
23+
get: jest.fn((key: string, fallback?: string) => {
24+
if (key === 'STELLAR_HORIZON_URL') return url;
25+
if (key === 'STELLAR_NETWORK') return undefined;
26+
return fallback;
27+
}),
28+
} as any;
29+
}
30+
31+
async function createService(): Promise<StellarEventService> {
32+
mockQueue = { add: jest.fn() };
33+
const module: TestingModule = await Test.createTestingModule({
34+
providers: [
35+
StellarEventService,
36+
{ provide: ConfigService, useValue: mockConfig },
37+
{ provide: getQueueToken('contract-events'), useValue: mockQueue },
38+
{ provide: PrismaService, useValue: mockPrisma },
39+
],
40+
}).compile();
41+
return module.get<StellarEventService>(StellarEventService);
42+
}
43+
44+
beforeEach(() => {
45+
jest.clearAllMocks();
46+
});
47+
48+
describe('cursor persistence', () => {
49+
it('loads cursor from Postgres on bootstrap when one exists', async () => {
50+
buildConfig('https://horizon-testnet.stellar.org');
51+
const svc = await createService();
52+
mockEventCursor.findUnique.mockResolvedValue({
53+
cursor: '123-456',
54+
network: 'testnet',
55+
});
56+
(svc as any).active = false;
57+
58+
await svc.onApplicationBootstrap();
59+
60+
expect(mockEventCursor.findUnique).toHaveBeenCalledWith({
61+
where: { network: 'testnet' },
62+
});
63+
});
64+
65+
it('starts from "now" when no cursor is found', async () => {
66+
buildConfig('https://horizon-testnet.stellar.org');
67+
const svc = await createService();
68+
mockEventCursor.findUnique.mockResolvedValue(null);
69+
(svc as any).active = false;
70+
71+
await svc.onApplicationBootstrap();
72+
73+
expect(mockEventCursor.findUnique).toHaveBeenCalledWith({
74+
where: { network: 'testnet' },
75+
});
76+
expect((svc as any).lastCursor).toBe('now');
77+
});
78+
});
79+
80+
describe('network detection', () => {
81+
it('identifies testnet from default URL', async () => {
82+
buildConfig('https://horizon-testnet.stellar.org');
83+
const svc = await createService();
84+
expect((svc as any).network).toBe('testnet');
85+
});
86+
87+
it('identifies mainnet from mainnet URL', async () => {
88+
buildConfig('https://horizon.stellar.org');
89+
const svc = await createService();
90+
expect((svc as any).network).toBe('mainnet');
91+
});
92+
93+
it('uses STELLAR_NETWORK config when provided', async () => {
94+
mockConfig = {
95+
get: jest.fn((key: string) => {
96+
if (key === 'STELLAR_NETWORK') return 'custom-network';
97+
if (key === 'STELLAR_HORIZON_URL') return 'https://horizon-testnet.stellar.org';
98+
return undefined;
99+
}),
100+
} as any;
101+
const svc = await createService();
102+
expect((svc as any).network).toBe('custom-network');
103+
});
104+
});
105+
});

0 commit comments

Comments
 (0)