/*
 * _____________________________________________________________________________
 * 
 * INAF - OATS National Institute for Astrophysics - Astronomical Observatory of
 * Trieste INAF - IA2 Italian Center for Astronomical Archives
 * _____________________________________________________________________________
 * 
 * Copyright (C) 2017 Istituto Nazionale di Astrofisica
 * 
 * This program is free software; you can redistribute it and/or modify it under
 * the terms of the GNU General Public License Version 3 as published by the
 * Free Software Foundation.
 * 
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
 * details.
 * 
 * You should have received a copy of the GNU General Public License along with
 * this program; if not, write to the Free Software Foundation, Inc., 51
 * Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 */
package it.inaf.ia2.tsm.datalayer.mysql;

import it.inaf.ia2.tsm.Column;
import it.inaf.ia2.tsm.Key;
import it.inaf.ia2.tsm.TapSchema;
import it.inaf.ia2.tsm.datalayer.ADQL;
import it.inaf.ia2.tsm.datalayer.DBBrokerTemplate;
import it.inaf.ia2.tsm.model.ColumnModel;
import it.inaf.ia2.tsm.model.TableModel;
import it.inaf.ia2.tsm.model.TypesMapping;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javax.sql.DataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author Sonia Zorba {@literal <zorba at oats.inaf.it>}
 */
public class MySQLDBBroker extends DBBrokerTemplate {

    private static final Logger LOG = LoggerFactory.getLogger(MySQLDBBroker.class);

    public MySQLDBBroker(DataSource dataSource, String tapSchemaVersion) {
        super(dataSource, '`', tapSchemaVersion);
    }

    @Override
    protected String getAllSchemaNamesQuery() {
        return "SHOW DATABASES";
    }

    @Override
    protected String getAllTablesNamesQuery(String schemaName) {
        return "SHOW TABLES FROM " + escape(schemaName);
    }

    @Override
    protected String getColumnNamesQuery(String schemaName, String tableName) {
        return String.format("SHOW COLUMNS FROM %s.%s", escape(schemaName), escape(tableName));
    }

    private Integer getSize(String typeWithSize) {
        int beginIndex = typeWithSize.indexOf('(');
        if (beginIndex != -1) {
            int endIndex = typeWithSize.indexOf(')');
            if (endIndex != -1) {
                try {
                    return Integer.parseInt(typeWithSize.substring(beginIndex + 1, endIndex));
                } catch (NumberFormatException e) {
                    return null;
                }
            }
        }
        return null;
    }

    @Override
    public Map<String, Map<String, Object>> getAllColumnsOriginalMetadata(String schemaName, String tableName) throws SQLException {

        Map<String, Map<String, Object>> allColumnsMetadata = new HashMap<>();

        String query = String.format("SHOW COLUMNS FROM %s.%s", escape(schemaName), escape(tableName));

        LOG.debug("Executing query {}", query);

        try (Connection connection = dataSource.getConnection();
                Statement statement = connection.createStatement();
                ResultSet resultSet = statement.executeQuery(query)) {

            while (resultSet.next()) {

                Map<String, Object> cm = new HashMap<>();

                // Column name
                String columnName = resultSet.getString("Field");
                cm.put(Column.COLUMN_NAME_KEY, columnName);

                // Key info
                String keyType = resultSet.getString("Key");

                boolean primaryKey = "PRI".equals(keyType);
                cm.put(Column.PRIMARY_KEY, primaryKey);

                boolean indexed = "PRI".equals(keyType) || "UNI".equals(keyType) || "MUL".equals(keyType);
                cm.put(Column.INDEXED_KEY, indexed);

                // Datatype and Size
                String type = resultSet.getString("Type").toUpperCase();
                String datatype = TypesMapping.getADQLTypeFromMySQLType(type, getTapSchemaVersion());
                Integer size = getSize(type);
                if (size != null && size > 0 && !datatype.contains("(")) {
                    // Adding size at the end of datatype
                    datatype += String.format("(%s)", size);
                }

                cm.put(Column.DATATYPE_KEY, datatype);
                cm.put(Column.SIZE_KEY, size);

                String arraySize = null;
                if (size != null) {
                    arraySize = String.valueOf(size);
                    // variable length columns must have a "*" symbol on arraysize
                    if (datatype.startsWith(ADQL.VARCHAR.name())) {
                        arraySize += "*";
                    }
                }
                cm.put(Column.ARRAYSIZE_KEY, arraySize);

                allColumnsMetadata.put(columnName, cm);
            }
        }

        return allColumnsMetadata;
    }

    @Override
    protected void createTable(String schemaName, TableModel tableModel, Connection conn) throws SQLException {

        StringBuilder querySb = new StringBuilder();

        querySb.append("CREATE TABLE IF NOT EXISTS ");
        querySb.append(escape(schemaName));
        querySb.append(".");
        querySb.append(escape(tableModel.getName()));
        querySb.append(" (\n");

        boolean first = true;
        for (ColumnModel cm : tableModel.getColumns()) {
            if (cm.isMandatory()) {
                if (!first) {
                    querySb.append(",\n");
                }
                first = false;

                querySb.append(cm.getName());
                querySb.append(" ");

                String mySQLType = TypesMapping.getMySQLTypeFromADQLType(cm.getType()).toUpperCase();
                querySb.append(mySQLType);
                if (mySQLType.equals("VARCHAR") || mySQLType.equals("CHAR")) {
                    appendSize(querySb, cm.getSize());
                }

                if (cm.isNullable()) {
                    querySb.append(" NULL");
                } else {
                    querySb.append(" NOT NULL");
                }
            }
        }

        querySb.append(")");

        String query = querySb.toString();

        try (Statement stat = conn.createStatement()) {
            LOG.debug("Executing query: {}", query);
            stat.executeUpdate(query);
        }
    }

    @Override
    protected String getAddPrimaryKeyQuery(String tapSchemaName, String tableName, String[] keyColumns) {
        return String.format("ALTER TABLE %s.%s ADD PRIMARY KEY(%s)", escape(tapSchemaName), escape(tableName), buildColumnsList(keyColumns));
    }

    @Override
    protected String getAddForeignKeyQuery(String tapSchemaName, String tableName, String[] fromKeyColumns, String targetTableName, String[] toKeyColumns) {
        return String.format("ALTER TABLE %s.%s ADD FOREIGN KEY (%s) REFERENCES %s.%s(%s)", escape(tapSchemaName), escape(tableName),
                buildColumnsList(fromKeyColumns), escape(tapSchemaName), escape(targetTableName), buildColumnsList(toKeyColumns));
    }

    @Override
    public List<Key> getKeys(TapSchema tapSchema, String schemaName) throws SQLException {
        StringBuilder sb = new StringBuilder();

        sb.append("SELECT\n");
        sb.append("c.CONSTRAINT_NAME AS constraint_name,\n");
        sb.append("k.TABLE_SCHEMA AS from_schema,\n");
        sb.append("k.TABLE_NAME AS from_table,\n");
        sb.append("k.COLUMN_NAME AS from_column,\n");
        sb.append("k.REFERENCED_TABLE_SCHEMA AS target_schema,\n");
        sb.append("k.REFERENCED_TABLE_NAME AS target_table,\n");
        sb.append("k.REFERENCED_COLUMN_NAME AS target_column\n");
        sb.append("FROM information_schema.TABLE_CONSTRAINTS c \n");
        sb.append("LEFT JOIN information_schema.KEY_COLUMN_USAGE k \n");
        sb.append("ON c.CONSTRAINT_NAME = k.CONSTRAINT_NAME AND c.TABLE_SCHEMA = k.TABLE_SCHEMA\n");
        sb.append("WHERE c.CONSTRAINT_TYPE = 'FOREIGN KEY' \n");
        sb.append("AND k.TABLE_SCHEMA = '");
        sb.append(schemaName);
        sb.append("' OR k.REFERENCED_TABLE_SCHEMA = '");
        sb.append(schemaName);
        sb.append("'");

        String query = sb.toString();

        LOG.debug("Executing query: {}", query);

        List<Key> keys = new ArrayList<>();
        try (Connection connection = dataSource.getConnection();
                Statement statement = connection.createStatement();
                ResultSet resultSet = statement.executeQuery(query)) {

            while (resultSet.next()) {
                //String constraintName = resultSet.getString("constraint_name");

                String fromSchemaName = resultSet.getString("from_schema");
                String fromTableName = resultSet.getString("from_table");
                String targetSchemaName = resultSet.getString("target_schema");
                String targetTableName = resultSet.getString("target_table");

                Map<String, Object> keyMetadata = new HashMap<>();
                keyMetadata.put(Key.FROM_TABLE_KEY, fromSchemaName + "." + fromTableName);
                keyMetadata.put(Key.TARGET_TABLE_KEY, targetSchemaName + "." + targetTableName);

                Key key = null;
                // Searching for a partial built key
                for (Key k : keys) {
                    if (k.getFromSchemaName().equals(fromSchemaName)
                            && k.getFromTableSimpleName().equals(fromTableName)
                            && k.getTargetSchemaName().equals(targetSchemaName)
                            && k.getTargetTableSimpleName().equals(targetTableName)) {
                        key = k;
                        break;
                    }
                }
                if (key == null) {
                    key = new Key(tapSchema, keyMetadata);
                    keys.add(key);
                }
                key.addKeyColumn(resultSet.getString("from_column"), resultSet.getString("target_column"));
            }
        }

        return keys;
    }

    @Override
    protected String getSchemaTablesQuery(String schemaName) {
        return "SHOW TABLES FROM " + escape(schemaName);
    }

    @Override
    protected String getCreateDatabaseQuery(String databaseName) {
        return "CREATE DATABASE IF NOT EXISTS " + escape(databaseName);
    }

    @Override
    protected String getTableTypesQuery(String schemaName) {
        return "SELECT table_name, table_type FROM information_schema.tables WHERE table_schema = '" + schemaName + "'";
    }
}