import { App } from './../app' ; import { RedlockMutex } from 'redis-semaphore' ; import { StatsDriver } from './stats-driver' ; const dayjs = require ( 'dayjs' ) ; const Redis = require ( 'ioredis' ) ; export class RedisStats implements StatsDriver { type { Redis } protected redis : typeof Redis ; type { string } protected setListName = 'stats:apps' ; param { any } options constructor ( protected options ) { this . redis = new Redis ( options . database . redis ) ; } param { App } app return { Promise < number > } markNewConnection ( app ) { if ( ! this . canRegisterStats ( app ) ) { return new Promise ( resolve => resolve ( 0 ) ) ; } return this . registerApp ( app ) . then ( ( ) => { return this . redis . hincrby ( this . getKey ( app , 'stats' ) , 'connections' , 1 ) . then ( currentConnections => { return this . recalculatePeakConnections ( app , currentConnections ) . then ( peakConnections => currentConnections ) ; } ) ; } ) ; } param { App } app param { string | null } reason return { Promise < number > } markDisconnection ( app , reason ? ) { if ( ! this . canRegisterStats ( app ) ) { return new Promise ( resolve => resolve ( 0 ) ) ; } return this . registerApp ( app ) . then ( ( ) => { return this . redis . hincrby ( this . getKey ( app , 'stats' ) , 'connections' , - 1 ) . then ( currentConnections => { return this . recalculatePeakConnections ( app , currentConnections ) . then ( peakConnections => currentConnections ) ; } ) ; } ) ; } param { App } app return { Promise < number > } markApiMessage ( app ) { if ( ! this . canRegisterStats ( app ) ) { return new Promise ( resolve => resolve ( 0 ) ) ; } return this . registerApp ( app ) . then ( ( ) => { return this . redis . hincrby ( this . getKey ( app , 'stats' ) , 'api_messages' , 1 ) ; } ) ; } param { App } app return { Promise < number > } markWsMessage ( app ) { if ( ! this . canRegisterStats ( app ) ) { return new Promise ( resolve => resolve ( 0 ) ) ; } return this . registerApp ( app ) . then ( ( ) => { return this . redis . hincrby ( this . getKey ( app , 'stats' ) , 'ws_messages' , 1 ) ; } ) ; } param { App | string } app return { Promise < any > } getStats ( app ) { return this . redis . hgetall ( this . getKey ( app , 'stats' ) ) . then ( response => { return { connections : parseInt ( response . connections || 0 ) , peak_connections : parseInt ( response . peak_connections || 0 ) , api_messages : parseInt ( response . api_messages || 0 ) , ws_messages : parseInt ( response . ws_messages || 0 ) , } ; } ) ; } param { App | string } app param { number | null } time return { Promise < any > } takeSnapshot ( app , time ? ) { let mutex = this . mutex ( app ) ; return mutex . acquire ( ) . then ( ( ) => { return this . getStats ( app ) . then ( stats => { let record = { time , stats , } ; return this . redis . zadd ( this . getKey ( app , 'snapshots' ) , time , JSON . stringify ( record ) ) . then ( ( ) => { mutex . release ( ) ; return record ; } ) ; } ) . then ( record => { return this . hasActivity ( app ) . then ( hasActivity => { if ( ! hasActivity ) { return this . resetAppTraces ( app ) . then ( ( ) => record ) ; } /**
                     * Reset the messages stats to 0 only
                     * if the app still has activity.
                     */ this . resetMessagesStats ( app , record . stats . connections ) ; return record ; } ) ; } ) ; } ) ; } param { App | string } app param { number | null } start param { number | null } end return { Promise < any > } getSnapshots ( app , start ? , end ? ) { start = start ? start : dayjs ( ) . subtract ( 7 , 'day' ) . unix ( ) , end = end ? end : dayjs ( ) . unix ( ) ; return this . redis . zrangebyscore ( this . getKey ( app , 'snapshots' ) , start , end ) . then ( points => points . map ( point => JSON . parse ( point ) ) ) ; } param { App | string } app param { number | null } time return { Promise < boolean > } deleteStalePoints ( app , time ? ) { let score = time - this . options . stats . retention . period - 1 ; return this . redis . zremrangebyscore ( this . getKey ( app , 'snapshots' ) , 0 , score ) . then ( ( ) => { return true ; } ) ; } param { App | string } app return { Promise < boolean > } registerApp ( app ) { let appKey = app instanceof App ? app . key : app ; return this . redis . sadd ( this . setListName , appKey ) . then ( ( ) => true ) ; } return { Promise < string [ ] > } getRegisteredApps ( ) { return this . redis . smembers ( this . setListName ) ; } param { App | string } app param { array } suffixes protected getKey ( app , ... suffixes ) { let appKey = app instanceof App ? app . key : app ; let compiledSuffixes = suffixes . join ( ':' ) ; return appKey + ( compiledSuffixes ? `:${ compiledSuffixes }` : '' ) ; } param { App | string } app param { number } currentConnections return { Promise < number > } protected recalculatePeakConnections ( app , currentConnections ) { return this . redis . hget ( this . getKey ( app , 'stats' ) , 'peak_connections' ) . then ( peakConnections => { peakConnections = peakConnections ? Math . max ( peakConnections , currentConnections ) : currentConnections ; return this . redis . hset ( this . getKey ( app , 'stats' ) , 'peak_connections' , peakConnections ) . then ( ( ) => { return peakConnections ; } ) ; } ) ; } param { App | string } app param { number } connections return { void } protected resetMessagesStats ( app , connections ) { return Promise . all ( [ this . registerApp ( app ) . then ( ( ) => this . redis . hset ( this . getKey ( app , 'stats' ) , 'api_messages' , 0 ) ) , this . registerApp ( app ) . then ( ( ) => this . redis . hset ( this . getKey ( app , 'stats' ) , 'ws_messages' , 0 ) ) , this . registerApp ( app ) . then ( ( ) => this . redis . hset ( this . getKey ( app , 'stats' ) , 'peak_connections' , connections ) ) , ] ) ; } param { App | string } app return { Promise < boolean > } protected resetAppTraces ( app ) { return new Promise ( resolve => { let appKey = app instanceof App ? app . key : app ; return this . redis . zremrangebyscore ( this . getKey ( app , 'snapshots' ) , 0 , Infinity ) . then ( ( ) => { return this . redis . srem ( this . setListName , appKey ) . then ( ( ) => true ) ; } ) ; } ) ; } param { App | string } app return { Promise < boolean > } protected hasActivity ( app ) { return this . getSnapshots ( app , 0 , Infinity ) . then ( snapshots => { let activeSnapshots = snapshots . filter ( point => { /**
                 * Check if the sum of all stats are > 0.
                 * If it is, it means that the app still has activity.
                 */ return Object . values ( point . stats ) . reduce ( ( sum , num ) => sum += num , 0 ) > 0 ; } ) ; return activeSnapshots . length > 0 ; } ) ; } param { App } app return { boolean } protected canRegisterStats ( app ) { return this . options . stats . enabled && ! ! app . enableStats ; } param { App | string } app return { RedlockMutex } protected mutex ( app ) { return new RedlockMutex ( [ this . redis ] , this . getKey ( app , 'lock' ) , { lockTimeout : this . options . stats . snapshots . interval * 1000 , refreshInterval : 0 , } ) ; } }