import { UpdateChannelParams , CreateChannelParams , SyncChannelParams , JoinChannelParams , CloseChannelParams , GetStateParams , Participant as APIParticipant , ChannelId , Message , ChannelResult , } from '@statechannels/client-api-schema' ; import { deserializeAllocations , validatePayload , Outcome , convertToParticipant , makeAddress , Address as CoreAddress , PrivateKey , NULL_APP_DATA , } from '@statechannels/wallet-core' ; import * as Either from 'fp-ts/lib/Either' ; import Knex from 'knex' ; import _ from 'lodash' ; import { ethers , utils } from 'ethers' ; import { Logger } from 'pino' ; import { Payload as WirePayload } from '@statechannels/wire-format' ; import { ValidationErrorItem } from 'joi' ; import { Bytes32 } from '../type-aliases' ; import * as UpdateChannel from '../handlers/update-channel' ; import * as JoinChannel from '../handlers/join-channel' ; import { PushMessageError } from '../errors/engine-error' ; import { timerFactory , recordFunctionMetrics , setupMetrics } from '../metrics' ; import { ChainRequest } from '../chain-service' ; import { WALLET_VERSION } from '../version' ; import { ObjectiveManager } from '../objectives' ; import { SingleAppUpdater } from '../handlers/single-app-updater' ; import { LedgerManager } from '../protocols/ledger-manager' ; import { WalletObjective , ObjectiveModel } from '../models/objective' ; import { getMessages } from '../message-service/utils' ; import { Store , AppHandler , MissingAppHandler } from './store' ; import { SingleChannelOutput , MultipleChannelOutput , EngineInterface , hasNewObjective , EngineConfig , } from './types' ; import { EngineResponse } from './engine-response' ; export class ConfigValidationError extends Error { constructor ( public errors ) { super ( 'Engine configuration validation failed' ) ; } } /**
 * A single-threaded Nitro engine
 */ export class SingleThreadedEngine implements EngineInterface { knex : Knex ; store : Store ; chainNetworkId : string ; objectiveManager : ObjectiveManager ; ledgerManager : LedgerManager ; public static async create ( engineConfig , logger ) { return new SingleThreadedEngine ( engineConfig , logger ) ; } readonly protected constructor ( private config , protected logger ) { this . knex = Knex ( config . dbConfig ) ; this . chainNetworkId = utils . hexlify ( this . config . chainNetworkID ) ; this . store = new Store ( this . knex , this . config . metrics . timingMetrics , this . config . skipEvmValidation , this . chainNetworkId , this . logger ) ; if ( this . config . metrics . timingMetrics ) { setupMetrics ( this . config . metrics . metricsOutputFile as string ) ; } this . objectiveManager = ObjectiveManager . create ( { store : this . store , logger : this . logger , timingMetrics : this . config . metrics . timingMetrics , } ) ; this . ledgerManager = LedgerManager . create ( { store : this . store } ) ; } remarks param privateKey returns public async addSigningKey ( privateKey ) { await this . store . addSigningKey ( privateKey ) ; } remarks param appDefinition param bytecode returns public async registerAppBytecode ( appDefinition , bytecode ) { return this . store . upsertBytecode ( utils . hexlify ( this . chainNetworkId ) , makeAddress ( appDefinition ) , bytecode ) ; } remarks returns public async destroy ( ) { await this . knex . destroy ( ) ; } param channelIds returns public async syncChannels ( channelIds ) { const response = EngineResponse . initialize ( ) ; await Promise . all ( channelIds . map ( channelId => this . _syncChannel ( channelId , response ) ) ) ; return response . multipleChannelOutput ( ) ; } param objectiveIds returns public async syncObjectives ( objectiveIds ) { const response = EngineResponse . initialize ( ) ; const objectives = await this . store . getObjectivesByIds ( objectiveIds ) ; const fetchedObjectiveIds = objectives . map ( o => o . objectiveId ) ; const missingObjectives = _ . difference ( objectiveIds , fetchedObjectiveIds ) ; if ( missingObjectives . length > 0 ) { this . logger . error ( { objectiveIds : missingObjectives } , 'The following objectives cannot be found' ) ; throw new Error ( 'Could not find all objectives' ) ; } for ( const o of objectives ) { const channel = await this . store . getChannelState ( o . data . targetChannelId ) ; await this . _syncChannel ( channel . channelId , response ) ; const { participants } = channel ; response . queueSendObjective ( o , channel . myIndex , participants ) ; if ( channel . fundingStrategy === 'Ledger' && channel . fundingLedgerChannelId ) { await this . _syncChannel ( channel . fundingLedgerChannelId , response ) ; } } return response . allMessages ; } param channelId returns public async syncChannel ( { channelId } ) { const response = EngineResponse . initialize ( ) ; await this . _syncChannel ( channelId , response ) ; return response . singleChannelOutput ( ) ; } private async _syncChannel ( channelId , response ) { const { states , channelState } = await this . store . getStates ( channelId ) ; const { myIndex , participants } = channelState ; states . forEach ( s => response . queueState ( s , myIndex , channelId ) ) ; response . queueChannelRequest ( channelId , myIndex , participants ) ; response . queueChannelState ( channelState ) ; } param challengeState returns public async challenge ( channelId ) { const response = EngineResponse . initialize ( ) ; await this . knex . transaction ( async ( tx ) => { const channel = await this . store . getChannel ( channelId , tx ) ; if ( ! channel ) { throw new Error ( `No channel found for channel id ${ channelId }` ) ; } if ( ! channel . isLedger ) { throw new Error ( 'Only ledger channels support challenging' ) ; } const objective = await ObjectiveModel . insert ( { type : 'SubmitChallenge' , participants : [ ] , data : { targetChannelId : channelId } , } , tx , 'approved' ) ; response . queueCreatedObjective ( objective , channel . myIndex , channel . participants ) ; response . queueChannel ( channel ) ; } ) ; await this . takeActions ( [ channelId ] , response ) ; return response . singleChannelOutput ( ) ; } returns public async getSigningAddress ( ) { return await this . store . getOrCreateSigningAddress ( ) ; } remarks The channel will have a null app definition and null app data.
     * This method is otherwise identical to  SingleThreadedEngine . createChannel .
     *
     *  returns async createLedgerChannel ( args , fundingStrategy = 'Direct' ) { const response = EngineResponse . initialize ( ) ; const channelId = await this . _createChannel ( response , { ... args , appDefinition : ethers . constants . AddressZero , appData : NULL_APP_DATA , fundingStrategy , } , 'ledger' ) ; await this . takeActions ( [ channelId ] , response ) ; const result = response . singleChannelOutput ( ) ; if ( ! hasNewObjective ( result ) ) { throw new Error ( 'No new objective created for create channel' ) ; } else { return result ; } } remarks param args returns async createChannel ( args ) { const response = EngineResponse . initialize ( ) ; await this . _createChannel ( response , args , 'app' ) ; const result = response . singleChannelOutput ( ) ; if ( ! hasNewObjective ( result ) ) { throw new Error ( 'No new objective created for create channel' ) ; } else { return result ; } } /**
     * Creates multiple channels with the same parameters. See  SingleThreadedEngine . createChannel .
     *
     *  param args param numberOfChannels returns async createChannels ( args , numberOfChannels ) { const response = EngineResponse . initialize ( ) ; for ( const _i of _ . range ( numberOfChannels ) ) { await this . _createChannel ( response , args , 'app' ) ; } return response . multipleChannelOutput ( ) ; } private async _createChannel ( response , args , role = 'app' ) { const { participants : serializedParticipants , appDefinition , appData , allocations , fundingStrategy , fundingLedgerChannelId , challengeDuration , } = args ; const participants = serializedParticipants . map ( convertToParticipant ) ; const outcome = deserializeAllocations ( allocations ) ; const channelNonce = await this . store . nextNonce ( participants . map ( p => p . signingAddress ) ) ; const constants = { appDefinition : makeAddress ( appDefinition ) , chainId : this . config . chainNetworkID , challengeDuration , channelNonce , participants , } ; const { channel , firstSignedState : signedState , objective } = await this . store . createChannel ( constants , appData , outcome , fundingStrategy , role , fundingLedgerChannelId ) ; response . queueState ( signedState , channel . myIndex , channel . channelId ) ; response . queueCreatedObjective ( objective , channel . myIndex , channel . participants ) ; response . queueChannelState ( channel ) ; return channel . channelId ; } remarks param channelIds returns async joinChannels ( channelIds ) { const response = EngineResponse . initialize ( ) ; const objectives = await this . store . getObjectives ( channelIds ) ; await Promise . all ( objectives . filter ( o => o . type === 'OpenChannel' ) . map ( async ( { objectiveId } ) => await this . store . approveObjective ( objectiveId ) ) ) ; await this . takeActions ( channelIds , response ) ; return response . multipleChannelOutput ( ) ; } private async approveObjective ( objectiveId , targetChannelId ) { return this . store . transaction ( async ( tx ) => { await this . store . getAndLockChannel ( targetChannelId , tx ) ; const objective = await this . store . getObjective ( objectiveId , tx ) ; if ( objective . status === 'pending' ) { const approved = await this . store . approveObjective ( objectiveId , tx ) ; return approved ; } else { return objective ; } } ) ; } async approveObjectives ( objectiveIds ) { const channelIds = [ ] ; const response = EngineResponse . initialize ( ) ; let objectives = await this . store . getObjectivesByIds ( objectiveIds ) ; for ( const objective of objectives ) { const { objectiveId , data } = objective ; await this . approveObjective ( objectiveId , data . targetChannelId ) ; channelIds . push ( data . targetChannelId ) ; } await this . takeActions ( channelIds , response ) ; objectives = await this . store . getObjectivesByIds ( objectiveIds ) ; return { objectives , messages : getMessages ( response . multipleChannelOutput ( ) ) , chainRequests : response . chainRequests , channelResults : response . channelResults , } ; } remarks param channelId returns async joinChannel ( { channelId } ) { const response = EngineResponse . initialize ( ) ; const channel = await this . store . getChannelState ( channelId ) ; if ( ! channel ) throw new JoinChannel . JoinChannelError ( JoinChannel . JoinChannelError . reasons . channelNotFound , channelId ) ; const objectives = await this . store . getObjectives ( [ channelId ] ) ; if ( objectives . length !== 1 ) { const msg = 'Expecting exactly one objective' ; this . logger . error ( { channelId , objectives } , msg ) ; throw new Error ( msg ) ; } const objective = objectives [ 0 ] ; if ( objective . type === 'OpenChannel' ) { await this . store . approveObjective ( objective . objectiveId ) ; } else { } await this . takeActions ( [ channelId ] , response ) ; return response . singleChannelOutput ( false ) ; } param channelIds returns public async crank ( channelIds ) { const response = EngineResponse . initialize ( ) ; await this . takeActions ( channelIds , response ) ; return response . multipleChannelOutput ( ) ; } remarks param channelId param allocations param appData returns async updateChannel ( { channelId , allocations , appData , } ) { const timer = timerFactory ( this . config . metrics . timingMetrics , `updateChannel ${ channelId }` ) ; const handleMissingChannel = ( ) => { throw new UpdateChannel . UpdateChannelError ( UpdateChannel . UpdateChannelError . reasons . channelNotFound , { channelId } ) ; } ; const criticalCode = async ( tx , channel ) => { const response = EngineResponse . initialize ( ) ; const { myIndex } = channel ; const outcome = recordFunctionMetrics ( deserializeAllocations ( allocations ) , this . config . metrics . timingMetrics ) ; const nextState = getOrThrow ( recordFunctionMetrics ( UpdateChannel . updateChannel ( { channelId , appData , outcome } , channel . protocolState ) , this . config . metrics . timingMetrics ) ) ; const signedState = await timer ( 'signing state' , async ( ) => { try { return this . store . signState ( channel , nextState , tx ) ; } catch ( err ) { this . logger . error ( { err , nextState } , 'Unable to update channel' ) ; throw err ; } } ) ; response . queueState ( signedState , myIndex , channelId ) ; const channelState = await this . store . getChannelState ( channelId , tx ) ; response . queueChannelState ( channelState ) ; return response . singleChannelOutput ( ) ; } ; return this . store . lockApp ( channelId , criticalCode , handleMissingChannel , true ) ; } remarks Signs, stores, and sends an isFinal=true state for each channel in the list.
     * Creates, approves and cranks a CloseChannel objective for each channel in the list. See  SingleThreadedEngine . closeChannel .
     *
     *  param channelId returns async closeChannels ( channelIds ) { const response = EngineResponse . initialize ( ) ; for ( const channelId of channelIds ) await this . _closeChannel ( channelId , response ) ; await this . takeActions ( channelIds , response ) ; return response . multipleChannelOutput ( ) ; } remarks param channelId returns async closeChannel ( { channelId , } ) { const response = EngineResponse . initialize ( ) ; await this . _closeChannel ( channelId , response ) ; await this . takeActions ( [ channelId ] , response ) ; const result = response . singleChannelOutput ( ) ; if ( ! hasNewObjective ( result ) ) { throw new Error ( 'No new objective created for closeChannel' ) ; } else { return result ; } } private async _closeChannel ( channelId , response ) { await this . objectiveManager . commenceCloseChannel ( channelId , response ) ; } param participants returns async getLedgerChannels ( participants ) { const response = EngineResponse . initialize ( ) ; const channelStates = await this . store . getLedgerChannels ( participants . map ( convertToParticipant ) ) ; channelStates . forEach ( cs => response . queueChannelState ( cs ) ) ; return response . multipleChannelOutput ( ) ; } returns async getChannels ( ) { const response = EngineResponse . initialize ( ) ; const channelStates = await this . store . getChannels ( ) ; channelStates . forEach ( cs => response . queueChannelState ( cs ) ) ; return response . multipleChannelOutput ( ) ; } returns async getObjective ( objectiveId ) { return this . store . getObjective ( objectiveId ) ; } returns async getApprovedObjectives ( ) { return this . store . getApprovedObjectives ( ) ; } privateRemarks returns async getState ( { channelId } ) { const response = EngineResponse . initialize ( ) ; try { const channel = await this . store . getChannelState ( channelId ) ; response . queueChannelState ( channel ) ; return response . singleChannelOutput ( ) ; } catch ( err ) { this . logger . error ( { err } , 'Could not get channel' ) ; throw err ; } } remarks param rawPayload returns async pushMessage ( rawPayload ) { const wirePayload = validatePayload ( rawPayload ) ; const response = EngineResponse . initialize ( ) ; try { await this . _pushMessage ( wirePayload , response ) ; return response . multipleChannelOutput ( ) ; } catch ( err ) { this . logger . error ( { err } , 'Error during pushMessage' ) ; throw new PushMessageError ( 'Error during pushMessage' , { thisWalletVersion : WALLET_VERSION , payloadWalletVersion : wirePayload . walletVersion , cause : err , } ) ; } } remarks param rawPayload returns async pushUpdate ( rawPayload ) { const wirePayload = validatePayload ( rawPayload ) ; const response = EngineResponse . initialize ( ) ; await SingleAppUpdater . create ( this . store ) . update ( wirePayload , response ) ; return response . singleChannelOutput ( ) ; } private async _pushMessage ( wirePayload , response ) { const store = this . store ; const { channelIds : channelIdsFromStates , channelResults : fromStoring , storedObjectives , } = await this . store . pushMessage ( wirePayload ) ; response . createdObjectives = storedObjectives ; const channelIdsFromRequests = [ ] ; const requests = wirePayload . requests || [ ] ; for ( const request of requests ) { const { channelId } = request ; channelIdsFromRequests . push ( channelId ) ; switch ( request . type ) { case 'GetChannel' : { const { states : signedStates , channelState } = await store . getStates ( channelId ) ; signedStates . forEach ( s => response . queueState ( s , channelState . myIndex , channelId ) ) ; continue ; } default : continue ; } } fromStoring . forEach ( cr => response . queueChannelResult ( cr ) ) ; const channelIds = _ . uniq ( channelIdsFromStates . concat ( channelIdsFromRequests ) ) ; await this . takeActions ( channelIds , response ) ; } remarks param channels param response private async takeActions ( channelIds , response ) { while ( channelIds . length > 0 ) { const objectiveIds = await this . store . getApprovedObjectiveIds ( channelIds ) ; for ( const objectiveId of objectiveIds ) { await this . objectiveManager . crank ( objectiveId , response ) ; } const ledgersFromRequests = await this . store . getLedgersWithNewRequestsIds ( ) ; const ledgersFromChannels = await this . store . filterChannelIdsByIsLedger ( channelIds ) ; const ledgersToProcess = _ . uniq ( ledgersFromChannels . concat ( ledgersFromRequests ) ) ; channelIds = [ ] ; for ( const ledgerChannelId of ledgersToProcess ) { const touchedChannels = await this . ledgerManager . crank ( ledgerChannelId , response ) ; channelIds = [ ... channelIds , ... touchedChannels ] ; } } } } export function getOrThrow < E , T > ( result ) { return Either . getOrElseW < E , T > ( ( err ) : T => { throw err ; } ) ( result ) ; }