let dataConsumer = {},
	common = require('./common.js'),
	logger = require('../../logger'),
	campaignCommon = require('./campaign.common.js'),
	semusiConfig = require('./../config'),
	mongojs = require('mongojs'),
	cacheApi = require('./semusi.cache.js'),
	botsSession = require('./../bots/bots.session.js'),
	moment = require('moment'),
	async = require('async'),
	winstonLogger = require('./winston'),
	errorHandler = require('./error.handler.js'),
	//dynamodb = require('/datadrive/sap/dynamodbdata/dynamodb.js'),
	postback = require('./../ref_plugins/postback_url.js'),
	dataProducer = require('./common.data.producer.js'),
	psqlUtils = require('./pgsql.utils'),
	firehose = require('./firehose.js');

    let db;
    if(semusiConfig.IsProduction){
    	db = mongojs(semusiConfig.mongodb.productionUrl, []);
    }
    else{
    	db = mongojs(semusiConfig.mongodb.localUrl,  []);
    }

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

(function (dataConsumer) {

	dataConsumer.processBeginSessionData = function(data,finalcallback){
		let errorSQS = false;
		let ausersArray = [], tsessionArray = [], tusessionArray = [];
		async.forEach(data, function(element, callback){
			let dataArray = element.dataArray;
			let collectionis =db.collection('app_users'+element.appid);
			//let users_sessions =db.collection('app_users_sessions').initializeUnorderedBulkOp();
			let bulk = collectionis.initializeUnorderedBulkOp();
			let bulk_timely_sessions = db.collection('timely_sessions').initializeUnorderedBulkOp();
			let bulk_timely_user_sessions = db.collection('timely_users_session').initializeUnorderedBulkOp();
			async.forEach(dataArray, function(datais, inCallback){
			    let session = {st:datais.timestamp};
			    if(datais.context && datais.context.where && datais.context.where.location && datais.context.where.location.place){
			        session.p = datais.context.where.location.place;
			    }

			    // update last seen into app users collection
			    let updatedAt = common.getCurrentEpochTime();
					// push app users data into array
          ausersArray.push({appid: element.appid, op:'update', data:{"_id":datais.user_id, $set:{"ls":datais.timestamp,"updatedAt":updatedAt},$inc:{'sc':1}}});
				  // prepare app users query
					bulk.find({"_id":datais.user_id}).update({$set:{"ls":datais.timestamp,"updatedAt":updatedAt},$inc:{'sc':1}});

			    // Manage session optimise collection and
			    let timelySession  = {};
			    let platform = '';
			    // validate session event property
			    if(datais.timestamp && datais.context && datais.context.where
			        && datais.context.where.location && datais.context.where.location.place &&  datais.context.who
			        && datais.context.who.device && datais.context.who.device.p){
			        // variable initialization
			        let timelySession  = {}, time =common.getDayTime(datais.timestamp);
			        let place = datais.context.where.location.place;
			        platform = datais.context.who.device.p.toLowerCase();
			        // populate data into object
			        common.fillTimeObjectWithEpoch(platform, datais.timestamp, timelySession, time+"."+place);
			        common.fillTimeObjectWithEpoch(platform, datais.timestamp, timelySession, time+".total");
			        common.fillTimeObjectWithEpoch(platform, datais.timestamp, timelySession, "total");

							// manage MAU, WAU and DAU
              let isUnique = false;
              async.parallel([
              	function(callback){
              		// manage MAU key into redis server
                      let mKey = element.appid+moment.unix(datais.timestamp).format('YYYYM');
                       
                      cacheApi.addToSadd(mKey, datais.user_id, function(data){
                          if(data){
                                  let mauKey = common.getDauMauKey(platform, datais.timestamp, 'mau');
                                  timelySession[mauKey.m] = 1;
                                  timelySession[mauKey.d] = 1;
                                  cacheApi.expire(mKey,60*60*24*30); // set expaire key time in seconds for one day
                          }
                          callback();
                      });
              	},
              	function(callback){
              		// manage WAU key into redis server
                      let wKey = element.appid+moment.unix(datais.timestamp).format('YYYYW');
                       
                      cacheApi.addToSadd(wKey, datais.user_id, function(data){
                          if(data){
                                  let wauKey = common.getDauMauKey(platform, datais.timestamp, 'wau');
                                  timelySession[wauKey] = 1;
                                  cacheApi.expire(wKey,60*60*24*7); // set expaire key time in seconds for one day
                          }
                          callback();
                      });
              	},
              	function(callback){
                      // manage DAU key into redis server
                      let key = element.appid+moment.unix(datais.timestamp).format('YYYYMD');
                      cacheApi.addToSadd(key, datais.user_id, function(data){
                           
                          if(data){
                              isUnique = true
                              let dauKey = common.getDauMauKey(platform, datais.timestamp, 'dau');
                              timelySession[dauKey] = 1;
                              cacheApi.expire(key,60*60*24); // set expaire key time in seconds for one day
                               
                          }
                          callback();
                      });
              	}
              ],function(error){
              		 
									// process session into optimize or aggregate collection
									tsessionArray.push({appid:element.appid, op:'update', isUpsert:true, data:{'_id':db.ObjectId(element.appid),'$inc': timelySession}});
					 				// prepare timely session query
                  // process session into optimize or aggregate collection
									bulk_timely_sessions.find({'_id':db.ObjectId(element.appid)}).upsert().update({'$inc': timelySession});

									// check active user sc boundary
                  datais.appid = element.appid
									common.checkActiveUserSc(datais, 1, function(err, stdata){
											if(stdata){
												logger.info(`stdata => ${stdata}`);
													// add session optimise data into array
                          tusessionArray.push({appid:element.appid, op:'update', isUpsert:true, data:stdata, key:'$inc', keyData:'updateSessionUserObj'});
                          for(let j=0; j<stdata.length; j++){
														bulk_timely_user_sessions.find({'_id':stdata[j].id}).upsert().update({'$inc': stdata[j].updateSessionUserObj});
													}
											}
											inCallback();
									});
              });
					}
			    else{
			        inCallback();
			    }
			}, function(err){
			    async.parallel([
			        //app-users update
			        function(callback){
			            bulk.execute(function(err, result) {
			                if (err){
												  // send data into queue if mongo got error
												  errorQueue(ausersArray, 'mongo', 'app_users'+element.appid, 'processBeginSessionData', function(status){
															if(status){
																	errorSQS = true;
															}
															err.processBeginSessionData = 'processBeginSessionData app_users';
					                    errorHandler.sendError(err);
					                    logger.info("Error:processBeginSessionData:app_users");
										logger.error(`error=> ${err}`);
					                    callback(err);
													});
			                }else
			                    {
			                     
			                    callback();
			                }
			            });
			        },
			        //timely sessions update
			        function(callback){
			            bulk_timely_sessions.execute(function(err, result) {
			                if (err){
													// send data into queue if mongo got error
													errorQueue(tsessionArray, 'mongo', 'timely_sessions', 'processBeginSessionData', function(status){
															if(status){
																	errorSQS = true;
															}
															err.processBeginSessionData = 'processBeginSessionData timely session';
					                    errorHandler.sendError(err);
										logger.info("Error:processBeginSessionData:sessions");
					                    logger.error(`error=> ${err}`);
					                    callback(err);
													});
			                }else
			                    {
			                     
			                    callback();
			                }
			            });
			        },
			        function(callback){
			            try{
			                bulk_timely_user_sessions.execute(function(err, result){
			                    if (err){
								  // send data into queue if mongo got error
									errorQueue(tusessionArray, 'mongo', 'timely_users_session', 'processBeginSessionData', function(status){
										if(status){
												errorSQS = true;
										}
										err.processBeginSessionData = 'processBeginSessionData timely user session';
				                        errorHandler.sendError(err);
				                        logger.info("Error:processBeginSessionData:bulk_timely_user_sessions");
				                        logger.error(`error=> ${err}`);
				                        callback(err);
									});
			                    }else
			                        {
			                         
			                        callback();
			                    }
			                });
			            }
			            catch(e){
			                e.processBeginSessionData = 'processBeginSessionData timely user session catch';
			                errorHandler.sendError(e);
							logger.error(`error=> ${e}`);
			                callback();
			            }
			        }
			    ],function(err){
			        if(err){
			            err.processBeginSessionData = 'Error:processBeginSessionData:async:parallel';
			            errorHandler.sendError(err);
			             logger.error(`Error:processBeginSessionData:async:parallel:: ${err}`);
			            callback(err);
			        }
			        else{
			            callback();
			        }

			    });
			});
		},function(error){
			 // empty array
			 ausersArray = [];
			 tsessionArray = [];
			 tusessionArray = [];

			if (error){
				error.processBeginSessionData = 'Error:processBeginSessionData';
				errorHandler.sendError(error);
				logger.error(`Error:processBeginSessionData => ${error}`);
				if(errorSQS){
						finalcallback(true);
				}else{
						finalcallback(false);
				}
			}else{
				finalcallback(true);
			}
		});
	};

	dataConsumer.processBeginChatBotSessionData = function(data,finalcallback){
		async.forEach(data, function(element, callback){
			let dataArray = element.dataArray;
			for (let i=0;i<dataArray.length;i++){
				botsSession.checkNewSession(dataArray[i].params, dataArray[i].event);
			}
			callback();

		},function(error){
			if (error){
				error.processBeginSessionData = 'Error:processBeginChatBotSessionData';
				errorHandler.sendError(error);
				logger.error(`Error:processBeginChatBotSessionData => ${error}`);
				finalcallback(false);
			}else{
				finalcallback(true);
			}
		});
	};

	dataConsumer.processEndSessionData = function(data,finalcallback){
		let errorSQS = false;
		let ausersArray = [], tusessionArray = [];
		async.forEach(data, function(element, callback){
			let dataArray = element.dataArray;
			let collectionis =db.collection('app_users'+element.appid)
			let bulk = collectionis.initializeUnorderedBulkOp();
			let bulk_timely_user_sessions = db.collection('timely_users_session').initializeUnorderedBulkOp();

			async.forEach(dataArray, function(datais, inCallback){
					    // update tsd and lsd in app user collection
					    let updatedAt = common.getCurrentEpochTime();
							// push app users data into array
              ausersArray.push({appid:element.appid, op:'update', data:{"_id":datais.user_id, $inc:{'tsd':datais.duration},$set:{'lsd':datais.duration, 'ls_st':parseInt(datais.timestamp-datais.duration), 'updatedAt':updatedAt}}})
							// prepare bulk query
					    bulk.find({"_id":datais.user_id}).update({$inc:{'tsd':datais.duration},$set:{'lsd':datais.duration, 'ls_st':parseInt(datais.timestamp-datais.duration), 'updatedAt':updatedAt}});

					    // check active user Asl boundary
					    common.checkActiveUserAsl(datais, function(err, stdata){
					        if(stdata){
											// push end session optimisation query into array
                    	tusessionArray.push({appid:element.appid, op:'update', isUpsert:true, data:stdata, key:'$inc', keyData:'updateSessionUserObj'})
                      for(let j=0; j<stdata.length; j++){
					                bulk_timely_user_sessions.find({'_id':stdata[j].id}).upsert().update({'$inc': stdata[j].updateSessionUserObj});
					            }
					        }
					        inCallback();
					    });

					}, function(err){
					    async.parallel([
					        //app-users update
					        function(callback){
					            bulk.execute(function(err, result) {
					                if (err){
															// send data into queue if mongo got error
															errorQueue(ausersArray, 'mongo', 'app_users'+element.appid, 'processEndSessionData', function(status){
																	if(status){
																			errorSQS = true;
																	}
																	err.processEndSessionData = 'Error:processEndSessionData:app_users';
							                    errorHandler.sendError(err);
												logger.error(`Error:processEndSessionData:app_users ${err}`);
							                    callback(err);
															});
					                }else
					                    {
					                     
					                    callback();
					                }
					            });
					        },
					        function(callback){
					            try{
					                bulk_timely_user_sessions.execute(function(err, result){
					                    if (err){
																	// send data into queue if mongo got error
																	errorQueue(tusessionArray, 'mongo', 'timely_users_session', 'processEndSessionData', function(status){
																			if(status){
																					errorSQS = true;
																			}
																			err.processEndSessionData = 'Error:processBeginSessionData:bulk_timely_user_sessions';
							                        errorHandler.sendError(err);
													logger.error(`Error:processBeginSessionData:bulk_timely_user_sessions:: ${err}`);
							                        callback(err);
																	});
					                    }else
					                        {
					                         
					                        callback();
					                    }
					                });
					            }
					            catch(e){
					                e.processEndSessionData = 'Error:processBeginSessionData:bulk_timely_user_sessions catch';
					                errorHandler.sendError(e);
									logger.error(`error=> ${e}`);
					                callback();
					            }
					        }
					    ],function(err){
					        if(err){
					            err.processEndSessionData = 'Error:processEndSessionData:async:parallel';
					            errorHandler.sendError(err);
								logger.error(`Error:processEndSessionData:async:parallel ::${err}`);
					            callback(err);
					        }
					        else{
					            callback();
					        }
					    });
					});

	    },function(error){
					// empty array
					ausersArray = [];
					tusessionArray = [];
					if (error){
						error.processEndSessionData = 'Error:processEndSessionData';
						errorHandler.sendError(error);
						logger.error(`Error:processEndSessionData:: ${error}`);
						if(errorSQS){
								finalcallback(true);
						}else{
								finalcallback(false);
						}
					}else{
						 
						finalcallback(true);
					}
		});
	};

	dataConsumer.processEvents = function(data,finalcallback){
		let errorSQS = false;
		let ausersArray = [], events_Array = [], events_statsArray = [], events_stats_TransArray = [],
		timely_eventsArray = [], aggregates_events_Array = [];

		async.forEach(data, function(element, callback){
			// read next element if appid is null
			if(element.appid == null){
					callback();
			}

			let dataArray = element.dataArray;
			let eventAttributes = [];
			let eventCityAttributes = [];
			let postbackArr = [];
			let counts = 0;
			let eventCollection =db.collection('events_'+element.appid);
			let appuserCollection =db.collection('app_users'+element.appid);
			let bulk_events_stats =db.collection('events_stats').initializeUnorderedBulkOp();
			let bulk_events_stats_transaction =db.collection('events_stats_transaction').initializeUnorderedBulkOp();
			let bulk_timely_events =db.collection('timely_events').initializeUnorderedBulkOp();
			let bulk_timely_user_events =db.collection('aggregates_events_'+element.appid).initializeUnorderedBulkOp();
			let bulk_insert = eventCollection.initializeUnorderedBulkOp();
			let bulk_update = appuserCollection.initializeUnorderedBulkOp();
			// sqs event list loop
			async.forEach(dataArray, function(datais, iCallback){
				// events loop
				async.forEach(datais.events, function(eventData, jCallback){
					// replace space and forward slace from event name
					eventData.key = eventData.key.replace(/(?:\s|\ |\/)/g, '_');

					//eventData.key =  escape(eventData.key);
					// delete unrequired properties from segment
					if(eventData.segment){
						if(eventData.segment.updatedAt)
							delete eventData.segment.updatedAt;
						if(eventData.segment.createdAt)
							delete eventData.segment.createdAt;
						if(eventData.segment.did && eventData.segment.did == '1111')
							delete eventData.segment.did;
					}

					// validate live or search event
					campaignCommon.checkLiveEventCampaign({key: eventData.key, appid: element.appid, did: eventData.did, etime: (eventData.context && eventData.context.when && eventData.context.when.timestamp)? eventData.context.when.timestamp : eventData.mtimestamp/1000 });

					if(semusiConfig.validateEventsApps.indexOf(element.appid) >= 0){
						 
						let geoReserve = {lat:'', lon:''}, ips = '', radio = {}, cells = [], wifi = [];
						// set lat and log
						if(eventData.context && eventData.context.where && eventData.context.where.gpsloc){
							geoReserve.lat = eventData.context.where.gpsloc.Lat
							geoReserve.lon = eventData.context.where.gpsloc.Lng
						}
						// set network and wifi details
						if(eventData.context && eventData.context.who && eventData.context.who.network){
							// network detail
							radio.type = semusiConfig.contextNetPType[eventData.context.who.network.ptype];
							radio.mnc = eventData.context.who.network.mnc;
							radio.mcc = eventData.context.who.network.mcc;
							// cell detail
							cells.push({lac:eventData.context.who.network.lac, cid:eventData.context.who.network.cid});
							// wifi detail
							wifi.push({bssid:eventData.context.who.network.bssid});
						}
						// set ip address
						if(eventData.context && eventData.context.where && eventData.context.where.ip){
							ips = eventData.context.where.ip;
						}

					}
					else{
							// check clickid for older version
							if('57f7a1a9662688ff05897eb1' == element.appid
								&&  ((eventData.segment
									&& eventData.segment.Reg_Step == 'Completed'
									&& (eventData.key =="New_Registration" || eventData.key == "Wallet Registration"))
								|| (eventData.key == 'UPI_Registration' && eventData.segment && eventData.segment.Step_Completed == 'Success'))) {

								// CHECK REFNAME IN API CALL
								if(eventData.context && eventData.context.who && ( eventData.context.who.refname == 'taskbucks' || eventData.context.who.refname == 'rationalheads' )) {
									postbackArr.push({
													key:eventData.context.who.refname,
													event:eventData.key,
													segment:(eventData.segment)? eventData.segment:{},
													did:eventData.did,
													user_id:eventData.user_id,
													eTime:{
															etime:moment(parseInt(eventData.eventTime)*1000).format("YYYY-M-D h:mm:ss a"),
															ptime:moment().format("YYYY-M-D h:mm:ss a")
														},
													eventTime:eventData.eventTime
												});
								}
							}

							// send postbacks with specific events
              semusiConfig.postbackEventsApps.forEach(function(postBackApp){
                  // valide appid and event
									//eventData.key = eventData.key.toString();

									if( postBackApp.appid == element.appid
                      && ( eventData.key && postBackApp.events.indexOf(eventData.key.toLowerCase()) >=0 )
                      && ( eventData.context && eventData.context.who && postBackApp.refferers.indexOf(eventData.context.who.refname.toLowerCase()) >=0 )
                  ){
                      postbackArr.push({
                          key:eventData.context.who.refname,
                          event:eventData.key,
                          segment:(eventData.segment)? eventData.segment:{},
                          did:eventData.did,
                          user_id:eventData.user_id,
                          eTime:{
                              etime:moment(parseInt(eventData.eventTime)*1000).format("YYYY-M-D h:mm:ss a"),
                              ptime:moment().format("YYYY-M-D h:mm:ss a")
                          },
                          eventTime:eventData.eventTime
                      });
                  }
              });

						/*=================================================================*/
						// Manage campaign events into optimise collection
						if(eventData.key == 'Campaign_Received'
							|| eventData.key == 'Campaign_Viewed'
							|| eventData.key == 'Campaign_Clicked'
							|| eventData.key == 'Campaign_Deleted'){

							if(eventData.context.where.geo == undefined){
								eventData.context.where.geo = {cty:'NA', cc:'NA'};
							}
							else{
								if(eventData.context.where.geo.cc == undefined){
									eventData.context.where.geo.cc = 'NA';
								}
								if(eventData.context.where.geo.cty == undefined){
									eventData.context.where.geo.cty = 'NA';
								}
							}

							if(eventData.context && eventData.context.who
								&& eventData.context.who.device && eventData.context.who.device.p
								&& eventData.context.who.device.d && eventData.context.where
								&& eventData.context.where.geo && eventData.context.where.geo.cty
								&& eventData.context.where.geo.cc && eventData.context.where.location
								&& eventData.context.where.location.place){
								let platform = eventData.context.who.device.p.toLowerCase();
								let device = eventData.context.who.device.d;
								device = device.replace(/\./g, "_");
								let country = eventData.context.where.geo.cc;
								let city = eventData.context.where.geo.cty;
								let place = eventData.context.where.location.place;

								let events_stats_obj = {};
								//events_stats_obj[platform+"."+eventData.key] = 1;
								common.fillCampaignEventObj(events_stats_obj, platform, eventData.eventTime, device, country, city, place, eventData.key);

								if(eventData.camp_type == "transactional"){
										// push events stats into array
										events_stats_TransArray.push({appid:element.appid+eventData.campid, op:'update', isUpsert:true, data:{'_id':element.appid+eventData.campid, '$inc': events_stats_obj}});
										// prepare bulk query transaction
										bulk_events_stats_transaction.find({'_id':element.appid+eventData.campid}).upsert().update({'$inc': events_stats_obj});
								}
								else{
										// push events stats into array
										events_statsArray.push({appid:element.appid+eventData.campid, op:'update', isUpsert:true, data:{'_id':element.appid+eventData.campid, '$inc': events_stats_obj}});
										// prepare bulk query
										bulk_events_stats.find({'_id':element.appid+eventData.campid}).upsert().update({'$inc': events_stats_obj});
								}

							}
						}
						/*=============================================================================================================================================================*/

						eventData['createdAt'] = common.getCurrentEpochTime();
						//createCallBackLogFile(element.appid, eventData, 'events_fslog');
						//bulk_insert.insert(eventData);
						//events_Array.push({appid:element.appid, data:eventData});
            // put events into log
						winstonLogger.init('events_'+element.appid+"_"+eventData.key, true);
						winstonLogger.writeData(eventData);

						if(eventData.key && eventData.eventTime &&
							eventData.context && eventData.context.who
							&& eventData.context.who.device && eventData.context.who.device.p )
						{
							let platform = eventData.context.who.device.p.toLowerCase();
		    				let epoch = (eventData.eventTime*1000) // convert epoch from gmt
						    let id = common.timelyEventId(element.appid, epoch);
						    let updateEvents = {};
						    common.fillEventObjectWithEpoch(epoch, updateEvents, platform, eventData.key);
						    common.fillEventObjectWithEpoch(epoch, updateEvents, platform, "total");
								// push timely event into array
							  timely_eventsArray.push({appid:element.appid, op:'update', isUpsert:true, data:{'_id':id, '$inc': updateEvents}})
							  // prepare bulk query
								bulk_timely_events.find({'_id':id}).upsert().update({'$inc': updateEvents});

							// optimise user basis new event collection
						    let userEvents = {};
						  	common.fillNewEventObject(userEvents,epoch,eventData.key);
								// push timely events into array
								aggregates_events_Array.push({appid:element.appid, op:'update', isUpsert:true, data:{'_id':datais.did, '$inc': userEvents, '$set':{'updatedAt':common.getCurrentEpochTime()}}});
								// prepare bulk query
								bulk_timely_user_events.find({'_id':datais.did}).upsert().update({'$inc': userEvents, '$set':{'updatedAt':common.getCurrentEpochTime()}});

							//Maintain aggregate collection of attribute values.
							if(eventData.key.toLowerCase() != '_app_crash' && eventData.segment){
								let keys = Object.keys(eventData.segment);
								keys.forEach(function(key){
									let attValue = (typeof(eventData.segment[key])=="string") ? eventData.segment[key].replace(/\./g, ":") : "";
									if(attValue!=""){
										let eventName = eventData.key;
										eventAttributes.push(createEventSegment(eventName, key, epoch, platform, attValue));
									}
								});
							}

							let tmp_obj = {k:eventData.key, t:eventData.eventTime, p:moment().valueOf()}
							// set event attributes or segments
							if(eventData.segment){
								tmp_obj.segment = eventData.segment;
							}
							if(datais.user_id){
								// store last 10 events
								updateLastEvents(element.appid, eventData.did, datais.user_id, tmp_obj);
							}

						}

						 
						// iterate async foreach
						jCallback();
					}
				}, function(error){
					// send postback webhook
					if(datais.postback){
						postback.webHook(element.appid, datais.postback.url, datais.postback.data);
					}

					// update last event into app user collection
					let updatedAt = common.getCurrentEpochTime();
					let currentEvent = datais.events[datais.events.length-1];
					// push single event into array
          ausersArray.push(
          	{
							appid:element.appid,
							op:'update',
							data:{
								"_id":datais.user_id,
								"$set":{
										"le":{
											k:currentEvent.key,
											t:currentEvent.eventTime,
											s:currentEvent.segment,
											c:currentEvent.context
										},
										'updatedAt':updatedAt
								}
							}
            }
					);

					bulk_update.find({"_id":datais.user_id}).update({"$set":{"le":{k:currentEvent.key,t:currentEvent.eventTime,s:currentEvent.segment,c:currentEvent.context},'updatedAt':updatedAt}},{safe:true});
					//Update event segment list.
					processEventSegments(datais.events,element.appid);
					processEventAttributeAggregation(eventAttributes, 'timely_eventAttributes', element.appid);

					iCallback();
				});
			}, function(err){
				async.parallel([
					//bulk_insert exceution
					function(callback){
						try{
							bulk_insert.execute(function(err, result) {
								if (err){
									// data send to error queue
		              errorQueue(events_Array, 'mongo', 'events_'+element.appid, 'processEvents', function(status){
											if(status){
													errorSQS = true;
											}
											err.processEvents = 'Error:processEvents:events_appid';
											errorHandler.sendError(err);
											logger.error(`Error:processEvents:events_appid ${err}`);
											callback(err);
									});
								}else
									{
						   	    	 
						   	    	callback();
								}
			    			});
						}
						catch(e){

						}

					},
					//bulk_update execution
					function(callback){
						bulk_update.execute(function(err, result) {
							if (err){
								// data send to error queue
                errorQueue(ausersArray, 'mongo', 'app_users'+element.appid, 'processEvents', function(status){
										if(status){
												errorSQS = true;
										}
										err.processEvents = 'Error:processEvents:app_users';
										errorHandler.sendError(err);
										logger.error(`Error:processEvents:app_users=>>>> ${err}`);
										callback(err);
								});
							}else
								{
					   	    	 
					   	    	callback();
							}
				    	});
					},
					//bulk_update timely event collection execution
					function(callback){
						bulk_timely_events.execute(function(err, result) {
							if (err){
								// data send to error queue
                errorQueue(timely_eventsArray, 'mongo', 'timely_events', 'processEvents', function(status){
										if(status){
												errorSQS = true;
										}
										err.processEvents = 'Error:processEvents:timely_event_collection';
										errorHandler.sendError(err);
										logger.error(`Error:processEvents:timely_event_collection =>>>> ${err}`);
										callback(err);
								});
							}else
								{
					   	    	 
					   	    	callback();
							}
				    	});
					},
					//bulk_update new timely event collection execution
					function(callback){
						bulk_timely_user_events.execute(function(err, result) {
							if (err){
								// data send to error queue
                errorQueue(aggregates_events_Array, 'mongo', 'aggregates_events_'+element.appid, 'processEvents', function(status){
										if(status){
												errorSQS = true;
										}
										err.processEvents = 'Error:processEvents:timely_new_event_collection';
										errorHandler.sendError(err);
										logger.error(`Error:processEvents:timely_new_event_collection =>>>> ${err}`);
										callback(err);
								});
							}else
								{
					   	    	 
					   	    	callback();
							}
				    	});
					},
					//bulk campaign events stats
					function(callback){
						try{
							bulk_events_stats.execute(function(err, result) {
								if (err){
									// data send to error queue
                  errorQueue(events_Array, 'mongo', 'events_stats', 'processEvents', function(status){
											if(status){
													errorSQS = true;
											}
											err.processEvents = 'Error:processEvents:bulk_events_stats';
											errorHandler.sendError(err);
											logger.error(`Error:processEvents:bulk_events_statsn =>>>> ${err}`);
											callback(err);
									});
								}else
									{
						   	    	 
						   	    	callback();
								}
					    	});
						}
						catch(e){
							logger.error(`exception:bulk_events_stats =>>>> ${e}`);
							callback();
						}
					},
					//bulk campaign events stats
					function(callback){
						try{
							bulk_events_stats_transaction.execute(function(err, result) {
								if (err){
                      // data send to error queue
                      errorQueue(events_stats_TransArray, 'mongo', 'events_stats_transaction', 'processEvents', function(status){
                      	if(status){
											errorSQS = true;
										}
										err.processEvents = 'Error:processEvents:events_stats_transaction';
										errorHandler.sendError(err);
										logger.error(`Error:processEvents:events_stats_transaction=>>>> ${err}`);
										callback(err);
                                    });
								}else
									{
						   	    	 
						   	    	callback();
								}
					    	});
						}
						catch(e){
							logger.error(`exception:events_stats_transaction =>>>> ${e}`);
							callback();
						}
					},
					//send event from postback url
					function(callback){
						if(semusiConfig.sendPostbackApps.indexOf(element.appid) >= 0){
							if(postbackArr){
								async.forEach(postbackArr, function(postData, loopCallback){
									cacheApi.getKey("app_id"+element.appid+postData.did,function(err,result){
		        						if(err || result==null){
											// create log file or get file object
											winstonLogger.init('postback_'+element.appid, true);
											// check clickid in to database
	        								common.db.collection('app_users'+element.appid).findOne({_id:postData.user_id},{"history":1},function(err, userData){
	        									if(err){
															postData.clickid = '';
															postData.erro = 'error while get click id from db';
															winstonLogger.writeData({erro:"error while get click id from db", did:postData.did, eventname:postData.event, etimes:JSON.stringify(postData.eTime), ref:postData.key});
														}
	        									else{
	        										if(userData){
	        											if(userData.history.length > 0 && userData.history[userData.history.length-1].clickId){
		        											postData.clickid = userData.history[userData.history.length-1].clickId;
		        											postData.msg = 'get from db';
		        											postback.sendEvents(element.appid, {clickid:postData.clickid, eventname:postData.event, etimes:postData.eTime, ref:postData.key, did:postData.did, segment:postData.segment});
		        										}
		        										else{
		        											postData.clickid = '';
		        											postData.erro = 'clickid not found in history';
																	winstonLogger.writeData({erro:"clickid not found in history", did:postData.did, eventname:postData.event, etimes:JSON.stringify(postData.eTime), ref:postData.key});
		        										}
	        										}
	        										else{
	        											postData.clickid = '';
		        										postData.erro = 'record not found in collection';
																winstonLogger.writeData({erro:"record not found in collection", did:postData.did, eventname:postData.event, etimes:JSON.stringify(postData.eTime), ref:postData.key});
	        										}
	        									}
	        									loopCallback();
		        							});
		        						}
		        						else{
		        							postData.clickid = result;
		        							postData.msg = 'get from cache';
		        							postback.sendEvents(element.appid, {clickid:result, eventname:postData.event, etimes:postData.eTime, ref:postData.key, did:postData.did, segment:postData.segment});
		        							loopCallback();
		        						}
		        					});

									// write post into keen
									keen.writeEvents(element.appid,postbackArr,function(result){
											 
									});
								}, function(err){
									callback();
								});
	                    	}
	                    	else{
	                    		callback();
	                    	}
						}
						else{
							callback();
						}
					}
				],function(err){
					if(err){
						err.processEvents = 'Error:processEvents:async:parallel';
						errorHandler.sendError(err);
						logger.error(`Error:processEvents:async:parallel =>>>> ${err}`);
						callback(err);
					}
					else{
						callback();
					}

				});
			});
	    },function(error){
				// empty arrays
				ausersArray = []; events_Array = []; events_statsArray = [];
        timely_eventsArray = []; aggregates_events_Array=[];

			if (error){
					error.processEvents = 'Error:processEvents';
					errorHandler.sendError(error);
					logger.error(`Error:processEvents =>>>> ${error}`);
					// push data into error queue
					errorQueue(data, 'keen', '', 'processEvents', function(status){
							if(status){
									finalcallback(true);
							}else{
									finalcallback(false);
							}
					});
			}else{
				//If events are processed successfully in mongo send a copy to Keen.
				data.forEach(function(item){
						item.dataArray.forEach(function(evt){
								// stop pushing epic cricket event into keen
								if(semusiConfig.keenApps.indexOf(item.appid) >= 0){
										keen.writeEvents(item.appid,evt.events,function(result){
												 
										});
								}

								//if('57e8f2d84e25c0b702996cb0' == item.appid){
										// send event into firehose data stream
										let record = {data:evt.events};
										record.appid = item.appid;
										// put record into firehose datastream
										// firehose.putRecord(semusiConfig.dStreamName, record, function(err, res){
										 
										//
										 
										// });
								//}
						});
				});
				 
				finalcallback(true);
			}
		});
	};

	dataConsumer.processCompetingApps = function(data,finalcallback){
		let errorSQS = false;
		let auserArray = [], tusessionArray = [];
		async.forEach(data, function(element, callback){
			let dataArray = element.dataArray;

			let bulk_timely_user_sessions = db.collection('timely_users_session').initializeUnorderedBulkOp();
			let bulk = db.collection('app_users'+element.appid).initializeUnorderedBulkOp();
			for (let i=0;i<dataArray.length;i++){
				let datais = dataArray[i];
				let collectionis =db.collection('app_users'+element.appid);
				if(datais.cKey == "CSession_Start"){
					// dump app users into array
          auserArray.push({appid:element.appid, op:'update', data:{_id:datais.user_id,"competingapps_apps_id":datais.capp_id,$set:{"competingapps.apps.$.ls":datais.timestamp},$inc:{'competingapps.apps.$.sc':1} }});
					// prepare app users query
					bulk.find({_id:datais.user_id,"competingapps.apps.id":datais.capp_id}).update({$set:{"competingapps.apps.$.ls":datais.timestamp},$inc:{'competingapps.apps.$.sc':1}});
				}
				if(datais.cKey == "CSession_End"){
					updateCompetingApps(element.appid,datais);
					//let tsd = datais.duration;
					//bulk.find({_id:datais.user_id,"competingapps.apps.id":datais.capp_id}).update({$set:{"competingapps.apps.$.lsd":datais.duration,'competingapps.apps.$.ls_st':datais.timestamp},$inc:{'competingapps.apps.$.tsd':tsd}});
				}
				if(datais.timelyUsersSession){
					// push session into array
          tusessionArray.push({appid:element.appid, op:'update', isUpsert:true, data:datais.timelyUsersSession, key:'$inc', keyData:'updateSessionUserObj', flag:true});

					for(let j=0; j<datais.timelyUsersSession.length; j++){
						bulk_timely_user_sessions.find({'_id':datais.timelyUsersSession[j].id}).upsert().update({'$inc': datais.timelyUsersSession[j].updateSessionUserObj},true);
					}
				}
				if(datais.endtimelyUsersSession){
					// push session into array
          tusessionArray.push({appid:element.appid, op:'update', isUpsert:true, data:datais.endtimelyUsersSession, key:'$inc', keyData:'updateSessionUserObj'});

					for(let j=0; j<datais.endtimelyUsersSession.length; j++){
						bulk_timely_user_sessions.find({'_id':datais.endtimelyUsersSession[j].id}).upsert().update({'$inc': datais.endtimelyUsersSession[j].updateSessionUserObj});
					}
				}

			}
			async.parallel([
				function(callback){
					try{
						bulk_timely_user_sessions.execute(function(err, result){
							if (err){
								// data send to error queue
                errorQueue(tusessionArray, 'mongo', 'timely_users_session', 'processCompetingApps', function(status){
										if(status){
												errorSQS = true;
										}
										err.processBeginSessionData = 'processCompetingApps timely user session';
										errorHandler.sendError(err);
										logger.error(`Error:processCompetingApps:bulk_timely_user_sessions =>>>> ${err}`);
										callback(err);
								});
							}else
								{
					   	    	 
					   	    	callback();
							}
				    	});
					}
					catch(e){
						e.processCompetingApps = 'processCompetingApps timely user session catch';
						errorHandler.sendError(e);
						logger.error(`Error: = >>>> ${e}`);
						callback();
					}
				},
				function(callback){
					bulk.execute(function(err, result) {
						if (err){
							// data send to error queue
              errorQueue(auserArray, 'mongo', 'app_users'+element.appid, 'processCompetingApps', function(status){
									if(status){
											errorSQS = true;
									}
									err.processCompetingApps = 'Error:processCompetingApps:app_users';
									errorHandler.sendError(err);
									logger.error(`Error:processCompetingApps:app_users = >>>> ${err}`);
									callback(err);
							});
						}else
							{
				   	    	 
				   	    	callback();
						}
			    	});
				}
			],function(err){
				if(err){
					err.processCompetingApps = 'Error:processCompetingApps:async:parallel';
					errorHandler.sendError(err);
					logger.error(`Error:processCompetingApps:async:parallel: = >>>> ${err}`);
					callback(err);
				}
				else{
					callback();
				}

			});

		},function(error){
			tusessionArray = []; auserArray = [];
			if (error){
				error.processCompetingApps = 'Error:processCompetingApps';
				errorHandler.sendError(error);
				logger.error(`Error:processCompetingApps = >>>> ${error}`);
				if(errorSQS){
						finalcallback(true);
				}else{
						finalcallback(false);
				}

			}else{
					finalcallback(true);
			}
		});
	};

	dataConsumer.processDeviceUpdates = function(data,finalcallback){
		let errorSQS = false;
		let auserArray = [];
		async.forEach(data, function(element, callback){
			let dataArray = element.dataArray;
			let collectionis =db.collection('app_users'+element.appid)
			let bulk = collectionis.initializeUnorderedBulkOp();
			for (let i=0;i<dataArray.length;i++){
				let datais = dataArray[i];
				if (datais.user){
					let params = {app_id:datais.appid, user_id:datais.user_id, av:datais.user.av};
					if(datais.user.pv){
						params['pv'] = datais.user.pv;
					}
					updateTimelyDashbaord(params);
					datais.user['updatedAt'] = common.getCurrentEpochTime();
					// push user data into array
          auserArray.push({appid:element.appid, op:'update', data:{"_id":datais.user_id, $set:datais.user}})
					// prepare update user data query
					bulk.find({"_id":datais.user_id}).update({$set:datais.user});
					updateCustomFieldCollection(datais);
					//updateGCMIdToDynamoDB(datais.user,element.appid);
				}

				if(datais.user_info){
					let params = {app_id:datais.appid, user_id:datais.user_id, user_info:datais.user_info};
					updateUserInfo(params);
				}
				if(datais.user.hb){
					let params = {app_id:datais.appid, user_id:datais.user_id, hb:datais.user.hb};
					updateUserHb(params);
				}

				if (datais.source){
					updateReferal(datais);
				}
			}
			bulk.execute(function(err, result) {
					if (err){
						// data send to error queue
            errorQueue(auserArray, 'mongo', 'app_users'+element.appid, 'processDeviceUpdates', function(status){
								if(status){
										errorSQS = true;
								}
								err.processDeviceUpdates = 'Error:processDeviceUpdates:app_users';
								errorHandler.sendError(err);
								logger.error(`Error:processDeviceUpdates:app_users = >>>> ${err}`);
								callback(err);
						});
					}else{
			   	    	 
			   	    	callback();
					}
		    	});
		},function(error){
				auserArray = [];
				if (error){
						error.processDeviceUpdates = 'Error:processDeviceUpdates';
						errorHandler.sendError(error);
						logger.error(`Error:processDeviceUpdates >>>> ${error}`);
						if(errorSQS){
								finalcallback(true);
						}else{
								finalcallback(false);
						}
				}else{
					finalcallback(true);
				}
			});
	};

	let updateCompetingApps = function(appid,datais){
		datais.duration = (datais.duration)? datais.duration : 0;
		common.db.collection('app_users'+appid).update({_id:datais.user_id,"competingapps.apps.id":datais.capp_id},{$set:{"competingapps.apps.$.lsd":datais.duration,'competingapps.apps.$.ls_st':datais.timestamp},$inc:{'competingapps.apps.$.tsd':datais.duration}}, false,function(err,updated){
			if(!err){
				 
				// data send to error queue
        errorQueue([{appid:appid, op:'update', data:{_id:datais.user_id,"competingapps_apps_id":datais.capp_id,$set:{"competingapps.apps.$.lsd":datais.duration,'competingapps.apps.$.ls_st':datais.timestamp},$inc:{'competingapps.apps.$.tsd':datais.duration}}, flag:false}], 'mongo', 'app_users'+appid, 'processCompetingApps');

			}else{
				logger.error(`error=> ${err}`);	 
			}
		});
	}

	let updateUserInfo = function(params){
		let updatedAt = common.getCurrentEpochTime();
		common.db.collection('app_users'+params.app_id).update({"_id": params.user_id},{$set:{user_info:params.user_info,updatedAt:updatedAt}},function(err, result) {
			if(!err){
				if(result){
					logger.info('Success UserInfo')
				}
			}else{
				// data send to error queue
        errorQueue([{appid:params.app_id, op:'update',  data:{"_id": params.user_id,$set:{user_info:params.user_info,updatedAt:updatedAt}}}], 'mongo', 'app_users'+params.app_id, 'updateUserInfo');
		logger.error(`There is something error while updating the user:: ${err}`)
				err.updateUserInfo = 'There is something error while updating the user';
				errorHandler.sendError(err);
			}

		})
	}

	let updateUserHb = function(params){
		let updatedAt = common.getCurrentEpochTime();
		common.db.collection('app_users'+params.app_id).update({"_id": params.user_id},{$set:{hb:params.hb,updatedAt:updatedAt}},function(err, result) {
			if(!err){
				if(result){
					 
				}
			}else{
				// data send to error queue
        errorQueue([{appid:params.app_id, op:'update', data:{"_id": params.user_id,$set:{hb:params.hb,updatedAt:updatedAt}}}], 'mongo', 'app_users'+params.app_id, 'updateUserHb');

		logger.error(`There is something error while updating the HB:: ${err}`)
				err.updateUserHb = 'There is something error while updating the HB';
				errorHandler.sendError(err);
			}

		})
	}

	let updateTimelyDashbaord = function(params){
        common.db.collection('app_users'+params.app_id).find({"_id": params.user_id}, {'pv':1,'av':1,'p':1,'cc':1,'cty':1,'history':1,'tz':1}).toArray(function(err, result) {
            if(!err){
                let data = result[0];
                if(data){

	                if(data.pv && data.av && data.history && data.p && data.cc && data.cty){
	                    let history = data.history[data.history.length-1];
	                    if(params.av && params.pv && (data.av != params.av || data.pv != params.pv)){
	                    	data.pv = (data.pv)? data.pv: 'NA';
	                    	data.av = (data.av)? data.av: 'NA';
	                    	data.cc = (data.cc)? data.cc: 'NA';
	                    	data.cty = (data.cty)? data.cty: 'NA';

	                    	common.substractTimlyDashboard(data, params.app_id, function(result){
	                            if(result){
	                            	data.av = (params.av)? params.av: data.av;
			                        data.pv = (params.pv)? params.pv: data.pv;
	                                common.addTimlyDashboard(data, params.app_id);
	                            }
	                        });
	                    }
	                }
                }
            }
            else{
				logger.error(`error data not found while update meta data :: ${err}`) 
            }
        });
    }

	dataConsumer.processAppUpdates = function(data,callback){
		let apps = [];
		let collectionis =db.collection('apps');
		let bulk = collectionis.initializeUnorderedBulkOp();
		for (let lenData=0;lenData<data.length;lenData++){
			let datais = data[lenData];
			datais.app['modifiedOn'] = common.getCurrentEpochTime();
			// push apps data into array
      apps.push({appid:datais.appid, op:'update', data:{"_id":db.ObjectId(datais.appid), $set:datais.app}});
      // prepare apps bulk query
			bulk.find({"_id":db.ObjectId(datais.appid)}).update({$set:datais.app});
		}
		bulk.execute(function(err, result) {
			if (err){
				err.processAppUpdates = 'Error:processAppUpdates';
				errorHandler.sendError(err);
				logger.error(`Error:processAppUpdates => ${err}`);
				// data send to error queue
        errorQueue(apps, 'mongo', 'apps', 'processAppUpdates', function(status){
						if(status){
								callback(true);
						}else{
								callback(false);
						}
				});
			}else
				{
	   	    	 
	   	    	callback(true);
			}
    	});
    };

	dataConsumer.processCampaignResponses = function(data,finalcallback){
		let errorSQS = false;
		let auserArray = [], campaignArray = [];
		async.forEach(data, function(element, callback){
			let appid = element.appid;
			let dataArray = element.dataArray;
			let responseCollection =db.collection('campaign_responses');
			let appuserCollection =db.collection('app_users'+appid);
			let bulk_CampaignResponses = responseCollection.initializeUnorderedBulkOp();
			let bulk_AppUserUpdate = appuserCollection.initializeUnorderedBulkOp();

			if(dataArray){
				for (let i=0;i<dataArray.length;i++){
					let datais = dataArray[i];
					for(let j=0;j<datais.responses.length;j++){
						try{
							let campaignKey={};
							campaignKey["campaigns."+datais.responses[j].response[0].value]= datais.responses[j].campid;
							 
							datais.responses[j]['updatedAt'] = common.getCurrentEpochTime();
							let updatedAt = common.getCurrentEpochTime();

							if(datais.responses[j].response[0].qid==100 ||datais.responses[j].response[0].qid==200){
								// push campaign data into array
                campaignArray.push({appid:appid, op:'update', isUpsert:true, find:{'appid':datais.responses[j].appid,'campid':datais.responses[j].campid,"user_id":datais.responses[j].user_id,"response.qid":datais.responses[j].response[0].qid}, data:{'$set': datais.responses[j]}});
                // prepare campaign update query
								bulk_CampaignResponses.find({'appid':datais.responses[j].appid,'campid':datais.responses[j].campid,"user_id":datais.responses[j].user_id,"response.qid":datais.responses[j].response[0].qid}).upsert().update({'$set': datais.responses[j]});
								//Link campaigns to users which are actioned by them.
								bulk_AppUserUpdate.find({"_id":datais.responses[j].user_id}).update({$addToSet:campaignKey,$set:{'updatedAt':updatedAt}});
								// push campaign response data into array
                auserArray.push({appid:appid, op:'update', data:{"_id":datais.responses[j].user_id, $addToSet:campaignKey,$set:{'updatedAt':updatedAt}}});

							}
							else if(datais.responses[j].response[0].qid==300 ||datais.responses[j].response[0].qid==400){
								// push campaign data into array
                campaignArray.push({appid:appid, op:'update', isUpsert:true, find:{'appid':datais.responses[j].appid,'campid':datais.responses[j].campid,"user_id":datais.responses[j].user_id,$or:[{"response.qid":300},{"response.qid":400}]}, data:{'$set': datais.responses[j]}});
								// prepare campaign update query
								bulk_CampaignResponses.find({'appid':datais.responses[j].appid,'campid':datais.responses[j].campid,"user_id":datais.responses[j].user_id,$or:[{"response.qid":300},{"response.qid":400}]}).upsert().update({'$set': datais.responses[j]});
								//Link campaigns to users which are actioned by them.
								bulk_AppUserUpdate.find({"_id":datais.responses[j].user_id}).update({$addToSet:campaignKey,$set:{'updatedAt':updatedAt}});
								// push campaign response data into array
								auserArray.push({appid:appid, op:'update', isUpsert:true, data:{"_id":datais.responses[j].user_id, $addToSet:campaignKey,$set:{'updatedAt':updatedAt}}});
							}
							else{
								datais.responses[j]['createdAt'] = common.getCurrentEpochTime();
								// push campaign response data into array
                campaignArray.push({appid:appid, op:'insert', data:datais.responses[j]});
                //prepare campaign response insertion query
								bulk_CampaignResponses.insert(datais.responses[j]);
							}
						}
						catch(e){
							logger.error(`error => ${e}`);	 
						}
					}
				}
				async.parallel([
					//bulk_insert exceution
					function(callback){
						bulk_CampaignResponses.execute(function(err, result) {
							if (err){
									// data send to error queue
	                errorQueue(campaignArray, 'mongo', 'campaign_responses', 'processCampaignResponses', function(status){
											if(status){
													errorSQS = true;
											}
											err.processAppUpdates = 'Error:campaign responses';
											errorHandler.sendError(err);
											logger.error(` Error:campaign responses => ${err}`);
											callback(err);
									});
							}else
								{
					   	    	 
					   	    	callback();
							}
	    				});
					},
					//bulk_update execution
					function(callback){
						bulk_AppUserUpdate.execute(function(err, result) {
							if (err){
								// data send to error queue
                errorQueue(auserArray, 'mongo', 'app_users'+appid, 'processCampaignResponses', function(status){
										if(status){
												errorSQS = true;
										}
										err.processAppUpdates = 'Error:campaign update to appusers';
										errorHandler.sendError(err);
										logger.error(` Error:campaign update to appusers => ${err}`);
										callback(err);
								});
							}else
								{
					   	    	 
					   	    	callback();
							}
			    		});
					}
				],function(err){
					if(err){
						err.processAppUpdates = 'Error:processCampaignResponses:async:parallel';
						errorHandler.sendError(err);
						logger.error(` Error:processCampaignResponses:async:parallel => ${err}`);
						callback(err);
					}
					else{
						callback();
					}

				});
			}
	    },function(error){
					auserArray = []; campaignArray = [];
					if (error){
						error.processAppUpdates = 'Error:processCampaignResponses';
						errorHandler.sendError(error);
						logger.error(` Error:processCampaignResponses => ${error}`);
						if(errorSQS){
								finalcallback(true);
						}else {
								finalcallback(false);
						}
					}else{
						finalcallback(true);
					}
		});
	};

	dataConsumer.processCampaignData = function(data,finalcallback){
		let edate = common.getCurrentEpochTime();
    	let start = moment(edate*1000).subtract(120,'days').toDate();
    	edate = moment(edate*1000).add(1,'days').toDate();

		async.forEach(data, function(element, callback){
			let appid = element.appid;
			let dataArray = element.dataArray;
			let importMulti = cacheApi.client.multi();

			if(dataArray){
				//for (let i=0;i<dataArray.length;i++){
				async.forEach(dataArray, function(datais, inCallback){
					let criteria = [];
					let events = [];
					// add who into criteria
					if( typeof datais.segmentinfo == 'string'){
							datais.segmentinfo = JSON.parse(datais.segmentinfo);
					}

					if(datais.segmentinfo && datais.segmentinfo.who){
						datais.segmentinfo.who.forEach(function(obj){
							criteria.push(obj);
						})
					}
					// add what into criteria
					if(datais.segmentinfo && datais.segmentinfo.what){
						datais.segmentinfo.what.forEach(function(obj){
							events.push(obj);
						})
					}
					// add where into criteria
					if(datais.segmentinfo && datais.segmentinfo.where){
						datais.segmentinfo.where.forEach(function(obj){
							criteria.push(obj);
						})
					}
					// add when into criteria
					if(datais.segmentinfo && datais.segmentinfo.when){
						datais.segmentinfo.when.forEach(function(obj){
							criteria.push(obj);
						})
					}

					let eventLists = [];
		            let steps = [];
		            let filters = [];

		            // prepare filter and copy of criteria
		            common.prepareFilter(criteria, filters);

	                events.forEach(function(item){
	                    let obj={
	                        event_collection: 'event_'+datais.appid+"_"+item.operand,
	                        with_actors: true,
	                        actor_property: 'did',
	                        timeframe: {
	                            start: start,
	                            end: edate
	                        }
	                    };

	                    // validate segment information
	                    if(item.attr && item.attr != ''){
							logger.info(`${item.operand} :: item.attr ==> ${item.attr}`)
	                    	let filter = {
	                    			property_name:'segment.'+item.attr,
	                    			operator:item.operator,
	                    			property_value:item.value
	                    	}
	                    	obj.filters = [filter].concat(filters);
	                    }
	                    else{
	                       obj.filters = filters;
	                    }

	                    if(!item.value){
	                       obj.inverted = true;
	                    }
	                    eventLists.push({k:item.operand,p:false,s:true});
	                    steps.push(obj);
	                });

	                // if(steps.length > 0){
	                // 	steps[steps.length-1].with_actors = true;
	                // }
	                let keenUsers = [];
	                let userDid = [];
	                async.parallel([
	                	function(aCallback){
	                		if(steps.length > 0){
	                			 
	                			keen.getFunnelResultsWithActorProperty(steps,function(data){
	                				 
				                    if(data.actors && data.actors.length > 0){
				                    	 
				                    	//userDid = userDid.concat(data.actors[data.actors.length-1]);
				                    	keenUsers = data.actors;
				                    }
				                    aCallback();
				                });
	                		}
	                		else{
	                			aCallback();
	                		}
	                	},
	                	function(aCallback){
	                		 
	                		// if(criteria.length > 0){
	                		// 	let match = common.constructMatchFromCriteria(criteria);
		                	 
		                	// 	if(match){
		                	// 		match.active = true;
		                	// 		common.db.collection('app_users'+datais.appid).find(match,{did:1}).toArray(function(err,result){
					               
					              //       if(!err && result.length>0){
					              //           result.forEach(function(user){
					              //           	userDid.push(user.did);
					              //           })
					              //       }
					              //       aCallback();
					              //   });
		                	// 	}
		                	// 	else{
		                	// 		aCallback();
		                	// 	}
	                		// }
	                		// else{
	                		// 	aCallback();
	                		// }
	                		aCallback();
	                	}
	                	],function(err){
	                		userDid = [];
							// add unique users into cache
							logger.info("process data into cache")
							if(keenUsers && keenUsers.length > 0){
								for(k = keenUsers.length-1; k >= 0; k--){
									//userDid = userDid.concat(keenUsers[k]);
									let users = _(keenUsers[k]).filter(function(user) {
										 
								         if(userDid.indexOf(user) === -1){
								        	 userDid.push(user);
								         	return user;
								         }
								    });

									 
									// add last event unique users into cache
			                		 
			                		let obj = {st:false}
			                		// check event status next event
			                		if(eventLists[k+1]){
														eventLists[k+1].p = true;
														eventLists[k+1].s = events[k+1].value;
														if(!events[k+1].value){
																eventLists[k+1].s = true;
														}
			                		}

			                		 
							  				 // add users into cache
			                		users.forEach(function(user){
													obj.e = eventLists;
													 
													importMulti.hmset("campaignuser_"+datais.appid+""+datais.campid, user, JSON.stringify(obj));
									});
								}
								userDid = [];
							}

							// add events into cache
							events.forEach(function(event){
								importMulti.hmset("campaignevent_"+datais.appid+""+datais.campid, event.operand, JSON.stringify(event));
							});

							importMulti.exec(function(err,results){
								if (err) {
									logger.error(` error whiel import bulk record in cache => ${err}`);
									logger.info(`datais.campid ::: ${datais.campid}`); 
								} else {
							    	logger.info(`done campaign ::: ${results}`); 
								}
							   inCallback();
							});
	                });
				},function(err){
					callback();
				});
			}
	    },function(error){
			if (error){
				error.processAppUpdates = 'Error:processCampaignResponses';
				errorHandler.sendError(error);
				logger.error(`Error:processCampaignResponses::: ${error}`);
				finalcallback(false);
			}else{
				finalcallback(true);
			}
		});
	};

	function processEventSegments(events,app_id){
		events.forEach(function(evt){
			let list = [];
			let clist = [];
			if(evt.segment){
				let segmentList = Object.keys(evt.segment);
		        //Create event attribute list
		        segmentList.forEach(function(item){
		        	if(item.length>0){
		        		let obj = {};
			            obj.name = item;
			            obj.type = typeof(evt.segment[item]);

						if(semusiConfig.validateEventsApps.indexOf(app_id) >= 0){
							 if(obj.name == 'City'){
								 clist.push({name:evt.segment[item], type:typeof(evt.segment[item])});
							 }
						}
						list.push(obj);
		        	}
		        });
		    }
		    let updatedAt = common.getCurrentEpochTime();
	        db.collection('eventsegments').findOne({_id:db.ObjectId(app_id),"events.event":evt.key},function(err,result){
	        	if(result){
	        		db.collection('eventsegments').update({"_id" :db.ObjectId(app_id),"events.event":evt.key},{$addToSet: { 'events.$.list': {$each: list}, 'clist':{$each:clist} },$set:{'updatedAt':updatedAt} },function(err1,result1){});
	        	}
	        	else if(!err && !result){
	        		db.collection('eventsegments').update({"_id":db.ObjectId(app_id)},{$addToSet: {'events': {"event":evt.key, list:list}},$set:{'updatedAt':updatedAt, clist:clist}},{'upsert':true},function(err,result){});
	        	}
	        });
		});
  };

    function processEventAttributeAggregation(attributeData,collection,app_id){
			let attrArray = [];
    	let bulk_timely_eventAttributes =db.collection(collection+app_id).initializeUnorderedBulkOp();
    	async.forEach(attributeData,function(item,callback){
				async.parallel([
					function(callback1){
						cacheApi.addToSet(item.id+moment(item.timestamp).format("D"),item.attValue,function(data){
							if(parseInt(data)<=50){
								// push event attributes into array
                attrArray.push({appid:app_id, op:'update', isUpsert:true, data:{'_id':item.id, '$inc': item.data}})
								// prepare event attribute query
								bulk_timely_eventAttributes.find({'_id':item.id}).upsert().update({'$inc': item.data});
								//Set expiry to redis key.
								if(parseInt(data)==1){
									cacheApi.expire(item.id+moment(item.timestamp).format("D"),60*60*24*3);
								}
							}
							else{
								 
							}
							callback1();

						});
					}

				],function(err){
					if(err){
						err.processEventAttributeAggregation = 'Error:processEventAttributeAggregation';
						errorHandler.sendError(err);
						logger.error(`Error:processEventAttributeAggregation::: ${err}`);
						callback(err);
					}
					else{
						callback();
					}

				});
	    	},
	    	function(err){
	    		if(err){
	    			//callback(err);
	    		}
	    		else{
	    			//callback();
	    			bulk_timely_eventAttributes.execute(function(err, result) {
								if (err){
									// send data into error queue
		              errorQueue(attrArray, 'mongo', collection+app_id, 'processEventAttribute');

									err.processEventAttributeAggregation = 'Error:processEventAttributeAggregation query';
									errorHandler.sendError(err);
									logger.error(`Error:processEventAttributeAggregation::: ${err}`);
								}else{
						   	    	 
								}
								attrArray = [];
			    	});
	    		}
	    	}
    	);
    }

    function updateCustomFieldCollection(data){
		let keys = Object.keys(data.user);
		let userfields=[];
		keys.forEach(function(key){
			if(key.indexOf('_custom') > -1){
				let customlet = {};
                customVar.name = key;
                customVar.type = typeof(data.user[key]);
                userfields.push(customVar);
			}
		});

		if(userfields.length>0){
			let updatedAt = common.getCurrentEpochTime();
            db.collection('app_customfields').update({"_id":db.ObjectId(data.appid)},{'$addToSet':{ 'fields': { '$each': userfields } }, '$set':{'updatedAt':updatedAt}}, {'upsert':true}, function () {});
        }
	};

	function updateReferal(data,callback){
		if(data.source!=="self" || data.attribution!==""){
			db.collection('app_users'+data.appid).findOne({_id:data.user_id},{"history":{"$slice":-1},'pv':1,'av':1,'p':1,'cc':1,'cty':1,'tz':1},function(err,appUser){
				if(!err && appUser){
					if(appUser.history && appUser.history.length>0 && appUser.history[0].type=="I"){
						let refObj = appUser.history[0];
						refObj.updated="api";
						common.populateHistoryObject(decodeURIComponent(data.source).replace(/\+/g, " "),refObj);
						if(refObj.refname=="self"){
							//Try to get referrer from install attribution
							common.populateHistoryObject(decodeURIComponent(data.attribution).replace(/\+/g, " "),refObj);
							if(refObj.refname==undefined || refObj.refname==""){
								refObj.refname="self";
							}
						}
						if(refObj.refname!=="self")	{
							db.collection('app_users' + data.appid).update({_id:data.user_id}, {'$pop':{'history':1}}, function (err,poppedUser) {
								let updatedAt = common.getCurrentEpochTime();
								db.collection('app_users' + data.appid).update({'_id':data.user_id}, {'$addToSet':{'history':refObj},'$set':{'updatedAt':updatedAt}}, function (err,pushedUser) {
									if(err){
										//callback(false);
										err.processEventAttributeAggregation = 'Error, while updating referal.';
										errorHandler.sendError(err);
										logger.error(`Error, while updating referal => ${err}`);
									}
									else{
										// check old and new source
										if(appUser.history && appUser.history[0] && appUser.history[0].refname != refObj.refname ){
											try{
	                                            // send email while update refferer
	                                            errorHandler.sendLog({oldhistory:appUser.history, newHistory:refObj}, 'updateReferal name appid '+data.appid);
											}
											catch (e){
											}

											if(data.appid == '57f7a1a9662688ff05897eb1'){
												logger.info(`old refferer => ${appUser.history[0].refname}`);
												logger.info(`newrefferer => ${refObj.refname}`); 
											}

											 
											let datas = {}
											datas.pv = (appUser.pv)? appUser.pv : 'NA';
											datas.av = (appUser.av)? appUser.av : 'NA';
											datas.cc = (appUser.cc)? appUser.cc : 'NA';
											datas.cty = (appUser.cty)? appUser.cty : 'NA';
											datas.p = appUser.p;
											datas.history = appUser.history;

											common.substractTimlyDashboard(datas, data.appid, function(result){
					                            if(result){
					                            	 
					                            	datas.history = [refObj];
					                                common.addTimlyDashboard(datas, data.appid);
					                                 
					                            }
					                        });
										}
									}
								});
							});
						}
					}
				}
			});
		}
	}
	function updateGCMIdToDynamoDB(user,appid){
            if(user.gcmid){
                    dynamodb.updateGCMId(user.gcmid,appid);
            }
    };

		// generate event segement with city for radio meter
		function createEventSegment(eventName, key, epoch, platform, attValue){
			  let obj ={};
			  let eventAttributeObj={};
				let docId = common.crypto.createHash(semusiConfig.shaV1Hash).update(eventName+key+moment(epoch).format("YYYYM")+"").digest('hex');
				common.fillEventAttributeObjectWithEpoch(epoch,eventAttributeObj,platform,attValue);
				obj.id = docId;
				obj.data = eventAttributeObj;
				obj.timestamp = epoch;
				obj.attValue = attValue;
				return obj;
		}

    // add last 5 events into app user collection
    function updateLastEvents(appid, did, user_id, event){
    	if(appid == '583e6143ca3b1959380b54b4'){
    		 
	    	if(event.k != 'Session_Start'
	    		&& event.k != 'Session_End'
	    		&& event.k != 'CSession_Start'
	    		&& event.k != 'CSession_End'
	    		&& event.k != 'App_Foreground'
	    		&& event.k != 'App_Background'){
	    		 
    			//db.collection('app_users'+appid).findAndModify({query:{_id:user_id, $or:[{el:{$exists:false}}, {$where:'this.el.length<5'}]}, update:{$addToSet:{'el':event}}},function(err,result){
		    	db.collection('app_users'+appid).findAndModify({query:{_id:user_id, $or:[{el:{$exists:false}}, {el:{$size:0}}, {el:{$size:1}}, {el:{$size:2}}, {el:{$size:3}}, {el:{$size:4}}, {el:{$size:5}}, {el:{$size:6}}, {el:{$size:7}}, {el:{$size:8}}, {el:{$size:9}}]}, update:{$addToSet:{'el':event}}},function(err,result){
						if(result == null){
							 
							let bulk =db.collection('app_users'+appid).initializeUnorderedBulkOp();
							bulk.find({_id:user_id}).update({$addToSet:{'el':event}});
							bulk.find({_id:user_id}).update({$pop:{el:-1}});
							bulk.execute(function(err, res){
								 
							});
						}
					});
	    	}
	    }
    	else{
    		db.collection('app_users'+appid).findAndModify({query:{_id:user_id, $or:[{el:{$exists:false}}, {el:{$size:0}}, {el:{$size:1}}, {el:{$size:2}}, {el:{$size:3}}, {el:{$size:4}}, {el:{$size:5}}, {el:{$size:6}}, {el:{$size:7}}, {el:{$size:8}}, {el:{$size:9}}]}, update:{$addToSet:{'el':event}}},function(err,result){
				if(result == null){
					let bulk =db.collection('app_users'+appid).initializeUnorderedBulkOp();
					bulk.find({_id:user_id}).update({$addToSet:{'el':event}});
					bulk.find({_id:user_id}).update({$pop:{el:-1}});
					bulk.execute(function(err, res){
						 
					});
				}
			});
    	}
    }

		let errorQueue = function(eventsData, type, col, method, callback){
	        let obj = {
	            method  : method,
	            col		: col,
	            type 	: type,
	            events  : eventsData,
	            key 	: "errors"
	        };
					dataProducer.writeDataToSQS(obj, function(status){
							if(callback){
								return callback(status);
							}
							return;
					});
		}
}(dataConsumer));

module.exports = dataConsumer;
