module(..., package.seeall)
local m_cassandra = require "lua_cassandra"
local m_table = require "helpers.table"

function Init (options)
	assert(type(options) == "table", "options must be a table")
	assert(options.cassandra, "options.cassandra is missing")
	assert(options.cassandra.host, "missing options.cassandra.host field")
	assert(options.cassandra.port, "missing options.cassandra.port field")
	LogDebug("[HELPERS.cassandra.Init] CASSANDRA connecting to:  " .. tostring(options.cassandra.host) .. ":" .. tostring(options.cassandra.port))
	m_cassandra.seed(options.cassandra.host, options.cassandra.port)
	return m_cassandra
end

function GetClient ()
	return m_cassandra
end

function GetAllColumns (options)
	assert(options, "You must supply options")

	local function XOR (A, B)
		return (A and not B) or (B and not A)
	end

	assert(XOR(options.key, options.startKey or options.endKey), "You must supply key or <startKey, endKey> but not both")

	--LogDebug(" 1 ENTRA GetAllColumns columnFamily: " .. tostring(options.columnFamily))
	--LogDebug(" 2 ENTRA GetAllColumns key: " .. tostring(options.key))
	--LogDebug(" 3 ENTRA GetAllColumns columns: " .. tostring(options.columns))

	local has_close_keyspace = true
	local keyspace
	if type( options.keyspace ) == "string" then
		keyspace = m_cassandra:CassandraConnection( options.keyspace )
	else
		keyspace = options.keyspace
		has_close_keyspace = false
	end

	local func, specific_columns
	if options.superColumns then
		func = keyspace.GetSuperRangeSlices
		specific_columns = {}
	else
		func = keyspace.GetRangeSlices
		specific_columns = options.columns or {}
	end

	local get_rows_chunk = function( start_key, end_key, load_chunk_rows )
		local ok, rows = pcall(func,
			keyspace,
			options.columnFamily, --column family
			"", --supercolumn
			start_key, --startkey
			end_key, --endkey
			{ --predicate
				column_names = specific_columns,
				slice_range = {
					start = options.startColumn or "",
					finish = options.endColumn or "",
					reversed = options.reversed or false,
					count = options.columnCount or 10000
				}
			},
			load_chunk_rows --total keys
		)
		if ok then
			return rows
		end
	end

	local get_value = function (column, super_column, key)
		if type(options.transform_value) == "function" then
			return options.transform_value( column, super_column, key )
		else
			return options.valueAsNumber and column.value_as_number or m_table:JSON_Decode(column.value)
		end
	end

	local all_result = {}
	local chunk_result = {}
	local previsous_chunk_result = {}
	local start_chunk_key = options.key or options.startKey or ""
	local end_chunk_key = options.key or options.endKey or ""
	local previous_chunk_key
	local top_rows = options.rowCount
	local total_rows = 0
	local continue = true
	local chunks_loop = 0

	-- Funcion encargada de determinar si la funcion tiene que retornar todos los registros de los chunks, o si tiene que invocar al callback
	-- con los rows que se han obtenido en el chunk
	local function HasRetriveChunks ()
		return options.callbacks and options.callbacks.chunk_iter
	end

	local rows_per_chunk = options.rows_per_chunk or 30000
	local sleep_between_chunk_sec = options.sleep_between_chunk_sec or nil
	--LogDebug("rows_per_chunk: " .. tostring(rows_per_chunk) .. ", sleep_between_chunk_sec: " .. tostring(sleep_between_chunk_sec))

	while continue do
		chunks_loop = chunks_loop + 1
		if type(sleep_between_chunk_sec) == "number" and chunks_loop > 1 then
			LogInfo("[GetAllColumns] Wait moment please, there is a " .. tostring(sleep_between_chunk_sec) .. " seconds pause between chunk. Chunk " .. tostring(chunks_loop) .. " of " .. tostring(rows_per_chunk) .. " rows.")
			socket.sleep(sleep_between_chunk_sec)
		end

		local load_chunk_rows = rows_per_chunk
		-- El primer chunk siempre retorna la totalidad especificada en el parametro recibido "rows_per_chunk",
		--  pero a partir del segundo chunk tengo que incrementar en 1 el valor recibido como parametro (solo una vez),
		--  para que los siguientes, chunks retornen la misma cantidad.
		-- Esto es asi ya que el punto de partida para obtener informacion es el ultimo row del chunk anterior, y si no hacemos esto,
		--   en el siguiente chunk va a venir repetido el ultimo row del chunk anterior.
		if chunks_loop > 1 then
			load_chunk_rows = rows_per_chunk + 1
		end
		previous_chunk_key = start_chunk_key

		local rows = get_rows_chunk( start_chunk_key, end_chunk_key, load_chunk_rows )

		--LogDebug("-> Chunk Number: " .. tostring(chunks_loop) .. ", Column Family: " .. tostring(options.columnFamily))
		if not rows then
			break
		end

		local total_rows_in_chunk = 0

		for key, columns in rows:Iterator() do
			-- Si el row fue agregado en el chunk anterior, no lo tiene que agregar.
			----LogDebug("---> key: " .. tostring(key))
			----LogDebug("---> columns: " .. json.encode(columns))
			if not previsous_chunk_result[key] then
				for _, column in pairs(columns) do
					all_result[key] = all_result[key] or {}
					if not options.superColumns then
						--LogDebug("   " .. tostring(_) .. " ---> column.name: " .. tostring(column.name) .. ", column.value: " .. tostring(column.value))
						all_result[key][column.name] = get_value(column, nil, key)
					else
						local super_column = column
						all_result[key][super_column.name] = {}
						for _, col_of_super_column in pairs(super_column.columns) do
							all_result[key][super_column.name][col_of_super_column.name] = get_value(col_of_super_column, super_column, key)
						end
					end
				end
				chunk_result[key] = m_table:DeepCopy(all_result[key])
			end

			total_rows = total_rows + 1
			total_rows_in_chunk = total_rows_in_chunk + 1
			--LogDebug(tostring(total_rows) .. "' - CHUNK_KEY: '" .. tostring(start_chunk_key) .. "' - chunk row: " .. tostring(total_rows_in_chunk))

			start_chunk_key = key -- actualizo el chunk con el que se ha procesado, para cuando se vuela a pedir los siguientes chunks.

			local is_top_rows = total_rows == top_rows
			if is_top_rows then
				--LogDebug("BREAK - is_top_rows: " .. tostring(is_top_rows))
				if HasRetriveChunks() then
				-- Invoco al callback con los rows del chunk y luego limpio la tabla, para continuar con los chunks siguientes.
					--LogDebug("1-->RETORNO al callback del chunk: " .. tostring(total_rows_in_chunk))
					options.callbacks.chunk_iter(chunk_result, chunks_loop)
				end
				previsous_chunk_result = nil
				previsous_chunk_result = m_table:DeepCopy(chunk_result)
				chunk_result = {}
				continue = false
				break
			end
		end

		if continue then
			if HasRetriveChunks() then
			-- Invoco al callback con los rows del chunk y luego limpio la tabla, para continuar con los chunks siguientes.
				--LogDebug("2-->RETORNO al callback del chunk: " .. tostring(total_rows_in_chunk))
				options.callbacks.chunk_iter(chunk_result, chunks_loop)
			end
			previsous_chunk_result = nil
			previsous_chunk_result = m_table:DeepCopy(chunk_result)
			chunk_result = {}

			if total_rows_in_chunk < rows_per_chunk then
				--LogDebug("BREAK - total_rows_in_chunk: " .. tostring(total_rows_in_chunk) .. " rows_per_chunk: " .. tostring(rows_per_chunk))
				continue = false
			else
				local is_equal_chunks = previous_chunk_key == start_chunk_key

				if is_equal_chunks then
					--LogDebug("BREAK - is_equal_chunks: " .. tostring(is_equal_chunks) .. " start_chunk_key: " .. start_chunk_key)
					continue = false
				end
			end
		end
	end

	if has_close_keyspace then
		keyspace:Close()
	end

	if HasRetriveChunks() then
	-- Como la funcion que retorna los rows de cada chunk se encuentra en el callback, me voy y no retorno nada mas.
		return
	end

	-- Como no se fueron retornando los row por cada callback, retorno todo de una.

	if options.key then
		all_result = all_result[options.key]
	end

	--LogDebug("all_result: " ..  m_table:JSON_Encode(all_result))
	--LogDebug(" 2 SALE GetAllColumns " .. tostring(options.columnFamily))

	return all_result
end

function SaveRows (options)
	assert(options, "You must supply options")
	assert(options.keyspace, "You must supply a keyspace")
	assert(options.columnFamily, "You must supply a columnFamily")
	assert(options.rows, "You must supply one or more rows")

	local keyspace, close_connection
	if type(options.keyspace) == "string" then
		close_connection = true
		keyspace = m_cassandra:CassandraConnection(options.keyspace)
	else
		close_connection = false
		keyspace = options.keyspace
	end

	local columnFamily = options.columnFamily
	local superColumns = options.superColumns
	local removeBefore = options.removeBefore

	for rowId, row in pairs(options.rows) do
		if removeBefore and not superColumns then
			keyspace:RemoveColumn(rowId, columnFamily)
		end
		for columndId, column in pairs(row) do
			if superColumns then
				if removeBefore then
					keyspace:RemoveSuperColumn(rowId, columnFamily, columndId)
				end
				for k, v in pairs(column) do
					local value = type(v) == "table" and m_table:JSON_Encode(v) or tostring(v)
					keyspace:InsertSuperColumn(rowId, columnFamily, columndId, tostring(k), value)
				end
			else
				--standard column
				local value = type(column) == "table" and m_table:JSON_Encode(column) or tostring(column)
				keyspace:InsertColumn(rowId, columnFamily, columndId, value)
			end
		end
	end

	if close_connection then
		keyspace:Close()
	end
end

--[[
	If parameter row has the form: ["row1", "row2", ...]
	This function will remove one or more rows,
	
	If parameter row has the form: {
		row1 : ["column1", "column2", ...],
		row2 : ["column1", "column3", ...],
	}
	This function will remove a supercolumn
	
	
	If parameter row has the form: {
		row1: {
			supercolumn1: ["column1", "column2", ...],
			supercolumn2: ["column1", "column2", ...],
		},
		row2: {...}
	}
	
	This function will remove only each supercolumn
]]
function RemoveColumns (options)
	assert(options, "You must supply options")
	assert(options.keyspace, "You must supply a keyspace")
	assert(options.columnFamily, "You must supply a columnFamily")
	assert(options.rows, "You must supply one or more rows")

	local keyspace, close_connection
	if type(options.keyspace) == "string" then
		close_connection = true
		keyspace = m_cassandra:CassandraConnection(options.keyspace)
	else
		close_connection = false
		keyspace = options.keyspace
	end

	for rowId, row in pairs(options.rows) do
		if type(row) == "string" then
			keyspace:RemoveColumn(row, options.columnFamily)
		elseif type(row) == "table" then
			for columnName, column in pairs(row) do
				if type(column) == "string" then
					if options.superColumns then
						keyspace:RemoveColumn(rowId, options.columnFamily, nil, column)
					else
						keyspace:RemoveColumn(rowId, options.columnFamily, column)
					end
				elseif type(column) == "table" then
					if not options.superColumns then
						error("You must specify options.superColumns to remove in this level")
					end
					for k, v in pairs(column) do
						keyspace:RemoveColumn(rowId, options.columnFamily, v, columnName)
					end
				else
					error("Unexpected column type in RemoveColumn")
				end
			end
		else
			error("Unexpected row type in RemoveColumn")
		end
	end

	if close_connection then
		keyspace:Close()
	end
end

------------------------

function LoadAllVCCs (ksAgent)
	assert(ksAgent)
	local vccs = GetAllColumns({
		keyspace     = ksAgent,
		columnFamily = "VCCs",
		startKey     = "",
		endKey       = ""
	})
	--LogDebug("[HELPERS.cassandra:LoadAllVCCs] loaded all vccs: " .. json.encode(vccs or {}))
	return vccs
end
