/*
 * Copyright (C) 2016-2022 ActionTech.
 * based on code by MyCATCopyrightHolder Copyright (c) 2013, OpenCloudDB/MyCAT.
 * License: http://www.gnu.org/licenses/gpl.html GPL version 2 or higher.
 */
package com.actiontech.dble.services.manager.handler;

import com.actiontech.dble.DbleServer;
import com.actiontech.dble.cluster.AbstractGeneralListener;
import com.actiontech.dble.cluster.ClusterGeneralConfig;
import com.actiontech.dble.config.ErrorCode;
import com.actiontech.dble.config.model.ClusterConfig;
import com.actiontech.dble.net.IOProcessor;
import com.actiontech.dble.net.mysql.OkPacket;
import com.actiontech.dble.net.service.AbstractService;
import com.actiontech.dble.services.FrontendService;
import com.actiontech.dble.services.manager.ManagerService;
import com.actiontech.dble.util.DelayServiceControl;
import org.apache.curator.shaded.com.google.common.collect.Lists;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.List;
import java.util.StringJoiner;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author mycat
 */
public final class ClusterManageHandler {
    private static final Logger LOGGER = LogManager.getLogger(ClusterManageHandler.class);
    private static final Pattern PATTERN_DETACH = Pattern.compile("@@detach(\\s*timeout\\s*=\\s*([0-9]+))?\\s*$", Pattern.CASE_INSENSITIVE);
    private static final Pattern PATTERN_ATTACH = Pattern.compile("@@attach(\\s*timeout\\s*=\\s*([0-9]+))?\\s*$", Pattern.CASE_INSENSITIVE);
    private static final int DEFAULT_PAUSE_TIME = 10;
    private static volatile boolean detached = false;

    private ClusterManageHandler() {
    }


    public static synchronized void handle(String sql, ManagerService selfService, int offset) {

        String options = sql.substring(offset).trim();
        {
            Pattern pattern = PATTERN_DETACH;
            final Matcher matcher = pattern.matcher(options);
            if (matcher.matches()) {
                final String timeout = matcher.group(2);
                if (timeout == null) {
                    detach(DEFAULT_PAUSE_TIME, selfService);
                } else {
                    detach(Integer.parseInt(timeout), selfService);
                }

                return;
            }


        }
        {
            Pattern pattern = PATTERN_ATTACH;
            final Matcher matcher = pattern.matcher(options);
            if (matcher.matches()) {
                final String timeout = matcher.group(2);
                if (timeout == null) {
                    attach(DEFAULT_PAUSE_TIME, selfService);
                } else {
                    attach(Integer.parseInt(timeout), selfService);
                }
                return;
            }
        }
        selfService.writeErrMessage(ErrorCode.ER_YES, "Syntax incorrect");

    }

    public static boolean isDetached() {
        return detached;
    }

    private static synchronized void attach(int maxPauseTime, ManagerService selfService) {
        if (!ClusterConfig.getInstance().isClusterEnable()) {
            selfService.writeErrMessage(ErrorCode.ER_YES, "illegal state: attach cluster operation only can be used in cluster mode");
            return;
        }
        if (!detached) {
            selfService.writeErrMessage(ErrorCode.ER_YES, "illegal state: cluster is not detached");
            return;
        }


        //block other manager&ddl session send query
        ClusterGeneralConfig.getInstance().setNeedBlocked(true);
        try {
            //if the session is execute query already ,wait until the response return.
            LOGGER.info("cluster attach begin waiting");
            WaitResult waitResult = waitOtherSessionBlocked(selfService, maxPauseTime, false);
            LOGGER.info("cluster attach after waiting");
            if (!waitResult.success) {
                selfService.writeErrMessage(ErrorCode.ER_YES, "attach cluster pause timeout. " + waitResult.errorMessage);
                return;
            }
            ClusterGeneralConfig.getInstance().getClusterSender().attachCluster();
            detached = false;
            LOGGER.info("cluster attach complete");
        } catch (Exception e) {
            LOGGER.info("attach cluster err", e);
            selfService.writeErrMessage(ErrorCode.ER_YES, "attach cluster err：" + e);
            return;
        } finally {
            //resume task consume
            ClusterGeneralConfig.getInstance().setNeedBlocked(false);
            DelayServiceControl.getInstance().wakeUpAllBlockServices();
        }
        OkPacket okPacket = OkPacket.getDefault();
        okPacket.write(selfService.getConnection());

    }


    private static synchronized void detach(int maxPauseTime, ManagerService selfService) {
        if (!ClusterConfig.getInstance().isClusterEnable()) {
            selfService.writeErrMessage(ErrorCode.ER_YES, "illegal state: attach cluster operation only can be used in cluster mode");
            return;
        }
        if (detached) {
            selfService.writeErrMessage(ErrorCode.ER_YES, "illegal state: cluster is already detached");
            return;
        }


        //block other manager&ddl session send query
        ClusterGeneralConfig.getInstance().setNeedBlocked(true);
        try {
            //if the session is execute query already ,wait until the response return.
            LOGGER.info("cluster detach begin waiting");
            WaitResult waitResult = waitOtherSessionBlocked(selfService, maxPauseTime, true);
            LOGGER.info("cluster detach after waiting");
            if (!waitResult.success) {
                selfService.writeErrMessage(ErrorCode.ER_YES, "detach cluster pause timeout. " + waitResult.errorMessage);
                return;
            }

            ClusterGeneralConfig.getInstance().getClusterSender().detachCluster();
            detached = true;
            LOGGER.info("cluster detach complete");
        } catch (Exception e) {
            LOGGER.info("detach cluster err, try to resume", e);
            boolean resumeResult = false;
            try {
                ClusterGeneralConfig.getInstance().getClusterSender().attachCluster();
                LOGGER.info("detach resume success");
                resumeResult = true;
            } catch (Exception e2) {
                LOGGER.info("detach cluster resume err", e2);
            }
            selfService.writeErrMessage(ErrorCode.ER_YES, "detach cluster err：" + e + ", resume status:" + resumeResult);
            return;
        } finally {
            //resume task consume
            ClusterGeneralConfig.getInstance().setNeedBlocked(false);
            DelayServiceControl.getInstance().wakeUpAllBlockServices();
        }
        OkPacket okPacket = OkPacket.getDefault();
        okPacket.write(selfService.getConnection());


    }

    /**
     * return false is wait timeout
     *
     * @param maxPauseTime
     * @return
     */
    private static WaitResult waitOtherSessionBlocked(ManagerService selfService, int maxPauseTime, boolean isDetach) {
        List<FrontendService> waitServices = Lists.newArrayList();

        final long startTime = System.currentTimeMillis();
        /*
        if query is executing ,the isDoing must be true, then wait for complete.
        if the query is going to execute, it will be block,won't set Doing from false to true.
         */
        for (IOProcessor p : DbleServer.getInstance().getFrontProcessors()) {
            p.getFrontends().
                    values().
                    forEach(fc -> {
                        if (!fc.isAuthorized()) {
                            return;
                        }
                        final AbstractService frontService = fc.getService();
                        if (frontService == selfService) {
                            return;
                        }
                        if (frontService instanceof FrontendService) {
                            if (((FrontendService) frontService).getClusterDelayService().isDoing()) {
                                waitServices.add(((FrontendService) frontService));
                            }
                        }
                    });
        }

        while (System.currentTimeMillis() - startTime < maxPauseTime * 1000L) {

            waitServices.removeIf(tmpService -> !tmpService.getClusterDelayService().isDoing());
            if (waitServices.isEmpty()) {
                break;
            }
            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10));
        }
        if (!waitServices.isEmpty()) {
            final StringJoiner msg = new StringJoiner(",", "{", "}");
            waitServices.forEach(service -> {
                msg.add(String.valueOf(service.getConnection().getId()));
            });
            LOGGER.warn("cluster operation timeout because of {} connections, connections id is {}", waitServices.size(), msg);
            return WaitResult.ofError("some frontend connection is doing operation.");
        }
        if (!isDetach) {
            return WaitResult.ofSuccess();
        }

        /*
        detach should also guarantee ucore/zk listener isn't doing.
         */
        ClusterGeneralConfig.getInstance().getClusterSender().markDetach(true);
        while (System.currentTimeMillis() - startTime < maxPauseTime * 1000L && AbstractGeneralListener.getDoingCount().get() != 0) {
            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10));
        }

        if (AbstractGeneralListener.getDoingCount().get() != 0) {
            LOGGER.warn("cluster operation timeout because of other server is doing cluster-related operation");
            ClusterGeneralConfig.getInstance().getClusterSender().markDetach(false);
            return WaitResult.ofError("some backend operation is doing. please check the dble.log");
        }
        return WaitResult.ofSuccess();
    }


    static final class WaitResult {
        boolean success;
        String errorMessage;

        private WaitResult(String errorMessage) {
            this.success = false;
            this.errorMessage = errorMessage;
        }

        private WaitResult(boolean success) {
            this.success = success;
        }

        public static WaitResult ofSuccess() {
            return new WaitResult(true);
        }

        public static WaitResult ofError(String reason) {
            return new WaitResult(reason);
        }

    }
}
