import * as Either from 'fp-ts/lib/Either' ; import * as TaskEither from 'fp-ts/lib/TaskEither' ; import { pipe } from 'fp-ts/lib/pipeable' ; import * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey' ; import { AcknowledgeResponse } from './index' ; import { ElasticsearchClient } from '../../../elasticsearch' ; import { IndexMapping } from '../../mappings' ; import { catchRetryableEsClientErrors , RetryableEsClientError , } from './catch_retryable_es_client_errors' ; import { DEFAULT_TIMEOUT , INDEX_AUTO_EXPAND_REPLICAS , WAIT_FOR_ALL_SHARDS_TO_BE_ACTIVE , } from './constants' ; import { waitForIndexStatusYellow } from './wait_for_index_status_yellow' ; function aliasArrayToRecord ( aliases ) { const result = { } ; for ( const alias of aliases ) { result [ alias ] = { } ; } return result ; } internal export interface CreateIndexParams { client : ElasticsearchClient ; indexName : string ; mappings : IndexMapping ; aliases ? : string [ ] ; } remarks export const createIndex = ( { client , indexName , mappings , aliases = [ ] , } ) : TaskEither . TaskEither < RetryableEsClientError , 'create_index_succeeded' > => { const createIndexTask = ( ) => { const aliasesObject = aliasArrayToRecord ( aliases ) ; return client . indices . create ( { index : indexName , wait_for_active_shards : WAIT_FOR_ALL_SHARDS_TO_BE_ACTIVE , timeout : DEFAULT_TIMEOUT , body : { mappings , aliases : aliasesObject , settings : { index : { number_of_shards : 1 , auto_expand_replicas : INDEX_AUTO_EXPAND_REPLICAS , refresh_interval : '1s' , priority : 10 , } , } , } , } , { maxRetries : 0 } ) . then ( ( res ) => { /**
             * - acknowledged=false, we timed out before the cluster state was
             *   updated on all nodes with the newly created index, but it
             *   probably will be created sometime soon.
             * - shards_acknowledged=false, we timed out before all shards were
             *   started
             * - acknowledged=true, shards_acknowledged=true, index creation complete
             */ return Either . right ( { acknowledged : Boolean ( res . body . acknowledged ) , shardsAcknowledged : res . body . shards_acknowledged , } ) ; } ) . catch ( ( error ) => { if ( error ?. body ?. error ?. type === 'resource_already_exists_exception' ) { /**
                 * If the target index already exists it means a previous create
                 * operation had already been started. However, we can't be sure
                 * that all shards were started so return shardsAcknowledged: false
                 */ return Either . right ( { acknowledged : true , shardsAcknowledged : false , } ) ; } else { throw error ; } } ) . catch ( catchRetryableEsClientErrors ) ; } ; return pipe ( createIndexTask , TaskEither . chain ( ( res ) => { if ( res . acknowledged && res . shardsAcknowledged ) { return TaskEither . right ( 'create_index_succeeded' ) ; } else { return pipe ( waitForIndexStatusYellow ( { client , index : indexName , timeout : DEFAULT_TIMEOUT } ) , TaskEither . map ( ( ) => { /** When the index status is 'yellow' we know that all shards were started */ return 'create_index_succeeded' ; } ) ) ; } } ) ) ; } ;