diff --git a/backend/src/services/checkpoint.service.ts b/backend/src/services/checkpoint.service.ts index 618fa0bc..e8d58086 100644 --- a/backend/src/services/checkpoint.service.ts +++ b/backend/src/services/checkpoint.service.ts @@ -397,9 +397,9 @@ async function recomputeStreamState( let deposited = BigInt(stream.depositedAmount); let withdrawn = 0n; let isActive = stream.isActive; - let isPaused = stream.isPaused; - let pausedAt = stream.pausedAt; - let totalPausedDuration = stream.totalPausedDuration; + let isPaused = false; + let pausedAt: bigint | null = null; + let totalPausedDuration = 0; let lastUpdateTime = stream.lastUpdateTime; let endTime = stream.endTime; let sawRelevantEvent = false; @@ -411,6 +411,9 @@ async function recomputeStreamState( sawRelevantEvent = true; if (event.amount) deposited = BigInt(event.amount); isActive = true; + isPaused = false; + pausedAt = null; + totalPausedDuration = 0; endTime = rate === 0n ? null : startTime + deposited / rate; lastUpdateTime = event.timestamp; break; @@ -454,17 +457,24 @@ async function recomputeStreamState( const paused = metadata['pausedAt']; if (typeof paused === 'number') pausedAt = BigInt(paused); else if (typeof paused === 'string') pausedAt = BigInt(paused); + else if (typeof paused === 'bigint') pausedAt = paused; + else pausedAt = event.timestamp; lastUpdateTime = event.timestamp; break; } case 'RESUMED': { sawRelevantEvent = true; isPaused = false; - pausedAt = null; const total = metadata['totalPausedDuration']; if (typeof total === 'number') totalPausedDuration = total; + else if (typeof total === 'string') totalPausedDuration = parseInt(total, 10) || 0; + else if (pausedAt !== null) { + totalPausedDuration += Number(event.timestamp - pausedAt); + } + pausedAt = null; const newEndTime = metadata['newEndTime']; if (typeof newEndTime === 'string') endTime = BigInt(newEndTime); + else if (typeof newEndTime === 'number' || typeof newEndTime === 'bigint') endTime = BigInt(newEndTime); lastUpdateTime = event.timestamp; break; } diff --git a/backend/src/services/indexerService.ts b/backend/src/services/indexerService.ts index abb7cda3..1eca8781 100644 --- a/backend/src/services/indexerService.ts +++ b/backend/src/services/indexerService.ts @@ -240,7 +240,7 @@ export function serializeDeadLetterPayload(event: rpc.Api.EventResponse): string ledgerClosedAt: event.ledgerClosedAt, transactionIndex: event.transactionIndex, operationIndex: event.operationIndex, - inSuccessfulContractCall: event.inSuccessfulContractCall, + inSuccessfulContractCall: event.inSuccessfulContractCall ?? false, topic: event.topic.map((scval) => Buffer.from(scval.toXDR()).toString('base64')), value: Buffer.from(event.value.toXDR()).toString('base64'), }; diff --git a/backend/tests/reorg-resilience.test.ts b/backend/tests/reorg-resilience.test.ts index 2303f8eb..9414a8b2 100644 --- a/backend/tests/reorg-resilience.test.ts +++ b/backend/tests/reorg-resilience.test.ts @@ -310,6 +310,133 @@ describe('rollbackAboveLedger', () => { expect(tx.stream.delete).not.toHaveBeenCalled(); expect(result.streamsDeleted).toBe(0); }); + + it('restores pause state to unpaused when a PAUSED event is rolled back and only CREATED survives', async () => { + tx.streamEvent.findMany + .mockResolvedValueOnce([{ streamId: 42n, ledgerSequence: 1003 }]) + .mockResolvedValueOnce([ + streamEventRow({ + ledgerSequence: 1000, + eventType: 'CREATED', + amount: '1000', + timestamp: 1_700_000_000n, + }), + ]); + tx.streamEvent.deleteMany.mockResolvedValue({ count: 1 }); + // Reverted DB row reflects the reverted fork where it was paused + tx.stream.findUnique.mockResolvedValue( + streamRow({ + isPaused: true, + pausedAt: 1_700_000_020n, + totalPausedDuration: 0, + }), + ); + tx.stream.update.mockResolvedValue({}); + tx.ledgerCheckpoint.updateMany.mockResolvedValue({ count: 1 }); + tx.indexerState.upsert.mockResolvedValue({}); + + await checkpoint.rollbackAboveLedger(1002); + + expect(tx.stream.update).toHaveBeenCalledTimes(1); + const update = tx.stream.update.mock.calls[0]![0]; + expect(update.where).toEqual({ streamId: 42n }); + expect(update.data.isPaused).toBe(false); + expect(update.data.pausedAt).toBeNull(); + expect(update.data.totalPausedDuration).toBe(0); + }); + + it('restores stream to paused state and previous pausedAt when RESUMED is rolled back', async () => { + tx.streamEvent.findMany + .mockResolvedValueOnce([{ streamId: 42n, ledgerSequence: 1004 }]) + .mockResolvedValueOnce([ + streamEventRow({ + ledgerSequence: 1000, + eventType: 'CREATED', + amount: '1000', + timestamp: 1_700_000_000n, + }), + streamEventRow({ + ledgerSequence: 1002, + eventType: 'PAUSED', + timestamp: 1_700_000_020n, + metadata: JSON.stringify({ pausedAt: '1700000020' }), + }), + ]); + tx.streamEvent.deleteMany.mockResolvedValue({ count: 1 }); + // Current DB row was updated by reverted RESUMED + tx.stream.findUnique.mockResolvedValue( + streamRow({ + isPaused: false, + pausedAt: null, + totalPausedDuration: 30, + }), + ); + tx.stream.update.mockResolvedValue({}); + tx.ledgerCheckpoint.updateMany.mockResolvedValue({ count: 1 }); + tx.indexerState.upsert.mockResolvedValue({}); + + await checkpoint.rollbackAboveLedger(1003); + + expect(tx.stream.update).toHaveBeenCalledTimes(1); + const update = tx.stream.update.mock.calls[0]![0]; + expect(update.where).toEqual({ streamId: 42n }); + expect(update.data.isPaused).toBe(true); + expect(update.data.pausedAt).toBe(1_700_000_020n); + expect(update.data.totalPausedDuration).toBe(0); + }); + + it('restores pause state and totalPausedDuration through repeated pause cycles when later cycles are rolled back', async () => { + tx.streamEvent.findMany + .mockResolvedValueOnce([ + { streamId: 42n, ledgerSequence: 1004 }, + { streamId: 42n, ledgerSequence: 1005 }, + ]) + .mockResolvedValueOnce([ + streamEventRow({ + ledgerSequence: 1000, + eventType: 'CREATED', + amount: '1000', + timestamp: 1_700_000_000n, + }), + streamEventRow({ + ledgerSequence: 1001, + eventType: 'PAUSED', + timestamp: 1_700_000_010n, + metadata: JSON.stringify({ pausedAt: '1700000010' }), + }), + streamEventRow({ + ledgerSequence: 1002, + eventType: 'RESUMED', + timestamp: 1_700_000_025n, + metadata: JSON.stringify({ + pausedDuration: 15, + totalPausedDuration: 15, + newEndTime: '1700000115', + }), + }), + ]); + tx.streamEvent.deleteMany.mockResolvedValue({ count: 2 }); + // Current DB row has cycle 2 reverted values + tx.stream.findUnique.mockResolvedValue( + streamRow({ + isPaused: false, + pausedAt: null, + totalPausedDuration: 45, + }), + ); + tx.stream.update.mockResolvedValue({}); + tx.ledgerCheckpoint.updateMany.mockResolvedValue({ count: 2 }); + tx.indexerState.upsert.mockResolvedValue({}); + + await checkpoint.rollbackAboveLedger(1003); + + expect(tx.stream.update).toHaveBeenCalledTimes(1); + const update = tx.stream.update.mock.calls[0]![0]; + expect(update.where).toEqual({ streamId: 42n }); + expect(update.data.isPaused).toBe(false); + expect(update.data.pausedAt).toBeNull(); + expect(update.data.totalPausedDuration).toBe(15); + }); }); describe('verifyAndRecover', () => { diff --git a/frontend/src/lib/api-types.generated.ts b/frontend/src/lib/api-types.generated.ts index bb0db700..02d2c791 100644 --- a/frontend/src/lib/api-types.generated.ts +++ b/frontend/src/lib/api-types.generated.ts @@ -2577,6 +2577,197 @@ export interface paths { patch?: never; trace?: never; }; + "/v1/analytics/tvl": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * Current protocol TVL partitioned by token asset + * @description Read-only protocol statistics. No authentication is required because + * every payload here is public (the same numbers a block explorer + * would show). Subject to the API's global rate limiter. + */ + get: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description TVL snapshot */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + /** @example true */ + success: boolean; + /** @description Current TVL snapshot partitioned by token asset */ + data: Record; + }; + }; + }; + /** @description Too many requests */ + 429: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description TVL snapshot unavailable */ + 500: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + }; + }; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/v1/analytics/historical": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * Pre-aggregated time-series points for frontend charts + * @description Historical TVL and volume series at a fixed interval. No + * authentication is required; subject to the API's global rate limiter. + */ + get: { + parameters: { + query?: { + /** @description Lookback window */ + period?: "7d" | "30d" | "90d" | "1y"; + /** @description Granularity of each point */ + interval?: "1h" | "1d"; + }; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Time-series points */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + /** @example true */ + success: boolean; + /** @description Aggregated time-series points for the requested period and interval */ + data: unknown[]; + }; + }; + }; + /** @description Invalid period or interval */ + 400: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Too many requests */ + 429: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Historical analytics unavailable */ + 500: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + }; + }; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; + "/v1/analytics/defillama": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + /** + * DefiLlama adapter payload + * @description TVL and activity numbers in the shape DefiLlama's adapter format + * expects, so the protocol can be listed without a separate adapter. + * No authentication is required; subject to the global rate limiter. + */ + get: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description DefiLlama adapter payload */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": { + /** @example true */ + success: boolean; + /** @description DefiLlama-shaped adapter payload */ + data: Record; + }; + }; + }; + /** @description Too many requests */ + 429: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description DefiLlama adapter unavailable */ + 500: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + }; + }; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/v1/admin/metrics": { parameters: { query?: never;