package com.atlassian.stash.internal.liquibase;

import com.google.common.io.CharStreams;
import liquibase.change.custom.CustomTaskChange;
import liquibase.database.Database;
import liquibase.exception.CustomChangeException;
import liquibase.exception.SetupException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowCallbackHandler;

import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.Charset;
import java.sql.Clob;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;

/**
 * Converts values stored as {@code CLOB}s (OIDs) in Postgres to {@code String}s stored directly in the column.
 * <p>
 *
 * @since 1.3
 */
@SuppressWarnings("unused") //All properties are used via Reflection from Liquibase
public class PostgresClobToStringChange extends AbstractCustomChange implements CustomTaskChange {

    public static final String SELECT_SQL = "select %1$s, %2$s from %3$s";
    public static final String UPDATE_SQL = "update %3$s set %2$s = ? where %1$s = ?";

    private String columnName;
    private String primaryKeyColumnName;
    private String schemaName;
    private String tableName;
    private int total;
    private int updated;

    @Override
    public void execute(Database database) throws CustomChangeException {
        String table = database.escapeTableName(schemaName, tableName);
        String column = database.escapeColumnName(schemaName, tableName, columnName);
        String primaryKeyColumn = database.escapeColumnName(schemaName, tableName, primaryKeyColumnName);

        final String select = String.format(SELECT_SQL, primaryKeyColumn, column, table);
        final String update = String.format(UPDATE_SQL, primaryKeyColumn, column, table);

        final JdbcTemplate template = LiquibaseUtils.getJdbcTemplate(database);
        template.query(select, new RowCallbackHandler() {

            @Override
            public void processRow(ResultSet resultSet) throws SQLException {
                ++total;

                try {
                    Clob clob = resultSet.getClob(2);

                    //If we can get the value as a CLOB, it means it's stored as an OID. We need to convert it to a
                    //String and update the row accordingly.
                    Object id = resultSet.getObject(1);
                    String value = toString(clob);

                    //Note the use of LONGVARCHAR here is to encourage Spring to use Statement.setString
                    template.update(update, new Object[]{value, id},
                            new int[]{Types.LONGVARCHAR, resultSet.getMetaData().getColumnType(1)});
                    ++updated;
                } catch (SQLException e) {
                    if ("22003".equals(e.getSQLState())) { //PSQLState.NUMERIC_VALUE_OUT_OF_RANGE
                        //If the SQL state is 22003, it means the value stored in the column is not a number. Ergo,
                        //the CLOB's text is already in the column and everything is correct.
                        //
                        //For more details, see AbstractJdbc2ResultSet.toLong, around lines 2789 and 2796.
                    } else {
                        throw e;
                    }
                }
            }

            /**
             * Reads the provided {@code clob} into a {@code String}.
             *
             * @param clob the clob to read
             * @return the entire contents of the {@code CLOB}
             * @throws SQLException if a stream cannot be opened from the {@code CLOB}, or if it cannot be read
             */
            private String toString(Clob clob) throws SQLException {
                InputStream stream = clob.getAsciiStream();

                try {
                    return CharStreams.toString(new InputStreamReader(stream, Charset.forName("ASCII")));
                } catch (IOException e) {
                    throw new SQLException("Failed to read CLOB contents", e);
                }
            }
        });
    }

    @Override
    public String getConfirmationMessage() {
        return String.format("Converted CLOBs to Strings for %1$d of %2$d rows in %3$s", updated, total, tableName);
    }

    @Override
    public void setUp() throws SetupException {
        notBlank(columnName, "columnName");
        notBlank(primaryKeyColumnName, "primaryKeyColumnName");
        notBlank(tableName, "tableName");
    }

    public void setColumnName(String columnName) {
        this.columnName = columnName;
    }

    public void setPrimaryKeyColumnName(String primaryKeyColumnName) {
        this.primaryKeyColumnName = primaryKeyColumnName;
    }

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

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