let dataConsumer = {},
	common = require('./common.js');  
	//awsHelper = require('./../parts/data/aws.data.js'),
    //kafkaProducer = require('./../parts/data/kafka.producer.js');

(function (dataConsumer) {

	dataConsumer.processBeginSessionData = function(data,callback){

		let session = {st:data.timestamp};
		if(data.context && data.context.places && data.context.places._t){
			session.p = data.context.places._t;
		}

		let updatedAt = common.getCurrentEpochTime();
		common.db.collection('app_user'+data.appid).update({"_id":data.user_id},{$set:{"ls":data.timestamp, 'updatedAt':updatedAt},$addToSet:{'sessions':session},$inc:{'sc':1}},function(err,result){
			if(err){
				callback(false);		
			}
			else{
				callback(true);			
			}
		});	
	};

	dataConsumer.processEndSessionData = function(data,callback){

		let updatedAt = common.getCurrentEpochTime();
		common.db.collection('app_user'+data.appid).update({"_id":data.user_id},{$inc:{'tsd':data.duration}, $set:{'updatedAt':updatedAt}},function(err,result){
			if(err){
				callback(false);		
			}
			else{
				callback(true);			
			}
		});
	};

	dataConsumer.processEvents = function(data,callback){     
	    data.events['createdAd'] = common.getCurrentEpochTime();
        common.db.collection('events_'+data.appid).insert(data.events,function(err, result) {
        	if(err){
        		callback(false);
        	}
        	else{
        		callback(true);
        		//Update last event to app users collection.
	            let currentEvent = data.events[data.events.length-1];
	            let updatedAt = common.getCurrentEpochTime();
           		common.db.collection('app_users'+data.appid).update({did:currentEvent.did},{"$set":{"le":{k:currentEvent.key,t:currentEvent.eventTime,s:currentEvent.segment,c:currentEvent.context}, 'updatedAt':updatedAt}},{safe:true},function(err,res){});
           		//Update event segment list.
           		processEventSegments(data.events,data.appid);
        	}

        });                
	};

	dataConsumer.processDeviceUpdates = function(data,callback){
		if(data.user){
			data.user = common.getCurrentEpochTime();
			common.db.collection('app_users'+data.appid).update({_id:data.user_id},{$set:data.user},function(err,result){
				if(err){
	        		callback(false);
	        	}
	        	else{
	        		callback(true);	
	        		updateCustomFieldCollection(data);
	        	}
			});
		}
		if(data.source){
			updateReferal(data,callback);
		}
		
	};

	dataConsumer.processAppUpdates = function(data,callback){
		data.app['modifiedOn'] = common.getCurrentEpochTime();
		common.db.collection('apps').update({_id:common.db.ObjectID(data.appid)},{$set:data.app},function(err,result){
			if(err){
				callback(false);
			}
			else{
				callback(true);	
			}			
		});
	};

	dataConsumer.processCampaignResponses = function(data,callback){
		data.responses['createdAd'] = common.getCurrentEpochTime();
		common.db.collection('campaign_responses').insert(data.responses,function(err,result){
			if(err){
				callback(false);
			}
			else{
				callback(true);
			}
		});		
	}

	function updateReferal(data,callback){
		let updatedAt = common.getCurrentEpochTime();
		common.db.collection('app_users'+data.appid).findOne({_id:data.user_id},{"history":{"$slice":-1}},function(err,appUser){
			if(!err && appUser){
				if(appUser.history && appUser.history[0].type=="I"){
					common.db.collection('app_users' + data.appid).update({_id:data.user_id}, {'$pop':{'history':1}, '$set':{'updatedAt':updatedAt}}, function (err,poppedUser) {
						let refObj = {};
						common.populateHistoryObject(data.source,refObj);
						refObj.type="I";
						refObj.dtEntry = appUser.history[0].dtEntry;
						refObj.ip_address = appUser.history[0].ip_address;
						refObj.isnew = appUser.history[0].isnew;
						refObj.installer = (appUser.history[0].installer)?appUser.history[0].installer:"";

						let updatedAt = common.getCurrentEpochTime();
						common.db.collection('app_users' + data.appid).update({'_id':data.user_id}, {'$addToSet':{'history':refObj}, '$set':{'updatedAt':updatedAt}}, function (err,pushedUser) {
							if(err){
								callback(false);
							}
							else{
								callback(true);
							}
						});
					});
				}
				else{
					callback(true);
				}
			}
			else{
				callback(false)
			}

		});
	}

	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();
            common.db.collection('app_customfields').update({"_id":data.appid},{'$addToSet':{ 'fields': { '$each': userfields } }, '$set':{'updatedAt':updatedAt}}, {'upsert':true}, function () {});
        }
	}

	function processEventSegments(events,appid){
		events.forEach(function(evt){
			let segmentList = Object.keys(evt.segment);
	        let list = [];
	       
	        //Create event attribute list.
	        segmentList.forEach(function(item){
	            let obj = {};
	            obj.name = item;
	            obj.type = typeof(segment[item]);
	            list.push(obj);
	        });
	        
	        let updatedAt = common.getCurrentEpochTime();
	        common.db.collection('eventsegments').update({"_id":appid,"events.event":{$nin:[evt.key]}},{$push: {'events': {"event":evt.key, list:[]}}, $set:{'updatedAt':updatedAt}},{upsert:true},function(err,result){
	            common.db.collection('eventsegments').update({"_id" : params.app_id,"events.event":evt.key},{$addToSet: { 'events.$.list': {$each: list} }, $set:{'updatedAt':updatedAt} },function(err1,result1){

	            });
	        });	
		});
        
         
    }


}(dataConsumer));

module.exports = dataConsumer;