/*
 * Copyright OpenSearch Contributors
 * SPDX-License-Identifier: Apache-2.0
 */


package org.opensearch.jdbc;

import org.opensearch.jdbc.protocol.JdbcQueryRequest;
import org.opensearch.jdbc.types.OpenSearchType;
import org.opensearch.jdbc.types.TypeConverters;
import org.opensearch.jdbc.internal.util.SqlParser;
import org.opensearch.jdbc.logging.Logger;
import org.opensearch.jdbc.protocol.JdbcDateTimeFormatter;
import org.opensearch.jdbc.protocol.JdbcQueryParam;

import java.io.InputStream;
import java.io.Reader;
import java.math.BigDecimal;
import java.net.URL;
import java.sql.Array;
import java.sql.Blob;
import java.sql.Clob;
import java.sql.Date;
import java.sql.JDBCType;
import java.sql.NClob;
import java.sql.ParameterMetaData;
import java.sql.PreparedStatement;
import java.sql.Ref;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.RowId;
import java.sql.SQLDataException;
import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
import java.sql.SQLNonTransientException;
import java.sql.SQLXML;
import java.sql.Time;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.Arrays;
import java.util.Calendar;
import java.util.HashMap;
import java.util.Map;

public class PreparedStatementImpl extends StatementImpl implements PreparedStatement {
    // TODO - support String representations containing Timestamp With Timezone
    protected String sql;
    private JdbcQueryParam[] parameters;

    public PreparedStatementImpl(ConnectionImpl connection, String sql, Logger log) throws SQLException {
        super(connection, log);
        this.sql = sql;

        try {
            parameters = new JdbcQueryParam[SqlParser.countParameterMarkers(sql)];
        } catch (IllegalArgumentException iae) {
            logAndThrowSQLException(
                    log,
                    new SQLNonTransientException( "Error preparing SQL statement: "+iae.getMessage(), iae));
        }
    }

    @Override
    public ResultSet executeQuery() throws SQLException {
        log.debug(() -> logEntry("executeQuery()"));
        checkOpen();
        ResultSet rs = executeQueryX(getFetchSize());
        log.debug(() -> logExit("executeQuery", rs));
        return rs;
    }

    protected ResultSet executeQueryX(int fetchSize) throws SQLException {
        checkParamsFilled();
        JdbcQueryRequest jdbcQueryRequest = new JdbcQueryRequest(sql, fetchSize);
        jdbcQueryRequest.setParameters(Arrays.asList(parameters));
        return executeQueryRequest(jdbcQueryRequest);
    }

    @Override
    public int executeUpdate() throws SQLException {
        throw new SQLFeatureNotSupportedException("Updates are not supported");
    }

    @Override
    public void setNull(int parameterIndex, int sqlType) throws SQLException {
        log.debug(() -> logEntry("setNull(%d, %d)", parameterIndex, sqlType));
        checkOpen();
        setParameter(parameterIndex, OpenSearchType.fromJdbcType(JDBCType.valueOf(sqlType)).getTypeName(), null);
        log.debug(() -> logExit("setNull"));
    }

    @Override
    public void setBoolean(int parameterIndex, boolean x) throws SQLException {
        log.debug(() -> logEntry("setBoolean(%d, %s)", parameterIndex, x));
        checkOpen();
        setObjectX(parameterIndex, x, Types.BOOLEAN);
        log.debug(() -> logExit("setBoolean"));
    }

    @Override
    public void setByte(int parameterIndex, byte x) throws SQLException {
        log.debug(() -> logEntry("setByte(%d, %d)", parameterIndex, x));
        checkOpen();
        setObjectX(parameterIndex, x, Types.TINYINT);
        log.debug(() -> logExit("setByte"));
    }

    @Override
    public void setShort(int parameterIndex, short x) throws SQLException {
        log.debug(() -> logEntry("setShort(%d, %d)", parameterIndex, x));
        checkOpen();
        setObjectX(parameterIndex, x, Types.SMALLINT);
        log.debug(() -> logExit("setShort"));
    }

    @Override
    public void setInt(int parameterIndex, int x) throws SQLException {
        log.debug(() -> logEntry("setInt(%d, %d)", parameterIndex, x));
        checkOpen();
        setObjectX(parameterIndex, x, Types.INTEGER);
        log.debug(() -> logExit("setInt"));
    }

    @Override
    public void setLong(int parameterIndex, long x) throws SQLException {
        log.debug(() -> logEntry("setLong(%d, %d)", parameterIndex, x));
        checkOpen();
        setObjectX(parameterIndex, x, Types.BIGINT);
        log.debug(() -> logExit("setLong"));
    }

    @Override
    public void setFloat(int parameterIndex, float x) throws SQLException {
        log.debug(() -> logEntry("setFloat(%d, %f)", parameterIndex, x));
        checkOpen();
        setObjectX(parameterIndex, x, Types.REAL);
        log.debug(() -> logExit("setFloat"));
    }

    @Override
    public void setDouble(int parameterIndex, double x) throws SQLException {
        log.debug(() -> logEntry("setDouble(%d, %f)", parameterIndex, x));
        checkOpen();
        setObjectX(parameterIndex, x, Types.DOUBLE);
        log.debug(() -> logExit("setDouble"));
    }

    @Override
    public void setBigDecimal(int parameterIndex, BigDecimal x) throws SQLException {
        throw new SQLFeatureNotSupportedException("BigDecimal is not supported");
    }

    @Override
    public void setString(int parameterIndex, String x) throws SQLException {
        log.debug(() -> logEntry("setString(%d, %s)", parameterIndex, x));
        checkOpen();
        setParameter(parameterIndex, OpenSearchType.fromJdbcType(JDBCType.VARCHAR).getTypeName(), x);
        log.debug(() -> logExit("setString"));
    }

    @Override
    public void setBytes(int parameterIndex, byte[] x) throws SQLException {
        throw new SQLFeatureNotSupportedException("Setting byte arrays is not supported");
    }

    @Override
    public void setDate(int parameterIndex, Date x) throws SQLException {
        log.debug(() -> logEntry("setDate(%d, %s)", parameterIndex, x));
        checkOpen();
        setObjectX(parameterIndex, x, Types.DATE);
        log.debug(() -> logExit("setDate"));
    }

    @Override
    public void setTime(int parameterIndex, Time x) throws SQLException {
        throw new SQLFeatureNotSupportedException("Setting Time is not supported");
    }

    @Override
    public void setTimestamp(int parameterIndex, Timestamp x) throws SQLException {
        log.debug(() -> logEntry("setTimestamp(%d, %s)", parameterIndex, x));
        checkOpen();
        setObjectX(parameterIndex, x, Types.TIMESTAMP);
        log.debug(() -> logExit("setTimestamp"));
    }

    @Override
    public void setAsciiStream(int parameterIndex, InputStream x, int length) throws SQLException {
        throw new SQLFeatureNotSupportedException("Setting ASCII Stream is not supported");
    }

    @Override
    public void setUnicodeStream(int parameterIndex, InputStream x, int length) throws SQLException {
        throw new SQLFeatureNotSupportedException("Setting Unicode Stream is not supported");
    }

    @Override
    public void setBinaryStream(int parameterIndex, InputStream x, int length) throws SQLException {
        throw new SQLFeatureNotSupportedException("Setting Binary Stream is not supported");
    }

    @Override
    public void clearParameters() throws SQLException {
        log.debug(() -> logEntry("clearParameters()"));
        for (int i = 0; i < parameters.length; i++)
            parameters[i] = null;
        log.debug(() -> logExit("clearParameters"));
    }

    @Override
    public void setObject(int parameterIndex, Object x, int targetSqlType) throws SQLException {
        log.debug(() -> logEntry("setObject(%d, %s, %d)", parameterIndex, x, targetSqlType));
        checkOpen();
        setObjectX(parameterIndex, x, targetSqlType, null);
        log.debug(() -> logExit("setObject"));
    }

    private void setObjectX(int parameterIndex, Object x, int targetSqlType) throws SQLException {
        setObjectX(parameterIndex, x, targetSqlType, null);
    }

    private void setObjectX(int parameterIndex, Object x, int targetSqlType, Map<String, Object> conversionParams)
            throws SQLException {
        JDBCType jdbcType = JDBCType.valueOf(targetSqlType);
        OpenSearchType openSearchType = OpenSearchType.fromJdbcType(jdbcType);

        Object value = TypeConverters.getInstance(jdbcType).convert(x, null, conversionParams);

        // flow date/times in JDBC escape format
        if (jdbcType == JDBCType.TIMESTAMP) {
            value = JdbcDateTimeFormatter.JDBC_FORMAT.format((Timestamp) value);
        } else if (jdbcType == JDBCType.DATE) {
            value = JdbcDateTimeFormatter.JDBC_FORMAT.format((Date) value);
        }

        setParameter(parameterIndex, openSearchType.getTypeName(), value);
    }

    @Override
    public void setObject(int parameterIndex, Object x) throws SQLException {
        log.debug(() -> logEntry("setObject(%d, %s)", parameterIndex, x));
        checkOpen();
        setObjectX(parameterIndex, x, javaToSqlType(x));
        log.debug(() -> logExit("setObject"));
    }

    private JDBCType javaToJDBCType(Object x) throws SQLException {
        // Based on JDBC spec, Table B-4
        // TODO support java.time.* and JDBCType.TIME_WITH_TIMEZONE/JDBCType.TIMESTAMP_WITH_TIMEZONE

        if (x instanceof String)
            return JDBCType.VARCHAR;
        else if (x instanceof Boolean)
            return JDBCType.BOOLEAN;
        else if (x instanceof Byte)
            return JDBCType.TINYINT;
        else if (x instanceof Short)
            return JDBCType.SMALLINT;
        else if (x instanceof Integer)
            return JDBCType.INTEGER;
        else if (x instanceof Long)
            return JDBCType.BIGINT;
        else if (x instanceof Float)
            return JDBCType.REAL;
        else if (x instanceof Double)
            return JDBCType.DOUBLE;
        else if (x instanceof byte[])
            return JDBCType.VARBINARY;
        else if (x instanceof java.sql.Date)
            return JDBCType.DATE;
        else if (x instanceof java.sql.Timestamp)
            return JDBCType.TIMESTAMP;
        else
            throw new SQLDataException("Objects of type " + x.getClass().getName() + " not supported.");
    }

    private int javaToSqlType(Object x) throws SQLException {
        return javaToJDBCType(x).getVendorTypeNumber();
    }

    @Override
    public boolean execute() throws SQLException {
        log.debug(() -> logEntry("execute()"));
        checkOpen();
        executeQueryX(getFetchSize());
        log.debug(() -> logExit("execute", true));
        return true;
    }

    @Override
    public void addBatch() throws SQLException {
        throw new SQLFeatureNotSupportedException("addBatch is not supported");
    }

    @Override
    public void setCharacterStream(int parameterIndex, Reader reader, int length) throws SQLException {
        throw new SQLFeatureNotSupportedException("setCharacterStream is not supported");
    }

    @Override
    public void setRef(int parameterIndex, Ref x) throws SQLException {
        throw new SQLFeatureNotSupportedException("setRef is not supported");
    }

    @Override
    public void setBlob(int parameterIndex, Blob x) throws SQLException {
        throw new SQLFeatureNotSupportedException("setBlob is not supported");
    }

    @Override
    public void setClob(int parameterIndex, Clob x) throws SQLException {
        throw new SQLFeatureNotSupportedException("setClob is not supported");
    }

    @Override
    public void setArray(int parameterIndex, Array x) throws SQLException {
        throw new SQLFeatureNotSupportedException("setArray is not supported");
    }

    @Override
    public ResultSetMetaData getMetaData() throws SQLException {
        // can only return metadata after query execution
        log.debug(() -> logEntry("getMetaData ()"));
        ResultSetMetaData rsmd = resultSet != null ? resultSet.getMetaData() : null;
        log.debug(() -> logExit("getMetaData", rsmd));
        return rsmd;
    }

    @Override
    public void setDate(int parameterIndex, Date x, Calendar cal) throws SQLException {
        log.debug(() -> logEntry("setDate (%d, %s, %s)", parameterIndex, x,
                cal == null ? "null" : "Calendar TZ= " + cal.getTimeZone()));
        checkOpen();
        Map<String, Object> conversionParams = new HashMap<>();
        conversionParams.put("calendar", cal);

        setObjectX(parameterIndex, x, Types.DATE, conversionParams);
        log.debug(() -> logExit("setDate"));
    }

    @Override
    public void setTime(int parameterIndex, Time x, Calendar cal) throws SQLException {
        // TODO - implement
    }

    @Override
    public void setTimestamp(int parameterIndex, Timestamp x, Calendar cal) throws SQLException {
        log.debug(() -> logEntry("setTimestamp (%d, %s, %s)", parameterIndex, x,
                cal == null ? "null" : "Calendar TZ= " + cal.getTimeZone()));
        checkOpen();
        Map<String, Object> conversionParams = new HashMap<>();
        conversionParams.put("calendar", cal);
        setObjectX(parameterIndex, x, Types.TIMESTAMP, conversionParams);
        log.debug(() -> logExit("setTimestamp"));
    }

    @Override
    public void setNull(int parameterIndex, int sqlType, String typeName) throws SQLException {
        // TODO - implement?
    }

    @Override
    public void setURL(int parameterIndex, URL x) throws SQLException {
        throw new SQLFeatureNotSupportedException("setURL not supported");
    }

    @Override
    public ParameterMetaData getParameterMetaData() throws SQLException {
        // can't determine parameterMetadata as we don't have a query
        // "prepare" phase that could return us this info from the server
        // where the query gets parsed
        return null;
    }

    @Override
    public void setRowId(int parameterIndex, RowId x) throws SQLException {
        throw new SQLFeatureNotSupportedException("setRowId not supported");
    }

    @Override
    public void setNString(int parameterIndex, String value) throws SQLException {
        throw new SQLFeatureNotSupportedException("setNString not supported");
    }

    @Override
    public void setNCharacterStream(int parameterIndex, Reader value, long length) throws SQLException {
        throw new SQLFeatureNotSupportedException("setNCharacterStream not supported");
    }

    @Override
    public void setNClob(int parameterIndex, NClob value) throws SQLException {
        throw new SQLFeatureNotSupportedException("setNClob not supported");
    }

    @Override
    public void setClob(int parameterIndex, Reader reader, long length) throws SQLException {
        throw new SQLFeatureNotSupportedException("setClob not supported");
    }

    @Override
    public void setBlob(int parameterIndex, InputStream inputStream, long length) throws SQLException {
        throw new SQLFeatureNotSupportedException("setBlob not supported");
    }

    @Override
    public void setNClob(int parameterIndex, Reader reader, long length) throws SQLException {
        throw new SQLFeatureNotSupportedException("setNClob not supported");
    }

    @Override
    public void setSQLXML(int parameterIndex, SQLXML xmlObject) throws SQLException {
        throw new SQLFeatureNotSupportedException("setSQLXML not supported");
    }

    @Override
    public void setObject(int parameterIndex, Object x, int targetSqlType, int scaleOrLength) throws SQLException {
        // currently ignore scaleOrLength
        setObjectX(parameterIndex, x, targetSqlType);
    }

    @Override
    public void setAsciiStream(int parameterIndex, InputStream x, long length) throws SQLException {
        throw new SQLFeatureNotSupportedException("setAsciiStream not supported");
    }

    @Override
    public void setBinaryStream(int parameterIndex, InputStream x, long length) throws SQLException {
        throw new SQLFeatureNotSupportedException("setBinaryStream not supported");
    }

    @Override
    public void setCharacterStream(int parameterIndex, Reader reader, long length) throws SQLException {
        throw new SQLFeatureNotSupportedException("setCharacterStream not supported");
    }

    @Override
    public void setAsciiStream(int parameterIndex, InputStream x) throws SQLException {
        throw new SQLFeatureNotSupportedException("setAsciiStream not supported");
    }

    @Override
    public void setBinaryStream(int parameterIndex, InputStream x) throws SQLException {
        throw new SQLFeatureNotSupportedException("setBinaryStream not supported");
    }

    @Override
    public void setCharacterStream(int parameterIndex, Reader reader) throws SQLException {
        throw new SQLFeatureNotSupportedException("setCharacterStream not supported");
    }

    @Override
    public void setNCharacterStream(int parameterIndex, Reader value) throws SQLException {
        throw new SQLFeatureNotSupportedException("setNCharacterStream not supported");
    }

    @Override
    public void setClob(int parameterIndex, Reader reader) throws SQLException {
        throw new SQLFeatureNotSupportedException("setClob not supported");
    }

    @Override
    public void setBlob(int parameterIndex, InputStream inputStream) throws SQLException {
        throw new SQLFeatureNotSupportedException("setBlob not supported");
    }

    @Override
    public void setNClob(int parameterIndex, Reader reader) throws SQLException {
        throw new SQLFeatureNotSupportedException("setNClob not supported");
    }

    private void checkParamsFilled() throws SQLException {
        int filled = 0;

        for (int i = 0; i < parameters.length; i++) {
            if (parameters[i] != null) {
                filled++;
            }
        }

        if (filled < parameters.length)
            logAndThrowSQLException(log, new SQLDataException(
                    String.format("Missing parameter values. The PreparedStatement " +
                                    "requires %d parameter values but only %d were found.",
                            parameters.length, filled)));
    }

    protected void setParameter(int index, String type, Object value) throws SQLException {
        checkParamIndex(index);
        parameters[index - 1] = new JdbcQueryParam(type, value);
    }

    private void checkParamIndex(int index) throws SQLException {
        if (parameters == null || index < 1 || index > parameters.length)
            logAndThrowSQLException(log, new SQLDataException("Invalid parameter index " + index));
    }
}