feat(request-node): use subgraph for confirmation

packages/data-access/README.md

@@ -139,10 +139,6 @@ Transactions are indexed with `topics`. When persisting a transaction on data-ac
 
 To save on costs, transactions are batched together. This is the responsibility of Data Access layer. This package creates `blocks` that are collections of `transactions`
 
-### Cache
-
-In order to speed up recovery of transactions, data-access has a local cache of topics=>transaction. This is the reason why this package should be initialized with `dataAccess.initialize()` before being operational. This cache can be persisted in a file using a [Keyv store](https://github.com/lukechilds/keyv#official-storage-adapters).
-
 ### Synchronization
 
 Blocks can be added into the storage by other peers running their own data-access instance. Therefore, to remain consistent with the global state of the network, this package need to synchronize with these new blocks.`dataAccess.synchronizeNewDataIds()` allows to synchronize manually with all unsynchronized blocks. The synchronization can also be done automatically, `dataAccess.startAutoSynchronization()` allows to automatically synchronize with new blocks, the interval time between each synchronization can be defined in data-access constructor. `dataAccess.stopAutoSynchronization()` allows to stop automatic synchronization.

packages/integration-test/test/scheduled/native-token.test.ts

@@ -1,13 +1,12 @@
 import { PaymentNetworkFactory } from '@requestnetwork/payment-detection';
 import { ExtensionTypes, RequestLogicTypes } from '@requestnetwork/types';
-import { PnReferenceBased } from '@requestnetwork/types/dist/extension-types';
 import { AdvancedLogic } from '@requestnetwork/advanced-logic';
 import { CurrencyManager } from '@requestnetwork/currency';
 import { omit } from 'lodash';
 
 const advancedLogic = new AdvancedLogic(new CurrencyManager(CurrencyManager.getDefaultList()));
 
-const createCreationActionParams: PnReferenceBased.ICreationParameters = {
+const createCreationActionParams: ExtensionTypes.PnReferenceBased.ICreationParameters = {
   paymentAddress: 'payment.testnet',
   salt: 'a1a2a3a4a5a6a7a8',
   paymentNetworkName: 'aurora-testnet',

packages/payment-processor/src/payment/batch-conversion-proxy.ts

@@ -29,7 +29,6 @@ import { IPreparedTransaction } from './prepared-transaction';
 import { IConversionPaymentSettings } from './index';
 import { getConversionPathForErc20Request } from './any-to-erc20-proxy';
 import { checkErc20Allowance, encodeApproveAnyErc20 } from './erc20';
-import { IState } from 'types/dist/extension-types';
 import { CurrencyDefinition, CurrencyManager, ICurrencyManager } from '@requestnetwork/currency';
 import {
   BatchPaymentNetworks,
@@ -128,7 +127,7 @@ const computeRequestDetails = ({
   extension,
 }: {
   enrichedRequest: EnrichedRequest;
-  extension: IState<any> | undefined;
+  extension: ExtensionTypes.IState<any> | undefined;
 }) => {
   const paymentNetworkId = enrichedRequest.paymentNetworkId;
   const allowedCurrencies = mapPnToAllowedCurrencies[paymentNetworkId];
@@ -177,7 +176,7 @@ function encodePayBatchConversionRequest(
     'pn-eth-fee-proxy-contract': [],
   };
 
-  const requestExtensions: Record<BatchPaymentNetworks, IState<any> | undefined> = {
+  const requestExtensions: Record<BatchPaymentNetworks, ExtensionTypes.IState<any> | undefined> = {
     'pn-any-to-erc20-proxy': undefined,
     'pn-any-to-eth-proxy': undefined,
     'pn-erc20-fee-proxy-contract': undefined,

packages/request-node/README.md

@@ -242,8 +242,6 @@ Default values correspond to the basic configuration used to run a server in a t
 - `--storageConcurrency` Maximum number of concurrent calls to Ethereum or IPFS
   - Default value: `'200'`
   - Environment variable name: `$STORAGE_MAX_CONCURRENCY`
-- `--initializationStorageFilePath` Path to a file to persist the ethereum metadata and transaction index for faster initialization
-  - Environment variable name: `$INITIALIZATION_STORAGE_FILE_PATH`
 - `--logLevel` The maximum level of messages we will log
   - Environment variable name: `$LOG_LEVEL`
   - Available levels: ERROR, WARN, INFO and DEBUG

packages/request-node/package.json

@@ -58,8 +58,6 @@
     "graphql-request": "6.1.0",
     "http-shutdown": "1.2.2",
     "http-status-codes": "2.1.4",
-    "keyv": "4.0.3",
-    "keyv-file": "0.2.0",
     "shelljs": "0.8.5",
     "tslib": "2.5.0",
     "yargs": "17.6.2"
@@ -68,7 +66,6 @@
     "@types/cors": "2.8.9",
     "@types/express": "4.17.17",
     "@types/jest": "29.5.6",
-    "@types/keyv": "3.1.1",
     "@types/node": "18.11.9",
     "@types/supertest": "2.0.10",
     "@types/yargs": "17.0.14",

packages/request-node/src/config.ts

@@ -153,16 +153,6 @@ export const getLogLevel = (): LogTypes.LogLevel => {
 /** logMode defines the log format to display: `human` is a more readable log, `machine` is better for parsing */
 export const getLogMode = makeOption('logMode', 'LOG_MODE', defaultValues.log.mode);
 
-/**
- * Get the initialization storage (a json-like file) path.
- * @returns the path to the json-like file that stores the initialization data (ethereum metadata and transaction index).
- */
-export const getInitializationStorageFilePath = makeOption(
-  'initializationStorageFilePath',
-  'INITIALIZATION_STORAGE_FILE_PATH',
-  '',
-);
-
 /**
  * Get the delay to wait before sending a timeout when performing a persistTransaction request
  * persistTransaction is called when a request is created or updated and need to wait for Ethereum to commit the transaction
@@ -240,7 +230,6 @@ export const getConfigDisplay = (): string => {
   TheGraph url: ${getGraphNodeUrl()}
   IPFS url: ${getIpfsUrl()}
   IPFS timeout: ${getIpfsTimeout()}
-  Initialization storage path: ${getInitializationStorageFilePath()}
   Storage block confirmations: ${getBlockConfirmations()}
 `;
 };

packages/request-node/src/dataAccess.ts

@@ -1,29 +1,19 @@
 import { providers, Wallet } from 'ethers';
 import { NonceManager } from '@ethersproject/experimental';
-import { DataAccessTypes, LogTypes, StorageTypes } from '@requestnetwork/types';
-import { EvmChains } from '@requestnetwork/currency';
+import { CurrencyTypes, DataAccessTypes, LogTypes, StorageTypes } from '@requestnetwork/types';
 
 import * as config from './config';
 import { TheGraphDataAccess } from '@requestnetwork/thegraph-data-access';
 import { PendingStore } from '@requestnetwork/data-access';
-import {
-  EthereumStorage,
-  EthereumTransactionSubmitter,
-  getEthereumStorageNetworkNameFromId,
-} from '@requestnetwork/ethereum-storage';
+import { EthereumStorage, EthereumTransactionSubmitter } from '@requestnetwork/ethereum-storage';
 
 export function getDataAccess(
+  network: CurrencyTypes.EvmChainName,
   ipfsStorage: StorageTypes.IIpfsStorage,
   logger: LogTypes.ILogger,
 ): DataAccessTypes.IDataAccess {
   const graphNodeUrl = config.getGraphNodeUrl();
 
-  const network = getEthereumStorageNetworkNameFromId(config.getStorageNetworkId()) as any;
-  if (!network) {
-    throw new Error(`Storage network not supported: ${config.getStorageNetworkId()}`);
-  }
-  EvmChains.assertChainSupported(network);
-
   const wallet = Wallet.fromMnemonic(config.getMnemonic()).connect(
     new providers.StaticJsonRpcProvider(config.getStorageWeb3ProviderUrl()),
   );

packages/request-node/src/dataStorage.ts

@@ -1,7 +1,7 @@
 import * as config from './config';
 
 import { IpfsStorage } from '@requestnetwork/ethereum-storage';
-import { LogTypes, StorageTypes } from 'types/dist';
+import { LogTypes, StorageTypes } from '@requestnetwork/types';
 
 export function getDataStorage(logger: LogTypes.ILogger): StorageTypes.IIpfsStorage {
   return new IpfsStorage({

packages/request-node/src/request/confirmedTransactionStore.ts

@@ -1,50 +1,52 @@
-import { DataAccessTypes } from '@requestnetwork/types';
-import Keyv, { Store } from 'keyv';
+import { DataAccessTypes, StorageTypes } from '@requestnetwork/types';
+import { SubgraphClient } from '@requestnetwork/thegraph-data-access';
 
 /**
- * Class for storing confirmed transactions information
- * When 'confirmed' event is received from a 'persistTransaction', the event data are stored.
- * The client can call the getConfirmed entry point, to get the confirmed event.
+ * Class for storing confirmed transaction information
+ * When 'confirmed' event is received from a 'persistTransaction', the event data is
+ * stored and indexed by the storage subgraph. The client can call the
+ * getConfirmedTransaction endpoint, to get the confirmed event.
  */
 export default class ConfirmedTransactionStore {
-  private store: Keyv<DataAccessTypes.IReturnPersistTransactionRaw | Error>;
-
   /**
    * Confirmed transactions store constructor
    */
-  constructor(store?: Store<DataAccessTypes.IReturnPersistTransaction | Error>) {
-    this.store = new Keyv<DataAccessTypes.IReturnPersistTransaction | Error>({
-      namespace: 'ConfirmedTransactions',
-      store,
-    });
-  }
+  constructor(
+    private readonly subgraphClient: SubgraphClient,
+    private readonly networkName: string,
+  ) {}
 
   public async getConfirmedTransaction(
     transactionHash: string,
-  ): Promise<DataAccessTypes.IReturnPersistTransactionRaw | Error | undefined> {
-    return this.store.get(transactionHash);
-  }
-
-  /**
-   * Stores the result of a transaction confirmation
-   *
-   * @param transactionHash hash of the transaction
-   * @param result result of the event "confirmed"
-   */
-  public async addConfirmedTransaction(
-    transactionHash: string,
-    result: DataAccessTypes.IReturnPersistTransactionRaw,
-  ): Promise<void> {
-    await this.store.set(transactionHash, result);
-  }
-
-  /**
-   * Stores the error
-   *
-   * @param transactionHash hash of the transaction
-   * @param error error of the event "error"
-   */
-  public async addFailedTransaction(transactionHash: string, error: Error): Promise<void> {
-    await this.store.set(transactionHash, error);
+  ): Promise<DataAccessTypes.IReturnPersistTransactionRaw | undefined> {
+    const { transactions, blockNumber } =
+      await this.subgraphClient.getTransactionsByDataHash(transactionHash);
+    if (transactions.length === 0) {
+      return;
+    }
+    const transaction = transactions[0];
+    return {
+      meta: {
+        transactionStorageLocation: transaction.hash,
+        topics: transaction.topics,
+        storageMeta: {
+          state: StorageTypes.ContentState.CONFIRMED,
+          storageType: StorageTypes.StorageSystemType.ETHEREUM_IPFS,
+          timestamp: transaction.blockTimestamp,
+          ethereum: {
+            blockConfirmation: blockNumber - transaction.blockNumber,
+            blockTimestamp: transaction.blockTimestamp,
+            blockNumber: transaction.blockNumber,
+            networkName: this.networkName,
+            smartContractAddress: transaction.smartContractAddress,
+            transactionHash: transaction.transactionHash,
+          },
+          ipfs: {
+            size: Number(transaction.size),
+          },
+        },
+      },
+      result: {},
+    };
   }
 }

packages/request-node/src/request/getConfirmedTransactionHandler.ts

@@ -3,7 +3,7 @@ import { Request, Response } from 'express';
 import { StatusCodes } from 'http-status-codes';
 import ConfirmedTransactionStore from './confirmedTransactionStore';
 
-export default class getConfirmedTransactionHandler {
+export default class GetConfirmedTransactionHandler {
   constructor(
     private logger: LogTypes.ILogger,
     private store: ConfirmedTransactionStore,

packages/request-node/src/request/persistTransaction.ts

@@ -3,7 +3,6 @@ import { Request, Response } from 'express';
 import { StatusCodes } from 'http-status-codes';
 import { getPersistTransactionTimeout } from '../config';
 
-import ConfirmedTransactionStore from './confirmedTransactionStore';
 import { normalizeKeccak256Hash } from '@requestnetwork/utils';
 
 /**
@@ -14,7 +13,6 @@ export default class PersistTransactionHandler {
    * Persist transaction constructor
    */
   constructor(
-    private confirmedTransactionStore: ConfirmedTransactionStore,
     private dataAccess: DataAccessTypes.IDataWrite,
     private logger: LogTypes.ILogger,
   ) {
@@ -68,12 +66,7 @@ export default class PersistTransactionHandler {
         clientRequest.body.topics,
       );
 
-      // when the transaction is confirmed, store the information to be served when requested
-      dataAccessResponse.on('confirmed', async (dataAccessConfirmedResponse) => {
-        await this.confirmedTransactionStore.addConfirmedTransaction(
-          transactionHash.value,
-          dataAccessConfirmedResponse,
-        );
+      dataAccessResponse.on('confirmed', async () => {
         this.logger.info(`Transaction confirmed: ${transactionHash.value}`, [
           'metric',
           'successRate',
@@ -82,10 +75,6 @@ export default class PersistTransactionHandler {
 
       // when the transaction fails, log an error
       dataAccessResponse.on('error', async (e: unknown) => {
-        await this.confirmedTransactionStore.addFailedTransaction(
-          transactionHash.value,
-          e as Error,
-        );
         this.logger.error(`persistTransaction error: ${e}\n
           transactionHash: ${transactionHash.value}, channelId: ${
             clientRequest.body.channelId

packages/request-node/src/requestNode.ts

@@ -4,7 +4,6 @@ import cors from 'cors';
 import { Server } from 'http';
 import express, { NextFunction, Request, Response } from 'express';
 import { StatusCodes } from 'http-status-codes';
-import { Store } from 'keyv';
 import ConfirmedTransactionStore from './request/confirmedTransactionStore';
 import GetConfirmedTransactionHandler from './request/getConfirmedTransactionHandler';
 import GetTransactionsByChannelIdHandler from './request/getTransactionsByChannelId';
@@ -39,7 +38,6 @@ export class RequestNode {
   private initialized: boolean;
   private logger: LogTypes.ILogger;
   private persistTransactionHandler: PersistTransactionHandler;
-  private confirmedTransactionStore: ConfirmedTransactionStore;
   private requestNodeVersion: string;
 
   private getTransactionsByChannelIdHandler: GetTransactionsByChannelIdHandler;
@@ -55,18 +53,17 @@ export class RequestNode {
   constructor(
     dataAccess: DataAccessTypes.IDataAccess,
     ipfsStorage: StorageTypes.IIpfsStorage,
-    store?: Store<DataAccessTypes.IReturnPersistTransaction>,
+    confirmedTransactionStore: ConfirmedTransactionStore,
     logger?: LogTypes.ILogger,
   ) {
     this.initialized = false;
 
     this.logger = logger || new SimpleLogger();
     this.dataAccess = dataAccess;
 
-    this.confirmedTransactionStore = new ConfirmedTransactionStore(store);
     this.getConfirmedTransactionHandler = new GetConfirmedTransactionHandler(
       this.logger,
-      this.confirmedTransactionStore,
+      confirmedTransactionStore,
     );
     this.getTransactionsByChannelIdHandler = new GetTransactionsByChannelIdHandler(
       this.logger,
@@ -75,11 +72,7 @@ export class RequestNode {
     this.getChannelByTopicHandler = new GetChannelsByTopicHandler(this.logger, this.dataAccess);
     this.getStatusHandler = new GetStatusHandler(this.logger, this.dataAccess);
     this.ipfsAddHandler = new IpfsAddHandler(this.logger, ipfsStorage);
-    this.persistTransactionHandler = new PersistTransactionHandler(
-      this.confirmedTransactionStore,
-      this.dataAccess,
-      this.logger,
-    );
+    this.persistTransactionHandler = new PersistTransactionHandler(this.dataAccess, this.logger);
 
     this.express = express();
     this.mountRoutes();

packages/request-node/src/server.ts

@@ -3,22 +3,38 @@ import { Logger } from './logger';
 import withShutdown from 'http-shutdown';
 import { RequestNode } from './requestNode';
 import { getDataAccess } from './dataAccess';
-import KeyvFile from 'keyv-file';
 import { getDataStorage } from './dataStorage';
+import ConfirmedTransactionStore from './request/confirmedTransactionStore';
+import { EvmChains } from '@requestnetwork/currency';
+import { getEthereumStorageNetworkNameFromId } from '@requestnetwork/ethereum-storage';
+import { SubgraphClient } from '@requestnetwork/thegraph-data-access';
 
 // Initialize the node logger
 const logger = new Logger(config.getLogLevel(), config.getLogMode());
 
+const getNetwork = () => {
+  const network = getEthereumStorageNetworkNameFromId(config.getStorageNetworkId()) as any;
+  if (!network) {
+    throw new Error(`Storage network not supported: ${config.getStorageNetworkId()}`);
+  }
+  EvmChains.assertChainSupported(network);
+  return network;
+};
+
 export const getRequestNode = (): RequestNode => {
-  const initializationStoragePath = config.getInitializationStorageFilePath();
-  const store = initializationStoragePath
-    ? new KeyvFile({
-        filename: initializationStoragePath,
-      })
-    : undefined;
+  const network = getNetwork();
   const storage = getDataStorage(logger);
-  const dataAccess = getDataAccess(storage, logger);
-  return new RequestNode(dataAccess, storage, store, logger);
+  const dataAccess = getDataAccess(network, storage, logger);
+
+  // we access the subgraph client directly, not through the data access,
+  // because this feature is specific to RN use with Request Node. Without a node,
+  // the confirmation process would be different, so this doesn't fit in the data access layer
+  const confirmedTransactionStore = new ConfirmedTransactionStore(
+    new SubgraphClient(config.getGraphNodeUrl()),
+    network,
+  );
+
+  return new RequestNode(dataAccess, storage, confirmedTransactionStore, logger);
 };
 
 export const startNode = async (): Promise<void> => {

packages/request-node/test/getChannelsByTopic.test.ts

@@ -17,7 +17,7 @@ const otherTopics = [`01eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee${tim
 );
 const nonExistentTopic = '010000000000000000000000000000000000000000000000000000000000000000';
 const transactionData = {
-  data: 'this is sample data for a transaction to test getChannelsByTopic',
+  data: `this is sample data for a transaction to test getChannelsByTopic ${Date.now()}`,
 };
 const otherTransactionData = {
   data: 'this is other sample data for a transaction to test getChannelsByTopic',