/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * 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.
 */
package org.apache.solr.cloud.api.collections;

import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.impl.CloudLegacySolrClient;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.UpdateRequest;
import org.apache.solr.client.solrj.response.CollectionAdminResponse;
import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.cloud.AbstractDistribZkTestBase;
import org.apache.solr.cloud.SolrCloudTestCase;
import org.apache.solr.common.SolrInputDocument;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.Slice;
import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.embedded.JettySolrRunner;
import org.junit.Test;

/** Test of the Collections API with the MiniSolrCloudCluster. */
public class TestCollectionsAPIViaSolrCloudCluster extends SolrCloudTestCase {

  private static final int numShards = 2;
  private static final int numReplicas = 2;
  private static final int nodeCount = 5;
  private static final String configName = "solrCloudCollectionConfig";
  private static final Map<String, String>
      collectionProperties // ensure indexes survive core shutdown
      = Collections.singletonMap("solr.directoryFactory", "solr.StandardDirectoryFactory");

  @Override
  public void setUp() throws Exception {
    configureCluster(nodeCount)
        .addConfig(configName, configset("cloud-minimal"))
        .addConfig("cloud-minimal-userproperties", configset("cloud-minimal-userproperties"))
        .configure();
    super.setUp();
  }

  @Override
  public void tearDown() throws Exception {
    cluster.shutdown();
    super.tearDown();
  }

  private void createCollection(String collectionName, String createNodeSet) throws Exception {
    if (random().nextBoolean()) { // process asynchronously
      CollectionAdminRequest.createCollection(collectionName, configName, numShards, numReplicas)
          .setCreateNodeSet(createNodeSet)
          .setProperties(collectionProperties)
          .processAndWait(cluster.getSolrClient(), 30);
    } else {
      CollectionAdminRequest.createCollection(collectionName, configName, numShards, numReplicas)
          .setCreateNodeSet(createNodeSet)
          .setProperties(collectionProperties)
          .process(cluster.getSolrClient());
    }

    if (createNodeSet != null
        && createNodeSet.equals(CollectionHandlingUtils.CREATE_NODE_SET_EMPTY)) {
      cluster.waitForActiveCollection(collectionName, numShards, 0);
    } else {
      cluster.waitForActiveCollection(collectionName, numShards, numShards * numReplicas);
    }
  }

  @Test
  public void testCollectionCreateSearchDelete() throws Exception {
    final CloudSolrClient client = cluster.getSolrClient();
    final String collectionName = "testcollection";

    assertNotNull(cluster.getZkServer());
    List<JettySolrRunner> jettys = cluster.getJettySolrRunners();
    assertEquals(nodeCount, jettys.size());
    for (JettySolrRunner jetty : jettys) {
      assertTrue(jetty.isRunning());
    }

    // shut down a server
    JettySolrRunner stoppedServer = cluster.stopJettySolrRunner(0);

    cluster.waitForJettyToStop(stoppedServer);

    assertTrue(stoppedServer.isStopped());
    assertEquals(nodeCount - 1, cluster.getJettySolrRunners().size());

    // create a server
    JettySolrRunner startedServer = cluster.startJettySolrRunner();
    cluster.waitForAllNodes(30);
    assertTrue(startedServer.isRunning());
    assertEquals(nodeCount, cluster.getJettySolrRunners().size());

    // create collection
    createCollection(collectionName, null);

    // modify/query collection
    new UpdateRequest().add("id", "1").commit(client, collectionName);
    QueryResponse rsp = client.query(collectionName, new SolrQuery("*:*"));
    assertEquals(1, rsp.getResults().getNumFound());

    // remove a server not hosting any replicas
    ZkStateReader zkStateReader = ZkStateReader.from(client);
    zkStateReader.forceUpdateCollection(collectionName);
    ClusterState clusterState = zkStateReader.getClusterState();
    Map<String, JettySolrRunner> jettyMap = new HashMap<>();
    for (JettySolrRunner jetty : cluster.getJettySolrRunners()) {
      String key =
          jetty
              .getBaseUrl()
              .toString()
              .substring((jetty.getBaseUrl().getProtocol() + "://").length());
      jettyMap.put(key, jetty);
    }
    Collection<Slice> slices = clusterState.getCollection(collectionName).getSlices();
    // track the servers not host replicas
    for (Slice slice : slices) {
      jettyMap.remove(slice.getLeader().getNodeName().replace("_solr", "/solr"));
      for (Replica replica : slice.getReplicas()) {
        jettyMap.remove(replica.getNodeName().replace("_solr", "/solr"));
      }
    }
    assertTrue("Expected to find a node without a replica", jettyMap.size() > 0);
    JettySolrRunner jettyToStop = jettyMap.entrySet().iterator().next().getValue();
    jettys = cluster.getJettySolrRunners();
    for (int i = 0; i < jettys.size(); ++i) {
      if (jettys.get(i).equals(jettyToStop)) {
        cluster.stopJettySolrRunner(i);
        assertEquals(nodeCount - 1, cluster.getJettySolrRunners().size());
      }
    }

    // re-create a server (to restore original nodeCount count)
    startedServer = cluster.startJettySolrRunner(jettyToStop);
    cluster.waitForAllNodes(30);
    assertTrue(startedServer.isRunning());
    assertEquals(nodeCount, cluster.getJettySolrRunners().size());

    CollectionAdminRequest.deleteCollection(collectionName).process(client);
    AbstractDistribZkTestBase.waitForCollectionToDisappear(
        collectionName, ZkStateReader.from(client), true, 330);

    // create it again
    createCollection(collectionName, null);

    cluster.waitForActiveCollection(collectionName, numShards, numShards * numReplicas);

    // check that there's no left-over state
    assertEquals(0, client.query(collectionName, new SolrQuery("*:*")).getResults().getNumFound());

    // modify/query collection
    new UpdateRequest().add("id", "1").commit(client, collectionName);
    assertEquals(1, client.query(collectionName, new SolrQuery("*:*")).getResults().getNumFound());
  }

  @Test
  // 12-Jun-2018 @BadApple(bugUrl="https://issues.apache.org/jira/browse/SOLR-12028") // 09-Apr-2018
  public void testCollectionCreateWithoutCoresThenDelete() throws Exception {

    final String collectionName = "testSolrCloudCollectionWithoutCores";
    final CloudSolrClient client = cluster.getSolrClient();

    assertNotNull(cluster.getZkServer());
    assertFalse(cluster.getJettySolrRunners().isEmpty());

    // create collection
    createCollection(collectionName, CollectionHandlingUtils.CREATE_NODE_SET_EMPTY);

    // check the collection has no cores.
    int coreCount = 0;
    DocCollection docCollection = client.getClusterState().getCollection(collectionName);
    for (Map.Entry<String, Slice> entry : docCollection.getSlicesMap().entrySet()) {
      coreCount += entry.getValue().getReplicasMap().entrySet().size();
    }
    assertEquals(0, coreCount);

    // delete the collection
    CollectionAdminRequest.deleteCollection(collectionName).process(client);
    AbstractDistribZkTestBase.waitForCollectionToDisappear(
        collectionName, ZkStateReader.from(client), true, 330);
  }

  @Test
  public void testStopAllStartAll() throws Exception {

    final String collectionName = "testStopAllStartAllCollection";
    final CloudSolrClient client = cluster.getSolrClient();

    assertNotNull(cluster.getZkServer());
    List<JettySolrRunner> jettys = new ArrayList<>(cluster.getJettySolrRunners()); // make a copy
    assertEquals(nodeCount, jettys.size());
    for (JettySolrRunner jetty : jettys) {
      assertTrue(jetty.isRunning());
    }

    final SolrQuery query = new SolrQuery("*:*");
    final SolrInputDocument doc = new SolrInputDocument();

    // create collection
    createCollection(collectionName, null);

    ZkStateReader zkStateReader = ZkStateReader.from(client);

    // modify collection
    final int numDocs = 1 + random().nextInt(10);
    for (int ii = 1; ii <= numDocs; ++ii) {
      doc.setField("id", "" + ii);
      client.add(collectionName, doc);
      if (ii * 2 == numDocs) client.commit(collectionName);
    }
    client.commit(collectionName);

    // query collection
    assertEquals(numDocs, client.query(collectionName, query).getResults().getNumFound());

    // the test itself
    zkStateReader.forceUpdateCollection(collectionName);
    final ClusterState clusterState = zkStateReader.getClusterState();

    final Set<Integer> leaderIndices = new HashSet<>();
    final Set<Integer> followerIndices = new HashSet<>();
    {
      final Map<String, Boolean> shardLeaderMap = new HashMap<>();
      for (final Slice slice : clusterState.getCollection(collectionName).getSlices()) {
        for (final Replica replica : slice.getReplicas()) {
          shardLeaderMap.put(replica.getNodeName().replace("_solr", "/solr"), Boolean.FALSE);
        }
        shardLeaderMap.put(slice.getLeader().getNodeName().replace("_solr", "/solr"), Boolean.TRUE);
      }
      for (int ii = 0; ii < jettys.size(); ++ii) {
        final URL jettyBaseUrl = jettys.get(ii).getBaseUrl();
        final String jettyBaseUrlString =
            jettyBaseUrl.toString().substring((jettyBaseUrl.getProtocol() + "://").length());
        final Boolean isLeader = shardLeaderMap.get(jettyBaseUrlString);
        if (Boolean.TRUE.equals(isLeader)) {
          leaderIndices.add(ii);
        } else if (Boolean.FALSE.equals(isLeader)) {
          followerIndices.add(ii);
        } // else neither leader nor follower i.e. node without a replica (for our collection)
      }
    }
    final List<Integer> leaderIndicesList = new ArrayList<>(leaderIndices);
    final List<Integer> followerIndicesList = new ArrayList<>(followerIndices);

    // first stop the followers (in no particular order)
    Collections.shuffle(followerIndicesList, random());
    for (Integer ii : followerIndicesList) {
      if (!leaderIndices.contains(ii)) {
        cluster.stopJettySolrRunner(jettys.get(ii));
      }
    }

    // then stop the leaders (again in no particular order)
    Collections.shuffle(leaderIndicesList, random());
    for (Integer ii : leaderIndicesList) {
      cluster.stopJettySolrRunner(jettys.get(ii));
    }

    // calculate restart order
    final List<Integer> restartIndicesList = new ArrayList<>();
    Collections.shuffle(leaderIndicesList, random());
    restartIndicesList.addAll(leaderIndicesList);
    Collections.shuffle(followerIndicesList, random());
    restartIndicesList.addAll(followerIndicesList);
    if (random().nextBoolean()) Collections.shuffle(restartIndicesList, random());

    // and then restart jettys in that order
    for (Integer ii : restartIndicesList) {
      final JettySolrRunner jetty = jettys.get(ii);
      if (!jetty.isRunning()) {
        cluster.startJettySolrRunner(jetty);
        assertTrue(jetty.isRunning());
      }
    }
    cluster.waitForAllNodes(30);
    cluster.waitForActiveCollection(collectionName, numShards, numShards * numReplicas);

    zkStateReader.forceUpdateCollection(collectionName);

    // re-query collection
    assertEquals(numDocs, client.query(collectionName, query).getResults().getNumFound());
  }

  @Test
  public void testUserDefinedProperties() throws Exception {
    String confName = "cloud-minimal-userproperties";
    String collectionName = "testUserDefinedPropertiesCollection";

    switch (random().nextInt(3)) {
      case 0:
        CollectionAdminRequest.Create createOp =
            CollectionAdminRequest.createCollection(collectionName, confName, 1, 2);
        createOp.withProperty("my.custom.prop", "customProp");
        CollectionAdminResponse rsp = createOp.process(cluster.getSolrClient());
        assertNull(rsp.getErrorMessages());
        assertSame(0, rsp.getStatus());
        break;

      case 1:
        // Sometimes use v1 API
        String url =
            String.format(
                Locale.ROOT,
                "%s/admin/collections?action=CREATE&name=%s&collection.configName=%s&numShards=%s&nrtReplicas=%s&tlogReplicas=%s&pullReplicas=%s&&property.my.custom.prop=%s",
                cluster.getRandomJetty(random()).getBaseUrl(),
                collectionName,
                confName,
                2,
                2,
                2,
                0,
                "customProp");

        HttpGet createCollectionGet = new HttpGet(url);
        HttpResponse httpResponseV1 =
            ((CloudLegacySolrClient) cluster.getSolrClient())
                .getHttpClient()
                .execute(createCollectionGet);
        assertEquals(200, httpResponseV1.getStatusLine().getStatusCode());
        break;

      case 2:
        // Sometimes use V2 API
        url = cluster.getRandomJetty(random()).getBaseUrl().toString() + "/____v2/collections";
        String requestBody =
            String.format(
                Locale.ROOT,
                "{\"name\":\"%s\", \"config\":\"%s\", \"numShards\":%s, \"nrtReplicas\":%s, \"tlogReplicas\":%s, \"pullReplicas\":%s,"
                    + " \"properties\": { \"my.custom.prop\": \"test\" } }",
                collectionName,
                confName,
                2,
                2,
                2,
                0);
        HttpPost createCollectionPost = new HttpPost(url);
        createCollectionPost.setHeader("Content-type", "application/json");
        createCollectionPost.setEntity(new StringEntity(requestBody));
        HttpResponse httpResponseV2 =
            ((CloudLegacySolrClient) cluster.getSolrClient())
                .getHttpClient()
                .execute(createCollectionPost);
        assertEquals(200, httpResponseV2.getStatusLine().getStatusCode());
        break;
    }
  }
}
