import { Allocation , ChannelResult , CreateChannelParams , Uint256 , } from '@statechannels/client-api-schema' ; import _ from 'lodash' ; import EventEmitter from 'eventemitter3' ; import { calculateObjectiveId , makeAddress , makeDestination , Address , makePrivateKey , } from '@statechannels/wallet-core' ; import { utils } from 'ethers' ; import { setIntervalAsync , clearIntervalAsync } from 'set-interval-async/dynamic' ; import P from 'pino' ; import { MessageHandler , MessageServiceFactory , MessageServiceInterface , } from '../message-service/types' ; import { getMessages } from '../message-service/utils' ; import { defaultConfig , Engine , extractDBConfigFromWalletConfig , hasNewObjective , EngineConfig , IncomingWalletConfig , isMultipleChannelOutput , MultipleChannelOutput , SingleChannelOutput , validateEngineConfig , SyncConfiguration , } from '../engine' ; import { AllocationUpdatedArg , ChainEventSubscriberInterface , ChainService , ChainServiceInterface , ChallengeRegisteredArg , ChannelFinalizedArg , HoldingUpdatedArg , MockChainService , } from '../chain-service' ; import * as ChannelState from '../protocols/state' ; import { createLogger } from '../logger' ; import { ConfigValidationError , SingleThreadedEngine } from '../engine/engine' ; import { ObjectiveResult , WalletEvents , ObjectiveDoneResult , UpdateChannelResult , CreateLedgerChannelParams , } from './types' ; export class Wallet extends EventEmitter < WalletEvents > { param incomingConfig param messageServiceFactory returns public static async create ( incomingConfig , messageServiceFactory ) { const populatedConfig = _ . assign ( { } , defaultConfig , incomingConfig ) ; const logger = createLogger ( populatedConfig ) ; logger . trace ( { engineConfig : populatedConfig } , 'Wallet initializing' ) ; const { errors , valid } = validateEngineConfig ( populatedConfig ) ; if ( ! valid ) { errors . forEach ( error => logger . error ( { error } , `Validation error occurred ${ error . message }` ) ) ; throw new ConfigValidationError ( errors ) ; } const { skipEvmValidation , metricsConfiguration } = populatedConfig ; const engineConfig = { skipEvmValidation , dbConfig : extractDBConfigFromWalletConfig ( populatedConfig ) , metrics : metricsConfiguration , chainNetworkID : utils . hexlify ( populatedConfig . networkConfiguration . chainNetworkID ) , workerThreadAmount : 0 , } ; const engine = await SingleThreadedEngine . create ( engineConfig , logger ) ; if ( populatedConfig . privateKey ) { await engine . addSigningKey ( makePrivateKey ( populatedConfig . privateKey ) ) ; } const chainService = populatedConfig . chainServiceConfiguration . attachChainService ? new ChainService ( { ... populatedConfig . chainServiceConfiguration , logger , } ) : new MockChainService ( ) ; await chainService . checkChainId ( populatedConfig . networkConfiguration . chainNetworkID ) ; const networkId = utils . hexlify ( populatedConfig . networkConfiguration . chainNetworkID ) ; const wallet = new Wallet ( messageServiceFactory , engine , chainService , networkId , logger , populatedConfig . syncConfiguration ) ; await wallet . registerExistingChannelsWithChainService ( ) ; return wallet ; } private _messageService : MessageServiceInterface ; private _syncInterval = setIntervalAsync ( async ( ) => { await this . syncObjectives ( ) ; } , this . _syncOptions . pollInterval ) ; private _chainListener : ChainEventSubscriberInterface ; private constructor ( messageServiceFactory , private _engine , private _chainService , private _chainNetworkId , private _logger , private _syncOptions ) { super ( ) ; const handler = async ( message ) => { const result = await this . _engine . pushMessage ( message . data ) ; await this . handleEngineOutput ( result ) ; } ; this . _messageService = messageServiceFactory ( handler ) ; this . _chainListener = { holdingUpdated : this . createChainEventlistener ( 'holdingUpdated' , e => this . _engine . store . updateFunding ( e . channelId , e . amount , e . asset ) ) , allocationUpdated : this . createChainEventlistener ( 'allocationUpdated' , async ( e ) => { const transferredOut = e . externalPayouts . map ( ai => ( { toAddress : makeDestination ( ai . destination ) , amount : ai . amount as Uint256 , } ) ) ; await this . _engine . store . updateTransferredOut ( e . channelId , e . asset , transferredOut , e . newHoldings ) ; } ) , challengeRegistered : this . createChainEventlistener ( 'challengeRegistered' , e => this . _engine . store . insertAdjudicatorStatus ( e . channelId , e . finalizesAt , e . challengeStates ) ) , channelFinalized : this . createChainEventlistener ( 'channelFinalized' , e => this . _engine . store . markAdjudicatorStatusAsFinalized ( e . channelId , e . blockNumber , e . blockTimestamp , e . finalizedAt ) ) , } ; } remarks param appDefinition returns public async registerAppDefinition ( appDefinition ) { const bytecode = await this . _chainService . fetchBytecode ( appDefinition ) ; await this . _engine . store . upsertBytecode ( utils . hexlify ( this . _chainNetworkId ) , makeAddress ( appDefinition ) , bytecode ) ; } param channelParams param fundingStrategy returns public async createLedgerChannel ( channelParams ) { try { const createResult = await this . _engine . createLedgerChannel ( channelParams , channelParams . fundingStrategy ) ; await this . registerChannels ( [ createResult . channelResult ] ) ; const { newObjective , channelResult } = createResult ; const done = this . createObjectiveDoneResult ( newObjective . objectiveId ) ; await this . handleEngineOutput ( createResult ) ; return { channelId : channelResult . channelId , currentStatus : newObjective . status , objectiveId : newObjective . objectiveId , done , } ; } catch ( error ) { this . _logger . error ( { err : error } , 'Uncaught InternalError in CreateLedgerChannel' ) ; return { channelId : 'ERROR' , currentStatus : 'failed' as const , objectiveId : 'ERROR' , done : Promise . resolve ( { type : 'InternalError' as const , error } ) , } ; } } remarks param objectiveIds returns public async approveObjectives ( objectiveIds ) { const { objectives , messages , chainRequests , channelResults , } = await this . _engine . approveObjectives ( objectiveIds ) ; await this . registerChannels ( channelResults ) ; const results = objectives . map ( async ( o ) => ( { objectiveId : o . objectiveId , currentStatus : o . status , channelId : o . data . targetChannelId , done : this . createObjectiveDoneResult ( o . objectiveId ) , } ) ) ; const completedObjectives = objectives . filter ( o => o . status === 'succeeded' ) ; completedObjectives . forEach ( o => this . emit ( 'ObjectiveCompleted' , o ) ) ; await this . _chainService . handleChainRequests ( chainRequests ) ; await this . messageService . send ( messages ) ; return Promise . all ( results ) ; } param channelParameters returns public async createChannels ( channelParameters ) { return Promise . all ( channelParameters . map ( async ( p ) => { try { const createResult = await this . _engine . createChannel ( p ) ; await this . registerChannels ( [ createResult . channelResult ] ) ; const { newObjective , channelResult } = createResult ; const done = this . createObjectiveDoneResult ( newObjective . objectiveId ) ; await this . handleEngineOutput ( createResult ) ; return { channelId : channelResult . channelId , currentStatus : newObjective . status , objectiveId : newObjective . objectiveId , done , } ; } catch ( error ) { this . _logger . error ( { err : error } , 'Uncaught InternalError in CreateChannel' ) ; return { channelId : 'ERROR' , currentStatus : 'failed' as const , objectiveId : 'ERROR' , done : Promise . resolve ( { type : 'InternalError' as const , error } ) , } ; } } ) ) ; } param channelId param allocations param appData returns public async updateChannel ( channelId , allocations , appData ) { try { const updateResponse = await this . _engine . updateChannel ( { channelId , allocations , appData } ) ; await this . handleEngineOutput ( updateResponse ) ; return { type : 'Success' , channelId , result : updateResponse . channelResult } ; } catch ( error ) { return { type : 'InternalError' , error , channelId } ; } } private async syncObjectives ( ) { const staleDate = Date . now ( ) - this . _syncOptions . staleThreshold ; const timeOutDate = Date . now ( ) - this . _syncOptions . timeOutThreshold ; const objectives = await this . _engine . store . getApprovedObjectives ( ) ; const timedOutObjectives = objectives . filter ( o => o . progressLastMadeAt . getTime ( ) < timeOutDate ) ; for ( const o of timedOutObjectives ) { this . emit ( 'ObjectiveTimedOut' , o ) ; } const staleObjectives = objectives . filter ( o => o . progressLastMadeAt . getTime ( ) < staleDate ) . map ( o => o . objectiveId ) ; const syncMessages = await this . _engine . syncObjectives ( staleObjectives ) ; await this . _messageService . send ( syncMessages ) ; } param channelIds returns public async closeChannels ( channelIds ) { return Promise . all ( channelIds . map ( async ( channelId ) => { const objectiveId = calculateObjectiveId ( 'CloseChannel' , channelId ) ; const done = this . createObjectiveDoneResult ( objectiveId ) ; const closeResult = await this . _engine . closeChannel ( { channelId } ) ; const { newObjective , channelResult } = closeResult ; const latest = await this . _engine . getObjective ( newObjective . objectiveId ) ; await this . handleEngineOutput ( closeResult ) ; return { channelId : channelResult . channelId , currentStatus : latest . status , objectiveId : newObjective . objectiveId , done , } ; } ) ) ; } returns public async jumpStartObjectives ( ) { const objectives = await this . _engine . getApprovedObjectives ( ) ; const objectiveIds = objectives . map ( o => o . objectiveId ) ; await this . _engine . store . updateObjectiveProgressDate ( objectiveIds ) ; const syncMessages = await this . _engine . syncObjectives ( objectiveIds ) ; const results = objectives . map ( async ( o ) => { const done = this . createObjectiveDoneResult ( o . objectiveId ) ; return { objectiveId : o . objectiveId , currentStatus : o . status , channelId : o . data . targetChannelId , done , } ; } ) ; await this . messageService . send ( syncMessages ) ; return Promise . all ( results ) ; } private emitEvents ( result ) { if ( isMultipleChannelOutput ( result ) ) { for ( const c of result . channelResults ) { this . emit ( 'ChannelUpdated' , c ) ; } for ( const o of result . completedObjectives ) { this . emit ( 'ObjectiveCompleted' , o ) ; } for ( const o of result . newObjectives ) { if ( o . status === 'pending' ) { this . emit ( 'ObjectiveProposed' , o ) ; } } } else { this . emit ( 'ChannelUpdated' , result . channelResult ) ; if ( hasNewObjective ( result ) ) { if ( result . newObjective . status === 'pending' ) { this . emit ( 'ObjectiveProposed' , result . newObjective ) ; } } } } param output private async handleEngineOutput ( output ) { this . emitEvents ( output ) ; await this . _messageService . send ( getMessages ( output ) ) ; await this . _chainService . handleChainRequests ( output . chainRequests ) ; } public get messageService ( ) : MessageServiceInterface { return this . _messageService ; } private async createObjectiveDoneResult ( objectiveId ) { return new Promise < ObjectiveDoneResult > ( resolve => { this . on ( 'ObjectiveTimedOut' , o => { if ( o . objectiveId === objectiveId ) { this . _logger . trace ( { objective : o } , 'Objective Timed out' ) ; resolve ( { type : 'ObjectiveTimedOutError' , objectiveId : o . objectiveId , lastProgressMadeAt : o . progressLastMadeAt , } ) ; } } ) ; this . on ( 'ObjectiveCompleted' , o => { if ( o . objectiveId === objectiveId ) { this . _logger . trace ( { objective : o } , 'Objective Suceeded' ) ; resolve ( { type : 'Success' , channelId : o . data . targetChannelId } ) ; } } ) ; } ) ; } remarks returns private async registerExistingChannelsWithChainService ( ) { const channelsToRegister = ( await this . _engine . store . getNonFinalizedChannels ( ) ) . map ( ChannelState . toChannelResult ) ; await this . registerChannels ( channelsToRegister ) ; } private async registerChannels ( channelsToRegister ) { const channelsWithAssetHolders = channelsToRegister . map ( cr => ( { assetHolderAddresses : cr . allocations . map ( a => makeAddress ( a . asset ) ) , channelId : cr . channelId , } ) ) ; for ( const { channelId , assetHolderAddresses } of channelsWithAssetHolders ) { this . _chainService . registerChannel ( channelId , assetHolderAddresses , this . _chainListener ) ; } } private createChainEventlistener < K extends keyof ChainEventSubscriberInterface , EH extends ChainEventSubscriberInterface [ K ] > ( eventName , storeUpdater ) { return async ( event ) => { const { channelId } = event ; this . _logger . trace ( { event } , `${ eventName } being handled` ) ; try { await storeUpdater ( event ) ; const result = await this . _engine . crank ( [ channelId ] ) ; await this . handleEngineOutput ( result ) ; } catch ( err ) { this . _logger . error ( { err , event } , `Error handling ${ eventName }` ) ; throw err ; } } ; } public async getSigningAddress ( ) { return this . _engine . getSigningAddress ( ) ; } public async getChannels ( ) { const { channelResults } = await this . _engine . getChannels ( ) ; return channelResults ; } param peerUrl public async registerPeerMessageService ( peerUrl ) { await this . messageService . registerPeer ( peerUrl ) ; } async destroy ( ) { await clearIntervalAsync ( this . _syncInterval ) ; this . _chainService . destructor ( ) ; await this . _messageService . destroy ( ) ; await this . _engine . destroy ( ) ; } }