chore: update docker-compose and improve data access logic

- Modified docker-compose.yml to adjust the build context and dockerfile path for the graph-deploy service.
- Enhanced data access logic in data-read.ts to streamline transaction filtering and pagination.
- Updated GraphQL queries in queries.ts to correctly retrieve transactions by topics and ensure proper ordering.
- Refactored subgraph-client.ts to handle channel-based transaction retrieval, improving data organization and pagination handling.

These changes improve the overall structure and functionality of the data access layer.

Author: rodrigopavezi

docker-compose.yml

@@ -52,16 +52,16 @@ services:
     restart: on-failure:20
   graph-deploy:
     build:
-      context: https://github.com/RequestNetwork/docker-images.git#main
-      dockerfile: request-subgraph-storage/Dockerfile
+      context: https://github.com/RequestNetwork/docker-images.git#main:request-subgraph-storage
+      dockerfile: ./Dockerfile
     depends_on:
       - ipfs
       - postgres
       - graph-node
       - ganache
     environment:
       GRAPH_NODE: 'http://graph-node:8020'
-      IPFS_HOST: 'ipfs:5001'
+      IPFS_HOST: 'http://ipfs:5001'
       KEEP_ALIVE: 0
       SUBGRAPH_FILE: 'subgraph-private.yaml'
     restart: on-failure:20

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

@@ -70,6 +70,7 @@ export class DataAccessRead implements DataAccessTypes.IDataRead {
     }
 
     const pending = this.pendingStore?.findByTopics(topics) || [];
+
     const pendingItems = pending.map((item) => ({
       hash: item.storageResult.id,
       channelId: item.channelId,
@@ -112,33 +113,25 @@ export class DataAccessRead implements DataAccessTypes.IDataRead {
 
     const transactions = result.transactions.concat(...pendingItems);
 
-    // Apply timestamp filtering FIRST
-    const filteredTransactions = updatedBetween
-      ? transactions.filter(
-          (tx) =>
-            tx.blockTimestamp >= (updatedBetween.from || 0) &&
-            tx.blockTimestamp <= (updatedBetween.to || Number.MAX_SAFE_INTEGER),
-        )
-      : transactions;
-
-    // Then get unique channels from filtered transactions
-    const channels = [...new Set(filteredTransactions.map((x) => x.channelId))];
+    // 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);
 
-    // Get all transactions for these channels
     const filteredTxs = transactions.filter((tx) => channels.includes(tx.channelId));
-
-    // Apply pagination to the filtered results
-    const start = ((page || 1) - 1) * (pageSize || filteredTxs.length);
-    const end = start + (pageSize || filteredTxs.length);
-    const paginatedTxs = filteredTxs.slice(start, end);
-
     return {
       meta: {
-        storageMeta: paginatedTxs.reduce((acc, tx) => {
+        storageMeta: filteredTxs.reduce((acc, tx) => {
           acc[tx.channelId] = [this.toStorageMeta(tx, result.blockNumber, this.network)];
           return acc;
         }, {} as Record<string, StorageTypes.IEntryMetadata[]>),
-        transactionsStorageLocation: paginatedTxs.reduce((prev, curr) => {
+        transactionsStorageLocation: filteredTxs.reduce((prev, curr) => {
           if (!prev[curr.channelId]) {
             prev[curr.channelId] = [];
           }
@@ -148,15 +141,17 @@ export class DataAccessRead implements DataAccessTypes.IDataRead {
         pagination:
           page && pageSize
             ? {
-                total: filteredTxs.length,
+                total: result.transactions.length + pendingItems.length,
                 page,
                 pageSize,
-                hasMore: end < filteredTxs.length,
+                hasMore:
+                  (page - 1) * pageSize + filteredTxs.length - pendingItemsOnCurrentPage <
+                  result.transactions.length,
               }
             : undefined,
       },
       result: {
-        transactions: paginatedTxs.reduce((prev, curr) => {
+        transactions: filteredTxs.reduce((prev, curr) => {
           if (!prev[curr.channelId]) {
             prev[curr.channelId] = [];
           }

packages/thegraph-data-access/src/queries.ts

@@ -79,14 +79,17 @@ ${TransactionsBodyFragment}
 
 query GetTransactionsByTopics($topics: [String!]!, $first: Int!, $skip: Int!) {
   ${metaQueryBody}
-  transactions(
+  channels(
     where: { topics_contains: $topics }
-    first: $first
-    skip: $skip
-    orderBy: blockTimestamp
-    orderDirection: asc
-  ) {
-    ...TransactionsBody
+  ){
+    transactions(
+      orderBy: blockTimestamp, 
+      orderDirection: asc
+      first: $first
+      skip: $skip
+    ) {
+      ...TransactionsBody
+    }
   }
 }`;
 

packages/thegraph-data-access/src/subgraph-client.ts

@@ -71,27 +71,28 @@ export class SubgraphClient implements StorageTypes.IIndexer {
     const effectivePage = page ?? 1;
     const skip = (effectivePage - 1) * effectivePageSize;
 
-    const topicsArray = Array.isArray(topics) ? topics : [topics];
-
-    const response = await this.graphql.request<Meta & { transactions: Transaction[] }>(
-      GetTransactionsByTopics,
-      {
-        topics: topicsArray,
-        first: effectivePageSize,
-        skip,
-      },
-    );
+    const { _meta, channels } = await this.graphql.request<
+      Meta & { channels: { transactions: Transaction[] }[] }
+    >(GetTransactionsByTopics, {
+      topics,
+      first: effectivePageSize,
+      skip,
+    });
 
-    const indexedTransactions = response.transactions.map(this.toIndexedTransaction);
+    const transactionsByChannel = channels
+      .map(({ transactions }) => transactions)
+      .flat()
+      .sort((a, b) => a.blockTimestamp - b.blockTimestamp);
 
+    const indexedTransactions = transactionsByChannel.map(this.toIndexedTransaction);
     return {
       transactions: indexedTransactions,
-      blockNumber: response._meta.block.number,
+      blockNumber: _meta.block.number,
       pagination: {
         page: effectivePage,
         pageSize: effectivePageSize,
         total: indexedTransactions.length,
-        hasMore: indexedTransactions.length === effectivePageSize,
+        hasMore: skip + effectivePageSize < indexedTransactions.length,
       },
     };
   }