module(..., package.seeall)
local m_helpers = require "helpers"
local m_json = require "cjson"
require "LuaOrm.Date"
local m_datasource = {}

function SetDatasource (datasource)
	m_datasource.cassandra = datasource.cassandra
end

function GetDatasource ()
	return m_datasource
end

local m_CFs = {
	Audit             = "Audit",
	AuditIndexByDates = "AuditIndexByDates"
}

local function GetElapsedTimeSec (start)
	return m_helpers.utils.ElapsedTime.GetElapsedTimeSec(start)
end

--[[
	"vcc1::2022-10-20
]]--
local function MakeVccDateKey (vcc, date)
	return ("%s::%s"):format(vcc, date)
end

--[[
	"vcc1::2022-10-2::Entity::User::a.jose.molina1"
]]--
local function MakeAuditKey (vccDateKey, scEntity, fieldKey)
	return ("%s::%s::%s"):format(vccDateKey, scEntity, fieldKey)
end

--[[
	param key: "vcc1::2022-10-20::Entity::User::a.jose.molina1"
	response 
]]--
local function ParseAuditKey (key)
	assert(type(key) == "string")
	local splitkey = m_helpers.utils.SplitString(key, "::")
	local parsedKey = {
		vcc   = splitkey[1],
		date  = splitkey[2],
		type  = splitkey[3],
		name  = splitkey[4],
		value = splitkey[5]
	}
	--LogDebug("[audits:ParseAuditKey] key: %s, parsedKey: %s", key, json.encode(parsedKey))
	return parsedKey
end

local function GetTotal (params, decode)
	assert(params.key, "audit.model.GetTotal You must supply key")
	assert(params.connection, "audit.model.GetTotal You must supply Cassandra Connection")
	local connection = params.connection
	local res, value = pcall(connection.GetColumnValue, connection, params.key, params.columnFamily, params.columnName)
	--LogDebug("%s", value)
	if res and value then
		if decode then
			return m_json.decode(value)
		end
		return value
	end
end

local function GetAudit (params)
	assert(params, "audit.model.GetAudit You must supply params")
	assert(params.connection, "audit.model.GetAudit You must supply connection")
	assert(params.auditKey, "audit.model.GetAudit You must supply auditKey")

	local connection = params.connection
	local reversed = params.order == "desc" and true or false

	local rows = connection:GetRangeSlices(m_CFs.Audit, --column family
		"",  --supercolumn
		params.auditKey, --startkey
		params.auditKey, --endkey
		{ --predicate
			column_names = {},
			slice_range = {
				start = "",
				finish = "",
				reversed = reversed,
				count = 100000
			}
		},
		1 --total keys
	)
	--LogDebug("****************************")
	local auditRows = {}
	if rows then
		for id, row in rows:Iterator() do
			--LogDebug("   %s -> %s", id, #(row))
			if #row > 0 then
				for k, v in ipairs(row) do
					--LogDebug("      ->   %s", v.name)
					local columnValue = v.value
					if v.name:lower() ~= "total" then
						local decodedValue = m_json.decode(v.value)
						decodedValue.cassandraTimeStamp = v.timestamp
						--LogDebug("         columnName: %s, v.value.timeStampStr: '%s', cassandra.cassandraTimeStamp: %s", v.name, decodedValue.timeStampStr or "NONE", decodedValue.cassandraTimeStamp)
						table.insert(auditRows, decodedValue)
					end
				end
			end
		end
	end

	return auditRows
end

local function GetAuditIndexByDates (params)
	assert(params, "audit.model.GetAuditIndexByDates You must supply params")
	assert(params.connection, "audit.model.GetAuditIndexByDates You must supply connection")
	assert(params.key, "audit.model.GetAuditIndexByDates You must supply key")

	local connection = params.connection
	local reversed = params.order == "desc" and true or false

	local rows = connection:GetRangeSlices(m_CFs.AuditIndexByDates, --column family
		"",  --supercolumn
		params.key, --startkey
		params.key, --endkey
		{ --predicate
			column_names = {},
			slice_range = {
				start = "",
				finish = "",
				reversed = reversed,
				count = 100000
			}
		},
		1 --total keys
	)
	--LogDebug("****************************")
	local auditRowsByDates = {}
	if rows then
		for id, row in rows:Iterator() do
			--LogDebug("   %s -> %s", id, #(row))
			if #row > 0 then
				for k, v in ipairs(row) do
					--LogDebug("         columnName: %s, va.value: '%s', cassandra.cassandraTimeStamp: %s", v.name, v.value, v.timestamp)
					table.insert(auditRowsByDates, {
						auditKey  = v.name,
						admin     = v.value,
						cassandraTimeStamp = v.timestamp
					})
				end
			end
		end
	end

	return auditRowsByDates
end

function GetRowsByDate (params)
	assert(params, "audit.model.GetRowsByDate You must supply params")
	assert(params.connection, "audit.model.GetRowsByDate You must supply connection")
	assert(params.vcc, "audit.model.GetRowsByDate You must supply vcc")
	assert(params.initialDate, "audit.model.GetRowsByDate You must supply initialDate")
	assert(params.endDate, "audit.model.GetRowsByDate You must supply endDate")

	--LogDebug("GetRowsByDate ENTRA")
	local connection = params.connection
	local order = (params.order or "desc"):lower()
	local currentDate = LuaOrm.Date(params.initialDate)
	local endDate = LuaOrm.Date(params.endDate)
	local rows = {}
	local i = 0
	local daysToIncrement = order == "desc" and -1 or 1
	local reversed = order == "desc" and true or false

	local allPlainRows = {}
	while (currentDate:time() <= endDate:time()) do
		i = i + 1
		local d = order == "desc" and endDate or currentDate
		local dateString = d:toString("%Y-%m-%d")
		local vccDateKey = MakeVccDateKey(params.vcc, dateString)
		LogDebug("-------------------------------")
		LogDebug("  %d - dateString: %s - KEY: '%s'", i, dateString, vccDateKey)
		LogDebug("-------------------------------")
		
		--[[------------------------------------------------------------------
			En este caso el utilizar las funciones genercias "GetAllColumns", no incrementa el tiempo ya que no tiene que encodear ni recorrer de forma recursiva.
		----------------------------------------------------------------------]]--
		local indexesByDate = GetAuditIndexByDates({
			connection = connection,
			columnFamily = m_CFs.AuditIndexByDates,
			key = vccDateKey,
			order = order
		})
		-- local indexesByDate = m_helpers.cassandra.GetAllColumns({
		-- 	keyspace = connection,
		-- 	columnFamily = m_CFs.AuditIndexByDates,
		-- 	key = vccDateKey,
		-- 	reversed = reversed,
		-- 	rowCount = 100000
		-- })
		------------------------
		for i, info in ipairs(indexesByDate or {}) do
			local auditKey = info.auditKey
			local admin = info.admin
			local auditRows = GetAudit({
				connection = params.connection,
				auditKey = auditKey,
				order = order
			})
			--LogDebug(" %d -> auditKey: %s, admin: %s", i, auditKey, admin)
			for x, auditRow in ipairs(auditRows) do
				auditRow.__info = {
					dateString = dateString,
					admin = admin,
					auditKey = auditKey
				}
				table.insert(allPlainRows, auditRow)
				--LogDebug("  EMMA  %d -> auditRow.timestamp: %s, auditRow.timeStampStr: %s", x, auditRow.cassandraTimeStamp, auditRow.timeStampStr)
			end
			rows[dateString] = rows[dateString] or {}
			rows[dateString][admin] = rows[dateString][admin] or {}
			rows[dateString][admin][auditKey] = rows[dateString][admin][auditKey] or auditRows
		end
		d = LuaOrm.Date({year = d:year(), month = d:month(), day = (d:day() + daysToIncrement), hour = 0})
		if order == "desc" then
			endDate = d
		else
			currentDate = d
		end
	end

	--LogDebug("--> GetRowsByDate-> rows: %s", m_json.encode(rows))
	return rows
end

--[[
	filters = {
		user = user,
		entity = entity,
		operation = operation,
		fieldKey = fieldKey,
		vcc = vcc
	}
]]--
function FilterRecords (filters, records)
	local result = {}

	--LogDebug("A--> filters: %s", m_json.encode(filters))
	--LogDebug("B--> records: %s", m_json.encode(records))

	local function addDataRecords (entity, keyField, dataByEntity)
		local data = {}
		local added = {}
			for i, records in ipairs(dataByEntity) do --By entity
				local index
				local key = ("%s::%s"):format(entity, keyField)
				if not added[key] then
					table.insert(data, {
						entity   = entity,
						keyField = keyField,
						records  = {}
					})
					added[key] = #data
					index = 1
				else
					index = tonumber(added[key])
				end
				table.insert(data[index].records, records)
			end
		return data
	end

	local function FilterDataRecords (info, dataRecords)
		if not info then
			return
		end
		for key, rows in pairs(info) do
			local parsedKey = ParseAuditKey(key)
			local entity = parsedKey.name
			if not filters.entity or filters.entity:lower() == entity:lower() then
				local keyField = parsedKey.value
				--LogDebug("1-->entity: %s", entity)
				--LogDebug("1.1-->keyField: %s", keyField)
				--LogDebug("2-->rows: %s", json.encode(rows))
				if not filters.fieldKey or filters.fieldKey:lower() == keyField:lower() then
					local returnData = addDataRecords(entity, keyField, rows)
					--LogDebug("3-->returnData: %s", json.encode(returnData))
					for _, value in pairs(returnData) do
						table.insert(dataRecords, value)
					end
				end
			end
		end
	end

	local startSubProcessTime = os.time()
	local totalRecords = 0
	local totalDateKeys = 0
	for dateKey, data in pairs(records) do
		local dataRecords = {}
		if filters.admin then
			FilterDataRecords(data[filters.admin], dataRecords)
		else
			for admin, info in pairs(data) do
				FilterDataRecords(info, dataRecords)
			end
		end
		--LogDebug("1 - despues de aplicar filtro principal. dateKey: %s, filteredRows: %s", dateKey, #dataRecords)
		--LogDebug("1.1 - despues de aplicar filtro principal. dateKey: %s, filteredRows: %s", dateKey, m_json.encode(dataRecords))

		if filters.operation or filters.subEntity then
			local recordsByOperation = {}
			for _, value_data in pairs(dataRecords) do
				local records = {}
				local has_match = false
				for _, value_record in pairs(value_data.records) do
					local can_insert = value_record.operation and filters.subEntity and value_record.operation == filters.operation and filters.subEntity == value_record.subEntity or false
					can_insert = not can_insert and not filters.operation and filters.subEntity and filters.subEntity == value_record.subEntity or can_insert
					can_insert = not can_insert and filters.operation and not filters.subEntity and filters.operation == value_record.operation or can_insert
					if  can_insert then
						table.insert(records, value_record)
						has_match = true
					end
				end
				if has_match then
					local aux_data = value_data
					aux_data.records = records
					table.insert(recordsByOperation, aux_data)
				end
			end
			local data = {} 
			for _, value in pairs(recordsByOperation) do
				table.insert(data, value)
			end
			dataRecords = data
		end
		--LogDebug("2 - despues de aplicar filtro secundario. dateKey: %s, filteredRows: %s", dateKey, #dataRecords)
		--LogDebug("2.1 - despues de aplicar filtro secundario. dateKey: %s, filteredRows: %s", dateKey, m_json.encode(dataRecords))
		if #dataRecords > 0 then
			for i, dataRecord in ipairs(dataRecords) do
				--LogDebug("  %d--> entity: %s, keyField: %s, records: %d", i, dataRecord.entity, dataRecord.keyField, #dataRecord.records)
				totalRecords = totalRecords + #dataRecord.records
				-- -- ordeno los rows
				-- dataRecord.records = m_helpers.table:SortArray(dataRecord.records, function(row)
				-- 		assert(row)
				-- 		local timeStampStr, timeStamp = m_helpers.date:ConvertUTCDateToDate(row.timeStampStr, nil, "%Y-%m-%d %H:%M:%S", true, filters.timeZoneInfo)
				-- 		row.timeStampStr = timeStampStr
				-- 		row.timeStamp = row.timeStamp or timeStamp
				-- 		local dataType = "timeStamp"
				-- 		return row.timeStamp, dataType
				-- 	end,
				-- 	filters.desc
				-- )
			end
			totalDateKeys = totalDateKeys + 1
			result[dateKey] = dataRecords
		end
	end

	LogDebug("C--> (After Filtered Array) totalDateKeys: %s, totalRecords: %s. ELAPSED Time '%s' seconds.", totalDateKeys, totalRecords, GetElapsedTimeSec(startSubProcessTime))
	--LogDebug("C1--> result: %s", m_json.encode(result))

	startSubProcessTime = os.time()
	local filteredRecords = {}
	for key, rows in pairs(result or {}) do
		--LogDebug("   key: %s", key)
		for _, row in ipairs(rows) do
			--LogDebug("     _: %s", _)
			for i, record in ipairs(row.records) do
				--counter = counter + 1
				record.Id = record.Id or uuid.new()
				record.entity = row.entity
				record.keyField = row.keyField
				--LogDebug("     %d -  Id: %s, entity: %s, keyField: %s, timeStampStr: %s, timeStamp: %s", i, record.Id, record.entity, record.keyField, record.timeStampStr, record.timeStamp)
				table.insert(filteredRecords, record)
			end
		end
	end
	LogDebug("D--> (After Plain Array) filteredRecords: %s. ELAPSED Time '%s' seconds.", #filteredRecords, GetElapsedTimeSec(startSubProcessTime))

	startSubProcessTime = os.time()
	local sortedFilteredRecords = m_helpers.table:SortArray(filteredRecords, function(row)
			assert(row)
			assert(row.timeStamp)
			--assert(row.cassandraTimeStamp)
			local dataType = "timeStamp"
			local timestamp = row.timeStamp -- row.cassandraTimeStamp
			return timestamp, dataType
		end,
		filters.desc
	)
	LogDebug("E--> (After Sorted Array) sortedFilteredRecords: %s. ELAPSED Time '%s' seconds.", #sortedFilteredRecords, GetElapsedTimeSec(startSubProcessTime))

	local from, to
	if filters.paging then
		local numberPage = tonumber(filters.paging.numberPage)
		local rowsPerPage = tonumber(filters.paging.rowsPerPage)
		--LogDebug("--> numberPage: %s, rowsPerPage: %s", numberPage, rowsPerPage)
		if numberPage == 1 then
			from = 1
		else
			from = ((numberPage - 1) * rowsPerPage) + 1
		end
		to = from + (rowsPerPage - 1)
		--LogDebug("--> from: %s, to: %s", from, to)
	end

	startSubProcessTime = os.time()
	local pageRecords = m_helpers.table:TruncateArray(sortedFilteredRecords, from, to, function(row)
		assert(row)
		assert(row.timeStampStr)
		local timeStampStr = m_helpers.date:ConvertUTCDateToDate(row.timeStampStr, nil, "%Y-%m-%d %H:%M:%S", true, filters.timeZoneInfo)
		row.displayTime = timeStampStr
		return row
	end)
	local totalPageRecords = #pageRecords
	LogDebug("F--> (After Truncate Array) totalPageRecords: %s. ELAPSED Time '%s' seconds.", totalPageRecords, GetElapsedTimeSec(startSubProcessTime))

	return pageRecords, totalRecords, totalPageRecords
end

function RemoveByDate (params)
	assert(params, "audit.model.RemoveRecords You must supply params")
	assert(params.vccs, "audit.model.RemoveRecords - params.vccs is required")

	local startProcessTime = os.time()
	local keyspace = m_datasource.cassandra:CassandraConnection("Historical")
	local now = LuaOrm.Date()
	local nowTime = now:time()
	local daysToIncrement = 1
	local vccs = params.vccs
	local maxExtraDays = 15

	LogInfo("[Audits::RemoveByDate] +START -Removing vccs audits by date. startProcessTime: %s", startProcessTime)

	for vcc, vccData in pairs(vccs) do
		local days = vccData.Audit and vccData.Audit.CleanUpPeriod or 365
		local limitCleanUpDate = (now:add(unpack({0,0, -(days) }))):toString("%Y-%m-%d")
		local currentDate = LuaOrm.Date(limitCleanUpDate)
		local extraDays = 0
		local i = 0

		LogInfo("[audit.RemoveByDate] vcc: %s, Audit.CleanUpPeriod days: %s, limitCleanUpDate: %s", vcc, days, limitCleanUpDate)

		repeat
			i = i + 1
			local currentDateString = currentDate:toString("%Y-%m-%d")
			local vccDateKey = MakeVccDateKey(vcc, currentDateString)
			--LogDebug(" %d - currentDateString: %s - KEY: '%s'", i, currentDateString, vccDateKey)
			-- En este caso el utilizar las funciones genercias "GetAllColumns", no incrementa el tiempo ya que no tiene que encodear ni recorrer de forma recursiva.

			local indexesByDate = m_helpers.cassandra.GetAllColumns({
				keyspace = keyspace,
				columnFamily = m_CFs.AuditIndexByDates,
				key = vccDateKey,
				rowCount = 100000
			})

			if indexesByDate == nil or next(indexesByDate) == nil then
				extraDays = extraDays + 1
			else
				for auditKey, admin in pairs(indexesByDate or {}) do
					--LogDebug(" auditKey: %s, admin: %s", auditKey, admin)
					keyspace:RemoveColumn(auditKey, m_CFs.Audit)
				end
				keyspace:RemoveColumn(vccDateKey, m_CFs.AuditIndexByDates)
			end

			currentDate = LuaOrm.Date({year = currentDate:year(), month = currentDate:month(), day = (currentDate:day() - daysToIncrement), hour = 0})

		until(extraDays > maxExtraDays)
	end

	keyspace:Close()
	LogInfo("[Audits::RemoveByDate] -END Removed vccs audits by date.. elapsedTime %s seconds, endProcessTime: %s", GetElapsedTimeSec(startProcessTime))
end

function NewRecord (params)
	assert(params, "audit.model.NewRecord You must supply params")
	assert(params.data, "audit.model.NewRecord You must supply a data")
	assert(params.entity, "audit.model.NewRecord You must supply a entity")
	assert(params.fieldKey, "audit.model.NewRecord You must supply a keyField")
	assert(params.user, "audit.model.NewRecord You must supply a user")
	assert(params.vcc, "audit.model.NewRecord You must supply a vcc")
	assert(params.date, "audit.model.NewRecord You must supply a date")
	assert(params.connection, "audit.model.NewRecord You must supply Cassandra Connection")

	local vcc = params.vcc
	local date = params.date
	local newData = params.data
	local entity = params.entity
	local fieldKey = params.fieldKey
	local scEntity = "Entity::" .. entity 		--Super Column Entity
	local cfEntity = m_CFs.Audit 			--Column Family Entity
	local admin = params.user:lower() 					--Super Column User
	local connection = params.connection
	params.connection = nil

	--LogDebug("[AUDIT:NewRecord] (ENTRA) --> connection: %s", connection)
	--LogDebug("[AUDIT:NewRecord] (ENTRA) --> params: %s", m_json.encode(params))

	local vccDateKey = MakeVccDateKey(vcc, date)

	--[[
		KEY vcc::fecha::Entity::(User, Campaign...)::IdUser, IdCampaña(column)
	]]--
	local keyCFByEntity = MakeAuditKey(vccDateKey, scEntity, fieldKey)
	local total = GetTotal({
		connection = connection,
		columnFamily = cfEntity,
		key = keyCFByEntity,
		columnName = "Total"
	})
	--LogDebug("[AUDIT:NewRecord] total: %s", total)
	total = tonumber(total or "0") + 1
	local rows = {
		[keyCFByEntity] = {
			Total = total,
			[tostring(total)] = newData
		}
	}
	--LogDebug("---> rows: %s", m_json.encode(rows))
	local options = {
		keyspace = connection,
		columnFamily = cfEntity,
		rows = rows
	}
	-- save AuditByEntityTempora
	m_helpers.cassandra.SaveRows(options)

	---- Index Audit by Date. -------------
	--
	local cfIndexByDates = m_CFs.AuditIndexByDates 				--Column Family Indexes By Users
	local rows = {
		[vccDateKey] = {
			[keyCFByEntity] = admin -- el usuario administrador que ha actualizado la entidad.
		}
	}
	local options = {
		keyspace = connection,
		columnFamily = cfIndexByDates,
		rows = rows
	}
	m_helpers.cassandra.SaveRows(options)
end
