import { Transaction , TransactionOrKnex , UniqueViolationError } from 'objection' ; import { Outcome , SignedStateWithHash , SignedStateVarsWithHash , calculateChannelId , StateVariables , ChannelConstants , Participant , deserializeObjective , isOpenChannel , isCloseChannel , SignedState , OpenChannel , makePrivateKey , makeAddress , Address , PrivateKey , isSubmitChallenge , BN , } from '@statechannels/wallet-core' ; import { Payload as WirePayload , SignedState as WireSignedState } from '@statechannels/wire-format' ; import _ from 'lodash' ; import { ChannelResult , FundingStrategy } from '@statechannels/client-api-schema' ; import { ethers } from 'ethers' ; import Knex from 'knex' ; import { Logger } from 'pino' ; import pMap from 'p-map' ; import { Channel , ChannelError , CHANNEL_COLUMNS } from '../models/channel' ; import { SigningWallet } from '../models/signing-wallet' ; import { addHash , addState , clearOldStates , fastDeserializeState } from '../state-utils' ; import { ChannelState } from '../protocols/state' ; import { EngineError , Values } from '../errors/engine-error' ; import { Bytes32 , Uint256 , Bytes } from '../type-aliases' ; import { timerFactory , recordFunctionMetrics , setupDBMetrics , timerFactorySync } from '../metrics' ; import { isReverseSorted , pick } from '../utilities/helpers' ; import { Funding , TransferredOutEntry } from '../models/funding' ; import { Nonce } from '../models/nonce' ; import { ObjectiveModel , WalletObjective , isSupportedObjective } from '../models/objective' ; import { AppBytecode } from '../models/app-bytecode' ; import { LedgerRequest } from '../models/ledger-request' ; import { shouldValidateTransition , validateTransition } from '../utilities/validate-transition' ; import { defaultTestWalletConfig } from '../config' ; import { createLogger } from '../logger' ; import { ChainServiceRequest } from '../models/chain-service-request' ; import { AdjudicatorStatusModel } from '../models/adjudicator-status' ; import { WaitingFor as OpenChannelWaitingFor } from '../protocols/channel-opener' ; export type AppHandler < T > = ( tx : Transaction , channelRecord : Channel ) => T ; export type MissingAppHandler < T > = ( channelId : string ) => T ; const throwMissingChannel = ( channelId ) => { throw new ChannelError ( ChannelError . reasons . channelMissing , { channelId } ) ; } ; export class Store { readonly logger : Logger ; constructor ( public readonly knex , readonly timingMetrics , readonly skipEvmValidation , readonly chainNetworkID , logger ? ) { this . logger = logger ?? createLogger ( defaultTestWalletConfig ( ) ) ; if ( timingMetrics ) { this . getOrCreateSigningAddress = recordFunctionMetrics ( this . getOrCreateSigningAddress ) ; this . lockApp = recordFunctionMetrics ( this . lockApp ) ; this . signState = recordFunctionMetrics ( this . signState ) ; this . getChannelState = recordFunctionMetrics ( this . getChannelState ) ; this . getStates = recordFunctionMetrics ( this . getStates ) ; this . getChannels = recordFunctionMetrics ( this . getChannels ) ; this . pushMessage = recordFunctionMetrics ( this . pushMessage ) ; this . addSignedState = recordFunctionMetrics ( this . addSignedState ) ; setupDBMetrics ( this . knex ) ; } } async addSigningKey ( privateKey ) { await this . getOrCreateSigningAddress ( new ethers . Wallet ( privateKey ) ) ; } async getOrCreateSigningAddress ( wallet = ethers . Wallet . createRandom ( ) ) { try { const signingWallet = await SigningWallet . query ( this . knex ) . insert ( { privateKey : makePrivateKey ( wallet . privateKey ) , address : makeAddress ( wallet . address ) , } ) . returning ( '*' ) ; return signingWallet . address ; } catch ( error ) { if ( error instanceof UniqueViolationError ) { return ( await SigningWallet . query ( this . knex ) . first ( ) ) . address ; } throw error ; } } param channelId param criticalCode param onChannelMissing param fetchSigningWallet async lockApp < T > ( channelId , criticalCode , onChannelMissing = throwMissingChannel , fetchSigningWallet = false ) { return this . knex . transaction ( async ( tx ) => { const timer = timerFactory ( this . timingMetrics , `lock app ${ channelId }` ) ; const channel = await timer ( 'getting channel' , ( ) => { const query = Channel . query ( tx ) . where ( { channelId } ) . forUpdate ( ) . first ( ) ; if ( fetchSigningWallet ) return query . withGraphFetched ( 'signingWallet' ) ; return query ; } ) ; if ( ! channel ) return onChannelMissing ( channelId ) ; return timer ( 'critical code' , async ( ) => criticalCode ( tx , channel ) ) ; } ) ; } async transaction < T > ( callback ) { return this . knex . transaction ( async ( tx ) => callback ( tx ) ) ; } async signState ( channel , vars , tx ) { const signingWallet = channel . signingWallet || ( await channel . $relatedQuery ( 'signingWallet' , tx ) . first ( ) ) ; if ( ! signingWallet ) { throw new Error ( 'No signing wallets' ) ; } const timer = timerFactory ( this . timingMetrics , `signState ${ channel . channelId }` ) ; const state = addHash ( { ... channel . channelConstants , ... vars } ) ; const { supported , latestSignedByMe } = channel ; if ( latestSignedByMe && latestSignedByMe . turnNum >= state . turnNum ) throw new StoreError ( StoreError . reasons . staleState ) ; const signatureEntry = await timer ( 'signing' , async ( ) => signingWallet . signState ( state ) ) ; const signedState = { ... state , signatures : [ signatureEntry ] } ; if ( supported && shouldValidateTransition ( state , channel ) ) { const bytecode = await this . getBytecode ( supported . appDefinition , tx ) ; if ( ! this . skipEvmValidation && ! bytecode ) this . logger . error ( 'Missing bytecode' , { error : new Error ( `No byte code for ${ supported . appDefinition }` ) , } ) ; if ( ! ( await timer ( 'validating transition' , async ( ) => validateTransition ( supported , signedState , this . logger , bytecode , this . skipEvmValidation ) ) ) ) { throw new StoreError ( 'Invalid state transition' , { from : channel . supported , to : signedState , } ) ; } } await timer ( 'adding MY state' , async ( ) => this . addMyState ( channel , signedState , tx ) ) ; return signedState ; } async addMyState ( channel , signedState , tx ) { if ( signedState . signatures . length > 1 || signedState . signatures [ 0 ] . signer !== channel . myAddress ) { throw new Error ( 'This state not exclusively signed by me' ) ; } const timer = timerFactory ( this . timingMetrics , `addMyState ${ channel . channelId }` ) ; channel . vars = await timer ( 'adding state' , async ( ) => addState ( channel . vars , signedState ) ) ; channel . vars = clearOldStates ( channel . vars , channel . isSupported ? channel . support : undefined ) ; await timer ( 'validating invariants' , async ( ) => validateInvariants ( channel . vars , channel . myAddress ) ) ; const result = await timer ( 'updating' , async ( ) => Channel . query ( tx ) . where ( { channelId : channel . channelId } ) . patch ( { vars : channel . vars } ) . returning ( '*' ) . first ( ) ) ; return result ; } async getChannelState ( channelId , tx ? ) { const channel = await this . getChannel ( channelId , tx ) ; if ( ! channel ) throw new Error ( `Channel not found with id ${ channelId }` ) ; return channel . protocolState ; } async getChannel ( channelId , tx ? ) { return Channel . query ( tx ?? this . knex ) . where ( { 'channels.channel_id' : channelId } ) . withGraphJoined ( 'funding' ) . withGraphJoined ( 'chainServiceRequests' ) . withGraphJoined ( 'adjudicatorStatus' ) . first ( ) ; } async getAndLockChannel ( channelId , tx ) { const channel = await Channel . query ( tx ) . where ( { channelId } ) . withGraphJoined ( 'adjudicatorStatus' ) . withGraphFetched ( 'signingWallet' ) . forUpdate ( ) . first ( ) ; if ( ! channel ) { throw new ChannelError ( 'No channel found with id' , { channelId } ) ; } return channel ; } async getStates ( channelId , tx ? ) { const channel = await Channel . forId ( channelId , tx || this . knex ) ; if ( ! channel ) throw new StoreError ( StoreError . reasons . channelMissing ) ; const { vars , channelConstants , protocolState : channelState } = channel ; return { states : vars . map ( ss => _ . merge ( ss , channelConstants ) ) , channelState } ; } async getObjectivesByIds ( objectiveIds ) { return ObjectiveModel . forIds ( objectiveIds , this . knex ) ; } async getChannels ( ) { return ( await Channel . query ( this . knex ) ) . map ( channel => channel . protocolState ) ; } /**
     * Returns all channels that are not finalized on chain
     */ async getNonFinalizedChannels ( ) { return ( await Channel . query ( this . knex ) ) . filter ( c => ! c . adjudicatorStatus || c . adjudicatorStatus . toResult ( ) . channelMode === 'Open' ) . map ( channel => channel . protocolState ) ; } async getLedgerChannels ( participants ) { const ledgers = await Channel . getLedgerChannels ( participants , this . knex ) ; return ledgers . map ( c => c . protocolState ) ; } async pushMessage ( message ) { return this . knex . transaction ( async ( tx ) => { const channelResults = [ ] ; const sortedSignedStates = _ . sortBy ( message . signedStates || [ ] , a => a . channelNonce ) ; const stateChannelIds = message . signedStates ?. map ( ss => ss . channelId ) || [ ] ; for ( const ss of sortedSignedStates || [ ] ) { const channel = await this . addSignedState ( ss . channelId , ss , tx ) ; channelResults . push ( channel . channelResult ) ; } const deserializedObjectives = message . objectives ?. map ( deserializeObjective ) || [ ] ; const storedObjectives = [ ] ; for ( const o of deserializedObjectives ) { if ( isSupportedObjective ( o ) ) { const status = o . type === 'CloseChannel' ? 'approved' : undefined ; storedObjectives . push ( await ObjectiveModel . insert ( o , tx , status ) ) ; } else this . logger . warn ( 'Unsupported objective received' ) ; } const objectiveChannelIds = storedObjectives . filter ( objective => isSubmitChallenge ( objective ) || isOpenChannel ( objective ) || isCloseChannel ( objective ) ) . map ( objective => objective . data . targetChannelId ) ; return { channelIds : stateChannelIds . concat ( objectiveChannelIds ) , channelResults , storedObjectives , } ; } ) ; } async getObjectives ( channelIds , tx ? ) { return await ObjectiveModel . forChannelIds ( channelIds , tx || this . knex ) ; } async getApprovedObjectiveIds ( channelIds , tx ? ) { return await ObjectiveModel . approvedObjectiveIds ( channelIds , tx || this . knex ) ; } public async getApprovedObjectives ( ) { const results = await ObjectiveModel . query ( this . knex ) . where ( { status : 'approved' } ) ; return results . map ( o => o . toObjective ( ) ) ; } public async updateObjectiveProgressDate ( objectiveIds ) { return this . knex . transaction ( async ( tx ) => Promise . all ( objectiveIds . map ( oId => ObjectiveModel . updateProgressDate ( oId , tx ) ) ) ) ; } async getLedgersWithNewRequestsIds ( tx ? ) { return LedgerRequest . ledgersWithNewRequestsIds ( tx || this . knex ) ; } async approveObjective ( objectiveId , tx ? ) { const objective = await ObjectiveModel . approve ( objectiveId , tx || this . knex ) ; if ( objective . type === 'OpenChannel' ) { const channelId = objective . data . targetChannelId ; if ( objective . data . role === 'ledger' ) { await Channel . setLedger ( channelId , tx || this . knex ) ; } const { fundingStrategy , fundingLedgerChannelId } = objective . data ; await Channel . query ( tx || this . knex ) . where ( { channelId } ) . patch ( { fundingStrategy , fundingLedgerChannelId } ) ; } return objective ; } async markObjectiveStatus < O extends WalletObjective > ( objective , status , tx ? ) { if ( status === 'succeeded' ) { return ( await ObjectiveModel . succeed ( objective . objectiveId , tx || this . knex ) ) . toObjective ( ) as O ; } else { return ( await ObjectiveModel . failed ( objective . objectiveId , tx || this . knex ) ) . toObjective ( ) as O ; } } private bytecodeCache : Record < string , string | undefined > = { } ; async getBytecode ( appDefinition , tx ) { return ( this . bytecodeCache [ appDefinition ] ?? ( this . bytecodeCache [ appDefinition ] = await AppBytecode . getBytecode ( this . chainNetworkID , appDefinition , tx ) ) ) ; } async upsertBytecode ( chainNetworkId , appDefinition , bytecode ) { await AppBytecode . upsertBytecode ( chainNetworkId , appDefinition , bytecode , this . knex ) ; } async getObjective ( objectiveId , tx ? ) { tx = tx || this . knex ; return await ObjectiveModel . forId ( objectiveId , tx ) ; } param objectiveId param tx async getAndLockObjective ( objectiveId , tx ) { return ( await ObjectiveModel . query ( tx ) . findById ( objectiveId ) . forUpdate ( ) ) . toObjective ( ) ; } async isLedger ( channelId , tx ? ) { return Channel . isLedger ( channelId , tx || this . knex ) ; } async getLedgerRequest ( channelId , type , tx ? ) { return LedgerRequest . getRequest ( channelId , type , tx || this . knex ) ; } async getChannelIdsPendingLedgerFundingFrom ( ledgerChannelIds , tx ? ) { const query = await LedgerRequest . query ( tx || this . knex ) . whereIn ( 'ledgerChannelId' , ledgerChannelIds ) . where ( { status : 'pending' } ) . distinct ( 'channelToBeFunded' ) . select ( 'channelToBeFunded' ) ; return _ . map ( query , 'channelToBeFunded' ) ; } async getLedgerChannelIdsFundingChannels ( channelToBeFunded , tx ? ) { const query = await LedgerRequest . query ( tx || this . knex ) . whereIn ( 'channelToBeFunded' , channelToBeFunded ) . where ( { status : 'pending' } ) . orWhere ( { status : 'queued' } ) . distinct ( 'ledgerChannelId' ) . select ( 'ledgerChannelId' ) ; return _ . map ( query , 'ledgerChannelId' ) ; } async filterChannelIdsByIsLedger ( maybeLedgerChannelIds , tx ? ) { const query = await Channel . query ( tx || this . knex ) . whereIn ( 'channelId' , maybeLedgerChannelIds ) . where ( { isLedgerChannel : true } ) . select ( 'channelId' ) ; return _ . map ( query , 'channelId' ) ; } async getActiveLedgerRequests ( ledgerChannelId , tx ? ) { return LedgerRequest . getActiveRequests ( ledgerChannelId , tx || this . knex ) ; } async markLedgerRequests ( channelIds , type , status , tx ? ) { await pMap ( channelIds , request => LedgerRequest . setRequestStatus ( request , type , status , tx || this . knex ) ) ; } param channelId param state param tx async addSignedState ( channelId , wireState , tx ) { const syncTimer = timerFactorySync ( this . timingMetrics , `add signed state ${ channelId }` ) ; const asyncTimer = timerFactory ( this . timingMetrics , `add signed state ${ channelId }` ) ; const deserializedState = syncTimer ( 'validating signatures' , ( ) => fastDeserializeState ( channelId , wireState ) ) ; if ( ! BN . eq ( deserializedState . chainId , this . chainNetworkID ) ) throw Error ( `This channel can only be concluded on a chain with id ${ deserializedState . chainId }, but this wallet uses chain id: ${ this . chainNetworkID }.` ) ; const state = addHash ( deserializedState ) ; const channel = ( await Channel . forId ( channelId , tx ) ) || ( await createChannel ( state , 'Unknown' , undefined , tx ) ) ; const { supported } = channel ; if ( supported && shouldValidateTransition ( state , channel ) ) { const bytecode = await this . getBytecode ( supported . appDefinition , tx ) ; if ( bytecode == undefined ) this . logger . error ( 'Missing bytecode' , { error : new Error ( `No byte code for ${ supported . appDefinition }` ) , } ) ; const isValidTransition = syncTimer ( 'validating transition' , ( ) => validateTransition ( supported , state , this . logger , bytecode , this . skipEvmValidation ) ) ; if ( ! isValidTransition ) { throw new StoreError ( 'Invalid state transition' , { from : supported , to : state , } ) ; } } channel . vars = syncTimer ( 'adding state' , ( ) => addState ( channel . vars , state ) ) ; channel . vars = clearOldStates ( channel . vars , channel . isSupported ? channel . support : undefined ) ; syncTimer ( 'validating invariants' , ( ) => validateInvariants ( channel . vars , channel . myAddress ) ) ; return await asyncTimer ( 'updating' , ( ) => Channel . query ( tx ) . where ( { channelId : channel . channelId } ) . patch ( { vars : channel . vars } ) . returning ( '*' ) . first ( ) ) ; } async createChannel ( constants , appData , outcome , fundingStrategy , role = 'app' , fundingLedgerChannelId ? ) { return await this . knex . transaction ( async ( tx ) => { const channel = await createChannel ( constants , fundingStrategy , fundingLedgerChannelId , tx ) ; const { channelId , participants } = channel ; const firstSignedState = await this . signState ( channel , { ... constants , turnNum : 0 , isFinal : false , appData , outcome , } , tx ) ; if ( role === 'ledger' ) await Channel . setLedger ( channelId , tx ) ; const o = { type : 'OpenChannel' as const , participants , data : { targetChannelId : channelId , fundingStrategy , fundingLedgerChannelId , role , } , } ; const objective = await ObjectiveModel . insert ( o , tx , 'approved' , OpenChannelWaitingFor . theirPreFundSetup ) ; return { channel : await this . getChannelState ( channelId , tx ) , firstSignedState , objective } ; } ) ; } async setInitialSupport ( channelId , support , tx ? ) { await Channel . setInitialSupport ( channelId , support , tx ?? this . knex ) ; } async updateFunding ( channelId , fromAmount , asset ) { await Funding . updateFunding ( this . knex , channelId , fromAmount , asset ) ; } async insertAdjudicatorStatus ( channelId , finalizedAt , states ) { await AdjudicatorStatusModel . insertAdjudicatorStatus ( this . knex , channelId , finalizedAt , states ) ; } async markAdjudicatorStatusAsFinalized ( channelId , blockNumber , blockTimestamp , finalizedAt ) { this . knex . transaction ( async ( tx ) => { await this . getAndLockChannel ( channelId , tx ) ; await AdjudicatorStatusModel . setFinalized ( tx , channelId , blockNumber , blockTimestamp , finalizedAt ) ; } ) ; } async updateTransferredOut ( channelId , asset , transferredOut , newHoldings ) { this . lockApp ( channelId , async ( tx ) => { await Funding . updateTransferredOut ( tx , channelId , asset , transferredOut ) ; await Funding . updateFunding ( tx , channelId , newHoldings , asset ) ; } ) ; } async nextNonce ( signingAddresses ) { return await Nonce . next ( this . knex , signingAddresses ) ; } async fundingRequestExists ( channelId , tx ) { const request = await ChainServiceRequest . query ( tx ) . where ( { channelId , request : 'fund' } ) . first ( ) ; return ! ! request && request . isValid ( ) ; } async getFunding ( channelId , asset , tx ) { const funding = Funding . query ( tx ) . where ( { channelId , asset } ) . first ( ) ; return funding ; } } class StoreError extends EngineError { readonly type = EngineError . errors . StoreError ; static readonly reasons = { duplicateChannel : 'Expected the channel to NOT exist in the database' , duplicateTurnNums : 'multiple states with same turn number' , notSorted : 'states not sorted' , multipleSignedStates : 'Store signed multiple states for a single turn' , invalidSignature : 'Invalid signature' , notInChannel : 'Not in channel' , staleState : 'Stale state' , missingSigningKey : 'Missing a signing key' , invalidTransition : 'Invalid state transition' , channelMissing : 'Channel not found' , unimplementedObjective : 'Unimplemented objective' , unimplementedFundingStrategy : 'Unimplemented funding strategy' , invalidFundingLedgerChannelId : 'Ledger channel intended to fund channel does not exist' , expiredFundingLedgerChannelId : 'Ledger channel intended to fund channel is closed or closing' , } as const ; constructor ( reason , public readonly data = undefined ) { super ( reason ) ; } } /**
 * Construct a new Channel and insert it into the database, ensuring that the Nonce
 * provided for this new channel is also the newest we have ever seen.
 */ async function createChannel ( constants , fundingStrategy , fundingLedgerChannelId , txOrKnex ) { if ( fundingLedgerChannelId ) { const ledger = await Channel . forId ( fundingLedgerChannelId , txOrKnex ) ; if ( ! ledger ) throw new StoreError ( StoreError . reasons . invalidFundingLedgerChannelId ) ; if ( ledger . supported && _ . some ( ledger . support , 'isFinal' ) ) throw new StoreError ( StoreError . reasons . expiredFundingLedgerChannelId ) ; } const addresses = constants . participants . map ( p => p . signingAddress ) ; const signingWallet = await SigningWallet . query ( txOrKnex ) . whereIn ( 'address' , addresses ) . first ( ) ; if ( ! signingWallet ) throw new StoreError ( StoreError . reasons . notInChannel ) ; await Nonce . ensureLatest ( constants . channelNonce , addresses , txOrKnex ) ; const channel = Channel . fromJson ( pick ( { ... constants , isLedgerChannel : false , channelId : calculateChannelId ( constants ) , fundingLedgerChannelId , fundingStrategy , signingAddress : signingWallet . address , vars : [ ] , } , ... CHANNEL_COLUMNS ) ) ; return await Channel . query ( txOrKnex ) . withGraphFetched ( 'signingWallet' ) . insert ( channel ) . returning ( '*' ) . first ( ) ; } function ensureSingleSignedState ( states , mySigningAddress ) { const multipleSignedByMe = _ . chain ( states ) . filter ( state => _ . some ( state . signatures , [ 'signer' , mySigningAddress ] ) ) . groupBy ( 'turnNum' ) . some ( states => states . length > 1 ) . value ( ) ; if ( multipleSignedByMe ) throw new StoreError ( StoreError . reasons . multipleSignedStates ) ; } function ensureReverseSortedOrderOfStates ( states ) { if ( ! isReverseSorted ( _ . map ( states , 'turnNum' ) ) ) throw new StoreError ( StoreError . reasons . notSorted ) ; } function ensureNoDuplicateTurnNums ( states ) { const turnNums = _ . map ( states , 'turnNum' ) ; const duplicateTurnNums = turnNums . some ( ( t , i ) => turnNums . indexOf ( t ) != i ) ; if ( duplicateTurnNums ) { throw new StoreError ( StoreError . reasons . duplicateTurnNums , { states } ) ; } } /**
 * There are currently two invariants being checked:
 *
 * 1. Only signed a single state for any given turn number.
 * 2. Storing states in reverse sorted order.
 */ function validateInvariants ( states , mySigningAddress ) { ensureSingleSignedState ( states , mySigningAddress ) ; ensureNoDuplicateTurnNums ( states ) ; ensureReverseSortedOrderOfStates ( states ) ; }