let dataConsumer = {},
	common = require('./common.js'),
	logger = require('../../logger'),
	semusiConfig = require('./../config'),
	mongojs = require('mongojs'),
	cacheApi = require('./semusi.cache.js'),	
	moment = require('moment'),
	async = require('async');
	//dynamodb = require('/datadrive/sap/dynamodbdata/dynamodb.js');
	
    let db;
    if(semusiConfig.IsProduction){
    	db = mongojs(semusiConfig.mongodb.productionUrl);
    }
    else{
    	db = mongojs(semusiConfig.mongodb.localUrl);
    }    

(function (dataConsumer) {

	dataConsumer.processBeginSessionData = function(data,finalcallback){
		async.forEach(data, function(element, callback){
			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_sessions = db.collection('sessions').initializeUnorderedBulkOp();
			for (let i=0;i<dataArray.length;i++){
				datais = dataArray[i]
				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;
				}
				//users_sessions.insert({appid:element.appid, user_id:datais.user_id, st:datais.timestamp, c:datais.context});
				let updateSessions = datais.commonSession;		
				bulk.find({"_id":datais.user_id}).update({$set:{"ls":datais.timestamp},$inc:{'sc':1}});
				//bulk_sessions.find({'_id':db.ObjectId(element.appid)}).upsert().update({'$inc': updateSessions});
				bulk_timely_sessions.find({'_id':db.ObjectId(element.appid)}).upsert().update({'$inc': datais.timelySession});
			}
			async.parallel([
				//app-users update
				function(callback){
					bulk.execute(function(err, result) {
						if (err){
							logger.error(`Error:processBeginSessionData:app_users == >> ${err}`);
							callback(err);
						}else
							{
				   	    	 
				   	    	callback();
						}
			    	});
				},
				//sessions update
				/*function(callback){
					bulk_sessions.execute(function(err, result) {
						if (err){
							 
							 
							callback(err);
						}else
							{
				   	    	 
				   	    	callback();
						}
			    	});
				},*/
				//timely sessions update
				function(callback){
					bulk_timely_sessions.execute(function(err, result) {
						if (err){
							logger.error(`Error:processBeginSessionData:sessions == >> ${err}`);
							callback(err);
						}else
							{
				   	    	 
				   	    	callback();
						}
			    	});
				}
				//users sessions manage
				/*function(callback){
					users_sessions.execute(function(err, result) {
						if (err){
							 
							 
							callback(err);
						}else
							{
				   	    	 
				   	    	callback();
						}
			    	});
				}*/
			],function(err){
				if(err){
					logger.error(`Error:processBeginSessionData:async:parallel == >> ${err}`);
					callback(err);
				}
				else{
					callback();	
				}
				
			});	 
			
		},function(error){
			if (error){
				logger.error(`Error:processBeginSessionData == >> ${error}`);
				finalcallback(false);
			}else{
				finalcallback(true);
			}
		});
	};

	dataConsumer.processEndSessionData = function(data,finalcallback){
		async.forEach(data, function(element, callback){
			dataArray = element.dataArray;
			let collectionis =db.collection('app_users'+element.appid)
			let bulk = collectionis.initializeUnorderedBulkOp();
			for (let i=0;i<dataArray.length;i++){
				datais = dataArray[i];
				bulk.find({"_id":datais.user_id}).update({$inc:{'tsd':datais.duration},$set:{'lsd':datais.duration}});
				}
			bulk.execute(function(err, result) {
				if (err){
					logger.error(`Error:processBeginSessionData:app_users == >> ${err}`);
					callback(err);
				}else
					{
		   	    	 
		   	    	callback();
				}
	    	});
	    },function(error){
			if (error){
				logger.error(`Error:processEndSessionData ==>> ${error}`); 
				finalcallback(false);
			}else{
				finalcallback(true);
			}
		});
	};

	/*dataConsumer.processEvents = function(data,finalcallback){
		async.forEach(data, function(element, callback){
			let appid = element.appid;
			dataArray = element.dataArray;
			let eventCollection =db.collection('events_'+element.appid)
			let appuserCollection =db.collection('app_users'+element.appid)
			let bulk_insert = eventCollection.initializeUnorderedBulkOp();
			let bulk_update = appuserCollection.initializeUnorderedBulkOp();
			for (let i=0;i<dataArray.length;i++){
				datais = dataArray[i];
				let currentEvent = datais.events[datais.events.length-1];
				for(let j=0;j<datais.events.length;j++){
					bulk_insert.insert(datais.events[j]);
				}
				bulk_update.find({"_id":datais.user_id}).update({"$set":{"le":{k:currentEvent.key,t:currentEvent.eventTime,s:currentEvent.segment,c:currentEvent.context}}},{safe:true});
				//Update event segment list.
				processEventSegments(datais.events,appid);
			}
			async.parallel([
				//bulk_insert exceution
				function(callback){
					bulk_insert.execute(function(err, result) {
						if (err){
							 
							 
							callback(err);
						}else
							{
				   	    	 
				   	    	callback();
						}
	    				});
				},
				//bulk_update execution
				function(callback){
					bulk_update.execute(function(err, result) {
						if (err){
							 
							 
							callback(err);
						}else
							{
				   	    	 
				   	    	callback();
						}
			    		});
				}
			],function(err){
				if(err){
					 
					callback(err);
				}
				else{
					callback();	
				}
				
			});	    	
	    },function(error){
			if (error){
				 
				finalcallback(false);
			}else{
				finalcallback(true);
			}
		});
	};*/

	dataConsumer.processEvents = function(data,finalcallback){
		async.forEach(data, function(element, callback){
			let appid = element.appid;
			dataArray = element.dataArray;
			let eventAttributes = [];
			let eventCollection =db.collection('events_'+element.appid)
			let appuserCollection =db.collection('app_users'+element.appid)
			let bulk_timely_events =db.collection('timely_events').initializeUnorderedBulkOp();
			let bulk_insert = eventCollection.initializeUnorderedBulkOp();
			let bulk_update = appuserCollection.initializeUnorderedBulkOp();
			for (let i=0;i<dataArray.length;i++){
				datais = dataArray[i];
				let currentEvent = datais.events[datais.events.length-1];
				for(let j=0;j<datais.events.length;j++){
					bulk_insert.insert(datais.events[j]);
					if(datais.events[j].key && datais.events[j].eventTime && datais.events[j].context && datais.events[j].context.who && datais.events[j].context.who.device && datais.events[j].context.who.device.p ){
						let platform = datais.events[j].context.who.device.p.toLowerCase();
	    				let epoch = (datais.events[j].eventTime*1000) // convert epoch from gmt
	   
					    let id = common.timelyEventId(appid, epoch);
					    let updateEvents = {};
					    common.fillEventObjectWithEpoch(epoch, updateEvents, platform, datais.events[j].key);
					    common.fillEventObjectWithEpoch(epoch, updateEvents, platform, "total");
						
						bulk_timely_events.find({'_id':id}).upsert().update({'$inc': updateEvents});
						//Maintain aggregate collection of attribute values.
						if(datais.events[j].key.toLowerCase() != '_app_crash' && datais.events[j].segment){
							let keys = Object.keys(datais.events[j].segment);
							keys.forEach(function(key){
								let obj ={};
								let attValue = (typeof(datais.events[j].segment[key])=="string") ? evt.segment[key].replace(/\./g, ":") : "";
								if(attValue!=""){
									let eventAttributeObj={};
									//Create document id.
									let eventName = datais.events[j].key;
				        			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.attKey = key;
				        			obj.attValue = attValue;
				        			eventAttributes.push(obj);				        			
								}
							});							
						}
					}
				}
				bulk_update.find({"_id":datais.user_id}).update({"$set":{"le":{k:currentEvent.key,t:currentEvent.eventTime,s:currentEvent.segment,c:currentEvent.context}}},{safe:true});
				//Update event segment list.
				processEventSegments(datais.events,appid);
				processEventAttributeAggregation(eventAttributes,appid);
			}
			async.parallel([
				//bulk_insert exceution
				function(callback){
					bulk_insert.execute(function(err, result) {
						if (err){
							logger.error(`Error:processEvents:events_appid == >> ${err}`);
							callback(err);
						}else
							{
				   	    	 
				   	    	callback();
						}
	    				});
				},
				//bulk_update execution
				function(callback){
					bulk_update.execute(function(err, result) {
						if (err){
							logger.error(`Error:processEvents:app_users == >> ${err}`);
							callback(err);
						}else
							{
				   	    	 
				   	    	callback();
						}
			    		});
				},
				//bulk_update timely session execution
				function(callback){
					bulk_timely_events.execute(function(err, result) {
						if (err){
							logger.error(`Error:processEvents:app_users == >> ${err}`);
							callback(err);
						}else
							{
				   	    	 
				   	    	callback();
						}
			    		});
				}
			],function(err){
				if(err){
					logger.error(`Error:processEvents:async:parallel == >> ${err}`);
					callback(err);
				}
				else{
					callback();	
				}
				
			});	    	
	    },function(error){
			if (error){
				logger.error(`Error:processEvents =>> ${error}`);
				finalcallback(false);
			}else{
				finalcallback(true);
			}
		});
	};

	dataConsumer.processDeviceUpdates = function(data,finalcallback){
		async.forEach(data, function(element, callback){
			dataArray = element.dataArray;
			let collectionis =db.collection('app_users'+element.appid)
			let bulk = collectionis.initializeUnorderedBulkOp();
			for (let i=0;i<dataArray.length;i++){
				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);

					bulk.find({"_id":datais.user_id}).update({$set:datais.user});
					updateCustomFieldCollection(datais);
					//updateGCMIdToDynamoDB(datais.user,element.appid);
				}

				if (datais.source){
					updateReferal(datais);
				}
			}
			bulk.execute(function(err, result) {
					if (err){
						logger.error(`Error:processDeviceUpdates:app_users =>> ${err}`);
						callback(err);
					}else
						{
			   	    	 
			   	    	callback();
					}
		    	});
		},function(error){
				if (error){
					logger.error(`Error:processDeviceUpdates:app_users =>> ${err}`);
					finalcallback(false);
				}else{
					finalcallback(true);
				}
			});
	};

	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 collectionis =db.collection('apps');
		let bulk = collectionis.initializeUnorderedBulkOp();
		for (let lenData=0;lenData<data.length;lenData++){
			datais = data[lenData];
			bulk.find({"_id":db.ObjectId(datais.appid)}).update({$set:datais.app});
		}
		bulk.execute(function(err, result) {
			if (err){
				logger.error(`Error:processAppUpdates==>> ${err}`); 
			}else
				{
	   	    	 
	   	    	callback(true);
			}
    	});
    };
	
	dataConsumer.processCampaignResponses = function(data,finalcallback){
		async.forEach(data, function(element, callback){
			let appid = element.appid;
			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++){
					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;
							logger.info(`campaignKey==>> ${campaignKey}`);
							logger.info(`datais . response :: ${datais.responses[j].user_id}`);
							if(datais.responses[j].response[0].qid==100 ||datais.responses[j].response[0].qid==200){
								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});
							}
							else if(datais.responses[j].response[0].qid==300 ||datais.responses[j].response[0].qid==400){
								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});
							}
							else{
								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){
								logger.error(`Error:campaign responses=> ${err}`);
								callback(err);
							}else
								{
					   	    	 
					   	    	callback();
							}
	    				});
					},
					//bulk_update execution
					function(callback){
						bulk_AppUserUpdate.execute(function(err, result) {
							if (err){
								logger.error(`Error:campaign update to appusers => ${err}`);
								callback(err);
							}else
								{
					   	    	 
					   	    	callback();
							}
			    		});
					}
				],function(err){
					if(err){
						logger.error(`Error:processCampaignResponses:async:parallel==>> ${err}`);
						callback(err);
					}
					else{
						callback();	
					}
					
				});
			}	    	
	    },function(error){
			if (error){
				logger.error(`Error:processCampaignResponses:==>> ${error}`);
				finalcallback(false);
			}else{
				finalcallback(true);
			}
		});
	};

	function processEventSegments(events,app_id){
		events.forEach(function(evt){
			let list = [];
			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]);
			            list.push(obj);	
		        	}		            
		        });
		    }    
	        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} } },function(err1,result1){});	
	        	}
	        	else if(!err && !result){
	        		db.collection('eventsegments').update({"_id":db.ObjectId(app_id)},{$addToSet: {'events': {"event":evt.key, list:list}}},{'upsert':true},function(err,result){});	
	        	}
	        });								        
		});                 
    };

    function processEventAttributeAggregation(attributeData,app_id){    	
    	let bulk_timely_eventAttributes =db.collection('timely_eventAttributes'+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)<=30){	
								 
								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);
								}
							}
							else{
								 
								 
							}							
							callback1();
							
						});		
					}
					
				],function(err){
					if(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){
							logger.error(`Error:processEventAttributeAggregation==>> ${err}`);
						}else{					   	    	
				   	    	 
						}

						process.exit(1);
			    	});	
	    		}
	    	}
    	);    	
    }

    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){
            db.collection('app_customfields').update({"_id":db.ObjectId(data.appid)},{'$addToSet':{ 'fields': { '$each': userfields } }}, {'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) {														
								db.collection('app_users' + data.appid).update({'_id':data.user_id}, {'$addToSet':{'history':refObj}}, function (err,pushedUser) {
									if(err){
										//callback(false);
										logger.error(`Error, while updating referal ==>> ${err}`);
									}
									else{
										logger.info(`source:: update:: ${datas}`);
										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){
												logger.info('substractTimlyDashboard ::');
				                            	datas.history = [refObj];
				                                common.addTimlyDashboard(datas, data.appid);
												logger.info(`add update timely dashboard :: ${datas}`);
				                            }
				                        });
									}
								});																								
							});
						}
					}
				}				
			});	
		}		
	}
	function updateGCMIdToDynamoDB(user,appid){
            if(user.gcmid){
                    dynamodb.updateGCMId(user.gcmid,appid);
            }
    };

	/////populate existing event attributes.
	function populateExistingeventAttributes(appid){
		let eventAttributes = [];
		//{eventTime:{$gte:1467315000,$lte:1467916199}}
		
		let epoch = moment().startOf('day').valueOf()/1000;
		
		db.collection('events_' + appid).find({eventTime:{$gte:epoch}}).toArray(function(err,events){
			events.forEach(function(evt){

				if(evt.key && evt.eventTime && evt.context && evt.context.who && evt.context.who.device && evt.context.who.device.p ){
					if(evt.segment){
						logger.info(`evt. segment :: ${evt.segment}`);
						let keys = Object.keys(evt.segment);
						let platform = evt.context.who.device.p.toLowerCase();
						let epoch = (evt.eventTime*1000);
						keys.forEach(function(key){							
							let obj ={};
							
							let attValue = (typeof(evt.segment[key])=="string") ? evt.segment[key].replace(/\./g, ":") : "";
							if(attValue!=""){
								
								let eventAttributeObj={};
								//Create document id.
								let eventName = evt.key;
			        			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;
			        			eventAttributes.push(obj);				        			
							}
						});							
					}
				}	
			});
			
			 
			while(eventAttributes.length>0){ 
				logger.info(`${eventAttributes.splice(0,4)}`);
				processEventAttributeAggregation(eventAttributes.splice(0,999),appid);
			}
			logger.info(`done!`);
			process.exit(1);
		});
		
	}
	populateExistingeventAttributes("56740ceb566583c105000001");
}(dataConsumer));

module.exports = dataConsumer;
