[WIP] Use utils.Nitro to setup Nitro client in watcher

environments/local.toml

@@ -19,6 +19,9 @@
   # Use -1 for skipping check on block range.
   maxEventsBlockRange = -1
 
+  # Flag to specify whether RPC endpoint supports block hash as block tag parameter
+  rpcSupportsBlockHashParam = false
+
   [server.p2p]
     enableRelay = true
     enablePeer = false
@@ -39,11 +42,30 @@
       peerIdFile = ''
       enableDebugInfo = true
       enableL2Txs = false
+      pingInterval = 4000
+      pingTimeout = 1500
 
       [server.p2p.peer.l2TxsConfig]
         privateKey = ''
         contractAddress = ''
 
+      [server.p2p.nitro]
+        store = ''
+        privateKey = ''
+        chainPrivateKey = ''
+
+        [server.p2p.nitro.payments]
+          ratesFile = ''
+          requestTimeoutInSecs = 10
+
+          [server.p2p.nitro.payments.cache]
+            maxAccounts = 1000
+            accountTTLInSecs = 1800
+            maxVouchersPerAccount = 1000
+            voucherTTLInSecs = 300
+            maxPaymentChannels = 10000
+            paymentChannelTTLInSecs = 1800
+
 [metrics]
   host = "127.0.0.1"
   port = 9000

environments/rates.toml

@@ -0,0 +1,14 @@
+freeQueriesLimit = 10
+
+freeQueriesList = []
+
+[queries]
+  multiNonce = ''
+  _owner = ''
+  isRevoked = ''
+  isPhisher = ''
+  isMember = ''
+
+[mutations]
+  invoke = ''
+  revoke = ''

package.json

@@ -10,8 +10,8 @@
     "clean": "rm -rf ./dist",
     "copy-assets": "copyfiles -u 1 src/**/*.gql dist/",
     "prepare": "husky install",
-    "server": "DEBUG='vulcanize:*, laconic:*' YARN_CHILD_PROCESS=true node --enable-source-maps dist/server.js",
-    "server:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/server.ts",
+    "server": "DEBUG='vulcanize:*, laconic:*, ts-nitro:*' YARN_CHILD_PROCESS=true node --enable-source-maps dist/server.js",
+    "server:dev": "DEBUG='vulcanize:*, laconic:*, ts-nitro:*' YARN_CHILD_PROCESS=true ts-node src/server.ts",
     "job-runner": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true node --enable-source-maps dist/job-runner.js",
     "job-runner:dev": "DEBUG=vulcanize:* YARN_CHILD_PROCESS=true ts-node src/job-runner.ts",
     "watch:contract": "DEBUG=vulcanize:* ts-node src/cli/watch-contract.ts",
@@ -39,19 +39,23 @@
   "homepage": "https://github.com/cerc-io/mobymask-v2-watcher-ts#readme",
   "dependencies": {
     "@apollo/client": "^3.3.19",
-    "@cerc-io/cli": "^0.2.41",
-    "@cerc-io/ipld-eth-client": "^0.2.41",
-    "@cerc-io/solidity-mapper": "^0.2.41",
-    "@cerc-io/util": "^0.2.41",
+    "@cerc-io/peer": "^0.2.50",
+    "@cerc-io/cli": "^0.2.50",
+    "@cerc-io/ipld-eth-client": "^0.2.50",
+    "@cerc-io/nitro-client": "^0.1.6",
+    "@cerc-io/nitro-util": "^0.1.6",
+    "@cerc-io/solidity-mapper": "^0.2.50",
+    "@cerc-io/util": "^0.2.50",
     "@ethersproject/providers": "^5.4.4",
     "apollo-type-bigint": "^0.1.3",
     "debug": "^4.3.1",
     "decimal.js": "^10.3.1",
     "ethers": "^5.4.4",
     "graphql": "^15.5.0",
     "json-bigint": "^1.0.0",
+    "pluralize": "^8.0.0",
     "reflect-metadata": "^0.1.13",
-    "typeorm": "^0.2.32",
+    "typeorm": "0.2.37",
     "yargs": "^17.0.1"
   },
   "devDependencies": {

src/config.ts

@@ -0,0 +1,4 @@
+import { BaseRatesConfig } from '@cerc-io/util';
+
+// Extend if required
+export type RatesConfig = BaseRatesConfig

src/indexer.ts

@@ -329,7 +329,15 @@ export class Indexer implements IndexerInterface {
   async _getStorageValueRPC (storageLayout: StorageLayout, blockHash: string, contractAddress: string, variable: string, ...mappingKeys: MappingKey[]): Promise<ValueResult> {
     const getStorageAt = async (params: { blockHash: string, contract: string, slot: string }) => {
       const { blockHash, contract, slot } = params;
-      const value = await this._ethProvider.getStorageAt(contract, slot, blockHash);
+      let value;
+
+      // Use block number if RPC doesn't support blockHash as blockTag param
+      if (this._serverConfig.rpcSupportsBlockHashParam === false) {
+        const { number: blockNumber } = await this._ethProvider.getBlock(blockHash);
+        value = await this._ethProvider.getStorageAt(contract, slot, blockNumber);
+      } else {
+        value = await this._ethProvider.getStorageAt(contract, slot, blockHash);
+      }
 
       return {
         value,

src/libp2p-utils.ts

@@ -4,7 +4,10 @@
 
 import debug from 'debug';
 import { ethers, Signer } from 'ethers';
+
 import { TransactionReceipt, TransactionResponse } from '@ethersproject/providers';
+import { PaymentsManager } from '@cerc-io/util';
+import { utils as nitroUtils } from '@cerc-io/nitro-client';
 
 import { abi as PhisherRegistryABI } from './artifacts/PhisherRegistry.json';
 
@@ -24,11 +27,12 @@ export function createMessageToL2Handler (
   { contractAddress, gasLimit }: {
     contractAddress: string,
     gasLimit?: number
-  }
+  },
+  paymentsManager: PaymentsManager
 ) {
   return (peerId: string, data: any): void => {
     log(`[${getCurrentTime()}] Received a message on mobymask P2P network from peer:`, peerId);
-    sendMessageToL2(signer, { contractAddress, gasLimit }, data);
+    sendMessageToL2(signer, { contractAddress, gasLimit }, data, paymentsManager);
   };
 }
 
@@ -38,9 +42,48 @@ export async function sendMessageToL2 (
     contractAddress: string,
     gasLimit?: number
   },
-  data: any
+  data: any,
+  paymentsManager: PaymentsManager
 ): Promise<void> {
-  const { kind, message } = data;
+  // Message envelope includes the payload as well as a payment (to, vhash, vsig)
+  const {
+    payload: { kind, message },
+    payment
+  } = data;
+
+  if (!paymentsManager.clientAddress) {
+    log('Ignoring payload, payments manager not subscribed to vouchers yet');
+    return;
+  }
+
+  // Ignore if the payload is not meant for us
+  if (payment.to === paymentsManager.clientAddress) {
+    log('Ignoring payload not meant for this client');
+    return;
+  }
+
+  // Retrieve signer address
+  const signerAddress = nitroUtils.getSignerAddress(payment.vhash, payment.vsig);
+
+  // Get the configured mutation cost
+  const mutationRates = paymentsManager.mutationRates;
+  if (kind in mutationRates) {
+    const configuredMutationCost = BigInt(mutationRates[kind as string]);
+
+    // Check for payment voucher received from the sender Nitro account
+    const [paymentVoucherReceived, paymentError] = await paymentsManager.authenticatePayment(payment.vhash, signerAddress, configuredMutationCost);
+
+    if (!paymentVoucherReceived) {
+      log(`Rejecting a mutation request from ${signerAddress}: ${paymentError}`);
+      return;
+    }
+
+    log(`Serving a paid mutation request for ${signerAddress}`);
+  } else {
+    // Serve a mutation request for free if rate is not configured
+    log(`Mutation rate not configured for "${kind}", serving a free mutation request to ${signerAddress}`);
+  }
+
   const contract = new ethers.Contract(contractAddress, PhisherRegistryABI, signer);
   let receipt: TransactionReceipt | undefined;
 
@@ -99,7 +142,11 @@ export async function sendMessageToL2 (
 
 export function parseLibp2pMessage (peerId: string, data: any): void {
   log(`[${getCurrentTime()}] Received a message on mobymask P2P network from peer:`, peerId);
-  const { kind, message } = data;
+
+  // Message envelope includes the payload
+  const {
+    payload: { kind, message }
+  } = data;
 
   switch (kind) {
     case MESSAGE_KINDS.INVOKE: {

src/nitro-addresses.json

@@ -0,0 +1,5 @@
+{
+  "nitroAdjudicatorAddress": "",
+  "virtualPaymentAppAddress": "",
+  "consensusAppAddress": ""
+}

src/server.ts

@@ -11,10 +11,17 @@ import { ethers } from 'ethers';
 
 import { ServerCmd } from '@cerc-io/cli';
 
+import { utils } from '@cerc-io/nitro-client';
+
 import { createResolvers } from './resolvers';
 import { Indexer } from './indexer';
 import { Database } from './database';
 import { createMessageToL2Handler, parseLibp2pMessage } from './libp2p-utils';
+import contractAddresses from './nitro-addresses.json';
+import { Config, PaymentsManager, getConfig } from '@cerc-io/util';
+import { Peer } from '@cerc-io/peer';
+
+import { RatesConfig } from './config';
 
 const log = debug('vulcanize:server');
 
@@ -23,17 +30,59 @@ export const main = async (): Promise<any> => {
   await serverCmd.init(Database);
   await serverCmd.initIndexer(Indexer);
 
+  let nitroPaymentsManager: PaymentsManager | undefined;
   let p2pMessageHandler = parseLibp2pMessage;
-  const { enableL2Txs, l2TxsConfig } = serverCmd.config.server.p2p.peer;
 
-  if (enableL2Txs) {
-    assert(l2TxsConfig);
-    const wallet = new ethers.Wallet(l2TxsConfig.privateKey, serverCmd.ethProvider);
-    p2pMessageHandler = createMessageToL2Handler(wallet, l2TxsConfig);
+  const { enablePeer, peer: { enableL2Txs, l2TxsConfig }, nitro: { payments } } = serverCmd.config.server.p2p;
+
+  if (enablePeer) {
+    const ratesConfig: RatesConfig = await getConfig(payments.ratesFile);
+    nitroPaymentsManager = new PaymentsManager(payments, ratesConfig);
+
+    if (enableL2Txs) {
+      assert(l2TxsConfig);
+      const wallet = new ethers.Wallet(l2TxsConfig.privateKey, serverCmd.ethProvider);
+      p2pMessageHandler = createMessageToL2Handler(wallet, l2TxsConfig, nitroPaymentsManager);
+    }
   }
 
   const typeDefs = fs.readFileSync(path.join(__dirname, 'schema.gql')).toString();
-  return serverCmd.exec(createResolvers, typeDefs, p2pMessageHandler);
+  await serverCmd.exec(createResolvers, typeDefs, p2pMessageHandler, nitroPaymentsManager);
+
+  if (enablePeer) {
+    assert(serverCmd.peer);
+    assert(nitroPaymentsManager);
+
+    const nitro = await setupNitro(serverCmd.config, serverCmd.peer);
+    log(`Nitro client started with address: ${nitro.client.address}`);
+
+    // Start subscription for payment vouchers received by the client
+    nitroPaymentsManager.subscribeToVouchers(nitro.client);
+  }
+};
+
+const setupNitro = async (config: Config, peer: Peer): Promise<utils.Nitro> => {
+  const {
+    server: {
+      p2p: {
+        nitro
+      }
+    },
+    upstream: {
+      ethServer: {
+        rpcProviderEndpoint
+      }
+    }
+  } = config;
+
+  return utils.Nitro.setupClient(
+    nitro.privateKey,
+    rpcProviderEndpoint,
+    nitro.chainPrivateKey,
+    contractAddresses,
+    peer,
+    path.resolve(nitro.store)
+  );
 };
 
 main().then(() => {