/*
 * Copyright (C) 2016-2022 ActionTech.
 * License: http://www.gnu.org/licenses/gpl.html GPL version 2 or higher.
 */

package com.actiontech.dble.cluster.general;

import com.actiontech.dble.backend.mysql.view.FileSystemRepository;
import com.actiontech.dble.backend.mysql.view.KVStoreRepository;
import com.actiontech.dble.backend.mysql.view.Repository;
import com.actiontech.dble.cluster.ClusterSender;
import com.actiontech.dble.cluster.DistributeLock;
import com.actiontech.dble.cluster.general.bean.ClusterAlertBean;
import com.actiontech.dble.cluster.general.bean.SubscribeRequest;
import com.actiontech.dble.cluster.general.bean.SubscribeReturnBean;
import com.actiontech.dble.cluster.general.kVtoXml.ClusterToXml;
import com.actiontech.dble.cluster.general.listener.ClusterClearKeyListener;
import com.actiontech.dble.cluster.general.response.*;
import com.actiontech.dble.cluster.path.ClusterPathUtil;
import com.actiontech.dble.cluster.values.OnlineType;
import com.actiontech.dble.config.model.SystemConfig;
import com.actiontech.dble.singleton.OnlineStatus;
import com.actiontech.dble.singleton.ProxyMeta;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * Created by szf on 2019/3/11.
 */
public abstract class AbstractConsulSender implements ClusterSender {
    public static final Logger LOGGER = LoggerFactory.getLogger(AbstractConsulSender.class);

    protected ConcurrentHashMap<String, Thread> lockMap = new ConcurrentHashMap<>();

    protected static final String ERROR_MSG = "ALL the url to cluster connect failure,";

    public abstract String getRenewThreadPrefix();

    /**
     * only init the connection preperties for the clusterSender
     * mainly used by shell to upload xml config
     * do not start any Thread in this function!
     */
    public abstract void initConInfo();

    /**
     * lock a path,so that other DbleServer in cluster can not hold the path in the same time
     * return sessionId of the lock,if the return String is not null or "" means lock success
     * the lock value is also import ,the lock also need to be regarded as a KV
     * <p>
     * NOTICE: the lock should only influences it self,the child path should be available to other DbleServer to writeDirectly
     *
     * @param path
     * @param value
     * @return
     * @throws Exception
     */
    public abstract String lock(String path, String value) throws Exception;


    /**
     * use the locked path and sessionId to unlock a path
     * and the KV of the lock should be removed
     *
     * @param key
     * @param sessionId
     */
    public abstract void unlockKey(String key, String sessionId);


    /**
     * alert something into cluster
     *
     * @param alert
     */
    public abstract void alert(ClusterAlertBean alert);

    /**
     * notify cluster some alert is resolved
     *
     * @param alert
     * @return
     */
    public abstract boolean alertResolve(ClusterAlertBean alert);

    public abstract SubscribeReturnBean subscribeKvPrefix(SubscribeRequest request) throws Exception;


    @Override
    public void createSelfTempNode(String path, String value) throws Exception {
        String selfPath;
        if (path.endsWith(ClusterPathUtil.SEPARATOR)) {
            selfPath = path + SystemConfig.getInstance().getInstanceName();
        } else {
            selfPath = path + ClusterPathUtil.SEPARATOR + SystemConfig.getInstance().getInstanceName();
        }
        setKV(selfPath, value);

        LOGGER.info("writeDirectly self node for path:" + selfPath);
    }


    @Override
    public void writeConfToCluster() throws Exception {
        ClusterClearKeyListener ucoreListen = new ClusterClearKeyListener(this);


        new XmlDbLoader().registerPrefixForUcore(ucoreListen);
        new XmlShardingLoader().registerPrefixForUcore(ucoreListen);
        new XmlUserLoader().registerPrefixForUcore(ucoreListen);
        new SequencePropertiesLoader().registerPrefixForUcore(ucoreListen);

        ucoreListen.initAllNode();
        new DbGroupHaResponse().notifyCluster();
    }

    @Override
    public DistributeLock createDistributeLock(String path, String value) {
        return new ConsulDistributeLock(path, value, this);
    }

    @Override
    public DistributeLock createDistributeLock(String path, String value, int maxErrorCnt) {
        return new ConsulDistributeLock(path, value, maxErrorCnt, this);
    }


    @Override
    public Map<String, OnlineType> getOnlineMap() {
        return ClusterToXml.getOnlineMap();
    }

    @Override
    public void forceResumePause() throws Exception {
        ClusterXmlLoader loader = ClusterToXml.getListener().getResponse(ClusterPathUtil.getPauseResultNodePath());
        loader.notifyCluster();
    }

    /**
     * only be called when the dble start with cluster disconnect
     * & dble connect to cluster after a while
     * init some of the status from cluster
     *
     * @throws Exception
     */
    protected void firstReturnToCluster() throws Exception {
        if (ProxyMeta.getInstance().getTmManager() != null) {
            if (ProxyMeta.getInstance().getTmManager().getRepository() != null &&
                    ProxyMeta.getInstance().getTmManager().getRepository() instanceof FileSystemRepository) {
                LOGGER.warn("Dble first reconnect to ucore ,local view repository change to KVStoreRepository");
                Repository newViewRepository = new KVStoreRepository();
                ProxyMeta.getInstance().getTmManager().setRepository(newViewRepository);
                Map<String, Map<String, String>> viewCreateSqlMap = newViewRepository.getViewCreateSqlMap();
                ProxyMeta.getInstance().getTmManager().reloadViewMeta(viewCreateSqlMap);
                //init online status
                LOGGER.warn("Dble first reconnect to ucore ,online status rebuild");
            }
            OnlineStatus.getInstance().nodeListenerInitClusterOnline();
        }
    }

    public List<String> fetchRenewThread() {
        String onlineStr = ClusterPathUtil.getOnlinePath(SystemConfig.getInstance().getInstanceName());
        List<String> renewThread = lockMap.values().stream().
                filter(c -> !c.getName().endsWith(onlineStr) && c.isAlive()).
                map(Thread::getName).collect(Collectors.toList());
        return renewThread;
    }

    public boolean killRenewThread(String path) {
        Thread renewThread = lockMap.get(path);
        if (renewThread != null) {
            if (renewThread.isAlive() && !renewThread.isInterrupted()) {
                renewThread.interrupt();
                LOGGER.info("manual kill cluster renew thread: [{}]", renewThread.getName());
            } else {
                if (!renewThread.isAlive()) {
                    LOGGER.info("try manual kill cluster renew thread: [{}], but it already terminated", renewThread.getName());
                } else {
                    LOGGER.info("try manual kill cluster renew thread: [{}], but it already killed", renewThread.getName());
                }
            }
            return true;
        }
        return false;
    }
}
