import type * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey' ; import _ from 'lodash' ; import { first } from 'rxjs/operators' ; import { TaskInstance , TaskStatus , TaskLifecycleResult , SerializedConcreteTaskInstance , } from './task' ; import { elasticsearchServiceMock } from '../../../../src/core/server/mocks' ; import { TaskStore , SearchOpts } from './task_store' ; import { savedObjectsRepositoryMock } from 'src/core/server/mocks' ; import { SavedObjectsSerializer , SavedObjectTypeRegistry , SavedObjectAttributes , SavedObjectsErrorHelpers , } from 'src/core/server' ; import { TaskTypeDictionary } from './task_type_dictionary' ; import { mockLogger } from './test_utils' ; const savedObjectsClient = savedObjectsRepositoryMock . create ( ) ; const serializer = new SavedObjectsSerializer ( new SavedObjectTypeRegistry ( ) ) ; beforeEach ( ( ) => jest . resetAllMocks ( ) ) ; const mockedDate = new Date ( '2019-02-12T21:01:22.479Z' ) ; ( global as any ) . Date = class Date { constructor ( ) { return mockedDate ; } static now ( ) { return mockedDate . getTime ( ) ; } } ; const taskDefinitions = new TaskTypeDictionary ( mockLogger ( ) ) ; taskDefinitions . registerTaskDefinitions ( { report : { title : 'report' , createTaskRunner : jest . fn ( ) , } , dernstraight : { title : 'dernstraight' , createTaskRunner : jest . fn ( ) , } , yawn : { title : 'yawn' , createTaskRunner : jest . fn ( ) , } , } ) ; describe ( 'TaskStore' , ( ) => { describe ( 'schedule' , ( ) => { let store ; beforeAll ( ( ) => { store = new TaskStore ( { index : 'tasky' , taskManagerId : '' , serializer , esClient : elasticsearchServiceMock . createClusterClient ( ) . asInternalUser , definitions : taskDefinitions , savedObjectsRepository : savedObjectsClient , } ) ; } ) ; async function testSchedule ( task ) { savedObjectsClient . create . mockImplementation ( async ( type , attributes ) => ( { id : 'testid' , type , attributes , references : [ ] , version : '123' , } ) ) ; const result = await store . schedule ( task as TaskInstance ) ; expect ( savedObjectsClient . create ) . toHaveBeenCalledTimes ( 1 ) ; return result ; } test ( 'serializes the params and state' , async ( ) => { const task = { id : 'id' , params : { hello : 'world' } , state : { foo : 'bar' } , taskType : 'report' , traceparent : 'apmTraceparent' , } ; const result = await testSchedule ( task ) ; expect ( savedObjectsClient . create ) . toHaveBeenCalledWith ( 'task' , { attempts : 0 , schedule : undefined , params : '{"hello":"world"}' , retryAt : null , runAt : '2019-02-12T21:01:22.479Z' , scheduledAt : '2019-02-12T21:01:22.479Z' , scope : undefined , startedAt : null , state : '{"foo":"bar"}' , status : 'idle' , taskType : 'report' , user : undefined , traceparent : 'apmTraceparent' , } , { id : 'id' , refresh : false , } ) ; expect ( result ) . toEqual ( { id : 'testid' , attempts : 0 , schedule : undefined , params : { hello : 'world' } , retryAt : null , runAt : mockedDate , scheduledAt : mockedDate , scope : undefined , startedAt : null , state : { foo : 'bar' } , status : 'idle' , taskType : 'report' , user : undefined , version : '123' , traceparent : 'apmTraceparent' , } ) ; } ) ; test ( 'returns a concrete task instance' , async ( ) => { const task = { params : { hello : 'world' } , state : { foo : 'bar' } , taskType : 'report' , } ; const result = await testSchedule ( task ) ; expect ( result ) . toMatchObject ( { ... task , id : 'testid' , } ) ; } ) ; test ( 'sets runAt to now if not specified' , async ( ) => { await testSchedule ( { taskType : 'dernstraight' , params : { } , state : { } } ) ; expect ( savedObjectsClient . create ) . toHaveBeenCalledTimes ( 1 ) ; const attributes = savedObjectsClient . create . mock . calls [ 0 ] [ 1 ] as SerializedConcreteTaskInstance ; expect ( new Date ( attributes . runAt as string ) . getTime ( ) ) . toEqual ( mockedDate . getTime ( ) ) ; } ) ; test ( 'ensures params and state are not null' , async ( ) => { await testSchedule ( { taskType : 'yawn' } ) ; expect ( savedObjectsClient . create ) . toHaveBeenCalledTimes ( 1 ) ; const attributes = savedObjectsClient . create . mock . calls [ 0 ] [ 1 ] as SerializedConcreteTaskInstance ; expect ( attributes . params ) . toEqual ( '{}' ) ; expect ( attributes . state ) . toEqual ( '{}' ) ; } ) ; test ( 'errors if the task type is unknown' , async ( ) => { await expect ( testSchedule ( { taskType : 'nope' , params : { } , state : { } } ) ) . rejects . toThrow ( /Unsupported task type "nope"/i ) ; } ) ; test ( 'pushes error from saved objects client to errors$' , async ( ) => { const task = { id : 'id' , params : { hello : 'world' } , state : { foo : 'bar' } , taskType : 'report' , } ; const firstErrorPromise = store . errors$ . pipe ( first ( ) ) . toPromise ( ) ; savedObjectsClient . create . mockRejectedValue ( new Error ( 'Failure' ) ) ; await expect ( store . schedule ( task ) ) . rejects . toThrowErrorMatchingInlineSnapshot ( `"Failure"` ) ; expect ( await firstErrorPromise ) . toMatchInlineSnapshot ( `[Error: Failure]` ) ; } ) ; } ) ; describe ( 'fetch' , ( ) => { let store ; let esClient ; beforeAll ( ( ) => { esClient = elasticsearchServiceMock . createClusterClient ( ) . asInternalUser ; store = new TaskStore ( { index : 'tasky' , taskManagerId : '' , serializer , esClient , definitions : taskDefinitions , savedObjectsRepository : savedObjectsClient , } ) ; } ) ; async function testFetch ( opts ? , hits = [ ] ) { esClient . search . mockResolvedValue ( asApiResponse ( { hits : { hits , total : hits . length } } ) ) ; const result = await store . fetch ( opts ) ; expect ( esClient . search ) . toHaveBeenCalledTimes ( 1 ) ; return { result , args : esClient . search . mock . calls [ 0 ] [ 0 ] , } ; } test ( 'empty call filters by type, sorts by runAt and id' , async ( ) => { const { args } = await testFetch ( ) ; expect ( args ) . toMatchObject ( { index : 'tasky' , body : { sort : [ { 'task.runAt' : 'asc' } ] , query : { term : { type : 'task' } } , } , } ) ; } ) ; test ( 'allows custom queries' , async ( ) => { const { args } = await testFetch ( { query : { term : { 'task.taskType' : 'bar' } , } , } ) ; expect ( args ) . toMatchObject ( { body : { query : { bool : { must : [ { term : { type : 'task' } } , { term : { 'task.taskType' : 'bar' } } ] , } , } , } , } ) ; } ) ; test ( 'pushes error from call cluster to errors$' , async ( ) => { const firstErrorPromise = store . errors$ . pipe ( first ( ) ) . toPromise ( ) ; esClient . search . mockRejectedValue ( new Error ( 'Failure' ) ) ; await expect ( store . fetch ( ) ) . rejects . toThrowErrorMatchingInlineSnapshot ( `"Failure"` ) ; expect ( await firstErrorPromise ) . toMatchInlineSnapshot ( `[Error: Failure]` ) ; } ) ; } ) ; describe ( 'update' , ( ) => { let store ; let esClient ; beforeAll ( ( ) => { esClient = elasticsearchServiceMock . createClusterClient ( ) . asInternalUser ; store = new TaskStore ( { index : 'tasky' , taskManagerId : '' , serializer , esClient , definitions : taskDefinitions , savedObjectsRepository : savedObjectsClient , } ) ; } ) ; test ( 'refreshes the index, handles versioning' , async ( ) => { const task = { runAt : mockedDate , scheduledAt : mockedDate , startedAt : null , retryAt : null , id : 'task:324242' , params : { hello : 'world' } , state : { foo : 'bar' } , taskType : 'report' , attempts : 3 , status : 'idle' as TaskStatus , version : '123' , ownerId : null , traceparent : 'myTraceparent' , } ; savedObjectsClient . update . mockImplementation ( async ( type , id , attributes ) => { return { id , type , attributes , references : [ ] , version : '123' , } ; } ) ; const result = await store . update ( task ) ; expect ( savedObjectsClient . update ) . toHaveBeenCalledWith ( 'task' , task . id , { attempts : task . attempts , schedule : undefined , params : JSON . stringify ( task . params ) , retryAt : null , runAt : task . runAt . toISOString ( ) , scheduledAt : mockedDate . toISOString ( ) , scope : undefined , startedAt : null , state : JSON . stringify ( task . state ) , status : task . status , taskType : task . taskType , user : undefined , ownerId : null , traceparent : 'myTraceparent' , } , { version : '123' , refresh : false } ) ; expect ( result ) . toEqual ( { ... task , schedule : undefined , retryAt : null , scope : undefined , startedAt : null , user : undefined , version : '123' , } ) ; } ) ; test ( 'pushes error from saved objects client to errors$' , async ( ) => { const task = { runAt : mockedDate , scheduledAt : mockedDate , startedAt : null , retryAt : null , id : 'task:324242' , params : { hello : 'world' } , state : { foo : 'bar' } , taskType : 'report' , attempts : 3 , status : 'idle' as TaskStatus , version : '123' , ownerId : null , traceparent : '' , } ; const firstErrorPromise = store . errors$ . pipe ( first ( ) ) . toPromise ( ) ; savedObjectsClient . update . mockRejectedValue ( new Error ( 'Failure' ) ) ; await expect ( store . update ( task ) ) . rejects . toThrowErrorMatchingInlineSnapshot ( `"Failure"` ) ; expect ( await firstErrorPromise ) . toMatchInlineSnapshot ( `[Error: Failure]` ) ; } ) ; } ) ; describe ( 'bulkUpdate' , ( ) => { let store ; beforeAll ( ( ) => { store = new TaskStore ( { index : 'tasky' , taskManagerId : '' , serializer , esClient : elasticsearchServiceMock . createClusterClient ( ) . asInternalUser , definitions : taskDefinitions , savedObjectsRepository : savedObjectsClient , } ) ; } ) ; test ( 'pushes error from saved objects client to errors$' , async ( ) => { const task = { runAt : mockedDate , scheduledAt : mockedDate , startedAt : null , retryAt : null , id : 'task:324242' , params : { hello : 'world' } , state : { foo : 'bar' } , taskType : 'report' , attempts : 3 , status : 'idle' as TaskStatus , version : '123' , ownerId : null , traceparent : '' , } ; const firstErrorPromise = store . errors$ . pipe ( first ( ) ) . toPromise ( ) ; savedObjectsClient . bulkUpdate . mockRejectedValue ( new Error ( 'Failure' ) ) ; await expect ( store . bulkUpdate ( [ task ] ) ) . rejects . toThrowErrorMatchingInlineSnapshot ( `"Failure"` ) ; expect ( await firstErrorPromise ) . toMatchInlineSnapshot ( `[Error: Failure]` ) ; } ) ; } ) ; describe ( 'remove' , ( ) => { let store ; beforeAll ( ( ) => { store = new TaskStore ( { index : 'tasky' , taskManagerId : '' , serializer , esClient : elasticsearchServiceMock . createClusterClient ( ) . asInternalUser , definitions : taskDefinitions , savedObjectsRepository : savedObjectsClient , } ) ; } ) ; test ( 'removes the task with the specified id' , async ( ) => { const id = randomId ( ) ; const result = await store . remove ( id ) ; expect ( result ) . toBeUndefined ( ) ; expect ( savedObjectsClient . delete ) . toHaveBeenCalledWith ( 'task' , id ) ; } ) ; test ( 'pushes error from saved objects client to errors$' , async ( ) => { const firstErrorPromise = store . errors$ . pipe ( first ( ) ) . toPromise ( ) ; savedObjectsClient . delete . mockRejectedValue ( new Error ( 'Failure' ) ) ; await expect ( store . remove ( randomId ( ) ) ) . rejects . toThrowErrorMatchingInlineSnapshot ( `"Failure"` ) ; expect ( await firstErrorPromise ) . toMatchInlineSnapshot ( `[Error: Failure]` ) ; } ) ; } ) ; describe ( 'get' , ( ) => { let store ; beforeAll ( ( ) => { store = new TaskStore ( { index : 'tasky' , taskManagerId : '' , serializer , esClient : elasticsearchServiceMock . createClusterClient ( ) . asInternalUser , definitions : taskDefinitions , savedObjectsRepository : savedObjectsClient , } ) ; } ) ; test ( 'gets the task with the specified id' , async ( ) => { const task = { runAt : mockedDate , scheduledAt : mockedDate , startedAt : null , retryAt : null , id : randomId ( ) , params : { hello : 'world' } , state : { foo : 'bar' } , taskType : 'report' , attempts : 3 , status : 'idle' as TaskStatus , version : '123' , ownerId : null , } ; savedObjectsClient . get . mockImplementation ( async ( type , objectId ) => ( { id : objectId , type , attributes : { ... _ . omit ( task , 'id' ) , ... _ . mapValues ( _ . pick ( task , [ 'params' , 'state' ] ) , ( value ) => JSON . stringify ( value ) ) , } , references : [ ] , version : '123' , } ) ) ; const result = await store . get ( task . id ) ; expect ( result ) . toEqual ( task ) ; expect ( savedObjectsClient . get ) . toHaveBeenCalledWith ( 'task' , task . id ) ; } ) ; test ( 'pushes error from saved objects client to errors$' , async ( ) => { const firstErrorPromise = store . errors$ . pipe ( first ( ) ) . toPromise ( ) ; savedObjectsClient . get . mockRejectedValue ( new Error ( 'Failure' ) ) ; await expect ( store . get ( randomId ( ) ) ) . rejects . toThrowErrorMatchingInlineSnapshot ( `"Failure"` ) ; expect ( await firstErrorPromise ) . toMatchInlineSnapshot ( `[Error: Failure]` ) ; } ) ; } ) ; describe ( 'getLifecycle' , ( ) => { test ( 'returns the task status if the task exists ' , async ( ) => { expect . assertions ( 5 ) ; return Promise . all ( Object . values ( TaskStatus ) . map ( async ( status ) => { const task = { runAt : mockedDate , scheduledAt : mockedDate , startedAt : null , retryAt : null , id : randomId ( ) , params : { hello : 'world' } , state : { foo : 'bar' } , taskType : 'report' , attempts : 3 , status : status as TaskStatus , version : '123' , ownerId : null , traceparent : 'myTraceparent' , } ; savedObjectsClient . get . mockImplementation ( async ( type , objectId ) => ( { id : objectId , type , attributes : { ... _ . omit ( task , 'id' ) , ... _ . mapValues ( _ . pick ( task , [ 'params' , 'state' ] ) , ( value ) => JSON . stringify ( value ) ) , } , references : [ ] , version : '123' , } ) ) ; const store = new TaskStore ( { index : 'tasky' , taskManagerId : '' , serializer , esClient : elasticsearchServiceMock . createClusterClient ( ) . asInternalUser , definitions : taskDefinitions , savedObjectsRepository : savedObjectsClient , } ) ; expect ( await store . getLifecycle ( task . id ) ) . toEqual ( status ) ; } ) ) ; } ) ; test ( 'returns NotFound status if the task doesnt exists ' , async ( ) => { savedObjectsClient . get . mockRejectedValueOnce ( SavedObjectsErrorHelpers . createGenericNotFoundError ( 'type' , 'id' ) ) ; const store = new TaskStore ( { index : 'tasky' , taskManagerId : '' , serializer , esClient : elasticsearchServiceMock . createClusterClient ( ) . asInternalUser , definitions : taskDefinitions , savedObjectsRepository : savedObjectsClient , } ) ; expect ( await store . getLifecycle ( randomId ( ) ) ) . toEqual ( TaskLifecycleResult . NotFound ) ; } ) ; test ( 'throws if an unknown error takes place ' , async ( ) => { savedObjectsClient . get . mockRejectedValueOnce ( SavedObjectsErrorHelpers . createBadRequestError ( ) ) ; const store = new TaskStore ( { index : 'tasky' , taskManagerId : '' , serializer , esClient : elasticsearchServiceMock . createClusterClient ( ) . asInternalUser , definitions : taskDefinitions , savedObjectsRepository : savedObjectsClient , } ) ; return expect ( store . getLifecycle ( randomId ( ) ) ) . rejects . toThrow ( 'Bad Request' ) ; } ) ; } ) ; } ) ; const asApiResponse = ( body ) => elasticsearchServiceMock . createSuccessTransportRequestPromise ( { hits : body . hits , took : 0 , timed_out : false , _shards : { failed : 0 , successful : body . hits . hits . length , total : 0 , skipped : 0 , } , } ) ; const randomId = ( ) => `id-${ _ . random ( 1 , 20 ) }` ;