import { inject , injectable } from 'inversify' ; import { cosmosContainerClientTypes , CosmosContainerClient , client , CosmosOperationResponse } from 'azure-services' ; import { System , HashGenerator , RetryHelper , ServiceConfiguration } from 'common' ; import { WebsiteScanResult , ItemType , WebsiteScanResultPart , WebsiteScanResultBase , WebsiteScanResultPartModel , websiteScanResultBaseKeys , websiteScanResultPartModelKeys , websiteScanResultBaseTransientKeys , } from 'storage-documents' ; import { GlobalLogger } from 'logger' ; import _ from 'lodash' ; import pLimit from 'p-limit' ; import moment from 'moment' ; import { PartitionKeyFactory } from '../factories/partition-key-factory' ; import { WebsiteScanResultAggregator } from './website-scan-result-aggregator' ; interface DbDocument { baseDocument : Partial < WebsiteScanResultBase > ; partDocument : Partial < WebsiteScanResultPart > ; } @ injectable ( ) export class WebsiteScanResultProvider { private readonly maxRetryCount : number = 5 ; private readonly msecBetweenRetries : number = 1000 ; public maxConcurrencyLimit = 5 ; constructor ( @ inject ( cosmosContainerClientTypes . OnDemandScanRunsCosmosContainerClient ) private readonly cosmosContainerClient , @ inject ( WebsiteScanResultAggregator ) private readonly websiteScanResultAggregator , @ inject ( ServiceConfiguration ) protected readonly serviceConfig , @ inject ( PartitionKeyFactory ) private readonly partitionKeyFactory , @ inject ( HashGenerator ) private readonly hashGenerator , @ inject ( GlobalLogger ) private readonly logger , @ inject ( RetryHelper ) private readonly retryHelper = new RetryHelper < WebsiteScanResult > ( ) ) { } public async read ( websiteScanId , readCompleteDocument = false ) { const baseDocument = ( await this . cosmosContainerClient . readDocument < WebsiteScanResultBase > ( websiteScanId , this . getPartitionKey ( websiteScanId ) ) ) . item ; const partDocument = readCompleteDocument ? await this . readPartDocument ( baseDocument ) : { } ; const partDocumentModel = _ . pick ( partDocument , websiteScanResultPartModelKeys ) as Partial < WebsiteScanResultPartModel > ; return { ... baseDocument , ... partDocumentModel } ; } /**
     * Writes documents to a storage if documents do not exist; otherwise, merges documents with corresponding storage documents.
     *
     * Source document properties that resolve to undefined are skipped if a destination document value exists.
     * Will remove all falsey (false, null, 0, "", undefined, and NaN) values from document's array type properties
     */ public async mergeOrCreateBatch ( websiteScanResults ) { const limit = pLimit ( this . maxConcurrencyLimit ) ; await Promise . all ( websiteScanResults . map ( ( result ) => { return limit ( async ( ) => this . mergeOrCreate ( result . scanId , result . websiteScanResult ) ) ; } ) ) ; } /**
     * Writes document to a storage if document does not exist; otherwise, merges the document with the current storage document.
     *
     * Source document properties that resolve to undefined are skipped if a destination document value exists.
     * Will remove all falsey (false, null, 0, "", undefined, and NaN) values from document's array type properties
     */ public async mergeOrCreate ( scanId , websiteScanResult , readCompleteDocument = false ) { const dbDocument = this . convertToDbDocument ( scanId , websiteScanResult ) ; return ( await this . retryHelper . executeWithRetries ( async ( ) => { if ( readCompleteDocument ) { const baseDocument = await this . mergeOrCreateImpl ( dbDocument ) ; return this . read ( baseDocument . id , true ) ; } else { return this . mergeOrCreateImpl ( dbDocument ) ; } } , async ( err ) => this . logger . logError ( `Failed to update website scan result Cosmos DB document. Retrying on error.` , { baseId : dbDocument ?. baseDocument . id , partId : dbDocument ?. partDocument . id , baseDocument : JSON . stringify ( dbDocument . baseDocument ) , partDocument : JSON . stringify ( dbDocument . partDocument ) , error : System . serializeError ( err ) , } ) , this . maxRetryCount , this . msecBetweenRetries ) ) as WebsiteScanResultBase ; } /**
     *
     * Sets the required storage document properties.
     */ public normalizeToDbDocument ( websiteScanResult ) { const documentId = this . getWebsiteScanId ( websiteScanResult ) ; const partitionKey = websiteScanResult . partitionKey ?? this . getPartitionKey ( documentId ) ; return { ... ( websiteScanResult as WebsiteScanResult ) , id : documentId , partitionKey : partitionKey , itemType : ItemType . websiteScanResult , } ; } private async mergeOrCreateImpl ( dbDocument ) { const baseDocument = await this . mergeAndWriteBaseDocument ( dbDocument ) ; await this . mergeAndWritePartDocument ( dbDocument . partDocument ) ; return baseDocument ; } private async mergeAndWriteBaseDocument ( dbDocument ) { const operationResult = await this . createBaseDocumentIfNotExists ( dbDocument ) ; if ( operationResult . created ) { return operationResult . scanResult ; } const storageDocument = operationResult . scanResult ; const originalDocument = _ . cloneDeep ( storageDocument ) ; const mergedDocument = this . websiteScanResultAggregator . mergeBaseDocument ( dbDocument . baseDocument , storageDocument ) ; if ( ! this . same ( originalDocument , mergedDocument ) ) { return ( await this . cosmosContainerClient . writeDocument ( mergedDocument as WebsiteScanResultBase ) ) . item ; } else { return storageDocument ; } } private async mergeAndWritePartDocument ( partDocument ) { const operationResponse = await this . cosmosContainerClient . readDocument < WebsiteScanResultPart > ( partDocument . id , partDocument . partitionKey , false ) ; const mergedDocument = await this . websiteScanResultAggregator . mergePartDocument ( partDocument , operationResponse . item ?? { } ) ; await this . cosmosContainerClient . writeDocument ( mergedDocument ) ; } private async readPartDocument ( websiteScanResult ) { const query = { query : 'SELECT * FROM c WHERE c.partitionKey = @partitionKey and c.baseId = @baseId and c.itemType = @itemType' , parameters : [ { name : '@baseId' , value : websiteScanResult . id , } , { name : '@partitionKey' , value : websiteScanResult . partitionKey , } , { name : '@itemType' , value : ItemType . websiteScanResultPart , } , ] , } ; let documentCount = 0 ; let elapsedToAggregate = 0 ; let partDocument = { } ; let continuationToken ; do { const response = ( await this . cosmosContainerClient . queryDocuments < WebsiteScanResultPart > ( query , continuationToken ) ) as CosmosOperationResponse < WebsiteScanResultPart [ ] > ; client . ensureSuccessStatusCode ( response ) ; continuationToken = response . continuationToken ; documentCount = documentCount + response . item . length ; const start = process . hrtime ( ) ; partDocument = await this . websiteScanResultAggregator . mergePartDocuments ( response . item , partDocument ) ; const elapsed = process . hrtime ( start ) ; elapsedToAggregate = elapsedToAggregate + elapsed [ 0 ] * 1000 + elapsed [ 1 ] / 1000000 ; } while ( continuationToken !== undefined ) ; this . logger . logInfo ( `Elapsed ${ moment . utc ( moment . duration ( elapsedToAggregate ) . asMilliseconds ( ) ) . format ( 'mm:ss.SSS' ) } while merged ${ documentCount } website scan result partials documents` , { elapsed : `${ elapsedToAggregate }` , documentCount : `${ documentCount }` } ) ; return partDocument ; } private async createBaseDocumentIfNotExists ( dbDocument ) { const operationResponse = await this . cosmosContainerClient . readDocument < WebsiteScanResultBase > ( dbDocument . baseDocument . id , dbDocument . baseDocument . partitionKey , false ) ; if ( operationResponse . item !== undefined ) { return { created : false , scanResult : operationResponse . item } ; } await this . setDeepScanLimit ( dbDocument ) ; const websiteScanResultDocument = this . websiteScanResultAggregator . mergeBaseDocument ( dbDocument . baseDocument , { } ) ; const scanResult = ( await this . cosmosContainerClient . writeDocument ( websiteScanResultDocument as WebsiteScanResultBase ) ) . item ; return { created : true , scanResult } ; } private getWebsiteScanId ( websiteScanResult ) { if ( websiteScanResult . id !== undefined ) { return websiteScanResult . id ; } if ( websiteScanResult . baseUrl === undefined && websiteScanResult . scanGroupId === undefined ) { throw new Error ( `WebsiteScanResult instance should have either id or baseUrl and scanGroupId properties defined. ${ JSON . stringify ( websiteScanResult ) }` ) ; } return this . hashGenerator . getWebsiteScanResultDocumentId ( websiteScanResult . baseUrl , websiteScanResult . scanGroupId ) ; } private getPartitionKey ( websiteScanResultId ) { return this . partitionKeyFactory . createPartitionKeyForDocument ( ItemType . websiteScanResult , websiteScanResultId ) ; } private convertToDbDocument ( scanId , websiteScanResult ) { const websiteScanResultNormalized = this . normalizeToDbDocument ( websiteScanResult ) ; const baseDocument = _ . pick ( websiteScanResultNormalized , websiteScanResultBaseKeys ) as Partial < WebsiteScanResultBase > ; const part = _ . pick ( websiteScanResultNormalized , websiteScanResultPartModelKeys ) as Partial < WebsiteScanResultPartModel > ; const partDocument = { id : this . hashGenerator . getWebsiteScanResultPartDocumentId ( websiteScanResultNormalized . id , scanId ) , partitionKey : websiteScanResultNormalized . partitionKey , itemType : ItemType . websiteScanResultPart , baseId : websiteScanResultNormalized . id , scanId , ... part , } ; return { baseDocument , partDocument } ; } private async setDeepScanLimit ( dbDocument ) { const partDocument = await this . websiteScanResultAggregator . mergePartDocument ( dbDocument . partDocument , { } ) ; const config = await this . serviceConfig . getConfigValue ( 'crawlConfig' ) ; if ( partDocument . knownPages ?. length >= config . deepScanDiscoveryLimit ) { dbDocument . baseDocument . deepScanLimit = partDocument . knownPages . length + 1 > config . deepScanUpperLimit ? config . deepScanUpperLimit : partDocument . knownPages . length + 1 ; } else { dbDocument . baseDocument . deepScanLimit = config . deepScanDiscoveryLimit ; } } private same ( storageDocument , mergedDocument ) { const transientKeys = new Set ( websiteScanResultBaseTransientKeys ) ; const keys = websiteScanResultBaseKeys . filter ( ( key ) => ! transientKeys . has ( key ) ) ; return JSON . stringify ( _ . pick ( storageDocument , keys ) ) === JSON . stringify ( _ . pick ( mergedDocument , keys ) ) ; } }