feat: pagination (#1490)

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

packages/data-access/src/combined-data-access.ts

@@ -26,14 +26,18 @@ export abstract class CombinedDataAccess implements DataAccessTypes.IDataAccess
   async getChannelsByTopic(
     topic: string,
     updatedBetween?: DataAccessTypes.ITimestampBoundaries | undefined,
+    page?: number,
+    pageSize?: number,
   ): Promise<DataAccessTypes.IReturnGetChannelsByTopic> {
-    return await this.reader.getChannelsByTopic(topic, updatedBetween);
+    return await this.reader.getChannelsByTopic(topic, updatedBetween, page, pageSize);
   }
   async getChannelsByMultipleTopics(
     topics: string[],
     updatedBetween?: DataAccessTypes.ITimestampBoundaries,
+    page?: number | undefined,
+    pageSize?: number | undefined,
   ): Promise<DataAccessTypes.IReturnGetChannelsByTopic> {
-    return await this.reader.getChannelsByMultipleTopics(topics, updatedBetween);
+    return await this.reader.getChannelsByMultipleTopics(topics, updatedBetween, page, pageSize);
   }
   async persistTransaction(
     transactionData: DataAccessTypes.ITransaction,

packages/data-access/src/data-read.ts

@@ -49,15 +49,26 @@ export class DataAccessRead implements DataAccessTypes.IDataRead {
   async getChannelsByTopic(
     topic: string,
     updatedBetween?: DataAccessTypes.ITimestampBoundaries | undefined,
+    page?: number | undefined,
+    pageSize?: number | undefined,
   ): Promise<DataAccessTypes.IReturnGetChannelsByTopic> {
-    return this.getChannelsByMultipleTopics([topic], updatedBetween);
+    return this.getChannelsByMultipleTopics([topic], updatedBetween, page, pageSize);
   }
 
   async getChannelsByMultipleTopics(
     topics: string[],
     updatedBetween?: DataAccessTypes.ITimestampBoundaries,
+    page?: number,
+    pageSize?: number,
   ): Promise<DataAccessTypes.IReturnGetChannelsByTopic> {
-    const result = await this.storage.getTransactionsByTopics(topics);
+    // Validate pagination parameters
+    if (page !== undefined && page < 1) {
+      throw new Error(`Page number must be greater than or equal to 1, but it is ${page}`);
+    }
+    if (pageSize !== undefined && pageSize < 1) {
+      throw new Error(`Page size must be greater than 0, but it is ${pageSize}`);
+    }
+
     const pending = this.pendingStore?.findByTopics(topics) || [];
 
     const pendingItems = pending.map((item) => ({
@@ -73,6 +84,33 @@ export class DataAccessRead implements DataAccessTypes.IDataRead {
       topics: item.topics || [],
     }));
 
+    // Calculate adjusted pagination
+    let adjustedPage = page;
+    let adjustedPageSize = pageSize;
+    let pendingItemsOnCurrentPage = 0;
+    if (page !== undefined && pageSize !== undefined) {
+      const totalPending = pendingItems.length;
+      const itemsPerPage = (page - 1) * pageSize;
+
+      if (totalPending > itemsPerPage) {
+        pendingItemsOnCurrentPage = Math.min(totalPending - itemsPerPage, pageSize);
+        adjustedPageSize = pageSize - pendingItemsOnCurrentPage;
+        adjustedPage = 1;
+        if (adjustedPageSize === 0) {
+          adjustedPageSize = 1;
+          pendingItemsOnCurrentPage--;
+        }
+      } else {
+        adjustedPage = page - Math.floor(totalPending / pageSize);
+      }
+    }
+
+    const result = await this.storage.getTransactionsByTopics(
+      topics,
+      adjustedPage,
+      adjustedPageSize,
+    );
+
     const transactions = result.transactions.concat(...pendingItems);
 
     // list of channels having at least one tx updated during the updatedBetween boundaries
@@ -106,6 +144,17 @@ export class DataAccessRead implements DataAccessTypes.IDataRead {
           },
           {} as Record<string, string[]>,
         ),
+        pagination:
+          page && pageSize
+            ? {
+                total: result.transactions.length + pendingItems.length,
+                page,
+                pageSize,
+                hasMore:
+                  (page - 1) * pageSize + filteredTxs.length - pendingItemsOnCurrentPage <
+                  result.transactions.length,
+              }
+            : undefined,
       },
       result: {
         transactions: filteredTxs.reduce((prev, curr) => {

packages/data-access/src/in-memory-indexer.ts

@@ -55,8 +55,38 @@ export class InMemoryIndexer implements StorageTypes.IIndexer {
     };
   }
 
-  async getTransactionsByTopics(topics: string[]): Promise<StorageTypes.IGetTransactionsResponse> {
-    const channelIds = topics.map((topic) => this.#topicToChannelsIndex.get(topic)).flat();
+  async getTransactionsByTopics(
+    topics: string[],
+    page?: number,
+    pageSize?: number,
+  ): Promise<StorageTypes.IGetTransactionsResponse> {
+    if (page !== undefined && page < 1) {
+      throw new Error('Page must be greater than or equal to 1');
+    }
+    if (pageSize !== undefined && pageSize <= 0) {
+      throw new Error('Page size must be greater than 0');
+    }
+
+    // Efficiently get total count without creating intermediate array
+    const channelIdsSet = new Set(topics.flatMap((topic) => this.#topicToChannelsIndex.get(topic)));
+    const total = channelIdsSet.size;
+    let channelIds = Array.from(channelIdsSet);
+
+    if (page && pageSize) {
+      const start = (page - 1) * pageSize;
+      // Return empty result if page exceeds available data
+      if (start >= total) {
+        return {
+          blockNumber: 0,
+          transactions: [],
+          pagination:
+            page && pageSize
+              ? { total, page, pageSize, hasMore: page * pageSize < total }
+              : undefined,
+        };
+      }
+      channelIds = channelIds.slice(start, start + pageSize);
+    }
     const locations = channelIds
       .map((channel) => this.#channelToLocationsIndex.get(channel))
       .flat();

packages/request-client.js/src/api/request-network.ts

@@ -17,7 +17,7 @@ import {
   SignatureProviderTypes,
   TransactionTypes,
 } from '@requestnetwork/types';
-import { deepCopy, supportedIdentities } from '@requestnetwork/utils';
+import { deepCopy, supportedIdentities, validatePaginationParams } from '@requestnetwork/utils';
 import { CurrencyManager, UnsupportedCurrencyError } from '@requestnetwork/currency';
 import * as Types from '../types';
 import ContentDataExtension from './content-data-extension';
@@ -268,7 +268,12 @@ export default class RequestNetwork {
   public async fromIdentity(
     identity: IdentityTypes.IIdentity,
     updatedBetween?: Types.ITimestampBoundaries,
-    options?: { disablePaymentDetection?: boolean; disableEvents?: boolean },
+    options?: {
+      disablePaymentDetection?: boolean;
+      disableEvents?: boolean;
+      page?: number;
+      pageSize?: number;
+    },
   ): Promise<Request[]> {
     if (!this.supportedIdentities.includes(identity.type)) {
       throw new Error(`${identity.type} is not supported`);
@@ -286,7 +291,12 @@ export default class RequestNetwork {
   public async fromMultipleIdentities(
     identities: IdentityTypes.IIdentity[],
     updatedBetween?: Types.ITimestampBoundaries,
-    options?: { disablePaymentDetection?: boolean; disableEvents?: boolean },
+    options?: {
+      disablePaymentDetection?: boolean;
+      disableEvents?: boolean;
+      page?: number;
+      pageSize?: number;
+    },
   ): Promise<Request[]> {
     const identityNotSupported = identities.find(
       (identity) => !this.supportedIdentities.includes(identity.type),
@@ -309,11 +319,23 @@ export default class RequestNetwork {
   public async fromTopic(
     topic: any,
     updatedBetween?: Types.ITimestampBoundaries,
-    options?: { disablePaymentDetection?: boolean; disableEvents?: boolean },
+    options?: {
+      disablePaymentDetection?: boolean;
+      disableEvents?: boolean;
+      page?: number;
+      pageSize?: number;
+    },
   ): Promise<Request[]> {
+    validatePaginationParams(options?.page, options?.pageSize);
+
     // Gets all the requests indexed by the value of the identity
     const requestsAndMeta: RequestLogicTypes.IReturnGetRequestsByTopic =
-      await this.requestLogic.getRequestsByTopic(topic, updatedBetween);
+      await this.requestLogic.getRequestsByTopic(
+        topic,
+        updatedBetween,
+        options?.page,
+        options?.pageSize,
+      );
     // From the requests of the request-logic layer creates the request objects and gets the payment networks
     const requestPromises = requestsAndMeta.result.requests.map(
       async (requestFromLogic: {
@@ -361,11 +383,23 @@ export default class RequestNetwork {
   public async fromMultipleTopics(
     topics: any[],
     updatedBetween?: Types.ITimestampBoundaries,
-    options?: { disablePaymentDetection?: boolean; disableEvents?: boolean },
+    options?: {
+      disablePaymentDetection?: boolean;
+      disableEvents?: boolean;
+      page?: number;
+      pageSize?: number;
+    },
   ): Promise<Request[]> {
+    validatePaginationParams(options?.page, options?.pageSize);
+
     // Gets all the requests indexed by the value of the identity
     const requestsAndMeta: RequestLogicTypes.IReturnGetRequestsByTopic =
-      await this.requestLogic.getRequestsByMultipleTopics(topics, updatedBetween);
+      await this.requestLogic.getRequestsByMultipleTopics(
+        topics,
+        updatedBetween,
+        options?.page,
+        options?.pageSize,
+      );
 
     // From the requests of the request-logic layer creates the request objects and gets the payment networks
     const requestPromises = requestsAndMeta.result.requests.map(

packages/request-client.js/src/http-data-access.ts

@@ -1,7 +1,7 @@
 import { ClientTypes, DataAccessTypes } from '@requestnetwork/types';
 import { EventEmitter } from 'events';
 import httpConfigDefaults from './http-config-defaults';
-import { normalizeKeccak256Hash, retry } from '@requestnetwork/utils';
+import { normalizeKeccak256Hash, retry, validatePaginationParams } from '@requestnetwork/utils';
 import { stringify } from 'qs';
 import { utils } from 'ethers';
 
@@ -175,11 +175,19 @@ export default class HttpDataAccess implements DataAccessTypes.IDataAccess {
   public async getChannelsByTopic(
     topic: string,
     updatedBetween?: DataAccessTypes.ITimestampBoundaries,
+    page?: number,
+    pageSize?: number,
   ): Promise<DataAccessTypes.IReturnGetChannelsByTopic> {
-    return await this.fetchAndRetry('/getChannelsByTopic', {
+    validatePaginationParams(page, pageSize);
+
+    const params = {
       topic,
       updatedBetween,
-    });
+      ...(page !== undefined && { page }),
+      ...(pageSize !== undefined && { pageSize }),
+    };
+
+    return await this.fetchAndRetry('/getChannelsByTopic', params);
   }
 
   /**
@@ -191,10 +199,16 @@ export default class HttpDataAccess implements DataAccessTypes.IDataAccess {
   public async getChannelsByMultipleTopics(
     topics: string[],
     updatedBetween?: DataAccessTypes.ITimestampBoundaries,
+    page?: number,
+    pageSize?: number,
   ): Promise<DataAccessTypes.IReturnGetChannelsByTopic> {
+    validatePaginationParams(page, pageSize);
+
     return await this.fetchAndRetry('/getChannelsByMultipleTopics', {
       topics,
       updatedBetween,
+      page,
+      pageSize,
     });
   }
 

packages/request-client.js/test/api/request-network.test.ts

@@ -95,7 +95,12 @@ describe('api/request-network', () => {
     it('can get requests with payment network fromIdentity', async () => {
       const mockDataAccessWithTxs: DataAccessTypes.IDataAccess = {
         ...mockDataAccess,
-        async getChannelsByTopic(topic: string): Promise<any> {
+        async getChannelsByTopic(
+          topic: string,
+          updatedBetween?: DataAccessTypes.ITimestampBoundaries,
+          page?: number,
+          pageSize?: number,
+        ): Promise<any> {
           expect(topic).toBe('01f1a21ab419611dbf492b3136ac231c8773dc897ee0eb5167ef2051a39e685e76');
           return {
             meta: {
@@ -137,7 +142,14 @@ describe('api/request-network', () => {
       };
 
       const requestnetwork = new RequestNetwork({ dataAccess: mockDataAccessWithTxs });
-      const requests: Request[] = await requestnetwork.fromIdentity(TestData.payee.identity);
+      const requests: Request[] = await requestnetwork.fromIdentity(
+        TestData.payee.identity,
+        undefined,
+        {
+          page: 1,
+          pageSize: 10,
+        },
+      );
 
       expect(requests.length).toBe(2);
       expect(requests[0].requestId).toBe(TestData.actionRequestId);
@@ -201,7 +213,12 @@ describe('api/request-network', () => {
     it('can get requests with payment network from multiple Identities', async () => {
       const mockDataAccessWithTxs: DataAccessTypes.IDataAccess = {
         ...mockDataAccess,
-        async getChannelsByMultipleTopics(topics: [string]): Promise<any> {
+        async getChannelsByMultipleTopics(
+          topics: [string],
+          updatedBetween?: DataAccessTypes.ITimestampBoundaries,
+          page?: number,
+          pageSize?: number,
+        ): Promise<any> {
           expect(topics).toEqual([
             '01f1a21ab419611dbf492b3136ac231c8773dc897ee0eb5167ef2051a39e685e76',
           ]);
@@ -245,9 +262,14 @@ describe('api/request-network', () => {
       };
 
       const requestnetwork = new RequestNetwork({ dataAccess: mockDataAccessWithTxs });
-      const requests: Request[] = await requestnetwork.fromMultipleIdentities([
-        TestData.payee.identity,
-      ]);
+      const requests: Request[] = await requestnetwork.fromMultipleIdentities(
+        [TestData.payee.identity],
+        undefined,
+        {
+          page: 1,
+          pageSize: 10,
+        },
+      );
 
       expect(requests.length).toBe(2);
       expect(requests[0].requestId).toBe(TestData.actionRequestId);

packages/request-logic/src/request-logic.ts

@@ -346,13 +346,17 @@ export default class RequestLogic implements RequestLogicTypes.IRequestLogic {
   public async getRequestsByTopic(
     topic: string,
     updatedBetween?: RequestLogicTypes.ITimestampBoundaries,
+    page?: number,
+    pageSize?: number,
   ): Promise<RequestLogicTypes.IReturnGetRequestsByTopic> {
     // hash all the topics
     const hashedTopic = MultiFormat.serialize(normalizeKeccak256Hash(topic));
 
     const getChannelsResult = await this.transactionManager.getChannelsByTopic(
       hashedTopic,
       updatedBetween,
+      page,
+      pageSize,
     );
     return this.computeMultipleRequestFromChannels(getChannelsResult);
   }
@@ -365,6 +369,8 @@ export default class RequestLogic implements RequestLogicTypes.IRequestLogic {
   public async getRequestsByMultipleTopics(
     topics: string[],
     updatedBetween?: RequestLogicTypes.ITimestampBoundaries,
+    page?: number,
+    pageSize?: number,
   ): Promise<RequestLogicTypes.IReturnGetRequestsByTopic> {
     // hash all the topics
     const hashedTopics = topics.map((topic) =>
@@ -374,6 +380,8 @@ export default class RequestLogic implements RequestLogicTypes.IRequestLogic {
     const getChannelsResult = await this.transactionManager.getChannelsByMultipleTopics(
       hashedTopics,
       updatedBetween,
+      page,
+      pageSize,
     );
     return this.computeMultipleRequestFromChannels(getChannelsResult);
   }