import { NetworkContext , Node as NodeTypes } from "@counterfactual/types" ; import { Provider } from "ethers/providers" ; import { SigningKey } from "ethers/utils" ; import EventEmitter from "eventemitter3" ; import log from "loglevel" ; import { Memoize } from "typescript-memoize" ; import { createRpcRouter } from "./api" ; import { Opcode , ProtocolMessage , ProtocolRunner } from "./engine" ; import { getFreeBalanceAddress } from "./models/free-balance" ; import { EthereumNetworkName , getNetworkContextForNetworkName } from "./network-configuration" ; import { getPrivateKeysGeneratorAndXPubOrThrow , PrivateKeysGetter } from "./private-keys-generator" ; import ProcessQueue from "./process-queue" ; import { RequestHandler } from "./request-handler" ; import RpcRouter from "./rpc-router" ; import { Store } from "./store" ; import { NODE_EVENTS , NodeMessageWrappedProtocolMessage } from "./types" ; import { timeout } from "./utils" ; import AutoNonceWallet from "./utils/auto-nonce-wallet" ; import { Deferred } from "./utils/deferred" ; export interface NodeConfig { STORE_KEY_PREFIX : string ; } const REASONABLE_NUM_BLOCKS_TO_WAIT = 1 ; export class Node { private readonly incoming : EventEmitter ; private readonly outgoing : EventEmitter ; private readonly networkContext : NetworkContext ; private readonly ioSendDeferrals = new Map < string , Deferred < NodeMessageWrappedProtocolMessage > > ( ) ; /**
     * These properties don't have initializers in the constructor, since they must be initialized
     * asynchronously. This is done via the `asynchronouslySetupUsingRemoteServices` function.
     * Since we have a private constructor and only allow instances of the Node to be created
     * via `create` which immediately calls `asynchronouslySetupUsingRemoteServices`, these are
     * always non-null when the Node is being used.
     */ private readonly store : Store ; private signer ! : SigningKey ; protected requestHandler ! : RequestHandler ; public rpcRouter ! : RpcRouter ; private readonly protocolRunner ! : ProtocolRunner ; static async create ( messagingService , storeService , networkOrNetworkContext , nodeConfig , provider , lockService ? , publicExtendedKey ? , privateKeyGenerator ? , blocksNeededForConfirmation ? ) { const [ privateKeysGenerator , extendedPubKey ] = await getPrivateKeysGeneratorAndXPubOrThrow ( storeService , privateKeyGenerator , publicExtendedKey ) ; const node = new Node ( extendedPubKey , privateKeysGenerator , messagingService , storeService , nodeConfig , provider , networkOrNetworkContext , blocksNeededForConfirmation , lockService ) ; return await node . asynchronouslySetupUsingRemoteServices ( ) ; } private constructor ( private readonly publicExtendedKey , private readonly privateKeyGetter , private readonly messagingService , private readonly storeService , private readonly nodeConfig , private readonly provider , networkContext , readonly blocksNeededForConfirmation = REASONABLE_NUM_BLOCKS_TO_WAIT , private readonly lockService ? ) { this . incoming = new EventEmitter ( ) ; this . outgoing = new EventEmitter ( ) ; this . networkContext = typeof networkContext === "string" ? getNetworkContextForNetworkName ( networkContext ) : networkContext ; this . store = new Store ( storeService , `${ this . nodeConfig . STORE_KEY_PREFIX }/${ this . publicIdentifier }` ) ; this . protocolRunner = this . buildProtocolRunner ( ) ; log . info ( `Waiting for ${ this . blocksNeededForConfirmation } block confirmations` ) ; } private async asynchronouslySetupUsingRemoteServices ( ) { this . signer = new SigningKey ( await this . privateKeyGetter . getPrivateKey ( "0" ) ) ; log . info ( `Node signer address: ${ this . signer . address }` ) ; log . info ( `Node public identifier: ${ this . publicIdentifier }` ) ; this . requestHandler = new RequestHandler ( this . publicIdentifier , this . incoming , this . outgoing , this . store , this . messagingService , this . protocolRunner , this . networkContext , this . provider , new AutoNonceWallet ( this . signer . privateKey , this . provider ) , this . blocksNeededForConfirmation ! , new ProcessQueue ( this . lockService ) ) ; await this . requestHandler . store . connectDB ( ) ; this . registerMessagingConnection ( ) ; this . rpcRouter = createRpcRouter ( this . requestHandler ) ; this . requestHandler . injectRouter ( this . rpcRouter ) ; return this ; } @ Memoize ( ) get publicIdentifier ( ) : string { return this . publicExtendedKey ; } @ Memoize ( ) async signerAddress ( ) { return await this . requestHandler . getSignerAddress ( ) ; } @ Memoize ( ) get freeBalanceAddress ( ) : string { return getFreeBalanceAddress ( this . publicIdentifier ) ; } /**
     * Instantiates a new _ProtocolRunner_ object and attaches middleware
     * for the OP_SIGN, IO_SEND, and IO_SEND_AND_WAIT opcodes.
     */ private buildProtocolRunner ( ) { const protocolRunner = new ProtocolRunner ( this . store , this . networkContext , this . provider ) ; protocolRunner . register ( Opcode . OP_SIGN , async ( args ) => { if ( args . length !== 1 && args . length !== 2 ) { throw Error ( "OP_SIGN middleware received wrong number of arguments." ) ; } const [ commitment , overrideKeyIndex ] = args ; const keyIndex = overrideKeyIndex || 0 ; const signingKey = new SigningKey ( await this . privateKeyGetter . getPrivateKey ( keyIndex ) ) ; return signingKey . signDigest ( commitment . hashToSign ( ) ) ; } ) ; protocolRunner . register ( Opcode . IO_SEND , async ( args ) => { const [ data ] = args ; const fromXpub = this . publicIdentifier ; const to = data . toXpub ; await this . messagingService . send ( to , { data , from : fromXpub , type : NODE_EVENTS . PROTOCOL_MESSAGE_EVENT } as NodeMessageWrappedProtocolMessage ) ; } ) ; protocolRunner . register ( Opcode . IO_SEND_AND_WAIT , async ( args ) => { const [ data ] = args ; const to = data . toXpub ; const deferral = new Deferred < NodeMessageWrappedProtocolMessage > ( ) ; this . ioSendDeferrals . set ( data . processID , deferral ) ; const counterpartyResponse = deferral . promise ; await this . messagingService . send ( to , { data , from : this . publicIdentifier , type : NODE_EVENTS . PROTOCOL_MESSAGE_EVENT } as NodeMessageWrappedProtocolMessage ) ; const msg = await Promise . race ( [ counterpartyResponse , timeout ( 60000 ) ] ) ; if ( ! msg || ! ( "data" in ( msg as NodeMessageWrappedProtocolMessage ) ) ) { throw Error ( `IO_SEND_AND_WAIT timed out after 30s waiting for counterparty reply in ${ data . protocol }` ) ; } this . ioSendDeferrals . delete ( data . processID ) ; return ( msg as NodeMessageWrappedProtocolMessage ) . data ; } ) ; return protocolRunner ; } param event param callback on ( event , callback ) { this . rpcRouter . subscribe ( event , async ( res ) => callback ( res ) ) ; } param event param callback off ( event , callback ? ) { this . rpcRouter . unsubscribe ( event , callback ? async ( res ) => callback ( res ) : undefined ) ; } param event param callback once ( event , callback ) { this . rpcRouter . subscribeOnce ( event , async ( res ) => callback ( res ) ) ; } param event param req emit ( event , req ) { this . rpcRouter . emit ( event , req ) ; } param method param req async call ( method , req ) { return this . requestHandler . callMethod ( method , req ) ; } /**
     * When a Node is first instantiated, it establishes a connection
     * with the messaging service. When it receives a message, it emits
     * the message to its registered subscribers, usually external
     * subscribed (i.e. consumers of the Node).
     */ private registerMessagingConnection ( ) { this . messagingService . onReceive ( this . publicIdentifier , async ( msg ) => { await this . handleReceivedMessage ( msg ) ; this . rpcRouter . emit ( msg . type , msg , "outgoing" ) ; } ) ; } /**
     * Messages received by the Node fit into one of three categories:
     *
     * (a) A NodeMessage which is _not_ a NodeMessageWrappedProtocolMessage;
     *     this is a standard received message which is handled by a named
     *     controller in the _events_ folder.
     *
     * (b) A NodeMessage which is a NodeMessageWrappedProtocolMessage _and_
     *     has no registered _ioSendDeferral_ callback. In this case, it means
     *     it will be sent to the protocol message event controller to dispatch
     *     the received message to the instruction executor.
     *
     * (c) A NodeMessage which is a NodeMessageWrappedProtocolMessage _and_
     *     _does have_ an _ioSendDeferral_, in which case the message is dispatched
     *     solely to the deffered promise's resolve callback.
     */ private async handleReceivedMessage ( msg ) { if ( ! Object . values ( NODE_EVENTS ) . includes ( msg . type ) ) { console . error ( `Received message with unknown event type: ${ msg . type }` ) ; } const isProtocolMessage = ( msg ) => msg . type === NODE_EVENTS . PROTOCOL_MESSAGE_EVENT ; const isExpectingResponse = ( msg ) => this . ioSendDeferrals . has ( msg . data . processID ) ; if ( isProtocolMessage ( msg ) && isExpectingResponse ( msg as NodeMessageWrappedProtocolMessage ) ) { await this . handleIoSendDeferral ( msg as NodeMessageWrappedProtocolMessage ) ; } else if ( this . requestHandler . isLegacyEvent ( msg . type ) ) { await this . requestHandler . callEvent ( msg . type , msg ) ; } else { await this . rpcRouter . emit ( msg . type , msg ) ; } } private async handleIoSendDeferral ( msg ) { const key = msg . data . processID ; if ( ! this . ioSendDeferrals . has ( key ) ) { throw Error ( "Node received message intended for machine but no handler was present" ) ; } const promise = this . ioSendDeferrals . get ( key ) ! ; try { promise . resolve ( msg ) ; } catch ( error ) { console . error ( `Error while executing callback registered by IO_SEND_AND_WAIT middleware hook` , { error , msg } ) ; } } }