Skip to content

Commit 57a00b5

Browse files
Rampop01Rampop01
committed
fix: support indexing governance events and fix migration lock provider
1 parent 9d27a98 commit 57a00b5

10 files changed

Lines changed: 1581 additions & 1585 deletions

File tree

‎backend/package.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@
4646
"@types/supertest": "^7.2.0",
4747
"@types/swagger-jsdoc": "^6.0.4",
4848
"@types/swagger-ui-express": "^4.1.6",
49+
"@vitest/coverage-v8": "^2.1.8",
4950
"eventsource": "^2.0.2",
5051
"nodemon": "^3.1.11",
5152
"prisma": "^7.4.1",

‎backend/prisma/migrations/migration_lock.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,3 @@
11
# Please do not edit this file manually
22
# It should be added in your version-control system (e.g., Git)
3-
provider = "sqlite"
3+
provider = "postgresql"

‎backend/src/__tests__/integration/streams.test.ts‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ const { mockPrisma, mockSseService } = vi.hoisted(() => ({
99
addClient: vi.fn(),
1010
broadcastToStream: vi.fn(),
1111
broadcastToUser: vi.fn(),
12+
broadcastToAdmin: vi.fn(),
1213
},
1314
mockPrisma: {
1415
stream: {
@@ -243,4 +244,127 @@ describeIfDatabase('Stream Lifecycle Integration Tests', () => {
243244
expect(Array.isArray(res.body.data)).toBe(true);
244245
expect(res.body.data.length).toBe(1);
245246
});
247+
248+
it('Indexer processes fee_config_updated -> persists FEE_CONFIG_UPDATED record and broadcasts SSE', async () => {
249+
const adminKey = Keypair.random().publicKey();
250+
const oldTreasury = Keypair.random().publicKey();
251+
const newTreasury = Keypair.random().publicKey();
252+
253+
const event = {
254+
id: 'fee-config-event-1',
255+
txHash: 'hash-fee-config',
256+
ledger: 105,
257+
inSuccessfulContractCall: true,
258+
topic: [
259+
xdr.ScVal.scvSymbol('fee_config_updated'),
260+
],
261+
value: xdr.ScVal.scvMap([
262+
new xdr.ScMapEntry({
263+
key: xdr.ScVal.scvSymbol('admin'),
264+
val: nativeToScVal(adminKey, { type: 'address' }),
265+
}),
266+
new xdr.ScMapEntry({
267+
key: xdr.ScVal.scvSymbol('old_treasury'),
268+
val: nativeToScVal(oldTreasury, { type: 'address' }),
269+
}),
270+
new xdr.ScMapEntry({
271+
key: xdr.ScVal.scvSymbol('new_treasury'),
272+
val: nativeToScVal(newTreasury, { type: 'address' }),
273+
}),
274+
new xdr.ScMapEntry({
275+
key: xdr.ScVal.scvSymbol('old_fee_rate_bps'),
276+
val: nativeToScVal(100, { type: 'u32' }),
277+
}),
278+
new xdr.ScMapEntry({
279+
key: xdr.ScVal.scvSymbol('new_fee_rate_bps'),
280+
val: nativeToScVal(200, { type: 'u32' }),
281+
}),
282+
]),
283+
} as any;
284+
285+
await sorobanEventWorker.processEvent(event);
286+
287+
expect(mockPrisma.user.upsert).toHaveBeenCalledWith(
288+
expect.objectContaining({
289+
where: { publicKey: 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF' },
290+
})
291+
);
292+
293+
expect(mockPrisma.stream.upsert).toHaveBeenCalledWith(
294+
expect.objectContaining({
295+
where: { streamId: 0 },
296+
})
297+
);
298+
299+
expect(mockPrisma.streamEvent.upsert).toHaveBeenCalledWith(
300+
expect.objectContaining({
301+
where: { transactionHash_eventType: { transactionHash: 'hash-fee-config', eventType: 'FEE_CONFIG_UPDATED' } },
302+
create: expect.objectContaining({
303+
streamId: 0,
304+
eventType: 'FEE_CONFIG_UPDATED',
305+
transactionHash: 'hash-fee-config',
306+
ledgerSequence: 105,
307+
}),
308+
})
309+
);
310+
311+
expect(mockSseService.broadcastToAdmin).toHaveBeenCalledWith(
312+
'stream.fee_config_updated',
313+
expect.objectContaining({
314+
admin: adminKey,
315+
oldTreasury,
316+
newTreasury,
317+
oldFeeRateBps: 100,
318+
newFeeRateBps: 200,
319+
})
320+
);
321+
});
322+
323+
it('Indexer processes admin_transferred -> persists ADMIN_TRANSFERRED record and broadcasts SSE', async () => {
324+
const previousAdmin = Keypair.random().publicKey();
325+
const newAdmin = Keypair.random().publicKey();
326+
327+
const event = {
328+
id: 'admin-transfer-event-1',
329+
txHash: 'hash-admin-transfer',
330+
ledger: 106,
331+
inSuccessfulContractCall: true,
332+
topic: [
333+
xdr.ScVal.scvSymbol('admin_transferred'),
334+
],
335+
value: xdr.ScVal.scvMap([
336+
new xdr.ScMapEntry({
337+
key: xdr.ScVal.scvSymbol('previous_admin'),
338+
val: nativeToScVal(previousAdmin, { type: 'address' }),
339+
}),
340+
new xdr.ScMapEntry({
341+
key: xdr.ScVal.scvSymbol('new_admin'),
342+
val: nativeToScVal(newAdmin, { type: 'address' }),
343+
}),
344+
]),
345+
} as any;
346+
347+
await sorobanEventWorker.processEvent(event);
348+
349+
expect(mockPrisma.streamEvent.upsert).toHaveBeenCalledWith(
350+
expect.objectContaining({
351+
where: { transactionHash_eventType: { transactionHash: 'hash-admin-transfer', eventType: 'ADMIN_TRANSFERRED' } },
352+
create: expect.objectContaining({
353+
streamId: 0,
354+
eventType: 'ADMIN_TRANSFERRED',
355+
transactionHash: 'hash-admin-transfer',
356+
ledgerSequence: 106,
357+
}),
358+
})
359+
);
360+
361+
expect(mockSseService.broadcastToAdmin).toHaveBeenCalledWith(
362+
'stream.admin_transferred',
363+
expect.objectContaining({
364+
previousAdmin,
365+
newAdmin,
366+
})
367+
);
368+
});
246369
});
370+

‎backend/src/controllers/stream.controller.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -294,7 +294,7 @@ export const getStreamEvents = async (req: Request, res: Response) => {
294294

295295
const whereClause: any = { streamId: parsedStreamId };
296296
if (eventType) {
297-
const validEventTypes = ['CREATED', 'TOPPED_UP', 'WITHDRAWN', 'CANCELLED', 'COMPLETED', 'PAUSED', 'RESUMED', 'FEE_COLLECTED'];
297+
const validEventTypes = ['CREATED', 'TOPPED_UP', 'WITHDRAWN', 'CANCELLED', 'COMPLETED', 'PAUSED', 'RESUMED', 'FEE_COLLECTED', 'FEE_CONFIG_UPDATED', 'ADMIN_TRANSFERRED'];
298298
if (!validEventTypes.includes(eventType)) {
299299
return res.status(400).json({
300300
error: 'Invalid eventType parameter',

‎backend/src/routes/v1/events.routes.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ const EVENT_TYPES = new Set([
1717
'PAUSED',
1818
'RESUMED',
1919
'FEE_COLLECTED',
20+
'FEE_CONFIG_UPDATED',
21+
'ADMIN_TRANSFERRED',
2022
]);
2123

2224
const MAX_EVENT_LIMIT = 200;

‎backend/src/workers/soroban-event-worker.ts‎

Lines changed: 146 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,11 @@ export function decodeU64(val: xdr.ScVal): bigint {
2222
return BigInt(val.u64().toString());
2323
}
2424

25+
/** Decode an ScVal U32 to a JavaScript number. */
26+
export function decodeU32(val: xdr.ScVal): number {
27+
return val.u32();
28+
}
29+
2530
/**
2631
* Decode an ScVal I128 to a decimal string suitable for DB storage.
2732
* I128 in XDR is split into hi (signed Int64) and lo (unsigned Uint64).
@@ -141,6 +146,32 @@ export class SorobanEventWorker {
141146
this.pollTimer = setTimeout(() => this.poll(), this.pollIntervalMs);
142147
}
143148

149+
private async ensureSystemStream(tx: any): Promise<void> {
150+
const systemUser = 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF';
151+
await tx.user.upsert({
152+
where: { publicKey: systemUser },
153+
create: { publicKey: systemUser },
154+
update: {},
155+
});
156+
await tx.stream.upsert({
157+
where: { streamId: 0 },
158+
create: {
159+
streamId: 0,
160+
sender: systemUser,
161+
recipient: systemUser,
162+
tokenAddress: 'CDAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAHF',
163+
ratePerSecond: '0',
164+
depositedAmount: '0',
165+
withdrawnAmount: '0',
166+
startTime: 0,
167+
lastUpdateTime: 0,
168+
endTime: 0,
169+
isActive: false,
170+
},
171+
update: {},
172+
});
173+
}
174+
144175
private async poll(): Promise<void> {
145176
this.activeBatch = this.fetchAndProcessEvents().catch((err) => {
146177
logger.error('[SorobanWorker] Unhandled error during poll:', err);
@@ -247,14 +278,26 @@ export class SorobanEventWorker {
247278
public async processEvent(
248279
event: rpc.Api.EventResponse,
249280
): Promise<void> {
250-
if (!event.topic || event.topic.length < 2) return;
281+
if (!event.topic || event.topic.length < 1) return;
251282

252283
const topic0: xdr.ScVal | undefined = event.topic[0];
253-
const topic1: xdr.ScVal | undefined = event.topic[1];
254-
if (!topic0 || !topic1) return;
284+
if (!topic0) return;
255285

256286
const eventName = decodeSymbol(topic0);
257287

288+
if (eventName === 'fee_config_updated' || eventName === 'admin_transferred') {
289+
if (eventName === 'fee_config_updated') {
290+
await this.handleFeeConfigUpdated(event);
291+
} else {
292+
await this.handleAdminTransferred(event);
293+
}
294+
return;
295+
}
296+
297+
if (event.topic.length < 2) return;
298+
const topic1: xdr.ScVal | undefined = event.topic[1];
299+
if (!topic1) return;
300+
258301
switch (eventName) {
259302
case 'stream_created':
260303
await this.handleStreamCreated(event, topic1);
@@ -286,6 +329,106 @@ export class SorobanEventWorker {
286329
}
287330
}
288331

332+
private async handleFeeConfigUpdated(
333+
event: rpc.Api.EventResponse,
334+
): Promise<void> {
335+
const body = decodeMap(event.value);
336+
337+
if (
338+
!body['admin'] ||
339+
!body['old_treasury'] ||
340+
!body['new_treasury'] ||
341+
body['old_fee_rate_bps'] === undefined ||
342+
body['new_fee_rate_bps'] === undefined
343+
) {
344+
throw new Error('FeeConfigUpdated: missing body fields');
345+
}
346+
347+
const admin = decodeAddress(body['admin']);
348+
const oldTreasury = decodeAddress(body['old_treasury']);
349+
const newTreasury = decodeAddress(body['new_treasury']);
350+
const oldFeeRateBps = decodeU32(body['old_fee_rate_bps']);
351+
const newFeeRateBps = decodeU32(body['new_fee_rate_bps']);
352+
const timestamp = Math.floor(Date.now() / 1000);
353+
354+
await prisma.$transaction(async (tx: any) => {
355+
await this.ensureSystemStream(tx);
356+
357+
await tx.streamEvent.upsert({
358+
where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'FEE_CONFIG_UPDATED' } },
359+
create: {
360+
streamId: 0,
361+
eventType: 'FEE_CONFIG_UPDATED',
362+
transactionHash: event.txHash,
363+
ledgerSequence: event.ledger,
364+
timestamp,
365+
metadata: JSON.stringify({
366+
admin,
367+
old_treasury: oldTreasury,
368+
new_treasury: newTreasury,
369+
old_fee_rate_bps: oldFeeRateBps,
370+
new_fee_rate_bps: newFeeRateBps,
371+
}),
372+
},
373+
update: {},
374+
});
375+
});
376+
377+
sseService.broadcastToAdmin('stream.fee_config_updated', {
378+
admin,
379+
oldTreasury,
380+
newTreasury,
381+
oldFeeRateBps,
382+
newFeeRateBps,
383+
transactionHash: event.txHash,
384+
ledger: event.ledger,
385+
timestamp,
386+
});
387+
}
388+
389+
private async handleAdminTransferred(
390+
event: rpc.Api.EventResponse,
391+
): Promise<void> {
392+
const body = decodeMap(event.value);
393+
394+
if (!body['previous_admin'] || !body['new_admin']) {
395+
throw new Error('AdminTransferred: missing body fields');
396+
}
397+
398+
const previousAdmin = decodeAddress(body['previous_admin']);
399+
const newAdmin = decodeAddress(body['new_admin']);
400+
const timestamp = Math.floor(Date.now() / 1000);
401+
402+
await prisma.$transaction(async (tx: any) => {
403+
await this.ensureSystemStream(tx);
404+
405+
await tx.streamEvent.upsert({
406+
where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'ADMIN_TRANSFERRED' } },
407+
create: {
408+
streamId: 0,
409+
eventType: 'ADMIN_TRANSFERRED',
410+
transactionHash: event.txHash,
411+
ledgerSequence: event.ledger,
412+
timestamp,
413+
metadata: JSON.stringify({
414+
previous_admin: previousAdmin,
415+
new_admin: newAdmin,
416+
transactionHash: event.txHash,
417+
}),
418+
},
419+
update: {},
420+
});
421+
});
422+
423+
sseService.broadcastToAdmin('stream.admin_transferred', {
424+
previousAdmin,
425+
newAdmin,
426+
transactionHash: event.txHash,
427+
ledger: event.ledger,
428+
timestamp,
429+
});
430+
}
431+
289432
// ─── Event Handlers ────────────────────────────────────────────────────────
290433

291434
private async handleStreamCreated(

‎backend/tests/integration/stream-actions.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -157,7 +157,7 @@ describe('stream action routes', () => {
157157
});
158158
});
159159

160-
it('POST /v1/streams/:streamId/resume resumes a paused sender-owned stream', async () =>
160+
it('POST /v1/streams/:streamId/resume resumes a paused sender-owned stream', async () => {
161161
const sender = makeKeypair();
162162
const token = await getValidJwt(sender);
163163

0 commit comments

Comments
 (0)