/**
 * Process error queue data that generate during mongo keen and fs error
 * @Author Semusi Technologies Pvt Ltd.
 * @Copyright appICE.io @2017
 */

// define global module
const errorQueueWorker = {},
    AWS = require('aws-sdk'),
    common = require('./common.js'),
    logger = require('../../logger'),
    semusiConfig = require('./../config'),
    mongojs = require('mongojs'),
    async = require('async'),
    winstonLogger = require('./winston'),
    errorHandler = require('./error.handler.js'),
    dataProducer = require('./common.data.producer.js'),
    Redshift = require('node-redshift'),
    redshiftDB = new Redshift(semusiConfig.redshiftClient),
    firehose = require('./firehose.js');

let db;
// establish mongo db connection
if(semusiConfig.IsProduction){
    // connect production mongo db
    db = mongojs(semusiConfig.mongodb.productionUrl, [], { connectionTimeout: (1000*60) });
}
else{
    // connect local mongo db
    db = mongojs(semusiConfig.mongodb.localUrl, [], { connectionTimeout: (1000*60) });
}

// create firehose bucket if does not exists
firehose.createBucketIfItDoesNotExist(semusiConfig.firehostBucket, function(err, bucket){
  if(err){
    logger.error(`create firehose bucket error => ${err}`); 
  }
  else{
    // active firehose data stream
    firehose.waitForDStreamToBecomeActive(semusiConfig.dStreamName, function(err, data){
      if(err){
        logger.error(`create firehose bucket error => ${err}`);
      }  
    });
  }
});

AWS.config.region = semusiConfig.AWS_REGION;
AWS.config.update({accessKeyId:semusiConfig.AWS_ACCESS_KEY_ID,secretAccessKey:semusiConfig.AWS_SECRET_KEY});
const sqs = new AWS.SQS();

// define error queue worker module
(function(module, errorQueueWorker){
    // process begin session data
    errorQueueWorker.processData = (function (data,finalcallback) {

        // start process data with async for each loop
        async.forEach(data, function(element, callback){

            // create collection reference
            let bulk =db.collection(element.col).initializeUnorderedBulkOp();

            // check for which data will process
            if(element.type == 'mongo'){

                // start loop for prepare bulk query
                for(let i=0; i<element.events.length; i++){

                    // check operation
                    if(element.events[i].op == 'update'){

                        // check upsert into query
                        if(element.events[i].isUpsert){

                            // check process data type
                            if(Array.isArray(element.events[i].data) && element.events[i].data.length > 0 ){

                                // get mongo operation key and key data
                                let updateObj = {};
                                updateObj[element.events[i].data.key] = {};

                                // process array type data
                                for(let j=0; j<element.events[i].data.length; j++){

                                    // get key data from object
                                    updateObj[element.events[i].data.key] = element.events[i].data[j][element.events[i].data.keyData];
                                    // prepare bulk query
                                    bulk.find({'_id':element.events[i].data[j].id}).upsert().update(updateObj);
                                }
                            }
                            else{
                                errorQueueWorker.prepareObjectBulkQuery(element, i, bulk);
                            }
                        }
                        else{
                            errorQueueWorker.prepareObjectBulkQuery(element, i, bulk);
                        }
                    }
                    else{
                        // process insert operation data
                        bulk.insert(element.events[i].data);
                    }
                }

                // execute bulk querys
                bulk.execute(function(err, result) {
                    if (err){
                        // send error into callback
                        //callback(err);
                        // dataProducer.writeDataToSQS(element, function(status){
                        //     if(status){
                        //         return callback();
                        //     }
                        //     return;
                        // });
                        logger.error(` error while preocess query => ${err}`);
                        callback(false);
                    }
                    else
                    {
                        // send success into callback
                        logger.info(`${element.col} :: processBeginSessionData sessions RESULT:  ${result}`);
                        callback();
                    }
                });
            }
            else if(element.type == 'keen'){
                // process keen data
                if(Array.isArray(element.events) && element.events.length > 0 ) {
                    try{
                        async.forEach(element.events, function(event, lcallback){
                            let data = event.dataArray;
                            //If events are processed successfully in mongo send a copy to Keen.
                            data.forEach(function (item) {
                                if(item.appid != "583293316940eec96800559f"){
                                    keen.writeEvents(item.appid, item.events, function (result) {
                                        logger.info(`result  :: +> ${result}`);  
                                    });
                                }
                                let record = {data:item.events};
            										record.appid = item.appid;
            										// put record into firehose datastream
            										firehose.putRecord(semusiConfig.dStreamName, record, function(err, res){
            												if(err){
                                                                if(err) logger.error(`put record error => ${err}`);     
                                                            }  

            												 
            										});
                            });
                            lcallback();
                          }, function(err){
                            logger.info("done ingestion into Keen!")
                            logger.error(`${err}`);
                            callback();
                        });
                    }
                    catch(e){
                        logger.error(`data not found for keen => ${e}`);
                        callback();
                    }
                }
                else {
                    logger.info("done ingestion into Keen!")
                    callback();
                }
            }
            else if(element.type == 'redshift'){
                // element.columns = element.columns.replace(/_COMMA_/g, ",");
                // execute redshift command
                redshiftDB.query(element.copyCmd, {raw: true})
                    .then((data) => {
                         
                        callback();
                    })
                    .catch((err) =>{
                        logger.info(`element.copyCmd =>> ${element.copyCmd}`);
                        logger.error(`error ==>> ${err}`);
                        // send error mail to developers
                        err.redshiftError = 'Error while process redshift data';
                        err.copyCmd = element.copyCmd
                        // errorHandler.sendError(err);

                        // resend data into sqs for furthur process
                        // let boxParams = {
                        //     MessageBody: JSON.stringify({copyCmd: element.copyCmd, type:'redshift', columns:element.columns, key:"redshift"}),
                        //     QueueUrl: semusiConfig.apiQueueMap['errors'],
                        //     DelaySeconds: 0
                        // };
                        // sqs.sendMessage(boxParams, function (err, data){
                        //     if (err) {
                         
                        //     }
                        //     callback();
                        // });
                        callback();
                    });
            }
            else{
                callback();
            }
        },function (error) {
            if(error){
                // send error into callback
                logger.error(`error=> ${error}`);
                finalcallback(false);
            }
            else{
                // send success into callback
                finalcallback(true);
            }
            return;
        });
    });

    // prepare query if data has object data type
    errorQueueWorker.prepareObjectBulkQuery = function(element, i, bulk){
        // process object type of data
        if(element.events[i].data._id){
            let match = {"_id":element.events[i].data._id};
            // delete key from operation data
            delete element.events[i].data._id;
            // check competingapps apps id condition
            if(element.events[i].data.competingapps_apps_id){
                match['competingapps.apps.id'] = element.events[i].data.competingapps_apps_id;
                // delete key from operation data
                delete element.events[i].data.competingapps_apps_id;
            }

             
             
            // prepare bulk query
            bulk.find(match).update(element.events[i].data, {safe:true});
        }
        else{
            let find = element.events[i].find;
            // prepare bulk query
            bulk.find(find).update(element.events[i].data, {safe:true});
        }

        return;
    }

    //
    // export errorQueueWorker into module
    module.exports = errorQueueWorker;
}(module, errorQueueWorker)); // self invoking module
