/*
 * 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.storm.zookeeper;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import org.apache.storm.callback.WatcherCallBack;
import org.apache.storm.cluster.DaemonType;
import org.apache.storm.cluster.VersionedData;
import org.apache.storm.shade.org.apache.curator.framework.CuratorFramework;
import org.apache.storm.shade.org.apache.curator.framework.api.CuratorEvent;
import org.apache.storm.shade.org.apache.curator.framework.api.CuratorEventType;
import org.apache.storm.shade.org.apache.curator.framework.api.CuratorListener;
import org.apache.storm.shade.org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.storm.shade.org.apache.zookeeper.CreateMode;
import org.apache.storm.shade.org.apache.zookeeper.KeeperException;
import org.apache.storm.shade.org.apache.zookeeper.WatchedEvent;
import org.apache.storm.shade.org.apache.zookeeper.data.ACL;
import org.apache.storm.shade.org.apache.zookeeper.data.Stat;
import org.apache.storm.utils.CuratorUtils;
import org.apache.storm.utils.Utils;
import org.apache.storm.utils.ZookeeperAuthInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ClientZookeeper {

    // A singleton instance allows us to mock delegated static methods in our
    // tests by subclassing.
    private static final ClientZookeeper INSTANCE = new ClientZookeeper();
    private static Logger LOG = LoggerFactory.getLogger(ClientZookeeper.class);
    private static ClientZookeeper _instance = INSTANCE;

    /**
     * Provide an instance of this class for delegates to use.  To mock out delegated methods, provide an instance of a subclass that
     * overrides the implementation of the delegated method.
     *
     * @param u a ClientZookeeper instance
     */
    public static void setInstance(ClientZookeeper u) {
        _instance = u;
    }

    /**
     * Resets the singleton instance to the default. This is helpful to reset the class to its original functionality when mocking is no
     * longer desired.
     */
    public static void resetInstance() {
        _instance = INSTANCE;
    }

    public static void mkdirs(CuratorFramework zk, String path, List<ACL> acls) {
        _instance.mkdirsImpl(zk, path, acls);
    }

    public static CuratorFramework mkClient(Map<String, Object> conf, List<String> servers, Object port,
                                            String root, final WatcherCallBack watcher, Map<String, Object> authConf, DaemonType type) {
        return _instance.mkClientImpl(conf, servers, port, root, watcher, authConf, type);
    }

    // Deletes the state inside the zookeeper for a key, for which the
    // contents of the key starts with nimbus host port information
    public static void deleteNodeBlobstore(CuratorFramework zk, String parentPath, String hostPortInfo) {
        String normalizedParentPath = normalizePath(parentPath);
        List<String> childPathList = null;
        if (existsNode(zk, normalizedParentPath, false)) {
            childPathList = getChildren(zk, normalizedParentPath, false);
            for (String child : childPathList) {
                if (child.startsWith(hostPortInfo)) {
                    LOG.debug("deleteNode child {}", child);
                    deleteNode(zk, normalizedParentPath + "/" + child);
                }
            }
        }
    }

    public static String createNode(CuratorFramework zk, String path, byte[] data, CreateMode mode, List<ACL> acls) {
        String ret = null;
        try {
            String npath = normalizePath(path);
            ret = zk.create().creatingParentsIfNeeded().withMode(mode).withACL(acls).forPath(npath, data);
        } catch (Exception e) {
            throw Utils.wrapInRuntime(e);
        }
        return ret;
    }

    public static String createNode(CuratorFramework zk, String path, byte[] data, List<ACL> acls) {
        return createNode(zk, path, data, CreateMode.PERSISTENT, acls);
    }

    public static List<String> tokenizePath(String path) {
        String[] toks = path.split("/");
        java.util.ArrayList<String> rtn = new ArrayList<String>();
        for (String str : toks) {
            if (!str.isEmpty()) {
                rtn.add(str);
            }
        }
        return rtn;
    }

    public static String toksToPath(List<String> toks) {
        StringBuffer buff = new StringBuffer();
        buff.append("/");
        int size = toks.size();
        for (int i = 0; i < size; i++) {
            buff.append(toks.get(i));
            if (i < (size - 1)) {
                buff.append("/");
            }
        }
        return buff.toString();
    }

    public static String normalizePath(String path) {
        String rtn = toksToPath(tokenizePath(path));
        return rtn;
    }

    public static boolean existsNode(CuratorFramework zk, String path, boolean watch) {
        Stat stat = null;
        try {
            if (watch) {
                stat = zk.checkExists().watched().forPath(normalizePath(path));
            } else {
                stat = zk.checkExists().forPath(normalizePath(path));
            }
        } catch (Exception e) {
            throw Utils.wrapInRuntime(e);
        }
        return stat != null;
    }

    public static void deleteNode(CuratorFramework zk, String path) {
        try {
            String npath = normalizePath(path);
            if (existsNode(zk, npath, false)) {
                zk.delete().deletingChildrenIfNeeded().forPath(normalizePath(path));
            }
        } catch (Exception e) {
            if (Utils.exceptionCauseIsInstanceOf(KeeperException.NodeExistsException.class, e)) {
                // do nothing
                LOG.info("delete {} failed.", path, e);
            } else {
                throw Utils.wrapInRuntime(e);
            }
        }
    }

    public static String parentPath(String path) {
        List<String> toks = tokenizePath(path);
        int size = toks.size();
        if (size > 0) {
            toks.remove(size - 1);
        }
        return toksToPath(toks);
    }

    public static boolean exists(CuratorFramework zk, String path, boolean watch) {
        return existsNode(zk, path, watch);
    }

    public static Stat setData(CuratorFramework zk, String path, byte[] data) {
        try {
            String npath = normalizePath(path);
            return zk.setData().forPath(npath, data);
        } catch (Exception e) {
            throw Utils.wrapInRuntime(e);
        }
    }

    public static Integer getVersion(CuratorFramework zk, String path, boolean watch) throws Exception {
        String npath = normalizePath(path);
        Stat stat = null;
        if (existsNode(zk, npath, watch)) {
            if (watch) {
                stat = zk.checkExists().watched().forPath(npath);
            } else {
                stat = zk.checkExists().forPath(npath);
            }
        }
        return stat == null ? null : Integer.valueOf(stat.getVersion());
    }

    public static List<String> getChildren(CuratorFramework zk, String path, boolean watch) {
        try {
            String npath = normalizePath(path);
            if (watch) {
                return zk.getChildren().watched().forPath(npath);
            } else {
                return zk.getChildren().forPath(npath);
            }
        } catch (Exception e) {
            throw Utils.wrapInRuntime(e);
        }
    }

    public static byte[] getData(CuratorFramework zk, String path, boolean watch) {
        try {
            String npath = normalizePath(path);
            if (existsNode(zk, npath, watch)) {
                if (watch) {
                    return zk.getData().watched().forPath(npath);
                } else {
                    return zk.getData().forPath(npath);
                }
            }
        } catch (Exception e) {
            if (Utils.exceptionCauseIsInstanceOf(KeeperException.NoNodeException.class, e)) {
                // this is fine b/c we still have a watch from the successful exists call
            } else {
                throw Utils.wrapInRuntime(e);
            }
        }
        return null;
    }

    /**
     * Get the data along with a version.
     *
     * @param zk    the zk instance to use
     * @param path  the path to get it from
     * @param watch should a watch be enabled
     * @return null if no data is found, else the data with the version.
     */
    public static VersionedData<byte[]> getDataWithVersion(CuratorFramework zk, String path, boolean watch) {
        VersionedData<byte[]> data = null;
        try {
            byte[] bytes = null;
            Stat stats = new Stat();
            String npath = normalizePath(path);
            if (existsNode(zk, npath, watch)) {
                if (watch) {
                    bytes = zk.getData().storingStatIn(stats).watched().forPath(npath);
                } else {
                    bytes = zk.getData().storingStatIn(stats).forPath(npath);
                }
                if (bytes != null) {
                    int version = stats.getVersion();
                    data = new VersionedData<>(version, bytes);
                }
            }
        } catch (Exception e) {
            if (Utils.exceptionCauseIsInstanceOf(KeeperException.NoNodeException.class, e)) {
                // this is fine b/c we still have a watch from the successful exists call
            } else {
                Utils.wrapInRuntime(e);
            }
        }
        return data;
    }

    public static void addListener(CuratorFramework zk, ConnectionStateListener listener) {
        zk.getConnectionStateListenable().addListener(listener);
    }

    public static void syncPath(CuratorFramework zk, String path) {
        try {
            zk.sync().forPath(normalizePath(path));
        } catch (Exception e) {
            throw Utils.wrapInRuntime(e);
        }
    }

    public void mkdirsImpl(CuratorFramework zk, String path, List<ACL> acls) {
        String npath = ClientZookeeper.normalizePath(path);
        if (npath.equals("/")) {
            return;
        }
        if (ClientZookeeper.existsNode(zk, npath, false)) {
            return;
        }
        byte[] byteArray = new byte[1];
        byteArray[0] = (byte) 7;
        try {
            ClientZookeeper.createNode(zk, npath, byteArray, CreateMode.PERSISTENT, acls);
        } catch (Exception e) {
            if (Utils.exceptionCauseIsInstanceOf(KeeperException.NodeExistsException.class, e)) {
                // this can happen when multiple clients doing mkdir at same time
            }
        }
    }

    public CuratorFramework mkClientImpl(Map<String, Object> conf, List<String> servers, Object port, String root,
                                         final WatcherCallBack watcher, Map<String, Object> authConf, DaemonType type) {
        CuratorFramework fk;
        if (authConf != null) {
            fk = CuratorUtils.newCurator(conf, servers, port, root, new ZookeeperAuthInfo(authConf), type.getDefaultZkAcls(conf));
        } else {
            fk = CuratorUtils.newCurator(conf, servers, port, root, null, type.getDefaultZkAcls(conf));
        }

        fk.getCuratorListenable().addListener((unused, e) -> {
            if (e.getType().equals(CuratorEventType.WATCHED)) {
                WatchedEvent event = e.getWatchedEvent();
                watcher.execute(event.getState(), event.getType(), event.getPath());
            }
        });
        LOG.info("Starting ZK Curator");
        fk.start();
        return fk;
    }
}