/*
 * 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.backend.heartbeat;

import com.actiontech.dble.DbleServer;
import com.actiontech.dble.alarm.AlarmCode;
import com.actiontech.dble.alarm.Alert;
import com.actiontech.dble.alarm.AlertUtil;
import com.actiontech.dble.alarm.ToResolveContainer;
import com.actiontech.dble.backend.datasource.PhysicalDbInstance;
import com.actiontech.dble.backend.mysql.VersionUtil;
import com.actiontech.dble.config.helper.GetAndSyncDbInstanceKeyVariables;
import com.actiontech.dble.config.helper.KeyVariables;
import com.actiontech.dble.config.model.SystemConfig;
import com.actiontech.dble.sqlengine.OneRawSQLQueryResultHandler;
import com.actiontech.dble.sqlengine.SQLQueryResult;
import com.actiontech.dble.sqlengine.SQLQueryResultListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * @author mycat
 */
public class MySQLDetector implements SQLQueryResultListener<SQLQueryResult<Map<String, String>>> {
    public static final Logger LOGGER = LoggerFactory.getLogger(MySQLDetector.class);
    private static final String[] MYSQL_SLAVE_STATUS_COLS = new String[]{
            "Seconds_Behind_Master",
            "Slave_IO_Running",
            "Slave_SQL_Running",
            "Slave_IO_State",
            "Master_Host",
            "Master_User",
            "Master_Port",
            "Connect_Retry",
            "Last_IO_Error"};
    private static final String[] MYSQL_READ_ONLY_COLS = new String[]{"@@read_only"};

    private final MySQLHeartbeat heartbeat;
    private volatile long lastSendQryTime;
    private volatile long lastReceivedQryTime;
    private final HeartbeatSQLJob sqlJob;

    public MySQLDetector(MySQLHeartbeat heartbeat) {
        this.heartbeat = heartbeat;
        String[] fetchCols = {};
        if (heartbeat.getSource().getDbGroupConfig().isShowSlaveSql()) {
            fetchCols = MYSQL_SLAVE_STATUS_COLS;
        } else if (heartbeat.getSource().getDbGroupConfig().isSelectReadOnlySql()) {
            fetchCols = MYSQL_READ_ONLY_COLS;
        }

        OneRawSQLQueryResultHandler resultHandler = new OneRawSQLQueryResultHandler(fetchCols, this);
        this.sqlJob = new HeartbeatSQLJob(heartbeat, resultHandler);
    }

    boolean isHeartbeatTimeout() {
        return System.currentTimeMillis() > Math.max(lastSendQryTime, lastReceivedQryTime) + heartbeat.getHeartbeatTimeout();
    }

    long getLastReceivedQryTime() {
        return lastReceivedQryTime;
    }

    public void heartbeat() {
        if (lastSendQryTime <= 0) {
            lastSendQryTime = System.currentTimeMillis();
            heartbeat.getSource().createConnectionSkipPool(null, sqlJob);
        } else {
            lastSendQryTime = System.currentTimeMillis();
            sqlJob.execute();
        }
    }

    public void quit() {
        sqlJob.terminate();
    }

    public boolean isQuit() {
        return sqlJob.isQuit();
    }

    @Override
    public void onResult(SQLQueryResult<Map<String, String>> result) {
        lastReceivedQryTime = System.currentTimeMillis();
        heartbeat.getRecorder().set((lastReceivedQryTime - lastSendQryTime));
        if (result.isSuccess()) {
            PhysicalDbInstance source = heartbeat.getSource();
            Map<String, String> resultResult = result.getResult();
            if (source.getDbGroupConfig().isShowSlaveSql()) {
                setStatusBySlave(source, resultResult);
            } else if (source.getDbGroupConfig().isSelectReadOnlySql()) {
                setStatusByReadOnly(source, resultResult);
            } else {
                setStatusForNormalHeartbeat(source);
            }
        }
    }

    private void setStatusForNormalHeartbeat(PhysicalDbInstance source) {
        if (checkRecoverFail(source)) return;
        heartbeat.setResult(MySQLHeartbeatStatus.OK);
    }

    /**
     * if recover failed, return true
     */
    private boolean checkRecoverFail(PhysicalDbInstance source) {
        if (heartbeat.isStop()) {
            LOGGER.warn("heartbeat[{}] had been stop", source.getConfig().getUrl());
            return true;
        }
        if (heartbeat.getStatus() == MySQLHeartbeatStatus.OK) { // ok->ok
            if (!heartbeat.getSource().isSalveOrRead() && source.isReadOnly()) { // writehost checkRecoverFail read only status is back?
                GetAndSyncDbInstanceKeyVariables task = new GetAndSyncDbInstanceKeyVariables(source, true);
                KeyVariables variables = task.call();
                if (variables != null) {
                    source.setReadOnly(variables.isReadOnly());
                } else {
                    LOGGER.warn("GetAndSyncDbInstanceKeyVariables failed, set heartbeat Error");
                    heartbeat.setErrorResult("GetAndSyncDbInstanceKeyVariables failed");
                    return true;
                }
            }
        } else if (heartbeat.getStatus() != MySQLHeartbeatStatus.TIMEOUT) { //error/init ->ok
            if (source.isNeedSkipHeartTest() && heartbeat.getStatus() == MySQLHeartbeatStatus.INIT) {
                source.setNeedSkipHeartTest(false);
                return false;
            }
            try {
                source.testConnection();
            } catch (Exception e) {
                LOGGER.warn("testConnection failed, set heartbeat Error");
                heartbeat.setErrorResult("testConnection failed");
                return true;
            }
            GetAndSyncDbInstanceKeyVariables task = new GetAndSyncDbInstanceKeyVariables(source, true);
            KeyVariables variables = task.call();
            if (variables == null ||
                    variables.isLowerCase() != DbleServer.getInstance().getSystemVariables().isLowerCaseTableNames() ||
                    variables.getMaxPacketSize() < SystemConfig.getInstance().getMaxPacketSize() ||
                    VersionUtil.getMajorVersion(variables.getVersion()) < VersionUtil.getMajorVersion(SystemConfig.getInstance().getFakeMySQLVersion())) {
                String url = heartbeat.getSource().getConfig().getUrl();
                Map<String, String> labels = AlertUtil.genSingleLabel("dbInstance", url);
                String errMsg;
                if (variables == null) {
                    errMsg = "GetAndSyncDbInstanceKeyVariables failed";
                } else if (variables.isLowerCase() != DbleServer.getInstance().getSystemVariables().isLowerCaseTableNames()) {
                    errMsg = "this dbInstance[=" + url + "]'s lower_case is wrong";
                } else if (VersionUtil.getMajorVersion(variables.getVersion()) < VersionUtil.getMajorVersion(SystemConfig.getInstance().getFakeMySQLVersion())) {
                    errMsg = "this dbInstance[=" + url + "]'s version[=" + variables.getVersion() + "] cannot be lower than the dble version[=" + SystemConfig.getInstance().getFakeMySQLVersion() + "]";
                } else {
                    errMsg = "this dbInstance[=" + url + "]'s max_allowed_packet is " + variables.getMaxPacketSize() + ", but dble's is " + SystemConfig.getInstance().getMaxPacketSize();
                }
                LOGGER.warn(errMsg + ", set heartbeat Error");
                if (variables != null) {
                    AlertUtil.alert(AlarmCode.DB_INSTANCE_LOWER_CASE_ERROR, Alert.AlertLevel.WARN, errMsg, "mysql", this.heartbeat.getSource().getConfig().getId(), labels);
                    ToResolveContainer.DB_INSTANCE_LOWER_CASE_ERROR.add(url);
                }
                heartbeat.setErrorResult(errMsg);
                return true;
            } else {
                String url = heartbeat.getSource().getConfig().getUrl();
                if (ToResolveContainer.DB_INSTANCE_LOWER_CASE_ERROR.contains(url)) {
                    Map<String, String> labels = AlertUtil.genSingleLabel("dbInstance", url);
                    AlertUtil.alertResolve(AlarmCode.DB_INSTANCE_LOWER_CASE_ERROR, Alert.AlertLevel.WARN, "mysql", this.heartbeat.getSource().getConfig().getId(), labels,
                            ToResolveContainer.DB_INSTANCE_LOWER_CASE_ERROR, url);
                }
                if (heartbeat.getStatus() == MySQLHeartbeatStatus.INIT || !source.isSalveOrRead()) { // writehost checkRecoverFail read only
                    source.setReadOnly(variables.isReadOnly());
                }
            }
        }
        return false;
    }

    private void setStatusBySlave(PhysicalDbInstance source, Map<String, String> resultResult) {
        String slaveIoRunning = resultResult != null ? resultResult.get("Slave_IO_Running") : null;
        String slaveSqlRunning = resultResult != null ? resultResult.get("Slave_SQL_Running") : null;
        if (slaveIoRunning != null && slaveIoRunning.equals(slaveSqlRunning) && slaveSqlRunning.equals("Yes")) {
            heartbeat.setDbSynStatus(MySQLHeartbeat.DB_SYN_NORMAL);
            String secondsBehindMaster = resultResult.get("Seconds_Behind_Master");
            if (null != secondsBehindMaster && !"".equals(secondsBehindMaster) && !"NULL".equalsIgnoreCase(secondsBehindMaster)) {
                int behindMaster = Integer.parseInt(secondsBehindMaster);
                int delayThreshold = source.getDbGroupConfig().getDelayThreshold();
                if (delayThreshold > 0) {
                    String alertKey = source.getDbGroupConfig().getName() + "-" + source.getConfig().getInstanceName();
                    if (behindMaster > delayThreshold) {
                        Map<String, String> labels = AlertUtil.genSingleLabel("dbInstance", alertKey);
                        String errMsg = "found MySQL master/slave Replication delay !!! " + source.getConfig() + ", binlog sync time delay: " + behindMaster + "s";
                        MySQLHeartbeat.LOGGER.warn(errMsg);
                        AlertUtil.alert(AlarmCode.DB_SLAVE_INSTANCE_DELAY, Alert.AlertLevel.WARN, errMsg, "mysql", source.getConfig().getId(), labels);
                        ToResolveContainer.DB_SLAVE_INSTANCE_DELAY.add(alertKey);
                    } else {
                        if (ToResolveContainer.DB_SLAVE_INSTANCE_DELAY.contains(alertKey)) {
                            Map<String, String> labels = AlertUtil.genSingleLabel("dbInstance", alertKey);
                            AlertUtil.alertResolve(AlarmCode.DB_SLAVE_INSTANCE_DELAY, Alert.AlertLevel.WARN, "mysql", source.getConfig().getId(), labels, ToResolveContainer.DB_SLAVE_INSTANCE_DELAY, alertKey);
                        }
                    }
                }
                heartbeat.setSlaveBehindMaster(behindMaster);
            } else {
                heartbeat.setSlaveBehindMaster(null);
            }
        } else if (source.isSalveOrRead()) {
            //String Last_IO_Error = resultResult != null ? resultResult.get("Last_IO_Error") : null;
            MySQLHeartbeat.LOGGER.warn("found MySQL master/slave Replication err !!! " +
                    heartbeat.getSource().getConfig() + ", " + resultResult);
            heartbeat.setDbSynStatus(MySQLHeartbeat.DB_SYN_ERROR);
            heartbeat.setSlaveBehindMaster(null);
        }
        heartbeat.getAsyncRecorder().setBySlaveStatus(resultResult);
        if (checkRecoverFail(source)) return;
        heartbeat.setResult(MySQLHeartbeatStatus.OK);
    }

    private void setStatusByReadOnly(PhysicalDbInstance source, Map<String, String> resultResult) {
        String readonly = resultResult != null ? resultResult.get("@@read_only") : null;
        if (readonly == null) {
            heartbeat.setErrorResult("result of select @@read_only is null");
            return;
        } else if (readonly.equals("0")) {
            source.setReadOnly(false);
        } else {
            source.setReadOnly(true);
        }
        if (checkRecoverFail(source)) return;
        heartbeat.setResult(MySQLHeartbeatStatus.OK);
    }

}
