import { Injectable , Logger } from "@nestjs/common" ; import { InjectRepository } from "@nestjs/typeorm" ; import S3 from "aws-sdk/clients/s3" ; import { spawn } from "child_process" ; import { cpus } from "os" ; import { Topology } from "topojson-specification" ; import { Repository } from "typeorm" ; import { deserialize } from "v8" ; import { TypedArrays , IStaticFile , IStaticMetadata , S3URI } from "../../../../shared/entities" ; import { RegionConfig } from "../../region-configs/entities/region-config.entity" ; import { GeoUnitTopology } from "../entities/geo-unit-topology.entity" ; import { GeoUnitProperties } from "../entities/geo-unit-properties.entity" ; import _ from "lodash" ; import { getObject , s3Options } from "../../common/s3-wrapper" ; const MAX_RETRIES = 5 ; const BATCH_SIZE = cpus ( ) . length * 2 ; const STATE_ORDER = [ "TX" , "CA" , "PA" , "FL" , "NC" , "MO" , "NY" , "IL" , "TN" , "VA" ] ; type Layer = Promise < GeoUnitTopology | GeoUnitProperties | undefined > ; interface Layers { [ s3URI ] : null | Layer ; } async function asyncLoop < T > ( asyncFns , concurrent = 5 ) { const queue = [ ] ; for ( const fn of asyncFns ) { const p = fn ( ) . then ( res => { queue . splice ( queue . indexOf ( p ) , 1 ) ; return res ; } ) ; queue . push ( p ) ; if ( queue . length >= concurrent ) { await Promise . race ( queue ) ; } } await Promise . all ( queue ) ; } @ Injectable ( ) export class TopologyService { private _layers ? : Layers = undefined ; private readonly logger = new Logger ( TopologyService . name ) ; private readonly s3 = new S3 ( ) ; constructor ( @ InjectRepository ( RegionConfig ) repo ) { void repo . find ( ) . then ( regionConfigs => { const getLayers = async ( ) => { this . _layers = regionConfigs . reduce ( ( layers , regionConfig ) => ( { ... layers , [ regionConfig . s3URI ] : null } ) , { } ) ; const sortedRegions = _ . sortBy ( regionConfigs , region => [ ! STATE_ORDER . includes ( region . regionCode ) , STATE_ORDER . indexOf ( region . regionCode ) ] ) ; await asyncLoop ( sortedRegions . map ( region => ( ) => { const promise = this . fetchLayer ( region . s3URI , region . archived ) ; this . _layers = { ... this . _layers , [ region . s3URI ] : promise } ; return promise ; } ) , BATCH_SIZE ) ; } ; void getLayers ( ) ; } ) ; } public layers ( ) { return this . _layers && Object . freeze ( { ... this . _layers } ) ; } public async get ( s3URI ) { if ( ! this . _layers ) { return ; } if ( ! ( s3URI in this . _layers ) ) { this . _layers [ s3URI ] = this . fetchLayer ( s3URI ) . catch ( err => { this . logger . error ( err ) ; } ) ; } const layer = this . _layers [ s3URI ] ; if ( layer !== null ) { return layer ; } } private async fetchLayer ( s3URI , archived , numRetries = 0 ) { try { const [ topojsonResponse , staticMetadataResponse ] = await Promise . all ( [ getObject ( this . s3 , s3Options ( s3URI , "topo.buf" ) ) , getObject ( this . s3 , s3Options ( s3URI , "static-metadata.json" ) ) ] ) ; const staticMetadataBody = staticMetadataResponse . Body ?. toString ( "utf8" ) ; const topojsonBody = topojsonResponse . Body as Buffer ; if ( staticMetadataBody && topojsonBody ) { const staticMetadata = JSON . parse ( staticMetadataBody ) as IStaticMetadata ; const geoLevelHierarchy = staticMetadata . geoLevelHierarchy . map ( gl => gl . id ) ; if ( ! geoLevelHierarchy ) { this . logger . error ( `geoLevelHierarchy missing from static metadata for ${ s3URI }` ) ; } const topology = deserialize ( topojsonBody ) as Topology ; const [ demographics , geoLevels , voting ] = await Promise . all ( [ this . fetchStaticFiles ( s3URI , staticMetadata . demographics ) , this . fetchStaticFiles ( s3URI , staticMetadata . geoLevels ) , this . fetchStaticFiles ( s3URI , staticMetadata . voting || [ ] ) ] ) ; this . logger . debug ( `downloaded data for s3URI ${ s3URI }` ) ; const geoUnitTopology = new GeoUnitTopology ( topology , { groups : geoLevelHierarchy . slice ( ) . reverse ( ) } , staticMetadata , demographics , voting , geoLevels ) ; return archived ? GeoUnitProperties . fromTopology ( geoUnitTopology ) : geoUnitTopology ; } else { this . logger . error ( "Invalid TopoJSON or metadata bodies" ) ; } } catch ( err ) { this . logger . error ( `Failed to load topology for '${ s3URI }' ${ numRetries + 1 } times, err ${ err }` ) ; if ( numRetries < MAX_RETRIES ) { return this . fetchLayer ( s3URI , archived , numRetries + 1 ) ; } else { spawn ( "pkill" , [ "node" ] ) . once ( "exit" , ( ) => { process . exit ( 1 ) ; } ) ; } } } private async fetchStaticFiles ( path , files ) { const requests = files . map ( fileMeta => getObject ( this . s3 , s3Options ( path , fileMeta . fileName ) ) ) ; return new Promise ( ( resolve , reject ) => { Promise . all ( requests ) . then ( response => resolve ( response . map ( ( res , ind ) => { const bpe = files [ ind ] . bytesPerElement ; const unsigned = files [ ind ] . unsigned ; const typedArrayConstructor = unsigned || unsigned === undefined ? bpe === 1 ? Uint8Array : bpe === 2 ? Uint16Array : Uint32Array : bpe === 1 ? Int8Array : bpe === 2 ? Int16Array : Int32Array ; const buf = res . Body as Buffer ; const typedArray = new typedArrayConstructor ( buf . buffer ) ; return typedArray ; } ) ) ) . catch ( error => reject ( error . message ) ) ; } ) ; } }