/*
 * 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.client.solrj.impl;

import static org.apache.solr.client.solrj.impl.BaseHttpSolrClient.RemoteSolrException;

import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.QueryRequest;
import org.apache.solr.client.solrj.response.CollectionAdminResponse;
import org.apache.solr.common.cloud.Aliases;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.PerReplicaStates;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.SimpleOrderedMap;
import org.apache.solr.common.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class BaseHttpClusterStateProvider implements ClusterStateProvider {
  private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

  private String urlScheme;
  volatile Set<String> liveNodes;
  long liveNodesTimestamp = 0;
  volatile Map<String, List<String>> aliases;
  volatile Map<String, Map<String, String>> aliasProperties;
  long aliasesTimestamp = 0;

  private int cacheTimeout = 5; // the liveNodes and aliases cache will be invalidated after 5 secs

  public void init(List<String> solrUrls) throws Exception {
    for (String solrUrl : solrUrls) {
      urlScheme = solrUrl.startsWith("https") ? "https" : "http";
      try (SolrClient initialClient = getSolrClient(solrUrl)) {
        this.liveNodes = fetchLiveNodes(initialClient);
        liveNodesTimestamp = System.nanoTime();
        break;
      } catch (SolrServerException | IOException e) {
        log.warn("Attempt to fetch cluster state from {} failed.", solrUrl, e);
      }
    }

    if (this.liveNodes == null || this.liveNodes.isEmpty()) {
      throw new RuntimeException(
          "Tried fetching live_nodes using Solr URLs provided, i.e. "
              + solrUrls
              + ". However, "
              + "succeeded in obtaining the cluster state from none of them."
              + "If you think your Solr cluster is up and is accessible,"
              + " you could try re-creating a new CloudSolrClient using working"
              + " solrUrl(s) or zkHost(s).");
    }
  }

  protected abstract SolrClient getSolrClient(String baseUrl);

  @Override
  public ClusterState.CollectionRef getState(String collection) {
    for (String nodeName : liveNodes) {
      String baseUrl = Utils.getBaseUrlForNodeName(nodeName, urlScheme);
      try (SolrClient client = getSolrClient(baseUrl)) {
        ClusterState cs = fetchClusterState(client, collection, null);
        return cs.getCollectionRef(collection);
      } catch (SolrServerException | IOException e) {
        log.warn(
            "Attempt to fetch cluster state from {} failed.",
            Utils.getBaseUrlForNodeName(nodeName, urlScheme),
            e);
      } catch (RemoteSolrException e) {
        if ("NOT_FOUND".equals(e.getMetadata("CLUSTERSTATUS"))) {
          return null;
        }
        log.warn("Attempt to fetch cluster state from {} failed.", baseUrl, e);
      } catch (NotACollectionException e) {
        // Cluster state for the given collection was not found, could be an alias.
        // Lets fetch/update our aliases:
        getAliases(true);
        return null;
      }
    }
    throw new RuntimeException(
        "Tried fetching cluster state using the node names we knew of, i.e. "
            + liveNodes
            + ". However, "
            + "succeeded in obtaining the cluster state from none of them."
            + "If you think your Solr cluster is up and is accessible,"
            + " you could try re-creating a new CloudSolrClient using working"
            + " solrUrl(s) or zkHost(s).");
  }

  @SuppressWarnings("unchecked")
  private ClusterState fetchClusterState(
      SolrClient client, String collection, Map<String, Object> clusterProperties)
      throws SolrServerException, IOException, NotACollectionException {
    ModifiableSolrParams params = new ModifiableSolrParams();
    if (collection != null) {
      params.set("collection", collection);
    }
    params.set("action", "CLUSTERSTATUS");
    params.set("prs", "true");
    QueryRequest request = new QueryRequest(params);
    request.setPath("/admin/collections");
    SimpleOrderedMap<?> cluster = (SimpleOrderedMap<?>) client.request(request).get("cluster");
    Map<String, Object> collectionsMap;
    if (collection != null) {
      collectionsMap =
          Collections.singletonMap(
              collection, ((NamedList<?>) cluster.get("collections")).get(collection));
    } else {
      collectionsMap = ((NamedList<?>) cluster.get("collections")).asMap(10);
    }
    int znodeVersion;
    Map<String, Object> collFromStatus = (Map<String, Object>) (collectionsMap).get(collection);
    if (collection != null && collFromStatus == null) {
      throw new NotACollectionException(); // probably an alias
    }
    if (collection != null) { // can be null if alias
      znodeVersion = (int) collFromStatus.get("znodeVersion");
    } else {
      znodeVersion = -1;
    }
    Set<String> liveNodes = new HashSet<>((List<String>) (cluster.get("live_nodes")));
    this.liveNodes = liveNodes;
    liveNodesTimestamp = System.nanoTime();
    ClusterState cs = new ClusterState(liveNodes, new HashMap<>());
    for (Map.Entry<String, Object> e : collectionsMap.entrySet()) {
      @SuppressWarnings("rawtypes")
      Map m = (Map) e.getValue();
      cs = cs.copyWith(e.getKey(), fillPrs(znodeVersion, e, m));
    }

    if (clusterProperties != null) {
      Map<String, Object> properties = (Map<String, Object>) cluster.get("properties");
      if (properties != null) {
        clusterProperties.putAll(properties);
      }
    }
    return cs;
  }

  @SuppressWarnings({"rawtypes", "unchecked"})
  private DocCollection fillPrs(int znodeVersion, Map.Entry<String, Object> e, Map m) {
    DocCollection.PrsSupplier prsSupplier = null;
    if (m.containsKey("PRS")) {
      Map prs = (Map) m.remove("PRS");
      prsSupplier =
          () ->
              new PerReplicaStates(
                  (String) prs.get("path"),
                  (Integer) prs.get("cversion"),
                  (List<String>) prs.get("states"));
    }

    return ClusterState.collectionFromObjects(e.getKey(), m, znodeVersion, prsSupplier);
  }

  @Override
  public Set<String> getLiveNodes() {
    if (liveNodes == null) {
      throw new RuntimeException(
          "We don't know of any live_nodes to fetch the"
              + " latest live_nodes information from. "
              + "If you think your Solr cluster is up and is accessible,"
              + " you could try re-creating a new CloudSolrClient using working"
              + " solrUrl(s) or zkHost(s).");
    }
    if (TimeUnit.SECONDS.convert((System.nanoTime() - liveNodesTimestamp), TimeUnit.NANOSECONDS)
        > getCacheTimeout()) {
      for (String nodeName : liveNodes) {
        String baseUrl = Utils.getBaseUrlForNodeName(nodeName, urlScheme);
        try (SolrClient client = getSolrClient(baseUrl)) {
          Set<String> liveNodes = fetchLiveNodes(client);
          this.liveNodes = (liveNodes);
          liveNodesTimestamp = System.nanoTime();
          return liveNodes;
        } catch (Exception e) {
          log.warn("Attempt to fetch cluster state from {} failed.", baseUrl, e);
        }
      }
      throw new RuntimeException(
          "Tried fetching live_nodes using all the node names we knew of, i.e. "
              + liveNodes
              + ". However, "
              + "succeeded in obtaining the cluster state from none of them."
              + "If you think your Solr cluster is up and is accessible,"
              + " you could try re-creating a new CloudSolrClient using working"
              + " solrUrl(s) or zkHost(s).");
    } else {
      return liveNodes; // cached copy is fresh enough
    }
  }

  @SuppressWarnings({"rawtypes", "unchecked"})
  private static Set<String> fetchLiveNodes(SolrClient client) throws Exception {
    ModifiableSolrParams params = new ModifiableSolrParams();
    params.set("action", "CLUSTERSTATUS");
    QueryRequest request = new QueryRequest(params);
    request.setPath("/admin/collections");
    NamedList cluster = (SimpleOrderedMap) client.request(request).get("cluster");
    return (Set<String>) new HashSet((List<String>) (cluster.get("live_nodes")));
  }

  @Override
  public List<String> resolveAlias(String aliasName) {
    return resolveAlias(aliasName, false);
  }

  public List<String> resolveAlias(String aliasName, boolean forceFetch) {
    return Aliases.resolveAliasesGivenAliasMap(getAliases(forceFetch), aliasName);
  }

  @Override
  public String resolveSimpleAlias(String aliasName) throws IllegalArgumentException {
    return Aliases.resolveSimpleAliasGivenAliasMap(getAliases(false), aliasName);
  }

  private Map<String, List<String>> getAliases(boolean forceFetch) {
    if (this.liveNodes == null) {
      throw new RuntimeException(
          "We don't know of any live_nodes to fetch the"
              + " latest aliases information from. "
              + "If you think your Solr cluster is up and is accessible,"
              + " you could try re-creating a new CloudSolrClient using working"
              + " solrUrl(s) or zkHost(s).");
    }

    if (forceFetch
        || this.aliases == null
        || TimeUnit.SECONDS.convert((System.nanoTime() - aliasesTimestamp), TimeUnit.NANOSECONDS)
            > getCacheTimeout()) {
      for (String nodeName : liveNodes) {
        String baseUrl = Utils.getBaseUrlForNodeName(nodeName, urlScheme);
        try (SolrClient client = getSolrClient(baseUrl)) {

          CollectionAdminResponse response =
              new CollectionAdminRequest.ListAliases().process(client);
          this.aliases = response.getAliasesAsLists();
          this.aliasProperties = response.getAliasProperties(); // side-effect
          this.aliasesTimestamp = System.nanoTime();
          return Collections.unmodifiableMap(this.aliases);
        } catch (SolrServerException | RemoteSolrException | IOException e) {
          // Situation where we're hitting an older Solr which doesn't have LISTALIASES
          if (e instanceof RemoteSolrException && ((RemoteSolrException) e).code() == 400) {
            log.warn(
                "LISTALIASES not found, possibly using older Solr server. Aliases won't work {}",
                "unless you re-create the CloudSolrClient using zkHost(s) or upgrade Solr server",
                e);
            this.aliases = Collections.emptyMap();
            this.aliasProperties = Collections.emptyMap();
            this.aliasesTimestamp = System.nanoTime();
            return aliases;
          }
          log.warn("Attempt to fetch cluster state from {} failed.", baseUrl, e);
        }
      }

      throw new RuntimeException(
          "Tried fetching aliases using all the node names we knew of, i.e. "
              + liveNodes
              + ". However, "
              + "succeeded in obtaining the cluster state from none of them."
              + "If you think your Solr cluster is up and is accessible,"
              + " you could try re-creating a new CloudSolrClient using a working"
              + " solrUrl or zkHost.");
    } else {
      return Collections.unmodifiableMap(this.aliases); // cached copy is fresh enough
    }
  }

  @Override
  public Map<String, String> getAliasProperties(String alias) {
    getAliases(false);
    return Collections.unmodifiableMap(aliasProperties.getOrDefault(alias, Collections.emptyMap()));
  }

  @Override
  public ClusterState getClusterState() {
    for (String nodeName : liveNodes) {
      String baseUrl = Utils.getBaseUrlForNodeName(nodeName, urlScheme);
      try (SolrClient client = getSolrClient(baseUrl)) {
        return fetchClusterState(client, null, null);
      } catch (SolrServerException | BaseHttpSolrClient.RemoteSolrException | IOException e) {
        log.warn("Attempt to fetch cluster state from {} failed.", baseUrl, e);
      } catch (NotACollectionException e) {
        // not possible! (we passed in null for collection so it can't be an alias)
        throw new RuntimeException(
            "null should never cause NotACollectionException in "
                + "fetchClusterState() Please report this as a bug!");
      }
    }
    throw new RuntimeException(
        "Tried fetching cluster state using the node names we knew of, i.e. "
            + liveNodes
            + ". However, "
            + "succeeded in obtaining the cluster state from none of them."
            + "If you think your Solr cluster is up and is accessible,"
            + " you could try re-creating a new CloudSolrClient using working"
            + " solrUrl(s) or zkHost(s).");
  }

  @Override
  public Map<String, Object> getClusterProperties() {
    for (String nodeName : liveNodes) {
      String baseUrl = Utils.getBaseUrlForNodeName(nodeName, urlScheme);
      try (SolrClient client = getSolrClient(baseUrl)) {
        Map<String, Object> clusterProperties = new HashMap<>();
        fetchClusterState(client, null, clusterProperties);
        return clusterProperties;
      } catch (SolrServerException | BaseHttpSolrClient.RemoteSolrException | IOException e) {
        log.warn("Attempt to fetch cluster state from {} failed.", baseUrl, e);
      } catch (NotACollectionException e) {
        // not possible! (we passed in null for collection so it can't be an alias)
        throw new RuntimeException(
            "null should never cause NotACollectionException in "
                + "fetchClusterState() Please report this as a bug!");
      }
    }
    throw new RuntimeException(
        "Tried fetching cluster state using the node names we knew of, i.e. "
            + liveNodes
            + ". However, "
            + "succeeded in obtaining the cluster state from none of them."
            + "If you think your Solr cluster is up and is accessible,"
            + " you could try re-creating a new CloudSolrClient using working"
            + " solrUrl(s) or zkHost(s).");
  }

  @Override
  public String getPolicyNameByCollection(String coll) {
    throw new UnsupportedOperationException(
        "Fetching cluster properties not supported"
            + " using the HttpClusterStateProvider. "
            + "ZkClientClusterStateProvider can be used for this."); // TODO
  }

  @Override
  public Object getClusterProperty(String propertyName) {
    if (propertyName.equals(ClusterState.URL_SCHEME)) {
      return this.urlScheme;
    }
    return getClusterProperties().get(propertyName);
  }

  @Override
  public void connect() {}

  public int getCacheTimeout() {
    return cacheTimeout;
  }

  public void setCacheTimeout(int cacheTimeout) {
    this.cacheTimeout = cacheTimeout;
  }

  // This exception is not meant to escape this class it should be caught and wrapped.
  private static class NotACollectionException extends Exception {}

  @Override
  public String getQuorumHosts() {
    if (this.liveNodes == null) {
      return null;
    }
    return String.join(",", this.liveNodes);
  }
}
