

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of CMSDK version 2.0.0.170915_Beta4 (15 September 2017)
//_/_/  
//_/_/  developed by
//_/_/  
//_/_/  Communicative Machines Ltd. 
//_/_/    http://cmlabs.com/
//_/_/  Copyright (c) 2017
//_/_/  
//_/_/  in collaboration with 
//_/_/  
//_/_/  Icelandic Institute for Intelligent Machines ses
//_/_/    Menntavegur 1
//_/_/    101 Reykjavik
//_/_/    ICELAND
//_/_/    http://www.iiim.is/
//_/_/
//_/_/  and is released under the following
//_/_/  
//_/_/  EXTENDED BSD License
//_/_/
//_/_/  Redistribution and use in source and binary forms, with or without
//_/_/  modification, is permitted provided that the following conditions 
//_/_/  are met:
//_/_/
//_/_/  - Redistributions of source code must retain the above copyright 
//_/_/    and collaboration notice, this list of conditions and the following 
//_/_/    disclaimer.
//_/_/
//_/_/  - Redistributions in binary form must reproduce the above copyright 
//_/_/    notice, this list of conditions and the following disclaimer in the 
//_/_/    documentation and/or other materials provided with the distribution.
//_/_/
//_/_/  - Neither the name of its copyright holders nor the names of its 
//_/_/    contributors may be used to endorse or promote products derived from 
//_/_/    this software without specific prior written permission.
//_/_/
//_/_/  - CADIA Clause: The license granted in and to the software under this 
//_/_/    agreement is a limited-use license. The software may not be used in 
//_/_/    furtherance of: (i) intentionally causing bodily injury or severe 
//_/_/    emotional harm to any person; (ii) invading the personal privacy or 
//_/_/    violating the human rights of any person; or (iii) committing or 
//_/_/    preparing for any act of war. 
//_/_/
//_/_/  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
//_/_/  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
//_/_/  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A 
//_/_/  PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER 
//_/_/  OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 
//_/_/  EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 
//_/_/  PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
//_/_/  PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
//_/_/  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
//_/_/  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
//_/_/  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
#include "MessageIndex.h"

namespace	cmlabs {

MessageIndex::MessageIndex() {
	lastMaintenance = GetTimeNow();
	buffertime = 10000;
	maintenanceInterval = 10000;
	defaultTTL = PSYMINUTE;
	typeMap = NULL;
	timeMap = NULL;
}

MessageIndex::~MessageIndex() {
	timeMaps.clear();
	stringMaps.clear();
	integerMaps.clear();
	floatMaps.clear();

	if (typeMap)
		delete(typeMap);
	typeMap = NULL;

	if (timeMap)
		delete(timeMap);
	timeMap = NULL;

	std::multimap<uint64, DataMessage*>::iterator i = eolMap.begin(), e = eolMap.end();
	while (i != e) {
		delete(i->second);
		i++;
	}
	eolMap.clear();
}

bool MessageIndex::setDefaultTTL(uint64 defaultTTL) {
	this->defaultTTL = defaultTTL;
	return true;
}

bool MessageIndex::addIndexKey(const char* key, uint8 type) {
	switch(type) {
	case INDEX_TIME:
		if ((stricmp(key, "time") == 0) && !timeMap) {
			timeMap = new std::multimap<uint64, DataMessage*>;
			return true;
		}
		else {
			timeMaps[key];
			return true;
		}
	case INDEX_STRING:
		stringMaps[key];
		return true;
	case INDEX_INTEGER:
		integerMaps[key];
		return true;
	case INDEX_FLOAT:
		floatMaps[key];
		return true;
	default:
		return false;
	}
}

bool MessageIndex::removeIndexKey(const char* key, uint8 type) {
	switch(type) {
	case INDEX_TIME:
		if ((stricmp(key, "time") == 0) && timeMap) {
			delete(timeMap);
			timeMap = NULL;
			return true;
		}
		else
			return deleteIndexKey<uint64>(key, timeMaps);
	case INDEX_STRING:
		return deleteIndexKey<std::string>(key, stringMaps);
	case INDEX_INTEGER:
		return deleteIndexKey<int64>(key, integerMaps);
	case INDEX_FLOAT:
		return deleteIndexKey<float64>(key, floatMaps);
	default:
		return false;
	}
}


bool MessageIndex::setFileStorage(const char* dir, uint32 buffertime, const char* indexfilename) {
	this->dir = dir;
	this->indexfilename = indexfilename;
	this->buffertime = buffertime;
	return true;
}

uint32 MessageIndex::getCount() {
	return (uint32)eolMap.size();
}

bool MessageIndex::addMessage(DataMessage* msg) {
	if (GetTimeAgeMS(lastMaintenance) > maintenanceInterval)
		doMaintenance();

	// Add to each key in each type map
	addMessageToTime(msg);
	addMessageToString(msg);
	addMessageToInteger(msg);
	addMessageToFloat(msg);

	if (timeMap)
		timeMap->insert(Time_Pair(msg->getCreatedTime(), msg));
	if (typeMap)
		typeMap->insert(Type_Pair(msg->getType(), msg));

	// Add to EOL map
	// if msg->ttl is not set we need to use the default timeout to determine when to delete it
	if (!msg->getTTL())
		eolMap.insert(Time_Pair(msg->getCreatedTime() + defaultTTL, msg));
	else
		eolMap.insert(Time_Pair(msg->getEOL(), msg));
	return true;
}

char* MessageIndex::queryMessages(RetrieveSpec* spec, uint32 &size, uint32 &count) {
	if (GetTimeAgeMS(lastMaintenance) > maintenanceInterval)
		doMaintenance();

	size = count = 0;
	if (!spec)
		return NULL;

	std::map<std::string, std::multimap<uint64, DataMessage*> >::iterator it;
	std::map<std::string, std::multimap<std::string, DataMessage*> >::iterator is;
	std::map<std::string, std::multimap<int64, DataMessage*> >::iterator ii;
	std::map<std::string, std::multimap<float64, DataMessage*> >::iterator ifl;

//	printf("Querying for key '%s'...\n", spec->key);

	switch(spec->keytype) {
	case INDEX_TIME:
		if ((stricmp(spec->key, "time") == 0) && timeMap) {
			if (spec->startTime && spec->endTime)
				return queryMessages<uint64>(spec->startTime, spec->endTime, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, *timeMap);
			else if (spec->startTime)
				return queryMessages<uint64>(spec->startTime, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, *timeMap);
			else
				return queryMessages<uint64>(spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, *timeMap);
		}
		else {
			it = timeMaps.find(spec->key);
			if (it != timeMaps.end()) {
				if (spec->startTime && spec->endTime)
					return queryMessages<uint64>(spec->startTime, spec->endTime, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, it->second);
				else if (spec->startTime)
					return queryMessages<uint64>(spec->startTime, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, it->second);
				else
					return queryMessages<uint64>(spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, it->second);
			}
			else
				return NULL;
		}
	case INDEX_STRING:
		is = stringMaps.find(spec->key);
		if (is != stringMaps.end()) {
			std::string start = spec->startString;
			std::string end = spec->endString;
			if (start.size() && end.size())
				return queryMessages<std::string>(start, end, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, is->second);
			else if (spec->startTime)
				return queryMessages<std::string>(start, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, is->second);
			else
				return queryMessages<std::string>(spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, is->second);
		}
		else
			return NULL;
	case INDEX_INTEGER:
		ii = integerMaps.find(spec->key);
		if (ii != integerMaps.end()) {
		//	printf("Querying for key '%s' (of %u)...\n", spec->key, ii->second.size());
			if ((spec->startInt != INT64_NOVALUE) && (spec->endInt != INT64_NOVALUE))
				return queryMessages<int64>(spec->startInt, spec->endInt, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, ii->second);
			else if (spec->startInt != INT64_NOVALUE)
				return queryMessages<int64>(spec->startInt, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, ii->second);
			else
				return queryMessages<int64>(spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, ii->second);
		}
		else
			return NULL;
	case INDEX_FLOAT:
		ifl = floatMaps.find(spec->key);
		if (ifl != floatMaps.end()) {
			if ((spec->startFloat != FLOAT64_NOVALUE) && (spec->endFloat != FLOAT64_NOVALUE))
				return queryMessages<float64>(spec->startFloat, spec->endFloat, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, ifl->second);
			else if (spec->startFloat != FLOAT64_NOVALUE)
				return queryMessages<float64>(spec->startFloat, spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, ifl->second);
			else
				return queryMessages<float64>(spec->from, spec->to, spec->type, spec->maxcount, spec->maxage, size, count, ifl->second);
		}
		else
			return NULL;
	default:
		return NULL;
	}
}









bool MessageIndex::doMaintenance() {
	uint64 now = lastMaintenance = GetTimeNow();

	std::map<std::string, std::multimap<uint64, DataMessage*> >::iterator it, et = timeMaps.end();
	std::map<std::string, std::multimap<std::string, DataMessage*> >::iterator is, es = stringMaps.end();
	std::map<std::string, std::multimap<int64, DataMessage*> >::iterator ii, ei = integerMaps.end();
	std::map<std::string, std::multimap<float64, DataMessage*> >::iterator ifl, efl = floatMaps.end();

	std::multimap<uint64, DataMessage*>::iterator i = eolMap.begin(), e = eolMap.end();
	while (i != e) {
		if (i->first < now) {
			// remove message from all other maps
			if (timeMap) {
				uint64 ctval = i->second->getCreatedTime();
				if (ctval)
					removeMessage<uint64>(i->second, *timeMap, ctval);
			}

			it = timeMaps.begin();
			while (it != et) {
				uint64 tval = i->second->getTime(it->first.c_str());
				if (tval)
					removeMessage<uint64>(i->second, it->second, tval);
				it++;
			}

			is = stringMaps.begin();
			while (is != es) {
				std::string sval = i->second->getString(is->first.c_str());
				removeMessage<std::string>(i->second, is->second, sval);
				is++;
			}

			ii = integerMaps.begin();
			while (ii != ei) {
				int64 ival;
				if (i->second->getInt(ii->first.c_str(), ival))
					removeMessage<int64>(i->second, ii->second, ival);
				ii++;
			}

			ifl = floatMaps.begin();
			while (ifl != efl) {
				float64 fval;
				if (i->second->getDouble(ifl->first.c_str(), fval))
					removeMessage<float64>(i->second, ifl->second, fval);
				ifl++;
			}
			// delete message itself
			delete(i->second);
			// remove entry
			eolMap.erase(i++);
		}
		else
			break;
	}

	return true;
}











bool MessageIndex::addMessageToTime(DataMessage* msg) {
	uint64 val;
	std::map<std::string, std::multimap<uint64, DataMessage*> >::iterator i = timeMaps.begin(), e = timeMaps.end();
	while (i != e) {
		if (val = msg->getTime(i->first.c_str()))
			i->second.insert(Time_Pair(val, msg));
		i++;
	}
	return true;
}

bool MessageIndex::addMessageToString(DataMessage* msg) {
	const char* val;
	std::map<std::string, std::multimap<std::string, DataMessage*> >::iterator i = stringMaps.begin(), e = stringMaps.end();
	while (i != e) {
		if (val = msg->getString(i->first.c_str()))
			i->second.insert(String_Pair(val, msg));
		i++;
	}
	return true;
}

bool MessageIndex::addMessageToInteger(DataMessage* msg) {
	int64 val;
	std::map<std::string, std::multimap<int64, DataMessage*> >::iterator i = integerMaps.begin(), e = integerMaps.end();
	while (i != e) {
		if (msg->getInt(i->first.c_str(), val)) {
			i->second.insert(Int_Pair(val, msg));
		//	printf("------------------- WB Added %lld to key '%s'   %s\n", val, i->first.c_str(), PrintTimeString(msg->getCreatedTime()).c_str());
		}
		i++;
	}
	return true;
}

bool MessageIndex::addMessageToFloat(DataMessage* msg) {
	float64 val;
	std::map<std::string, std::multimap<float64, DataMessage*> >::iterator i = floatMaps.begin(), e = floatMaps.end();
	while (i != e) {
		if (msg->getDouble(i->first.c_str(), val))
			i->second.insert(Float_Pair(val, msg));
		i++;
	}
	return true;
}




template <typename T>
bool MessageIndex::deleteIndexKey(const char* key, std::map<std::string, std::multimap<T, DataMessage*> > &maps) {

	std::multimap<T, DataMessage*>* map = &(maps[key]);

	typename std::multimap<T, DataMessage*>::iterator i = map->begin(), e = map->end();
	while (i != e) {
		delete(i->second);
		i++;
	}
	maps.erase(key);
	return true;
}

template <typename T>
bool MessageIndex::removeMessage(DataMessage* msg, std::multimap<T, DataMessage*> &map, T &val) {
	typename std::multimap<T, DataMessage*>::iterator i, e = map.end();
	i = map.lower_bound(val);
	while ((i != e) && (i->first == val)) {
		if (((void*)msg) == ((void*)i->second))
			map.erase(i++);
		else
			i++;
	}
	return true;
}

template <typename T>
char* MessageIndex::queryMessages(T start, T end, uint32 from, uint32 to, PsyType &type, uint32 maxcount, uint64 maxage, uint32 &size, uint32 &count, std::multimap<T, DataMessage*> &map) {
	count = size = 0;
	typename std::multimap<T, DataMessage*>::iterator i1, i2, e = map.end();

	uint64 now = GetTimeNow();
	uint64 createdAfter = 0;
	if (maxage)
		createdAfter = now - maxage;

	if ( (i1 = map.lower_bound(start)) == e) {
	//	printf("queryMessages found (%u) results (of %u)...\n", count, map.size());
		return NULL;
	}
	i2 = map.upper_bound(end);

	uint32 allocSize = 4096;
	char* result = (char*) malloc(allocSize);

	DataMessage* msg;
	uint32 msgSize;
	while ((i1 != i2) && (i1 != e)) {
		if (msg = i1->second) {
			// Filters...
			if (msg->getEOL() < now) {
				i1++;
				continue;
			}
			if (from && (from != msg->getFrom())) {
				i1++;
				continue;
			}
			if (to && (to != msg->getTo())) {
				i1++;
				continue;
			}
			if ((type != NOTYPE) && (!msg->getType().matches(type))) {
				i1++;
				continue;
			}
			if (maxcount && (count >= maxcount))
				break;
			if (createdAfter && (msg->getCreatedTime() < createdAfter)) {
				i1++;
				continue;
			}

			msgSize = i1->second->getSize();
			while (size + msgSize > allocSize) {
				allocSize *= 2;
				void* newmem = realloc(result, allocSize);
				if (!newmem) {
					free(result);
					count = size = 0;
					return NULL;
				}
				result = (char*) newmem;
			}
			memcpy(result+size, msg->data, msgSize);
			size += msgSize;
			count++;
		}
		i1++;
	}
	// printf("queryMessages found %u results (of %u)...\n", count, map.size());
	return result;
}

template <typename T>
char* MessageIndex::queryMessages(T start, uint32 from, uint32 to, PsyType &type, uint32 maxcount, uint64 maxage, uint32 &size, uint32 &count, std::multimap<T, DataMessage*> &map) {
	count = size = 0;
	typename std::multimap<T, DataMessage*>::iterator i1, i2, e = map.end();

	uint64 now = GetTimeNow();
	uint64 createdAfter = 0;
	if (maxage)
		createdAfter = now - maxage;

	if ( (i1 = map.lower_bound(start)) == e)
		return NULL;

	uint32 allocSize = 4096;
	char* result = (char*) malloc(allocSize);

	DataMessage* msg;
	uint32 msgSize;
	while (i1 != e) {
		if (msg = i1->second) {
			// Filters...
			if (msg->getEOL() < now) {
				i1++;
				continue;
			}
			if (from && (from != msg->getFrom())) {
				i1++;
				continue;
			}
			if (to && (to != msg->getTo())) {
				i1++;
				continue;
			}
			if ((type != NOTYPE) && (!msg->getType().matches(type))) {
				i1++;
				continue;
			}
			if (maxcount && (count >= maxcount))
				break;
			if (createdAfter && (msg->getCreatedTime() < createdAfter)) {
				i1++;
				continue;
			}

			msgSize = i1->second->getSize();
			while (size + msgSize > allocSize) {
				allocSize *= 2;
				void* newmem = realloc(result, allocSize);
				if (!newmem) {
					free(result);
					count = size = 0;
					return NULL;
				}
				result = (char*) newmem;
			}
			memcpy(result+size, msg->data, msgSize);
			size += msgSize;
			count++;
		}
		i1++;
	}
//	printf("queryMessages found %u results...\n", count);
	return result;
}

template <typename T>
char* MessageIndex::queryMessages(uint32 from, uint32 to, PsyType &type, uint32 maxcount, uint64 maxage, uint32 &size, uint32 &count, std::multimap<T, DataMessage*> &map) {
	count = size = 0;
	typename std::multimap<T, DataMessage*>::reverse_iterator i1, e = map.rend();

	uint64 now = GetTimeNow();
	uint64 createdAfter = 0;
	if (maxage)
		createdAfter = now - maxage;

	if ( (i1 = map.rbegin()) == e) {
		// printf("queryMessages found (%u) results (of %u)...\n", count, map.size());
		return NULL;
	}

	uint32 allocSize = 4096;
	char* result = (char*) malloc(allocSize);

	DataMessage* msg;
	uint32 msgSize;
	while (i1 != e) {
		if (msg = i1->second) {
			// Filters...
			if (msg->getTTL() && msg->getEOL() < now) {
				i1++;
				continue;
			}
			if (from && (from != msg->getFrom())) {
				i1++;
				continue;
			}
			if (to && (to != msg->getTo())) {
				i1++;
				continue;
			}
			if ((type != NOTYPE) && (!msg->getType().matches(type))) {
				i1++;
				continue;
			}
			if (maxcount && (count >= maxcount))
				break;
			if (createdAfter && (msg->getCreatedTime() < createdAfter)) {
				i1++;
				continue;
			}

			msgSize = i1->second->getSize();
			while (size + msgSize > allocSize) {
				allocSize *= 2;
				void* newmem = realloc(result, allocSize);
				if (!newmem) {
					free(result);
					count = size = 0;
					return NULL;
				}
				result = (char*) newmem;
			}
			memcpy(result+size, msg->data, msgSize);
			size += msgSize;
			count++;
		}
		i1++;
	}
	// printf("queryMessages found %u results (of %u)...\n", count, map.size());
	return result;
}


bool MessageIndex::UnitTest() {
	MessageIndex* index = new MessageIndex();

	index->addIndexKey("time", INDEX_TIME);

	uint32 n, size, count;
	int64 val;
	DataMessage* msg;
	char* result, *data;
	RetrieveSpec spec;

	uint64 now = GetTimeNow();
	for (n=0; n<100; n++) {
		msg = new DataMessage();
		msg->setCreatedTime(now - 99000 + (n*1000));
		msg->setTTL(1000000);
		msg->setInt("Count", n);
		index->addMessage(msg);
	}


	memset(&spec, 0, sizeof(RetrieveSpec));
	utils::strcpyavail(spec.key, "time", MAXKEYNAMELEN, true);
	spec.keytype = INDEX_TIME;
	spec.maxage = 50000;
	
	result = index->queryMessages(&spec, size, count);
	if ( (count < 40) || (count > 50) ) {
		LogPrint(0,LOG_MEMORY,0,"Index didn't return around 50 msgs as expected");
		free(result);
		delete(index);
		return false;
	}

	data = result;
	for (n=0; n<count; n++) {
		if (GetObjID(data) != DATAMESSAGEID) {
			free(result);
			delete(index);
			return false;
		}
		msg = new DataMessage(data, true);
		if (!msg->getInt("Count", val) || (val != 99 - n)) {
			free(result);
			delete(index);
			return false;
		}
		data += msg->getSize();
		delete(msg);
	}
	free(result);

	// **********************************************

	memset(&spec, 0, sizeof(RetrieveSpec));
	utils::strcpyavail(spec.key, "time", MAXKEYNAMELEN, true);
	spec.keytype = INDEX_TIME;
	spec.maxcount = 50;
	
	result = index->queryMessages(&spec, size, count);
	if (count != 50) {
		LogPrint(0,LOG_MEMORY,0,"Index didn't return 50 msgs as expected");
		free(result);
		delete(index);
		return false;
	}

	data = result;
	for (n=0; n<count; n++) {
		if (GetObjID(data) != DATAMESSAGEID) {
			free(result);
			delete(index);
			return false;
		}
		msg = new DataMessage(data, true);
		if (!msg->getInt("Count", val) || (val != 99 - n)) {
			free(result);
			delete(index);
			return false;
		}
		data += msg->getSize();
		delete(msg);
	}
	free(result);

	// **********************************************

	utils::Sleep(950);

	memset(&spec, 0, sizeof(RetrieveSpec));
	utils::strcpyavail(spec.key, "time", MAXKEYNAMELEN, true);
	spec.keytype = INDEX_TIME;
	spec.maxcount = 100;
	
	result = index->queryMessages(&spec, size, count);
	if (count > 50) {
		LogPrint(0,LOG_MEMORY,0,"Index didn't return 50 msgs as expected");
		free(result);
		delete(index);
		return false;
	}
	free(result);

	delete(index);
	return true;
}


} // namespace cmlabs
