'use strict' ; import { IsReadableStream , ReadChunkFromStream } from '@smcloudstore/core/dist/StreamUtils' ; import { WaitPromise } from '@smcloudstore/core/dist/Utils' ; import { Readable , Stream } from 'stream' ; /**
 * Manages the upload of objects to Backblaze B2.
 *
 * This supports using Buffers and strings with the "simple APIs". It supports streams too, using either the "simple APIs" if the stream is less than `chunkSize`, or the large file APIs otherwise. The selection happens automatically.
 */ class B2Upload { /**
     * Size of each chunk that is uploaded when using B2's large file APIs, in bytes. Minimum value is 5MB; default is 9MB.
     *
     * Note: there seems to be a bug in the current version of the backblaze-b2 package when the request body upload is > 10 MB, because of a downstream dependency on axios@0.17; once backblaze-b2 updates its dependency on axios, this might be fixed.
     */ static chunkSize = 9 * 1024 * 1024 ; /** Backblaze recommends retrying all uploads at least two times (up to five) in case of errors, with an incrementing delay. We're retrying all uploads 3 times by default. */ static retries = 3 ; /** Instance of the B2 client library */ protected client : any ; /** Id of the target bucket */ protected bucketId : string ; /** Path where to store the object, inside the container */ protected path : string ; /** Data to upload */ protected data : Stream | string | Buffer ; /** Metadata for the object */ protected metadata : any ; /** Length (in bytes) of the input data */ protected length : number ; param client param bucketId param path param data param metadata param length constructor ( client , bucketId , path , data , metadata ? , length ? ) { this . client = client ; this . bucketId = bucketId ; this . path = path ; this . data = data ; this . metadata = metadata || { } ; this . length = length || 0 ; } returns async start ( ) { if ( typeof this . data == 'string' || ( typeof this . data == 'object' && Buffer . isBuffer ( this . data ) ) ) { if ( typeof this . data == 'string' ) { this . data = Buffer . from ( this . data as string , 'utf8' ) ; } if ( this . data . byteLength > 5 * 1024 * 1024 * 1024 ) { throw Error ( 'Maximum size for strings and Buffers is 5 GB' ) ; } return this . putFile ( ) ; } if ( ! IsReadableStream ( this . data ) ) { throw Error ( 'putObject requires a Stream, a Buffer or a string' ) ; } if ( B2Upload . chunkSize < 5 * 1024 * 1024 ) { throw Error ( 'chunkSize must be at least 5MB' ) ; } return ReadChunkFromStream ( this . data as Readable , B2Upload . chunkSize + 1 , true ) . then ( ( firstChunk ) => { if ( ! this . length ) { if ( ! firstChunk || ! firstChunk . byteLength ) { throw Error ( 'First chunk read from the stream has zero length' ) ; } this . length = firstChunk . byteLength ; } if ( this . length <= B2Upload . chunkSize ) { return this . putFile ( firstChunk ) ; } else { return this . putLargeFile ( ) ; } } ) ; } param data returns async private putFile ( data ? ) { if ( ! data ) { data = this . data as Buffer ; } if ( ! data || typeof data != 'object' || ! Buffer . isBuffer ( data ) ) { throw Error ( 'Argument data must be a Buffer' ) ; } let retryCounter = 0 ; const doUpload = ( ) => { return this . client . getUploadUrl ( this . bucketId ) . then ( ( response ) => { if ( ! response || ! response . data || ! response . data . authorizationToken || ! response . data . uploadUrl ) { throw Error ( 'Invalid response when requesting the upload url and upload authorization token' ) ; } const requestArgs = { data : data , filename : this . path , info : { } as any , mime : 'application/octet-stream' , uploadAuthToken : response . data . authorizationToken , uploadUrl : response . data . uploadUrl } ; if ( this . metadata ) { let i = 0 ; for ( const key in this . metadata ) { if ( ! this . metadata . hasOwnProperty ( key ) ) { continue ; } if ( key && key . toLowerCase ( ) == 'content-type' ) { requestArgs . mime = this . metadata [ key ] ; } else { if ( i == 10 ) { throw Error ( 'Cannot send more than 10 custom headers' ) ; } if ( ! key . match ( '^[A-Za-z0-9\-]+$' ) ) { throw Error ( 'Invalid header format: must be A-Za-z0-9' ) ; } if ( key . substr ( 0 , 10 ) != 'X-Bz-Info-' ) { requestArgs . info [ 'X-Bz-Info-' + key ] = this . metadata [ key ] ; } else { requestArgs . info [ key ] = this . metadata [ key ] ; } i ++ ; } } } return this . client . uploadFile ( requestArgs ) ; } ) . catch ( ( err ) => { if ( retryCounter < B2Upload . retries ) { retryCounter ++ ; return WaitPromise ( ( retryCounter + 1 ) * 500 ) . then ( ( ) => { return doUpload ( ) ; } ) ; } else { throw err ; } } ) ; } ; return doUpload ( ) ; } param stream returns async private putLargeFile ( stream ? ) { if ( ! stream ) { stream = this . data as Readable ; } if ( ! IsReadableStream ( stream ) ) { throw Error ( 'Argument stream must be a Readable Stream' ) ; } let streamEnded = false ; stream . on ( 'end' , ( ) => { streamEnded = true ; } ) ; let fileId = null ; const readChunk = ( partNumber , hashes ) : Promise < { fileId : string ; hashes : string [ ] ; } > => { if ( streamEnded ) { return Promise . resolve ( { fileId : fileId , hashes : hashes } ) ; } return ReadChunkFromStream ( stream , B2Upload . chunkSize ) . then ( ( data ) => { if ( ! data ) { return { fileId : fileId , hashes : hashes } ; } else { return this . putPart ( fileId , partNumber , data ) . then ( ( response ) => { if ( ! response || ! response . data ) { throw Error ( 'Invalid response when uploading a part' ) ; } if ( ! response . data . fileId || response . data . fileId != fileId ) { throw Error ( 'fileId for uploaded part does not match' ) ; } if ( response . data . partNumber === undefined || response . data . partNumber != partNumber ) { throw Error ( 'partNumber for uploaded part does not match' ) ; } if ( ! response . data . contentLength ) { throw Error ( 'Invalid contentLength of uploaded part' ) ; } if ( ! response . data . contentSha1 ) { throw Error ( 'Invalid contentSha1 of uploaded part' ) ; } hashes . push ( response . data . contentSha1 ) ; return readChunk ( partNumber + 1 , hashes ) ; } ) ; } } ) ; } ; return Promise . resolve ( ) . then ( ( ) => { let contentType = 'application/octet-stream' ; if ( this . metadata ) { for ( const key in this . metadata ) { if ( ! this . metadata . hasOwnProperty ( key ) ) { continue ; } const keyLowerCase = key . toLowerCase ( ) ; if ( keyLowerCase == 'content-type' ) { contentType = this . metadata [ key ] ; break ; } } } return this . client . startLargeFile ( { bucketId : this . bucketId , contentType : contentType , fileName : this . path } ) ; } ) . then ( ( response ) => { if ( ! response || ! response . data || ! response . data . fileId ) { throw Error ( 'Invalid response when requesting the file id' ) ; } fileId = response . data . fileId as string ; return readChunk ( 1 , [ ] ) ; } ) . then ( ( result ) => { return this . client . finishLargeFile ( { fileId : result . fileId , partSha1Array : result . hashes } ) ; } ) . catch ( ( err ) => { const cb = ( ) => { throw err ; } ; if ( fileId ) { return this . client . cancelLargeFile ( { fileId : fileId } ) . then ( cb , cb ) ; } else { cb ( ) ; } } ) ; } param fileId param partNumber param data returns async private putPart ( fileId , partNumber , data ) { let retryCounter = 0 ; const doUpload = ( ) => { return this . client . getUploadPartUrl ( { fileId : fileId } ) . then ( ( response ) => { if ( ! response || ! response . data || ! response . data . authorizationToken || ! response . data . uploadUrl ) { throw Error ( 'Invalid response when requesting the upload part url and upload authorization token' ) ; } return this . client . uploadPart ( { data : data , partNumber : partNumber , uploadAuthToken : response . data . authorizationToken , uploadUrl : response . data . uploadUrl } ) ; } ) . catch ( ( err ) => { if ( retryCounter < B2Upload . retries ) { retryCounter ++ ; return WaitPromise ( ( retryCounter + 1 ) * 500 ) . then ( ( ) => { return doUpload ( ) ; } ) ; } else { throw err ; } } ) ; } ; return doUpload ( ) ; } } export = B2Upload ;