/*
 *
 *  * 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 Affero 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 General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, see<http://www.gnu.org/licenses/>.
 *
 */

package org.exoplatform.commons.search.integration;

import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.fail;

import java.io.IOException;
import java.net.DatagramSocket;
import java.net.ServerSocket;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.SortedMap;

import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.Version;
import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoRequest;
import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
import org.elasticsearch.cluster.metadata.AliasOrIndex;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.env.Environment;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.ReindexPlugin;
import org.elasticsearch.ingest.IngestService;
import org.elasticsearch.ingest.attachment.IngestAttachmentPlugin;
import org.elasticsearch.node.Node;
import org.elasticsearch.node.NodeValidationException;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.transport.Netty4Plugin;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;

import org.exoplatform.commons.search.es.client.ElasticIndexingAuditTrail;
import org.exoplatform.commons.search.es.client.ElasticIndexingClient;
import org.exoplatform.commons.search.es.client.ElasticSearchingClient;
import org.exoplatform.commons.utils.PropertyManager;
import org.exoplatform.container.PortalContainer;
import org.exoplatform.services.log.ExoLogger;
import org.exoplatform.services.log.Log;

import sun.util.logging.resources.logging;

/**
 * Created by The eXo Platform SAS Author : eXoPlatform exo@exoplatform.com
 * 10/1/15
 */
public class BaseElasticsearchIT {
  private static final Log LOGGER = ExoLogger.getExoLogger(BaseElasticsearchIT.class);
  protected static Node node;

  private static boolean propertiesSet;

  protected static ElasticIndexingClient elasticIndexingClient;
  protected static ElasticSearchingClient elasticSearchingClient;

  /**
   * Port used for embedded Elasticsearch.
   * The port is needed to set the system properties exo.es.index.server.url and exo.es.search.server.url that
   * are used by the services ElasticIndexingClient and ElasticSearchingClient. These services read these
   * system properties at init (constructor) and cannot be changed later (no setter). Since the eXo Container is
   * created only once for all the tests, the same instance of these services are used for all the tests, so
   * the port must be the same for all the tests. That's why it must be static.
   */
  private static Integer esPort = null;

  @BeforeClass
  public static void setUp() {
    startEmbeddedES();
    // If the test is executed standalone, the properties are not present at the
    // beginning
    // of the test, so they have to be removed at the end of the test to avoid
    // java.lang.AssertionError: System properties invariant violated from
    // com.carrotsearch.randomizedtesting
    // If another test ran before and set the properties, we must not remove the
    // properties
    // If we do, we'll get the same exception.
    propertiesSet = StringUtils.isBlank(System.getProperty("exo.es.index.server.url"));
    PropertyManager.setProperty("exo.es.index.server.url", "http://localhost:" + esPort);
    PropertyManager.setProperty("exo.es.search.server.url", "http://localhost:" + esPort);

    elasticIndexingClient = new ElasticIndexingClient(new ElasticIndexingAuditTrail());
    elasticSearchingClient = new ElasticSearchingClient(new ElasticIndexingAuditTrail());
  }

  @Before
  public void setup() {
    // Make sure a container has been created
    PortalContainer.getInstance();

    // Reset data
    deleteAllIndexesInES();
  }

  @AfterClass
  public static void cleanUrlProperties() {
    // Close ES Node
    if(node != null) {
      LOGGER.info("Embedded ES instance - Stopping");
      try {
        node.close();
      } catch (Exception e) {
        LOGGER.error("Embedded ES instance can't be stopped", e);
      }
      LOGGER.info("Embedded ES instance - Stopped");
    }

    if (propertiesSet) {
      System.clearProperty("exo.es.index.server.url");
      System.clearProperty("exo.es.indexing.batch.number");
      System.clearProperty("exo.es.indexing.replica.number.default");
      System.clearProperty("exo.es.indexing.shard.number.default");
      System.clearProperty("exo.es.search.server.url");
      System.clearProperty("jboss.i18n.generate-proxies");
    } else {
      //Reset server.url properties to default values
      PropertyManager.setProperty("exo.es.index.server.url", "http://127.0.0.1:9200");
      PropertyManager.setProperty("exo.es.search.server.url", "http://127.0.0.1:9200");
    }
  }

  /**
   * Start an Elasticsearch instance
   */
  private static void startEmbeddedES() {
    if(esPort == null) {
      try {
        esPort = getAvailablePort();
      } catch (IOException e) {
        fail("Cannot get available port : " + e.getMessage());
      }
    }

    // Init ES
    LOGGER.info("Embedded ES instance - Starting on port " + esPort);
    Settings.Builder elasticsearchSettings = Settings.builder()
            .put("http.enabled", true)
            .put("network.host", "_local_")
            .put("http.port", esPort)
            .put("node.name", "esEmbeddedForTests" + esPort)
            .put("node.master", true)
            .put("node.data", true)
            .put("path.logs", "logs")
            .put("transport.type", "local")
            .put("http.type", "netty4")
            .put("script.inline", false)
            .put("script.stored", false)
            .put("script.file", false)
            .put("path.home", "target/es")
            .put("path.data", "target/es");

    Environment environment = new Environment(elasticsearchSettings.build());
    Collection plugins = new ArrayList<>();
    Collections.<Class<? extends Plugin>>addAll(plugins, Netty4Plugin.class, IngestAttachmentPlugin.class, ReindexPlugin.class);
    node = new EmbeddedNode(environment, Version.CURRENT, plugins);
    try {
      node.start();
    } catch (NodeValidationException e) {
      LOGGER.error("Embedded ES instance couldn't start", e);
    }
    //node = nodeBuilder().local(true).settings(elasticsearchSettings.build()).node();
    node.client().admin().cluster().prepareHealth().setWaitForYellowStatus().execute().actionGet();
    assertNotNull(node);
    assertFalse(node.isClosed());
    LOGGER.info("Embedded ES instance - Started");
    // Set URL of server in property
    NodesInfoRequest nodesInfoRequest = new NodesInfoRequest().transport(true);
    NodesInfoResponse response = node.client().admin().cluster().nodesInfo(nodesInfoRequest).actionGet();
    NodeInfo nodeInfo = response.getNodes().iterator().next();
    InetSocketTransportAddress address = (InetSocketTransportAddress) nodeInfo.getHttp().getAddress().publishAddress();
    String url = "http://" + address.address().getHostName() + ":" + address.address().getPort();
    PropertyManager.setProperty("exo.es.index.server.url", url);
    PropertyManager.setProperty("exo.es.search.server.url", url);
  }

  /**
   * Get a random available port
   * @return
   * @throws IOException
   */
  private static int getAvailablePort() throws IOException {
    ServerSocket ss = null;
    DatagramSocket ds = null;
    try {
      ss = new ServerSocket(0);
      ss.setReuseAddress(true);
      ds = new DatagramSocket(0);
      ds.setReuseAddress(true);
      return ss.getLocalPort();
    } finally {
      if (ds != null) {
        ds.close();
      }

      if (ss != null) {
        try {
          ss.close();
        } catch (IOException e) {
                /* should not be thrown */
        }
      }
    }
  }

  protected void deleteAllIndexesInES() {
    node.client().admin().indices().prepareRefresh().execute().actionGet();
    ImmutableOpenMap<String, IndexMetaData> indices = node.client().admin().cluster()
            .prepareState().execute().actionGet().getState().getMetaData().getIndices();
    Iterator<String> keysIt = indices.keysIt();
    while (keysIt.hasNext()) {
      String index = keysIt.next();
      node.client().admin().indices().prepareDelete(index).execute().actionGet();
    }
    node.client().admin().indices().prepareRefresh().execute().actionGet();
  }

  protected boolean typeExists(String index, String type) {
    return node.client().admin().indices().prepareTypesExists(index).setTypes(type).execute().actionGet().isExists();
  }

  protected boolean pipelineExists(String pipelineName) {
    return node.injector().getInstance(IngestService.class).getPipelineStore().get(pipelineName) != null;
  }

  protected long documentNumber() {
    return node.client().prepareSearch().setSize(0).execute().actionGet().getHits().getTotalHits();
  }

  protected long documentNumber(String type) {
    return node.client().prepareSearch("_all").setSize(0).setQuery(QueryBuilders.termQuery("_type", type)).execute().actionGet().getHits().getTotalHits();
  }
}
