/*
 * _____________________________________________________________________________
 * 
 * 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.pgsql;

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.PropertyModel;
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 java.util.regex.Pattern;
import javax.sql.DataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

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

    private final String pgDatabaseName;

    public PostgresDBBroker(DataSource dataSource, String pgDatabaseName) {
        super(dataSource, '"');
        this.pgDatabaseName = pgDatabaseName;
    }

    @Override
    protected void createTable(String tapSchemaName, TableModel tableModel, Connection conn) throws SQLException {
        StringBuilder querySb = new StringBuilder();

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

        boolean first = true;
        for (PropertyModel pm : tableModel.getProperties().values()) {
            if (!first) {
                querySb.append(",\n");
            }
            first = false;

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

            String pgsqlType = TypesMapping.getPostgresSQLTypeFromADQLType(pm.getType());
            querySb.append(pgsqlType);
            if (pm.getType().equals(ADQL.VARCHAR.name())
                    || pm.getType().equals(ADQL.CHAR.name())) {
                appendSize(querySb, pm.getSize());
            }

            if (pm.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 ONLY %s.%s ADD CONSTRAINT %s_pkey PRIMARY KEY (%s)",
                escape(tapSchemaName), escape(tableName), tableName, buildColumnsList(keyColumns));
    }

    @Override
    protected String getAddForeignKeyQuery(String tapSchemaName, String tableName, String[] fromKeyColumns, String targetTableName, String[] toKeyColumns) {
        // Building univocal constraint name
        StringBuilder constraintNameSb = new StringBuilder(tableName);
        for (String fromKeyColumn : fromKeyColumns) {
            constraintNameSb.append("_");
            constraintNameSb.append(fromKeyColumn);
        }
        return String.format("ALTER TABLE ONLY %s.%s ADD CONSTRAINT fk_%s FOREIGN KEY (%s) REFERENCES %s.%s(%s)",
                escape(tapSchemaName), escape(tableName), constraintNameSb.toString(), buildColumnsList(fromKeyColumns),
                escape(tapSchemaName), escape(targetTableName), buildColumnsList(toKeyColumns));
    }

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

    @Override
    protected String getSchemaTablesQuery(String schemaName) {
        return String.format("SELECT tablename FROM pg_catalog.pg_tables where schemaname = '%s'", schemaName);
    }

    @Override
    public Map<String, Map<String, Object>> getAllColumnsMetadata(String schemaName, String tableSimpleName) throws SQLException {

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

        StringBuilder querySb = new StringBuilder();

        querySb.append("SELECT c.column_name, c.data_type, r.contype AS column_type, c.character_maximum_length, c.numeric_precision\n"); //, c.numeric_precision_radix
        querySb.append("FROM information_schema.columns c\n");
        querySb.append("JOIN pg_catalog.pg_tables t ON c.table_schema = t.schemaname AND c.table_name = t.tablename\n");
        querySb.append("LEFT JOIN pg_catalog.pg_constraint r ON c.ordinal_position = ANY(r.conkey) AND r.conrelid = (t.schemaname || '.' || t.tablename)::regclass::oid\n");
        querySb.append("WHERE t.schemaname = '");
        querySb.append(schemaName);
        querySb.append("' AND t.tablename = '");
        querySb.append(tableSimpleName);
        querySb.append("'");

        String query = querySb.toString();
        LOG.debug("Executing query {}", query);

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

            while (resultSet.next()) {

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

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

                // Key info
                String columnType = resultSet.getString("column_type");
                boolean primaryKey = false;
                boolean indexed = false;
                if (columnType != null) {
                    primaryKey = "p".equals(columnType);
                    if ("p".equals(columnType)
                            || "f".equals(columnType)
                            || "u".equals(columnType)) {
                        indexed = true;
                    }
                }
                cm.put(Column.PRIMARY_KEY, primaryKey);
                cm.put(Column.INDEXED_KEY, indexed);

                int size = 0;
                String datatype;

                String type = resultSet.getString("data_type");
                datatype = TypesMapping.getADQLTypeFromPostgresType(type);
                if (datatype.equals(ADQL.VARCHAR.name()) || datatype.equals(ADQL.CHAR.name())) {
                    size = resultSet.getInt("character_maximum_length");
                }

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

                Integer arraySize = null; // TODO (v 1.1)

                allColumnsMetadata.put(columnName, cm);
            }
        }

        return allColumnsMetadata;
    }

    @Override
    public List<Key> getKeys(TapSchema tapSchema, String schemaName) throws SQLException {

        StringBuilder queryKeysSb = new StringBuilder();

        queryKeysSb.append("SELECT\n");
        queryKeysSb.append("conname AS constraint_name,\n");
        queryKeysSb.append("conrelid::regclass AS from_table, \n");
        queryKeysSb.append("confrelid::regclass AS target_table\n");
        queryKeysSb.append("FROM pg_catalog.pg_constraint\n");
        queryKeysSb.append("WHERE contype = 'f'\n");
        queryKeysSb.append("AND ((conrelid::regclass || '' LIKE '");
        queryKeysSb.append(schemaName);
        queryKeysSb.append(".%')\n");
        queryKeysSb.append("OR (confrelid::regclass || '' LIKE '");
        queryKeysSb.append(schemaName);
        queryKeysSb.append(".%'))");

        String queryKeys = queryKeysSb.toString();

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

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

            List<Key> keys = new ArrayList<>();
            while (resultSet.next()) {

                String constraintName = resultSet.getString("constraint_name");

                String fromTableCompleteName = resultSet.getString("from_table");
                String targetTableCompleteName = resultSet.getString("target_table");

                String fromSchema = fromTableCompleteName.split(Pattern.quote("."))[0];
                String targetSchema = targetTableCompleteName.split(Pattern.quote("."))[0];

                Map<String, Object> keyMetadata = new HashMap<>();
                keyMetadata.put(Key.FROM_TABLE_KEY, fromTableCompleteName);
                keyMetadata.put(Key.TARGET_TABLE_KEY, targetTableCompleteName);

                Key key = new Key(tapSchema, keyMetadata);
                keys.add(key);

                StringBuilder queryFromKCSb = new StringBuilder();
                queryFromKCSb.append("SELECT\n");
                queryFromKCSb.append("c.column_name AS key_column\n");
                queryFromKCSb.append("FROM information_schema.columns c\n");
                queryFromKCSb.append("JOIN pg_catalog.pg_constraint r ON c.ordinal_position = ANY(r.conkey)\n");
                queryFromKCSb.append("AND (c.table_schema || '.' || c.table_name) = (r.conrelid::regclass || '')\n");
                queryFromKCSb.append("WHERE r.conname = '");
                queryFromKCSb.append(constraintName);
                queryFromKCSb.append("' AND r.contype = 'f'\n");
                queryFromKCSb.append("AND c.table_schema = '");
                queryFromKCSb.append(fromSchema);
                queryFromKCSb.append("'\nAND table_catalog = '");
                queryFromKCSb.append(pgDatabaseName);
                queryFromKCSb.append("'");

                // conkey conrelid
                String queryFromKC = queryFromKCSb.toString();

                StringBuilder queryTargetKCSb = new StringBuilder();
                queryTargetKCSb.append("SELECT\n");
                queryTargetKCSb.append("c.column_name AS key_column\n");
                queryTargetKCSb.append("FROM information_schema.columns c\n");
                queryTargetKCSb.append("JOIN pg_catalog.pg_constraint r ON c.ordinal_position = ANY(r.confkey)\n");
                queryTargetKCSb.append("AND (c.table_schema || '.' || c.table_name) = (r.confrelid::regclass || '')\n");
                queryTargetKCSb.append("WHERE r.conname = '");
                queryTargetKCSb.append(constraintName);
                queryTargetKCSb.append("' AND r.contype = 'f'\n");
                queryTargetKCSb.append("AND c.table_schema = '");
                queryTargetKCSb.append(targetSchema);
                queryTargetKCSb.append("'\nAND table_catalog = '");
                queryTargetKCSb.append(pgDatabaseName);
                queryTargetKCSb.append("'");

                // as above, but with confkey and confrelid and different c.table_schema where condition
                String queryTargetKC = queryTargetKCSb.toString();

                try (Statement statFromKC = connection.createStatement();
                        Statement statTargetKC = connection.createStatement()) {

                    try (ResultSet rsFromKC = statFromKC.executeQuery(queryFromKC);
                            ResultSet rsTargetKC = statTargetKC.executeQuery(queryTargetKC)) {

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

                        while (rsFromKC.next()) {
                            if (rsTargetKC.next()) {
                                key.addKeyColumn(
                                        rsFromKC.getString("key_column"),
                                        rsTargetKC.getString("key_column")
                                );
                            }
                        }
                    }
                }
            }

            return keys;
        }
    }

    @Override
    protected String getTableTypesQuery(String schemaName) {
        StringBuilder sb = new StringBuilder();
        sb.append("SELECT tablename AS table_name, 'table' AS table_type\n");
        sb.append("FROM pg_catalog.pg_tables WHERE schemaname = '");
        sb.append(schemaName);
        sb.append("'\n");
        sb.append("UNION\n");
        sb.append("SELECT table_name AS table_name, 'view' AS table_type\n");
        sb.append("FROM INFORMATION_SCHEMA.views\n");
        sb.append("WHERE table_schema = '");
        sb.append(schemaName);
        sb.append("'");
        return sb.toString();
    }

    @Override
    protected String getColumnNamesQuery(String tapSchemaName, String tableName) {
        return String.format("SELECT column_name FROM information_schema.columns WHERE table_schema = '%s' AND table_name = '%s'",
                tapSchemaName, tableName);
    }

    @Override
    protected String getAllSchemaNamesQuery() {
        return "SELECT schema_name FROM information_schema.schemata";
    }

    @Override
    protected String getAllTablesNamesQuery(String schemaName) {
        return String.format("SELECT tablename FROM pg_catalog.pg_tables where schemaname = '%s'", schemaName);
    }
}