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

package com.actiontech.dble.meta;

import com.actiontech.dble.DbleServer;
import com.actiontech.dble.backend.datasource.PhysicalDbGroup;
import com.actiontech.dble.backend.datasource.ShardingNode;
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.ClusterHelper;
import com.actiontech.dble.cluster.DistributeLock;
import com.actiontech.dble.cluster.DistributeLockManager;
import com.actiontech.dble.cluster.general.kVtoXml.ClusterToXml;
import com.actiontech.dble.cluster.logic.ClusterOperation;
import com.actiontech.dble.cluster.path.ClusterMetaUtil;
import com.actiontech.dble.cluster.path.ClusterPathUtil;
import com.actiontech.dble.cluster.values.ClusterTime;
import com.actiontech.dble.cluster.zkprocess.zktoxml.listen.DDLChildListener;
import com.actiontech.dble.cluster.zkprocess.zktoxml.listen.DbGroupResponseListener;
import com.actiontech.dble.cluster.zkprocess.zktoxml.listen.DbGroupStatusListener;
import com.actiontech.dble.cluster.zkprocess.zktoxml.listen.ViewChildListener;
import com.actiontech.dble.config.ErrorCode;
import com.actiontech.dble.config.ServerConfig;
import com.actiontech.dble.config.model.ClusterConfig;
import com.actiontech.dble.config.model.SystemConfig;
import com.actiontech.dble.config.model.sharding.SchemaConfig;
import com.actiontech.dble.config.model.sharding.table.BaseTableConfig;
import com.actiontech.dble.meta.table.AbstractSchemaMetaHandler;
import com.actiontech.dble.meta.table.SchemaCheckMetaHandler;
import com.actiontech.dble.meta.table.ServerMetaHandler;
import com.actiontech.dble.plan.node.PlanNode;
import com.actiontech.dble.server.util.SchemaUtil;
import com.actiontech.dble.server.util.SchemaUtil.SchemaInfo;
import com.actiontech.dble.singleton.TraceManager;
import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.SQLNonTransientException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReentrantLock;

public class ProxyMetaManager {
    protected static final Logger LOGGER = LoggerFactory.getLogger(ProxyMetaManager.class);
    // catalog,table,table meta
    private final Map<String, SchemaMeta> catalogs;
    private final Map<String, String> lockTables;
    private ReentrantLock metaLock = new ReentrantLock();
    private ScheduledExecutorService scheduler;
    private ScheduledFuture<?> checkTaskHandler;
    private AtomicInteger metaCount = new AtomicInteger(0);
    private volatile Repository repository = null;
    private AtomicInteger version = new AtomicInteger(0);
    private long timestamp;
    private AtomicInteger tableIndex = new AtomicInteger();
    private static AtomicInteger fakeTableIndex = new AtomicInteger();

    public ProxyMetaManager() {
        this.catalogs = new ConcurrentHashMap<>();
        this.lockTables = new HashMap<>();
        this.timestamp = System.currentTimeMillis();
    }

    public ProxyMetaManager(ProxyMetaManager origin) {
        this.catalogs = new ConcurrentHashMap<>();
        this.lockTables = origin.lockTables;
        this.timestamp = System.currentTimeMillis();
        this.metaLock = origin.metaLock;
        this.metaCount = origin.metaCount;
        this.repository = origin.repository;
        this.version = origin.version;
        this.tableIndex = origin.tableIndex;
        for (Map.Entry<String, SchemaMeta> entry : origin.catalogs.entrySet()) {
            catalogs.put(entry.getKey(), entry.getValue().metaCopy());
        }
    }

    //no need to check user
    private static SchemaInfo getSchemaInfo(String schema, String table) {
        return SchemaUtil.getSchemaInfoWithoutCheck(schema, table);
    }

    public long getTimestamp() {
        return timestamp;
    }

    public void setTimestamp(long timestamp) {
        this.timestamp = timestamp;
    }

    public Map<String, String> getLockTables() {
        return lockTables;
    }

    private String genLockKey(String schema, String tbName) {
        return schema + "." + tbName;
    }

    public String metaCountCheck() {
        StringBuilder result = new StringBuilder();
        metaLock.lock();
        try {
            if (metaCount.get() != 0) {
                result.append("There is other session is doing DDL\n");
                for (String x : lockTables.values()) {
                    result.append(x).append("\n");
                }
                result.setLength(result.length() - 1);
            }
        } finally {
            metaLock.unlock();
        }
        return result.length() == 0 ? null : result.toString();
    }

    public ReentrantLock getMetaLock() {
        return metaLock;
    }

    public void addMetaLock(String schema, String tbName, String sql) throws SQLNonTransientException {
        metaLock.lock();
        try {
            String lockKey = genLockKey(schema, tbName);
            if (lockTables.containsKey(lockKey)) {
                String msg = "SCHEMA[" + schema + "], TABLE[" + tbName + "] is doing DDL";
                LOGGER.warn(msg);
                throw new SQLNonTransientException(msg, "HY000", ErrorCode.ER_DOING_DDL);
            } else {
                metaCount.incrementAndGet();
                version.incrementAndGet();
                lockTables.put(lockKey, sql);
            }
        } finally {
            metaLock.unlock();
        }
    }

    public boolean removeMetaLock(String schema, String tbName) {

        boolean isRemoved = false;
        metaLock.lock();
        try {
            if (lockTables.remove(genLockKey(schema, tbName)) != null) {
                isRemoved = true;
                metaCount.decrementAndGet();
            }
        } finally {
            metaLock.unlock();
        }
        return isRemoved;
    }

    public Map<String, SchemaMeta> getCatalogs() {
        return catalogs;
    }

    public boolean createDatabase(String schema) {
        SchemaMeta schemaMeta = catalogs.get(schema);
        if (schemaMeta == null) {
            schemaMeta = new SchemaMeta();
            catalogs.put(schema, schemaMeta);
            return true;
        } else {
            return false;
        }
    }

    /**
     * Checking the existence of the database in which the view is to be created
     */
    private boolean checkDbExists(String schema) {
        return schema != null && this.catalogs.containsKey(schema);
    }

    public boolean checkTableExists(String schema, String strTable) {
        return checkDbExists(schema) && strTable != null && this.catalogs.get(schema).getTableMetas().containsKey(strTable);
    }

    public void addTable(String schema, TableMeta tm, boolean isNewCreate) {
        String tbName = tm.getTableName();
        tm.setSchemaName(schema);
        SchemaMeta schemaMeta = catalogs.get(schema);
        if (schemaMeta != null) {
            tm.setId(tableIndex.incrementAndGet());
            schemaMeta.addTableMeta(tbName, tm);
        }
        if (isNewCreate) {
            SchemaUtil.tryAddDefaultShardingTableConfig(schema, tbName, tm.getCreateSql(), fakeTableIndex);
        }
    }

    public void addView(String schema, ViewMeta vm) {
        String viewName = vm.getViewName();
        SchemaMeta schemaMeta = catalogs.get(schema);
        if (schemaMeta != null) {
            schemaMeta.addViewMeta(viewName, vm);
        }
    }

    public void dropTable(String schema, String tbName) {
        SchemaMeta schemaMeta = catalogs.get(schema);
        if (schemaMeta != null)
            schemaMeta.dropTable(tbName);
        SchemaUtil.tryDropDefaultShardingTableConfig(schema, tbName);
    }

    public TableMeta getSyncTableMeta(String schema, String tbName) throws SQLNonTransientException {
        TraceManager.TraceObject traceObject = TraceManager.threadTrace("get-sync-meta");
        TraceManager.log(ImmutableMap.of("schema", schema, "table", tbName), traceObject);
        try {
            while (true) {
                int oldVersion = version.get();
                if (metaCount.get() == 0) {
                    TableMeta meta = getTableMeta(schema, tbName);
                    if (version.get() == oldVersion) {
                        return meta;
                    }
                } else {
                    metaLock.lock();
                    try {
                        if (lockTables.containsKey(genLockKey(schema, tbName))) {
                            String msg = "SCHEMA[" + schema + "], TABLE[" + tbName + "] is doing DDL";
                            LOGGER.info(msg);
                            throw new SQLNonTransientException(msg, "HY000", ErrorCode.ER_DOING_DDL);
                        } else {
                            return getTableMeta(schema, tbName);
                        }
                    } finally {
                        metaLock.unlock();
                    }
                }
            }
        } finally {
            TraceManager.finishSpan(traceObject);
        }
    }

    public PlanNode getSyncView(String schema, String vName) throws SQLNonTransientException {
        TraceManager.TraceObject traceObject = TraceManager.threadTrace("get-sync-view");
        TraceManager.log(ImmutableMap.of("schema", schema, "table", vName), traceObject);
        try {
            while (true) {
                int oldVersion = version.get();
                if (catalogs.get(schema) == null) {
                    return null;
                }
                if (metaCount.get() == 0) {
                    PlanNode viewNode = catalogs.get(schema).getView(vName);
                    if (version.get() == oldVersion) {
                        return viewNode;
                    }
                } else {
                    metaLock.lock();
                    try {
                        if (lockTables.containsKey(genLockKey(schema, vName))) {
                            String msg = "SCHEMA[" + schema + "], TABLE[" + vName + "] is doing DDL";
                            LOGGER.info(msg);
                            throw new SQLNonTransientException(msg, "HY000", ErrorCode.ER_DOING_DDL);
                        } else {
                            return catalogs.get(schema).getView(vName);
                        }
                    } finally {
                        metaLock.unlock();
                    }
                }
            }
        } finally {
            TraceManager.finishSpan(traceObject);
        }
    }

    private TableMeta getTableMeta(String schema, String tbName) {
        return catalogs.get(schema).getTableMeta(tbName);
    }

    protected static Set<String> getSelfNodes(ServerConfig config) {
        Set<String> selfNode = null;
        for (Map.Entry<String, PhysicalDbGroup> entry : config.getDbGroups().entrySet()) {
            PhysicalDbGroup host = entry.getValue();
            if (host.isAllFakeNode()) {
                for (Map.Entry<String, ShardingNode> nodeEntry : config.getShardingNodes().entrySet()) {
                    if (nodeEntry.getValue().getDbGroup().getGroupName().equals(host.getGroupName())) {
                        if (selfNode == null) {
                            selfNode = new HashSet<>(2);
                        }
                        selfNode.add(nodeEntry.getKey());
                    }
                }
            }
        }
        return selfNode;
    }

    public void init(ServerConfig config) throws Exception {
        LOGGER.info("init metaData start");
        tryAddSyncMetaLock();
        initMeta(config);
        releaseSyncMetaLock();
        LOGGER.info("init metaData end");
    }

    private void tryAddSyncMetaLock() throws Exception {
        if (ClusterConfig.getInstance().isClusterEnable()) {
            int times = 0;
            final ClusterHelper metaHelper = ClusterHelper.getInstance(ClusterOperation.META);
            DistributeLock lock = metaHelper.createDistributeLock(ClusterMetaUtil.getSyncMetaLockPath(), new ClusterTime(String.valueOf(System.currentTimeMillis())));
            while (!lock.acquire()) {
                LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100));
                if (times % 100 == 0) {
                    LOGGER.warn("tryAddSyncMetaLock failed");
                    times = 0;
                }
                times++;
            }
            DistributeLockManager.addLock(lock);


            times = 0;
            String ddlLockPath = ClusterPathUtil.getDDLLockPath();
            final ClusterHelper ddlHelper = ClusterHelper.getInstance(ClusterOperation.DDL);
            while (!Boolean.TRUE.equals(ClusterHelper.isExist(ddlLockPath)) && ddlHelper.getChildrenSize(ddlLockPath) > 0) {
                LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1000));
                if (times % 60 == 0) {
                    LOGGER.warn("waiting for DDL finished ");
                    times = 0;
                }
                times++;
            }

            times = 0;
            String viewLockPath = ClusterPathUtil.getViewLockPath();
            final ClusterHelper viewHelper = ClusterHelper.getInstance(ClusterOperation.VIEW);
            while (!Boolean.TRUE.equals(ClusterHelper.isExist(viewLockPath)) && viewHelper.getChildrenSize(viewLockPath) > 0) {
                LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1000));
                if (times % 60 == 0) {
                    LOGGER.warn("waiting for view finished");
                    times = 0;
                }
                times++;
            }
        }
    }

    private void releaseSyncMetaLock() throws Exception {
        if (ClusterConfig.getInstance().isClusterEnable()) {
            if (ClusterConfig.getInstance().useZkMode()) {
                //add watcher
                new DDLChildListener().registerPrefixForZk();
                //add tow ha status && ha lock watcher
                if (ClusterConfig.getInstance().isNeedSyncHa()) {
                    new DbGroupStatusListener().registerPrefixForZk();
                    new DbGroupResponseListener().registerPrefixForZk();
                }
                //add watcher
                new ViewChildListener().registerPrefixForZk();
            } else {
                ClusterToXml.startMetaListener();
            }
            // syncMeta UNLOCK
            DistributeLockManager.releaseLock(ClusterPathUtil.getSyncMetaLockPath());
        }
    }

    private void initViewMeta() {
        if (ClusterConfig.getInstance().isClusterEnable()) {
            loadViewFromKV();
        } else {
            loadViewFromFile();
        }
    }

    /**
     * recovery all the view info from KV system
     */
    private void loadViewFromKV() {
        repository = new KVStoreRepository();
        Map<String, Map<String, String>> viewCreateSqlMap = repository.getViewCreateSqlMap();
        loadViewMeta(viewCreateSqlMap);
    }

    /**
     * recovery all the view info for file system
     */
    private void loadViewFromFile() {
        repository = new FileSystemRepository();
        Map<String, Map<String, String>> viewCreateSqlMap = repository.getViewCreateSqlMap();
        loadViewMeta(viewCreateSqlMap);
    }


    /**
     * put the create sql in & parse all the tableNode & put into sharding-viewMeta
     *
     * @param viewCreateSqlMap
     */
    public void loadViewMeta(Map<String, Map<String, String>> viewCreateSqlMap) {
        for (Map.Entry<String, Map<String, String>> schemaName : viewCreateSqlMap.entrySet()) {
            for (Map.Entry<String, String> view : schemaName.getValue().entrySet()) {
                try {
                    ViewMeta vm = new ViewMeta(schemaName.getKey(), view.getValue(), this);
                    vm.init();
                    SchemaMeta schemaMeta = this.getCatalogs().get(schemaName.getKey());
                    if (schemaMeta == null) {
                        LOGGER.warn("View " + view.getKey() + " can not find it's schema,view " + view.getKey() + " not initialized");
                    } else {
                        schemaMeta.getViewMetas().put(vm.getViewName(), vm);
                    }
                } catch (Exception e) {
                    LOGGER.warn("load view meta error", e);
                }
            }
        }
    }

    public void reloadViewMeta(Map<String, Map<String, String>> viewCreateSqlMap) {
        for (Map.Entry<String, Map<String, String>> schemaName : viewCreateSqlMap.entrySet()) {
            ConcurrentMap<String, ViewMeta> schemaViewMeta = new ConcurrentHashMap<String, ViewMeta>();
            for (Map.Entry<String, String> view : schemaName.getValue().entrySet()) {
                try {
                    ViewMeta vm = new ViewMeta(schemaName.getKey(), view.getValue(), this);
                    vm.init();
                    schemaViewMeta.put(vm.getViewName(), vm);
                } catch (Exception e) {
                    LOGGER.warn("reload view meta error", e);
                }
            }
            this.getCatalogs().get(schemaName.getKey()).setViewMetas(schemaViewMeta);
        }
    }

    /**
     * init meta when dble server started
     * no interrupted ,init the view anyway
     *
     * @param config
     */
    private void initMeta(ServerConfig config) {
        Set<String> selfNode = getSelfNodes(config);
        ServerMetaHandler handler = new ServerMetaHandler(this, config, selfNode);
        handler.setFilter(null);
        handler.execute();
        initViewMeta();
        if (SystemConfig.getInstance().getCheckTableConsistency() == 1) {
            scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder().setNameFormat("MetaDataChecker-%d").build());
            checkTaskHandler = scheduler.scheduleWithFixedDelay(tableStructureCheckTask(selfNode), SystemConfig.getInstance().getCheckTableConsistencyPeriod(), SystemConfig.getInstance().getCheckTableConsistencyPeriod(), TimeUnit.MILLISECONDS);
        }
    }

    /**
     * init new Meta from reload config/metadata
     * can be interrupted and abandon the new config
     *
     * @param config
     * @param specifiedSchemas
     */
    public boolean initMeta(ServerConfig config, Map<String, Set<String>> specifiedSchemas) {
        Set<String> selfNode = getSelfNodes(config);
        ServerMetaHandler handler = new ServerMetaHandler(this, config, selfNode);
        handler.setFilter(specifiedSchemas);
        handler.register();
        // if the meta reload interrupted by reload release
        // do not reload the view meta or start a new scheduler
        if (handler.execute()) {
            initViewMeta();
            if (SystemConfig.getInstance().getCheckTableConsistency() == 1) {
                scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder().setNameFormat("MetaDataChecker-%d").build());
                checkTaskHandler = scheduler.scheduleWithFixedDelay(tableStructureCheckTask(selfNode), SystemConfig.getInstance().getCheckTableConsistencyPeriod(), SystemConfig.getInstance().getCheckTableConsistencyPeriod(), TimeUnit.MILLISECONDS);
            }
            return true;
        }
        return false;
    }

    public void terminate() {
        if (checkTaskHandler != null) {
            checkTaskHandler.cancel(false);
            scheduler.shutdown();
        }
        if (repository != null) {
            repository.terminate();
        }
    }

    //Check the Consistency of table Structure
    private Runnable tableStructureCheckTask(final Set<String> selfNode) {
        return new Runnable() {
            @Override
            public void run() {
                tableStructureCheck(selfNode);
            }
        };
    }

    private void tableStructureCheck(Set<String> selfNode) {
        for (SchemaConfig schema : DbleServer.getInstance().getConfig().getSchemas().values()) {
            if (!checkDbExists(schema.getName())) {
                continue;
            }
            Map<String, Set<String>> shardingNodeMap = new HashMap<>();
            for (Map.Entry<String, BaseTableConfig> entry : schema.getTables().entrySet()) {
                String tableName = entry.getKey();
                BaseTableConfig tbConfig = entry.getValue();
                for (String shardingNode : tbConfig.getShardingNodes()) {
                    Set<String> tables = shardingNodeMap.computeIfAbsent(shardingNode, k -> new HashSet<>());
                    tables.add(tableName);
                }
            }

            AbstractSchemaMetaHandler multiTablesMetaHandler = new SchemaCheckMetaHandler(this, schema, selfNode);
            multiTablesMetaHandler.execute();
        }
    }

    public Repository getRepository() {
        return repository;
    }

    public void setRepository(Repository repository) {
        this.repository = repository;
    }
}
