Merge pull request #165 from modecodes/157-blockchain-event-indexer

fix(contract-sync): close reliability gaps in Soroban event indexer

Author: SudiptaPaul-31

__tests__/contract-sync/listener.test.ts

@@ -80,4 +80,116 @@ describe('SorobanEventListener', () => {
     await listener.start()
     expect(mockGetLatestLedger).toHaveBeenCalledTimes(1)
   })
+
+  describe('checkpoint resumption', () => {
+    it('resumes from an initialLedger option instead of fetching the chain tip', async () => {
+      const onCheckpoint = vi.fn()
+      const resumable = new SorobanEventListener({
+        rpcUrl: 'https://soroban-testnet.stellar.org',
+        networkPassphrase: 'Test SDF Network ; September 2015',
+        contractAddresses: ['CA1234'],
+        pollIntervalMs: 1000,
+        initialLedger: 900,
+        onCheckpoint,
+      })
+      resumable.setCallback(callback as any)
+
+      await resumable.start()
+
+      expect(mockGetLatestLedger).not.toHaveBeenCalled()
+      resumable.stop()
+    })
+
+    it('setInitialLedger seeds the resume point before start()', async () => {
+      listener.setInitialLedger(900)
+      await listener.start()
+
+      expect(mockGetLatestLedger).not.toHaveBeenCalled()
+    })
+
+    it('setInitialLedger is a no-op once already running', async () => {
+      mockGetLatestLedger.mockResolvedValue({ sequence: 500 })
+      await listener.start()
+
+      listener.setInitialLedger(900)
+      mockGetEvents.mockResolvedValue({ events: [] })
+      mockGetLatestLedger.mockResolvedValue({ sequence: 600 })
+
+      await vi.advanceTimersByTimeAsync(1000)
+
+      // startSeq should derive from the original 500 checkpoint (501), not 900
+      expect(mockGetEvents).toHaveBeenCalledWith(
+        expect.objectContaining({ startLedger: 501 })
+      )
+    })
+
+    it('invokes onCheckpoint with the new high-water-mark after each poll', async () => {
+      const onCheckpoint = vi.fn()
+      listener = new SorobanEventListener({
+        rpcUrl: 'https://soroban-testnet.stellar.org',
+        networkPassphrase: 'Test SDF Network ; September 2015',
+        contractAddresses: ['CA1234'],
+        pollIntervalMs: 1000,
+        onCheckpoint,
+      })
+      listener.setCallback(callback as any)
+
+      mockGetLatestLedger.mockResolvedValueOnce({ sequence: 500 })
+      await listener.start()
+
+      mockGetEvents.mockResolvedValue({ events: [] })
+      mockGetLatestLedger.mockResolvedValue({ sequence: 505 })
+      await vi.advanceTimersByTimeAsync(1000)
+
+      expect(onCheckpoint).toHaveBeenCalledWith(505)
+    })
+
+    it('awaits the callback for each event before advancing the checkpoint', async () => {
+      // Uses real timers and calls the private poll() directly so the
+      // ordering can be observed deterministically without racing fake-timer
+      // microtask flushing against a manually-controlled promise.
+      vi.useRealTimers()
+
+      const order: string[] = []
+      const onCheckpoint = vi.fn((ledger: number) => order.push(`checkpoint:${ledger}`))
+      let resolveCallback!: () => void
+      const slowCallback = vi.fn(
+        () =>
+          new Promise<void>((resolve) => {
+            resolveCallback = () => {
+              order.push('callback-resolved')
+              resolve()
+            }
+          })
+      )
+
+      const realtimeListener = new SorobanEventListener({
+        rpcUrl: 'https://soroban-testnet.stellar.org',
+        networkPassphrase: 'Test SDF Network ; September 2015',
+        contractAddresses: ['CA1234'],
+        pollIntervalMs: 1000,
+        initialLedger: 500,
+        onCheckpoint,
+      })
+      realtimeListener.setCallback(slowCallback)
+
+      mockGetEvents.mockResolvedValue({
+        events: [{ topic: ['fund'], value: [], ledger: 501, txHash: 'tx1' }],
+      })
+      mockGetLatestLedger.mockResolvedValue({ sequence: 505 })
+
+      const pollPromise = (realtimeListener as any).poll()
+      await Promise.resolve() // let poll() reach the awaited callback
+      await Promise.resolve()
+      expect(slowCallback).toHaveBeenCalled()
+      expect(onCheckpoint).not.toHaveBeenCalled()
+
+      resolveCallback()
+      await pollPromise
+
+      expect(order).toEqual(['callback-resolved', 'checkpoint:505'])
+
+      vi.useFakeTimers()
+    })
+  })
 })

__tests__/contract-sync/queue.test.ts

@@ -107,6 +107,44 @@ describe('SyncQueue', () => {
     expect(deadLetters[0].retryCount).toBeGreaterThanOrEqual(3)
   })
 
+  it('calls onRetry with the item and error message when a retry is scheduled', async () => {
+    const onRetry = vi.fn()
+    const retryQueue = new SyncQueue({ maxRetries: 3, concurrency: 2, pollIntervalMs: 100, onRetry })
+    retryQueue.setHandler(async () => {
+      throw new Error('transient failure')
+    })
+
+    retryQueue.enqueue(makePayload({ txHash: 'retry001' }))
+    retryQueue.start()
+    await vi.advanceTimersByTimeAsync(100)
+    retryQueue.stop()
+
+    expect(onRetry).toHaveBeenCalledTimes(1)
+    const [item, error] = onRetry.mock.calls[0]
+    expect(item.id).toBe('retry001:fund:0')
+    expect(item.retryCount).toBe(1)
+    expect(error).toBe('transient failure')
+  })
+
+  it('calls onDeadLetter once retries are exhausted, and does not call onRetry for that attempt', async () => {
+    const onRetry = vi.fn()
+    const onDeadLetter = vi.fn()
+    const dlQueue = new SyncQueue({ maxRetries: 1, concurrency: 2, pollIntervalMs: 100, onRetry, onDeadLetter })
+    dlQueue.setHandler(async () => {
+      throw new Error('fatal')
+    })
+
+    dlQueue.enqueue(makePayload({ txHash: 'dead002' }))
+    dlQueue.start()
+    await vi.advanceTimersByTimeAsync(100)
+    dlQueue.stop()
+
+    expect(onRetry).not.toHaveBeenCalled()
+    expect(onDeadLetter).toHaveBeenCalledTimes(1)
+    expect(onDeadLetter.mock.calls[0][0].id).toBe('dead002:fund:0')
+    expect(onDeadLetter.mock.calls[0][0].status).toBe('dead_letter')
+  })
+
   it('returns empty dead letters when all succeed', async () => {
     handler.mockResolvedValue(undefined)
 

__tests__/contract-sync/service.test.ts

@@ -0,0 +1,157 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest'
+
+vi.mock('@/lib/db', () => ({
+  sql: Object.assign(vi.fn(), { unsafe: vi.fn() }),
+}))
+
+const { mockGetLatestLedger, mockGetEvents, MockSorobanServer } = vi.hoisted(() => {
+  const mGetLatestLedger = vi.fn().mockResolvedValue({ sequence: 100 })
+  const mGetEvents = vi.fn().mockResolvedValue({ events: [] })
+
+  const MServer = class {
+    getLatestLedger = mGetLatestLedger
+    getEvents = mGetEvents
+  }
+
+  return {
+    mockGetLatestLedger: mGetLatestLedger,
+    mockGetEvents: mGetEvents,
+    MockSorobanServer: MServer,
+  }
+})
+
+vi.mock('@stellar/stellar-sdk', () => ({
+  default: MockSorobanServer,
+}))
+
+import { sql } from '@/lib/db'
+import { ContractSyncService } from '@/lib/contract-sync/service'
+import type { SorobanEventPayload } from '@/lib/contract-sync/types'
+
+function makePayload(overrides: Partial<SorobanEventPayload> = {}): SorobanEventPayload {
+  return {
+    event: 'fund',
+    contractAddress: 'CA1234',
+    ledgerSequence: 1000,
+    timestamp: Date.now(),
+    txHash: 'abc123',
+    data: [],
+    ...overrides,
+  }
+}
+
+describe('ContractSyncService', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+  })
+
+  describe('onEvent (duplicate protection)', () => {
+    it('skips enqueueing an event that was already synced successfully', async () => {
+      const service = new ContractSyncService({ contractAddresses: ['CA1'] })
+      const enqueueSpy = vi.spyOn(service.getQueue(), 'enqueue')
+
+      ;(sql as unknown as ReturnType<typeof vi.fn>).mockResolvedValueOnce([{ '?column?': 1 }])
+
+      await (service as any).onEvent(makePayload())
+
+      expect(enqueueSpy).not.toHaveBeenCalled()
+      expect(sql).toHaveBeenCalledTimes(1)
+    })
+
+    it('enqueues and logs an event that has not been synced before', async () => {
+      const service = new ContractSyncService({ contractAddresses: ['CA1'] })
+      const enqueueSpy = vi.spyOn(service.getQueue(), 'enqueue')
+
+      ;(sql as unknown as ReturnType<typeof vi.fn>)
+        .mockResolvedValueOnce([]) // isAlreadySynced -> not found
+        .mockResolvedValueOnce([]) // createSyncLog insert
+
+      await (service as any).onEvent(makePayload({ txHash: 'new-tx' }))
+
+      expect(enqueueSpy).toHaveBeenCalledWith(expect.objectContaining({ txHash: 'new-tx' }))
+      expect(sql).toHaveBeenCalledTimes(2)
+    })
+  })
+
+  describe('checkpoint persistence', () => {
+    it('loadCheckpoint returns null when no row exists', async () => {
+      const service = new ContractSyncService({ contractAddresses: ['CA1'] })
+      ;(sql as unknown as ReturnType<typeof vi.fn>).mockResolvedValueOnce([])
+
+      const result = await (service as any).loadCheckpoint()
+      expect(result).toBeNull()
+    })
+
+    it('loadCheckpoint returns the persisted ledger as a number', async () => {
+      const service = new ContractSyncService({ contractAddresses: ['CA1'] })
+      ;(sql as unknown as ReturnType<typeof vi.fn>).mockResolvedValueOnce([{ last_ledger: '4242' }])
+
+      const result = await (service as any).loadCheckpoint()
+      expect(result).toBe(4242)
+    })
+
+    it('persistCheckpoint issues an upsert', async () => {
+      const service = new ContractSyncService({ contractAddresses: ['CA1'] })
+      ;(sql as unknown as ReturnType<typeof vi.fn>).mockResolvedValueOnce([])
+
+      await (service as any).persistCheckpoint(555)
+      expect(sql).toHaveBeenCalledTimes(1)
+    })
+
+    it('start() seeds the listener from a persisted checkpoint instead of the chain tip', async () => {
+      const service = new ContractSyncService({ contractAddresses: ['CA1'] })
+      ;(sql as unknown as ReturnType<typeof vi.fn>).mockResolvedValueOnce([{ last_ledger: '777' }])
+
+      await service.start()
+      service.stop()
+
+      expect(mockGetLatestLedger).not.toHaveBeenCalled()
+    })
+  })
+
+  describe('failure / dead-letter audit logging', () => {
+    it('records a failed sync attempt with the error message on retry', async () => {
+      const service = new ContractSyncService({ contractAddresses: ['CA1'], maxRetries: 5 })
+      const updateSpy = vi.spyOn(service as any, 'updateSyncLog').mockResolvedValue(undefined)
+
+      const queue = service.getQueue()
+      queue.setHandler(async () => {
+        throw new Error('boom')
+      })
+
+      vi.useFakeTimers()
+      queue.enqueue(makePayload({ txHash: 'fail-tx' }))
+      queue.start()
+      await vi.advanceTimersByTimeAsync(1000)
+      queue.stop()
+      vi.useRealTimers()
+
+      expect(updateSpy).toHaveBeenCalledWith(
+        'fail-tx:fund:0',
+        expect.objectContaining({ status: 'failed', errorMessage: 'boom' })
+      )
+    })
+
+    it('records dead_letter status once retries are exhausted', async () => {
+      const service = new ContractSyncService({ contractAddresses: ['CA1'], maxRetries: 1 })
+      const updateSpy = vi.spyOn(service as any, 'updateSyncLog').mockResolvedValue(undefined)
+
+      const queue = service.getQueue()
+      queue.setHandler(async () => {
+        throw new Error('permanent failure')
+      })
+
+      vi.useFakeTimers()
+      queue.enqueue(makePayload({ txHash: 'dead-tx' }))
+      queue.start()
+      await vi.advanceTimersByTimeAsync(1000)
+      queue.stop()
+      vi.useRealTimers()
+
+      expect(updateSpy).toHaveBeenCalledWith(
+        'dead-tx:fund:0',
+        expect.objectContaining({ status: 'dead_letter', errorMessage: 'permanent failure' })
+      )
+    })
+  })
+})