/*
 * Copyright (C) 2016-2022 ActionTech.
 * License: http://www.gnu.org/licenses/gpl.html GPL version 2 or higher.
 */
package com.actiontech.dble.services.manager.response;

import com.actiontech.dble.DbleServer;
import com.actiontech.dble.config.ErrorCode;
import com.actiontech.dble.net.IOProcessor;
import com.actiontech.dble.net.connection.BackendConnection;
import com.actiontech.dble.net.connection.FrontendConnection;
import com.actiontech.dble.net.mysql.OkPacket;
import com.actiontech.dble.net.service.AbstractService;
import com.actiontech.dble.plan.common.exception.MySQLOutPutException;
import com.actiontech.dble.route.RouteResultsetNode;
import com.actiontech.dble.services.manager.ManagerService;
import com.actiontech.dble.services.manager.handler.PacketResult;
import com.actiontech.dble.services.mysqlsharding.ShardingService;
import com.actiontech.dble.singleton.PauseShardingNodeManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public final class PauseStart {
    private static final Pattern PATTERN_FOR_PAUSE = Pattern.compile("@@shardingNode\\s*=\\s*'([a-zA-Z_0-9,]+)'\\s*and\\s*timeout\\s*=\\s*([0-9]+)\\s*(,\\s*queue\\s*=\\s*([0-9]+)){0,1}\\s*(,\\s*wait_limit\\s*=\\s*([0-9]+)){0,1}\\s*$", Pattern.CASE_INSENSITIVE);
    private static final OkPacket OK = new OkPacket();
    private static final int DEFAULT_CONNECTION_TIME_OUT = 120000;
    private static final int DEFAULT_QUEUE_LIMIT = 200;
    private static final Logger LOGGER = LoggerFactory.getLogger(PauseStart.class);

    private PauseStart() {
    }

    static {
        OK.setPacketId(1);
        OK.setAffectedRows(1L);
        OK.setServerStatus(2);
    }

    public static void execute(final ManagerService c, final String sql, final int offset) {
        c.getClusterDelayService().markDoingOrDelay(true);
        DbleServer.getInstance().getComplexQueryExecutor().execute(new Runnable() {
            @Override
            public void run() {
                pause(c, sql, offset);
            }
        });

    }


    public static void pause(ManagerService service, String sql, final int offset) {
        String options = sql.substring(offset).trim();
        Matcher ma = PATTERN_FOR_PAUSE.matcher(options);
        PacketResult packetResult = new PacketResult();
        try {
            if (!ma.matches()) {
                packetResult.setSuccess(false);
                packetResult.setErrorMsg("The sql did not match pause @@shardingNode ='dn......' and timeout = ([0-9]+)");
                packetResult.setErrorCode(ErrorCode.ER_UNKNOWN_ERROR);
                return;
            }
            String shardingNode = ma.group(1);
            int connectionTimeOut = ma.group(6) == null ? DEFAULT_CONNECTION_TIME_OUT : Integer.parseInt(ma.group(6)) * 1000;
            int queueLimit = ma.group(4) == null ? DEFAULT_QUEUE_LIMIT : Integer.parseInt(ma.group(4));
            Set<String> shardingNodes = new HashSet<>(Arrays.asList(shardingNode.split(",")));
            //check shardingNode
            for (String singleDn : shardingNodes) {
                if (DbleServer.getInstance().getConfig().getShardingNodes().get(singleDn) == null) {
                    packetResult.setSuccess(false);
                    packetResult.setErrorMsg("ShardingNode " + singleDn + " did not exists");
                    packetResult.setErrorCode(ErrorCode.ER_UNKNOWN_ERROR);
                    return;
                }
            }

            if (!PauseShardingNodeManager.getInstance().getDistributeLock()) {
                packetResult.setSuccess(false);
                packetResult.setErrorMsg("Other node is doing pause operation concurrently");
                packetResult.setErrorCode(ErrorCode.ER_UNKNOWN_ERROR);
                return;
            }


            try {
                try {
                    //clusterPauseNotice
                    PauseShardingNodeManager.getInstance().clusterPauseNotice(shardingNode, connectionTimeOut, queueLimit);
                } catch (Exception e) {
                    LOGGER.warn("pause failed", e);
                    packetResult.setSuccess(false);
                    packetResult.setErrorMsg(e.getMessage());
                    packetResult.setErrorCode(ErrorCode.ER_YES);
                    return;
                }


                PauseShardingNodeManager.getInstance().startPausing(connectionTimeOut, shardingNodes, shardingNode, queueLimit);

                //self pause the shardingNode
                long timeOut = Long.parseLong(ma.group(2)) * 1000;
                long beginTime = System.currentTimeMillis();
                boolean recycleFinish = waitForSelfPause(beginTime, timeOut, shardingNodes);

                LOGGER.info("wait finished " + recycleFinish);
                if (!recycleFinish) {
                    packetResult.setSuccess(false);
                    packetResult.setErrorCode(ErrorCode.ER_YES);
                    if (PauseShardingNodeManager.getInstance().tryResume()) {
                        try {
                            PauseShardingNodeManager.getInstance().resumeCluster();
                        } catch (Exception e) {
                            LOGGER.warn("resume cause error", e);
                        }
                        packetResult.setErrorMsg("The backend connection recycle failure, try it later");
                    } else {
                        packetResult.setErrorMsg("Pause resume when recycle connection, pause revert");
                    }
                } else {
                    try {
                        if (PauseShardingNodeManager.getInstance().waitForCluster(beginTime, timeOut, packetResult)) {
                            LOGGER.info("call pause success");
                        }
                    } catch (Exception e) {
                        LOGGER.warn("wait for other node failed.", e);
                        packetResult.setSuccess(false);
                        packetResult.setErrorMsg(e.getMessage());
                        packetResult.setErrorCode(ErrorCode.ER_YES);
                    }
                }
            } catch (MySQLOutPutException e) {
                packetResult.setSuccess(false);
                packetResult.setErrorMsg(e.getMessage());
                packetResult.setErrorCode(ErrorCode.ER_YES);
            } catch (Exception e) {
                packetResult.setSuccess(false);
                packetResult.setErrorMsg("Pause operation cause error: " + e.getMessage());
                packetResult.setErrorCode(ErrorCode.ER_YES);
            } finally {
                PauseShardingNodeManager.getInstance().releaseDistributeLock();
            }
        } finally {
            writePacket(packetResult.isSuccess(), service, packetResult.getErrorMsg(), packetResult.getErrorCode());
        }
    }

    private static void writePacket(boolean isSuccess, ManagerService service, String errorMsg, int errorCode) {
        if (isSuccess) {
            OK.write(service.getConnection());
        } else {
            service.writeErrMessage(errorCode, errorMsg);
        }
    }


    public static boolean waitForSelfPause(long beginTime, long timeOut, Set<String> shardingNodes) {
        boolean recycleFinish = false;
        while ((System.currentTimeMillis() - beginTime < timeOut) && PauseShardingNodeManager.getInstance().getIsPausing().get()) {
            boolean nextTurn = false;
            for (IOProcessor processor : DbleServer.getInstance().getFrontProcessors()) {
                for (Map.Entry<Long, FrontendConnection> entry : processor.getFrontends().entrySet()) {
                    AbstractService service = entry.getValue().getService();
                    if ((service instanceof ShardingService)) {
                        ShardingService shardingService = (ShardingService) service;
                        for (Map.Entry<RouteResultsetNode, BackendConnection> conEntry : shardingService.getSession2().getTargetMap().entrySet()) {
                            if (shardingNodes.contains(conEntry.getKey().getName())) {
                                nextTurn = true;
                                break;
                            }
                        }
                        if (nextTurn) {
                            break;
                        }
                    }
                }
                if (nextTurn) {
                    break;
                }
            }
            if (!nextTurn) {
                recycleFinish = true;
                break;
            } else {
                LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10L));
            }
        }
        return recycleFinish;
    }
}
