/* 
 * Copyright (C) 2003-2015 eXo Platform SAS.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program. If not, see http://www.gnu.org/licenses/ .
 */
package org.exoplatform.commons.search.index.impl;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.lang3.StringUtils;

import org.exoplatform.commons.api.persistence.ExoTransactional;
import org.exoplatform.container.ExoContainer;
import org.exoplatform.container.ExoContainerContext;
import org.picocontainer.Startable;

import org.exoplatform.commons.persistence.impl.EntityManagerService;
import org.exoplatform.commons.search.dao.IndexingOperationDAO;
import org.exoplatform.commons.search.domain.IndexingOperation;
import org.exoplatform.commons.search.domain.OperationType;
import org.exoplatform.commons.search.es.client.*;
import org.exoplatform.commons.search.index.IndexingOperationProcessor;
import org.exoplatform.commons.search.index.IndexingServiceConnector;
import org.exoplatform.commons.utils.PropertyManager;
import org.exoplatform.container.xml.InitParams;
import org.exoplatform.services.log.ExoLogger;
import org.exoplatform.services.log.Log;

/**
 * Created by The eXo Platform SAS Author : Thibault Clement
 * tclement@exoplatform.com 10/12/15
 */
public class ElasticIndexingOperationProcessor extends IndexingOperationProcessor implements Startable {

  private static final Log                   LOG                                 =
                                                 ExoLogger.getExoLogger(ElasticIndexingOperationProcessor.class);

  private static final String                BATCH_NUMBER_PROPERTY_NAME          = "exo.es.indexing.batch.number";

  private static final Integer               BATCH_NUMBER_DEFAULT                = 1000;

  private static final String                REQUEST_SIZE_LIMIT_PROPERTY_NAME    = "exo.es.indexing.request.size.limit";

  /** in bytes, default=10MB **/
  private static final Integer               REQUEST_SIZE_LIMIT_DEFAULT          = 10485760;

  private static final String                REINDEXING_BATCH_SIZE_PROPERTY_NAME = "exo.es.reindex.batch.size";

  private static final int                   REINDEXING_BATCH_SIZE_DEFAULT_VALUE = 100;

  private int                                reindexBatchSize                    = REINDEXING_BATCH_SIZE_DEFAULT_VALUE;

  // Service
  private final IndexingOperationDAO         indexingOperationDAO;

  private final ElasticIndexingClient        elasticIndexingClient;

  private final ElasticContentRequestBuilder elasticContentRequestBuilder;

  private final ElasticIndexingAuditTrail    auditTrail;

  private final EntityManagerService         entityManagerService;

  private Integer                            batchNumber                         = BATCH_NUMBER_DEFAULT;

  private Integer                            requestSizeLimit                    = REQUEST_SIZE_LIMIT_DEFAULT;

  private Map<String, Set<String>>           indexUpgrading                      = new HashMap<>();

  private ExecutorService                    executors                           = Executors.newCachedThreadPool();

  private String                             esVersion;

  private boolean                            interrupted                         = false;

  private boolean                            initialized                         = false;

  public ElasticIndexingOperationProcessor(IndexingOperationDAO indexingOperationDAO,
                                           ElasticIndexingClient elasticIndexingClient,
                                           ElasticContentRequestBuilder elasticContentRequestBuilder,
                                           ElasticIndexingAuditTrail auditTrail,
                                           EntityManagerService entityManagerService,
                                           InitParams initParams) {
    this.indexingOperationDAO = indexingOperationDAO;
    this.auditTrail = auditTrail;
    this.entityManagerService = entityManagerService;
    this.elasticIndexingClient = elasticIndexingClient;
    this.elasticContentRequestBuilder = elasticContentRequestBuilder;
    if (StringUtils.isNotBlank(PropertyManager.getProperty(BATCH_NUMBER_PROPERTY_NAME))) {
      this.batchNumber = Integer.valueOf(PropertyManager.getProperty(BATCH_NUMBER_PROPERTY_NAME));
    }
    if (StringUtils.isNotBlank(PropertyManager.getProperty(REQUEST_SIZE_LIMIT_PROPERTY_NAME))) {
      this.requestSizeLimit = Integer.valueOf(PropertyManager.getProperty(REQUEST_SIZE_LIMIT_PROPERTY_NAME));
    }
    if (StringUtils.isNotBlank(PropertyManager.getProperty(REINDEXING_BATCH_SIZE_PROPERTY_NAME))) {
      this.reindexBatchSize = Integer.valueOf(PropertyManager.getProperty(REINDEXING_BATCH_SIZE_PROPERTY_NAME));
    }
    if (initParams == null || !initParams.containsKey("es.version")) {
      throw new IllegalStateException("es.version parameter is mandatory");
    }
    this.esVersion = initParams.getValueParam("es.version").getValue();
    LOG.info("Use ES Version {}", esVersion);
  }

  @Override
  public void addConnector(IndexingServiceConnector indexingServiceConnector) {
    addConnector(indexingServiceConnector, false);
  }

  @Override
  public void addConnector(IndexingServiceConnector indexingServiceConnector, Boolean override) {
    if (getConnectors().containsKey(indexingServiceConnector.getConnectorName()) && override.equals(false)) {
      LOG.error("Impossible to add connector {}. A connector with the same name has already been registered.",
                indexingServiceConnector.getConnectorName());
    } else {
      getConnectors().put(indexingServiceConnector.getConnectorName(), indexingServiceConnector);
      LOG.info("An Indexing Connector has been added: {}", indexingServiceConnector.getConnectorName());
    }
  }

  /**
   * Handle the Indexing queue Get all data in the indexing queue, transform
   * them to ES requests, send requests to ES This method is ONLY called by the
   * job scheduler. This method is not annotated with @ExoTransactional because
   * we don't want it to be executed in one transaction. A request lifecycle is
   * started and ended for all jobs, it is done by
   * org.exoplatform.services.scheduler.impl.JobEnvironmentConfigListener. It
   * means that we have 1 entity manager per job execution. Because of that, we
   * have to take care of cleaning the persistence context regularly to avoid to
   * have too big sessions and bad performances. This method is synchronized to
   * make sure the queue is processed by only one thread at a time, since the
   * indexing queue does not support multi-thread processing for the moment.
   */
  @Override
  public synchronized void process() {
    if (!this.initialized) {
      LOG.debug("Skip ES queue processing until service is properly initialized");
      return;
    }
    this.interrupted = false;
    try {
      // Loop until the number of data retrieved from indexing queue is less
      // than
      // BATCH_NUMBER (default = 1000)
      int processedOperations;
      do {
        processedOperations = processBulk();
      } while (processedOperations >= batchNumber);
    } finally {
      if (this.interrupted) {
        LOG.info("Indexing queue processing interruption done");
      }
    }
  }

  /**
   * Set the indexing process as interrupted in order to terminate it as soon as
   * possible without finishing the whole process. Since the indexing process
   * can take time (for a reindexAll operation for example), it allows to
   * interrupt it gracefully (without killing the thread).
   */
  @Override
  public void interrupt() {
    LOG.info("Indexing queue processing has been interrupted. Please wait until the service exists cleanly...");
    this.interrupted = true;
  }

  private boolean isInterrupted() {
    if (Thread.currentThread().isInterrupted()) {
      LOG.info("Thread running indexing queue processing has been interrupted. Please wait until the service exists cleanly...");
      this.interrupted = true;
    }
    return this.interrupted;
  }

  private boolean isUpgradeInProgress() {
    return !indexUpgrading.isEmpty();
  }
  private int processBulk() {
    // Choose operation to delete from Queue one by one instead
    if (isUpgradeInProgress()) {
      LOG.info("Migration of indexes is in progress, indexation is suspended until migration finishes");
      return 0;
    }

    Map<OperationType, Map<String, List<IndexingOperation>>> indexingQueueSorted = new EnumMap<>(OperationType.class);
    List<IndexingOperation> indexingOperations;
    long maxIndexingOperationId = 0;

    // Get BATCH_NUMBER (default = 1000) first indexing operations
    indexingOperations = indexingOperationDAO.findAllFirst(batchNumber);
    if (indexingOperations.isEmpty()) {
      return 0;
    }

    // Get all Indexing operations and order them per operation and type in map:
    // <Operation, <Type, List<IndexingOperation>>>
    for (IndexingOperation indexingOperation : indexingOperations) {
      putIndexingOperationInMemoryQueue(indexingOperation, indexingQueueSorted);
      // Get the max ID of IndexingOperation of the bulk
      if (maxIndexingOperationId < indexingOperation.getId()) {
        maxIndexingOperationId = indexingOperation.getId();
      }
    }

    processInit(indexingQueueSorted);
    processDeleteAll(indexingQueueSorted);
    processCUD(indexingQueueSorted);

    if (isInterrupted()) {
      throw new IllegalStateException("Indexing queue processing interrupted");
    }

    // Removes the processed IDs from the “indexing queue” table that have
    // timestamp older than the timestamp of
    // start of processing
    indexingOperationDAO.deleteAllIndexingOperationsHavingIdLessThanOrEqual(maxIndexingOperationId);

    // clear entity manager content after each bulk
    entityManagerService.getEntityManager().clear();

    return indexingOperations.size();
  }

  /**
   * Add an indexing operation to the Temporary inMemory IndexingQueue
   * 
   * @param indexingOperation the operation to add to the Temporary inMemory
   *          IndexingQueue
   * @param indexingQueueSorted Temporary inMemory IndexingQueue
   */
  private void putIndexingOperationInMemoryQueue(IndexingOperation indexingOperation,
                                                 Map<OperationType, Map<String, List<IndexingOperation>>> indexingQueueSorted) {
    // Check if the Indexing Operation map already contains a specific operation
    if (!indexingQueueSorted.containsKey(indexingOperation.getOperation())) {
      // If not add a new operation in the map
      indexingQueueSorted.put(indexingOperation.getOperation(), new HashMap<>());
    }
    // Check if the operation map already contains a specific type
    if (!indexingQueueSorted.get(indexingOperation.getOperation()).containsKey(indexingOperation.getEntityIndex())) {
      // If not add a new type for the operation above
      indexingQueueSorted.get(indexingOperation.getOperation())
                         .put(indexingOperation.getEntityIndex(),
                              new ArrayList<>());
    }
    // Add the indexing operation in the specific Operation -> Type
    indexingQueueSorted.get(indexingOperation.getOperation()).get(indexingOperation.getEntityIndex()).add(indexingOperation);
  }

  /**
   * Process all the Create / Update / Delete operations
   * 
   * @param indexingQueueSorted Temporary inMemory IndexingQueue
   */
  private void processCUD(Map<OperationType, Map<String, List<IndexingOperation>>> indexingQueueSorted) {
    // Initialize bulk request for CUD operations
    StringBuilder bulkRequest = new StringBuilder();

    // Process Delete document operation
    if (indexingQueueSorted.containsKey(OperationType.DELETE)) {
      Map<String, List<IndexingOperation>> deleteIndexingOperationsMap = indexingQueueSorted.get(OperationType.DELETE);
      for (String entityIndex : deleteIndexingOperationsMap.keySet()) {
        ElasticIndexingServiceConnector connector = (ElasticIndexingServiceConnector) getConnectors().get(entityIndex);
        if (connector == null) {
          continue;
        }
        List<IndexingOperation> deleteIndexingOperationsList = deleteIndexingOperationsMap.get(entityIndex);
        if (deleteIndexingOperationsList == null || deleteIndexingOperationsList.isEmpty()) {
          continue;
        }
        Iterator<IndexingOperation> deleteIndexingOperationsIterator = deleteIndexingOperationsList.iterator();
        while (deleteIndexingOperationsIterator.hasNext()) {
          if (isInterrupted()) {
            return;
          }
          IndexingOperation deleteIndexQueue = deleteIndexingOperationsIterator.next();
          try {
            String deleteDocumentRequestContent = elasticContentRequestBuilder.getDeleteDocumentRequestContent(connector,
                                                                                                               deleteIndexQueue.getEntityId());
            if (deleteDocumentRequestContent != null) {
              bulkRequest.append(deleteDocumentRequestContent);
            }
            // Choose operation to delete from Queue one by one instead
          } catch (Exception e) {
            LOG.warn("Error while *deleting* index entry of entity, type = " + entityIndex + ", id ="
                + (deleteIndexQueue == null ? null : deleteIndexQueue.getEntityId()) + ", cause:", e);
          } finally {
            // Remove the delete operations from the map
            indexingQueueSorted.remove(OperationType.DELETE, deleteIndexQueue.getEntityId());
          }

          // Delete added indexation operation from queue even if the request
          // fails
          deleteIndexingOperationsIterator.remove();

          // Remove the object from other create or update operations planned
          // before the timestamp of the delete operation
          deleteOperationsByEntityIdForTypesBefore(new OperationType[] { OperationType.CREATE },
                                                   indexingQueueSorted,
                                                   deleteIndexQueue);
          deleteOperationsByEntityIdForTypes(new OperationType[] { OperationType.UPDATE }, indexingQueueSorted, deleteIndexQueue);
          // Check if the bulk request limit size is already reached
          bulkRequest = checkBulkRequestSizeReachedLimitation(bulkRequest);
        }
      }
    }

    // Process Create document operation
    if (indexingQueueSorted.containsKey(OperationType.CREATE)) {
      Map<String, List<IndexingOperation>> createIndexingOperationsMap = indexingQueueSorted.get(OperationType.CREATE);
      for (String entityIndex : createIndexingOperationsMap.keySet()) {
        ElasticIndexingServiceConnector connector = (ElasticIndexingServiceConnector) getConnectors().get(entityIndex);
        if (connector == null) {
          continue;
        }
        List<IndexingOperation> createIndexingOperationsList = createIndexingOperationsMap.get(entityIndex);
        if (createIndexingOperationsList == null || createIndexingOperationsList.isEmpty()) {
          continue;
        }
        Iterator<IndexingOperation> createIndexingOperationsIterator = createIndexingOperationsList.iterator();
        while (createIndexingOperationsIterator.hasNext()) {
          if (isInterrupted()) {
            return;
          }
          IndexingOperation createIndexQueue = createIndexingOperationsIterator.next();
          try {
            if (connector.isNeedIngestPipeline()) {
              String singleRequestOperation =
                                            elasticContentRequestBuilder.getCreatePipelineDocumentRequestContent(connector,
                                                                                                                 createIndexQueue.getEntityId());
              if (singleRequestOperation != null) {
                elasticIndexingClient.sendCreateDocOnPipeline(connector.getIndexAlias(),
                                                              createIndexQueue.getEntityId(),
                                                              connector.getPipelineName(),
                                                              singleRequestOperation);
              }
              // Delete this single operation since it's not indexed in bulk
              indexingQueueSorted.remove(OperationType.CREATE, createIndexQueue.getEntityId());
              indexingQueueSorted.remove(OperationType.UPDATE, createIndexQueue.getEntityId());
            } else {
              String singleRequestOperation =
                                            elasticContentRequestBuilder.getCreateDocumentRequestContent(connector,
                                                                                                         createIndexQueue.getEntityId());
              if (singleRequestOperation != null) {
                bulkRequest.append(singleRequestOperation);
              }
            }
          } catch (Exception e) {
            LOG.warn("Error while *creating* index entry of entity, type = " + entityIndex + ", id ="
                + (createIndexQueue == null ? null : createIndexQueue.getEntityId()) + ", cause:", e);
          } finally {

            // Delete added indexation operation from queue even if the request
            // fails
            createIndexingOperationsIterator.remove();

            // Remove the object from other update operations for this entityId
            deleteOperationsByEntityIdForTypes(new OperationType[] { OperationType.UPDATE },
                                               indexingQueueSorted,
                                               createIndexQueue);

            // Delete this single operation since it's not indexed in bulk
            indexingQueueSorted.remove(OperationType.CREATE, createIndexQueue.getEntityId());
          }

          // Check if the bulk request limit size is already reached
          bulkRequest = checkBulkRequestSizeReachedLimitation(bulkRequest);
        }
      }
    }

    // Process Update document operation
    if (indexingQueueSorted.containsKey(OperationType.UPDATE)) {
      Map<String, List<IndexingOperation>> updateIndexingOperationsMap = indexingQueueSorted.get(OperationType.UPDATE);
      for (String entityIndex : updateIndexingOperationsMap.keySet()) {
        ElasticIndexingServiceConnector connector = (ElasticIndexingServiceConnector) getConnectors().get(entityIndex);
        if (connector == null) {
          continue;
        }
        List<IndexingOperation> updateIndexingOperationsList = updateIndexingOperationsMap.get(entityIndex);
        if (updateIndexingOperationsList == null || updateIndexingOperationsList.isEmpty()) {
          continue;
        }
        Iterator<IndexingOperation> updateIndexingOperationsIterator = updateIndexingOperationsList.iterator();
        while (updateIndexingOperationsIterator.hasNext()) {
          if (isInterrupted()) {
            return;
          }
          IndexingOperation updateIndexQueue = updateIndexingOperationsIterator.next();
          try {
            if (connector.isNeedIngestPipeline()) {
              String singleRequestOperation =
                                            elasticContentRequestBuilder.getCreatePipelineDocumentRequestContent(connector,
                                                                                                                 updateIndexQueue.getEntityId());
              if (singleRequestOperation != null) {
                elasticIndexingClient.sendCreateDocOnPipeline(connector.getIndexAlias(),
                                                              updateIndexQueue.getEntityId(),
                                                              connector.getPipelineName(),
                                                              singleRequestOperation);
              }
            } else {
              String singleRequestOperation =
                                            elasticContentRequestBuilder.getUpdateDocumentRequestContent(connector,
                                                                                                         updateIndexQueue.getEntityId());
              if (singleRequestOperation != null) {
                bulkRequest.append(singleRequestOperation);
              }
            }
          } catch (Exception e) {
            LOG.warn("Error while *updating* index entry of entity, type = " + entityIndex + ", id ="
                + (updateIndexQueue == null ? null : updateIndexQueue.getEntityId()) + ", cause:", e);
          } finally {
            // Delete this single operation since it's not indexed in bulk
            indexingQueueSorted.remove(OperationType.UPDATE, updateIndexQueue.getEntityId());
          }

          // Delete added indexation operation from queue even if the request
          // fails
          updateIndexingOperationsIterator.remove();

          // Check if the bulk request limit size is already reached
          bulkRequest = checkBulkRequestSizeReachedLimitation(bulkRequest);
        }
      }
    }

    if (bulkRequest.length() > 0 && !isInterrupted()) {
      elasticIndexingClient.sendCUDRequest(bulkRequest.toString());
    }
  }

  /**
   * Process all the requests for “init of the ES create mapping” (Operation
   * type = I) in the indexing queue (if any)
   * 
   * @param indexingQueueSorted Temporary inMemory IndexingQueue
   */
  private void processInit(Map<OperationType, Map<String, List<IndexingOperation>>> indexingQueueSorted) {
    if (indexingQueueSorted.containsKey(OperationType.INIT)) {
      for (String entityIndex : indexingQueueSorted.get(OperationType.INIT).keySet()) {
        IndexingServiceConnector connector = getConnectors().get(entityIndex);
        if (isInterrupted() || connector == null) {
          return;
        }
        sendInitRequests(connector);
      }
      indexingQueueSorted.remove(OperationType.INIT);
    }
  }

  /**
   * Process all the requests for “remove all documents of type” (Operation type =
   * X) in the indexing queue (if any) = Delete type in ES
   * 
   * @param indexingQueueSorted Temporary inMemory IndexingQueue
   */
  private void processDeleteAll(Map<OperationType, Map<String, List<IndexingOperation>>> indexingQueueSorted) {
    if (indexingQueueSorted.containsKey(OperationType.DELETE_ALL)) {
      for (String entityType : indexingQueueSorted.get(OperationType.DELETE_ALL).keySet()) {
        if (isInterrupted()) {
          return;
        }
        if (indexingQueueSorted.get(OperationType.DELETE_ALL).containsKey(entityType)) {
          for (IndexingOperation indexingOperation : indexingQueueSorted.get(OperationType.DELETE_ALL).get(entityType)) {
            processDeleteAll(indexingOperation, indexingQueueSorted);
          }
        }
      }
      indexingQueueSorted.remove(OperationType.DELETE_ALL);
    }
  }

  /**
   * @param indexingOperation
   * @param indexingQueueSorted Temporary inMemory IndexingQueue
   */
  private void processDeleteAll(IndexingOperation indexingOperation,
                                Map<OperationType, Map<String, List<IndexingOperation>>> indexingQueueSorted) {
    // Remove the type (= remove all documents of this type) and recreate it
    ElasticIndexingServiceConnector connector =
                                              (ElasticIndexingServiceConnector) getConnectors().get(indexingOperation.getEntityIndex());
    // log in Audit Trail
    auditTrail.audit(ElasticIndexingAuditTrail.DELETE_ALL, null, connector.getCurrentIndex(), null, null, 0);
    // Call ES
    elasticIndexingClient.sendDeleteAllDocsRequest(connector.getCurrentIndex());
    // Remove all useless CUD operation that was plan before this delete all
    deleteOperationsForTypesBefore(new OperationType[] { OperationType.CREATE, OperationType.UPDATE, OperationType.DELETE },
                                   indexingQueueSorted,
                                   indexingOperation);
  }

  private void deleteOperationsForTypesBefore(OperationType[] operations,
                                              Map<OperationType, Map<String, List<IndexingOperation>>> indexingQueueSorted,
                                              IndexingOperation refIindexOperation) {
    for (OperationType operation : operations) {
      if (indexingQueueSorted.containsKey(operation)) {
        if (indexingQueueSorted.get(operation).containsKey(refIindexOperation.getEntityIndex())) {
          for (Iterator<IndexingOperation> iterator = indexingQueueSorted.get(operation)
                  .get(refIindexOperation.getEntityIndex())
                  .iterator(); iterator.hasNext();) {
            IndexingOperation indexingOperation = iterator.next();
            // Check id higher than the timestamp of the reference
            // indexing operation, the index operation is removed
            if (refIindexOperation.getId() > indexingOperation.getId()) {
              iterator.remove();
            }
          }
        }
      }
    }
  }

  /**
   * Reindex all the entities of the given entity type.
   *
   * @param index Entity type of the entities to reindex
   */
  @ExoTransactional
  private void reindexAllByEntityIndex(String index) {
    long startTime = System.currentTimeMillis();
    IndexingServiceConnector connector = getConnectors().get(index);
    if(connector != null) {

      // 1- Delete all documents in ES (and purge the indexing queue)
      indexingOperationDAO.create(new IndexingOperation(null, index, OperationType.DELETE_ALL));
      int offset = 0;
      int numberIndexed;
      do {
        if (isInterrupted()) {
          return;
        }
        // 2- Get all the documents ID
        List<String> ids = connector.getAllIds(offset, reindexBatchSize);
        if (ids == null) {
          numberIndexed = 0;
        } else {
          List<IndexingOperation> operations = new ArrayList<>(ids.size());
          for (String id : ids) {
            operations.add(new IndexingOperation(id, index, OperationType.CREATE));
          }
          // 3- Inject as a CUD operation
          indexingOperationDAO.createAll(operations);
          numberIndexed = ids.size();
          offset += reindexBatchSize;
        }
      } while (numberIndexed == reindexBatchSize);
      // 4- log in Audit Trail
      auditTrail.audit(ElasticIndexingAuditTrail.REINDEX_ALL, null, index, null, null, (System.currentTimeMillis() - startTime));
    }
  }

  private void deleteOperationsByEntityIdForTypesBefore(OperationType[] operations,
                                                        Map<OperationType, Map<String, List<IndexingOperation>>> indexingQueueSorted,
                                                        IndexingOperation indexQueue) {
    for (OperationType operation : operations) {
      if (indexingQueueSorted.containsKey(operation)) {
        if (indexingQueueSorted.get(operation).containsKey(indexQueue.getEntityIndex())) {
          for (Iterator<IndexingOperation> iterator = indexingQueueSorted.get(operation)
                                                                         .get(indexQueue.getEntityIndex())
                                                                         .iterator(); iterator.hasNext();) {
            IndexingOperation indexingOperation = iterator.next();
            // Check Id higher than the Id of the CUD indexing queue, the index
            // queue is removed
            if ((indexQueue.getId() > indexingOperation.getId())
                && indexingOperation.getEntityId().equals(indexQueue.getEntityId())) {
              iterator.remove();
            }
          }
        }
      }
    }
  }

  private void deleteOperationsByEntityIdForTypes(OperationType[] operations,
                                                  Map<OperationType, Map<String, List<IndexingOperation>>> indexingQueueSorted,
                                                  IndexingOperation indexQueue) {
    for (OperationType operation : operations) {
      if (indexingQueueSorted.containsKey(operation)) {
        if (indexingQueueSorted.get(operation).containsKey(indexQueue.getEntityIndex())) {
          for (Iterator<IndexingOperation> iterator = indexingQueueSorted.get(operation)
                                                                         .get(indexQueue.getEntityIndex())
                                                                         .iterator(); iterator.hasNext();) {
            IndexingOperation indexingOperation = iterator.next();
            if (indexingOperation.getEntityId().equals(indexQueue.getEntityId())) {
              iterator.remove();
            }
          }
        }
      }
    }
  }

  private void sendInitRequests(IndexingServiceConnector indexingServiceConnector) {
    ElasticIndexingServiceConnector connector = (ElasticIndexingServiceConnector) indexingServiceConnector;

    String indexAlias = connector.getIndexAlias();
    String index = connector.getCurrentIndex();
    String previousIndex = connector.getPreviousIndex();

    if (indexUpgrading.containsKey(indexAlias)) {
      boolean newIndexExists = elasticIndexingClient.sendIsIndexExistsRequest(index);

      // If the upgrade is incomplete (should point the alias on previous index)
      boolean aliasExistsOnPreviousIndex = elasticIndexingClient.sendGetIndexAliasesRequest(previousIndex).contains(indexAlias);
      if(!aliasExistsOnPreviousIndex) {
        boolean aliasExistsOnCurrentIndex = newIndexExists && elasticIndexingClient.sendGetIndexAliasesRequest(index).contains(indexAlias);
        // If the alias points to the new index, point it again to the previous one, else add new alias to previous
        elasticIndexingClient.sendCreateIndexAliasRequest(previousIndex, aliasExistsOnCurrentIndex ? index : null, indexAlias);
      }

      if (newIndexExists) {
        // Upgrade was interrupted, so remove it and upgrade again
        LOG.warn("ES index upgrade '{}' was interrupted, the new index {} will be recreated", previousIndex, index);
        elasticIndexingClient.sendDeleteIndexRequest(index);
        // Send request to create index
        elasticIndexingClient.sendCreateIndexRequest(index, elasticContentRequestBuilder.getCreateIndexRequestContent(connector), connector.getMapping());
      } else {
        elasticIndexingClient.sendCreateIndexRequest(index,
                elasticContentRequestBuilder.getCreateIndexRequestContent(connector), connector.getMapping());
        if(connector.isNeedIngestPipeline()) {
          elasticIndexingClient.sendCreateAttachmentPipelineRequest(index, connector.getPipelineName(), connector.getAttachmentProcessor());
        }
      }

      // Init reindex. Once the reindex finished, the index alias will be updated to new index
      executors.submit(new ReindexESType(ExoContainerContext.getCurrentContainer(), connector));
    } else {
      boolean useAlias = true;
      if (index == null) {
        index = indexAlias;
        useAlias = false;
      }

      // Send request to create index
      boolean newlyCreated =
              elasticIndexingClient.sendCreateIndexRequest(index,
                      elasticContentRequestBuilder.getCreateIndexRequestContent(connector),
                      connector.getMapping());
      if (newlyCreated && useAlias) {
        elasticIndexingClient.sendCreateIndexAliasRequest(index, null, indexAlias);
      }

      if (connector.isNeedIngestPipeline()) {
        elasticIndexingClient.sendCreateAttachmentPipelineRequest(index,
                connector.getPipelineName(),
                connector.getAttachmentProcessor());
      }
      // Make sure that the migration is not executed twice on the same connector
      connector.setPreviousIndex(null);
    }
  }

  /**
   * If the bulk request already reached a size limitation, the bulk request
   * need to be sent immediately
   *
   * @param bulkRequest to analyze
   * @return
   */
  private StringBuilder checkBulkRequestSizeReachedLimitation(StringBuilder bulkRequest) {
    if (bulkRequest.length() >= requestSizeLimit) {
      elasticIndexingClient.sendCUDRequest(bulkRequest.toString());
      // return an empty bulk request
      return new StringBuilder();
    } else {
      return bulkRequest;
    }
  }

  public Integer getBatchNumber() {
    return batchNumber;
  }

  public void setBatchNumber(Integer batchNumber) {
    this.batchNumber = batchNumber;
  }

  public Integer getRequestSizeLimit() {
    return requestSizeLimit;
  }

  public void setRequestSizeLimit(Integer requestSizeLimit) {
    this.requestSizeLimit = requestSizeLimit;
  }

  @Override
  public void start() {
    try {
      String version = elasticIndexingClient.sendGetESVersion();
      if (version == null || !version.startsWith(this.esVersion + ".")) {
        LOG.error("Expected Version of ES version is " + this.esVersion + " but was " + version
            + ". If this is a compatible version, you can configure 'exo.es.version.minor' to delete this error message.");
      }
      // ES index and type need to be created for all registered connectors
      initConnectors();

      this.initialized = true;
    } catch (Exception e) {
      LOG.error("Error while initializing ES connectors", e);
    }
  }

  @Override
  public void stop() {
    executors.shutdownNow();
  }

  public void setInitialized(boolean initialized) {
    this.initialized = initialized;
  }

  private void initConnectors() {
    for (Map.Entry<String, IndexingServiceConnector> entry: getConnectors().entrySet()) {
      ElasticIndexingServiceConnector connector = (ElasticIndexingServiceConnector) entry.getValue();
      String previousIndex = connector.getPreviousIndex();
      String index = connector.getCurrentIndex();
      String indexAlias = connector.getIndexAlias();

      boolean needsUpgrade = false;
      if (StringUtils.isNotBlank(previousIndex)) {
        // Need to check the upgrade status (incomplete/ not run == new index doesn't exist or indexAlias is not added to new index)
        needsUpgrade = elasticIndexingClient.sendIsIndexExistsRequest(previousIndex)
                && (!elasticIndexingClient.sendIsIndexExistsRequest(index)
                || !elasticIndexingClient.sendGetIndexAliasesRequest(index).contains(indexAlias));
      }

      if(needsUpgrade) {
        if(!indexUpgrading.containsKey(indexAlias)) {
          indexUpgrading.put(indexAlias, new HashSet<>());
        }
        indexUpgrading.get(indexAlias).add(entry.getKey());
      }
    }
    for (Map.Entry<String, IndexingServiceConnector> entry : getConnectors().entrySet()) {
      sendInitRequests(entry.getValue());
    }
  }

  public class ReindexESType implements Runnable {

    private ElasticIndexingServiceConnector indexingServiceConnector;

    private ExoContainer exoContainer;

    public ReindexESType(ExoContainer exoContainer, ElasticIndexingServiceConnector connector) {
      this.exoContainer = exoContainer;
      this.indexingServiceConnector = connector;
    }

    @Override
    public void run() {
      try {
          LOG.info("Reindexing index alias {} from old index {} to new index {}",
                  indexingServiceConnector.getIndexAlias(),
                  indexingServiceConnector.getPreviousIndex(),
                  indexingServiceConnector.getCurrentIndex());
          try {
            elasticIndexingClient.sendReindexTypeRequest(indexingServiceConnector.getCurrentIndex(), indexingServiceConnector.getPreviousIndex(), indexingServiceConnector.getReindexPipelineName());
            if(this.indexingServiceConnector.isReindexOnUpgrade()) {
              ExoContainerContext.setCurrentContainer(exoContainer);
              reindexAllByEntityIndex(indexingServiceConnector.getConnectorName());
            }
            LOG.info("Reindexation finished for index alias {} from old index {} to new index {}",
                    indexingServiceConnector.getIndexAlias(),
                    indexingServiceConnector.getPreviousIndex(),
                    indexingServiceConnector.getCurrentIndex());
          } catch (Exception e) {
            LOG.warn("Reindexation using pipeline error for index alias {} from old index {} to new index {}, for type {}. The reindexation will proceed from eXo DB",
                    indexingServiceConnector.getIndexAlias(),
                    indexingServiceConnector.getPreviousIndex(),
                    indexingServiceConnector.getCurrentIndex());
          }

        // This algorithm should be thread safe
        synchronized (indexUpgrading) {
          boolean indexMigrationInProgress = indexUpgrading.get(indexingServiceConnector.getIndexAlias()).size() > 1;

          if (indexMigrationInProgress) {
            LOG.info("The index {} has some types not completely migrated yet, the old index will be deleted after migration is finished",
                    indexingServiceConnector.getPreviousIndex());
          } else {
            LOG.info("Switching index alias {} from old index {} to new index {}", indexingServiceConnector.getIndexAlias(), indexingServiceConnector.getPreviousIndex(), indexingServiceConnector.getCurrentIndex());
            elasticIndexingClient.sendCreateIndexAliasRequest(indexingServiceConnector.getCurrentIndex(), indexingServiceConnector.getPreviousIndex(), indexingServiceConnector.getIndexAlias());

            indexUpgrading.remove(indexingServiceConnector.getIndexAlias());

            LOG.info("Remove old index {}", indexingServiceConnector.getPreviousIndex());
            elasticIndexingClient.sendDeleteIndexRequest(indexingServiceConnector.getPreviousIndex());

            if(indexUpgrading.isEmpty()) {
              LOG.info("ES indexes migration finished (except indexes that will be reindexed from DB)");
            }
          }
        }
      } catch (Exception e) {
        LOG.error("An error occurred while upgrading index " + indexingServiceConnector.getPreviousIndex(), e);
      } finally {
        indexUpgrading.remove(indexingServiceConnector.getIndexAlias());
      }
    }
  }
}
