module(..., package.seeall);

local table_helpers  = require "i6index.table_helpers"
local schema_helpers = require "i6index.schema_helpers"
local sockets = require "socket"
local utf8 = require "icu.utf8"

local cassandra = require "lua_cassandra"

--[[
	This module is responsible for the following tasks:
	* create and destroy the column families to persist documents defined in the
	  schema.
	* add and remove documents to the index.
--]]
--==============================================================================
-- true iff exists a column family named columnFamilyName in the current keyspace
local function existsColumnFamily(connection, columnFamilyName)
	local currentKeyspaceName = connection:GetCurrentKeyspaceName();
	local keyspaces = connection:DescribeKeyspaces();

	for _, keyspace in ipairs(keyspaces) do
		if keyspace.name == currentKeyspaceName then
			local columnFamilies = keyspace:DescribeColumnFamilies();
			for _, columnFamily in ipairs(columnFamilies) do
				if columnFamily.name == columnFamilyName then
					return true
				end
			end
		end
	end
	return false
end
--==============================================================================
local function existsKeyspace(connection, newKeyspace)
	local keyspaces = connection:DescribeKeyspaces();
	for _, keyspace in ipairs(keyspaces) do
		if keyspace.name == newKeyspace then
			return true
		end
	end
	return false
end
--==============================================================================
-- it returns the composite indexes names that results from the permutations
-- of the names with a given size
local function getCompositeIndexesPermutations(names, size, init)
	init = init or 1
	if size == 0 then return {{}} end
	local result = {}
	for i = init, #names - size + 1 do
		local deeper = getCompositeIndexesPermutations(names, size - 1, i + 1)
		for _, d in ipairs(deeper) do
			table.insert(result, { names[i], unpack(d) })
		end
	end
	return result
end

local function filterCompositeIndexes(compositeIndexes, forbiddenComposites)
	if forbiddenComposites == nil then
		return compositeIndexes
	end

	local filteredComposites = {}
	for _, composite in ipairs(compositeIndexes) do
		if not schema_helpers.isForbiddenComposite(composite, forbiddenComposites) then
			table.insert(filteredComposites, composite)
		end
	end
	return filteredComposites
end

function getCompositeIndexes(documentDefinition, level)
	local names = {}
	for name in pairs(documentDefinition.indexes) do table.insert(names, name) end
	table.sort(names)
	return filterCompositeIndexes(getCompositeIndexesPermutations(names, level),
	                              documentDefinition.forbiddenComposites)
end

--==============================================================================
local function createColumnFamiliesForDocument(connection, documentDefinition)

	local documentName = documentDefinition.name
	if not documentName then
		return nil, "document has no name"
	end

	if not documentDefinition.indexes then
		return nil, "documentDefinition has no indexes"
	end

	local compositeIndexLevel = documentDefinition.compositeIndexLevel or 1

	if not documentDefinition.sortField then
		return nil, "documentDefinition has no sortField"
	end

	local sortFieldColumnFamily     = schema_helpers.getCFNameFromFields(documentName, documentDefinition.sortField)
	local directIndexColumnFamily   = schema_helpers.getCFNameDirectIndex(documentName)
	local detailsIndexColumnFamily  = schema_helpers.getCFNameDetailsIndex(documentName)
	local documentTermsColumnFamily = schema_helpers.getCFNameTerms(documentName)

	if not existsColumnFamily(connection, detailsIndexColumnFamily) then
		connection:CreateColumnFamily(detailsIndexColumnFamily, {comparator_type = "UTF8Type", default_validation_class = "UTF8Type"})
	end

	if not existsColumnFamily(connection, sortFieldColumnFamily) then
		connection:CreateColumnFamily(sortFieldColumnFamily, {comparator_type = "CompositeType(UTF8Type, UTF8Type)", default_validation_class = "UTF8Type"})
	end

	if not existsColumnFamily(connection, directIndexColumnFamily) then
		connection:CreateColumnFamily(directIndexColumnFamily, {comparator_type = "CompositeType(UTF8Type, UTF8Type)"})
	end

	if not existsColumnFamily(connection, documentTermsColumnFamily) then
		connection:CreateColumnFamily(documentTermsColumnFamily, {comparator_type = "CompositeType(UTF8Type, UTF8Type)", default_validation_class = "CounterColumnType", replicate_on_write = true})
	end

	for indexName, indexType in pairs(documentDefinition.indexes) do
		local columnFamilyName = schema_helpers.getCFNameFromFields(documentDefinition.name, indexName)
		if not existsColumnFamily(connection, columnFamilyName) then
			local column_type = "CompositeType(UTF8Type, UTF8Type, UTF8Type)"
			connection:CreateColumnFamily(columnFamilyName, {comparator_type = column_type, default_validation_class = "UTF8Type"})
		end
	end

	for i = 2, compositeIndexLevel do
		local compositeIndexes, message = getCompositeIndexes(documentDefinition, compositeIndexLevel)
		if not compositeIndexes then
			return nil, message
		end

		local column_type="CompositeType(UTF8Type"
		for j = 2, i do
			column_type = column_type .. ", UTF8Type"
		end
		column_type = column_type .. ", UTF8Type, UTF8Type)"

		for _, compositeIndex in pairs(compositeIndexes) do
			local columnFamilyName = schema_helpers.getCFNameFromFields(documentDefinition.name, compositeIndex)
			if not existsColumnFamily(connection, columnFamilyName) then
				connection:CreateColumnFamily(columnFamilyName, {comparator_type = column_type, default_validation_class = "UTF8Type"})
			end
		end
	end

	return true
end
--==============================================================================
local function createColumnFamilies(connection, schema)
	if not schema.definitions then
		return nil, "there is no definitions in schema"
	end
	for _, documentDefinition in pairs(schema.definitions) do
		local success, message = createColumnFamiliesForDocument(connection, documentDefinition)
		if not success then
			return nil, "could not create column families for " .. documentDefinition.name .. " cause: " .. message
		end
	end
	return true
end

--==============================================================================
local function normalizeToken(token)
	return schema_helpers.trim(icu.utf8.lower(string.sub(tostring(token), 1, 128)))
end
--==============================================================================
-- returns tokens of the field/indexName in document to add
local function getIndexValues(documentDefinition, indexName, document)
	local indexType = documentDefinition.indexes[indexName]

	if not indexType then
		return nil, "index: "..indexName .. " does not exist in document: "  .. documentDefinition.name
	end

	local values = {}
	if indexType.multiValued then
		local getMultiValues = documentDefinition["get" .. schema_helpers.capitalizeFirst(indexName)]
		if not getMultiValues then
			getMultiValues = function (document) return document[indexName] end
		end
		local rawvalues = getMultiValues(document) or {}
		for _, value in pairs(rawvalues) do
			if schema_helpers.isValidAsValue(value) then
				table.insert(values, normalizeToken(value))
			end
		end
	else
		if schema_helpers.isValidAsValue(document[indexName]) then
			table.insert(values, normalizeToken(document[indexName]))
		end
	end

	local tokens
	if indexType.tokenized == true then
		local tokenizer = documentDefinition[indexName .. "Tokenizer"]

		if not tokenizer then
			return nil, "tokenizer for "..indexName .. " does not exists in document: "  .. documentDefinition.name
		end

		tokens = {}
		for _, value in pairs(values) do
			for _, token in pairs(tokenizer(value)) do
				table.insert(tokens, token)
			end
		end
	else
		tokens = values
	end

	return tokens
end

--==============================================================================
-- given a document returns the tokens to index in a composite index
local function getCompositeValues(documentDefinition, compositeIndex, document)
	local values = {}
	for _, indexName in pairs(compositeIndex) do
		local tokens, message = getIndexValues(documentDefinition, indexName, document)
		if not tokens then
			return nil, message
		end
		values = table_helpers.cartesianProduct(values, tokens)
	end
	return values
end

--==============================================================================
function create(schema)
	if not schema.keyspace then
		return nil, "there is no keyspace in schema"
	end

	local connection = cassandra:CassandraConnection("system")
	if not existsKeyspace(connection, schema.keyspace) then
		connection:CreateKeyspace(schema.keyspace)
	end
	connection:Close()

	connection = cassandra:CassandraConnection(schema.keyspace)
	local success, message = createColumnFamilies(connection, schema)
	connection:Close()

	if not success then
		return nil, message
	end

	return true
end
--==============================================================================
function drop(schema)
	if not schema.keyspace then
		return nil, "there is no keyspace in schema"
	end

	local connection = cassandra:CassandraConnection("system")
	for i = 1, 5 do
		if existsKeyspace(connection, schema.keyspace) then
			connection:DropKeyspace(schema.keyspace)
			-- TODO this is awful but we don't have any mechanism
			-- to wait for the spread of changes across the cluster
			socket.sleep(5)
		else
			break
		end
	end

	if existsKeyspace(connection, schema.keyspace) then
		return nil, "keyspace: " .. schema.keyspace .. " failed to drop"
	end
	connection:Close()
	return true
end

local function removeFromReverseIndex(connection, documentName, campaign, id)
	local last_name  = ""
	local block_size = 100
	local range_size = block_size
	local documentTermsCFName = schema_helpers.getCFNameTerms(documentName)
	local directIndexCFName   = schema_helpers.getCFNameDirectIndex(documentName)
	local ignoreFirst = false

	while block_size == range_size do
		range_size = 0
		local range = connection:GetSliceRange(id, directIndexCFName, "", {
		                                                column_names = {},
		                                                slice_range = {
		                                                    start  = last_name,
		                                                    finish = "",
		                                                    reversed = false,
		                                                    count = block_size
		                                                }
		                                            });
		for _, c in pairs(range) do
			range_size = range_size + 1
			if not ignoreFirst then
				local nameComposite = c.name_as_composite
				local indexCF  = nameComposite:GetString()
				if indexCF == documentTermsCFName then
					connection:IncrementColumnCounter(campaign, documentTermsCFName, c.value, -1)
					local counterValue = connection:GetCounterValue(campaign, documentTermsCFName, c.value)
					if counterValue ~= nil and counterValue <= 0  then
						connection:RemoveColumnCounter(campaign, documentTermsCFName, c.value)
					end
				else
					connection:RemoveColumn(campaign, indexCF, c.value)
				end
			else
				ignoreFirst = false
			end
			last_name = c.name
		end
		ignoreFirst = true
	end
end

--==============================================================================
function remove(connection, schema, documentName, id)
	local documentDefinition = schema.definitions[documentName]
	if not documentDefinition then
		return nil, "document ".. documentName .. " does not exists in schema"
	end

	local directIndexCF  = schema_helpers.getCFNameDirectIndex(documentName)
	local detailsIndexCF = schema_helpers.getCFNameDetailsIndex(documentName)
	local campaign = connection:GetColumnValue(id, detailsIndexCF, "campaign")
	if not campaign then
		return nil, "could not find campaign for document " .. id .. " of type " .. documentName
	end
	removeFromReverseIndex(connection, documentName, campaign, id)

	connection:RemoveColumn(id, directIndexCF)

	connection:RemoveColumn(id, detailsIndexCF)

	return true
end

--==============================================================================
function add(connection, schema, documentName, document, campaign, id)
	local documentDefinition = schema.definitions[documentName]
	if not documentDefinition then
		return nil, "document ".. documentName .. " does not exists in schema"
	end

	if not schema_helpers.isValidAsValue(campaign) then
		return nil, "empty campaign not allowed"
	end

	if not schema_helpers.isValidAsValue(id) then
		return nil, "empty id not allowed"
	end

	if not schema.keyspace then
		return nil, "there is no keyspace in schema"
	end

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

	local sortFieldValue, message = schema_helpers.getSortFieldValue(documentDefinition, document)
	if not sortFieldValue then
		return nil, message
	end

	local unique  = id

	local inserts = {}

	local directIndexColumnFamily   = schema_helpers.getCFNameDirectIndex(documentName)
	local detailsIndexColumnFamily  = schema_helpers.getCFNameDetailsIndex(documentName)
	local documentTermsColumnFamily = schema_helpers.getCFNameTerms(documentName)
	local bySortFieldColumnFamily   = schema_helpers.getCFNameFromFields(documentName, sortField)

	table.insert(inserts, {column_family    = detailsIndexColumnFamily,
	                           column_key   = unique,
	                           column_name  = "campaign",
	                           column_value = campaign});

	local compositeColumnName = cassandra.compose(sortFieldValue, unique)

	table.insert(inserts, {column_family    = bySortFieldColumnFamily,
	                           column_key   = campaign,
	                           column_name  = compositeColumnName,
	                           column_value = unique});

	table.insert(inserts, {column_family    = directIndexColumnFamily,
	                           column_key   = unique,
	                           column_name  = cassandra.compose(bySortFieldColumnFamily, ""),
	                           column_value = compositeColumnName});

	connection:IncrementColumnCounter(campaign, documentTermsColumnFamily,
	                           cassandra.compose(sortField, sortFieldValue))

	table.insert(inserts, {column_family = directIndexColumnFamily,
	                           column_key   = unique,
	                           column_name  = cassandra.compose(documentTermsColumnFamily, ""),
	                           column_value = cassandra.compose(sortField, sortFieldValue)});

	for indexName in pairs(documentDefinition.indexes) do
		local columnFamily = schema_helpers.getCFNameFromFields(documentName, indexName)
		local tokens, message = getIndexValues(documentDefinition, indexName, document)
		if not tokens then
			return nil, message
		end

		local entriesInColumnFamily = 1
		for _, token in pairs(tokens) do
			local tokenLower = normalizeToken(token)
			local compositeColumnName = cassandra.compose(tokenLower, sortFieldValue, unique)
			--reverse index entry
			table.insert(inserts, {column_family = columnFamily,
			                           column_key   = campaign,
			                           column_name  = compositeColumnName,
			                           column_value = unique})


			table.insert(inserts, {column_family = directIndexColumnFamily,
			                           column_key   = unique,
			                           column_name  = cassandra.compose(columnFamily,
			                                          tostring(entriesInColumnFamily)),
			                           column_value = compositeColumnName})

			local counterColumnName = cassandra.compose(indexName, token)
			connection:IncrementColumnCounter(campaign, documentTermsColumnFamily,
			                                  counterColumnName)

			table.insert(inserts, {column_family = directIndexColumnFamily,
			                           column_key   = unique,
			                           column_name  = cassandra.compose(documentTermsColumnFamily,
			                                          columnFamily .. tostring(entriesInColumnFamily)),
			                           column_value = counterColumnName})


			entriesInColumnFamily = entriesInColumnFamily + 1
			if (#inserts > 200) then
				connection:BatchInsert(inserts)
				inserts = {}
			end
		end
	end

	local compositeIndexLevel = documentDefinition.compositeIndexLevel or 1

	for i = 2, compositeIndexLevel do
		local compositeIndexes, message = getCompositeIndexes(documentDefinition, i)
		if not compositeIndexes then
			return nil, message
		end

		for _, compositeIndex in pairs(compositeIndexes) do
			local compositeValues, message = getCompositeValues(documentDefinition, compositeIndex, document)
			if not compositeValues then
				return nil, message
			end
			local columnFamily = schema_helpers.getCFNameFromFields(documentName, compositeIndex)

			local entriesInColumnFamily = 1
			for _, compositeValue in pairs(compositeValues) do
				table.insert(compositeValue, sortFieldValue)
				table.insert(compositeValue, unique)
				local compositeColumnName = cassandra.compose(unpack(compositeValue))

				table.insert(inserts, {column_family = columnFamily,
				                       column_key    = campaign,
				                       column_name   = compositeColumnName,
				                       column_value  = unique});

				table.insert(inserts, {column_family = directIndexColumnFamily,
				                       column_key    = unique,
				                       column_name   = cassandra.compose(columnFamily,
				                                       tostring(entriesInColumnFamily)),
				                       column_value  = compositeColumnName});

				if (#inserts > 200) then
					connection:BatchInsert(inserts)
					inserts = {}
				end

				entriesInColumnFamily = entriesInColumnFamily + 1
			end
		end
	end

	connection:BatchInsert(inserts)

	return true
end

