/**
 * 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.atlas.ha;

import org.apache.atlas.security.SecurityProperties;
import org.apache.commons.configuration.Configuration;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * A wrapper for getting configuration entries related to HighAvailability.
 */
public final class HAConfiguration {


    public static final String ATLAS_SERVER_ZK_ROOT_DEFAULT = "/apache_atlas";

    private HAConfiguration() {
    }

    public static final String ATLAS_SERVER_HA_PREFIX = "atlas.server.ha.";
    public static final String ZOOKEEPER_PREFIX = "zookeeper.";
    public static final String ATLAS_SERVER_HA_ZK_ROOT_KEY = ATLAS_SERVER_HA_PREFIX + ZOOKEEPER_PREFIX + "zkroot";
    public static final String ATLAS_SERVER_HA_ENABLED_KEY = ATLAS_SERVER_HA_PREFIX + "enabled";
    public static final String ATLAS_SERVER_ADDRESS_PREFIX = "atlas.server.address.";
    public static final String ATLAS_SERVER_IDS = "atlas.server.ids";
    public static final String HA_ZOOKEEPER_CONNECT = ATLAS_SERVER_HA_PREFIX + ZOOKEEPER_PREFIX + "connect";
    public static final int DEFAULT_ZOOKEEPER_CONNECT_SLEEPTIME_MILLIS = 1000;
    public static final String HA_ZOOKEEPER_RETRY_SLEEPTIME_MILLIS =
            ATLAS_SERVER_HA_PREFIX + ZOOKEEPER_PREFIX + "retry.sleeptime.ms";
    public static final String HA_ZOOKEEPER_NUM_RETRIES = ATLAS_SERVER_HA_PREFIX + ZOOKEEPER_PREFIX + "num.retries";
    public static final int DEFAULT_ZOOKEEPER_CONNECT_NUM_RETRIES = 3;
    public static final String HA_ZOOKEEPER_SESSION_TIMEOUT_MS =
            ATLAS_SERVER_HA_PREFIX + ZOOKEEPER_PREFIX + "session.timeout.ms";
    public static final int DEFAULT_ZOOKEEPER_SESSION_TIMEOUT_MILLIS = 20000;
    public static final String HA_ZOOKEEPER_ACL = ATLAS_SERVER_HA_PREFIX + ZOOKEEPER_PREFIX + "acl";
    public static final String HA_ZOOKEEPER_AUTH = ATLAS_SERVER_HA_PREFIX + ZOOKEEPER_PREFIX + "auth";

    /**
     * Return whether HA is enabled or not.
     * @param configuration underlying configuration instance
     * @return
     */
    public static boolean isHAEnabled(Configuration configuration) {
        boolean ret = false;

        if (configuration.containsKey(HAConfiguration.ATLAS_SERVER_HA_ENABLED_KEY)) {
            ret = configuration.getBoolean(ATLAS_SERVER_HA_ENABLED_KEY);
        } else {
            String[] ids = configuration.getStringArray(HAConfiguration.ATLAS_SERVER_IDS);

            ret = ids != null && ids.length > 1;
        }

        return ret;
    }

    /**
     * Get the web server address that a server instance with the passed ID is bound to.
     *
     * This method uses the property {@link SecurityProperties#TLS_ENABLED} to determine whether
     * the URL is http or https.
     *
     * @param configuration underlying configuration
     * @param serverId serverId whose host:port property is picked to build the web server address.
     * @return
     */
    public static String getBoundAddressForId(Configuration configuration, String serverId) {
        String hostPort = configuration.getString(ATLAS_SERVER_ADDRESS_PREFIX +serverId);
        boolean isSecure = configuration.getBoolean(SecurityProperties.TLS_ENABLED);
        String protocol = (isSecure) ? "https://" : "http://";
        return protocol + hostPort;
    }

    public static List<String> getServerInstances(Configuration configuration) {
        String[] serverIds = configuration.getStringArray(ATLAS_SERVER_IDS);
        List<String> serverInstances = new ArrayList<>(serverIds.length);
        for (String serverId : serverIds) {
            serverInstances.add(getBoundAddressForId(configuration, serverId));
        }
        return serverInstances;
    }

    /**
     * A collection of Zookeeper specific configuration that is used by High Availability code.
     */
    public static class ZookeeperProperties {
        private String connectString;
        private String zkRoot;
        private int retriesSleepTimeMillis;
        private int numRetries;
        private int sessionTimeout;
        private String acl;
        private String auth;

        public ZookeeperProperties(String connectString, String zkRoot, int retriesSleepTimeMillis, int numRetries,
                                   int sessionTimeout, String acl, String auth) {
            this.connectString = connectString;
            this.zkRoot = zkRoot;
            this.retriesSleepTimeMillis = retriesSleepTimeMillis;
            this.numRetries = numRetries;
            this.sessionTimeout = sessionTimeout;
            this.acl = acl;
            this.auth = auth;
        }

        public String getConnectString() {
            return connectString;
        }

        public int getRetriesSleepTimeMillis() {
            return retriesSleepTimeMillis;
        }

        public int getNumRetries() {
            return numRetries;
        }

        public int getSessionTimeout() {
            return sessionTimeout;
        }

        public String getAcl() {
            return acl;
        }

        public String getAuth() {
            return auth;
        }

        public String getZkRoot() {
            return zkRoot;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            ZookeeperProperties that = (ZookeeperProperties) o;
            return retriesSleepTimeMillis == that.retriesSleepTimeMillis &&
                    numRetries == that.numRetries &&
                    sessionTimeout == that.sessionTimeout &&
                    Objects.equals(connectString, that.connectString) &&
                    Objects.equals(zkRoot, that.zkRoot) &&
                    Objects.equals(acl, that.acl) &&
                    Objects.equals(auth, that.auth);
        }

        @Override
        public int hashCode() {
            return Objects.hash(connectString, zkRoot, retriesSleepTimeMillis, numRetries, sessionTimeout, acl, auth);
        }

        public boolean hasAcl() {
            return getAcl()!=null;
        }

        public boolean hasAuth() {
            return getAuth()!=null;
        }
    }

    public static ZookeeperProperties getZookeeperProperties(Configuration configuration) {
        String zookeeperConnectString = configuration.getString("atlas.kafka." + ZOOKEEPER_PREFIX + "connect");
        if (configuration.containsKey(HA_ZOOKEEPER_CONNECT)) {
            zookeeperConnectString = configuration.getString(HA_ZOOKEEPER_CONNECT);
        }

        String zkRoot = configuration.getString(ATLAS_SERVER_HA_ZK_ROOT_KEY, ATLAS_SERVER_ZK_ROOT_DEFAULT);
        int retriesSleepTimeMillis = configuration.getInt(HA_ZOOKEEPER_RETRY_SLEEPTIME_MILLIS,
                DEFAULT_ZOOKEEPER_CONNECT_SLEEPTIME_MILLIS);

        int numRetries = configuration.getInt(HA_ZOOKEEPER_NUM_RETRIES, DEFAULT_ZOOKEEPER_CONNECT_NUM_RETRIES);

        int sessionTimeout = configuration.getInt(HA_ZOOKEEPER_SESSION_TIMEOUT_MS,
                DEFAULT_ZOOKEEPER_SESSION_TIMEOUT_MILLIS);

        String acl = configuration.getString(HA_ZOOKEEPER_ACL);
        String auth = configuration.getString(HA_ZOOKEEPER_AUTH);
        return new ZookeeperProperties(zookeeperConnectString, zkRoot, retriesSleepTimeMillis, numRetries,
                sessionTimeout, acl, auth);
    }
}