import { DB } from '@pigi/core-db' ; import { BigNumber , DefaultSignatureVerifier , getLogger , ONE , SignatureVerifier , } from '@pigi/core-utils' ; import { abiEncodeTransaction , isTransferTransaction , RollupStateMachine , SignatureError , SignedTransaction , TransactionResult , } from '@pigi/rollup-core' ; import AsyncLock from 'async-lock' ; import FastPriorityQueue from 'fastpriorityqueue' ; import { RollupBlockBuilder } from '../types' ; const log = getLogger ( 'aggregator' ) ; export class Aggregator { public static readonly NEXT_TX_NUMBER_KEY = Buffer . from ( 'next_tx_num' ) ; private static readonly lock_key = 'lock' ; private readonly lock : AsyncLock ; private readonly transactionResultQueue : FastPriorityQueue < TransactionResult > ; private nextTransactionToProcess : BigNumber ; public static async create ( db , stateMachine , blockBuilder , signatureVerifier = DefaultSignatureVerifier . instance ( ) ) { const aggregator = new Aggregator ( db , stateMachine , blockBuilder , signatureVerifier ) ; await aggregator . init ( ) ; return aggregator ; } private constructor ( private readonly db , private readonly stateMachine , private readonly blockBuilder , private readonly signatureVerifier = DefaultSignatureVerifier . instance ( ) ) { this . lock = new AsyncLock ( ) ; this . transactionResultQueue = new FastPriorityQueue < TransactionResult > ( ( a , b ) => a . transactionNumber . lt ( b . transactionNumber ) ) ; this . nextTransactionToProcess = ONE ; } /**
     * Gets any unprocessed transaction results that may exist sends them to BlockBuilder
     * to make sure all previous transactions are accounted for before handling new ones.
     */ private async init ( ) { const nextTx = await this . db . get ( Aggregator . NEXT_TX_NUMBER_KEY ) ; if ( ! nextTx ) { log . info ( `No stored next transaction to process. Starting fresh.` ) ; return ; } this . nextTransactionToProcess = new BigNumber ( nextTx ) ; const results = await this . stateMachine . getTransactionResultsSince ( this . nextTransactionToProcess . sub ( ONE ) ) ; for ( const res of results ) { this . transactionResultQueue . add ( res ) ; } return this . processTransactionResultQueue ( ) ; } param signedTransaction returns public async handleTransaction ( signedTransaction ) { this . verifySignature ( signedTransaction ) ; const result = await this . stateMachine . applyTransaction ( signedTransaction ) ; this . transactionResultQueue . add ( result ) ; return this . processTransactionResultQueue ( ) ; } /**
     * Processes the transaction results queue in order, sending them to the Block Builder.
     */ private async processTransactionResultQueue ( ) { return this . lock . acquire ( Aggregator . lock_key , async ( ) => { while ( this . transactionResultQueue . peek ( ) && this . transactionResultQueue . peek ( ) . transactionNumber . eq ( this . nextTransactionToProcess ) ) { const res = this . transactionResultQueue . poll ( ) ; await this . blockBuilder . addTransactionResult ( res ) ; await this . db . put ( Aggregator . NEXT_TX_NUMBER_KEY , res . transactionNumber . toBuffer ( ) ) ; this . nextTransactionToProcess = res . transactionNumber . add ( ONE ) ; } } ) ; } private verifySignature ( signedTransaction ) { const signer = this . signatureVerifier . verifyMessage ( abiEncodeTransaction ( signedTransaction . transaction ) , signedTransaction . signature ) ; if ( signer !== signedTransaction . transaction . sender ) { log . debug ( `Signer does not match transaction sender. 
        Verified signer: [${ signer }].
        Signed Transaction: [${ JSON . stringify ( signedTransaction ) }].` ) ; throw new SignatureError ( ) ; } } }