import logger from '@mds-core/mds-logger' ; import flatten from 'flat' ; import { NotFoundError , nullKeys , stripNulls , now , isInsideBoundingBox , routeDistance } from '@mds-core/mds-utils' ; import { UUID , Timestamp , Device , VehicleEvent , Telemetry , BoundingBox , EVENT_STATUS_MAP , VEHICLE_STATUSES } from '@mds-core/mds-types' ; import redis from 'redis' ; import bluebird from 'bluebird' ; import { parseTelemetry , parseEvent , parseDevice , parseCachedItem } from './unflatteners' ; import { CacheReadDeviceResult , CachedItem , StringifiedCacheReadDeviceResult , StringifiedEventWithTelemetry , StringifiedTelemetry , StringifiedEvent } from './types' ; const { env } = process ; const { unflatten } = flatten ; declare module 'redis' { interface RedisClient { dbsizeAsync : ( ) => Promise < number > ; delAsync : ( ... arg1 : string [ ] ) => Promise < number > ; flushdbAsync : ( ) => Promise < 'OK' > ; hdelAsync : ( ... args : ( string | number ) [ ] ) => Promise < number > ; hgetallAsync : ( arg1 : string ) => Promise < { [ key ] : string ; } > ; hgetAsync : ( key : string , field : string ) => Promise < string > ; hscanAsync : ( key : string , cursor : number , condition : string , pattern : string ) => Promise < [ string , string [ ] ] > ; hsetAsync : ( key : string , field : string , value : string ) => Promise < number > ; hmsetAsync : ( ... args : unknown [ ] ) => Promise < 'OK' > ; infoAsync : ( ) => Promise < string > ; keysAsync : ( arg1 : string ) => Promise < string [ ] > ; zaddAsync : ( arg1 : string | number , arg2 : number , arg3 : string ) => Promise < number > ; zrangebyscoreAsync : ( key : string , min : number | string , max : number | string ) => Promise < string [ ] > ; georadiusAsync : ( key : string , longitude : number , latitude : number , radius : number , unit : string ) => Promise < UUID [ ] > ; } interface Multi { hgetallAsync : ( arg1 : string ) => Promise < { [ key ] : string ; } > ; execAsync : ( ) => Promise < object [ ] > ; } } bluebird . promisifyAll ( redis . RedisClient . prototype ) ; bluebird . promisifyAll ( redis . Multi . prototype ) ; let cachedClient ; function decorateKey ( key ) { return env . TENANT_ID ? `${ env . TENANT_ID }:${ key }` : key ; } async function getClient ( ) { if ( ! cachedClient ) { const port = Number ( env . REDIS_PORT ) || 6379 ; const host = env . REDIS_HOST || 'localhost' ; logger . info ( `connecting to redis on ${ host }:${ port }` ) ; cachedClient = redis . createClient ( { port , host , password : env . REDIS_PASS } ) ; logger . info ( 'redis client created' ) ; cachedClient . on ( 'connect' , ( ) => { logger . info ( 'redis cache connected' ) ; } ) ; cachedClient . on ( 'error' , async ( err ) => { logger . error ( `redis cache error ${ err }` ) ; } ) ; try { const size = await cachedClient . dbsizeAsync ( ) ; logger . info ( `redis cache has ${ size } keys` ) ; } catch ( err ) { logger . error ( 'redis failed to get dbsize' , err ) ; } } return cachedClient ; } async function info ( ) { const results = await ( await getClient ( ) ) . infoAsync ( ) ; const lines = results . split ( '\r\n' ) ; const data = { } ; lines . map ( line => { const [ key , val ] = line . split ( ':' ) ; if ( val !== undefined ) { if ( Number . isNaN ( Number ( val ) ) ) { data [ key ] = val ; } else { data [ key ] = parseFloat ( val ) ; } } } ) ; return data ; } async function updateVehicleList ( device_id , timestamp ? ) { const when = timestamp || now ( ) ; return ( await getClient ( ) ) . zaddAsync ( decorateKey ( 'device-ids' ) , when , device_id ) ; } async function hread ( suffix , device_id ) { if ( ! device_id ) { throw new Error ( `hread: tried to read ${ suffix } for device_id ${ device_id }` ) ; } const key = decorateKey ( `device:${ device_id }:${ suffix }` ) ; const flat = await ( await getClient ( ) ) . hgetallAsync ( key ) ; if ( flat ) { return unflatten ( { ... flat , device_id } ) ; } throw new NotFoundError ( `${ suffix } for ${ device_id } not found` ) ; } async function addGeospatialHash ( device_id , coordinates ) { const client = await getClient ( ) ; const [ lat , lng ] = coordinates ; const res = await client . geoadd ( decorateKey ( 'locations' ) , lng , lat , device_id ) ; return res ; } async function getEventsInBBox ( bbox ) { const start = now ( ) ; const client = await getClient ( ) ; const [ pt1 , pt2 ] = bbox ; const points = bbox . map ( pt => { return { lat : pt [ 0 ] , lng : pt [ 1 ] } ; } ) ; const [ lng , lat ] = [ ( pt1 [ 0 ] + pt2 [ 0 ] ) / 2 , ( pt1 [ 1 ] + pt2 [ 1 ] ) / 2 ] ; const radius = routeDistance ( points ) ; const events = client . georadiusAsync ( decorateKey ( 'locations' ) , lng , lat , radius , 'm' ) ; const finish = now ( ) ; const timeElapsed = finish - start ; logger . info ( `mds-agency-cache getEventsInBBox ${ JSON . stringify ( bbox ) } time elapsed: ${ timeElapsed }ms` ) ; return events ; } async function hreads ( suffixes , ids , prefix = 'device' ) { if ( suffixes === undefined ) { throw new Error ( 'hreads: no suffixes' ) ; } if ( ids === undefined ) { throw new Error ( 'hreads: no ids' ) ; } const multi = ( await getClient ( ) ) . multi ( ) ; await Promise . all ( suffixes . map ( suffix => ids . map ( id => { return multi . hgetallAsync ( decorateKey ( `${ prefix }:${ id }:${ suffix }` ) ) ; } ) ) ) ; const replies = await multi . execAsync ( ) ; return replies . map ( ( flat , index ) => { if ( flat ) { const flattened = { ... flat , [ `${ prefix }_id` ] : ids [ index % ids . length ] } ; return unflatten ( flattened ) ; } return unflatten ( null ) ; } ) ; } async function hwrite ( suffix , item ) { if ( typeof item . device_id !== 'string' ) { logger . error ( `hwrite: invalid device_id ${ item . device_id }` ) ; throw new Error ( `hwrite: invalid device_id ${ item . device_id }` ) ; } const { device_id } = item ; const key = `device:${ device_id }:${ suffix }` ; const flat = flatten ( item ) ; const nulls = nullKeys ( flat ) ; const hmap = stripNulls ( flat ) as { [ key ] : unknown ; device_id : UUID ; } ; delete hmap . device_id ; if ( nulls . length > 0 ) { await ( await getClient ( ) ) . hdelAsync ( decorateKey ( key ) , ... nulls ) ; } const client = await getClient ( ) ; await Promise . all ( ( suffix === 'event' ? [ `provider:${ item . provider_id }:latest_event` , key ] : [ key ] ) . map ( k => client . hmsetAsync ( k , hmap ) ) ) ; return updateVehicleList ( device_id ) ; } async function writeDevice ( device ) { if ( ! device ) { throw new Error ( 'null device not legal to write' ) ; } return hwrite ( 'device' , device ) ; } async function readKeys ( pattern ) { return ( await getClient ( ) ) . keysAsync ( decorateKey ( pattern ) ) ; } async function getMostRecentEventByProvider ( ) { const provider_ids = ( await readKeys ( 'provider:*:latest_event' ) ) . map ( key => { const [ , provider_id ] = key . split ( ':' ) ; return provider_id ; } ) ; const result = await hreads ( [ 'latest_event' ] , provider_ids , 'provider' ) ; return result . map ( elem => { const max = parseInt ( elem . timestamp || '0' ) ; return { provider_id : elem . provider_id , max } ; } ) ; } async function wipeDevice ( device_id ) { const keys = [ decorateKey ( `device:${ device_id }:event` ) , decorateKey ( `device:${ device_id }:telemetry` ) , decorateKey ( `device:${ device_id }:device` ) ] ; if ( keys . length > 0 ) { logger . info ( 'del' , ... keys ) ; return ( await getClient ( ) ) . delAsync ( ... keys ) ; } logger . info ( 'no keys found for' , device_id ) ; return 0 ; } async function writeEvent ( event ) { try { if ( event . event_type === 'deregister' ) { return await wipeDevice ( event . device_id ) ; } const prev_event = parseEvent ( ( await hread ( 'event' , event . device_id ) ) as StringifiedEventWithTelemetry ) ; if ( prev_event . timestamp < event . timestamp ) { try { if ( event . telemetry ) { const { lat , lng } = event . telemetry . gps ; await addGeospatialHash ( event . device_id , [ lat , lng ] ) ; } return hwrite ( 'event' , event ) ; } catch ( err ) { logger . error ( 'hwrites' , err . stack ) ; throw err ; } } else { return null ; } } catch ( _ ) { try { if ( event . telemetry ) { const { lat , lng } = event . telemetry . gps ; await addGeospatialHash ( event . device_id , [ lat , lng ] ) ; } return hwrite ( 'event' , event ) ; } catch ( err ) { logger . error ( 'hwrites' , err . stack ) ; throw err ; } } } async function readEvent ( device_id ) { const rawEvent = await hread ( 'event' , device_id ) ; const event = parseEvent ( rawEvent as StringifiedEventWithTelemetry ) ; return event ; } async function readEvents ( device_ids ) { const events = await hreads ( [ 'event' ] , device_ids ) ; return events . map ( e => { return parseEvent ( e as StringifiedEventWithTelemetry ) ; } ) . filter ( e => ! ! e ) ; } async function readAllEvents ( ) { let start = now ( ) ; const keys = await readKeys ( 'device:*:event' ) ; let finish = now ( ) ; let timeElapsed = finish - start ; logger . info ( `MDS-DAILY /admin/events -> cache.readAllEvents() readKeys() time elapsed: ${ timeElapsed }ms` ) ; const device_ids = keys . map ( key => { const [ , device_id ] = key . split ( ':' ) ; return device_id ; } ) ; start = now ( ) ; const result = ( await hreads ( [ 'event' ] , device_ids ) ) . map ( event => { return parseEvent ( event as StringifiedEventWithTelemetry ) ; } ) ; finish = now ( ) ; timeElapsed = finish - start ; logger . info ( `MDS-DAILY /admin/events -> cache.readAllEvents() hreads() time elapsed: ${ timeElapsed }ms` ) ; return result ; } async function readDevice ( device_id ) { if ( ! device_id ) { throw new Error ( 'null device not legal to read' ) ; } const rawDevice = await hread ( 'device' , device_id ) ; const device = parseDevice ( rawDevice as StringifiedCacheReadDeviceResult ) ; return device ; } async function readDevices ( device_ids ) { return ( ( await hreads ( [ 'device' ] , device_ids ) ) as StringifiedCacheReadDeviceResult [ ] ) . map ( device => { return parseDevice ( device ) ; } ) ; } async function readDeviceStatus ( device_id ) { const promises = [ readEvent ( device_id ) , readDevice ( device_id ) ] . map ( ( p ) => p . catch ( ( err ) => { if ( err . name !== 'NotFoundError' ) { throw err ; } } ) ) ; try { const results = await Promise . all ( promises ) ; const deviceStatusMap = { } ; results . filter ( ( item ) : item is CachedItem => item !== undefined ) . map ( item => { deviceStatusMap [ item . device_id ] = deviceStatusMap [ item . device_id ] || { } ; Object . assign ( deviceStatusMap [ item . device_id ] , item ) ; } ) ; const statuses = Object . values ( deviceStatusMap ) ; return statuses . find ( ( status ) => status . telemetry ) || statuses [ 0 ] || null ; } catch ( err ) { logger . error ( 'Error reading device status' , err ) ; throw err ; } } async function readDevicesStatus ( query ) { logger . info ( 'readDevicesStatus' , JSON . stringify ( query ) , 'start' ) ; const start = query . since || 0 ; const stop = now ( ) ; const strictChecking = query . strict ; logger . info ( 'redis zrangebyscore device-ids' , start , stop ) ; const client = await getClient ( ) ; const geoStart = now ( ) ; const { bbox } = query ; const deviceIdsInBbox = await getEventsInBBox ( bbox ) ; const deviceIdsRes = deviceIdsInBbox . length === 0 ? await client . zrangebyscoreAsync ( decorateKey ( 'device-ids' ) , start , stop ) : deviceIdsInBbox ; const skip = query . skip || 0 ; const take = query . take || 100000000000 ; const deviceIds = deviceIdsRes . slice ( skip , skip + take ) ; const geoFinish = now ( ) ; const timeElapsed = geoFinish - geoStart ; logger . info ( `mds-agency-cache readDevicesStatus bbox fetch ${ JSON . stringify ( bbox ) } time elapsed: ${ timeElapsed }ms` ) ; const eventsStart = now ( ) ; const events = ( ( await hreads ( [ 'event' ] , deviceIds ) ) as StringifiedEvent [ ] ) . reduce ( ( acc , item ) => { try { const parsedItem = parseEvent ( item ) ; if ( EVENT_STATUS_MAP [ parsedItem . event_type ] === VEHICLE_STATUSES . removed || ! parsedItem . telemetry || ( strictChecking && ! isInsideBoundingBox ( parsedItem . telemetry , query . bbox ) ) ) return acc ; return [ ... acc , parsedItem ] ; } catch ( err ) { return acc ; } } , [ ] ) . filter ( item => Boolean ( item ) ) ; const eventsFinish = now ( ) ; const eventsTimeElapsed = eventsFinish - eventsStart ; logger . info ( `mds-agency-cache readDevicesStatus bbox check ${ JSON . stringify ( bbox ) } time elapsed: ${ eventsTimeElapsed }ms` ) ; const devicesStart = now ( ) ; const eventDeviceIds = events . map ( event => event . device_id ) ; const devices = ( await hreads ( [ 'device' ] , eventDeviceIds ) ) . reduce ( ( acc , item ) => { try { const parsedItem = parseCachedItem ( item ) ; return [ ... acc , parsedItem ] ; } catch ( err ) { return acc ; } } , [ ] ) . filter ( item => Boolean ( item ) ) ; const all = [ ... devices , ... events ] ; const deviceStatusMap = { } ; all . map ( item => { deviceStatusMap [ item . device_id ] = deviceStatusMap [ item . device_id ] || { } ; Object . assign ( deviceStatusMap [ item . device_id ] , item ) ; } ) ; const values = Object . values ( deviceStatusMap ) ; const valuesWithTelemetry = values . filter ( ( item ) => item . telemetry ) ; const devicesFinish = now ( ) ; const devicesTimeElapsed = devicesFinish - devicesStart ; logger . info ( `mds-agency-cache readDevicesStatus device processing ${ JSON . stringify ( bbox ) } time elapsed: ${ devicesTimeElapsed }ms` ) ; return valuesWithTelemetry ; } async function readTelemetry ( device_id ) { const telemetry = await hread ( 'telemetry' , device_id ) ; return parseTelemetry ( telemetry as StringifiedTelemetry ) ; } async function writeOneTelemetry ( telemetry ) { const { lat , lng } = telemetry . gps ; try { const prevTelemetry = await readTelemetry ( telemetry . device_id ) ; if ( prevTelemetry . timestamp < telemetry . timestamp ) { try { await addGeospatialHash ( telemetry . device_id , [ lat , lng ] ) ; return hwrite ( 'telemetry' , telemetry ) ; } catch ( err ) { logger . error ( 'hwrite' , err . stack ) ; return Promise . reject ( err ) ; } } else { return Promise . resolve ( ) ; } } catch ( err ) { logger . info ( 'writeOneTelemetry: no prior telemetry found:' , err . message ) ; try { await addGeospatialHash ( telemetry . device_id , [ lat , lng ] ) ; return hwrite ( 'telemetry' , telemetry ) ; } catch ( err2 ) { logger . error ( 'writeOneTelemetry hwrite2' , err . stack ) ; return Promise . reject ( err2 ) ; } } } async function writeTelemetry ( telemetries ) { await Promise . all ( telemetries . map ( telemetry => writeOneTelemetry ( telemetry ) ) ) ; } async function readAllTelemetry ( ) { const keys = await readKeys ( 'device:*:telemetry' ) ; const device_ids = keys . map ( key => { const [ , device_id ] = key . split ( ':' ) ; return device_id ; } ) ; return ( ( await hreads ( [ 'telemetry' ] , device_ids ) ) as StringifiedTelemetry [ ] ) . reduce ( ( acc , telemetry ) => { try { return [ ... acc , parseTelemetry ( telemetry ) ] ; } catch ( err ) { logger . info ( JSON . parse ( err ) ) ; return acc ; } } , [ ] ) ; } async function seed ( dataParam ) { logger . info ( 'cache seed' ) ; const data = dataParam || { devices : [ ] , events : [ ] , telemetry : [ ] } ; await data . devices . map ( writeDevice ) ; await data . events . map ( writeEvent ) ; if ( data . telemetry . length !== 0 ) { await writeTelemetry ( data . telemetry . sort ( ( a , b ) => a . timestamp - b . timestamp ) ) ; } logger . info ( 'cache seed redis done' ) ; } async function reset ( ) { logger . info ( 'cache reset' ) ; await ( await getClient ( ) ) . flushdbAsync ( ) ; return logger . info ( 'redis flushed' ) ; } async function initialize ( ) { await getClient ( ) ; await reset ( ) ; } async function startup ( ) { await getClient ( ) ; } function shutdown ( ) { if ( cachedClient ) { cachedClient . quit ( ) ; cachedClient = null ; } } async function health ( ) { return Promise . resolve ( 'we good' ) ; } async function cleanup ( ) { try { const keys = await readKeys ( 'device:*' ) ; logger . warn ( 'cleanup: read' , keys . length ) ; const report = { telemetry : 0 , device : 0 , event : 0 } ; try { let badKeys = [ ] ; keys . map ( key => { const [ , , suffix ] = key . split ( ':' ) ; if ( suffix ) { badKeys . push ( key ) ; report [ suffix ] += 1 ; } } ) ; badKeys = badKeys . slice ( 0 , 10000 ) ; const result = await ( await getClient ( ) ) . delAsync ( ... badKeys ) ; report . deleted = result ; return report ; } catch ( ex ) { logger . error ( 'cleanup: exception' , ex ) ; throw ex ; } } catch ( ex ) { logger . error ( 'cleanup: exception' , ex ) ; return Promise . reject ( ex ) ; } } export default { initialize , health , info , seed , reset , startup , shutdown , writeDevice , writeEvent , writeOneTelemetry , writeTelemetry , readDevice , readDevices , readDeviceStatus , readDevicesStatus , readEvent , readEvents , readAllEvents , readTelemetry , readAllTelemetry , readKeys , wipeDevice , updateVehicleList , cleanup , getMostRecentEventByProvider } ;