import { Observable , forkJoin , merge , from } from 'rxjs' ; import { map , mergeMap , filter } from 'rxjs/operators' ; import { ObservableDbRef , DbRefBuilder , DateUtil } from '@machinelabs/core' ; import { ShortMonth , toShortMonth , HardwareType , PlanId , PlanCredits } from '@machinelabs/models' ; import { CostCalculator } from '../costs/cost-calculator' ; import { CostReport } from '../costs/cost-report' ; import { UsageStatistic } from './usage-statistic' ; import { LiveMetricsService } from '../live/live-metrics.service' ; const hoursToSeconds = ( hours ) => hours * 60 * 60 ; export class UsageStatisticService { constructor ( private costCalculator , private liveMetricsService , private db ) { } getUsageStatisticForAllCurrentlyActiveUsers ( ) { return this . liveMetricsService . getLiveExecutionsRef ( ) . onceValue ( ) . pipe ( map ( snapshot => snapshot . val ( ) ) , filter ( val => ! ! val ) , mergeMap ( val => Object . keys ( val ) ) , mergeMap ( userId => { return forkJoin ( this . db . userRef ( userId ) . onceValue ( ) . pipe ( map ( snapshot => snapshot . val ( ) ) ) , this . getCostReportForCurrentMonth ( userId ) ) . pipe ( map ( ( [ fullUser , costReport ] ) => this . calculateStatisticForPlan ( fullUser . common . id , fullUser . plan . plan_id , costReport ) ) ) ; } ) ) ; } getUsageStatisticForCurrentMonth ( userId , planId ) { const costReport$ = this . getCostReportForCurrentMonth ( userId ) ; return this . calculateStatistic ( userId , planId , costReport$ ) ; } getUsageStatistic ( userId , planId , year , month ) { const costReport$ = this . getCostReport ( userId , year , month ) ; return this . calculateStatistic ( userId , planId , costReport$ ) ; } getCostReportForCurrentMonth ( userId ) { return this . getCostReport ( userId , DateUtil . getCurrentUtcYear ( ) , DateUtil . getCurrentUtcShortMonth ( ) ) ; } getCostReport ( userId , year , month ) { const executions$ = merge ( this . db . userExecutionsByMonthRef ( userId , year , month ) . onceValue ( ) , this . db . userExecutionsLiveRef ( userId ) . onceValue ( ) ) . pipe ( map ( snapshot => snapshot . val ( ) ) , map ( val => ( val ? Object . keys ( val ) : [ ] ) ) , mergeMap ( executionIds => from ( executionIds ) . pipe ( mergeMap ( id => this . db . executionRef ( id ) . onceValue ( ) ) ) ) , map ( snapshot => snapshot . val ( ) ) , filter ( execution => execution !== null ) ) ; return this . costCalculator . calc ( executions$ ) ; } calculateStatistic ( userId , planId , costReport ) { return costReport . pipe ( map ( report => this . calculateStatisticForPlan ( userId , planId , report ) ) ) ; } calculateStatisticForPlan ( userId , planId , report ) { const credits = PlanCredits . get ( planId ) ; if ( ! credits || ! report ) { return null ; } return { userId : userId , planId : planId , costReport : report , cpuSecondsLeft : hoursToSeconds ( credits . cpuHoursPerMonth ) - report . getSecondsPerHardware ( HardwareType . CPU ) , gpuSecondsLeft : hoursToSeconds ( credits . gpuHoursPerMonth ) - report . getSecondsPerHardware ( HardwareType . GPU ) } ; } }