fix: add pagination to tap collector

packages/indexer-common/src/allocations/__tests__/tap-pagination.test.ts

@@ -0,0 +1,229 @@
+import { Address, Eventual, createLogger, createMetrics } from '@graphprotocol/common-ts'
+import {
+  Allocation,
+  AllocationsResponse,
+  NetworkSubgraph,
+  QueryFeeModels,
+  QueryResult,
+  TapCollector,
+  TapSubgraphResponse,
+  TapTransaction,
+  TransactionManager,
+} from '@graphprotocol/indexer-common'
+import { NetworkContracts as TapContracts } from '@semiotic-labs/tap-contracts-bindings'
+import { TAPSubgraph } from '../../tap-subgraph'
+import { NetworkSpecification } from 'indexer-common/src/network-specification'
+import { createMockAllocation } from '../../indexer-management/__tests__/helpers.test'
+import { getContractAddress } from 'ethers/lib/utils'
+
+const timeout = 30_000
+
+// mock allocation subgraph responses
+//
+// firstPage // 1000
+// secondPage // 1000
+// thirdPage // 999
+const allocations: Allocation[] = []
+const from = '0x8ba1f109551bD432803012645Ac136ddd64DBA72'
+
+for (let i = 0; i < 2999; i++) {
+  const mockAllocation = createMockAllocation()
+  allocations.push({
+    ...mockAllocation,
+    id: getContractAddress({ from, nonce: i }) as Address,
+  })
+}
+
+// mock transactions subgraph response
+//
+// firstPage // 1000
+// secondPage // 1000
+const transactions: TapTransaction[] = []
+for (let i = 0; i < 2000; i++) {
+  transactions.push({
+    id: i.toString(),
+    sender: { id: 'sender' },
+    allocationID: 'allocation id',
+    timestamp: i,
+  })
+}
+
+// Make global Jest variables available
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+declare const __LOG_LEVEL__: never
+let tapCollector: TapCollector
+
+function paginateArray<T>(
+  array: T[],
+  getId: (item: T) => string,
+  pageSize: number,
+  lastId?: string,
+): T[] {
+  // Sort the array by ID to ensure consistent pagination.
+  array.sort((a, b) => getId(a).localeCompare(getId(b)))
+
+  // Find the index of the item with the given lastId.
+  let startIndex = 0
+  if (lastId) {
+    startIndex = array.findIndex((item) => getId(item) === lastId) + 1
+  }
+
+  // Slice the array to return only the requested page size.
+  return array.slice(startIndex, startIndex + pageSize)
+}
+
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+let mockQueryNetworkSubgraph: jest.Mock<any, any, any>
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+let mockQueryTapSubgraph: jest.Mock<any, any, any>
+
+jest.spyOn(TapCollector.prototype, 'startRAVProcessing').mockImplementation()
+const setup = () => {
+  const logger = createLogger({
+    name: 'Indexer API Client',
+    async: false,
+    level: __LOG_LEVEL__ ?? 'error',
+  })
+  const metrics = createMetrics()
+  // Clearing the registry prevents duplicate metric registration in the default registry.
+  metrics.registry.clear()
+
+  mockQueryTapSubgraph = jest
+    .fn()
+    .mockImplementation(
+      async (_, variables): Promise<QueryResult<TapSubgraphResponse>> => {
+        console.log('MOCKING IMPLEMENTATION FOR TAP SUBGRAPH')
+        const pageSize: number = variables.pageSize
+        const lastId: string | undefined = variables.lastId
+
+        const paginatedTransactions = paginateArray(
+          transactions,
+          (tx) => tx.id,
+          pageSize,
+          lastId,
+        )
+
+        return {
+          data: {
+            transactions: paginatedTransactions,
+            _meta: {
+              block: {
+                hash: 'blockhash',
+                timestamp: 100000,
+              },
+            },
+          },
+        }
+      },
+    )
+
+  mockQueryNetworkSubgraph = jest
+    .fn()
+    .mockImplementation(
+      async (_, variables): Promise<QueryResult<AllocationsResponse>> => {
+        const pageSize: number = variables.pageSize
+        const lastId: string | undefined = variables.lastId
+
+        const paginatedAllocations = paginateArray(
+          allocations,
+          (allocation) => allocation.id,
+          pageSize,
+          lastId,
+        )
+
+        return {
+          data: {
+            allocations: paginatedAllocations,
+            meta: {
+              block: {
+                hash: 'blockhash',
+              },
+            },
+          },
+        }
+      },
+    )
+  {
+    const transactionManager = null as unknown as TransactionManager
+    const models = null as unknown as QueryFeeModels
+    const tapContracts = null as unknown as TapContracts
+    const allocations = null as unknown as Eventual<Allocation[]>
+    const networkSpecification = {
+      indexerOptions: { voucherRedemptionThreshold: 0, finalityTime: 0 },
+      networkIdentifier: 'test',
+    } as unknown as NetworkSpecification
+
+    const tapSubgraph = {
+      query: mockQueryTapSubgraph,
+    } as unknown as TAPSubgraph
+    const networkSubgraph = {
+      query: mockQueryNetworkSubgraph,
+    } as unknown as NetworkSubgraph
+
+    tapCollector = TapCollector.create({
+      logger,
+      metrics,
+      transactionManager,
+      models,
+      tapContracts,
+      allocations,
+      networkSpecification,
+
+      networkSubgraph,
+      tapSubgraph,
+    })
+  }
+}
+
+describe('TAP Pagination', () => {
+  beforeAll(setup, timeout)
+  test(
+    'test `getAllocationsfromAllocationIds` pagination',
+    async () => {
+      {
+        const allocations = await tapCollector['getAllocationsfromAllocationIds']([])
+        expect(mockQueryNetworkSubgraph).toBeCalledTimes(3)
+        expect(allocations.length).toEqual(2999)
+      }
+      mockQueryNetworkSubgraph.mockClear()
+
+      const mockAllocation = createMockAllocation()
+      allocations.push({
+        ...mockAllocation,
+        id: getContractAddress({ from, nonce: 3000 }) as Address,
+      })
+      {
+        const allocations = await tapCollector['getAllocationsfromAllocationIds']([])
+        expect(mockQueryNetworkSubgraph).toBeCalledTimes(4)
+        expect(allocations.length).toEqual(3000)
+      }
+    },
+    timeout,
+  )
+  test(
+    'test `findTransactionsForRavs` pagination',
+    async () => {
+      {
+        const transactionsResponse = await tapCollector['findTransactionsForRavs']([])
+        expect(mockQueryTapSubgraph).toBeCalledTimes(3)
+        expect(transactionsResponse.transactions.length).toEqual(2000)
+      }
+
+      mockQueryTapSubgraph.mockClear()
+      for (let i = 0; i < 500; i++) {
+        transactions.push({
+          id: i.toString(),
+          sender: { id: 'sender' },
+          allocationID: 'allocation id',
+          timestamp: i,
+        })
+      }
+      {
+        const transactionsResponse = await tapCollector['findTransactionsForRavs']([])
+        expect(mockQueryTapSubgraph).toBeCalledTimes(3)
+        expect(transactionsResponse.transactions.length).toEqual(2500)
+      }
+    },
+    timeout,
+  )
+})

packages/indexer-common/src/allocations/__tests__/tap.test.ts

@@ -101,6 +101,7 @@ const setupEach = async () => {
         _meta: {
           block: {
             timestamp: Date.now(),
+            hash: 'str',
           },
         },
       }
@@ -460,6 +461,7 @@ describe('TAP', () => {
           return {
             transactions: [
               {
+                id: 'test',
                 allocationID: ALLOCATION_ID_2.toString().toLowerCase().replace('0x', ''),
                 timestamp: redeemDateSecs,
                 sender: {
@@ -470,6 +472,7 @@ describe('TAP', () => {
             _meta: {
               block: {
                 timestamp: nowSecs,
+                hash: 'test',
               },
             },
           }

packages/indexer-common/src/allocations/__tests__/validate-queries.test.ts

@@ -0,0 +1,90 @@
+import {
+  defineQueryFeeModels,
+  GraphNode,
+  Network,
+  QueryFeeModels,
+  TapCollector,
+} from '@graphprotocol/indexer-common'
+import {
+  connectDatabase,
+  createLogger,
+  createMetrics,
+  Logger,
+  Metrics,
+} from '@graphprotocol/common-ts'
+import { testNetworkSpecification } from '../../indexer-management/__tests__/util'
+import { Sequelize } from 'sequelize'
+
+// Make global Jest variables available
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+declare const __DATABASE__: any
+declare const __LOG_LEVEL__: never
+let logger: Logger
+let tapCollector: TapCollector
+let metrics: Metrics
+let queryFeeModels: QueryFeeModels
+let sequelize: Sequelize
+const timeout = 30000
+
+const setup = async () => {
+  logger = createLogger({
+    name: 'Indexer API Client',
+    async: false,
+    level: __LOG_LEVEL__ ?? 'error',
+  })
+  metrics = createMetrics()
+  // Clearing the registry prevents duplicate metric registration in the default registry.
+  metrics.registry.clear()
+  sequelize = await connectDatabase(__DATABASE__)
+  queryFeeModels = defineQueryFeeModels(sequelize)
+  sequelize = await sequelize.sync({ force: true })
+
+  const graphNode = new GraphNode(
+    logger,
+    'https://test-admin-endpoint.xyz',
+    'https://test-query-endpoint.xyz',
+    'https://test-status-endpoint.xyz',
+  )
+
+  const network = await Network.create(
+    logger,
+    testNetworkSpecification,
+    queryFeeModels,
+    graphNode,
+    metrics,
+  )
+  tapCollector = network.tapCollector!
+}
+
+jest.spyOn(TapCollector.prototype, 'startRAVProcessing').mockImplementation()
+describe('Validate TAP queries', () => {
+  beforeAll(setup, timeout)
+
+  test(
+    'test `getAllocationsfromAllocationIds` query is valid',
+    async () => {
+      const mockedFunc = jest.spyOn(tapCollector.networkSubgraph, 'query')
+      const result = await tapCollector['getAllocationsfromAllocationIds']([])
+      expect(result).toEqual([])
+      // this subgraph is in an eventual
+      // we check if it was called more than 0 times
+      expect(mockedFunc).toBeCalled()
+      mockedFunc.mockReset()
+    },
+    timeout,
+  )
+
+  test(
+    'test `findTransactionsForRavs` query is valid',
+    async () => {
+      const mockedFunc = jest.spyOn(tapCollector.tapSubgraph, 'query')
+
+      const result = await tapCollector['findTransactionsForRavs']([])
+      expect(result.transactions).toEqual([])
+      expect(result._meta.block.hash.length).toEqual(66)
+      expect(mockedFunc).toBeCalledTimes(1)
+      mockedFunc.mockReset()
+    },
+    timeout,
+  )
+})