Fix event poll cursor for multi-page ledgers

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<typeof setTimeout> | 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);

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();
   });