local json = require "cjson"
require "socket.http" 
require "LuaOrm.Date"
require "lfs"
local reporterFtpController = require "reporter.ftpController"
local reporterDataBase = require "reporter.database"
local manager = require "reporter.manage"
local m_helpers = require "helpers"	

-- some constants
local AGENT_KS                = "Agent"
local VCCS_CF                 = "VCCs"
local USERS_CF                = "Users"
local TIMEZONES_CF            = "Timezones"

local SUMMARIZER_KS           = "SummarizerData"
local SCHEDULES_CF            = "ReportSchedules"
local SCHEDULES_BY_AGENT_CF   = "ReportSchedulesByAgent"
local SCHEDULES_HISTORICAL_CF = "ReportSchedulesHistorical"
local EXPORTED_CONTENT_CF     = "ReportSchedulesExportedContent"
local SCHEDULES_REDIS         = "ReportSchedules"

local SUNDAY    = 1
local MONDAY    = 2
local TUESDAY   = 3
local WEDNESDAY = 4
local THURSDAY  = 5
local FRIDAY    = 6
local SATURDAY  = 7

local ACTIVE    = "active"
local EXPIRED   = "expired"
local INACTIVE  = "inactive"
local RUNNING   = "running"

local RUNNING_TIMEOUT = 60 * 60 * 12

local DEFAULT_SUBJECT = "Scheduled Report %REPORT%"
local DEFAULT_BODY = "Your report %REPORT% has been executed and it's attached in this email. %EXPORT_FILE_LINK%"

-- Tope maximo de intentos fallidos de reagendar
local MAX_ALLOWED_FAILED = 3

-- in case it's available
local LogInfo = LogInfo or function(msg, ...)
	print("Info", msg:format(...))
end
local LogWarning = LogWarning or function(msg, ...)
	print("Warning", msg:format(...))
end

local function GetScheduleById (utils, id)
	return utils.GetAllColumns {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_CF,
		key = id
	}
end

local function GetSchedulesByAgent (utils, agentAtVcc)
	return utils.GetAllColumns {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_BY_AGENT_CF,
		key = agentAtVcc
	}
end

local function ValidateHour (strTime)
	if type(strTime) ~= "string" then
		return nil
	end
	local t = LuaOrm.Date(strTime..".999")
	if t:milliseconds() == 999 then
		return t:toString("%H:%M:%S")
	end
	-- invalid hour
	return nil
end

local function ValidateDate (strDate)
	if type(strDate) ~= "string" then
		return nil
	end
	local d = LuaOrm.Date(strDate)
	if d:yearDate():time() > 0 then
		return d:toString("!%Y-%m-%d")
	end
	-- invalid date
	return nil
end

local function MakeDateHour (string_date, string_hour)
	string_date = ValidateDate( string_date )
	assert( type( string_date ) == "string", "invalid date: " .. tostring( string_date ))
	string_hour = ValidateHour( string_hour )
	assert( type( string_hour ) == "string", "invalid hour: " .. tostring( string_hour ))

	return LuaOrm.Date( string_date .. " " .. string_hour )
end

local function MakeStartDateHour( string_date, string_hour )
	assert( type( string_date ) == "string", "invalid start date")
	string_hour = string_hour or "00:00:00"
	return MakeDateHour( string_date, string_hour )
end

local function MakeEndDateHour (string_date, string_hour)
	assert( type( string_date ) == "string", "invalid end date")
	string_hour = string_hour or "23:59:59"
	return MakeDateHour( string_date, string_hour )
end

-- Using the schedule data calculates the next timestamp when it should execute
-- This algorithm ensures that the next execution will follow the desired repeat policy
local function CalculateNextTimestamp (scheduleData)
	if scheduleData.EndsAfter and scheduleData.ExecutionCount then
		if tonumber(scheduleData.ExecutionCount) >= tonumber(scheduleData.EndsAfter) then
			return nil
		end
	end

	local now = LuaOrm.Date()

	if scheduleData.EndsOnDate then
		local endDate = MakeEndDateHour( scheduleData.EndsOnDate, scheduleData.EndsOnHour )
		if now:time() >= endDate:time() then
			return nil
		end
	end

	local lastExecution = LuaOrm.Date(tonumber(scheduleData.LastExecutionDate) or now)--:dayDate()
	local nextExecution = LuaOrm.Date(lastExecution)

	if scheduleData.RepeatPolicy == "EveryNHours" then
		local startDate = MakeStartDateHour( scheduleData.StartsOnDate, scheduleData.StartsOnHour )

		if lastExecution:time() < startDate:time() then
			-- first execution
			return startDate:time()
		else
			-- subsequent executions
			local newTimestamp = LuaOrm.Date({year = lastExecution:year(), month = lastExecution:month(), day = lastExecution:day(), hour = lastExecution:hour()+scheduleData.RepeatEveryNHours, min = 0, sec = 0}):time()

			return newTimestamp
		end
	elseif scheduleData.RepeatPolicy == "Daily" then
		local h = LuaOrm.Date(scheduleData.RepeatHour or now)
		local todayExecution = LuaOrm.Date {year = now:year(), month = now:month(), day = now:day(), hour = h:hour(), min = h:minute()}

		if todayExecution:time() <= now:time() then
			--we can not execute today, so we schedule for tomorrow
			nextExecution = nextExecution:add(0, 0, 1)
		end

	elseif scheduleData.RepeatPolicy == "CustomRepeat" then

		scheduleData.CustomRepeatDays = tonumber(scheduleData.CustomRepeatDays)
		local startDate = MakeStartDateHour( scheduleData.StartsOnDate, scheduleData.StartsOnHour )
		if startDate:time() > lastExecution:time() then
			nextExecution = startDate
		else
			local diff = startDate:interval(lastExecution, "D")
			local alignDiff = diff % scheduleData.CustomRepeatDays
			nextExecution = nextExecution:add(0, 0, scheduleData.CustomRepeatDays - alignDiff)
		end

	elseif scheduleData.RepeatPolicy == "Weekly" then

		local daysToExecute = scheduleData.RepeatWeekday - lastExecution:dayOfWeek()
		if daysToExecute <= 0 then
			daysToExecute = daysToExecute + 7
		end

		nextExecution = nextExecution:add(0, 0, daysToExecute)

	elseif scheduleData.RepeatPolicy == "Monthly" then

		local lastExecutionMonthEnd = lastExecution:monthEndDate():dayDate()
		local potentialRepetitionDate = LuaOrm.Date {year = lastExecution:year(), month = lastExecution:month(), day = scheduleData.RepeatDay}:dayDate()

		if potentialRepetitionDate:time() > lastExecutionMonthEnd:time() then
			potentialRepetitionDate = lastExecutionMonthEnd
		end

		if lastExecution:time() < potentialRepetitionDate:time() then
			nextExecution = potentialRepetitionDate
		else
			nextExecution = LuaOrm.Date {year = lastExecution:year(), month = lastExecution:month() + 1, day = scheduleData.RepeatDay}
			local limitDate = lastExecution:monthEndDate():add(0, 0, 1):monthEndDate():dayDate()

			if nextExecution:time() > limitDate:time() then
				--we can not skip a month
				nextExecution = limitDate
			end
		end

	else
		local d = LuaOrm.Date(scheduleData.OneTimeDate)
		local h = LuaOrm.Date(scheduleData.OneTimeHour)

		nextExecution = LuaOrm.Date {year = d:year(), month = d:month(), day = d:day(), hour = h:hour(), min = h:minute()}

		local next_exec = nextExecution:time()
		local now = os.time()
		if next_exec < now then
			local format_date = "%Y/%m/%d %H:%M"
			local str_exe = os.date( format_date, next_exec )
			local str_now = os.date( format_date, now )
			LogError("[REPORTER.SCHEDULER:CalculateNextTimestamp] You must define OneTime's execution in the future. Next execution time: '" .. tostring(str_exe) .. "', Now time: '" .. tostring(str_now) .. "'")
			return nil
		else
			return next_exec
		end
	end

	-- enforce the hour of execution
	local d = nextExecution
	local h = LuaOrm.Date(scheduleData.RepeatHour or os.time())

	local result = LuaOrm.Date {year = d:year(), month = d:month(), day = d:day(), hour = h:hour(), min = h:minute()}

	return result:time()
end

local function AddScheduleHistorical (utils, scheduleId, date, success, data)
	utils.SaveRows {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_HISTORICAL_CF,
		rows = {
			[scheduleId] = {
				[LuaOrm.Date(date):toString()] = {
					success = success,
					data = data
				}
			}
		}
	}
end

local function GetScheduleHistorical (utils, scheduleId, params)
	params = params or {}
	local columnCount = nil
	local reversed = nil
	if params.history then
		columnCount = params.history.maxRows
		reversed = (params.history.order or ""):lower() == "desc" or nil
	end
	local hitorical = utils.GetAllColumns {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_HISTORICAL_CF,
		key = scheduleId,
		columnCount = columnCount,
		reversed = reversed
	}
	return hitorical or {}
end

local function RegisterExportedContent (utils, vcc, scheduleId, size, removeDate, contentId)
	utils.SaveRows {
		keyspace = SUMMARIZER_KS,
		columnFamily = EXPORTED_CONTENT_CF,
		rows = {
			[vcc] = {
				[contentId] = {
					size = size,
					scheduleId = scheduleId,
					removeDate = LuaOrm.Date(removeDate):toString(),
				}
			}
		}
	}
end

local function GetVCCExportSpaceUtilized (utils, vcc)
	local records = utils.GetAllColumns {
		keyspace = SUMMARIZER_KS,
		columnFamily = EXPORTED_CONTENT_CF,
		key = vcc,
	} or {}

	local totalSize = 0
	for _, record in pairs(records) do
		totalSize = totalSize + record.size
	end

	return totalSize
end

local function GetVCCExportSpaceQuota (utils, vcc)
	if m_helpers.utils.HasLicenseSystem() then
		local options = {}
		if inconcert and inconcert.environment and inconcert.environment.apps and inconcert.environment.apps.agent and inconcert.environment.apps.agent.agentservice and inconcert.environment.apps.agent.cluster then
			options.host = inconcert.environment.apps.agent.agentservice.host
			options.port = inconcert.environment.apps.agent.agentservice.port
			options.cluster = inconcert.environment.apps.agent.cluster
		end	
		local vccInfo = m_helpers.agentservice.GetVccLicense(options, vcc)
		return 1024 * 1024 * (vccInfo.ExportSpaceQuota or 0)
	else	
		local vcc = utils.GetAllColumns {
				keyspace = AGENT_KS,
				columnFamily = VCCS_CF,
				key = vcc
		} or {}

		--space quota is in MB and we need in Bytes
		return 1024 * 1024 * (tonumber(vcc.ExportSpaceQuota) or 0)
	end
end

local function GetMailFromConfig (utils, vcc, from)
	local encodedVccAccounts = utils.GetColumnValue({
		keyspace = AGENT_KS,
		columnFamily = VCCS_CF,
		key = vcc,
		column = "Accounts"
	})

	if not encodedVccAccounts then
		return
	end

	local vccAccounts = json.decode( encodedVccAccounts )
	local outgoingMailAccounts = vccAccounts.MAIL
	if not outgoingMailAccounts then
		return
	end

	from = ( from or "" ):upper()
	local outgoingMail = outgoingMailAccounts[ from ]
	return outgoingMail or nil
end

local function GetUserTimezone (utils, userId, vcc)
	local fullUserId = utils.AppendVcc(utils.GetId(userId), vcc)

	local timezoneId = utils.GetColumnValue {
		keyspace     = AGENT_KS,
		columnFamily = USERS_CF,
		key          = fullUserId,
		column       = "TimeZoneId"
	}

	if not timezoneId then
		return
	end

	local timeZone = utils.GetAllColumns {
		keyspace = AGENT_KS,
		columnFamily = TIMEZONES_CF,
		key = timezoneId
	}

	if not timeZone then
		return
	end

	-- Retorna el mismo formato con el cual se encuentra en "inconcert.authUser.timezone"
	return {
		id      = timezoneId,                                  -- ej: "UYT",
		offset  = tonumber(timeZone.TimeZone),                 -- ej: -3,
		minutes = tonumber(timeZone.TimeZoneMinutes),          -- ej: 0,
		dst     = (tonumber(timeZone.DaylightSavingTime) ~= 0) -- ej: true
	}
end

local function GetSchedule (utils, sch_id)
	local schedule = GetScheduleById( utils, sch_id )
	if not schedule then
		LogError("Error deleting schedule id: '%s'. Schedule does not exist.", sch_id)
		return false, { result = "SCHEDULE_DOES_NOT_EXIST" }
	end
	return true, schedule
end

local function GetSchedules (utils, params)
	local schedules = utils.GetAllColumns {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_CF,
		startKey = "",
		endKey = ""
	}
	local vcc_schedules = {}
	local vcc = params and params.vcc and params.vcc:lower()
	for sch_id, schedule in pairs( schedules ) do
		if schedule.Vcc and (not vcc or vcc == schedule.Vcc:lower()) then
			vcc_schedules[sch_id] = schedule
		end
	end

	return vcc_schedules
end

local function GetSchedulesWithHistory (utils, params)
	local schedules = GetSchedules(utils, params)
	for sch_id, schedule in pairs(schedules) do
		local history = GetScheduleHistorical(utils, sch_id, params)
		schedule.__history = history
	end
	return schedules
end

local function UpdateState (utils, schedule, state)
	if not schedule then
		return
	end
	local sch_id = schedule.Id
	local now = LuaOrm.Date()
	local now_string = now:toString()
	local message_date = ""
	if state == RUNNING then
		message_date = " Running date " .. now_string
	end

	LogInfo("[REPORTER.SCHEDULER:UpdateState] Scheduler: Update State '%s', schedule id: '%s', %s.", state, sch_id, message_date )

	-- Elimino la columa running date, solo sirve mientras el schedule esta corriendo
	local columns_to_delete = {
		"RunningDate"
	}
	utils.RemoveColumns {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_CF,
		rows = {
			[ sch_id ] = columns_to_delete
		}
	}

	if state == RUNNING then
		schedule.RunningDate = now_string
	else
		schedule.RunningDate = nil
	end

	schedule.State = state
	utils.SaveRows {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_CF,
		rows = {
			[ sch_id ] = schedule
		}
	}
end

local function UpdateSchedulesByAgent (utils, schedule)
	LogInfo("Updating Schedules by Agent")
	-- actualizo o remuevo de la CF "ReportSchedulesByAgent"
	--  actualizo si existen mas de uno, y elimino si es la ultima
	local user_id = utils.AppendVcc( schedule.User, schedule.Vcc )
	local schedules_by_agent = GetSchedulesByAgent( utils, user_id )

	if not schedules_by_agent then
		return
	end
	local sch_id = schedule.Id

	for report_id, schedules_ids in pairs( schedules_by_agent ) do

		if report_id == schedule.Report then
			schedules_ids[ sch_id ] = nil
			local is_empty_list = true
			for _, v in pairs( schedules_ids ) do
				is_empty_list = false
				break
			end

			if is_empty_list then
			-- elimino
				utils.RemoveColumns {
					keyspace = SUMMARIZER_KS,
					columnFamily = SCHEDULES_BY_AGENT_CF,
					rows = { [ user_id ] = { report_id } }
				}
			else
			-- actualizo
				utils.SaveRows {
					keyspace = SUMMARIZER_KS,
					columnFamily = SCHEDULES_BY_AGENT_CF,
					rows = {
						[ user_id ] = {
							[ report_id ] = schedules_ids
						}
					}
				}
			end

			break
		end
	end
end

local function GetSendType (schedule)
	assert(schedule)
	return schedule.SendType or "mail"
end

local function IsSendType (utils, scheduleId, sendType)
	assert(utils)
	assert(scheduleId)
	assert(sendType)
	local success, schedule = GetSchedule(utils, scheduleId)
	--LogDebug("[REPORTER.SCHEDULER:IsSendType]   sendType:'%s' -> success: %s, schedule: '%s'", sendType, success, json.encode(schedule or {}))
	if not success then
		return false
	end
	return (schedule.SendType or ""):lower() == sendType:lower()
end

local function CanAddNewScheduler (utils, scheduleData)
--	LogDebug("[REPORTER.SCHEDULER:CanAddNewScheduler] %s", json.encode(scheduleData))

	local keyspace = utils.GetDatasource():CassandraConnection(SUMMARIZER_KS)
	--Primero busco el reporte como custom
	local report = manager.getReport(keyspace, scheduleData.Vcc, scheduleData.Report)


	if not report then
		-- Si no es custom busco uno system
		LogDebug("[REPORTER.SCHEDULER:CanAddNewScheduler] => Busco el reporte como system.")
		report = manager.getReport(keyspace, "system", scheduleData.Report)
		if not report then
			keyspace:Close()
			local message = string.format("Report %s does not exist.", scheduleData.Report)
			LogError("[REPORTER.SCHEDULER:CanAddNewScheduler] Error Adding schedule '%s'. %s", scheduleData.Id, message)
			return false, {result = "REPORT_DOES_NOT_EXIST", message = message}
		end
	end
	keyspace:Close()

	if report.schedulerInfo then
		if report.schedulerInfo.onlyAllowedOneInVcc then
			local newReport = scheduleData.Report:lower()
			local vccSchedulers = GetSchedules(utils, {vcc = scheduleData.Vcc})

			for currentSchId, currentScheduler in pairs(vccSchedulers or {}) do
				if (currentScheduler.Report or ""):lower() == newReport then
					--LogDebug("---> %s", json.encode(currentScheduler))
					local message = "only is allowed a one scheduler of Users connected by vcc"
					LogError("[REPORTER.SCHEDULER:CanAddNewScheduler] Error Adding schedule '%s'. %s. Current Scheduler: %s", scheduleData.Id, message, currentScheduler.Id)
					return false, {result = "ONLY_ALLOWED_A_ONE_SCHEDULER_OF_THIS_TYPE", message = message}
				end
			end
		end
	end

	-- is ok
	return true
end

local function AddSchedule (utils, redis, scheduleData)
	-- check basic data
	assert(type(scheduleData.Id) == "string", "invalid Id")
	assert(type(scheduleData.User) == "string", "invalid User")
	assert(type(scheduleData.Vcc) == "string", "invalid VCC")
	assert(type(scheduleData.Report) == "string", "invalid Report")
	assert(type(scheduleData.Params) == "table", "invalid Params")

	local sendType = GetSendType(scheduleData)
	if sendType == "mail" then
		assert(type(scheduleData.MailTo) == "string", "invalid MailTo")
		if scheduleData.AttachmentName then
			assert(type(scheduleData.AttachmentName) == "string" and scheduleData.AttachmentName ~= "", "invalid Mail AttachmentName")
		end
	elseif sendType == "ftp" then
		assert(type(scheduleData.ftp) == "table", "invalid FTP Config")
		if scheduleData.FileName then
			assert(type(scheduleData.FileName) == "string" and scheduleData.FileName ~= "", "invalid FTP FileName")
		end
	elseif sendType == "sftp" then
		assert(type(scheduleData.sftp) == "table", "invalid SFTP Config")
		if scheduleData.FileName then
			assert(type(scheduleData.FileName) == "string" and scheduleData.FileName ~= "", "invalid SFTP FileName")
		end
	elseif sendType == "database" then
		assert(type(scheduleData.database) == "table", "invalid DATABASE Config")
	else
		assert(sendType == "mail" or sendType == "ftp" or sendType == "database", "invalid Send Type Config")
	end

	local now = LuaOrm.Date()
	local old_schedule = GetScheduleById( utils, scheduleData.Id )

	if not old_schedule then
		LogInfo("[REPORTER.SCHEDULER:AddSchedule] Add new schedule '%s'", scheduleData.Id)

		local success, err = CanAddNewScheduler(utils, scheduleData)
		if not success then
			return false, err
		end

		scheduleData.CreationDate = now:toString()
		scheduleData.ExecutionCount = 0
	else
		LogInfo("[REPORTER.SCHEDULER:AddSchedule] Updating schedule '%s'", scheduleData.Id)

		-- Clean Up Old Values
		local columns_to_delete = {
			"RepeatPolicy", "RepeatHour", "CustomRepeatDays", "EndsAfter",
			"OneTimeDate", "OneTimeHour", "EndsOnDate", "EndsOnHour", "RepeatWeekday",
			"RepeatDay", "StartsOnDate", "StartsOnHour", "DaysToKeepContent", "RepeatEveryNHours",
			-- mail options
			"MailTo", "AttachmentName", "EmailSubject", "EmailBody", "EmailFrom",
			-- ftp options
			"ftp"
		}

		utils.RemoveColumns {
			keyspace = SUMMARIZER_KS,
			columnFamily = SCHEDULES_CF,
			rows = {
				[scheduleData.Id] = columns_to_delete
			}
		}
	end

	-- check repeat policy parameters
	if scheduleData.RepeatPolicy then

		if scheduleData.RepeatPolicy ~= "EveryNHours" then
			scheduleData.RepeatHour = ValidateHour(scheduleData.RepeatHour)
			assert(scheduleData.RepeatHour, "invalid RepeatHour")
		end
		if scheduleData.RepeatPolicy == "EveryNHours" then
			assert(type(scheduleData.RepeatEveryNHours) == "number" and
				scheduleData.RepeatEveryNHours == math.floor(scheduleData.RepeatEveryNHours) and
				scheduleData.RepeatEveryNHours >= 1, "invalid RepeatEveryNHours")

			scheduleData.StartsOnDate = ValidateDate(scheduleData.StartsOnDate)
			assert(scheduleData.StartsOnDate, "invalid StartsOnDate")

			scheduleData.StartsOnHour = ValidateHour(scheduleData.StartsOnHour)
			assert(scheduleData.StartsOnHour, "invalid StartsOnHour")
		end
		if scheduleData.RepeatPolicy == "Weekly" then
			assert(scheduleData.RepeatWeekday >= 1 and scheduleData.RepeatWeekday <= 7, "invalid RepeatWeekday")
		end
		if scheduleData.RepeatPolicy == "Monthly" then
			assert(scheduleData.RepeatDay >= 1 and scheduleData.RepeatDay <= 31, "invalid RepeatWeekday")
		end
		if scheduleData.RepeatPolicy == "CustomRepeat" then
			assert(scheduleData.CustomRepeatDays > 1, "invalid CustomRepeatDays")
			assert(ValidateDate(scheduleData.StartsOnDate), "invalid StartsOnDate")
			assert(ValidateHour(scheduleData.StartsOnHour), "invalid StartsOnHour")
		end

		if scheduleData.EndsOnDate then
			scheduleData.EndsOnDate = ValidateDate(scheduleData.EndsOnDate)
			assert(scheduleData.EndsOnDate, "invalid EndsOnDate")
			scheduleData.EndsOnHour = ValidateHour(scheduleData.EndsOnHour)
			assert(scheduleData.EndsOnHour, "invalid EndsOnHour")
		end

		if scheduleData.EndsAfter then
			assert(scheduleData.EndsAfter > 0, "invalid EndsAfter")
		end
	else
		scheduleData.OneTimeDate = ValidateDate(scheduleData.OneTimeDate)
		assert(scheduleData.OneTimeDate, "invalid OneTimeDate")
		scheduleData.OneTimeHour = ValidateHour(scheduleData.OneTimeHour)
		assert(scheduleData.OneTimeHour, "invalid OneTimeHour")
		scheduleData.EndsAfter = 1
	end

	local function BelongsTo(e, set)
		for _, a in ipairs(set) do
			if a == e then
				return true
			end
		end
		return false
	end

	if scheduleData.QueryRange.enabled then
		assert(type(scheduleData.QueryRange) == "table", "invalid QueryRange")
		local possibleTypes = {"SinceLastReport", "HoursBefore", "DaysBefore", "LastDay", "LastWeek", "LastMonth", "Range"}

		assert(BelongsTo(scheduleData.QueryRange.type, possibleTypes), "invalid type in QueryRange")

		if scheduleData.QueryRange.type == "HoursBefore" then
			assert(type(scheduleData.QueryRange.hours) == "number" and scheduleData.QueryRange.hours > 0, "hours in QueryRange must be positive")
		elseif scheduleData.QueryRange.type == "DaysBefore" then
			assert(type(scheduleData.QueryRange.days) == "number" and scheduleData.QueryRange.days > 0, "days in QueryRange must be positive")
		elseif scheduleData.QueryRange.type == "Range" then
			assert(ValidateDate(scheduleData.QueryRange.from), "from in QueryRange must be a valid date")
			assert(ValidateDate(scheduleData.QueryRange.to), "to in QueryRange must be a valid date")
		end
	end

	if scheduleData.StartsOnDate then
		local startDate = MakeStartDateHour( scheduleData.StartsOnDate, scheduleData.StartsOnHour )
		assert( startDate:time() >= now:time(), "StartsOnDate + StartsOnHour can not be set before today" )

		if scheduleData.EndsOnDate then
			local endDate = MakeEndDateHour( scheduleData.EndsOnDate, scheduleData.EndsOnHour )
			assert( startDate:time() <= endDate:time(), "StartsOnDate + StartsOnHour can not be after EndsOnDate  + EndsOnHour" )
		end
	end

	local nextTime = CalculateNextTimestamp(scheduleData)
	scheduleData.NextExecutionTime = nextTime

	utils.SaveRows {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_CF,
		rows = {
			[scheduleData.Id] = scheduleData
		}
	}

	local userId = scheduleData.User.."@"..scheduleData.Vcc

	local schedulesByAgent = utils.GetAllColumns {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_BY_AGENT_CF,
		key = userId,
	} or {}

	local schedulesByReport = schedulesByAgent[scheduleData.Report] or {}
	schedulesByReport[scheduleData.Id] = true

	utils.SaveRows {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_BY_AGENT_CF,
		rows = {
			[userId] = {
				[scheduleData.Report] = schedulesByReport
			}
		}
	}

	if old_schedule and old_schedule.User ~= scheduleData.User then
		-- si estoy actualizando, y es otro usuario quien esta actualizando debo actualizar la CF "ReportSchedulesByAgent"
		UpdateSchedulesByAgent( utils, old_schedule )
	end

	if nextTime then
		LogInfo("[REPORTER.SCHEDULER:AddSchedule] +Add / Update schedule '%s' to redis. Next execution time '%s'.", scheduleData.Id, LuaOrm.Date(nextTime):toString("%Y-%m-%d %H:%M:%S") )
		redis:zadd( SCHEDULES_REDIS, nextTime, scheduleData.Id )
		UpdateState( utils, scheduleData, ACTIVE ) -- siempre que se guarda (agrega o actualiza) el estado del sch pasa a estar "active" ( Si es que la fecha no ha caducado. )
	else
		LogInfo("[REPORTER.SCHEDULER:AddSchedule] -Remove schedule '%s' to redis. Next execution time '%s'.", scheduleData.Id, nextTime)
		local removed_from_redis = redis:zrem( SCHEDULES_REDIS, scheduleData.Id )
		if not removed_from_redis then
			LogWarning("[REPORTER.SCHEDULER:AddSchedule] Could not remove the schedule '%s' from redis.", scheduleData.Id)
		end
		UpdateState( utils, scheduleData, EXPIRED ) -- si la fecha caduco, lo dejo "expired"
	end

	return true
end

local function DeleteSchedule (utils, redis, scheduleData)

	assert(type(scheduleData) == "table", "invalid schedule information")
	assert(type(scheduleData.Id) == "string", "invalid Id")

	local sch_id = scheduleData.Id
	LogInfo("Deleting schedule id: '%s'.", sch_id)

	local schedule = GetScheduleById( utils, sch_id )
	if not schedule then
		LogError("Error deleting schedule id: '%s'. Schedule does not exist.", sch_id)
		return false, { result = "SCHEDULE_DOES_NOT_EXIST" }
	end

	-- remuevo de la CF "ReportSchedules"
	utils.RemoveColumns {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_CF,
		rows = { sch_id }
	}

	-- -- actualizo o remuevo de la CF "ReportSchedulesByAgent"
	UpdateSchedulesByAgent( utils, schedule )

	-- elimino de redis "ReportSchedules"
	local removed_from_redis = redis:zrem( SCHEDULES_REDIS, sch_id )
	if not removed_from_redis then
		LogWarning("Could not remove the schedule '%s' from redis.", sch_id)
	end

	LogInfo("Successfully deleted schedule id: '%s'.", sch_id)

	return true
end

local function UpdateSchedule (utils, redis, scheduleData)

	local nextTime = CalculateNextTimestamp(scheduleData)

	scheduleData.NextExecutionTime = nextTime

	if nextTime then
		redis:zadd(SCHEDULES_REDIS, nextTime , scheduleData.Id)
		UpdateState( utils, scheduleData, ACTIVE )  -- deja el estado nuevamente en active
	else
		UpdateState( utils, scheduleData, EXPIRED )  -- si expiro, se actualiza el estado
	end

	utils.SaveRows {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_CF,
		rows = {
			[scheduleData.Id] = scheduleData
		}
	}
end

local function GetNextSchedule (utils, redis)
	local data = redis:zrange(SCHEDULES_REDIS, 0, 0, "WITHSCORES")

	if type(data) == "table" then
		local _, pair = next(data)
		if pair then
			local scheduleId = pair[1]
			local timestamp = tonumber(pair[2])

			if timestamp <= os.time() then
				local removed = redis:zrem(SCHEDULES_REDIS, scheduleId)
				if removed then
					LogInfo("[REPORTER.SCHEDULER:GetNextSchedule] running schedule %s", scheduleId)
					return GetScheduleById(utils, scheduleId)
				else
					--someone has taken this schedule first
					--it is worth to try again in case there are more to execute
					--note: tail call is optimized by Lua
					return GetNextSchedule(utils, redis)
				end
			else
				LogInfo("[REPORTER.SCHEDULER:GetNextSchedule] next schedule %s will run on %s UTC", scheduleId, LuaOrm.Date(timestamp))
			end
		end
	end
	return nil
end

local function SendMail (irLink, to, subject, body, attachments, from, utils, vcc)

	local fromConfig
	if from then
		fromConfig = GetMailFromConfig( utils, vcc, from )
	else
		fromConfig = inconcert.environment.apps.agent.mailAccount
	end

	if not fromConfig then
		LogError("[REPORTER.SCHEDULER:SendMail] Unable to send mail. From account address does not exist. '%s'", fromConfig)
		return
	end

	LogInfo("[REPORTER.SCHEDULER:SendMail] Send Mail from:%s to:%s subject:%s\n\nbody:\n%s", fromConfig.address, to, subject, body )
	local response = irLink:SendIsolatedMail({
		account			= fromConfig,
		to              = to,
		cc              = cc or "",
		bcc             = bcc or "",
		subject         = subject,
		body            = body,
		content_type    = "text/plain",
		attachments     = (attachments and json.encode(attachments)) or ""
	})

	if response.result == "OK" then
		LogInfo("[REPORTER.SCHEDULER:SendMail] Successfully sending Isolated Mail.")
		return true
	else
		LogWarning("[REPORTER.SCHEDULER:SendMail] Failure sending Isolated Mail - %s", response.reason)
		return false
	end
end

--[[
	(1-**)
		Cuando en el Scheduler en "Rango" y "Frecuencia" ha configurado:
			"Rango"     -> "Desde la última ejecución".
			"Frecuencia"-> "Cada N Horas".
		El intervalo de la Fecha de Inicio debe de ser desde la hora del día de la última ejecución.
			Ej, si la última ejecución fue el 23/09/2020 a las 07:15, cuando se vuelva a ejecutar este Scheduler, la fecha y hora de Inicio va a ser:
				"2020-09-23 07:00:00.000"
			Como se puede ver, va a comenzar a la hora con 0 minutos de la última ejecución.
		Nota:
			Las fechas vienen dadas en UTC, por lo que se debe de indicar que son UTC para que cuando se ejecute el reporte, NO realice una doble conversión a UTC.

	(2-**) Solo para los Reportes de Summary
		Cuando en el Scheduler en el "Filtro" y el "Rango" ha configurado:
			"Filtro"    -> "Slice = Day".
			"Rango"     -> "Desde la última ejecución".
		El intervalo de la Fecha de Inicio debe de ser desde el día de la última ejecución.
			Ej, si la última ejecución fue el 23/09/2020 a las 07:15, cuando se vuelva a ejecutar este Scheduler, la fecha y hora de Inicio va a ser:
				"2020-09-23 00:00:00.000"
			Como se puede ver, va a comenzar a la hora 0 con 0 minutos del mismo día de la última ejecución.

	(3-**) Solo para los Reportes de Summary
		Cuando en el Scheduler en el "Filtro" y el "Rango" ha configurado:
			"Filtro"    -> "Slice = Month".
			"Rango"     -> "Desde la última ejecución".
		El intervalo de la Fecha de Inicio debe de ser desde el día de la última ejecución.
			Ej, si la última ejecución fue el 23/09/2020 a las 07:15, cuando se vuelva a ejecutar este Scheduler, la fecha y hora de Inicio va a ser:
				"2020-09-01 00:00:00.000"
			Como se puede ver, va a comenzar el primer día del y a la hora 0 con 0 minutos del mes de la última ejecución.
]]--
local function GetRangeSinceLastReport (scheduleToRun, endFormatDate)
	local summarySlicesSizeConfig = {
		["15min"] = {}, -- idem a hour
		["30min"] = {}, -- idem a hour
		hour = {
			ReportType = nil,
			QueryRange = {
				SinceLastReport = true
			},
			RepeatPolicy = {
				EveryNHours = true
			}
		},
		day = {
			ReportType = nil,
			QueryRange = {
				SinceLastReport = true
			},
			RepeatPolicy = {
				EveryNHours = false
			}
		},
		month = {
			ReportType = nil,
			QueryRange = {
				SinceLastReport = true
			},
			RepeatPolicy = {
				EveryNHours = false
			}
		}
	}
	summarySlicesSizeConfig["15min"] = summarySlicesSizeConfig.hour
	summarySlicesSizeConfig["30min"] = summarySlicesSizeConfig.hour

	local sliceSize = ((scheduleToRun.Params or {}).sliceSize or ""):lower()
	local currentSummarySliceConfig = summarySlicesSizeConfig[sliceSize] or {}

	local reportType = (scheduleToRun.ReportType or ""):lower()

	local dateFrom = LuaOrm.Date(tonumber(scheduleToRun.LastExecutionDate) or scheduleToRun.CreationDate)
	local dateTo = LuaOrm.Date() -- now
	local isUTC = false
	local stringFrom

	--LogDebug("------------------------------------------------------\n[SCHEDULER:GetRangeSinceLastReport] --> reportType: %s, Range: SinceLastReport, RepeatPolicy: %s, enabled: %s , sliceSize: %s", reportType, scheduleToRun.RepeatPolicy, currentSummarySliceConfig.RepeatPolicy.EveryNHours, sliceSize)

	if reportType ~= "summary" then
		--LogDebug("1")
		if scheduleToRun.RepeatPolicy == "EveryNHours" then
			--(1-**)
			stringFrom = dateFrom:toString("%Y-%m-%d %H:00:00.000") -- "2020-09-22 10:00:00.000"
			isUTC = true
		end
	else
		--LogDebug("2")
		if scheduleToRun.RepeatPolicy == "EveryNHours" and currentSummarySliceConfig.RepeatPolicy.EveryNHours then
			--(1-**)
			stringFrom = dateFrom:toString("%Y-%m-%d %H:00:00.000") -- "2020-09-22 10:00:00.000"
			isUTC = true
		else
			if sliceSize == "day" then
			--(2-**)
				stringFrom = dateFrom:toString("%Y-%m-%d 00:00:00.000") -- "2020-09-22 00:00:00.000"
				isUTC = false
			elseif sliceSize == "month" then
			--(3-**)
				stringFrom = dateFrom:toString("%Y-%m-01 00:00:00.000") -- "2020-09-01 00:00:00.000"
				isUTC = false -- no tiene que ser UTC
			end
		end
	end

	if not stringFrom then
		stringFrom = dateFrom:dayDate():toString(endFormatDate)
	end
	local stringTo =  dateTo:dayEndDate():toString(endFormatDate) -- "2020-09-22 23:59:59.999"

	--LogDebug("[SCHEDULER:GetRangeSinceLastReport] ==> reportType: %s, Frequency: '%s'. Between Dates: '%s -> %s'\n------------------------------------------------------", reportType, scheduleToRun.RepeatPolicy, stringFrom, stringTo)

	return stringFrom, stringTo, isUTC
end

-- creates a date interval based on the scheduler's QueryRange
local function GetQueryInterval (scheduleToRun, formatDate)
	--LogDebug("[SCHEDULER:GetQueryInterval] formatDate: '%s', scheduleToRun: '%s'", formatDate, json.encode(scheduleToRun or {}))

	local isUTC = false
	local now = LuaOrm.Date()
	local dateFrom, dateTo
	local queryRange = scheduleToRun.QueryRange.type

	if queryRange == "Range" then
		dateFrom = LuaOrm.Date(scheduleToRun.QueryRange.from)
		dateTo = LuaOrm.Date(scheduleToRun.QueryRange.to)
	elseif queryRange == "DaysBefore" then
		dateFrom = now:add(0, 0, -scheduleToRun.QueryRange.days)
		dateTo = now
	elseif queryRange == "HoursBefore" then
		dateFrom = now:add(0, 0, 0, -scheduleToRun.QueryRange.hours):toString("%Y-%m-%d %H:00:00.000")
		dateTo = now:add(0, 0, 0, -1):toString("%Y-%m-%d %H:59:59.000")
		return dateFrom, dateTo, true
	elseif queryRange == "SinceLastReport" then
		dateFrom, dateTo, isUTC = GetRangeSinceLastReport(scheduleToRun, formatDate)
		return dateFrom, dateTo, isUTC
	elseif queryRange == "LastDay" then
		dateFrom = now:add(0, 0, -1)
		dateTo = now:add(0, 0, -1)
	elseif queryRange == "LastWeek" then
		dateTo = now:add(0, 0, -now:dayOfWeek())
		dateFrom = dateTo:add(0, 0, -6)
	elseif queryRange == "LastMonth" then
		dateTo = now:add(0, 0, -now:day())
		dateFrom = dateTo:monthDate()
	elseif queryRange == "ALL" then
		return nil, nil
	else
		if scheduleToRun.QueryRange.enabled then
			error("Invalid QueryRange")
		end
		return nil, nil
	end

	local queryDateFrom = dateFrom:dayDate():toString(formatDate)
	local queryDateTo = dateTo:dayEndDate():toString(formatDate)

	--LogDebug("[SCHEDULER:GetQueryInterval] 2 ==> Range: '%s', Frequency: '%s'. Between Dates: '%s -> %s'", queryRange, scheduleToRun.RepeatPolicy, dateFrom, dateFrom)

	return queryDateFrom, queryDateTo, isUTC
end

local function ActivateSchedule (utils, redis, scheduleData)
	assert(type(scheduleData) == "table", "invalid schedule information")
	assert(type(scheduleData.Id) == "string", "invalid Id")
	assert(type(scheduleData.User) == "string", "invalid User")
	assert(type(scheduleData.Vcc) == "string", "invalid VCC")

	local sch_id = scheduleData.Id
	LogInfo("Activating schedule id: '%s'.", sch_id)

	local schedule = GetScheduleById( utils, sch_id )
	if not schedule then
		LogError("Error activating schedule id: '%s'. Schedule does not exist.", sch_id)
		return false, { result = "SCHEDULE_DOES_NOT_EXIST" }
	end

	local NEW_STATE = ACTIVE
	if schedule.State == NEW_STATE then
		LogError("Error activating schedule id: '%s'. Schedule is already in that '%s' state.", sch_id, NEW_STATE)
		return false, { result = "SCHEDULE_ALREADY_IN_THAT_STATE" }
	end

	schedule.State = NEW_STATE -- siempre que se guarda (agrega o actualiza) el estado del sch pasa a estar "active"
	utils.SaveRows {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_CF,
		rows = {
			[schedule.Id] = schedule
		}
	}

	local nextTime = schedule.NextExecutionTime
	if nextTime then
		redis:zadd( SCHEDULES_REDIS, nextTime , sch_id )
	end

	LogInfo("Successfully activated schedule id: '%s'.", sch_id)

	return true
end

local function DeactivateSchedule (utils, redis, scheduleData)

	assert(type(scheduleData) == "table", "invalid schedule information")
	assert(type(scheduleData.Id) == "string", "invalid Id")
	assert(type(scheduleData.User) == "string", "invalid User")
	assert(type(scheduleData.Vcc) == "string", "invalid VCC")

	local sch_id = scheduleData.Id
	LogInfo("[REPORTER.SCHEDULER:DeactivateSchedule] Deactivating schedule id: '%s'.", sch_id)

	local schedule = GetScheduleById( utils, sch_id )
	if not schedule then
		LogError("[REPORTER.SCHEDULER:DeactivateSchedule] Error deactivating schedule id: '%s'. Schedule does not exist.", sch_id)
		return false, { result = "SCHEDULE_DOES_NOT_EXIST" }
	end

	local NEW_STATE = INACTIVE
	if schedule.State == NEW_STATE then
		LogError("[REPORTER.SCHEDULER:DeactivateSchedule] Error deactivating schedule id: '%s'. Schedule is already in that '%s' state.", sch_id, NEW_STATE)
		return false, { result = "SCHEDULE_ALREADY_IN_THAT_STATE" }
	end

	schedule.State = NEW_STATE -- siempre que se guarda (agrega o actualiza) el estado del sch pasa a estar "active"
	utils.SaveRows {
		keyspace = SUMMARIZER_KS,
		columnFamily = SCHEDULES_CF,
		rows = {
			[schedule.Id] = schedule
		}
	}

	-- elimino de redis "ReportSchedules"
	local removed_from_redis = redis:zrem( SCHEDULES_REDIS, sch_id )
	if not removed_from_redis then
		LogWarning("[REPORTER.SCHEDULER:DeactivateSchedule] Could not remove the schedule '%s' from redis.", sch_id)
	end

	LogInfo("[REPORTER.SCHEDULER:DeactivateSchedule] Successfully deactivated schedule id: '%s'.", sch_id)

	return true
end

local function UrlEncode (str)
	str = string.gsub(str, "\n", "\r\n")
	str = string.gsub(str, "([^%w%-%_%.%~])", function (c)
		return string.format ("%%%02X", string.byte(c))
	end)
	return str
end

-- Gets the name of the exported file
-- prefers the name from the content disposition
-- over the report's name
local function GetFilenameFromHeaders (headers, scheduleToRun)
	local fileName
	local sendType = GetSendType(scheduleToRun)

	if sendType == "mail" and scheduleToRun.AttachmentName then
		fileName = scheduleToRun.AttachmentName
	elseif sendType == "ftp" and scheduleToRun.ftp then
		fileName = scheduleToRun.ftp.FileName
	elseif sendType == "sftp" and scheduleToRun.sftp then
		fileName = scheduleToRun.sftp.FileName
	end

	if not fileName then
		local disposition = headers["content-disposition"]
		if disposition then
			fileName = disposition:match("filename=([^;]+)")
		else
			fileName = scheduleToRun.Report
		end
	end

	local fileExtension = scheduleToRun.ExportFormat or "CSV"
	fileName = fileName .. "." .. fileExtension

	return fileName
end

local function UploadToRepository (fileName, content)
	local id = uuid.new():gsub("%-", "")
	local wsapiUtil = require "wsapi.util"
	local encodedFileName = wsapiUtil.url_encode(fileName)

	--LogDebug("[SCHEDULER:UploadToRepository] fileName: '%s', encodedFileName: '%s'", fileName, encodedFileName)

	--TODO: poner ip y port del repositorio posta e intentar con varios de ellos si uno falla
	local url = "http://127.0.0.1:8082/inconcert/apps/repository/add/?id="..id.."&section=-1&contentType=text/csv&format=csv&fileName="..encodedFileName

	local requestBody = content
	local responseBody = {}
	local result, code, headers, httpStatus = socket.http.request{
		url = url,
		method = "POST",
		headers = {
			["Content-Type"] = "text/csv",
			["Content-Length"] = #requestBody,
		},
		sink = ltn12.sink.table(responseBody),
		source = ltn12.source.string(requestBody)
	}
	if code == 200 then
		return true, id
	else
		LogError("[SCHEDULER:UploadToRepository] -> url: '%s'\n result: '%s', code: '%s', headers: '%s', httpStatus: '%s'", url, result, code, headers, httpStatus)
		local err = {
			result = result,
			code = code,
			headers = headers,
			httpStatus = httpStatus
		}
		return false, err
	end
end

local function FillTemplate (template, fields)
	return template:gsub("%%(.-)%%", function(a) return fields[a] end)
end

local function RequestExportReport (utils, authFilter, prefix, scheduleToRun)
	LogDebug("[REPORTER.SCHEDULER:RequestExportReport] - Export scheduleToRun Id %s", scheduleToRun.Id)
	--LogDebug("[REPORTER.SCHEDULER:RequestExportReport] - Export scheduleToRun %s", json.encode(scheduleToRun))

	local params = {
		export_type = scheduleToRun.ExportFormat or "csv",
		utc_offset  = 0, --TODO: usar el offset real del usuario
		utz_endDate = false -- Flag to check if have to convert endDate to user timezone in makeQueryRunner function located in file: agent/inconcert/apps/reports/init.lua
	}
	params.filter = utils.DeepCopy(scheduleToRun.Params) or {}
	params.filter.sliceSize = params.filter.sliceSize or "15Min"

	local dateFrom, dateTo, isUTC = GetQueryInterval(scheduleToRun)

	if scheduleToRun.QueryRange.type == "HoursBefore" and scheduleToRun.SendType ~= "database" then
		-- #14979 changes the startDate to user localtime when is not database, so we need to normalize both dates (start & end).
		params.utz_endDate = true
	end

	params.filter.DateFrom = dateFrom
	params.filter.DateTo = dateTo

	-- #5408 los reportes de tipo "realtime" necesitan las fechas en estos 2 campos
	params.filter.startDate = dateFrom
	params.filter.endDate = dateTo
	params.filter.isUTC = isUTC

	-- Hack: since detail reports use another table for parameters we send both tables
	params.queryFields = utils.DeepCopy(params.filter)
	params.queryFields.startDate = dateFrom
	params.queryFields.DateFrom = nil
	params.queryFields.endDate = dateTo
	params.queryFields.DateTo = nil
	params.queryFields.isUTC = isUTC

	if params.queryFields.withContent then
		params.queryFields.withContent = nil
		params.queryFields.saveContentTo = uuid.new()
		params.exportContentType = scheduleToRun.ExportContentType
		--HACK: find a better way to export the whole content by chunks
		--by now repository can't upload by chunks
		params.resultSize = 10^6
	end
	params.queryFields.scheduleId = scheduleToRun.Id
	params.executionType = "BY SCHEDULER"

	--LogDebug("[REPORTER.SCHEDULER:RequestExportReport] 2----------------> params.queryFields.scheduleId: %s", params.queryFields.scheduleId)

	local user = scheduleToRun.User
	local vcc = scheduleToRun.Vcc
	local report = scheduleToRun.Report

	socket.http.TIMEOUT = RUNNING_TIMEOUT --12h
	params.timeout = socket.http.TIMEOUT

	local oneTimeToken = authFilter:GetOneTimeRequestToken(user, vcc)
	local responseBody = {}
	local encodedParams = json.encode(params)
	local baseURL = "http://"..prefix.."/export/"..report.."?data="
	local url = baseURL .. UrlEncode(encodedParams)

	--LogDebug("[REPORTER.SCHEDULER:RequestExportReport] Export report URL: %s", baseURL)
	--LogDebug("[REPORTER.SCHEDULER:RequestExportReport] encodedParams: %s", encodedParams)

	local result, code, headers, httpStatus = socket.http.request{
		url = url,
		method = "GET",
		headers = {
			["Cookie"] = authFilter.AUTH_SID.."="..oneTimeToken,
		},
		sink = ltn12.sink.table(responseBody)
	}

	--LogDebug("[REPORTER.SCHEDULER:RequestExportReport] responseBody: %s", responseBody)

	local fileContent = table.concat(responseBody)
	local fileName = GetFilenameFromHeaders(headers, scheduleToRun) or report

	if responseBody and responseBody[#responseBody] == "REQUEST_HTTP_TIMEOUT" then
		code = 408
		httpStatus = "timeout"
		headers = nil
	end

	if code ~= 200 then
		LogError("[SCHEDULER:RequestExportReport] -> url: '%s'\n result: '%s', code: '%s', headers: '%s', httpStatus: '%s'", url, result, code, headers, httpStatus)
	end

	return result, code, headers, httpStatus, params, fileName, fileContent
end

--[[
	Funcion encargada de ejecutar el reporte con envio por MAIL
]]--
local function ExecuteReportToMail (utils, irLink, authFilter, prefix, scheduleToRun, executionType)
	LogInfo("[REPORTER.SCHEDULER:ExecuteReportToMail] MAIL execution schedule Id '%s'.", scheduleToRun.Id)

	local result, code, headers, httpStatus, params, fileName, fileContent = RequestExportReport(utils, authFilter, prefix, scheduleToRun)
	local success = true
	local errMsg = {subject = nil, body = nil, reason = nil}
	local now = LuaOrm.Date()
	local userId = utils.GetCurrentLoggedUser()

	local function AddSuccessHistory()
		local data = {id = uploadId, type = "text/csv", name = fileName, sendType = "MAIL", executionType = executionType, userId = userId}
		AddScheduleHistorical(utils, scheduleToRun.Id, now, true, data)
	end

	local function AddErrorHistory(data)
		data.sendType = "MAIL"
		data.executionType = executionType
		data.userId = userId
		AddScheduleHistorical(utils, scheduleToRun.Id, now, false, data)
	end

	if code == 200 then
		--all right
		local result, uploadId = UploadToRepository(fileName, fileContent)

		if result then
			local vcc = scheduleToRun.Vcc
			local CONTENT_TYPE = {
				csv = "text/csv",
				pdf = "application/pdf"
			}
			local fields = {REPORT = fileName, SCHEDULE_ID = scheduleToRun.Id, DATE_FROM = dateFrom, DATE_TO = dateTo, EXPORT_FILE_LINK = ""}

			local attachments = {{id = uploadId, type = CONTENT_TYPE[params.export_type] or "application/octet-stream", name = fileName}}
			if params.queryFields.saveContentTo then
				-- Obtengo el cabezal del archivo del Repo para obtener el tamaño del archivo.

				local repositoryFileId = params.queryFields.saveContentTo.."[-1]"
				local fileOnRepo = utils.GetAllColumns {
					keyspace = "Repository",
					columnFamily = "Files",
					key = repositoryFileId
				} or {}

				LogInfo("[REPORTER.SCHEDULER:ExecuteReportToMail] Retrieved FileId: '%s', File: '%s'.", repositoryFileId, json.encode(fileOnRepo))

				if fileOnRepo.FileSize then
					fields["EXPORT_FILE_LINK"] = "http://"..inconcert.environment.apps.agent.base_domain.."/inconcert/apps/repository/get/?id="..params.queryFields.saveContentTo.."&section=-1&part=0"
					local removeDate = now:add(0, 0, tonumber(scheduleToRun.DaysToKeepContent) or 1)
					RegisterExportedContent(utils, vcc, scheduleToRun.Id, fileOnRepo.FileSize, removeDate, params.queryFields.saveContentTo)
				else
					LogWarning("[REPORTER.SCHEDULER:ExecuteReportToMail] FileId: '%s' does not exist in repository.", repositoryFileId)
					fields["EXPORT_FILE_LINK"] = "The content file is missing. Maybe your quota for exported content has been exceeded."
				end
			end

			local ok = SendMail(
				irLink,
				scheduleToRun.MailTo,
				FillTemplate(scheduleToRun.EmailSubject or DEFAULT_SUBJECT, fields),
				FillTemplate(scheduleToRun.EmailBody or DEFAULT_BODY, fields),
				attachments,
				scheduleToRun.EmailFrom,
				utils,
				vcc
			)

			if not ok then
				success = false
				errMsg.subject = "Failed to execute report " .. fileName
				errMsg.body = "Some problem occurred during sending mail of exported report: " .. fileName
				errMsg.reason = "Error sending mail of exported report."

				LogWarning("[REPORTER.SCHEDULER:ExecuteReportToMail] Some problem occurred during sending mail of exported report: %s", uploadId)

				AddErrorHistory({id = uploadId, reason = errMsg.reason})

				if params.queryFields.saveContentTo then
					-- Elimino el file del repository
					local url = "http://127.0.0.1:8082/inconcert/apps/repository/delete/?id="..params.queryFields.saveContentTo.."&section=-1"
					socket.http.request (url)
				end
			else
				AddSuccessHistory()
			end
		else
			local uploadErr = uploadId
			success = false
			errMsg.subject = "Failed to execute report " .. fileName
			errMsg.body = "Some problem occurred uploading your report " .. fileName
			errMsg.reason = "Error uploading report. " .. (uploadErr.httpStatus or uploadErr.code or "")

			LogWarning("[REPORTER.SCHEDULER:ExecuteReportToMail] Some problem occurred during uploading of exported report. Err: %s", uploadErr)

			AddErrorHistory({id = uploadId, reason = errMsg.reason})

			if params.queryFields.saveContentTo then
				-- Elimino el file del repository
				local url = "http://127.0.0.1:8082/inconcert/apps/repository/delete/?id="..params.queryFields.saveContentTo.."&section=-1"
				socket.http.request (url)
			end
		end
	else
		success = false
		errMsg.subject = "Failed to execute report " .. fileName
		errMsg.body = "Some problem occurred during execution of your report " .. fileName
		errMsg.reason = "Error during execution report. " .. (type(httpStatus) == "string" and httpStatus or "")

		LogWarning("[REPORTER.SCHEDULER:ExecuteReportToMail] Some problem occurred during execution of request. Err: %s", httpStatus)

		AddErrorHistory({id = uploadId, reason = errMsg.reason, result = result, code = code, headers = headers, httpStatus = httpStatus})

		if params.queryFields.saveContentTo then
			-- Elimino el file del repository
			local url = "http://127.0.0.1:8082/inconcert/apps/repository/delete/?id="..params.queryFields.saveContentTo.."&section=-1"
			socket.http.request (url)
		end
	end

	return success, errMsg
end

--[[
	Funcion encargada de ejecutar el reporte con upload hacia FTP
]]--
local function ExecuteReportToFTP (utils, authFilter, prefix, scheduleToRun, executionType)
	LogInfo("[REPORTER.SCHEDULER:ExecuteReportToFTP] FTP execution schedule Id '%s'.", scheduleToRun.Id)
	return reporterFtpController.ExecuteReport(utils, authFilter, prefix, scheduleToRun, executionType, "FTP")
end

--[[
	Funcion encargada de ejecutar el reporte con upload hacia SFTP
]]--
local function ExecuteReportToSFTP (utils, authFilter, prefix, scheduleToRun, executionType)
	LogInfo("[REPORTER.SCHEDULER:ExecuteReportToSFTP] SFTP execution schedule Id '%s'.", scheduleToRun.Id)
	return reporterFtpController.ExecuteReport(utils, authFilter, prefix, scheduleToRun, executionType, "SFTP")
end

--[[
	Funcion encargada de ejecutar el reporte hacia la DataBase
]]--
local function ExecuteReportToDataBase (utils, authFilter, prefix, scheduleToRun, executionType)
	LogInfo("[REPORTER.SCHEDULER:ExecuteReportToDataBase] DATABASE execution schedule Id '%s'.", scheduleToRun.Id)
	--LogInfo("[REPORTER.SCHEDULER:ExecuteReportToDataBase] DATABASE execution schedule '%s'.", json.encode(scheduleToRun))
	if scheduleToRun.ReportType == "dialercontacts" then 
		return reporterDataBase.ExecuteReportES(utils, authFilter, prefix, scheduleToRun, executionType)
	end
	return reporterDataBase.ExecuteReport(utils, authFilter, prefix, scheduleToRun, executionType)
end

--[[
	Funcion que incluye en el Sch, informacion del reporte, en el atritubuto "__ReportInfo".
	Ej: si el Scheduler esta utlizando un Reporte Custom o System.
]]--
local function IncludeReportInfoToSchedule (schedule, utils)
	assert(schedule)
	assert(utils)

	--LogDebug("[REPORTER:IncludeReportInfoToSchedule] Including Rpt Info to schedule id %s, vcc %s", schedule.Id, schedule.Vcc)

	local reportId = schedule.Report or ""
	for _, vcc in pairs({"system", schedule.Vcc}) do
		local encodedReport = utils.GetColumnValue({
			keyspace = SUMMARIZER_KS,
			columnFamily = "Reports",
			key = vcc,
			column = reportId
		})
		--LogDebug("[REPORTER:IncludeReportInfoToSchedule] ---> encodedReport: %s", encodedReport)
		if encodedReport then
			local reportInfo = json.decode(encodedReport)
			reportInfo.isCustomReport = vcc == schedule.Vcc
			schedule.__ReportInfo = reportInfo
			break
		end
	end
end

local function ExecuteReport (utils, irLink, authFilter, prefix, scheduleToRun, redis, executionType)
	executionType = executionType or "BY SCHEDULER"
	LogInfo("[REPORTER.SCHEDULER:ExecuteReport] START execution schedule Id '%s', Execution Type: '%s'.", scheduleToRun.Id, executionType)

	UpdateState(utils, scheduleToRun, RUNNING)
	-- Inlcuyo Informacion del Reporte en el Schedule
	IncludeReportInfoToSchedule(scheduleToRun, utils)

	local success = false
	local errMsg = {subject = nil , body = nil}
	local sendType = GetSendType(scheduleToRun)

	if sendType == "mail" then
		success, errMsg = ExecuteReportToMail(utils, irLink, authFilter, prefix, scheduleToRun, executionType)
	elseif sendType == "ftp" then
		success, errMsg = ExecuteReportToFTP(utils, authFilter, prefix, scheduleToRun, executionType)
	elseif sendType == "sftp" then
		success, errMsg = ExecuteReportToSFTP(utils, authFilter, prefix, scheduleToRun, executionType)
	elseif sendType == "database" then
		success, errMsg = ExecuteReportToDataBase(utils, authFilter, prefix, scheduleToRun, executionType)
	else
		LogWarning("[REPORTER.SCHEDULER:ExecuteReport] Sending type '%s' has not been implemented. Report: '%s'", sendType, scheduleToRun.Id)
	end

	scheduleToRun.ExecutionCount = (tonumber(scheduleToRun.ExecutionCount) or 0) + 1

	if success then
		scheduleToRun.ExecutionSuccess = (tonumber(scheduleToRun.ExecutionSuccess) or 0) + 1
		scheduleToRun.ExecutionTemporalFailed = 0
	else

		LogError("[REPORTER.SCHEDULER:ExecuteReport] Error executing schedule Id: '%s', err: '%s'.", scheduleToRun.Id, json.encode(errMsg))

		scheduleToRun.ExecutionFailed = (tonumber(scheduleToRun.ExecutionFailed) or 0) + 1
		scheduleToRun.ExecutionTemporalFailed = (tonumber(scheduleToRun.ExecutionTemporalFailed) or 0) + 1

		local hasExceededNumberOfAttempts = MAX_ALLOWED_FAILED <= scheduleToRun.ExecutionTemporalFailed
		if hasExceededNumberOfAttempts then
			LogWarning("[REPORTER.SCHEDULER:ExecuteReport] Schedule '%s' has has reached %d failed attempts.", scheduleToRun.Id, scheduleToRun.ExecutionTemporalFailed)
		end

		if scheduleToRun.MailTo or scheduleToRun.EmailFrom then
			local body = errMsg.body or ""
			if hasExceededNumberOfAttempts then
				body = body .. "\n\n" .. string.format("Schedule '%s' has has reached %d failed attempts.", scheduleToRun.Id, scheduleToRun.ExecutionTemporalFailed)
			end

			SendMail(
				irLink,
				scheduleToRun.MailTo, 
				errMsg.subject or ("Error executing report " .. scheduleToRun.Id),
				body,
				nil,
				scheduleToRun.EmailFrom,
				utils,
				scheduleToRun.Vcc
			)
		else
			LogWarning("[REPORTER.SCHEDULER:ExecuteReport] Unable to send mail notificaction error. Invalid mail addresses from: '%s', to: '%s'.", scheduleToRun.EmailFrom, scheduleToRun.MailTo)
		end
	end

	UpdateState(utils, scheduleToRun, ACTIVE)
	utils.CollectAllMemory()
	LogInfo("[REPORTER.SCHEDULER:ExecuteReport] END execution schedule Id '%s', Execution Type: '%s'..", scheduleToRun.Id, executionType)
end

local function RemoveOldExportedContent (utils)
	local records = utils.GetAllColumns {
		keyspace = SUMMARIZER_KS,
		columnFamily = EXPORTED_CONTENT_CF,
		startKey = "",
		endKey = "",
	} or {}

	local now = LuaOrm.Date()

	for vcc, record in pairs(records) do
		for id, data in pairs(record) do
			local removeDate = LuaOrm.Date(data.removeDate)
			if now:time() >= removeDate:time() then
				LogInfo("Removing exported content '%s' from vcc '%s'", id, vcc)

				local url = "http://127.0.0.1:8082/inconcert/apps/repository/delete/?id="..id.."&section=-1"
				socket.http.request (url)

				utils.RemoveColumns {
					keyspace = SUMMARIZER_KS,
					columnFamily = EXPORTED_CONTENT_CF,
					rows = {[vcc] = {id}}
				}
			end
		end
	end
end

local function UpdateLastExecutionDate (utils, redis, scheduleToRun, nowTime)
	assert(utils)
	assert(redis)
	assert(scheduleToRun)

	scheduleToRun.LastExecutionDate = nowTime or os.time()
	UpdateSchedule(utils, redis, scheduleToRun)
end

local function run (irLink, authFilter, utils, redis, prefix)
	LogDebug("[REPORTER.SCHEDULER:run] START Running ........")

	RemoveOldExportedContent(utils)

	local scheduleToRun = GetNextSchedule(utils, redis)

	local CanExecute = function( now_time )
		if scheduleToRun.StartsOnDate then
			local start = MakeStartDateHour( scheduleToRun.StartsOnDate, scheduleToRun.StartsOnHour )

			if now_time < start:time() then
				LogInfo("Execution of report %s was skipped. Start date is %s hour %s", scheduleToRun.Id, scheduleToRun.StartsOnDate, scheduleToRun.StartsOnHour)
				return false
			end
		end

		if scheduleToRun.EndsOnDate then
			local finish = MakeEndDateHour( scheduleToRun.EndsOnDate, scheduleToRun.EndsOnHour )

			if now_time > finish:time() then
				LogInfo("Execution of report %s was skipped. End date is %s hour %s", scheduleToRun.Id, scheduleToRun.EndsOnDate, scheduleToRun.EndsOnHour)
				return false
			end
		end

		return true
	end

	while scheduleToRun do
		local now_time = os.time()
		local can_execute = CanExecute( now_time )

		if not can_execute then
			UpdateLastExecutionDate(utils, redis, scheduleToRun, now_time)
		else
			local executionTime = tonumber(scheduleToRun.NextExecutionTime)

			if (now_time - executionTime) > 12 * 60 * 60 then

				--execution was delayed by too much time
				LogWarning("Execution of schedule %s was delayed by too much time. Notifying the user.", scheduleToRun.Id)

				UpdateLastExecutionDate(utils, redis, scheduleToRun, now_time)

				local message = "The execution for your report '"..scheduleToRun.Report.."' was delayed and will not run this time."
				if scheduleToRun.NextExecutionTime then
					local dateToRun = LuaOrm.Date(tonumber(scheduleToRun.NextExecutionTime)):toString("%Y-%m-%d %H:%M:%S")
					message = message.." It was scheduled to run on "..dateToRun.."."
				else
					message = message.." Repetition policy prevents it to run anymore."
				end

				if scheduleToRun.MailTo or scheduleToRun.EmailFrom then
					SendMail(
						irLink,
						scheduleToRun.MailTo,
						"Failed to execute report "..(scheduleToRun.AttachmentName or scheduleToRun.Report),
						message,
						nil,
						scheduleToRun.EmailFrom,
						utils,
						scheduleToRun.Vcc
					)
				end
			else
				LogInfo("[SCHDULER:run] Running report %s", scheduleToRun.Id)
				ExecuteReport(utils, irLink, authFilter, prefix, scheduleToRun, redis)
				UpdateLastExecutionDate(utils, redis, scheduleToRun, now_time)
			end
		end

		scheduleToRun = GetNextSchedule(utils, redis)
	end


	--TODO: implementar mecanismo de checkeo que revise que todas las tareas activas estan en redis
end

local function CleanupSchedules (utils, redis)
	local now = LuaOrm.Date()
	local now_time = LuaOrm.Date():time()

	local schedules = GetSchedules( utils )
	-- recorrer la lista de schedule en estado running y valida que no haya sobrepasado el tiempo de timeout (timeout)
	--  si llega a pasar el tiempo de timeout, actualiza el estado a "ACTIVE".
	local timeout = 60 * 15 -- 15 minutos
	for sch_id, schedule in pairs( schedules ) do
		--LogDebug("[REPORTER.SCHEDULER:CleanupSchedules] Scheduled '%s'. schedule.State: %s schedule.RunningDate: %s", sch_id,schedule.State,schedule.RunningDate)
		if schedule.State == RUNNING and schedule.RunningDate then
			local runningDate = LuaOrm.Date( schedule.RunningDate )
			local runningDate_time = runningDate:time()
			local expired = now_time > ( runningDate_time + timeout )
			local remaining = ( runningDate_time + timeout ) - now_time

			if expired then
				LogWarning("[REPORTER.SCHEDULER:CleanupSchedules] Scheduled '%s' report execution, has been canceled by timeout.", schedule.Id)
				local data = {
					id = sch_id, 
					sendType = "--",
					executionType = now,
					userId = "",
					reason = "detect running state for a lot of time. Resetting..."
				}
				AddScheduleHistorical(utils, sch_id, now, false, data)
				schedule.ExecutionTemporalFailed = (tonumber(schedule.ExecutionTemporalFailed) or 0) + 1
				UpdateSchedule( utils, redis, schedule )
			end
		end
	end

	return true
end

local function RequestRunReport (utils, authFilter, prefix, scheduleToRun)
	LogDebug("[REPORTER.SCHEDULER:RequestRunReport] Run report Schedule Id: %s", scheduleToRun.Id)
	--LogDebug("[REPORTER.SCHEDULER:RequestRunReport] scheduleToRun: %s", json.encode(scheduleToRun))

	local params = {
		export_type = scheduleToRun.ExportFormat or "csv",
		utc_offset = 0 --TODO: usar el offset real del usuario
	}
	params.filter = utils.DeepCopy(scheduleToRun.Params) or {}
	params.filter.sliceSize = params.filter.sliceSize or "15Min"

	local dateFrom, dateTo, isUTC = GetQueryInterval(scheduleToRun)

	params.filter.DateFrom = dateFrom
	params.filter.DateTo = dateTo

	-- #5408 los reportes de tipo "realtime" necesitan las fechas en estos 2 campos
	params.filter.startDate = dateFrom
	params.filter.endDate = dateTo
	params.filter.isUTC = isUTC

	-- Hack: since detail reports use another table for parameters we send both tables
	params.queryFields = utils.DeepCopy(params.filter)
	params.queryFields.startDate = dateFrom
	params.queryFields.DateFrom = nil
	params.queryFields.endDate = dateTo
	params.queryFields.DateTo = nil
	params.queryFields.isUTC = isUTC

	if params.queryFields.withContent then
		params.queryFields.withContent = nil
		params.queryFields.saveContentTo = uuid.new()
		params.exportContentType = scheduleToRun.ExportContentType
		--HACK: find a better way to export the whole content by chunks
		--by now repository can't upload by chunks
		params.resultSize = 10^6
	end
	params.resultSize = scheduleToRun.resultSize
	params.last_handle = scheduleToRun.last_handle
	
	params.queryFields.scheduleId = scheduleToRun.Id
	params.executionType = "BY SCHEDULER"

	--LogDebug("[REPORTER.SCHEDULER:RequestRunReport] 1----------------> params.queryFields.scheduleId: %s", params.queryFields.scheduleId)
	--LogDebug("[REPORTER.SCHEDULER:RequestRunReport] 1----------------> params: %s",json.encode(params))
	local user = scheduleToRun.User
	local vcc = scheduleToRun.Vcc
	local report = scheduleToRun.Report

	socket.http.TIMEOUT = RUNNING_TIMEOUT

	local oneTimeToken = authFilter:GetOneTimeRequestToken(user, vcc)
	local responseBody = {}
	local encodedParams = json.encode(params)
	local baseURL = "http://"..prefix.."/execute/"..report.."?data="
	local url = baseURL .. UrlEncode(encodedParams)

-- LogDebug("[REPORTER.SCHEDULER:RequestRunReport] Export report URL: %s", baseURL)
-- LogDebug("[REPORTER.SCHEDULER:RequestRunReport] encodedParams: %s", encodedParams)

	local result, code, headers, httpStatus = socket.http.request{
		url = url,
		method = "GET",
		headers = {
			["Cookie"] = authFilter.AUTH_SID.."="..oneTimeToken,
		},
		sink = ltn12.sink.table(responseBody)
	}

--LogDebug("[REPORTER.SCHEDULER:RequestRunReport] responseBody: %s", responseBody)

	local fileContent = table.concat(responseBody)
	local fileName = GetFilenameFromHeaders(headers, scheduleToRun) or report

	if code ~= 200 then
		LogError("[SCHEDULER:RequestRunReport] -> url: '%s'\n result: '%s', code: '%s', headers: '%s', httpStatus: '%s'", url, result, code, headers, httpStatus)
	end

	return result, code, headers, httpStatus, params, fileName, fileContent
end

local function Exec (irLink, authFilter, utils, redis, scheduleId, prefix)
	LogInfo("[REPORTER.SCHEDULER:Exec] Starting Manual Executing for Scheduler: '%s' ........", scheduleId)

	local scheduleToRun = GetScheduleById(utils, scheduleId)

	if not scheduleToRun then
		LogInfo("[REPORTER.SCHEDULER:Exec] Manual Execution Error.  Scheduler: '%s' does not exist.", scheduleId)
		local err = "SCHEDULER_DOES_NOT_EXIST"
		return false, err
	end

	ExecuteReport(utils, irLink, authFilter, prefix, scheduleToRun, redis, "MANUAL BY USER")
	UpdateLastExecutionDate(utils, redis, scheduleToRun, now_time)

	LogInfo("[REPORTER.SCHEDULER:Exec] End Manual Executing for Scheduler: '%s' ........", scheduleId)
	return true
end

--------------------------------------------------------------------------------
return {
	run = run,
	exec = Exec,

	get_schedules_with_history = GetSchedulesWithHistory,
	get_schedules              = GetSchedules,
	get_schedule               = GetSchedule,
	add_schedule               = AddSchedule,
	delete_schedule            = DeleteSchedule,
	activate_schedule          = ActivateSchedule,
	deactivate_schedule        = DeactivateSchedule,
	cleanup_schedules          = CleanupSchedules,
	update_state               = UpdateState,
	get_send_type              = GetSendType,
	is_send_type               = IsSendType,

	SUNDAY                     = SUNDAY,
	MONDAY                     = MONDAY,
	TUESDAY                    = TUESDAY,
	WEDNESDAY                  = WEDNESDAY,
	THURSDAY                   = THURSDAY,
	FRIDAY                     = FRIDAY,
	SATURDAY                   = SATURDAY,

	-- utilizados por el module FTP
	get_query_interval         = GetQueryInterval,
	get_user_timezone          = GetUserTimezone,
	add_schedule_historical    = AddScheduleHistorical,
	request_export_report      = RequestExportReport,

	-- utilizados por el module DATABASE
	request_run_report         = RequestRunReport
}
