local json = require "cjson"
module(..., package.seeall)
local m_cassandra = nil

---
-- OJO! este modulo es compartido entre distintas aplicaciones, las que tienen que llamar este SetupConnections
-- antes de poder usar funciones como GetAllColumns, SaveRows o RemoveColumns, pasando la tabla 'lib_cassandra'
-- a la que se le tiene que haber hecho 'seed' previamente, etc.
-- Si alguien por error setea algo incorrecto acá, rompe el funcionamiento de cualquier otra aplicación.
--
-- Los módulos así, como este, tipo "utils" NO DEBEN TENER ESTADO.
-- En SetupConnections se estaba pasando Redis, que por suerte no se estaba usando, lo cual hubiese sido mucho peor
-- porque ya se está pasando una instancia conectada.
-- Si otra aplicación pasa una conexion de redis a un server distinto, todas las demas aplicaciones dejan de andar, y
-- andá a encontrar el porqué del bug.
--

function SetupConnections (connections)
	m_cassandra = connections.cassandra
end

function GetDatasource ()
	return m_cassandra
end

function SanitizeId (id)
	assert(type(id) == "string")
	id = id or ""
	id = id:gsub('"', '')
	id = id:gsub("'", "")
	return id
end

function GetColumnValue ( options )
	assert( options, "You must supply options" )
	--assert( type( options.keyspace ) == "string" or type( options.keyspace ) == "table", "keyspace id must be a string or table")
	assert( type( options.columnFamily ) == "string", "columnFamily id must be a string")
	assert( type( options.key ) == "string", "columnFamily id must be a string")
	assert( type( options.column ) == "string", "columnFamily id must be a string")

	if type( options.keyspace ) == "string" then
		local keyspace = m_cassandra:CassandraConnection( options.keyspace )
		local columnValue = keyspace:GetColumnValue( options.key, options.columnFamily, options.column )
		keyspace:Close()
		return columnValue
	else
		return options.keyspace:GetColumnValue( options.key, options.columnFamily, options.column )
	end
end

--[[
	Generic function to retrieve all columns in a keyspace and columnFamily
--]]
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 " .. tostring(options.columnFamily))

	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 rows_per_chunk = options.rows_per_chunk or 30000

	local get_rows_chunk = function( start_key, end_key )

		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
				}
			},
			rows_per_chunk --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 JSON_Decode(column.value)
		end
	end

	local 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

	while continue do

		previous_chunk_key = start_chunk_key
		local rows = get_rows_chunk( start_chunk_key, end_chunk_key )
		chunks_loop = chunks_loop + 1
		--LogDebug("chunk " .. tostring(chunks_loop) .. " " .. tostring(options.columnFamily))
		if not rows then
			break
		end
		local total_rows_in_chunk = 0

		for key, columns in rows:Iterator() do
			for _, column in pairs(columns) do
				result[key] = result[key] or {}
				if not options.superColumns then
					result[key][column.name] = get_value(column, nil, key)
				else
					local super_column = column
					result[key][super_column.name] = {}
					for _, col_of_super_column in pairs(super_column.columns) do
						result[key][super_column.name][col_of_super_column.name] = get_value(col_of_super_column, super_column, key)
					end
				end
			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

			local is_top_rows = total_rows == top_rows
			if is_top_rows then
				--LogDebug("BREAK - is_top_rows: " .. tostring(is_top_rows))
				continue = false
				break
			end
		end

		if continue then
			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 options.key then
		result = result[options.key]
	end
	--LogDebug("result: " .. json.encode(result))
	--LogDebug(" 2 SALE GetAllColumns " .. tostring(options.columnFamily))
	return 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 json.encode(v) or tostring(v)
					keyspace:InsertSuperColumn(rowId, columnFamily, columndId, k, value)
				end
			else
				--standard column
				local value = type(column) == "table" and 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 GetRowCount (rows)
	local count = 0
	for _, row in pairs(rows or {}) do
		if type(row) ~= "table" or next(row) then
			count = count + 1
		end
	end
	return count
end

function JSON_Decode (value)
	local ok, result = pcall (function ()
		local result = json.decode(value)
		if type (result) == "table" then
			return result
		else
			return value
		end
	end)
	return ok and result or value
end

function DeepCopy (t)
	if type(t) == "table" then
		local result = {}
		for k, v in pairs (t) do
			result[k] = DeepCopy(v)
		end
		return result
	else
		return t
	end
end
