import express , { Express } from 'express' ; import _ from 'lodash' ; import chalk from 'chalk' ; import P from 'pino' ; import got from 'got' ; import ms from 'ms' ; import { StopWatch } from 'stopwatch-node' ; import prettyMilliseconds from 'pretty-ms' ; import { WalletConfig } from '../src/config' ; import { ObjectiveDoneResult , UpdateChannelResult , Wallet } from '../src/wallet' ; import { SocketIOMessageService } from '../src/message-service/socket-io-message-service' ; import { createLogger } from '../src/logger' ; import { LatencyOptions } from '../src/message-service/test-message-service' ; import { CloseChannelStep , CreateChannelStep , CreateLedgerChannelStep , JobChannelLink , LoadNodeConfig , Peers , Step , UpdateChannelStep , } from './types' ; export class WalletLoadNode { private steps : Step [ ] = [ ] ; private jobToChannelMap : JobChannelLink [ ] = [ ] ; private completedSteps = 0 ; private server : Express ; private proposedObjectives = new Set < string > ( ) ; private proposedObjectivePromises = new Array < Promise < ObjectiveDoneResult > > ( ) ; private constructor ( private serverWallet , private loadNodeConfig , private peers , private logger ) { this . serverWallet . on ( 'ObjectiveProposed' , async ( o ) => { if ( o . type === 'OpenChannel' ) { if ( ! this . proposedObjectives . has ( o . objectiveId ) ) { this . logger . trace ( { objectiveId : o . objectiveId } , 'Auto approving objective' ) ; const [ result ] = await this . serverWallet . approveObjectives ( [ o . objectiveId ] ) ; this . proposedObjectives . add ( o . objectiveId ) ; this . proposedObjectivePromises . push ( result . done ) ; } } } ) ; this . server = express ( ) ; this . server . use ( express . json ( { limit : '50mb' } ) ) ; this . server . get ( '/start' , async ( req , res ) => { req . setTimeout ( ms ( '1 day' ) ) ; this . logger . trace ( 'Starting job processing' ) ; console . log ( chalk . whiteBright ( 'Starting job processing..' ) ) ; this . outputChannelStats ( ) ; const fromPeer = req . query [ 'fromPeer' ] ; const stopWatch = new StopWatch ( ) ; stopWatch . start ( ) ; if ( ! fromPeer ) { await Promise . all ( [ this . runJobs ( ) , this . sendGetRequestToPeers ( '/start?fromPeer=true' ) ] ) ; } else { await this . runJobs ( ) ; } await Promise . all ( this . proposedObjectivePromises ) ; stopWatch . stop ( ) ; console . log ( `Job processing took ${ prettyMilliseconds ( stopWatch . getTotalTime ( ) ) }` ) ; res . end ( ) ; } ) ; this . server . get ( '/reset' , async ( req , res ) => { this . steps = [ ] ; this . jobToChannelMap = [ ] ; const fromPeer = req . query [ 'fromPeer' ] ; if ( ! fromPeer ) { await this . sendGetRequestToPeers ( '/reset?fromPeer=true' ) ; } this . logger . info ( 'Jobs have been reset' ) ; res . end ( ) ; } ) ; this . server . post ( '/channelId' , async ( req , res ) => { const { channelId , jobId } = req . body ; this . jobToChannelMap . push ( { jobId , channelId } ) ; res . end ( ) ; } ) ; this . server . post ( '/load' , async ( req , res ) => { const requests = req . body ; const fromPeer = this . parseBooleanQueryParam ( 'fromPeer' , req ) ; const startProcessing = this . parseBooleanQueryParam ( 'start' , req ) ; await this . updateJobQueue ( requests ) ; if ( ! fromPeer ) { await this . shareJobsWithPeers ( requests ) ; } if ( startProcessing ) { await Promise . all ( [ this . sendGetRequestToPeers ( '/start?fromPeer=true' ) , this . runJobs ( ) ] ) ; } res . end ( ) ; } ) ; this . server . get ( '/' , ( req , res ) => res . end ( ) ) ; } private outputChannelStats ( ) { const ourSteps = this . steps . filter ( s => s . serverId === this . loadNodeConfig . serverId ) ; const createdAmount = ourSteps . filter ( s => s . type === 'CreateChannel' ) . length ; const ledgerAmount = ourSteps . filter ( s => s . type === 'CreateLedgerChannel' ) . length ; const isLedgerFunding = this . steps . some ( s => s . type === 'CreateLedgerChannel' ) ; console . log ( chalk . whiteBright ( `This node will create ${ createdAmount } channel(s) using ${ isLedgerFunding ? 'ledger' : 'direct' } funding` ) ) ; if ( ledgerAmount > 0 ) { console . log ( chalk . whiteBright ( `This node will create ${ ledgerAmount } ledger channel(s)` ) ) ; } const closedAmount = ourSteps . filter ( s => s . type === 'CloseChannel' ) . length ; if ( closedAmount > 0 ) { console . log ( chalk . whiteBright ( `This node will close ${ closedAmount } channel(s)` ) ) ; } } private parseBooleanQueryParam ( paramName , req ) { const param = req . query [ paramName ] ; return ! ! param && param . toString ( ) . toLocaleLowerCase ( ) === 'true' ; } returns private runJobs ( ) { const startTimestamp = Date . now ( ) ; return new Promise < void > ( resolve => { this . steps . filter ( s => s . serverId === this . loadNodeConfig . serverId ) . forEach ( s => setTimeout ( async ( ) => { this . logger . trace ( { step : s } , 'Starting job' ) ; const result = await this . handleStep ( s ) ; if ( result . type !== 'Success' ) { this . logger . error ( { result , step : s } , 'Load step failed!' ) ; console . error ( chalk . redBright ( `Step failed for ${ s . jobId }-${ s . timestamp }` ) ) ; process . exit ( 1 ) ; } else { this . completedSteps ++ ; this . logger . trace ( { step : s , timestamp : Date . now ( ) - startTimestamp } , 'Completed job' ) ; const ourSteps = this . steps . filter ( s => s . serverId === this . loadNodeConfig . serverId ) ; if ( this . completedSteps === ourSteps . length ) { this . logger . trace ( { jobIds : this . jobToChannelMap . map ( e => e . jobId ) } , 'All jobs completed!' ) ; console . log ( chalk . greenBright ( `All jobs complete!` ) ) ; resolve ( ) ; } } } , s . timestamp ) ) ; } ) ; } private async shareChannelIdsWithPeers ( jobAndChannel ) { for ( const { loadServerPort } of this . peers ) { await got . post ( `http://localhost:${ loadServerPort }/channelId/` , { json : jobAndChannel } ) ; } } private async shareJobsWithPeers ( steps ) { for ( const { loadServerPort } of this . peers ) { await got . post ( `http://localhost:${ loadServerPort }/load/?fromPeer=true` , { json : steps } ) ; } } private async sendGetRequestToPeers ( urlPath ) { for ( const { loadServerPort } of this . peers ) { await got . get ( `http://localhost:${ loadServerPort }${ urlPath }` , { retry : 0 } ) ; } } private async updateJobQueue ( steps ) { this . steps = _ . uniq ( this . steps . concat ( steps ) ) ; console . log ( chalk . yellow ( `Updated job queue with ${ steps . length } steps` ) ) ; this . logger . info ( { steps } , 'Updated queue with steps' ) ; } private getChannelIdForJob ( jobId ) { const entry = this . jobToChannelMap . find ( e => e . jobId === jobId ) ; if ( ! entry ) { throw new Error ( `No channel id for ${ jobId }` ) ; } return entry . channelId ; } /**
     * These are the various handlers for different step types
     * TODO: Typing should not use any type for the request
     */ private stepHandlers : Record < Step [ 'type' ] , ( step : any ) => Promise < ObjectiveDoneResult > > = { CreateChannel : async ( step ) => { const { fundingInfo } = step ; if ( fundingInfo . type === 'Direct' ) { const [ result ] = await this . serverWallet . createChannels ( [ { ... step . channelParams , fundingStrategy : 'Direct' } , ] ) ; const { jobId } = step ; const { channelId } = result ; this . jobToChannelMap . push ( { jobId , channelId } ) ; await this . shareChannelIdsWithPeers ( { jobId , channelId } ) ; return result . done ; } else { const ledgerResult = this . jobToChannelMap . find ( j => j . jobId === fundingInfo . fundingLedgerJob ) ; if ( ! ledgerResult ) { throw new Error ( `Cannot find channel id for ledger job ${ fundingInfo . fundingLedgerJob }` ) ; } const { channelId : fundingLedgerChannelId } = ledgerResult ; const [ result ] = await this . serverWallet . createChannels ( [ { ... step . channelParams , fundingStrategy : 'Ledger' , fundingLedgerChannelId } , ] ) ; const { jobId } = step ; const { channelId } = result ; this . jobToChannelMap . push ( { jobId , channelId } ) ; await this . shareChannelIdsWithPeers ( { jobId , channelId } ) ; return result . done ; } } , CloseChannel : async ( step ) => { const channelId = this . getChannelIdForJob ( step . jobId ) ; const [ result ] = await this . serverWallet . closeChannels ( [ channelId ] ) ; return result . done ; } , UpdateChannel : async ( step ) => { const channelId = this . getChannelIdForJob ( step . jobId ) ; const { allocations , appData } = step . updateParams ; const result = await this . serverWallet . updateChannel ( channelId , allocations , appData ) ; return result ; } , CreateLedgerChannel : async ( step ) => { const result = await this . serverWallet . createLedgerChannel ( { ... step . ledgerChannelParams , fundingStrategy : 'Direct' , } ) ; const { jobId } = step ; const { channelId } = result ; this . jobToChannelMap . push ( { jobId , channelId } ) ; await this . shareChannelIdsWithPeers ( { jobId , channelId } ) ; return result . done ; } , } ; param step returns private async handleStep ( step ) { return this . stepHandlers [ step . type ] ( step ) ; } public async destroy ( ) { this . server . removeAllListeners ( ) ; await this . serverWallet . destroy ( ) ; } public listen ( ) { this . server . listen ( this . loadNodeConfig . loadServerPort ) ; } param port public async registerMessagePeer ( port ) { this . serverWallet . messageService . registerPeer ( `http://localhost:${ port }` ) ; } public setLatencyOptions ( latencyOptions ) { ( this . serverWallet . messageService as SocketIOMessageService ) . setLatencyOptions ( latencyOptions ) ; } public static async create ( walletConfig , loadNodeConfig , peers ) { const messageServiceFactory = await SocketIOMessageService . createFactory ( 'localhost' , loadNodeConfig . messagePort ) ; const serverWallet = await Wallet . create ( walletConfig , messageServiceFactory ) ; const logger = createLogger ( walletConfig ) ; return new WalletLoadNode ( serverWallet , loadNodeConfig , peers , logger ) ; } public get loadPort ( ) : number { return this . loadNodeConfig . loadServerPort ; } }