import { Injectable , Inject } from '@angular/core' ; import { Observable , Subject , of , merge as mergeStatic , throwError , timer } from 'rxjs' ; import { catchError , filter , map , merge , share , startWith , switchMap , take , takeUntil } from 'rxjs/operators' ; import { takeWhileInclusive } from '../../rx/takeWhileInclusive' ; import { snapshotToValue } from '../../rx/snapshotToValue' ; import { Lab } from '../../models/lab' ; import { Invocation , InvocationType } from '../../models/invocation' ; import { TimeoutError , RateLimitError } from './errors' ; import { Execution , ExecutionMessage , ExecutionWrapper , ExecutionInvocationInfo } from '../../models/execution' ; import { MessageStreamOptimizer } from './message-stream-optimizer' ; import * as shortid from 'shortid' ; import * as firebase from 'firebase' ; import { DbRefBuilder } from '../../firebase/db-ref-builder' ; import { AuthService } from '../../auth' ; import { parseLabDirectory } from '@machinelabs/core' ; import { stringifyDirectory } from '../../util/directory' ; import { ExecutionStatus , MessageKind } from '@machinelabs/models' ; @ Injectable ( ) export class RemoteLabExecService { PARTITION_SIZE = 1000 ; FULL_FETCH_TRESHOLD = 5000 ; messageStreamOptimizer : MessageStreamOptimizer ; constructor ( private db , private authService ) { this . messageStreamOptimizer = new MessageStreamOptimizer ( db , this . PARTITION_SIZE , this . FULL_FETCH_TRESHOLD ) ; } run ( lab , timeout = 15000 ) { const id = this . newInvocationId ( ) ; const timeout$ = timer ( timeout ) . pipe ( switchMap ( _ => throwError ( new TimeoutError ( id , 'Timeout' ) ) ) ) ; return this . startWithRateProof ( id ) . pipe ( switchMap ( login => this . db . invocationRef ( id ) . set ( { id : id , user_id : login . uid , timestamp : firebase . database . ServerValue . TIMESTAMP , type : InvocationType . StartExecution , data : { id : lab . id , directory : stringifyDirectory ( lab . directory ) } } ) ) , switchMap ( _ => this . db . executionMessageRef ( id ) . limitToFirst ( 1 ) . childAdded ( ) . pipe ( take ( 1 ) ) ) , snapshotToValue , switchMap ( message => { if ( message . kind === MessageKind . ExecutionRejected ) { return of ( { executionId : id , rejection : message . data , persistent : false } ) ; } else { return this . db . executionRef ( id ) . value ( ) . pipe ( snapshotToValue , takeWhileInclusive ( e => e === null ) , filter ( e => e !== null ) , map ( _ => ( { persistent : true , executionId : id , rejection : null } ) ) ) ; } } ) , startWith ( { executionId : id , persistent : false , rejection : null } ) , merge ( timeout$ ) , take ( 2 ) , catchError ( e => { const error = e instanceof TimeoutError ? e : new RateLimitError ( id , 'Rate limit exceeded' ) ; return throwError ( e ) ; } ) ) ; } runAndListen ( lab ) { return this . run ( lab ) . pipe ( map ( info => this . listen ( info . executionId ) ) ) ; } listen ( executionId ) { const execution$ = this . db . executionRef ( executionId ) . value ( ) . pipe ( snapshotToValue , map ( execution => { if ( typeof execution . lab . directory === 'string' ) { execution . lab . directory = parseLabDirectory ( execution . lab . directory ) ; } return execution ; } ) ) ; const messages$ = this . messageStreamOptimizer . listenForMessages ( executionId ) ; const controlMessages$ = this . observeControlMessages ( executionId ) ; return this . consumeExecution ( messages$ , controlMessages$ , execution$ ) ; } consumeExecution ( messages , controlMessages , execution ) { const controlMessages$ = controlMessages . pipe ( share ( ) ) ; const sharedMessages = messages . pipe ( share ( ) ) ; const sharedExecution = execution . pipe ( share ( ) ) ; const messagesNotFinishedOrRejected = ( msg ) => msg . kind !== MessageKind . ExecutionFinished && msg . kind !== MessageKind . ExecutionRejected ; const executingExecutions = ( exec ) => exec . status === ExecutionStatus . Executing ; const messages$ = sharedMessages . pipe ( takeWhileInclusive ( messagesNotFinishedOrRejected ) ) ; const rejectedMessage = controlMessages$ . pipe ( filter ( msg => msg . kind === MessageKind . ExecutionRejected ) ) ; const execution$ = sharedExecution . pipe ( filter ( exec => ! ! exec ) , takeWhileInclusive ( executingExecutions ) , takeUntil ( rejectedMessage ) ) ; return { messages : messages$ , controlMessages : controlMessages$ , execution : execution$ } ; } stop ( executionId ) { const id = this . newInvocationId ( ) ; return this . startWithRateProof ( id ) . pipe ( switchMap ( login => this . db . invocationRef ( id ) . set ( { id : id , type : InvocationType . StopExecution , data : { execution_id : executionId } , user_id : login . uid , timestamp : firebase . database . ServerValue . TIMESTAMP } ) ) ) . subscribe ( ) ; } private startWithRateProof ( id ) { return this . authService . requireAuthOnce ( ) . pipe ( switchMap ( login => this . db . userInvocationRateProofRef ( login . uid ) . set ( { timestamp : firebase . database . ServerValue . TIMESTAMP , key : id } ) . pipe ( map ( _ => login ) ) ) ) ; } private newInvocationId ( ) { return `${ Date . now ( ) }-${ shortid . generate ( ) }` ; } private observeControlMessages ( executionId ) { return mergeStatic ( this . observeControlMessage ( executionId , MessageKind . ExecutionRejected ) , this . observeControlMessage ( executionId , MessageKind . ExecutionStarted ) , this . observeControlMessage ( executionId , MessageKind . ExecutionFinished ) ) . pipe ( takeWhileInclusive ( ( msg ) => msg . kind === MessageKind . ExecutionStarted ) ) ; } private observeControlMessage ( executionId , kind ) { return this . db . executionMessageRef ( executionId ) . orderByChild ( 'kind' ) . equalTo ( kind ) . limitToFirst ( 1 ) . childAdded ( ) . pipe ( take ( 1 ) , snapshotToValue ) ; } }