package fr.ifremer.adagio.synchro.meta;

/*
 * #%L
 * Tutti :: Persistence
 * $Id: ReferentialSynchroDatabaseMetadata.java 1573 2014-02-04 16:41:40Z tchemit $
 * $HeadURL: http://svn.forge.codelutin.com/svn/tutti/trunk/tutti-persistence/src/main/java/fr/ifremer/adagio/core/service/technical/synchro/ReferentialSynchroDatabaseMetadata.java $
 * %%
 * Copyright (C) 2012 - 2014 Ifremer
 * %%
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * 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 Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 * #L%
 */

import static org.nuiton.i18n.I18n.t;

import java.lang.reflect.Field;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.hibernate.HibernateException;
import org.hibernate.cfg.AvailableSettings;
import org.hibernate.cfg.Configuration;
import org.hibernate.cfg.Environment;
import org.hibernate.dialect.Dialect;
import org.hibernate.exception.spi.SQLExceptionConverter;
import org.hibernate.internal.util.StringHelper;
import org.hibernate.internal.util.config.ConfigurationHelper;
import org.hibernate.mapping.Table;
import org.hibernate.tool.hbm2ddl.DatabaseMetadata;
import org.hibernate.tool.hbm2ddl.TableMetadata;

import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

import fr.ifremer.adagio.synchro.SynchroTechnicalException;
import fr.ifremer.adagio.synchro.dao.DaoUtils;
import fr.ifremer.adagio.synchro.intercept.SynchroInterceptor;
import fr.ifremer.adagio.synchro.intercept.SynchroInterceptorUtils;
import fr.ifremer.adagio.synchro.service.SynchroContext;

/**
 * Created on 1/14/14.
 * 
 * @author Tony Chemit <chemit@codelutin.com>
 * @since 3.5
 */
public class SynchroDatabaseMetadata {

    /** Logger. */
    private static final Log log =
            LogFactory.getLog(SynchroDatabaseMetadata.class);

    private static final String TABLE_CATALOG_PATTERN = "TABLE_CAT";
    private static final String TABLE_TYPE_PATTERN = "TABLE_TYPE";
    private static final String TABLE_SCHEMA_PATTERN = "TABLE_SCHEM";
    private static final String REMARKS_PATTERN = "REMARKS";
    private static final String TABLE_NAME_PATTERN = "TABLE_NAME";

    /**
     * Load the datasource schema for the given connection and dialect.
     * 
     * @param connection
     *            connection of the data source
     * @param dialect
     *            dialect to use
     * @param configuration
     * @param tableNames
     *            table names to includes (table patterns are accepted)
     * @return the database schema metadata
     */
    public static SynchroDatabaseMetadata loadDatabaseMetadata(Connection connection,
            Dialect dialect, Configuration configuration,
            SynchroContext context,
            Set<String> tableNames,
            boolean enableJoinMetadataLoading
            ) {
        SynchroDatabaseMetadata result = new SynchroDatabaseMetadata(connection, dialect, configuration, context);
        result.prepare(dialect, configuration, tableNames, null, null, enableJoinMetadataLoading);
        return result;
    }

    /**
     * Load the datasource schema for the given connection and dialect.
     * 
     * @param connection
     *            connection of the data source
     * @param dialect
     *            dialect to use
     * @param configuration
     * @param tableNames
     *            table names to includes (table patterns are accepted) (optional if tabkleFilter not null)
     * @param tableFilter
     *            filter tables (optional)
     * @param columnFilter
     *            filter columns (optional)
     * @return the database schema metadata
     */
    public static SynchroDatabaseMetadata loadDatabaseMetadata(Connection connection,
            Dialect dialect,
            Configuration configuration,
            SynchroContext context,
            Set<String> tableNames,
            Predicate<String> tableFilter,
            Predicate<SynchroColumnMetadata> columnFilter,
            boolean enableJoinMetadataLoading
            ) {
        SynchroDatabaseMetadata result = new SynchroDatabaseMetadata(connection, dialect, configuration, context);
        result.prepare(dialect, configuration, tableNames, tableFilter, columnFilter, enableJoinMetadataLoading);
        return result;
    }

    protected final DatabaseMetadata delegate;

    protected final Map<String, SynchroTableMetadata> tables;

    protected final DatabaseMetaData meta;

    protected final Configuration configuration;

    protected final Dialect dialect;

    protected final Set<String> sequences;

    protected final String[] types;

    private SQLExceptionConverter sqlExceptionConverter;

    protected List<SynchroInterceptor> interceptors;

    protected SynchroContext context;

    public SynchroDatabaseMetadata(Connection connection, Dialect dialect, Configuration configuration,
            SynchroContext context) {
        Preconditions.checkNotNull(connection);
        Preconditions.checkNotNull(dialect);
        Preconditions.checkNotNull(configuration);

        this.configuration = configuration;
        this.dialect = dialect;
        this.sqlExceptionConverter = DaoUtils.newSQLExceptionConverter(dialect);
        this.context = context;

        try {
            this.delegate = new DatabaseMetadata(connection, dialect, configuration, true);

            Field sqlExceptionConverterField = DatabaseMetadata.class.getDeclaredField("sqlExceptionConverter");
            sqlExceptionConverterField.setAccessible(true);
            sqlExceptionConverterField.set(this.delegate, sqlExceptionConverter);

            sequences = initSequences(connection, dialect);

            Field typesField = DatabaseMetadata.class.getDeclaredField("types");
            typesField.setAccessible(true);
            this.types = (String[]) typesField.get(this.delegate);

            this.meta = connection.getMetaData();

        } catch (SQLException e) {
            throw new SynchroTechnicalException(t("adagio.persistence.dbMetadata.instanciation.error", connection), e);
        } catch (Exception e) {
            throw new SynchroTechnicalException(t("adagio.persistence.dbMetadata.instanciation.error", connection), e);
        }
        tables = Maps.newTreeMap();
    }

    public int getTableCount() {
        return tables.size();
    }

    public SynchroContext getContext() {
        return this.context;
    }

    public Dialect getDialect() {
        return this.dialect;
    }

    public int getInExpressionCountLimit() {
        return dialect.getInExpressionCountLimit();
    }

    public boolean isSequence(String tableName) {
        String[] strings = StringHelper.split(".", (String) tableName);
        return sequences.contains(StringHelper.toLowerCase(strings[strings.length - 1]));
    }

    public void prepare(Dialect dialect,
            Configuration configuration,
            Set<String> tableNames,
            Predicate<String> tableFilter,
            Predicate<SynchroColumnMetadata> columnFilter,
            boolean enableJoinMetadataLoading) {
        Preconditions.checkArgument(CollectionUtils.isNotEmpty(tableNames) || tableFilter != null,
                "One of 'tableNames' or 'tableFilter' must be set and not empty");

        // Getting tables names to process
        boolean enableFilter = tableFilter != null;
        if (!enableFilter) {
            for (String tablePattern : tableNames) {
                enableFilter = tablePattern.contains("%");
                if (enableFilter)
                    break;
            }
        }

        Set<String> filteredTableNames = tableNames;
        if (enableFilter) {
            if (CollectionUtils.isEmpty(tableNames)) {
                filteredTableNames = getTableNames(tableFilter);
            }
            else {
                filteredTableNames = getTableNames(tableNames, tableFilter);
            }
        }

        // Getting schema
        String jdbcCatalog = configuration.getProperty(Environment.DEFAULT_CATALOG);
        String jdbcSchema = configuration.getProperty(Environment.DEFAULT_SCHEMA);

        for (String tableName : filteredTableNames) {

            if (log.isDebugEnabled()) {
                log.debug("Load metas of table: " + tableName);
            }

            getTable(dialect, tableName, jdbcSchema, jdbcCatalog, false, columnFilter, false);
        }

        Map<String, SynchroTableMetadata> tablesByNames = Maps.newHashMap();
        for (SynchroTableMetadata table : tables.values()) {
            tablesByNames.put(table.getName(), table);

            // Init joins metadata (must be call AFTER getTable())
            if (enableJoinMetadataLoading) {

                if (log.isDebugEnabled()) {
                    log.debug("Load joins of table: " + table.getName());
                }
                table.initJoins(this);
            }

            fireOnTableLoad(table);
        }
    }

    public SynchroTableMetadata getTable(String name) throws HibernateException {
        String defaultSchema = ConfigurationHelper.getString(AvailableSettings.DEFAULT_SCHEMA, configuration.getProperties());
        String defaultCatalog = ConfigurationHelper.getString(AvailableSettings.DEFAULT_CATALOG, configuration.getProperties());

        return getTable(this.dialect, name, defaultSchema, defaultCatalog, false, null, true);
    }

    public SynchroTableMetadata getLoadedTable(String name) throws HibernateException {
        String defaultSchema = ConfigurationHelper.getString(AvailableSettings.DEFAULT_SCHEMA, configuration.getProperties());
        String defaultCatalog = ConfigurationHelper.getString(AvailableSettings.DEFAULT_CATALOG, configuration.getProperties());
        return getLoadedTable(name, defaultSchema, defaultCatalog);
    }

    public SynchroTableMetadata getLoadedTable(String name,
            String schema,
            String catalog) throws HibernateException {
        String key = Table.qualify(catalog, schema, name).toLowerCase();
        return tables.get(key);
    }

    /**
     * Load tables names from database schema, using the given table filter.<br/>
     * This method call {@link #getTableNames(Set<String>,Predicate<String>)} with the table pattern "%".
     * 
     * @param tableFilter
     *            A filter predicate, to filter tables to retrieve. If null: process all tables found.
     * @return All tables names found in database, filtered using the given tableFilter
     * @see #getTableNames(Set<String>,Predicate<String>)
     */
    public Set<String> getTableNames(Predicate<String> tableFilter) {
        return getTableNames(Sets.newHashSet("%"), tableFilter);
    }

    /**
     * Load tables names from database schema, using the given table patterns list, and a optional filter. This use the
     * JDBC metadata API.<br/>
     * This will include Tables and View objects. Synonyms ar includes only if enable in connection properties {@see
     * org.hibernate.cfg.AvailableSettings.ENABLE_SYNONYMS}.
     * 
     * @param tablePatterns
     *            A list of table pattern. Use the pattern '%' to get all tables.
     * @param tableFilter
     *            Optional. A filter predicate, to filter tables to retrieve. If null: process tables found from pattern
     *            will be return.
     * @return All tables names found in database
     * @see org.hibernate.cfg.AvailableSettings.ENABLE_SYNONYMS
     * @see java.sql.DatabaseMetaData#getTables(String,String,String,String[])
     */
    public Set<String> getTableNames(Set<String> tablePatterns, Predicate<String> tableFilter) {
        Preconditions.checkArgument(CollectionUtils.isNotEmpty(tablePatterns));

        Set<String> tablenames = Sets.newHashSet();

        String defaultSchema = ConfigurationHelper.getString(AvailableSettings.DEFAULT_SCHEMA, configuration.getProperties());
        String defaultCatalog = ConfigurationHelper.getString(AvailableSettings.DEFAULT_CATALOG, configuration.getProperties());

        String[] types = null; // available types are: "TABLE", "VIEW", "SYSTEM TABLE", "GLOBAL TEMPORARY",
                                // "LOCAL TEMPORARY", "ALIAS", "SYNONYM"
        if (configuration != null
                && ConfigurationHelper.getBoolean(AvailableSettings.ENABLE_SYNONYMS, configuration.getProperties(), false)) {
            types = new String[] { "TABLE", "VIEW", "SYNONYM" };
        }
        else {
            types = new String[] { "TABLE", "VIEW" };
        }

        ResultSet res = null;
        try {
            if (log.isDebugEnabled()) {
                log.debug("Getting table names, using filter");
            }

            for (String tablePattern : tablePatterns) {
                // first pass on the main schema
                res = meta.getTables(defaultCatalog, defaultSchema, tablePattern, types);
                while (res.next()) {
                    String tableName = res.getString(TABLE_NAME_PATTERN); //$NON-NLS-1$
                    if (!delegate.isSequence(tableName) && (tableFilter == null || tableFilter.apply(tableName))) {
                        if (log.isTraceEnabled()) {
                            log.trace(" " + TABLE_CATALOG_PATTERN + "=" + res.getString(TABLE_CATALOG_PATTERN)
                                    + " " + TABLE_SCHEMA_PATTERN + "=" + res.getString(TABLE_SCHEMA_PATTERN)
                                    + " " + TABLE_NAME_PATTERN + "=" + res.getString(TABLE_NAME_PATTERN)
                                    + " " + TABLE_TYPE_PATTERN + "=" + res.getString(TABLE_TYPE_PATTERN)
                                    + " " + REMARKS_PATTERN + "=" + res.getString(REMARKS_PATTERN));
                        }
                        tablenames.add(tableName);
                    }
                }
            }
        } catch (SQLException e) {
            throw sqlExceptionConverter.convert(e, "Retrieving database table names", "n/a");
        } finally {
            DaoUtils.closeSilently(res);
        }

        return tablenames;
    }

    /**
     * Return all root tables (top level tables).<br/>
     * Return only tables previously loaded using methods getTable() or loadDatabaseMetadata()
     * 
     * @return All loaded tables metadata
     */
    public Set<String> getLoadedRootTableNames() {
        Set<String> tablenames = Sets.newHashSet();
        for (SynchroTableMetadata table : tables.values()) {
            if (table.isRoot()) {
                tablenames.add(table.getName());
            }
        }

        return tablenames;
    }

    /**
     * Return all tables (already loaded).<br/>
     * Return only tables previously loaded using methods getTable() or loadDatabaseMetadata()
     * 
     * @return All loaded tables names
     */
    public Set<String> getLoadedTableNames() {
        Set<String> tablenames = Sets.newHashSet();
        for (SynchroTableMetadata table : tables.values()) {
            tablenames.add(table.getName());
        }

        return tablenames;
    }

    /**
     * @see java.sql.DatabaseMetaData.getExportedKeys(String,String,String)
     */
    public ResultSet getExportedKeys(String catalog, String schema, String table) throws SQLException {
        return meta.getExportedKeys(catalog, schema, table);
    }

    /**
     * @see java.sql.DatabaseMetaData.getImportedKeys(String,String,String)
     */
    public ResultSet getImportedKeys(String catalog, String schema, String table) throws SQLException {
        return meta.getImportedKeys(catalog, schema, table);
    }

    /**
     * @see java.sql.DatabaseMetaData.getPrimaryKeys(String,String,String)
     */
    public ResultSet getPrimaryKeys(String catalog, String schema, String table) throws SQLException {
        return meta.getPrimaryKeys(catalog, schema, table);
    }

    /* -- Internal methods -- */

    protected SynchroTableMetadata getTable(
            Dialect dialect,
            String name,
            String schema,
            String catalog,
            boolean isQuoted,
            Predicate<SynchroColumnMetadata> columnFilter,
            boolean withJoinedTables) throws HibernateException {
        String key = Table.qualify(catalog, schema, name).toLowerCase();
        SynchroTableMetadata synchroTableMetadata = tables.get(key);
        if (synchroTableMetadata == null) {

            TableMetadata tableMetadata = delegate.getTableMetadata(
                    name.toLowerCase(), schema, catalog, isQuoted);
            Preconditions.checkNotNull(tableMetadata, String.format("Could not find db table '%s' (schema=%s, catalog=%s)", name, schema, catalog));

            List<SynchroInterceptor> interceptors = getInterceptors(tableMetadata);

            synchroTableMetadata = new SynchroTableMetadata(
                    this,
                    tableMetadata,
                    interceptors,
                    name, sequences, columnFilter);
            Preconditions.checkNotNull(synchroTableMetadata,
                    "Could not load metadata for table: " + name);

            tables.put(key, synchroTableMetadata);
        }
        return synchroTableMetadata;
    }

    protected Set<String> initSequences(Connection connection, Dialect dialect) throws SQLException {
        Set<String> sequences = Sets.newHashSet();
        if (dialect.supportsSequences()) {
            String sql = dialect.getQuerySequencesString();
            if (sql != null) {

                Statement statement = null;
                ResultSet rs = null;
                try {
                    statement = connection.createStatement();
                    rs = statement.executeQuery(sql);

                    while (rs.next()) {
                        sequences.add(StringHelper.toLowerCase(rs.getString(1)).trim());
                    }
                } finally {
                    rs.close();
                    statement.close();
                }

            }
        }
        return sequences;
    }

    protected List<SynchroInterceptor> getInterceptors(final TableMetadata table) {
        if (interceptors == null) {
            interceptors = SynchroInterceptorUtils.load(SynchroInterceptor.class, this.context);
        }

        Collection<SynchroInterceptor> filteredInterceptors = SynchroInterceptorUtils.filter(
                interceptors,
                this,
                table
                );

        return Lists.newArrayList(filteredInterceptors);
    }

    protected void fireOnTableLoad(SynchroTableMetadata table) {
        List<SynchroInterceptor> interceptors = table.getInterceptors();
        if (CollectionUtils.isNotEmpty(interceptors)) {
            for (SynchroInterceptor interceptor : interceptors) {
                interceptor.onTableLoad(table);
            }
        }
    }
}