/*
 * Licensed to ElasticSearch and Shay Banon under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. ElasticSearch licenses this
 * file to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.xbib.elasticsearch.river.jdbc.strategy.column;

import java.io.IOException;
import java.security.NoSuchAlgorithmException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.ESLoggerFactory;
import org.elasticsearch.common.unit.TimeValue;
import org.xbib.elasticsearch.river.jdbc.strategy.simple.SimpleRiverSource;
import org.xbib.elasticsearch.river.jdbc.strategy.simple.SimpleValueListener;
import org.xbib.elasticsearch.river.jdbc.support.Operations;
import org.xbib.elasticsearch.river.jdbc.support.ValueListener;

/**
 * River source implementation for the 'column' strategy
 * 
 * @author Piotr Śliwa <piotr.sliwa@zineinc.com>
 */
public class ColumnRiverSource extends SimpleRiverSource {
    
    private final ESLogger logger = ESLoggerFactory.getLogger(ColumnRiverSource.class.getName());
    
    private DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    
    @Override
    public String strategy() {
        return "column";
    }

    @Override
    public String fetch() throws SQLException, IOException {
        
        String sql = context.pollStatement();
        
        Connection connection = connectionForReading();
        
        List<OpInfo> opInfos = getOpInfos(connection);

        Timestamp lastRunTimestamp = getLastRunTimestamp();
        
        for(OpInfo opInfo : opInfos) {
            fetch(connection, sql, opInfo, lastRunTimestamp);
        }

        return null;
    }    

    private List<OpInfo> getOpInfos(Connection connection) throws SQLException {
        
        String quoteString = getIdentifierQuoteString(connection);
        
        List<OpInfo> opInfos = new LinkedList<OpInfo>();
        
        opInfos.add(new OpInfo(Operations.OP_CREATE, quoteColumn(context.columnCreatedAt(), quoteString)+" >= ?"));
        opInfos.add(new OpInfo(Operations.OP_INDEX, quoteColumn(context.columnUpdatedAt(), quoteString)+" >= ? AND ("+quoteColumn(context.columnCreatedAt(), quoteString)+" IS NULL OR "+quoteColumn(context.columnCreatedAt(), quoteString)+" < ?)", 2));

        if(context.columnDeletedAt() != null) {
            opInfos.add(new OpInfo(Operations.OP_DELETE, quoteColumn(context.columnDeletedAt(), quoteString)+" >= ?"));
        }

        return opInfos;
    }
    
    private String getIdentifierQuoteString(Connection connection) throws SQLException {
        
        if(!context.columnEscape()) {
            return "";
        }
        
        String quoteString = connection.getMetaData().getIdentifierQuoteString();
        quoteString = quoteString == null ? "" : quoteString;
        
        return quoteString;
    }
    
    private String quoteColumn(String column, String quote) {
        return quote+column+quote;
    }

    private Timestamp getLastRunTimestamp() {
        Map jdbcSettings = (Map) context.riverSettings().get("jdbc");
        
        if(jdbcSettings == null || jdbcSettings.get(ColumnRiverFlow.LAST_RUN_TIME) == null) {
            return new Timestamp(0);
        }
        
        TimeValue lastRunTime = (TimeValue) jdbcSettings.get(ColumnRiverFlow.LAST_RUN_TIME);
        
        return new Timestamp(lastRunTime.millis());
    }

    private void fetch(Connection connection, String sql, OpInfo opInfo, Timestamp lastRunTimestamp) throws IOException, SQLException {
        String fullSql = addWhereClauseToSqlQuery(sql, opInfo.where);            
        
        PreparedStatement stmt = connection.prepareStatement(fullSql);
        List<Object> params = createQueryParams(lastRunTimestamp, opInfo.paramsInWhere);
        
        logger.debug("sql: {}, params {}", fullSql, params);

        ResultSet result = null;
        
        try {
            bind(stmt, params);

            result = executeQuery(stmt);

            try {
                ValueListener listener = new ColumnValueListener(opInfo.opType)
                        .target(context.riverMouth())
                        .digest(context.digesting());

                merge(result, listener);
            } catch (Exception e) {
                throw new IOException(e);
            }
        } finally {
            close(result);
            close(stmt);
        }

        acknowledge();
    }
    
    private String addWhereClauseToSqlQuery(String sql, String whereClauseToAppend) {
        final String whereKeyword = "where ";
        int whereIndex = sql.toLowerCase().indexOf(whereKeyword);
            
        String fullSql = whereIndex > 0 ? 
                sql.substring(0, whereIndex + whereKeyword.length())+whereClauseToAppend+" AND "+sql.substring(whereIndex+whereKeyword.length()) 
                : sql+" WHERE "+whereClauseToAppend;
          
        return fullSql;
    }

    private List<Object> createQueryParams(Timestamp lastRunTimestamp, int lastRunTimestampParamsCount) {
        List<? extends Object> statementParams = context.pollStatementParams() != null ? context.pollStatementParams() : Collections.emptyList();
        List<Object> params = new ArrayList<Object>(statementParams.size() + lastRunTimestampParamsCount);
        
        for(int i=0; i<lastRunTimestampParamsCount; i++) {
            params.add(lastRunTimestamp);
        }
        
        for(Object param : statementParams) {
            params.add(param);
        }

        return params;
    }
    
    private static class OpInfo {
        final String where;
        final String opType;
        final int paramsInWhere;
        
        public OpInfo(String opType, String where, int paramsInWhere) {
            this.opType = opType;
            this.where = where;
            this.paramsInWhere = paramsInWhere;
        }
        
        public OpInfo(String opType, String where) {
            this(opType, where, 1);
        }
    }
}

@@ -2122,98 +2122,8 @@
 ));%0A
-    %0A    private DateFormat dateFormat = new SimpleDateFormat(%22yyyy-MM-dd HH:mm:ss%22);%0A    
 %0A   
@@ -2893,32 +2893,192 @@
 fo%3E();%0A        %0A
+        String noDeletedWhereClause = context.columnDeletedAt() != null ? %22 AND %22+quoteColumn(context.columnDeletedAt(), quoteString)+%22 IS NULL%22 : %22%22;%0A        %0A
         opInfos.
@@ -3165,32 +3165,53 @@
 eString)+%22 %3E= ?%22
++noDeletedWhereClause
 ));%0A        opIn
@@ -3442,16 +3442,37 @@
 +%22 %3C ?)%22
++noDeletedWhereClause
 , 2));%0A%0A

fix sql