package com.zz.util.persist;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.sql.Types;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.zz.util.EpocratesException;
import com.zz.util.ValidationException;



/**
 * Implements API support for common JDBC operations including convenience methods
 * to generate SQL and facilitate usage of prepared statements, connection management,
 * and batch handling.
 *
 * @author zzarate
 *
 */
public abstract class AbstractJdbcFacade {

	private static Log log = LogFactory.getLog( AbstractJdbcFacade.class );

	public final static int DEFAULT_BATCH_THRESHOLD = 10000;  // in testing this seemed to offer a good balance of speed/memory performance

	private String user = null;
	private String password = null;

	private int batchThreshold = DEFAULT_BATCH_THRESHOLD;

	public AbstractJdbcFacade( String user, String password ) {
		this.user = user;
		this.password = password;
	}


	protected String getUser() {
		return user;
	}


	protected String getPassword() {
		return password;
	}



	/**
	 * @param batchThreshold the batchThreshold to set
	 */
	public void setBatchThreshold( int batchThreshold ) {

		this.batchThreshold = batchThreshold;
	}


	/**
	 * @return the batchThreshold
	 */
	public int getBatchThreshold() {

		return batchThreshold;
	}


	/**
	 * Obtain connection to data source.  Concrete implementations are expected to determine
	 * how to obtain connection according to specific implementation (e.g. DriverManager, JNDI DataSource, etc.)
	 *
	 * @param user
	 * @param password
	 * @return
	 * @throws SQLException
	 */
	public abstract Connection getConnection() throws SQLException;


	/**
	 * Executes the argument select SQL and returns results as a list of lists.
	 *
	 * @param sql
	 * @return
	 * @throws SQLException
	 */
	public List<List<?>> select( String sql ) throws SQLException {

	    Connection connection = null;
		ResultSet resultSet = null;

		List<List<?>> list = null;


		if( sql == null || sql.trim().toUpperCase().startsWith( "SELECT" ) == false ) {

			throw( new SQLException("Invalid SELECT query: " + sql ) );
		}

        if( log.isDebugEnabled() ) {
        	log.debug( sql );
        }

        Statement statement = null;

        try {

			connection = getConnection();
			statement = connection.createStatement();

        	resultSet = statement.executeQuery( sql );

	        while( resultSet.next() ) {

	        	if( list == null ) {
	        		list = new ArrayList<List<?>>();
	        	}

	        	List<?> row = getRow( resultSet );

	        	list.add( row );
	        }

        }
        catch( SQLException e ) {

        	throw e;
        }
        catch( Exception e ) {

            throw( new SQLException( "Error executing query: " + sql, e ) );
        }
	    finally {

	    	if( statement != null ) {
	    		statement.close();
			}

	    	if( resultSet != null ) {
	    		resultSet.close();
			}

			if( connection != null ) {
				connection.close();
			}
	    }

		return list;
	}


	/**
	 * Executes the given SQL statement.  May be an INSERT, UPDATE, or DELETE statement
	 * or any SQL statement that returns no value.
	 *
	 * @param sql
	 * @return
	 * 	the number of rows updated by statement
	 *
	 * @throws SQLException
	 */
	public int execute( String sql ) throws SQLException {

		if( log.isDebugEnabled() ) {
			log.debug( sql );
		}

	    Connection connection = null;
	    Statement statement = null;

		int updated = 0;

		try {

			connection = getConnection();
			statement = connection.createStatement();
			updated = statement.executeUpdate( sql );
		}

		finally {

			if( statement != null ) {
				statement.close();
			}

			if( connection != null ) {
				connection.close();
			}
		}

		return updated;
	}



	/**
	 * Update DB rows corresponding to argument values. Method will generate relevant SQL
	 * and prepared statements given table meta-data arguments and will execute in batches
	 * if number of values exceeds batch threshold size.
	 *
	 * @param values
	 * @param table
	 * @param columns
	 * @param primaryKey
	 * @param batchSize
	 * @throws SQLException
	 */
	@SuppressWarnings({ "rawtypes", "unchecked" })
	public void update( List<List<?>> values, String table, List<String> columns, String primaryKey, int batchSize ) throws SQLException {

		StringBuffer sb = new StringBuffer();
		StringBuffer where = new StringBuffer();
		sb.append("UPDATE ").append( table ).append(" SET ");

		int primaryIndex = -1;

		int i = 0;
		for( String column : columns ) {

			if( i > 0 ) {
				sb.append(",");
			}


			sb.append( column ).append( "=" ).append("?");

			if( column.equalsIgnoreCase( primaryKey ) ) {

				where.append(" WHERE ").append( column ).append( "=" ).append( "?" );
				primaryIndex = i;
			}

			i++;
		}

		sb.append( where.toString() );

		String sql = sb.toString();


		// in order to update value where PRIMARY_KEY = primaryKeyValue
		// we need to augment the value set so that this can occur dynamically
		// in context of prepared statement update
		// (append primary key value to end of list so that it appears in
		// correct place relative to prepared statement where clause)
		for( List row : values ) {

			Object primaryValue = row.get( primaryIndex );
			row.add( primaryValue );
		}

		update( values, sql, batchSize );
	}


	/**
	 * Retrieve NEXT ID for given sequence.
	 *
	 * (NOTE: default implementation is compatible with all DBs)
	 *
	 * @param sequence
	 * @return
	 * @throws SQLException
	 */
	public Integer nextId( String sequence ) throws SQLException {

		try {

			if( sequence.toUpperCase().endsWith( ".NEXTVAL" ) == false ) {
				sequence = sequence + ".NEXTVAL";
			}

			List<List<?>> results = select( "select " + sequence + " from dual" );

			BigDecimal id = (BigDecimal)results.get( 0 ).get( 0 );

			Integer nextId = Integer.valueOf( id.intValue() );
			return nextId;
		}
		catch( SQLException e ) {

			throw e;
		}
		catch( Exception e ) {

			throw( new SQLException("Error getting next ID for sequence: " + sequence ) );
		}

	}




	/**
	 * Performs a batch update using the argument sql string as a prepared statement against the argument values.
	 *
	 * Batch updates are transactional, however it is important to note that very large batches may
	 * incur out of memory errors as the batch prepared statement list grows very large.  Commit is required to
	 * flush batch prior to it growing to unmanagable sizes.  It is left to the caller to determine batch "size"
	 * by limiting the argument value list size to an acceptable value.
	 *
	 * Finding optimal batch size may require some experimentation.  Larger batches should result in more optimal
	 * speed performance, but may result in OutOfMemoryErrors if the batch is too large.  This issue may even occur
	 * if multiple successful batches are executed in sequence.
	 *
	 * NOTE: In development of this functionality I was attempting to load a dataset with approximately 800,000 records.
	 * I found that queuing batches of 10,000 would work successfully, but would eventually run OOM after processing
	 * 500,000+ records as it appeared the memory required for the prepared statement batching would outrun
	 * garbage collection.  Ultimately went with batches of 1000 which seemed to offer the ultimate performance balance.
	 *
	 * @param values
	 * @param sql
	 *
	 * @throws SQLException
	 */
	public void update( List<List<?>> values, String sql ) throws SQLException {

		if( log.isDebugEnabled() ) {
			log.debug( sql );
		}

//		INSERT INTO CUSTOMERS VALUES (?,?,?)
//		statement.setInt(1,1);
//		statement.setString(2, "J Smith");
//		statement.setString(3, "617 555-1323");

	    Connection connection = null;
	    PreparedStatement statement = null;

		try {

			connection = getConnection();
			connection.setAutoCommit( false ); // allow for transactional batch rollback
			statement = connection.prepareStatement( sql );

			// each row should represent a insert statement
			for( List<?> row : values ) {

				for( int i = 0; i < row.size(); i++ ) {

					Object value = row.get( i );
					setStatementValue( statement, i+1, value );  // sql index starts at 1 so offset accordingly
				}

				statement.addBatch();
			}

			long t0 = System.currentTimeMillis();

			@SuppressWarnings("unused")
			int[] updateCounts = statement.executeBatch();
			connection.commit();
			statement.clearBatch();  // not sure if this is necessary

			long t1 = System.currentTimeMillis();

			if( log.isDebugEnabled() ) {
				String elapsedSeconds = ( new DecimalFormat( "#.##" ).format( ((float) (t1-t0) / 1000) ) );
				log.debug( values.size() + " records batch inserted (" + elapsedSeconds + "s) " );
			}
		}
		catch ( Exception e ) {

			try {
				connection.rollback();  // if failure occurs then rollback transaction
			}
			catch ( Exception e2 ) {

				throw ( new SQLException( "Error attempting to rollback batch insert", e2 ) );
			}

			throw ( new SQLException( "Error execting batch insert", e ) );
		}
		finally {

			if( statement != null ) {
				statement.close();
			}

			if( connection != null ) {
				connection.close();
			}
		}
	}


	/**
	 * Insert values is sub-batches of "threshold" size.
	 *
	 * @param values
	 * @param sql
	 * @param threshold
	 * @throws SQLException
	 */
	public void update( List<List<?>> values, String sql, int batchSize ) throws SQLException {

		if( batchSize < 0 || batchSize <= values.size() ) {

			update( values, sql );
			return;
		}

		try {

			int threshold = getBatchThreshold();

			if( log.isDebugEnabled() ) {
				log.debug( "Inserting " + values.size() + " values in batches of " + threshold );
			}

			long t0 = System.currentTimeMillis();


			List<List<?>> batch = new ArrayList<List<?>>( threshold );

			int count = 0;
			for( List<?> row : values ){

				batch.add( row );
				count++;

				// execute in batches to scale performance and avoid running out of memory
				if( batch.size() % threshold == 0 ) {

					update( batch, sql );
					batch.clear();

					if( log.isDebugEnabled() ) {
						log.debug( ((int) Math.floor( count * 100 / values.size() )) + "%" );
					}
				}

			}

			// modulus batch
			if( batch.isEmpty() == false ) {
				//jdbcBroker.insertBatch( sql, values );
				update( batch, sql );
				batch = null;
			}

			// commit transaction
			//connection.commit();

			long t1 = System.currentTimeMillis();

			if( log.isDebugEnabled() ) {
				String elapsedSeconds = ( new DecimalFormat( "#.##" ).format( ((float) (t1-t0) / 1000) ) );
				log.debug( values.size() + " records batch inserted (" + elapsedSeconds + "s) " );
			}
		}
		catch( Exception e ) {

			throw ( new SQLException( "Error execting batch insert", e ) );
		}
	}



	/**
	 * Retrieve prepared statement SQL for batch insert.
	 * Default implementation assumes a sequence is defined.
	 */
	public void insert( List<List<?>> values, String table, List<String> columns, String primaryKey, String sequence, int batchSize ) throws SQLException {

		StringBuffer sb = new StringBuffer();
		StringBuffer prep = new StringBuffer();
		sb.append("INSERT INTO ").append( table );
		sb.append(" (" );

		boolean first = true;
		for( String column : columns ) {


			if( first == false ) {
				sb.append(",");
				prep.append(",");
			}

			sb.append( column );
			if( sequence != null && column.equalsIgnoreCase( primaryKey ) ) {

				prep.append( sequence );  // insert assumes sequence implementation
			}
			else {
				prep.append("?");
			}

			first = false;
		}

		sb.append(") VALUES (").append( prep.toString() ).append(")");

		String sql = sb.toString();

		update( values, sql, batchSize );  // use batching by default
	}


	/**
	 * Converts column name list to a select clause, null is interpreted as all columns '*'
	 *
	 * If a prefix is specified then prepend then it will be prepended to each column name
	 *
	 * @param columns
	 * @param prefix
	 * @return
	 */
	protected String getSelectClause( List<String> columns, String prefix ) {

		if( prefix == null ) {
			prefix = "";
		}

		if( prefix.length() > 0 && prefix.endsWith(".") == false ) {
			prefix += ".";
		}

		if( columns == null || columns.size() == 0 ) {

			return prefix + "*";
		}

		StringBuffer clause = new StringBuffer();
		boolean first = true;
		for( String column : columns ) {

			if( first == false ) {
				clause.append( ", " );
			}

			clause.append( prefix ).append( column );
			first = false;
		}

		return clause.toString();
	}


	@SuppressWarnings("rawtypes")
	protected void setStatementValue( PreparedStatement statement, int index, Object value ) throws SQLException {

		try {

			if( value == null ) {
				statement.setString( index, null );
			}
			else if( value instanceof String ) {
				statement.setString( index, value.toString() );
			}
			else if( value instanceof Boolean ) {
				statement.setBoolean( index, ((Boolean) value).booleanValue() );
			}
			else if( value instanceof Long ) {
				statement.setLong( index, ((Long) value).longValue());
			}
			else if( value instanceof Integer ) {
				statement.setInt( index, ((Integer) value).intValue());
			}
			else if( value instanceof Double ) {
				statement.setDouble( index, ((Double) value).doubleValue());
			}
			else if( value instanceof Date ) {
				statement.setTimestamp( index, new Timestamp( ((Date) value).getTime() ) );
			}
			else if( value instanceof Timestamp ) {
				statement.setTimestamp( index, (Timestamp) value );
			}
			else if( value instanceof List ) {

				// list is expected for an IN expression
				List list = (List) value;
				Iterator iterator = list.iterator();

				for( int i = index; iterator.hasNext(); i++ ) {

					Object o = iterator.next();
					setStatementValue( statement, i, o );
				}
			}
			else {

				throw( new SQLException("Value of type " + value.getClass().getName() + "' is not supported.") );
			}
		}
		catch( SQLException e ) {

			throw e;
		}
		catch( Exception e ) {

			throw( new SQLException( "Error setting statement value.", e ) );
		}

	}


	@SuppressWarnings({ "rawtypes", "unchecked" })
	protected List getRow( ResultSet rs ) throws SQLException {

		List row = null;

		try {

			ResultSetMetaData meta = rs.getMetaData();

	    	int columns = meta.getColumnCount();

	    	row = new ArrayList( columns );

	    	for( int i = 1; i <= columns; i++ ) {

	    		int type = meta.getColumnType( i );

	    		Object value = null;
	    		if( Types.NULL == type ) {
	    			value = null;
	    		}
	    		else if( Types.VARCHAR == type ) {
	    			value = rs.getString( i );
	    		}
	    		else if( Types.INTEGER == type ) {
	    			value = Integer.valueOf( rs.getInt( i ) );
	    		}
	    		else if( Types.BIGINT == type ) {
	    			value = Long.valueOf( rs.getLong( i ) );
	    		}
	    		else if( Types.DOUBLE == type ) {
	    			value = rs.getDouble( i );
	    		}
	    		else if( Types.TIMESTAMP == type ) {
	    			value = rs.getTimestamp( i );
	    		}
	    		else if( Types.BOOLEAN == type ) {
	    			value = rs.getBoolean( i );
	    		}
	    		else if( Types.NUMERIC == type ) {
	    			value = rs.getBigDecimal( i );
	    		}
	    		else {

	    			throw( new SQLException( "Unhandled sql type: " + type ) );
	    		}

	    		row.add( value );
	    	}
		}
		catch( SQLException e ) {
			throw e;
		}
		catch( Exception e ) {
			throw( new SQLException( "Error building persisted object", e ) );
		}

		return row;
	}


	/**
	 * Return value as valid SQL.  Implmentation assumes toString() implementation for value class
	 * @param value
	 * @return
	 */
	public String toSql( Object value ) throws EpocratesException {

		if( value == null ) {

			return "null";
		}

		if( value instanceof Integer ) {

			return( value.toString() );
		}
		else if( value instanceof Double ) {

			return( value.toString() );
		}
		else if( value instanceof String ) {

			return( "\'" + value.toString() + "\'" );
		}
		else {

			throw( new ValidationException( "Value of type " + value.getClass().getName() + " could not be converted to SQL") );
		}
	}


	/**
	 * This implementation accomodates the need to state an explicit sort order by value.
	 *
	 * For exmaple, we have an IN statement:
	 * 	WHERE id IN( 5,2,6,8,12,1 )
	 * and we want the results to match this order using the following order by clause
	 * 	ORDER BY DECODE(id,5,1,2,2,6,3,8,4,12,5,1,6);
	 *
	 * @param query
	 * @return
	 */
	public String getIndexedSort( String field, List<?> values ) {

		StringBuilder sb = new StringBuilder();
		sb.append("DECODE(");
		sb.append( field );

		int i = 1;
		for( Object o : values ) {

			sb.append(",").append( o ).append(",").append( i );
			i++;
		}

		sb.append(")");

		return sb.toString();
	}


}