package com.actiontech.dble.services.manager.handler;

import com.actiontech.dble.DbleServer;
import com.actiontech.dble.config.ErrorCode;
import com.actiontech.dble.config.model.sharding.table.BaseTableConfig;
import com.actiontech.dble.net.mysql.OkPacket;
import com.actiontech.dble.services.manager.ManagerService;
import com.actiontech.dble.services.manager.dump.ErrorMsg;
import com.actiontech.dble.services.manager.response.DumpFileError;
import com.actiontech.dble.services.manager.split.loaddata.DumpFileReader;
import com.actiontech.dble.services.manager.split.loaddata.ShardingNodeWriter;
import com.actiontech.dble.util.ExecutorUtil;
import com.actiontech.dble.util.NameableExecutor;
import com.actiontech.dble.util.StringUtil;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public final class SplitLoadDataHandler {
    private static final Pattern SPLIT_STMT = Pattern.compile("([^\\s]+)\\s+([^\\s]+)(((\\s*(-s([^\\s]+))?)|(\\s+(-t([^\\s]+))?)|(\\s+(-ds(\\d+))?))+)", Pattern.CASE_INSENSITIVE);
    public static final Logger LOGGER = LoggerFactory.getLogger("dumpFileLog");

    private SplitLoadDataHandler() {
    }


    public static void handle(String stmt, ManagerService service, int offset) {
        Config config = parseOption(stmt.substring(offset).trim());
        if (config == null) {
            LOGGER.info("split load data syntax is error.");
            service.writeErrMessage(ErrorCode.ER_PARSE_ERROR, "You have an error in your SQL syntax");
            return;
        }
        if (StringUtil.isBlank(config.getSourcePath()) || StringUtil.isBlank(config.getTargetPath()) || StringUtil.isBlank(config.getSchemaName()) || StringUtil.isBlank(config.getTableName())) {
            LOGGER.info("split load data syntax is error.");
            service.writeErrMessage(ErrorCode.ER_PARSE_ERROR, "You have an error in your SQL syntax");
            return;
        }
        BaseTableConfig tableConfig = DbleServer.getInstance().getConfig().getSchemas().get(config.getSchemaName()).getTables().get(config.getTableName());
        if (null == tableConfig) {
            LOGGER.info("schema `{}` table `{}` configuration does not exist.", config.getSchemaName(), config.getTableName());
            service.writeErrMessage(ErrorCode.ER_PARSE_ERROR, "schema `{" + config.getSchemaName() + "}` table `{" + config.getTableName() + "}` configuration does not exist.");
            return;
        }
        config.setTableConfig(tableConfig);

        List<ErrorMsg> errorMsgList = Lists.newCopyOnWriteArrayList();
        AtomicBoolean errorFlag = new AtomicBoolean();
        AtomicInteger nodeCount = new AtomicInteger(-1);

        Map<String, ShardingNodeWriter> writerMap = new ConcurrentHashMap<>();
        NameableExecutor fileReadExecutor = ExecutorUtil.createFixed("splitLoadReader", 1);
        DumpFileReader reader = new DumpFileReader(errorMsgList, errorFlag, config, writerMap, nodeCount);

        // start read
        reader.open(config.getSourcePath());
        fileReadExecutor.execute(reader);


        //wait
        while (!errorFlag.get() && nodeCount.get() != 0) {
            LockSupport.parkNanos(1000);
        }

        //recycle thread
        fileReadExecutor.shutdownNow();
        for (ShardingNodeWriter shardingNodeWriter : writerMap.values()) {
            shardingNodeWriter.shutdown();
        }
        writerMap.clear();

        //handle error
        if (errorFlag.get()) {
            //todo whether the generated files are deleted
            DumpFileError.execute(service, errorMsgList);
            return;
        }

        OkPacket packet = new OkPacket();
        packet.setServerStatus(2);
        packet.setPacketId(1);
        packet.write(service.getConnection());
    }


    private static Config parseOption(String stmt) {
        Matcher m = SPLIT_STMT.matcher(stmt);
        if (m.matches()) {
            Config config = new Config();
            config.setSourcePath(m.group(1));
            config.setTargetPath(m.group(2));
            config.setSchemaName(m.group(7));
            config.setTableName(m.group(10));
            if (null != m.group(13)) {
                config.setDisruptorBufferSize(Integer.parseInt(m.group(13)));
            }
            return config;
        }
        return null;
    }

    public static class Config {
        private String sourcePath;
        private String targetPath;
        private String schemaName;
        private String tableName;
        private BaseTableConfig tableConfig;
        private int disruptorBufferSize = 512;

        public Config() {
        }

        public void setSourcePath(String sourcePath) {
            this.sourcePath = sourcePath;
        }

        public void setTargetPath(String targetPath) {
            this.targetPath = targetPath;
        }

        public void setSchemaName(String schemaName) {
            this.schemaName = schemaName;
        }

        public void setTableName(String tableName) {
            this.tableName = tableName;
        }

        public void setTableConfig(BaseTableConfig tableConfig) {
            this.tableConfig = tableConfig;
        }

        public int getDisruptorBufferSize() {
            return disruptorBufferSize;
        }

        public void setDisruptorBufferSize(int disruptorBufferSize) {
            this.disruptorBufferSize = disruptorBufferSize;
        }

        public String getSourcePath() {
            return sourcePath;
        }

        public String getTargetPath() {
            return targetPath;
        }

        public String getSchemaName() {
            return schemaName;
        }

        public String getTableName() {
            return tableName;
        }

        public BaseTableConfig getTableConfig() {
            return tableConfig;
        }
    }

}
