/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 *  contributor license agreements.  The ASF licenses this file to You
 * under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.  For additional information regarding
 * copyright in this work, please see the NOTICE file in the top level
 * directory of this distribution.
 */
package org.apache.usergrid.persistence.index.impl;


import com.codahale.metrics.Meter;
import com.codahale.metrics.Timer;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.inject.Inject;
import com.google.inject.assistedinject.Assisted;
import com.yammer.metrics.core.Clock;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.usergrid.persistence.core.future.BetterFuture;
import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.core.util.Health;
import org.apache.usergrid.persistence.core.util.ValidationUtils;
import org.apache.usergrid.persistence.index.*;
import org.apache.usergrid.persistence.index.exceptions.IndexException;
import org.apache.usergrid.persistence.index.query.CandidateResult;
import org.apache.usergrid.persistence.index.query.CandidateResults;
import org.apache.usergrid.persistence.index.query.Query;
import org.apache.usergrid.persistence.index.utils.UUIDUtils;
import org.apache.usergrid.persistence.map.MapManager;
import org.apache.usergrid.persistence.map.MapManagerFactory;
import org.apache.usergrid.persistence.map.MapScope;
import org.apache.usergrid.persistence.map.impl.MapScopeImpl;
import org.apache.usergrid.persistence.model.entity.Id;
import org.apache.usergrid.persistence.model.entity.SimpleId;
import org.apache.usergrid.persistence.model.util.UUIDGenerator;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.ListenableActionFuture;
import org.elasticsearch.action.ShardOperationFailedException;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.cluster.tasks.PendingClusterTasksRequest;
import org.elasticsearch.action.admin.cluster.tasks.PendingClusterTasksResponse;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexResponse;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateResponse;
import org.elasticsearch.action.deletebyquery.DeleteByQueryResponse;
import org.elasticsearch.action.deletebyquery.IndexDeleteByQueryResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequestBuilder;
import org.elasticsearch.client.AdminClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.*;
import org.elasticsearch.indices.IndexAlreadyExistsException;
import org.elasticsearch.indices.IndexMissingException;
import org.elasticsearch.indices.InvalidAliasNameException;
import org.elasticsearch.rest.action.admin.indices.alias.delete.AliasesMissingException;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rx.Observable;
import rx.functions.Func1;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.apache.usergrid.persistence.index.impl.IndexingUtils.*;


/**
 * Implements index using ElasticSearch Java API.
 */
public class EsEntityIndexImpl implements AliasedEntityIndex {

    private static final Logger logger = LoggerFactory.getLogger( EsEntityIndexImpl.class );

    private static final AtomicBoolean mappingsCreated = new AtomicBoolean( false );
    public static final String DEFAULT_TYPE = "_default_";

    private final IndexIdentifier.IndexAlias alias;
    private final IndexIdentifier indexIdentifier;
    private final IndexBufferProducer indexBatchBufferProducer;
    private final IndexFig indexFig;
    private final Timer addTimer;
    private final Timer addWriteAliasTimer;
    private final Timer addReadAliasTimer;
    private final Timer searchTimer;

    /**
     * We purposefully make this per instance. Some indexes may work, while others may fail
     */
    private FailureMonitor failureMonitor;

    private final ApplicationScope applicationScope;

    private final EsProvider esProvider;

    private final int cursorTimeout;

    private final IndexFig config;

    private final MetricsFactory metricsFactory;


    //number of times to wait for the index to refresh properly.
    private static final int MAX_WAITS = 10;
    //number of milliseconds to try again before sleeping
    private static final int WAIT_TIME = 250;

    private static final String VERIFY_TYPE = "verification";

    private static final ImmutableMap<String, Object> DEFAULT_PAYLOAD =
            ImmutableMap.<String, Object>builder().put( "field", "test" ).put(IndexingUtils.ENTITYID_ID_FIELDNAME, UUIDGenerator.newTimeUUID().toString()).build();

    private static final MatchAllQueryBuilder MATCH_ALL_QUERY_BUILDER = QueryBuilders.matchAllQuery();

    private EsIndexCache aliasCache;
    private Timer removeAliasTimer;
    private Timer mappingTimer;
    private Timer refreshTimer;
    private Timer cursorTimer;
    private Timer getVersionsTimer;
    private Timer allVersionsTimer;
    private Timer deletePreviousTimer;

    private final MapManager mapManager;

//    private final Timer indexTimer;


    @Inject
    public EsEntityIndexImpl( @Assisted final ApplicationScope appScope, final IndexFig config,
                              final IndexBufferProducer indexBatchBufferProducer, final EsProvider provider,
                              final EsIndexCache indexCache, final MetricsFactory metricsFactory,
                              final MapManagerFactory mapManagerFactory, final IndexFig indexFig ) {
        this.indexBatchBufferProducer = indexBatchBufferProducer;
        this.indexFig = indexFig;
        ValidationUtils.validateApplicationScope( appScope );
        this.applicationScope = appScope;
        this.esProvider = provider;
        this.config = config;
        this.cursorTimeout = config.getQueryCursorTimeout();
        this.indexIdentifier = IndexingUtils.createIndexIdentifier(config, appScope);
        this.alias = indexIdentifier.getAlias();
        this.failureMonitor = new FailureMonitorImpl( config, provider );
        this.aliasCache = indexCache;
        this.metricsFactory = metricsFactory;
        this.addTimer = metricsFactory
            .getTimer( EsEntityIndexImpl.class, "es.entity.index.add.index.timer" );
        this.removeAliasTimer = metricsFactory
            .getTimer( EsEntityIndexImpl.class, "es.entity.index.remove.index.alias.timer" );
        this.addReadAliasTimer = metricsFactory
            .getTimer( EsEntityIndexImpl.class, "es.entity.index.add.read.alias.timer" );
        this.addWriteAliasTimer = metricsFactory
            .getTimer( EsEntityIndexImpl.class, "es.entity.index.add.write.alias.timer" );
        this.mappingTimer = metricsFactory
            .getTimer( EsEntityIndexImpl.class, "es.entity.index.create.mapping.timer" );
        this.refreshTimer = metricsFactory
            .getTimer( EsEntityIndexImpl.class, "es.entity.index.refresh.timer" );
        this.searchTimer =metricsFactory
            .getTimer( EsEntityIndexImpl.class, "es.entity.index.search.timer" );
        this.cursorTimer = metricsFactory
            .getTimer( EsEntityIndexImpl.class, "es.entity.index.search.cursor.timer" );
        this.getVersionsTimer =metricsFactory
            .getTimer( EsEntityIndexImpl.class, "es.entity.index.get.versions.timer" );
        this.allVersionsTimer =  metricsFactory
            .getTimer( EsEntityIndexImpl.class, "es.entity.index.delete.all.versions.timer" );
        this.deletePreviousTimer = metricsFactory
            .getTimer( EsEntityIndexImpl.class, "es.entity.index.delete.previous.versions.timer" );

        final MapScope mapScope = new MapScopeImpl( appScope.getApplication(), "cursorcache" );

        mapManager = mapManagerFactory.createMapManager( mapScope );
    }

    @Override
    public void initializeIndex() {
        final int numberOfShards = config.getNumberOfShards();
        final int numberOfReplicas = config.getNumberOfReplicas();
        addIndex(null, numberOfShards, numberOfReplicas,config.getWriteConsistencyLevel());
    }

    @Override
    public void addIndex(final String indexSuffix,final int numberOfShards, final int numberOfReplicas, final String writeConsistency) {
        String normalizedSuffix =  StringUtils.isNotEmpty(indexSuffix) ? indexSuffix : null;
        try {

            //get index name with suffix attached
            String indexName = indexIdentifier.getIndex(normalizedSuffix);

            //Create index
            try {
                final AdminClient admin = esProvider.getClient().admin();
                Settings settings = ImmutableSettings.settingsBuilder()
                        .put("index.number_of_shards", numberOfShards)
                        .put("index.number_of_replicas", numberOfReplicas)
                        .put("action.write_consistency", writeConsistency )
                    .build();

                //Added For Graphite Metrics
                Timer.Context timeNewIndexCreation = addTimer.time();
                final CreateIndexResponse cir = admin.indices().prepareCreate(indexName)
                        .setSettings(settings)
                        .execute()
                        .actionGet();
                timeNewIndexCreation.stop();

                //create the mappings
                createMappings( indexName );

                //ONLY add the alias if we create the index, otherwise we're going to overwrite production settings

                /**
                 * DO NOT MOVE THIS LINE OF CODE UNLESS YOU REALLY KNOW WHAT YOU'RE DOING!!!!
                 */

                //We do NOT want to create an alias if the index already exists, we'll overwrite the indexes that
                //may have been set via other administrative endpoint

                addAlias(normalizedSuffix);

                testNewIndex();

                logger.info("Created new Index Name [{}] ACK=[{}]", indexName, cir.isAcknowledged());
            } catch (IndexAlreadyExistsException e) {
                logger.info("Index Name [{}] already exists", indexName);
            }


        } catch (IndexAlreadyExistsException expected) {
            // this is expected to happen if index already exists, it's a no-op and swallow
        } catch (IOException e) {
            throw new RuntimeException("Unable to initialize index", e);
        }
    }


    @Override
    public void addAlias(final String indexSuffix) {
        try {
            Boolean isAck;
            String indexName = indexIdentifier.getIndex(indexSuffix);
            final AdminClient adminClient = esProvider.getClient().admin();

            String[] indexNames = getIndexes(AliasType.Write);

            for ( String currentIndex : indexNames ) {

                final Timer.Context timeRemoveAlias = removeAliasTimer.time();

                try {
                    //Added For Graphite Metrics

                    isAck = adminClient.indices().prepareAliases().removeAlias( currentIndex, alias.getWriteAlias() )
                                       .execute().actionGet().isAcknowledged();

                    logger.info( "Removed Index Name [{}] from Alias=[{}] ACK=[{}]", currentIndex, alias, isAck );
                }
                catch ( AliasesMissingException aie ) {
                    logger.info( "Alias does not exist Index Name [{}] from Alias=[{}] ACK=[{}]", currentIndex, alias,
                        aie.getMessage() );
                    continue;
                }
                catch ( InvalidAliasNameException iane ) {
                    logger.info( "Alias does not exist Index Name [{}] from Alias=[{}] ACK=[{}]", currentIndex, alias,
                        iane.getMessage() );
                    continue;
                }
                finally {
                    timeRemoveAlias.stop();
                }
            }

            //Added For Graphite Metrics
            Timer.Context timeAddReadAlias = addReadAliasTimer.time();
            // add read alias
            isAck = adminClient.indices().prepareAliases().addAlias(
                    indexName, alias.getReadAlias()).execute().actionGet().isAcknowledged();
            timeAddReadAlias.stop();
            logger.info("Created new read Alias Name [{}] ACK=[{}]", alias.getReadAlias(), isAck);

            //Added For Graphite Metrics
            Timer.Context timeAddWriteAlias = addWriteAliasTimer.time();
            //add write alias
            isAck = adminClient.indices().prepareAliases().addAlias(
                    indexName, alias.getWriteAlias()).execute().actionGet().isAcknowledged();
            timeAddWriteAlias.stop();
            logger.info("Created new write Alias Name [{}] ACK=[{}]", alias.getWriteAlias(), isAck);

            aliasCache.invalidate(alias);

        } catch (Exception e) {
            logger.warn("Failed to create alias ", e);
        }
    }

    @Override
    public String[] getIndexes(final AliasType aliasType) {
        return aliasCache.getIndexes(alias, aliasType);
    }


    /**
     * Tests writing a document to a new index to ensure it's working correctly. See this post:
     * http://s.apache.org/index-missing-exception
     */
    private void testNewIndex() {

        // create the document, this ensures the index is ready
        // Immediately create a document and remove it to ensure the entire cluster is ready
        // to receive documents. Occasionally we see errors.
        // See this post: http://s.apache.org/index-missing-exception

        logger.debug( "Testing new index name: read {} write {}", alias.getReadAlias(), alias.getWriteAlias());

        final RetryOperation retryOperation = new RetryOperation() {
            @Override
            public boolean doOp() {
                final String tempId = UUIDGenerator.newTimeUUID().toString();

                esProvider.getClient().prepareIndex( alias.getWriteAlias(), VERIFY_TYPE, tempId )
                     .setSource(DEFAULT_PAYLOAD).get();

                logger.info( "Successfully created new document with docId {} "
                     + "in index read {} write {} and type {}",
                        tempId, alias.getReadAlias(), alias.getWriteAlias(), VERIFY_TYPE );

                // delete all types, this way if we miss one it will get cleaned up
                esProvider.getClient().prepareDeleteByQuery( alias.getWriteAlias())
                        .setTypes(VERIFY_TYPE)
                        .setQuery(MATCH_ALL_QUERY_BUILDER).get();

                logger.info( "Successfully deleted all documents in index {} read {} write {} and type {}",
                        alias.getReadAlias(), alias.getWriteAlias(), VERIFY_TYPE );

                return true;
            }
        };

        doInRetry( retryOperation );
    }


    /**
     * Setup ElasticSearch type mappings as a template that applies to all new indexes.
     * Applies to all indexes that* start with our prefix.
     */
    private void createMappings(final String indexName) throws IOException {

        XContentBuilder xcb = IndexingUtils.createDoubleStringIndexMapping(
            XContentFactory.jsonBuilder(), DEFAULT_TYPE );


        //Added For Graphite Metrics
        Timer.Context timePutIndex = mappingTimer.time();
        PutMappingResponse  pitr = esProvider.getClient().admin().indices().preparePutMapping( indexName ).setType(
            DEFAULT_TYPE ).setSource( xcb ).execute().actionGet();
        timePutIndex.stop();
        if ( !pitr.isAcknowledged() ) {
            throw new IndexException( "Unable to create default mappings" );
        }
    }


    @Override
    public EntityIndexBatch createBatch() {
        EntityIndexBatch batch = new EsEntityIndexBatchImpl(
                applicationScope, esProvider.getClient(),indexBatchBufferProducer, config, this, metricsFactory );
        return batch;
    }


    @Override
    public CandidateResults search( final IndexScope indexScope, final SearchTypes searchTypes,
            final Query query ) {

        final String context = IndexingUtils.createContextName(indexScope);
        final String[] entityTypes = searchTypes.getTypeNames();

        QueryBuilder qb = query.createQueryBuilder(context);


        SearchResponse searchResponse;

        if ( query.getCursor() == null ) {
            SearchRequestBuilder srb = esProvider.getClient().prepareSearch( alias.getReadAlias() )
                    .setTypes(entityTypes)
                    .setScroll(cursorTimeout + "m")
                    .setQuery(qb);

            final FilterBuilder fb = query.createFilterBuilder();

            //we have post filters, apply them
            if ( fb != null ) {
                logger.debug( "   Filter: {} ", fb.toString() );
                srb = srb.setPostFilter( fb );
            }


            srb = srb.setFrom( 0 ).setSize( query.getLimit() );

            for ( Query.SortPredicate sp : query.getSortPredicates() ) {

                final SortOrder order;
                if ( sp.getDirection().equals( Query.SortDirection.ASCENDING ) ) {
                    order = SortOrder.ASC;
                }
                else {
                    order = SortOrder.DESC;
                }

                // we do not know the type of the "order by" property and so we do not know what
                // type prefix to use. So, here we add an order by clause for every possible type
                // that you can order by: string, number and boolean and we ask ElasticSearch
                // to ignore any fields that are not present.

                final String stringFieldName = STRING_PREFIX + sp.getPropertyName();
                final FieldSortBuilder stringSort = SortBuilders.fieldSort( stringFieldName )
                        .order( order ).ignoreUnmapped( true );
                srb.addSort( stringSort );

                logger.debug( "   Sort: {} order by {}", stringFieldName, order.toString() );

                final String numberFieldName = NUMBER_PREFIX + sp.getPropertyName();
                final FieldSortBuilder numberSort = SortBuilders.fieldSort( numberFieldName )
                        .order( order ).ignoreUnmapped( true );
                srb.addSort( numberSort );
                logger.debug( "   Sort: {} order by {}", numberFieldName, order.toString() );

                final String booleanFieldName = BOOLEAN_PREFIX + sp.getPropertyName();
                final FieldSortBuilder booleanSort = SortBuilders.fieldSort( booleanFieldName )
                        .order( order ).ignoreUnmapped( true );
                srb.addSort( booleanSort );
                logger.debug( "   Sort: {} order by {}", booleanFieldName, order.toString() );
            }


            if ( logger.isDebugEnabled() ) {
                logger.debug( "Searching index (read alias): {}\n  scope: {} \n type: {}\n   query: {} ",
                    this.alias.getReadAlias(), context, entityTypes, srb );
            }

            try {
                //Added For Graphite Metrics
                Timer.Context timeSearch = searchTimer.time();
                searchResponse = srb.execute().actionGet();
                timeSearch.stop();
            }
            catch ( Throwable t ) {
                logger.error( "Unable to communicate with Elasticsearch", t );
                failureMonitor.fail( "Unable to execute batch", t );
                throw t;
            }


            failureMonitor.success();
        }
        else {
            String userCursorString = query.getCursor();
            if ( userCursorString.startsWith( "\"" ) ) {
                userCursorString = userCursorString.substring( 1 );
            }
            if ( userCursorString.endsWith( "\"" ) ) {
                userCursorString = userCursorString.substring( 0, userCursorString.length() - 1 );
            }

            //now get the cursor from the map  and validate
            final String esScrollCursor  = mapManager.getString( userCursorString );

            Preconditions.checkArgument(esScrollCursor != null, "Could not find a cursor for the value '{}' ",  esScrollCursor);


            logger.debug( "Executing query with cursor: {} ", esScrollCursor );


            SearchScrollRequestBuilder ssrb = esProvider.getClient()
                    .prepareSearchScroll(esScrollCursor).setScroll( cursorTimeout + "m" );

            try {
                //Added For Graphite Metrics
                Timer.Context timeSearchCursor = cursorTimer.time();
                searchResponse = ssrb.execute().actionGet();
                timeSearchCursor.stop();
            }
            catch ( Throwable t ) {
                logger.error( "Unable to communicate with elasticsearch", t );
                failureMonitor.fail( "Unable to execute batch", t );
                throw t;
            }


            failureMonitor.success();
        }

        return parseResults(searchResponse, query);
    }


    private CandidateResults parseResults( final SearchResponse searchResponse, final Query query ) {

        final SearchHits searchHits = searchResponse.getHits();
        final SearchHit[] hits = searchHits.getHits();
        final int length = hits.length;

        logger.debug("   Hit count: {} Total hits: {}", length, searchHits.getTotalHits());

        List<CandidateResult> candidates = new ArrayList<>( length );

        for ( SearchHit hit : hits ) {

            String[] idparts = hit.getId().split( SPLITTER );
            String id = idparts[0];
            String type = idparts[1];
            String version = idparts[2];

            Id entityId = new SimpleId( UUID.fromString( id ), type );

            candidates.add( new CandidateResult( entityId, UUID.fromString( version ) ) );
        }

        CandidateResults candidateResults = new CandidateResults( query, candidates );

        if ( candidates.size() >= query.getLimit() ) {
            //USERGRID-461 our cursor is getting too large, map it to a new time UUID

            final String userCursorString = org.apache.usergrid.persistence.index.utils.StringUtils.sanitizeUUID( UUIDGenerator.newTimeUUID() );

            final String esScrollCursor = searchResponse.getScrollId();

            //now set this into our map module
            final int minutes = indexFig.getQueryCursorTimeout();

            //just truncate it, we'll never hit a long value anyway
            mapManager.putString( userCursorString, esScrollCursor, ( int ) TimeUnit.MINUTES.toSeconds( minutes ) );

            candidateResults.setCursor( userCursorString );
            logger.debug(" User cursor = {},  Cursor = {} ", userCursorString, esScrollCursor);
        }

        return candidateResults;
    }


    public void refresh() {

        BetterFuture future = indexBatchBufferProducer.put(new IndexOperationMessage());
        future.get();
        //loop through all batches and retrieve promises and call get

        final RetryOperation retryOperation = new RetryOperation() {
            @Override
            public boolean doOp() {
                try {
                    String[] indexes = ArrayUtils.addAll(
                        getIndexes(AliasType.Read),
                        getIndexes(AliasType.Write)
                    );

                    if ( indexes.length == 0 ) {
                        logger.debug( "Not refreshing indexes, none found for app {}",
                                applicationScope.getApplication().getUuid() );
                        return true;
                    }
                    //Added For Graphite Metrics
                    Timer.Context timeRefreshIndex = refreshTimer.time();
                    esProvider.getClient().admin().indices().prepareRefresh( indexes ).execute().actionGet();
                    timeRefreshIndex.stop();
                    logger.debug("Refreshed indexes: {}", StringUtils.join(indexes, ", "));
                    return true;
                }
                catch ( IndexMissingException e ) {
                    logger.error( "Unable to refresh index. Waiting before sleeping.", e );
                    throw e;
                }
            }
        };

        doInRetry( retryOperation );
    }


    @Override
    public int getPendingTasks() {

        final PendingClusterTasksResponse tasksResponse = esProvider.getClient().admin()
                .cluster().pendingClusterTasks(new PendingClusterTasksRequest()).actionGet();

        return tasksResponse.pendingTasks().size();
    }


    @Override
    public CandidateResults getEntityVersions( final IndexScope scope, final Id id ) {

        //since we don't have paging inputs, there's no point in executing a query for paging.

        final String context = IndexingUtils.createContextName(scope);
        final SearchTypes searchTypes = SearchTypes.fromTypes(id.getType());

        final QueryBuilder queryBuilder =
                QueryBuilders.termQuery( IndexingUtils.ENTITY_CONTEXT_FIELDNAME, context );

        final SearchRequestBuilder srb = esProvider.getClient().prepareSearch( alias.getReadAlias() )
                .setTypes(searchTypes.getTypeNames())
                .setScroll(cursorTimeout + "m")
                .setQuery(queryBuilder);

        final SearchResponse searchResponse;
        try {
            //Added For Graphite Metrics
            Timer.Context timeEntityIndex = getVersionsTimer.time();
            searchResponse = srb.execute().actionGet();
            timeEntityIndex.stop();
        }
        catch ( Throwable t ) {
            logger.error( "Unable to communicate with elasticsearch" );
            failureMonitor.fail( "Unable to execute batch", t );
            throw t;
        }


        failureMonitor.success();

        return parseResults(searchResponse, new Query());
    }


    @Override
    public ListenableActionFuture deleteAllVersionsOfEntity(final Id entityId ) {

        String idString = IndexingUtils.idString(entityId).toLowerCase();

        final TermQueryBuilder tqb = QueryBuilders.termQuery(ENTITYID_ID_FIELDNAME, idString);

        //Added For Graphite Metrics
        final Timer.Context timeDeleteAllVersions =allVersionsTimer.time();
        final ListenableActionFuture<DeleteByQueryResponse> response = esProvider.getClient()
            .prepareDeleteByQuery( alias.getWriteAlias() ).setQuery( tqb ).execute();

        response.addListener( new ActionListener<DeleteByQueryResponse>() {

            @Override
            public void onResponse( DeleteByQueryResponse response) {
                timeDeleteAllVersions.stop();
                logger
                    .debug( "Deleted entity {}:{} from all index scopes with response status = {}", entityId.getType(),
                        entityId.getUuid(), response.status().toString() );

                checkDeleteByQueryResponse(tqb, response);
            }


            @Override
            public void onFailure( Throwable e ) {
                logger.error( "Deleted entity {}:{} from all index scopes with error {}", entityId.getType(),
                    entityId.getUuid(), e);


            }
        });
        return response;
    }


    @Override
    public ListenableActionFuture deletePreviousVersions( final Id entityId, final UUID version ) {

        String idString = IndexingUtils.idString( entityId ).toLowerCase();

        final FilteredQueryBuilder fqb = QueryBuilders.filteredQuery(
                QueryBuilders.termQuery(ENTITYID_ID_FIELDNAME, idString),
                FilterBuilders.rangeFilter(ENTITY_VERSION_FIELDNAME).lt(version.timestamp())
        );

        //Added For Graphite Metrics
        //Checks the time from the execute to the response below
        final Timer.Context timeDeletePreviousVersions = deletePreviousTimer.time();
        final ListenableActionFuture<DeleteByQueryResponse> response = esProvider.getClient()
            .prepareDeleteByQuery(alias.getWriteAlias()).setQuery(fqb).execute();

        //Added For Graphite Metrics
        response.addListener( new ActionListener<DeleteByQueryResponse>() {
            @Override
            public void onResponse( DeleteByQueryResponse response ) {
                timeDeletePreviousVersions.stop();
                //error message needs to be retooled so that it describes the entity more throughly
                logger
                    .debug( "Deleted entity {}:{} with version {} from all " + "index scopes with response status = {}",
                        entityId.getType(), entityId.getUuid(), version, response.status().toString() );

                checkDeleteByQueryResponse( fqb, response );
            }


            @Override
            public void onFailure( Throwable e ) {
                logger.error( "Deleted entity {}:{} from all index scopes with error {}", entityId.getType(),
                    entityId.getUuid(), e );
            }
        } );


        return response;
    }


    /**
     * Validate the response doesn't contain errors, if it does, fail fast at the first error we encounter
     */
    private void checkDeleteByQueryResponse(
            final QueryBuilder query, final DeleteByQueryResponse response ) {

        for ( IndexDeleteByQueryResponse indexDeleteByQueryResponse : response ) {
            final ShardOperationFailedException[] failures = indexDeleteByQueryResponse.getFailures();

            for ( ShardOperationFailedException failedException : failures ) {
                logger.error( String.format("Unable to delete by query %s. "
                        + "Failed with code %d and reason %s on shard %s in index %s",
                    query.toString(),
                    failedException.status().getStatus(),
                    failedException.reason(),
                    failedException.shardId(),
                    failedException.index() )
                );
            }

        }
    }


    /**
     * Completely delete an index.
     */
    public void deleteIndex() {
        AdminClient adminClient = esProvider.getClient().admin();

        DeleteIndexResponse response = adminClient.indices()
                .prepareDelete( indexIdentifier.getIndex(null) ).get();

        if ( response.isAcknowledged() ) {
            logger.info( "Deleted index: read {} write {}", alias.getReadAlias(), alias.getWriteAlias());
            //invlaidate the alias
            aliasCache.invalidate(alias);
        }
        else {
            logger.info( "Failed to delete index: read {} write {}", alias.getReadAlias(), alias.getWriteAlias());
        }
    }


    /**
     * Do the retry operation
     */
    private void doInRetry( final RetryOperation operation ) {
        for ( int i = 0; i < MAX_WAITS; i++ ) {

            try {
                if ( operation.doOp() ) {
                    return;
                }
            }
            catch ( Exception e ) {
                logger.error( "Unable to execute operation, retrying", e );
            }


            try {
                Thread.sleep( WAIT_TIME );
            }
            catch ( InterruptedException e ) {
                //swallow it
            }
        }
    }


    /**
     * Check health of cluster.
     */
    @Override
    public Health getClusterHealth() {

        try {
            ClusterHealthResponse chr = esProvider.getClient().admin()
                    .cluster().health(new ClusterHealthRequest()).get();
            return Health.valueOf( chr.getStatus().name() );
        }
        catch ( Exception ex ) {
            logger.error( "Error connecting to ElasticSearch", ex );
        }

        // this is bad, red alert!
        return Health.RED;
    }


    /**
     * Check health of this specific index.
     */
    @Override
    public Health getIndexHealth() {

        try {
            ClusterHealthResponse chr = esProvider.getClient().admin().cluster().health(
                    new ClusterHealthRequest(new String[]{indexIdentifier.getIndex(null)})).get();
            return Health.valueOf( chr.getStatus().name() );
        }
        catch ( Exception ex ) {
            logger.error( "Error connecting to ElasticSearch", ex );
        }

        // this is bad, red alert!
        return Health.RED;
    }


    /**
     * Interface for operations.
     */
    private static interface RetryOperation {

        /**
         * Return true if done, false if there should be a retry.
         */
        public boolean doOp();
    }
}