Skip to content

Commit ed9c7e3

Browse files
aji70aji70
authored
Merge pull request #798 from Mimah97/feat/soroban-rpc-ledger-events
feat(backend): add getLatestLedgerSequence & getNetworkEvents to Soro…
2 parents cb17c92 + 36ed0db commit ed9c7e3

7 files changed

Lines changed: 269 additions & 9 deletions

File tree

‎backend/.env.example‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,3 +104,5 @@ CORS_ALLOWED_HOSTS=myfans.example.com,www.myfans.example.com
104104
FEATURE_NEW_SUBSCRIPTION_FLOW=false
105105
FEATURE_CRYPTO_PAYMENTS=false
106106
FEATURE_REFERRAL_CODES=false
107+
# Set to "false" to disable the Soroban event poller (enabled by default).
108+
FEATURE_SOROBAN_POLLER=true

‎backend/src/common/services/soroban-rpc-ledger-events.spec.ts‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
/**
2+
* Unit tests for SorobanRpcService.getLatestLedgerSequence
3+
* and SorobanRpcService.getNetworkEvents
4+
*/
5+
import { SorobanRpcService } from './soroban-rpc.service';
6+
7+
function makeService(serverOverrides: Record<string, jest.Mock> = {}): SorobanRpcService {
8+
const svc = new SorobanRpcService();
9+
(svc as any).server = {
10+
getHealth: jest.fn().mockResolvedValue({ status: 'healthy', ledger: 1000 }),
11+
getEvents: jest.fn().mockResolvedValue({ events: [], latestLedger: 1000 }),
12+
...serverOverrides,
13+
};
14+
return svc;
15+
}
16+
17+
describe('SorobanRpcService – getLatestLedgerSequence', () => {
18+
it('returns the ledger number from getHealth', async () => {
19+
const svc = makeService({ getHealth: jest.fn().mockResolvedValue({ status: 'healthy', ledger: 42 }) });
20+
await expect(svc.getLatestLedgerSequence()).resolves.toBe(42);
21+
});
22+
23+
it('throws when server is null', async () => {
24+
const svc = makeService();
25+
(svc as any).server = null;
26+
await expect(svc.getLatestLedgerSequence()).rejects.toThrow('server not initialized');
27+
});
28+
29+
it('throws when getHealth rejects', async () => {
30+
const svc = makeService({ getHealth: jest.fn().mockRejectedValue(new Error('network error')) });
31+
await expect(svc.getLatestLedgerSequence()).rejects.toThrow('network error');
32+
});
33+
34+
it('throws when ledger field is missing or zero', async () => {
35+
const svc = makeService({ getHealth: jest.fn().mockResolvedValue({ status: 'healthy' }) });
36+
await expect(svc.getLatestLedgerSequence()).rejects.toThrow('invalid ledger sequence');
37+
});
38+
});
39+
40+
describe('SorobanRpcService – getNetworkEvents', () => {
41+
it('returns events and latestLedger from getEvents', async () => {
42+
const fakeEvents = [{ id: '100:0', topic: [], value: {} }];
43+
const svc = makeService({
44+
getEvents: jest.fn().mockResolvedValue({ events: fakeEvents, latestLedger: 200 }),
45+
});
46+
const result = await svc.getNetworkEvents({ startLedger: 100 });
47+
expect(result.events).toEqual(fakeEvents);
48+
expect(result.latestLedger).toBe(200);
49+
expect(result.nextToken).toBeUndefined();
50+
});
51+
52+
it('passes limit and cursor to getEvents', async () => {
53+
const getEvents = jest.fn().mockResolvedValue({ events: [], latestLedger: 300 });
54+
const svc = makeService({ getEvents });
55+
await svc.getNetworkEvents({ startLedger: 50, limit: 10, paginationToken: 'tok' });
56+
expect(getEvents).toHaveBeenCalledWith(
57+
expect.objectContaining({ startLedger: 50, limit: 10, cursor: 'tok' }),
58+
);
59+
});
60+
61+
it('returns empty events array when getEvents returns undefined events', async () => {
62+
const svc = makeService({ getEvents: jest.fn().mockResolvedValue({ latestLedger: 100 }) });
63+
const result = await svc.getNetworkEvents({ startLedger: 1 });
64+
expect(result.events).toEqual([]);
65+
});
66+
67+
it('throws when server is null', async () => {
68+
const svc = makeService();
69+
(svc as any).server = null;
70+
await expect(svc.getNetworkEvents({ startLedger: 1 })).rejects.toThrow('server not initialized');
71+
});
72+
73+
it('throws when getEvents rejects', async () => {
74+
const svc = makeService({ getEvents: jest.fn().mockRejectedValue(new Error('rpc down')) });
75+
await expect(svc.getNetworkEvents({ startLedger: 1 })).rejects.toThrow('rpc down');
76+
});
77+
});

‎backend/src/common/services/soroban-rpc.service.ts‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -418,4 +418,65 @@ export class SorobanRpcService {
418418
getRetryConfig(): RetryConfig {
419419
return { ...this.retryConfig };
420420
}
421+
422+
/**
423+
* Returns the latest ledger sequence number from the Soroban RPC node.
424+
* Throws on network failure so callers can decide how to handle stale state.
425+
*/
426+
async getLatestLedgerSequence(): Promise<number> {
427+
if (!this.server) {
428+
throw new Error('SorobanRpcService: server not initialized');
429+
}
430+
try {
431+
const health = await (this.server as rpc.Server).getHealth();
432+
const seq = (health as rpc.Api.GetHealthResponse & { ledger?: number }).ledger;
433+
if (typeof seq !== 'number' || seq <= 0) {
434+
throw new Error('SorobanRpcService: invalid ledger sequence in health response');
435+
}
436+
return seq;
437+
} catch (err) {
438+
this.logger.error(`getLatestLedgerSequence failed: ${err}`);
439+
throw err;
440+
}
441+
}
442+
443+
/**
444+
* Fetches contract events from the Soroban RPC node.
445+
*
446+
* @param startLedger First ledger to include (inclusive).
447+
* @param limit Max events per page (default 200, max 10 000).
448+
* @param paginationToken Opaque cursor returned by a previous call.
449+
*/
450+
async getNetworkEvents(opts: {
451+
startLedger: number;
452+
limit?: number;
453+
paginationToken?: string;
454+
}): Promise<{
455+
events: rpc.Api.EventResponse[];
456+
startLedger: number;
457+
latestLedger: number;
458+
nextToken?: string;
459+
}> {
460+
if (!this.server) {
461+
throw new Error('SorobanRpcService: server not initialized');
462+
}
463+
const { startLedger, limit = 200, paginationToken } = opts;
464+
try {
465+
const response = await (this.server as rpc.Server).getEvents({
466+
startLedger,
467+
filters: [],
468+
limit,
469+
...(paginationToken ? { cursor: paginationToken } : {}),
470+
});
471+
return {
472+
events: response.events ?? [],
473+
startLedger: response.latestLedger, // Soroban SDK field
474+
latestLedger: response.latestLedger,
475+
nextToken: (response as any).cursor ?? undefined,
476+
};
477+
} catch (err) {
478+
this.logger.error(`getNetworkEvents failed (startLedger=${startLedger}): ${err}`);
479+
throw err;
480+
}
481+
}
421482
}

‎backend/src/feature-flags/feature-flags.service.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,11 +14,16 @@ export class FeatureFlagsService {
1414
return process.env.FEATURE_REFERRAL_CODES === 'true';
1515
}
1616

17+
isSorobanPollerEnabled(): boolean {
18+
return process.env.FEATURE_SOROBAN_POLLER !== 'false';
19+
}
20+
1721
getAllFlags() {
1822
return {
1923
newSubscriptionFlow: this.isNewSubscriptionFlowEnabled(),
2024
cryptoPayments: this.isCryptoPaymentsEnabled(),
2125
referralCodes: this.isReferralCodesEnabled(),
26+
sorobanPoller: this.isSorobanPollerEnabled(),
2227
};
2328
}
2429
}

‎backend/src/subscriptions/services/subscription-event-poller-ledger.spec.ts‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
/**
2+
* Tests for SubscriptionEventPollerService:
3+
* - feature flag disables polling
4+
* - stale / disconnected RPC is handled gracefully (no throw)
5+
* - typed methods are called (no any-cast)
6+
*/
7+
import { SubscriptionEventPollerService } from './subscription-event-poller.service';
8+
import { RequestContextService } from '../../common/services/request-context.service';
9+
10+
function makePoller(overrides: {
11+
pollerEnabled?: boolean;
12+
getLatestLedgerSequence?: () => Promise<number>;
13+
getNetworkEvents?: () => Promise<any>;
14+
checkpoint?: number;
15+
}) {
16+
const requestContext = new RequestContextService();
17+
18+
const featureFlags = {
19+
isSorobanPollerEnabled: jest.fn().mockReturnValue(overrides.pollerEnabled ?? true),
20+
};
21+
22+
const indexRepo = {
23+
getLatestCheckpoint: jest.fn().mockResolvedValue(overrides.checkpoint ?? 0),
24+
findByEventId: jest.fn().mockResolvedValue(null),
25+
upsertEvent: jest.fn().mockResolvedValue({ id: '1', eventType: 'subscribed', fan: 'A', creator: 'B', planId: 0, expiryUnix: 9999999999 }),
26+
};
27+
28+
const sorobanRpc = {
29+
getLatestLedgerSequence: jest.fn().mockImplementation(
30+
overrides.getLatestLedgerSequence ?? (() => Promise.resolve(0)),
31+
),
32+
getNetworkEvents: jest.fn().mockImplementation(
33+
overrides.getNetworkEvents ?? (() => Promise.resolve({ events: [], startLedger: 0, latestLedger: 0 })),
34+
),
35+
};
36+
37+
const eventBus = { publish: jest.fn() };
38+
39+
const svc = new (SubscriptionEventPollerService as any)(
40+
{ get: () => 'CONTRACT_ID' },
41+
indexRepo,
42+
eventBus,
43+
sorobanRpc,
44+
requestContext,
45+
featureFlags,
46+
) as SubscriptionEventPollerService;
47+
48+
(svc as any).contractId = 'CONTRACT_ID';
49+
50+
return { svc, sorobanRpc, indexRepo, featureFlags, eventBus };
51+
}
52+
53+
describe('SubscriptionEventPollerService – feature flag', () => {
54+
it('skips poll when isSorobanPollerEnabled returns false', async () => {
55+
const { svc, sorobanRpc } = makePoller({ pollerEnabled: false });
56+
await svc.poll();
57+
expect(sorobanRpc.getLatestLedgerSequence).not.toHaveBeenCalled();
58+
});
59+
60+
it('proceeds when isSorobanPollerEnabled returns true', async () => {
61+
const { svc, sorobanRpc } = makePoller({ pollerEnabled: true, checkpoint: 5 });
62+
// latest == checkpoint → early return, but RPC was still called
63+
sorobanRpc.getLatestLedgerSequence.mockResolvedValue(5);
64+
await svc.poll();
65+
expect(sorobanRpc.getLatestLedgerSequence).toHaveBeenCalledTimes(1);
66+
});
67+
});
68+
69+
describe('SubscriptionEventPollerService – stale / disconnected RPC', () => {
70+
it('does not throw when getLatestLedgerSequence rejects', async () => {
71+
const { svc, sorobanRpc } = makePoller({
72+
getLatestLedgerSequence: () => Promise.reject(new Error('connection refused')),
73+
});
74+
await expect(svc.poll()).resolves.toBeUndefined();
75+
expect(sorobanRpc.getNetworkEvents).not.toHaveBeenCalled();
76+
});
77+
78+
it('does not throw when getNetworkEvents rejects mid-page', async () => {
79+
const { svc, sorobanRpc } = makePoller({
80+
getLatestLedgerSequence: () => Promise.resolve(100),
81+
getNetworkEvents: () => Promise.reject(new Error('rpc timeout')),
82+
checkpoint: 50,
83+
});
84+
await expect(svc.poll()).resolves.toBeUndefined();
85+
});
86+
87+
it('uses typed getLatestLedgerSequence (not any-cast)', async () => {
88+
const { svc, sorobanRpc } = makePoller({ checkpoint: 10 });
89+
sorobanRpc.getLatestLedgerSequence.mockResolvedValue(10); // no new ledgers
90+
await svc.poll();
91+
// Verify the real method was called, not a dynamic property
92+
expect(sorobanRpc.getLatestLedgerSequence).toHaveBeenCalledTimes(1);
93+
});
94+
});

‎backend/src/subscriptions/services/subscription-event-poller.service.ts‎

Lines changed: 27 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,9 @@ import { v4 as uuidv4 } from 'uuid';
1414
import { resolveSubscriptionContractId } from '../../common/contract-deployed-env';
1515
import { SubscriptionIndexEntity, SubscriptionStatus } from '../entities/subscription-index.entity';
1616
import { SubscriptionIndexRepository, UpsertEventData } from '../repositories/subscription-index.repository';
17-
import { SorobanRpcService } from '../../common/services/soroban-rpc.service'; // Assumed to exist
17+
import { SorobanRpcService } from '../../common/services/soroban-rpc.service';
1818
import { RequestContextService } from '../../common/services/request-context.service';
19+
import { FeatureFlagsService } from '../../feature-flags/feature-flags.service';
1920

2021
const TARGET_EVENTS = ['subscribed', 'extended', 'cancelled'] as const;
2122
type TargetEventType = typeof TARGET_EVENTS[number];
@@ -31,6 +32,7 @@ export class SubscriptionEventPollerService implements OnModuleInit {
3132
private readonly eventBus: EventBus,
3233
private readonly sorobanRpc: SorobanRpcService,
3334
private readonly requestContext: RequestContextService,
35+
private readonly featureFlags: FeatureFlagsService,
3436
) {}
3537

3638
async onModuleInit() {
@@ -68,14 +70,25 @@ export class SubscriptionEventPollerService implements OnModuleInit {
6870
}
6971

7072
private async _poll(): Promise<void> {
73+
if (!this.featureFlags.isSorobanPollerEnabled()) {
74+
this.logger.debug('Soroban poller disabled via feature flag; skipping.');
75+
return;
76+
}
77+
7178
const startTime = Date.now();
7279
let processed = 0;
7380
let errors = 0;
7481

7582
try {
7683
const checkpoint = await this.indexRepo.getLatestCheckpoint();
77-
const latestLedger = await (this.sorobanRpc as any).getLatestLedgerSequence();
78-
84+
let latestLedger: number;
85+
try {
86+
latestLedger = await this.sorobanRpc.getLatestLedgerSequence();
87+
} catch (rpcErr) {
88+
this.logger.warn(`getLatestLedgerSequence failed – skipping poll cycle: ${rpcErr}`);
89+
return;
90+
}
91+
7992
if (latestLedger <= checkpoint) {
8093
this.logger.debug(`No new ledgers (checkpoint: ${checkpoint}, latest: ${latestLedger})`);
8194
return;
@@ -84,11 +97,17 @@ export class SubscriptionEventPollerService implements OnModuleInit {
8497
// Paginated fetch from checkpoint+1
8598
let cursor: string | undefined;
8699
do {
87-
const eventsResponse = await (this.sorobanRpc as any).getNetworkEvents({
88-
startLedger: checkpoint + 1,
89-
limit: 200,
90-
paginationToken: cursor,
91-
});
100+
let eventsResponse: Awaited<ReturnType<SorobanRpcService['getNetworkEvents']>>;
101+
try {
102+
eventsResponse = await this.sorobanRpc.getNetworkEvents({
103+
startLedger: checkpoint + 1,
104+
limit: 200,
105+
paginationToken: cursor,
106+
});
107+
} catch (rpcErr) {
108+
this.logger.warn(`getNetworkEvents failed – aborting page fetch: ${rpcErr}`);
109+
break;
110+
}
92111

93112
const events = eventsResponse.events ?? [];
94113
this.logger.debug(`Fetched ${events.length} events from ${eventsResponse.startLedger}-${eventsResponse.latestLedger}`);

‎backend/src/subscriptions/subscriptions.module.ts‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import { ScheduleModule } from '@nestjs/schedule';
44
import { ConfigModule } from '@nestjs/config';
55
import { LoggingModule } from '../common/logging.module';
66
import { EventsModule } from '../events/events.module';
7+
import { FeatureFlagsModule } from '../feature-flags/feature-flags.module';
78
import { SubscriptionLifecycleIndexerController } from './subscription-lifecycle-indexer.controller';
89
import { SubscriptionLifecycleIndexerService } from './subscription-lifecycle-indexer.service';
910
import { SubscriptionIndexEntity } from './entities/subscription-index.entity';
@@ -28,8 +29,9 @@ import { LedgerClockService } from './ledger-clock.service';
2829
ConfigModule,
2930
ScheduleModule,
3031
TypeOrmModule.forFeature([SubscriptionIndexEntity, FanSpendingCapEntity]),
31-
EventsModule,
32+
EventsModule,
3233
LoggingModule,
34+
FeatureFlagsModule,
3335
],
3436
controllers: [SubscriptionsController, SpendingCapController, SubscriptionLifecycleIndexerController],
3537
providers: [

0 commit comments

Comments
 (0)