feat: pagination

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 | undefined,
+    pageSize?: number | undefined,
   ): 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,17 +49,28 @@ 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);
-    const pending = this.pendingStore?.findByTopics(topics) || [];
+    // Validate pagination parameters
+    if (page !== undefined && page < 1) {
+      throw new Error('Page number must be greater than or equal to 1');
+    }
+    if (pageSize !== undefined && pageSize <= 0) {
+      throw new Error('Page size must be positive');
+    }
 
+    // Get pending items first
+    const pending = this.pendingStore?.findByTopics(topics) || [];
     const pendingItems = pending.map((item) => ({
       hash: item.storageResult.id,
       channelId: item.channelId,
@@ -73,20 +84,55 @@ export class DataAccessRead implements DataAccessTypes.IDataRead {
       topics: item.topics || [],
     }));
 
-    const transactions = result.transactions.concat(...pendingItems);
+    // Adjust pagination to account for pending items
+    let adjustedPage = page;
+    let adjustedPageSize = pageSize;
+    if (page !== undefined && pageSize !== undefined) {
+      if (pendingItems.length >= (page - 1) * pageSize) {
+        // If pending items fill previous pages
+        adjustedPage = 0;
+        adjustedPageSize = 0;
+      } else {
+        // Adjust page size to account for pending items included
+        const pendingItemsInPreviousPages = Math.min(pendingItems.length, (page - 1) * pageSize);
+        const pendingItemsInCurrentPage = Math.min(
+          pendingItems.length - pendingItemsInPreviousPages,
+          pageSize,
+        );
+        adjustedPageSize = pageSize - pendingItemsInCurrentPage;
+        adjustedPage = Math.floor(
+          ((page - 1) * pageSize - pendingItemsInPreviousPages) / adjustedPageSize,
+        );
+      }
+    }
 
-    // list of channels having at least one tx updated during the updatedBetween boundaries
-    const channels = (
-      updatedBetween
-        ? transactions.filter(
-            (tx) =>
-              tx.blockTimestamp >= (updatedBetween.from || 0) &&
-              tx.blockTimestamp <= (updatedBetween.to || Number.MAX_SAFE_INTEGER),
-          )
-        : transactions
-    ).map((x) => x.channelId);
+    // Fetch transactions from storage with adjusted pagination
+    const result = await this.storage.getTransactionsByTopics(
+      topics,
+      adjustedPage,
+      adjustedPageSize,
+    );
+
+    // Combine pending and stored transactions
+    const transactions = [...pendingItems, ...result.transactions];
+
+    // Proceed with filtering and mapping as per existing logic
+    const filteredTxs = transactions.filter((tx) => {
+      if (!updatedBetween) return true;
+      return (
+        tx.blockTimestamp >= (updatedBetween.from || 0) &&
+        tx.blockTimestamp <= (updatedBetween.to || Number.MAX_SAFE_INTEGER)
+      );
+    });
+
+    const finalTransactions = filteredTxs.reduce((prev, curr) => {
+      if (!prev[curr.channelId]) {
+        prev[curr.channelId] = [];
+      }
+      prev[curr.channelId].push(this.toTimestampedTransaction(curr));
+      return prev;
+    }, {} as DataAccessTypes.ITransactionsByChannelIds);
 
-    const filteredTxs = transactions.filter((tx) => channels.includes(tx.channelId));
     return {
       meta: {
         storageMeta: filteredTxs.reduce(
@@ -106,15 +152,15 @@ export class DataAccessRead implements DataAccessTypes.IDataRead {
           },
           {} as Record<string, string[]>,
         ),
+        pagination: {
+          page: page,
+          pageSize: pageSize,
+          total: filteredTxs.length,
+          hasMore: filteredTxs.length > (page || 0) * (pageSize || 0),
+        },
       },
       result: {
-        transactions: filteredTxs.reduce((prev, curr) => {
-          if (!prev[curr.channelId]) {
-            prev[curr.channelId] = [];
-          }
-          prev[curr.channelId].push(this.toTimestampedTransaction(curr));
-          return prev;
-        }, {} as DataAccessTypes.ITransactionsByChannelIds),
+        transactions: finalTransactions,
       },
     };
   }

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

@@ -265,7 +265,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`);
@@ -283,7 +288,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),
@@ -306,11 +316,21 @@ 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[]> {
     // 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: {
@@ -358,11 +378,21 @@ 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[]> {
     // 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

@@ -175,10 +175,21 @@ export default class HttpDataAccess implements DataAccessTypes.IDataAccess {
   public async getChannelsByTopic(
     topic: string,
     updatedBetween?: DataAccessTypes.ITimestampBoundaries,
+    page?: number,
+    pageSize?: number,
   ): Promise<DataAccessTypes.IReturnGetChannelsByTopic> {
+    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');
+    }
+
     return await this.fetchAndRetry('/getChannelsByTopic', {
       topic,
       updatedBetween,
+      page,
+      pageSize,
     });
   }
 
@@ -191,10 +202,14 @@ export default class HttpDataAccess implements DataAccessTypes.IDataAccess {
   public async getChannelsByMultipleTopics(
     topics: string[],
     updatedBetween?: DataAccessTypes.ITimestampBoundaries,
+    page?: number,
+    pageSize?: number,
   ): Promise<DataAccessTypes.IReturnGetChannelsByTopic> {
     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);
   }