import Blockweave from 'blockweave' ; import Transaction from 'blockweave/dist/lib/transaction' ; import { b64UrlToBuffer , bufferTob64Url } from 'blockweave/dist/utils/buffer' ; import { pipeline } from 'stream/promises' ; import chunker from 'stream-chunker' ; import { backOff } from 'exponential-backoff' ; import Merkle , { MAX_CHUNK_SIZE } from 'blockweave/dist/utils/merkle' ; const FATAL_CHUNK_UPLOAD_ERRORS = [ 'invalid_json' , 'chunk_too_big' , 'data_path_too_big' , 'offset_too_big' , 'data_size_too_big' , 'chunk_proof_ratio_not_attractive' , 'invalid_proof' , ] ; interface ChunkUploadPayload { data_root : string ; data_size : string ; data_path : string ; offset : string ; chunk : string ; } const MAX_CONCURRENT_CHUNK_UPLOAD_COUNT = 128 ; const merkle = new Merkle ( ) ; param createTx export function uploadTransactionAsync ( tx , blockweave , createTx = true ) { return async ( source ) : Promise < void > => { if ( ! tx . chunks ) { throw Error ( 'Transaction has no computed chunks!' ) ; } if ( createTx ) { tx . data = new Uint8Array ( 0 ) ; const createTxRes = await blockweave . api . post ( `tx` , tx ) ; if ( ! ( createTxRes . status >= 200 && createTxRes . status < 300 ) ) { throw new Error ( `Failed to create transaction: ${ createTxRes . data }` ) ; } } const txChunkData = tx . chunks ; const { chunks , proofs } = txChunkData ; function prepareChunkUploadPayload ( chunkIndex , chunkData ) { const proof = proofs [ chunkIndex ] ; return { data_root : tx . data_root , data_size : tx . data_size , data_path : bufferTob64Url ( proof . proof ) , offset : proof . offset . toString ( ) , chunk : bufferTob64Url ( chunkData ) , } ; } await pipeline ( source , chunker ( MAX_CHUNK_SIZE , { flush : true } ) , async function ( chunkedSource ) { let chunkIndex = 0 ; let dataRebalancedIntoFinalChunk ; const activeChunkUploads = [ ] ; for await ( const chunkData of chunkedSource ) { const currentChunk = chunks [ chunkIndex ] ; const chunkSize = currentChunk . maxByteRange - currentChunk . minByteRange ; const expectedToBeFinalRebalancedChunk = dataRebalancedIntoFinalChunk != null ; let chunkPayload ; if ( chunkData . byteLength === chunkSize ) { chunkPayload = prepareChunkUploadPayload ( chunkIndex , chunkData ) ; } else if ( chunkData . byteLength > chunkSize ) { chunkPayload = prepareChunkUploadPayload ( chunkIndex , chunkData . slice ( 0 , chunkSize ) ) ; dataRebalancedIntoFinalChunk = chunkData . slice ( chunkSize ) ; } else if ( chunkData . byteLength < chunkSize && expectedToBeFinalRebalancedChunk ) { chunkPayload = prepareChunkUploadPayload ( chunkIndex , Buffer . concat ( [ dataRebalancedIntoFinalChunk ! , chunkData ] , dataRebalancedIntoFinalChunk ! . length + chunkData . length ) ) ; } else { throw Error ( 'Transaction data stream terminated incorrectly.' ) ; } const chunkValid = await merkle . validatePath ( txChunkData . data_root , parseInt ( chunkPayload . offset , 10 ) , 0 , parseInt ( chunkPayload . data_size , 10 ) , b64UrlToBuffer ( chunkPayload . data_path ) ) ; if ( ! chunkValid ) { throw new Error ( `Unable to validate chunk ${ chunkIndex }.` ) ; } if ( activeChunkUploads . length >= MAX_CONCURRENT_CHUNK_UPLOAD_COUNT ) { await Promise . all ( activeChunkUploads ) ; activeChunkUploads . length = 0 ; } activeChunkUploads . push ( backOff ( ( ) => blockweave . api . post ( 'chunk' , chunkPayload ) , { retry : ( err ) => ! FATAL_CHUNK_UPLOAD_ERRORS . includes ( err . message ) , } ) ) ; chunkIndex ++ ; } await Promise . all ( activeChunkUploads ) ; if ( chunkIndex < chunks . length ) { throw Error ( 'Transaction upload incomplete.' ) ; } } ) ; } ; }