import { QldbDriver , RetryConfig , Result } from 'amazon-qldb-driver-nodejs' ; import { BlockchainTimeModel , IBlockchain , ServiceVersionModel , TransactionModel , ValueTimeLockModel , } from '@sidetree/common' ; import { AnchoredDataSerializer } from '@sidetree/core' ; import { Timestamp } from 'aws-sdk/clients/apigateway' ; import QLDBSession from 'aws-sdk/clients/qldbsession' ; import moment from 'moment' ; const { version } = require ( '../package.json' ) ; interface ValueWithMetaData { blockAddress : { sequenceNo : number ; } ; data : { numberOfOperations : number ; anchorFileHash : string ; } ; metadata : { id : string ; txTime : Timestamp ; txId : string ; } ; } export interface TransactionModelQLDB extends TransactionModel { transactionTimestamp : number ; } export default class QLDBLedger implements IBlockchain { public qldbDriver : QldbDriver ; public transactionTable : string ; constructor ( ledgerName , tableName , config ? ) { let qldbConfig ; if ( config ) { qldbConfig = config ; } else { process . env . AWS_SDK_LOAD_CONFIG = '1' ; qldbConfig = { region : 'us-east-1' , } ; } const maxConcurrentTransactions = 10 ; const retryLimit = 4 ; const retryConfig = new RetryConfig ( retryLimit ) ; this . qldbDriver = new QldbDriver ( ledgerName , qldbConfig , maxConcurrentTransactions , retryConfig ) ; this . transactionTable = tableName ; } public getServiceVersion ( ) { return Promise . resolve ( { name : 'qldb' , version , } ) ; } private async execute ( query , args ? ) { const params = args ? [ args ] : [ ] ; return this . qldbDriver . executeLambda ( async ( txn ) => txn . execute ( query , ... params ) ) ; } private async executeWithRetry ( query , args ? , retryCount = 0 ) { return this . execute ( query , args ) . catch ( ( err ) => { if ( err . message . includes ( 'No open transaction' ) && retryCount < 1 ) { console . log ( `retrying query ${ query } with count ${ retryCount }` ) ; return this . executeWithRetry ( query , args , retryCount + 1 ) ; } else { throw new Error ( err . message ) ; } } ) ; } private async executeWithoutError ( query ) { return this . executeWithRetry ( query ) . catch ( ( err ) => err . message ) ; } public async reset ( ) { console . log ( 'resetting' , this . transactionTable ) ; await this . executeWithRetry ( `DELETE FROM ${ this . transactionTable }` ) ; } public async initialize ( ) { await this . executeWithoutError ( `CREATE TABLE ${ this . transactionTable }` ) ; } public async write ( anchorString ) { const { coreIndexFileUri , numberOfOperations , } = AnchoredDataSerializer . deserialize ( anchorString ) ; const now = new Date ( ) ; await this . executeWithRetry ( `INSERT INTO ${ this . transactionTable } ?` , { ... { anchorFileHash : coreIndexFileUri , numberOfOperations , } , created : now . getTime ( ) , createdHumanReadable : now . toISOString ( ) , } ) ; } private toSidetreeTransaction ( qldbResult ) { const { blockAddress , data , metadata } = qldbResult ; const transactionTimeHash = metadata . id . toString ( ) ; const transactionTime = Number ( blockAddress . sequenceNo ) ; const transactionTimestamp = metadata . txTime . getTime ( ) ; const { anchorFileHash } = data ; const numberOfOperations = Number ( data . numberOfOperations ) ; const anchorString = AnchoredDataSerializer . serialize ( { coreIndexFileUri : anchorFileHash , numberOfOperations , } ) ; return { transactionTime , transactionNumber : transactionTime , transactionTimeHash , transactionTimestamp , anchorString , transactionFeePaid : 0 , normalizedTransactionFee : 0 , writer : 'writer' , } as any ; } public async read ( sinceTransactionNumber ? , transactionTimeHash ? ) { let result ; if ( sinceTransactionNumber && transactionTimeHash ) { result = await this . executeWithRetry ( `SELECT * FROM _ql_committed_${ this . transactionTable } BY doc_id WHERE doc_id = '${ transactionTimeHash }' AND blockAddress.sequenceNo >= ${ sinceTransactionNumber }` ) ; } else if ( sinceTransactionNumber ) { console . warn ( 'reading since transactionNumber is a costly operation (full table scan), use with caution' ) ; result = await this . executeWithRetry ( `SELECT * FROM _ql_committed_${ this . transactionTable } WHERE blockAddress.sequenceNo >= ${ sinceTransactionNumber }` ) ; } else if ( transactionTimeHash ) { console . warn ( 'reading all transactions is a costly operation (full table scan), use with caution' ) ; result = await this . executeWithRetry ( `SELECT * FROM _ql_committed_${ this . transactionTable } BY doc_id WHERE doc_id = '${ transactionTimeHash }'` ) ; } else { result = await this . executeWithRetry ( `SELECT * FROM _ql_committed_${ this . transactionTable }` ) ; } const resultList = ( result as Result ) . getResultList ( ) ; const transactions = ( resultList as ValueWithMetaData [ ] ) . map ( this . toSidetreeTransaction ) as TransactionModelQLDB [ ] ; transactions . sort ( ( t1 , t2 ) => { return t1 . transactionNumber - t2 . transactionNumber ; } ) ; return { moreTransactions : false , transactions , } ; } public async getFirstValidTransaction ( _transactions ) { return Promise . resolve ( undefined ) ; } public approximateTime : BlockchainTimeModel = { time : 0 , hash : '' , } ; public async getLatestTime ( ) { console . warn ( 'getLatestTime is a costly operation (full table scan), use with caution' ) ; const currentDate = moment ( ) . format ( ) ; const result = await this . executeWithRetry ( `SELECT blockAddress, id FROM history(${ this . transactionTable }, \`${ currentDate }\`) AS h BY id` ) ; const resultList = ( result as Result ) . getResultList ( ) ; if ( resultList . length > 0 ) { const sequenceNumbers = resultList . map ( ( result ) => result . blockAddress . sequenceNo . toString ( ) ) . map ( ( sequenceNo ) => Number ( sequenceNo ) ) ; const time = Math . max ( ... sequenceNumbers ) ; const latestBlock = resultList . find ( ( result ) => Number ( result . blockAddress . sequenceNo ) === time ) ; const hash = latestBlock . id . toString ( ) ; this . approximateTime = { time , hash , } ; } return this . approximateTime ; } getFee ( _transactionTime ) { return Promise . resolve ( 0 ) ; } getValueTimeLock ( _lockIdentifier ) { return Promise . resolve ( undefined ) ; } getWriterValueTimeLock ( ) { return Promise . resolve ( undefined ) ; } }