import { injectable } from 'inversify' ; import { WebsiteScanResultBase , WebsiteScanResultPart } from 'storage-documents' ; import Parallel from 'paralleljs' ; import { System } from 'common' ; import _ from 'lodash' ; @ injectable ( ) export class WebsiteScanResultAggregator { public static parallelBlockSize = 6 ; public mergeBaseDocument ( sourceDocument , targetDocument ) { const propertiesToKeep = [ '_etag' , 'deepScanId' , 'deepScanLimit' ] ; const mergedDocument = _ . mergeWith ( targetDocument , sourceDocument , ( target , source , key ) => { if ( propertiesToKeep . includes ( key ) ) { return target ; } return this . mergeArray ( target , source , key , [ 'reports' , 'discoveryPatterns' ] ) ; } ) ; if ( mergedDocument . reports !== undefined ) { mergedDocument . reports = _ . uniqBy ( mergedDocument . reports , ( r ) => r . reportId ) ; } if ( mergedDocument . discoveryPatterns !== undefined ) { mergedDocument . discoveryPatterns = _ . uniq ( mergedDocument . discoveryPatterns ) ; } return mergedDocument ; } public async mergePartDocument ( sourceDocument , targetDocument ) { return this . mergePartDocuments ( [ sourceDocument ] , targetDocument ) ; } public async mergePartDocuments ( documents , baseDocument ? ) { if ( _ . isEmpty ( documents ) ) { return baseDocument ?? { } ; } if ( documents . length === 1 && _ . isEmpty ( baseDocument ) ) { return documents [ 0 ] ; } let mergeResult = [ ] ; mergeResult . push ( baseDocument ) ; mergeResult . push ( ... documents ) ; do { mergeResult = await this . mergePartDocumentsParallel ( mergeResult ) ; } while ( mergeResult . length > 1 ) ; return mergeResult [ 0 ] ; } private async mergePartDocumentsParallel ( documents ) { const partResults = await new Promise < Partial < WebsiteScanResultPart > [ ] [ ] > ( ( resolve , reject ) => { const parts = System . chunkArray ( documents , WebsiteScanResultAggregator . parallelBlockSize ) ; const parallel = new Parallel ( parts , { evalPath : `${ __dirname }/../eval.js` } ) ; parallel . map ( ( part ) => { const worker = require ( `${ __dirname }/parallel-workers.js` ) ; return worker . reducePartDocuments ( part ) ; } ) . then ( ( data ) => { resolve ( data ) ; } , ( error ) => { reject ( error ) ; } ) ; } ) ; return partResults . map ( ( p ) => p [ 0 ] ) ; } private mergeArray ( target , source , key , supportedKeys ) { if ( _ . isArray ( target ) || _ . isArray ( source ) ) { if ( ! supportedKeys . includes ( key ) ) { throw new Error ( `Merge of ${ key } array is not implemented.` ) ; } if ( target ) { return _ . compact ( target . concat ( source ) ) ; } else { return _ . compact ( source ) ; } } return undefined ; } }