module(..., package.seeall);
local resultset             = require "i6index.resultset"
local table_helpers         = require "i6index.table_helpers"
local schema_helpers        = require "i6index.schema_helpers"
local term_matching_helpers = require "i6index.term_matching_helpers"
local cassandra             = require "lua_cassandra"
local utf8                  = require "icu.utf8"

local lastUtf8Character  = string.char(0xf4, 0x80, 0x8f, 0xbf)
local firstUtf8Character = string.char(0)

--[[
     This module implements index's queries
--]]

--#############################################################################
--retruns at most maxDocuments documents in campaign, starting from lastHandle
-- * documentName must be a document name in the schema
--
-- * the return resultset is sorted in ascending lexicographical order, if reverseOrder is false,
--   else it will be sorted in descending lexicographical order
local function getDocumentsSingleCampaign(connection, schema, documentName, campaign, rangeStart,
                                          rangeEnd, lastHandle, maxDocuments, reverseOrder)

	local sortField = schema.definitions[documentName].sortField
	local rangeColumnCount = maxDocuments
	local startColumn = lastHandle

	if lastHandle ~= "" then
		rangeColumnCount = maxDocuments + 1
	else
		local startValues = {}
		if rangeStart ~= "" then
			table.insert(startValues, rangeStart)
			if reverseOrder then
				table.insert(startValues, lastUtf8Character)
			else
				table.insert(startValues, firstUtf8Character)
			end
		end
		startColumn = cassandra.compose(unpack(startValues))
	end

	local finishValues = {}
	if rangeEnd ~= "" then
		table.insert(finishValues, rangeEnd)
		if reverseOrder then
			table.insert(finishValues, firstUtf8Character)
		else
			table.insert(finishValues, lastUtf8Character)
		end
	end

	local finishColumn = cassandra.compose(unpack(finishValues))

	local byKeyColumnFamily = schema_helpers.getCFNameFromFields(documentName, sortField)

	local range = connection:GetSliceRange(campaign, byKeyColumnFamily, "", {
	                                            column_names = {},
	                                            slice_range  = {
	                                                start    = startColumn,
	                                                finish   = finishColumn,
	                                                reversed = reverseOrder,
	                                                count    = rangeColumnCount
	                                            }})
	local result  = resultset.new()

	for _, column in pairs(range) do
		if column.name ~= lastHandle then
			if result:length() < maxDocuments then
				result:append(column.name, column.value)
			end
		end
	end

	return result
end

--#############################################################################
-- same as getDocumentsSingleCampaign but for campaigns that is an array of campaigns
local function getDocumentsMultipleCampaigns(connection, schema, documentName, campaigns, rangeStart,
                                             rangeEnd, lastHandle, maxDocuments, reverseOrder)

	local comparatorFun = schema_helpers.getHandleComparator(reverseOrder, 0)

	local result = resultset.new()
	for _, campaign in pairs(campaigns) do
		local campaignResult = getDocumentsSingleCampaign(connection, schema, documentName,
		                                                  campaign, rangeStart, rangeEnd, lastHandle,
		                                                  maxDocuments, reverseOrder)

		result = campaignResult:union(result, comparatorFun, maxDocuments)
	end

	return result
end

--#############################################################################
--returns at most maxDocuments where indexNames has indexValues.
--note that indexNames and IndexValues are correlatives, i.e. if we are searching for documents with field
-- "name" with value "heineken", then indexNames[i] should be "name" and indexValues[i] will be "heineken".
-- Also note that when the sortfield is used for querying that name and value should be the last element
-- of those tables/arrays.
local function searchDocumentsSingleCampaign(connection, schema, documentName, campaign, indexNames,
                                             indexValues, rangeStart, rangeEnd, lastHandle, maxDocuments,
                                             reverseOrder)

	local sortField = schema.definitions[documentName].sortField

	local filteredIndexNames  = table_helpers.copy(indexNames)
	local filteredIndexValues = table_helpers.copy(indexValues)

	-- sortField is removed from indexname
	local sortFieldPresent = false
	for i = 1, #filteredIndexNames do
		if filteredIndexNames[i] == sortField then
			table.remove(filteredIndexNames,  i)
			sortFieldPresent = true
			break
		end
	end

	local columnFamily

	--special case, when querying by sortfield
	if #filteredIndexNames == 0 then
		columnFamily = schema_helpers.getCFNameFromFields(documentName, sortField)
	else
		columnFamily = schema_helpers.getCFNameFromFields(documentName, filteredIndexNames)
	end

	local startValues = table_helpers.copy(filteredIndexValues)

	local startColumn
	local lastSortFieldValue
	local lastUnique
	local rangeColumnCount = maxDocuments

	if lastHandle == "" then
		-- upper bounds of the slice to get
		if rangeStart ~= "" then
			-- if the query filters include the sortfield we should not
			--include the bounds because they always refer to the sortField
			if not sortFieldPresent then
				table.insert(startValues, rangeStart)
			end
		else
			if reverseOrder then
				table.insert(startValues, lastUtf8Character)
			else
				table.insert(startValues, firstUtf8Character)
			end
		end
	else
		--if lastHandle is not nil we have to remove it from the resultset
		-- thus we need to get one more document from GetSliceRange
		rangeColumnCount = maxDocuments + 1

		--lastHandle may refer to another matched term, but we will use the
		-- sortfield and the unique to build the startColumn of the
		-- GetSliceRange
		lastSortFieldValue, lastUnique = schema_helpers.getSortFieldAndUnique(lastHandle, #filteredIndexNames)
		if not sortFieldPresent then
			table.insert(startValues, lastSortFieldValue)
		end
		table.insert(startValues, lastUnique)
	end
	startColumn = cassandra.compose(unpack(startValues))

	local finishValues = table_helpers.copy(filteredIndexValues)
	if rangeEnd ~= "" then
		if not sortFieldPresent then
			table.insert(finishValues, rangeEnd)
		end
	else
		if reverseOrder then
			table.insert(finishValues, firstUtf8Character)
		else
			table.insert(finishValues, lastUtf8Character)
		end
	end
	local finishColumn = cassandra.compose(unpack(finishValues))

	local range = connection:GetSliceRange(campaign, columnFamily, "", {
	                                            column_names = {},
	                                            slice_range  = {
	                                                start    = startColumn,
	                                                finish   = finishColumn,
	                                                reversed = reverseOrder,
	                                                count    = rangeColumnCount
	                                            }})

	local result = resultset.new()
	local mustCheckBound = true
	local includeThisResult = true
	for _, column in pairs(range) do
		if mustCheckBound then
			--check the lower bound only with the first column of the range
			mustCheckBound = false
			local currentSortField, currentUnique = schema_helpers.getSortFieldAndUnique(column.name, #filteredIndexNames)
			if (currentSortField == lastSortFieldValue and lastUnique == currentUnique) then
				-- if it's equal we assume that the caller already have the document id
				includeThisResult = false
			else
				includeThisResult = true
			end
		else
			includeThisResult = true
		end

		if includeThisResult then
			if result:length() < maxDocuments then
				result:append(column.name, column.value)
			else
				break
			end
		end
	end

	return result
end

--#############################################################################
--same as searchDocumentsSingleCampaign but for multiple campaigns (duh!)
local function searchDocumentsMultipleCampaign(connection, schema, documentName, campaigns,
	                                indexNames, indexValues, rangeStart, rangeEnd, lastHandle,
	                                maxDocuments, reverseOrder)

	local comparatorFun = schema_helpers.getHandleComparator(reverseOrder, #indexValues)

	local partialResult = resultset.new()
	for _, campaign in pairs(campaigns) do
		local campaignResult, message = searchDocumentsSingleCampaign(connection, schema,
		        documentName, campaign, indexNames, indexValues, rangeStart, rangeEnd, lastHandle,
		        maxDocuments, reverseOrder)

		if not campaignResult then
			return nil, message
		end

		partialResult = campaignResult:union(partialResult, comparatorFun, maxDocuments)
	end

	return partialResult
end

--##############################################################################
-- This function given a dictionary with keys the field names and values the tokens
-- we want to query it returns the composite index name and value that we will
-- have to query to get the desired result. Example
-- fiels = {type = {"CHAT"}, address={"deborah.meltrozzo@fing.edu.uy"}
-- returns
-- IndexNames = {"address", "type"}, IndexValues={"CHAT", "deborah.meltrozzo@fing.edu.uy"}
-- notice that it's not the actual name of the index but has the correct order to
-- easily construct the actual name, see getCFName* in schema_helpers
local function computeCompositeIndexAndQuery(schema, documentName, fields)
	local documentDefinition = schema.definitions[documentName]
	local compositeIndexLevel = documentDefinition.compositeIndexLevel or 1
	local sortField = documentDefinition.sortField

	--if he is querying by the sortField we have to act accordingly because
	-- there will not be one index to query only by sortfield (the data is
	-- allways combined with other indexes)
	local fieldNames = {}
	local sortFieldPresent = false
	for fieldName in pairs(fields) do
		if field ~= sortField  then
			table.insert(fieldNames, fieldName)
		else
			sortFieldPresent = true
		end
	end

	if compositeIndexLevel < #fieldNames then
		return nil, nil, "there is no composite index to satisfy search criteria"
	end

	table.sort(fieldNames)

	local compositeIndexName  = {}
	local compositeIndexValue = {}

	for i = 1, #fieldNames do
		local fieldName = fieldNames[i]
		table.insert(compositeIndexValue, fields[fieldName][1])
		table.insert(compositeIndexName,  fieldName)
	end
	-- sortfield value is allways the last value of the compositeIndexValues array
	-- to construct the startColumn (token_1:token_2:..:token_n:sortField:unique)
	-- notice that the function searchDocumentsSingleCampaing is "sortfield" aware
	-- when processing the query
	if sortFieldPresent then
		table.insert(compositeIndexValue, fields[sortField][1])
	end

	if not schema_helpers.isForbiddenComposite(compositeIndexName, documentDefinition.forbiddenComposites) then
		return compositeIndexName, compositeIndexValue
	else
		return nil, nil, "there is no valid composite index to satisfy search criteria"
	end
end

local function lowerizeQueryFields(fields)
	local lowerCapsFields = {}
	if fields then
		for field, values in pairs(fields) do
			local lowerValues = {}
			for _, value in pairs(values) do
				table.insert(lowerValues, icu.utf8.lower(value))
			end
			lowerCapsFields[field] = lowerValues
		end
	end
	return lowerCapsFields
end

local function validateQueryFields(fields)
	if fields then
		for field, values in pairs(fields) do
			if #values == 0 then
				return nil, "illegal operation: there is no value for field: " .. tostring(field)
			end
		end
	end
	return true
end

local function lowerizeCampaigns(campaigns)
	local lowerCapsCampaigns = {}
	for _, campaign in ipairs(campaigns) do
		table.insert(lowerCapsCampaigns, icu.utf8.lower(campaign))
	end
	return lowerCapsCampaigns
end

local function normalizeQueryTable(schema, queryTable)

	local documentName = queryTable.documentName
	if not documentName then
		return nil, "there is no documentName in queryTable"
	end

	local campaigns = queryTable.campaigns
	if not campaigns then
		return nil, "there is no campaigns in queryTable"
	end

	local documentDefinition = schema.definitions[documentName]
	if not documentDefinition then
		return nil, "there is no " .. documentName .. " in schema"
	end

	local sortFieldName = documentDefinition.sortField
	if not sortFieldName then
		return nil, "there is no sortField in " .. documentName
	end

	queryTable.campaigns = lowerizeCampaigns(campaigns)

	local valid, message = validateQueryFields(queryTable.fields)
	if not valid then
		return nil, message
	end

	-- if sortField is present in the fields table, it should be trimmed to
	-- be coherent with the add document function
	if queryTable.fields and queryTable.fields[sortFieldName]  then
		queryTable.fields[sortFieldName][1] = schema_helpers.trim(queryTable.fields[sortFieldName][1])
	end

	queryTable.fields = lowerizeQueryFields(queryTable.fields)

	return true
end

--#############################################################################
--returns at most maxDocuments with fields matching prefixes given in fields parameter
local function queryPrefix(connection, schema, documentName, campaigns, fields, rangeStart, rangeEnd,
	                        lastHandle, maxDocuments, reverseOrder)

	local fieldsIsEmpty = table_helpers.isEmpty(fields)

	local indexNames, prefixes, message = computeCompositeIndexAndQuery(schema, documentName, fields)
	if not indexNames then
		return nil, tostring(message)
	end

	local comparatorFun = schema_helpers.getHandleComparator(reverseOrder, #indexNames)

	local result = resultset.new()
	for _, campaign in pairs(campaigns) do

		local allMatches = {}
		for i = 1, #prefixes do
			local indexName = indexNames[i]
			local prefix    = prefixes[i]

			local matches = term_matching_helpers.getMatchedTerms(connection, schema, documentName,
			                                                        campaign, indexName, prefix)
			table.insert(allMatches, matches)
		end

		local permutations, message = term_matching_helpers.makeMatchedTermPermutations(allMatches)
		if message then
			return nil, message
		end

		for _, indexValues in pairs(permutations) do
			local partialResult, message = searchDocumentsSingleCampaign(connection, schema, documentName,
			                                    campaign, indexNames, indexValues, rangeStart, rangeEnd,
			                                    lastHandle, maxDocuments, reverseOrder)
			if message then
				return nil, message
			end
			result = partialResult:union(result, comparatorFun, maxDocuments)
		end
	end
	return result
end

--#############################################################################
-- process single valued fields queries, with or without prefixes
local function querySingleValuedFields(connection, schema, documentName, campaigns, fields,
	                     rangeStart, rangeEnd, lastHandle, maxDocuments,
	                     reverseOrder, prefixMatches)
	if prefixMatches then
		return queryPrefix(connection, schema, documentName, campaigns, fields, rangeStart, rangeEnd,
	                       lastHandle, maxDocuments, reverseOrder)
	else
		local indexNames, indexValues, message = computeCompositeIndexAndQuery(schema, documentName, fields)
		if not indexNames then
			return nil, tostring(message)
		end

		return searchDocumentsMultipleCampaign(connection, schema, documentName, campaigns,
		                                       indexNames, indexValues, rangeStart, rangeEnd, lastHandle,
		                                       maxDocuments, reverseOrder)
	end
end


--#############################################################################
--This functions splits fields that have more than one value from the single valued fields.
local function splitMultiValuedFields(fields)
	local singleFields = {}
	local multiFields  = {}
	for field, fieldValues in pairs(fields) do
		if #fieldValues > 1 then
			multiFields[field]  = fieldValues
		else
			singleFields[field] = fieldValues
		end
	end
	return multiFields, singleFields
end

local function mergeFields(singleFields, multiValuedFieldsCombination)
	local fields = {}
	for field, fieldValues in pairs(singleFields) do                 fields[field] = fieldValues end
	for field, fieldValues in pairs(multiValuedFieldsCombination) do fields[field] = fieldValues end
	return fields
end

--#############################################################################
--this two functions are used to compute the combinations of the values of the
-- multivalued fields
local function initMultiValuedFieldsIndex(multiValuedFields)
	local idx = {}
	for field, fieldValues in pairs(multiValuedFields) do
		idx[field] = {current = 1, max = #fieldValues}
	end
	return idx
end

local function nextCombination(multiValuedFields, idx)
	local combination = {}
	local incNext = true
	for field, values in pairs(idx) do
		combination[field] = {multiValuedFields[field][values.current]}
		if incNext then
			if values.current == values.max then
				idx[field].current = 1
			else
				idx[field].current = idx[field].current + 1
				incNext = false
			end
		end
	end
	return combination
end
--#############################################################################
-- process queries with multivalued fields, computing all the combinations and using
-- querySingleValuedFields to resolve.
local function queryMultiValuedFields(connection, schema, documentName, campaigns, fields,
	                                  rangeStart, rangeEnd, lastHandle, maxDocuments, reverseOrder,
	                                  prefixMatches)

	local involvedFields = 0
	for field, _ in pairs(fields) do involvedFields = involvedFields + 1 end

	local multiFields, singleFields = splitMultiValuedFields(fields)

	local combinationsCount = 1
	for field, fieldValues in pairs(multiFields) do
		combinationsCount = combinationsCount * #fieldValues
	end

	local comparatorFun = schema_helpers.getHandleComparator(reverseOrder, involvedFields)
	local total = resultset.new()
	local idx = initMultiValuedFieldsIndex(multiFields)

	for i = 1, combinationsCount do
		local multiValuedFieldsCombination = nextCombination(multiFields, idx)
		local queryFields = mergeFields(singleFields, multiValuedFieldsCombination)

		local rs, errMessage = querySingleValuedFields(connection, schema, documentName, campaigns,
		                                               queryFields, rangeStart, rangeEnd,
		                                               lastHandle, maxDocuments, reverseOrder,
		                                               prefixMatches)
		if not rs then
			return nil, errMessage
		end

		total = total:union(rs, comparatorFun, maxDocuments)
	end

	return total
end

--#############################################################################
--returns at most maxDocuments documents that match *exactly* the values of fields
--function query(connection, schema, documentName, campaigns, fields, lastHandle, maxDocuments, orderString)
function query(connection, schema, queryTable)

	if not connection then return nil, "you must supply a connection to query" end
	if not schema      then return nil, "you must supply a schema to query"	 end
	if not queryTable then return nil, "you must supply a queryTable to query" end

	local valid, message = normalizeQueryTable(schema, queryTable)
	if not valid then
		return nil, message
	end

	local documentName  = queryTable["documentName"]
	local campaigns     = queryTable["campaigns"]
	local fields        = queryTable["fields"]
	local lastHandle    = queryTable["lastHandle"] or ""
	local maxDocuments  = queryTable["maxDocuments"]
	local orderString   = queryTable["orderString"]
	local prefixMatches = queryTable["prefixMatches"]
	local rangeStart    = queryTable["rangeStart"] or ""
	local rangeEnd      = queryTable["rangeEnd"]   or ""

	local sortField = schema.definitions[documentName].sortField

	local reverseOrder, message = schema_helpers.translateToReverseOrder(orderString)
	if message then
		return nil, message
	end

	rangeStart = icu.utf8.lower(rangeStart)
	rangeEnd   = icu.utf8.lower(rangeEnd)

	if rangeStart ~= "" and rangeEnd ~= "" then
		if rangeStart > rangeEnd then
			return resultset.new()
		end
	end

	-- if the filter includes the sortField we should check the bounds, if it's not within
	-- bounds the resultset is empty, else the bounds have no sense! so we will remove
	-- them
	if fields[sortField] then
		if rangeStart ~= "" then
			if fields[sortField][1] < rangeStart then
				return resultset.new()
			else
				rangeStart = ""
			end
		end
		if rangeEnd ~= "" then
			if rangeEnd < fields[sortField][1] then
				return resultset.new()
			else
				rangeEnd = ""
			end
		end
	end

	if reverseOrder then
		rangeStart, rangeEnd = rangeEnd, rangeStart
	end

	local multiValuedField = false
	for field, fieldValues in pairs(fields) do
		if #fieldValues > 1 then
			multiValuedField = true
			break
		end
	end

	local fieldsIsEmpty = table_helpers.isEmpty(fields)

	if fieldsIsEmpty then
		return getDocumentsMultipleCampaigns(connection, schema, documentName, campaigns,
		                                     rangeStart, rangeEnd, lastHandle, maxDocuments,
		                                     reverseOrder)
	else
		if multiValuedField then
			return queryMultiValuedFields(connection, schema, documentName, campaigns, fields,
			                 rangeStart, rangeEnd, lastHandle, maxDocuments, reverseOrder,
			                 prefixMatches)
		else
			return querySingleValuedFields(connection, schema, documentName, campaigns, fields,
			                 rangeStart, rangeEnd, lastHandle, maxDocuments, reverseOrder,
			                 prefixMatches)
		end
	end
end
