Integrate peer in watcher server and send txs to L2 chain (#344)

* Integrate peer in watcher and send txs to L2

* Remove passing debug log in functions

* Remove unnecessary relay advertise config option

---------

Co-authored-by: prathamesh0 <prathamesh.musale0@gmail.com>

packages/cli/src/peer.ts

@@ -27,7 +27,7 @@ interface Arguments {
 }
 
 export class PeerCmd {
-  async exec (pubSubTopic?: string, parseLibp2pMessage?: (log: debug.Debugger, peerId: string, data: any) => void): Promise<any> {
+  async exec (pubSubTopic?: string, parseLibp2pMessage?: (peerId: string, data: any) => void): Promise<any> {
     const argv: Arguments = _getArgv();
 
     const { Peer } = await import('@cerc-io/peer');
@@ -52,7 +52,7 @@ export class PeerCmd {
     if (pubSubTopic) {
       peer.subscribeTopic(pubSubTopic, (peerId, data) => {
         if (parseLibp2pMessage) {
-          parseLibp2pMessage(log, peerId.toString(), data);
+          parseLibp2pMessage(peerId.toString(), data);
         } else {
           log(`> ${peerId.toString()} > ${data}`);
         }

packages/cli/src/server.ts

@@ -105,7 +105,7 @@ export class ServerCmd {
   async exec (
     createResolvers: (indexer: IndexerInterface, eventWatcher: EventWatcher) => Promise<any>,
     typeDefs: TypeSource,
-    parseLibp2pMessage?: (log: debug.Debugger, peerId: string, data: any) => void
+    parseLibp2pMessage?: (peerId: string, data: any) => void
   ): Promise<{
     app: Application,
     server: ApolloServer
@@ -146,7 +146,7 @@ export class ServerCmd {
 
   async _startP2PNodes (
     p2pConfig: P2PConfig,
-    parseLibp2pMessage?: (log: debug.Debugger, peerId: string, data: any) => void
+    parseLibp2pMessage?: (peerId: string, data: any) => void
   ): Promise<void> {
     const { createRelayNode, Peer } = await import('@cerc-io/peer');
     const {
@@ -208,7 +208,7 @@ export class ServerCmd {
 
       peer.subscribeTopic(peerConfig.pubSubTopic, (peerId, data) => {
         if (parseLibp2pMessage) {
-          parseLibp2pMessage(log, peerId.toString(), data);
+          parseLibp2pMessage(peerId.toString(), data);
         }
       });
 

packages/mobymask-v2-watcher/environments/local.toml

@@ -1,7 +1,7 @@
 [server]
   host = "127.0.0.1"
   port = 3010
-  kind = "active"
+  kind = "lazy"
 
   # Checkpointing state.
   checkpointing = true
@@ -32,6 +32,11 @@
     [server.p2p.peer]
       relayMultiaddr = ''
       pubSubTopic = 'mobymask'
+      peerIdFile = ''
+
+      [server.p2p.peer.l2TxConfig]
+        privateKey = ''
+        contractAddress = ''
 
 [metrics]
   host = "127.0.0.1"

packages/mobymask-v2-watcher/src/libp2p-utils.ts

@@ -8,7 +8,7 @@ import { TransactionReceipt, TransactionResponse } from '@ethersproject/provider
 
 import { abi as PhisherRegistryABI } from './artifacts/PhisherRegistry.json';
 
-const log = debug('laconic:libp2p-utils');
+const log = debug('vulcanize:libp2p-utils');
 
 const contractInterface = new ethers.utils.Interface(PhisherRegistryABI);
 
@@ -17,11 +17,26 @@ const MESSAGE_KINDS = {
   REVOKE: 'revoke'
 };
 
-export async function sendMessageToL2 (
+const DEFAULT_GAS_LIMIT = 500000;
+
+export function createMessageToL2Handler (
   signer: Signer,
   { contractAddress, gasLimit }: {
     contractAddress: string,
-    gasLimit: number
+    gasLimit?: number
+  }
+) {
+  return (peerId: string, data: any): void => {
+    log(`[${getCurrentTime()}] Received a message on mobymask P2P network from peer:`, peerId);
+    sendMessageToL2(signer, { contractAddress, gasLimit }, data);
+  };
+}
+
+export async function sendMessageToL2 (
+  signer: Signer,
+  { contractAddress, gasLimit = DEFAULT_GAS_LIMIT }: {
+    contractAddress: string,
+    gasLimit?: number
   },
   data: any
 ): Promise<void> {
@@ -82,18 +97,18 @@ export async function sendMessageToL2 (
   }
 }
 
-export function parseLibp2pMessage (log: debug.Debugger, peerId: string, data: any): void {
+export function parseLibp2pMessage (peerId: string, data: any): void {
   log(`[${getCurrentTime()}] Received a message on mobymask P2P network from peer:`, peerId);
   const { kind, message } = data;
 
   switch (kind) {
     case MESSAGE_KINDS.INVOKE: {
-      _parseInvocation(log, message);
+      _parseInvocation(message);
       break;
     }
 
     case MESSAGE_KINDS.REVOKE: {
-      _parseRevocation(log, message);
+      _parseRevocation(message);
       break;
     }
 
@@ -112,7 +127,7 @@ export const getCurrentTime = (): string => {
   return `${now.getHours()}:${now.getMinutes()}:${now.getSeconds()}`;
 };
 
-function _parseInvocation (log: debug.Debugger, msg: any): void {
+function _parseInvocation (msg: any): void {
   log('Signed invocations:');
   log(JSON.stringify(msg, null, 2));
 
@@ -125,7 +140,7 @@ function _parseInvocation (log: debug.Debugger, msg: any): void {
   });
 }
 
-function _parseRevocation (log: debug.Debugger, msg: any): void {
+function _parseRevocation (msg: any): void {
   const { signedDelegation, signedIntendedRevocation } = msg;
   log('Signed delegation:');
   log(JSON.stringify(signedDelegation, null, 2));

packages/mobymask-v2-watcher/src/server.ts

@@ -6,13 +6,14 @@ import fs from 'fs';
 import path from 'path';
 import 'reflect-metadata';
 import debug from 'debug';
+import { ethers } from 'ethers';
 
 import { ServerCmd } from '@cerc-io/cli';
 
 import { createResolvers } from './resolvers';
 import { Indexer } from './indexer';
 import { Database } from './database';
-import { parseLibp2pMessage } from './libp2p-utils';
+import { createMessageToL2Handler, parseLibp2pMessage } from './libp2p-utils';
 
 const log = debug('vulcanize:server');
 
@@ -21,8 +22,16 @@ export const main = async (): Promise<any> => {
   await serverCmd.init(Database);
   await serverCmd.initIndexer(Indexer);
 
+  let p2pMessageHandler = parseLibp2pMessage;
+  const { l2TxConfig } = serverCmd.config.server.p2p.peer;
+
+  if (l2TxConfig) {
+    const wallet = new ethers.Wallet(l2TxConfig.privateKey, serverCmd.ethProvider);
+    p2pMessageHandler = createMessageToL2Handler(wallet, l2TxConfig);
+  }
+
   const typeDefs = fs.readFileSync(path.join(__dirname, 'schema.gql')).toString();
-  return serverCmd.exec(createResolvers, typeDefs, parseLibp2pMessage);
+  return serverCmd.exec(createResolvers, typeDefs, p2pMessageHandler);
 };
 
 main().then(() => {

packages/peer/src/relay.ts

@@ -85,9 +85,6 @@ export async function createRelayNode (init: RelayNodeInitConfig): Promise<Libp2
       hop: {
         enabled: true,
         timeout: HOP_TIMEOUT
-      },
-      advertise: {
-        enabled: true
       }
     },
     webRTCSignal: {

packages/util/src/config.ts

@@ -69,6 +69,18 @@ export interface RelayConfig {
   enableDebugInfo?: boolean;
 }
 
+// L2 tx config
+interface L2TxConfig {
+  // Address of contract for which txs are sent
+  contractAddress: string;
+
+  // Private key of tx signer (needs to have some balance)
+  privateKey: string;
+
+  // Gas limit for tx
+  gasLimit?: number
+}
+
 // Peer config
 export interface PeerConfig {
   // Multiaddr of the primary relay node for this peer
@@ -100,6 +112,9 @@ export interface PeerConfig {
 
   // Participate in exchange of debug info over floodsub
   enableDebugInfo?: boolean;
+
+  // Config for sending txs to L2
+  l2TxConfig?: L2TxConfig;
 }
 
 // P2P config