import Debug from "debug" ; import Bluebird from "bluebird" ; import { strToInt } from "./../../utils" ; import { Yildiz } from "../Yildiz" ; import { Metrics } from "../metrics/Metrics" ; import { LookupCache } from "../cache/LookupCache" ; import { UpsertConfig } from "../../interfaces/ServiceConfig" ; import { FetchJob } from "../cache/FetchJob" ; import { NodeHandler } from "./NodeHandler" ; import { GenericObject , AnyObject } from "../../interfaces/Generic" ; import { YildizSingleSchema , YildizResultSchema , YildizUpsertSchema , } from "../../interfaces/Yildiz" ; import { EdgeRaw , NodeEdgeUpsert } from "../../interfaces/Graph" ; const debug = Debug ( "yildiz:graphaccess" ) ; export class GraphAccess { private yildiz : Yildiz ; private metrics : Metrics ; private lookupCache : LookupCache ; private upsertConfig : UpsertConfig | { } ; private fetchJob : FetchJob ; private nodeHandler ! : NodeHandler ; private promiseConcurrency : number ; constructor ( yildiz ) { this . yildiz = yildiz ; this . metrics = this . yildiz . metrics ; this . lookupCache = this . yildiz . lookupCache ; this . fetchJob = this . yildiz . fetchJob ; this . upsertConfig = this . yildiz . config . upsert || { } ; this . promiseConcurrency = this . yildiz . config . promiseConcurrency || 1000 ; } private async fullDataResolver ( identifier , noMemoryCache ? ) { const node = await this . nodeHandler . getNodeByIdentifier ( strToInt ( identifier ) + "" , noMemoryCache ) ; if ( ! node ) { try { await Bluebird . all ( [ this . yildiz . redisClient . clearCacheRefresh ( identifier ) , this . yildiz . redisClient . clearLastAccess ( identifier ) , ] ) ; } catch ( error ) { debug ( error ) ; } return ; } const edges = [ ] ; const rawOtherNodeIdentifiers = [ ] ; const nodeKeys = Object . keys ( node ) . filter ( ( nodeKey ) => nodeKey . includes ( "ec" ) || nodeKey . includes ( "ed" ) ) ; if ( ! nodeKeys . length ) { if ( node . data ) { if ( typeof node . data === "object" ) { if ( Object . keys ( node . data ) . length ) { node . data = node . data ; } else { delete node . data ; } } else { node . data = node . data ; } } else { delete node . data ; } delete node . ttld ; return { identifier : node . identifier , value : node . value , nodes : node } ; } for ( const nodeKey of nodeKeys ) { const rightNodeId = nodeKey . split ( "#" ) [ 1 ] ; const relation = nodeKey . split ( "#" ) [ 2 ] ; const edgeObj = { leftNodeId : node . identifier , rightNodeId , data : node [ nodeKey ] , } ; if ( ( this . upsertConfig as UpsertConfig ) . getEdgeTime && node . identifier ) { edgeObj . edgeTime = await this . nodeHandler . getEdgeTime ( node . identifier , rightNodeId , relation ) ; } if ( relation ) { edgeObj . relation = relation ; } edges . push ( edgeObj ) ; rawOtherNodeIdentifiers . push ( nodeKey . split ( "#" ) [ 1 ] ) ; } const otherNodeIdentifiers = rawOtherNodeIdentifiers . filter ( ( item , pos ) => rawOtherNodeIdentifiers . indexOf ( item ) === pos ) ; const otherNodes = await this . getRightNodes ( otherNodeIdentifiers , noMemoryCache ) ; const currentNode = { identifier : node . identifier , value : node . value , } ; if ( node . data ) { if ( typeof node . data === "object" ) { if ( Object . keys ( node . data ) . length ) { currentNode . data = node . data ; } } else { currentNode . data = node . data ; } } const result = { identifier : node . identifier , value : node . value , edges , nodes : [ currentNode ] . concat ( otherNodes ) , } ; await this . nodeHandler . createCache ( result ) ; return result ; } private async resolveOrBump ( identifier ) { const noMemoryCache = true ; const bumped = await this . nodeHandler . bumpCacheByIdentifier ( identifier ) ; if ( ! bumped ) { return this . fullDataResolver ( identifier + "" , noMemoryCache ) ; } } public async init ( ) { this . nodeHandler = await this . yildiz . getNodeHandler ( ) ; } public setLastAccessFireAndForget ( values ) { const nodeIdentifiers = values . map ( ( value ) => strToInt ( value ) + "" ) ; this . fetchJob . bumpTTL ( nodeIdentifiers ) . catch ( ( error ) => { } ) ; } public async getRightNodes ( identifiers , noMemoryCache ? ) { const start = Date . now ( ) ; const { cache , nocache } = await this . lookupCache . classifyRightNode ( identifiers ) ; this . metrics . inc ( "resolvedRightNode_cache_hits" , cache . length ) ; this . metrics . inc ( "resolvedRightNode_cache_miss" , nocache . length ) ; if ( ! nocache . length ) { return cache ; } const nocacheNodes = ( await Bluebird . map ( nocache , ( identifier ) => this . nodeHandler . getNodeByIdentifier ( identifier , noMemoryCache ) , { concurrency : this . promiseConcurrency , } ) ) . filter ( ( node ) => ! ! node ) . map ( ( node ) => { if ( ! node ) { return ; } const nodeObj = { identifier : node . identifier , value : node . value , } ; if ( node . data ) { if ( typeof node . data === "object" ) { if ( Object . keys ( node . data ) . length ) { nodeObj . data = node . data ; } } else { nodeObj . data = node . data ; } } return nodeObj ; } ) ; await this . lookupCache . setRightNode ( nocacheNodes ) ; this . metrics . inc ( "fetch_rightNodes_count" ) ; this . metrics . inc ( "fetch_rightNodes_time" , Date . now ( ) - start ) ; return nocacheNodes . concat ( cache ) ; } public async buildAndCacheNodes ( identifiers ) { ( await Bluebird . map ( identifiers , ( identifier ) => this . resolveOrBump ( identifier + "" ) , { concurrency : this . promiseConcurrency , } ) ) ; } public async getBuiltNodes ( identifiers ) { return ( await Bluebird . map ( identifiers , ( identifier ) => this . fullDataResolver ( identifier + "" ) , { concurrency : this . promiseConcurrency , } ) ) . filter ( ( result ) => ! ! result ) as YildizSingleSchema [ ] ; } private buildResult ( resultArray ? ) { if ( ! resultArray || ! resultArray . length ) { return ; } const filteredResultArray = resultArray . filter ( ( singleResult ) => ! ! singleResult ) ; if ( ! filteredResultArray || ! filteredResultArray . length ) { return ; } const result = { } ; result . identifiers = filteredResultArray . map ( ( singleResult ) => { return { identifier : singleResult . identifier , value : singleResult . value , } ; } ) ; const seenNodes = { } ; result . nodes = [ ] . concat ( ... filteredResultArray . map ( ( flattenResults ) => flattenResults . nodes ) ) . filter ( ( singleResult ) => { if ( ! singleResult || ! singleResult . identifier ) { return ; } const exists = seenNodes [ singleResult . identifier ] ; if ( ! exists ) { seenNodes [ singleResult . identifier ] = true ; } return ! exists ; } ) ; const seenEdges = { } ; result . edges = [ ] . concat ( ... filteredResultArray . map ( ( flattenResults ) => flattenResults . edges ) ) . filter ( ( singleResult ) => { if ( ! singleResult ) { return ; } const key = `${ singleResult . leftNodeId }_${ singleResult . rightNodeId }_${ singleResult . relation }` ; const exists = seenEdges [ key ] ; if ( ! exists ) { seenEdges [ key ] = true ; } return ! exists ; } ) ; return result ; } public async bumpCacheIfExists ( values = [ ] ) { if ( ! Array . isArray ( values ) ) { throw new Error ( "values must be an array" ) ; } if ( ! values . length ) { throw new Error ( "values should not be empty" ) ; } const nodeIdentifiers = values . map ( ( value ) => strToInt ( value ) + "" ) ; const { cache , nocache } = await this . lookupCache . classifyExistence ( nodeIdentifiers ) ; this . metrics . inc ( "resolvedNodes_cache_hits" , cache . length ) ; this . metrics . inc ( "resolvedNodes_cache_miss" , nocache . length ) ; if ( ! cache . length ) { return ; } return await Bluebird . map ( cache , ( identifier ) => this . nodeHandler . bumpCacheByIdentifier ( identifier ) , { concurrency : this . promiseConcurrency , } ) . catch ( ( error ) => { } ) ; } public async edgeInfoForNodesRelatingToTranslateValues ( values = [ ] ) { if ( ! Array . isArray ( values ) ) { throw new Error ( "values must be an array" ) ; } if ( ! values . length ) { throw new Error ( "values should not be empty" ) ; } let cacheResults = [ ] ; const start = Date . now ( ) ; const nodeIdentifiers = values . map ( ( value ) => strToInt ( value ) + "" ) ; const { cache , nocache } = await this . lookupCache . classifyExistence ( nodeIdentifiers ) ; this . metrics . inc ( "resolvedNodes_cache_hits" , cache . length ) ; this . metrics . inc ( "resolvedNodes_cache_miss" , nocache . length ) ; if ( cache . length ) { cacheResults = await Bluebird . map ( cache , ( identifier ) => this . nodeHandler . getCacheByIdentifier ( identifier ) , { concurrency : this . promiseConcurrency , } ) ; const errorKeys = [ ] ; cacheResults . map ( ( cacheResult , index ) => { if ( ! cacheResult ) { errorKeys . push ( cache [ index ] ) ; nocache . push ( cache [ index ] ) ; } } ) ; if ( errorKeys . length ) { debug ( `Cache in Bigtable does not exist for keys ${ errorKeys . join ( "," ) }` ) ; } } if ( ! nocache . length ) { const resultCache = this . buildResult ( cacheResults ) ; this . metrics . set ( "translated_edge_info_duration" , Date . now ( ) - start ) ; return resultCache ; } const resultArray = await this . getBuiltNodes ( nocache ) ; const result = this . buildResult ( resultArray . concat ( cacheResults ) ) ; if ( ( this . upsertConfig as UpsertConfig ) . translatedEdgeDebug ) { debug ( `translated edge info took ${ Date . now ( ) - start } ms` ) ; } this . metrics . set ( "translated_edge_info_duration" , Date . now ( ) - start ) ; return result ; } public async upsertSingleEdgeRelationBetweenNodes ( body ) { const start = Date . now ( ) ; if ( ! body || typeof body !== "object" ) { return Promise . reject ( new Error ( "Body must be an object." ) ) ; } const { leftNodeIdentifierVal , rightNodeIdentifierVal , leftNodeData , rightNodeData , ttld , relation , edgeData , depthBeforeCreation , isPopularRightNode , edgeTime , } = body ; const presumedNodeLeft = strToInt ( leftNodeIdentifierVal ) + "" ; const presumedNodeRight = strToInt ( rightNodeIdentifierVal ) + "" ; const [ nodeLeftExists , nodeRightExists , ] = await Bluebird . all ( [ this . nodeHandler . doesNodeExist ( presumedNodeLeft ) , this . nodeHandler . doesNodeExist ( presumedNodeRight ) , ] ) ; await Bluebird . all ( [ nodeLeftExists ? Promise . resolve ( ) : this . nodeHandler . createNode ( presumedNodeLeft , leftNodeData , { } , ttld , leftNodeIdentifierVal ) , nodeRightExists ? Promise . resolve ( ) : this . nodeHandler . createNode ( presumedNodeRight , rightNodeData , { } , ttld , rightNodeIdentifierVal ) , ] ) ; let edge = null ; if ( ! depthBeforeCreation ) { const edgeId = await this . nodeHandler . createEdgeWithId ( presumedNodeLeft , presumedNodeRight , relation , edgeData , { } , ttld , depthBeforeCreation , isPopularRightNode , edgeTime ) ; edge = { edgeId , leftNodeId : presumedNodeLeft , rightNodeId : presumedNodeRight , } ; } else { const edges = await this . nodeHandler . edgeExistsId ( presumedNodeLeft , presumedNodeRight ) ; let edgeId = null ; if ( ! edges ) { edgeId = await this . nodeHandler . createEdgeWithId ( presumedNodeLeft , presumedNodeRight , relation , edgeData , { } , ttld , depthBeforeCreation , isPopularRightNode , edgeTime ) ; } else { edgeId = ( await this . nodeHandler . increaseEdgeDepthById ( presumedNodeLeft , presumedNodeRight , isPopularRightNode , edgeTime ) ) . edges ; } edge = { edgeId , leftNodeId : presumedNodeLeft , rightNodeId : presumedNodeRight , } ; } const diff = ( new Date ( ) ) . getTime ( ) - start ; const withOrWithout = depthBeforeCreation ? "with" : "without" ; this . yildiz . metrics . set ( `upsert_${ withOrWithout }_depthBeforeCreation_duration` , diff ) ; if ( diff > 1000 ) { debug ( `edge upsert took too long, ${ diff } ms` , body ) ; } else { if ( ( this . upsertConfig as UpsertConfig ) . upsertDebug ) { debug ( `edge upsert took ${ diff } ms` ) ; } } return edge ; } public async runUpsertRelationWithRetry ( body , retry ? ) { try { const result = await this . upsertSingleEdgeRelationBetweenNodes ( body ) ; return result ; } catch ( error ) { if ( error . message . toLowerCase ( ) . includes ( "redis" ) ) { this . yildiz . incStat ( "redis_error" ) ; } if ( ! retry ) { retry = 0 ; } if ( retry === 2 ) { throw error ; } retry ++ ; return this . runUpsertRelationWithRetry ( body , retry ) ; } } }