diff --git a/src/events.ts b/src/events.ts index 3167430..ecc38f2 100644 --- a/src/events.ts +++ b/src/events.ts @@ -83,7 +83,8 @@ export function subscribeToStream( ): Subscription { const server = createRpcServer(rpcUrl); const pollInterval = handlers.pollInterval ?? 5000; - let startLedger = 0; // updated after each successful poll + let startLedger = 0; // fallback cursor for the initial poll + let cursor: string | undefined; let stopped = false; let timer: ReturnType | undefined; @@ -92,7 +93,7 @@ export function subscribeToStream( try { const response = await server.getEvents({ - ...(startLedger > 0 ? { startLedger } : {}), + ...(cursor ? { cursor } : (startLedger > 0 ? { startLedger } : {})), filters: [{ type: 'contract', contractIds: [streamAddress], @@ -101,14 +102,19 @@ export function subscribeToStream( }); if (response.events.length > 0) { - // Update startLedger to avoid replaying events - const maxLedger = Math.max(...response.events.map(e => e.ledger)); - startLedger = maxLedger + 1; - for (const event of response.events) { dispatchEvent(event, handlers); } } + + if (response.cursor) { + cursor = response.cursor; + } else { + cursor = undefined; + if (response.latestLedger !== undefined) { + startLedger = response.latestLedger + 1; + } + } } catch (err) { // Swallow polling errors; the subscription continues console.warn('[conduit-sdk] event polling error:', err); diff --git a/src/tests/events-subscribe.test.ts b/src/tests/events-subscribe.test.ts index 4427883..3b7e11b 100644 --- a/src/tests/events-subscribe.test.ts +++ b/src/tests/events-subscribe.test.ts @@ -62,21 +62,46 @@ describe('subscribeToStream', () => { sub.unsubscribe(); }); - it('advances startLedger past the highest event ledger seen, and polls again after the interval', async () => { - mockGetEvents - .mockResolvedValueOnce({ events: [{ ledger: 100, topic: [], value: xdr.ScVal.scvVoid() }] }) - .mockResolvedValueOnce({ events: [] }); + it('uses the RPC cursor to continue polling when a ledger spans multiple pages', async () => { + const { Address, Keypair } = await import('@stellar/stellar-sdk'); + const sender = Keypair.random().publicKey(); + const makeEvent = () => ({ + ledger: 100, + topic: [xdr.ScVal.scvSymbol('clawback'), new Address(sender).toScVal()], + value: xdr.ScVal.scvI128( + new xdr.Int128Parts({ hi: xdr.Int64.fromString('0'), lo: xdr.Uint64.fromString('5000') }), + ), + }); + + mockGetEvents.mockImplementation((params?: { cursor?: string }) => { + if (params?.cursor === 'page-2') { + return Promise.resolve({ events: [makeEvent()], latestLedger: 100 }); + } + + return Promise.resolve({ + events: Array.from({ length: 101 }, () => makeEvent()), + cursor: 'page-2', + latestLedger: 100, + }); + }); const { subscribeToStream } = await import('../events.js'); - const sub = subscribeToStream('http://localhost:8000', 'CSTREAM', { pollInterval: 1000 }); + const received: unknown[] = []; + const sub = subscribeToStream('http://localhost:8000', 'CSTREAM', { + pollInterval: 1000, + onClawback: (event) => { received.push(event); }, + }); await vi.waitFor(() => expect(mockGetEvents).toHaveBeenCalledTimes(1)); + await vi.waitFor(() => expect(received).toHaveLength(101)); await vi.advanceTimersByTimeAsync(1000); await vi.waitFor(() => expect(mockGetEvents).toHaveBeenCalledTimes(2)); + await vi.waitFor(() => expect(received).toHaveLength(102)); expect(mockGetEvents.mock.calls[1]?.[0]).toEqual( - expect.objectContaining({ startLedger: 101 }), + expect.objectContaining({ cursor: 'page-2' }), ); + expect(mockGetEvents.mock.calls[1]?.[0]).not.toHaveProperty('startLedger'); sub.unsubscribe(); });