-------------------------------------------------------------------------------
-- Copyright (c) 2016 Sierra Wireless. All rights reserved.
--
-- Contributors:
--     Randall Pfeifer - Built from AMMER (m3da.lua) initial implementation
--
--	Protocol Send Handler
--	Sends formated via configured protocol.
--	Handles persisting data across reboot in a way that minimizes flash wear.
--	Implements send latency, so that data is send in chunks to help improve efficiency.
--	Witholds sending until system time is known, ensuring all data is correctly timestamped.
-------------------------------------------------------------------------------

-- ----------------------------------------------------------------------------
-- REQUIRES
-- ----------------------------------------------------------------------------

local persist = require 'persist'
local queue_new = require 'queue'
local lock = require 'sched.lock'
local uptime = require 'sched.timer.core'
local timer = require 'timer'


-- ----------------------------------------------------------------------------
-- CONSTANTS
-- ----------------------------------------------------------------------------
local EVTID_BASE = 64					-- base of event id
local EVTID_PERSIST = 'evtid_base'		-- persist to save current evtid base

local EVTQ_PERSIST = 'evts_queued'		-- persist count & duration of queued events

local DEVT_INFO		= "info"			-- DELS event to inform AMM of info messages
local DEVT_WARN		= "warning"			-- DELS event to inform AMM of warning messages
local DEVT_ERROR	= "error"			-- DELS event to inform AMM of error messages

local REFTIME = 1512086400000			-- 2017-12-01 00:00:00 UTC - time in recent past (mS) against
										-- which to compare current time to see if RTC is valid


-- ----------------------------------------------------------------------------
-- Local Functions
-- ----------------------------------------------------------------------------

------------------------------------------------------------------
-- getevtid - get the next event id
-- 
-- Inputs:
--
-- Outputs:
--
-- Returns:
--	event id
--
-- Notes
------------------------------------------------------------------
local function getevtid(self)
	local rtn = self.eventid_base + self.eventid
	self.eventid = self.eventid + 1
	if self.eventid >= EVTID_BASE then
		self.eventid = 0
		self.eventid_base = self.eventid_base + EVTID_BASE
		persist.save(EVTID_PERSIST, self.eventid_base)
	end
	return rtn
end

------------------------------------------------------------------
-- getSystimeMs - get the system time in mS
-- 
-- Inputs:
--
-- Outputs:
--
-- Returns:
--	system time, in mS
--
-- Notes
--	Currently has resolution of 1000 mS, since that's limit of ALEOS
--	os.time. Could synthesize better resolution in future.
--
--	Guarantees that each call returns a unique time, even if multiple
--	calls made within clocks resolution
------------------------------------------------------------------
local function getSystimeMs(self)
	local now = os.time() * 1000

	if now <= self.last_now then
		-- Guarantee that time is unique.
		-- If > 1 call within clock resolution. increment
		-- mS to ensure it is unique
		now = self.last_now + 1
	end
	self.last_now = now

	return now
end

------------------------------------------------------------------
-- chktime - check if the RTC is valid
-- 
-- Inputs:
--
-- Outputs:
--
-- Returns:
--
-- Notes
--	Call to check if the RTC is valid, usually once at startup and,
--	if clock not correct then, then again when it becomes valid
--	(via devtree notification).
--
--	If time is valid then self.starttime will be nil, and data is
--	Call when the RTC becomes valid (via devtree notification) to
--	correct time stamps of events which were generated before the
--	actual time was known.
------------------------------------------------------------------
local function chktime(self)
	local now = getSystimeMs(self)
	local flushed = false
	-- compare current time to some date in the past to see it clock is valid instead
	-- of relying on MSCIID_STS_TIME_VALID, because there's a small window where where
	-- this flag may not be accurate.
	if now < self.reftime then
		-- Time not valid yet (app starting).
		if not self.starttime then
			self.starttime = now
			log(self.app_name, "WARNING", "RTC invalid - withholding transmission until time is known")
		end
--		log(self.app_name, "WARNING", "---------> RTC invalid: "..now)
	else
		if self.starttime then
			-- Time is now valid and a start time set (time not valid at app start)
			-- Push all of the data with invalid timestamps to AMM, correcting each
			local offset = now - self.starttime
			log(self.app_name, "INFO", "Time updated: sending "..self.tsqueue:len().." delayed events")

			-- Messing with the system messes with paho mqtt lib - fails to correctly send
			-- the initial 'catch-up' transmission. Re-start mqtt to avoid this problem.
			self.protocol:stop()
			self.protocol:run()

			while self.tsqueue:len() > 0 do
				flushed = true
				-- pop event, adjust timestamp, and send it
				local data = self.tsqueue:pop()
				data.timestamp = data.timestamp + offset

--				local msg = "Popped Event: "
--				for k,v in pairs(data) do msg = msg..k.."="..tostring(v)..", " end
--				log(self.app_name, "DEBUG", msg)

				if self:pushRow(data) ~= "ok" then
					log(self.app_name, "DETAIL", "->Event not pushed '"..tostring(data).."'")
				end
			end
			self.eqstat[#self.eqstat] = {}
			self.starttime = nil
		end
		-- Check for queued bad-time messages, send Miscellanious info message if there are any.
		-- These are created if GW reboots before time becomes valid, so they're lost. Gotta warn AMM.
		if #self.eqstat[#self.eqstat] > 0 then
			log(self.app_name, "ERROR", "Problem with dropped events list")
		else
			local now = getSystimeMs(self)
			local sent = false
			for i, losses in ipairs(self.eqstat) do
				local count = losses.count
				if count then
					flushed = true
					local span = losses.last - losses.first
					local msg = "Time unavailable: lost "..count.." events spanning "..span.." mS"
					-- push inform message directly instead of via inform()
					-- to avoid issues with lock/resursion issues calling send() here
					log(self.app_name, "WARNING", msg)
					-- disable sending for now - Bypasses msg config & rename
--					self:pushRow({	timestamp=now, warning = msg})
				end
			end
			if flushed then
				self.eqstat = { {} }
				persist.save(EVTQ_PERSIST, self.eqstat)
			end
		end
		
	end
end

-- ----------------------------------------------------------------------------
-- METHODS
-- ----------------------------------------------------------------------------

------------------------------------------------------------------
-- send - Member func to send the data to AMM
-- 
-- Inputs:
--	data		- table of DELS data parameters to send:
--					messageid
--					param0
--					param1
--					extra
--
-- Outputs:
--
-- Returns:
--
-- Notes
--	Sends data if form of set of parameter=value pairs to recieving
--	endpoint via configured protocol object. Tf the data contains a
--	parameter "timestamp" it will override the inserted timestamp.
--	If it's 0, the inserted timestamp will be the gateway's boot time.
--
--	If system time is not currently valid, the event(s) will be held
--	in a queue until the correct time is known, so that the timestamps
--	can be corrected and the events then sent with the correct time.
--	Call with null data when the time becomes valid to send the time-
--	corrected events immediately instead of waiting for next send.

--
--	Also keep a persisted record of the number of enqueued events and
--	the timestamp of 1st and last event. If a reboot occurs before the
--	time becomes known, it is impossible to re-construct the correct
--	timestamps, but an event can be created from this data to notify
--	of the duration and the number of affected events lost (discarded).
------------------------------------------------------------------
local function m_send(self, data)
    if self.running then
		-- Lock to prevent sending events while dumping the chktime queue
		self.lock:acquire()

		-- Check if there are any events whitheld due to invalid time stamps
		chktime(self)

		if type(data) == "table" then
			if data.timestamp and data.timestamp ~= 0 then
				-- An explicit timestamp specified. If valid, bypass time correction stuff and just send
				local timestamp = tonumber(data.timestamp)
				if timestamp and timestamp > REFTIME then
--					data.eventid = getevtid(self)
					if self:pushRow(data) ~= "ok" then
						log(self.app_name, "DETAIL", "->Event not pushed '"..tostring(data).."'")
					end
				else
					log(self.app_name, "WARNING", "Invalid explicit timestamp specified for event " .. tostring(data.timestamp))
				end
			else
				if data.timestamp == 0 then
					-- An event marked with a pre-existing timestamp == 0 adjusts timestamp
					-- to indicate boot time instead of app start time
					local ut = math.floor(uptime.time() * 1000)
					data.timestamp = getSystimeMs(self) - ut
					log (self.app_name, "DETAIL", "Started " .. os.date("%X", math.floor(ut / 1000)) .. " after boot")
				else
					-- timestamp not overridden, use current time
					data.timestamp = getSystimeMs(self)
				end

--				data.eventid = getevtid(self)

				if log.musttrace(self.app_name, "DETAIL") then
					local tmp = "Send:"
					for k,v in pairs(data) do tmp = tmp.."  "..tostring(k).."="..tostring(v) end
					log(self.app_name, "DETAIL", tmp)
				end

				if self.starttime then
					-- current time is invalid - enque data so time stamp can be fixed later
					self.tsqueue:push(data)
					if self.tsqueue:len() > self.maxqueue then
						-- Limit queue size by dropping oldest (log it, though)
						local dropped = self.tsqueue:pop()
						local msg = "Too many events queued - dropped: "
						for k,v in pairs(dropped) do msg = msg..k.."="..tostring(v)..", " end
						log(self.app_name, "DETAIL", msg)
					end

					if not self.eqstat[#self.eqstat].count then
						self.eqstat[#self.eqstat].count = 1
						self.eqstat[#self.eqstat].first = data.timestamp
					else
						self.eqstat[#self.eqstat].count = self.eqstat[#self.eqstat].count + 1
					end
--[[
					if log.musttrace(self.app_name, "DEBUG") then
						local msg = "Pushed Event: "
						for k,v in pairs(data) do msg = msg..k.."="..tostring(v)..", " end
						log(self.app_name, "DEBUG", msg)
					end
--]]
					self.eqstat[#self.eqstat].last = data.timestamp
					persist.save(EVTQ_PERSIST, self.eqstat)
					log(self.app_name, "DEBUG", "Invalid time: "..self.eqstat[#self.eqstat].count.." events spanning "..self.eqstat[#self.eqstat].last - self.eqstat[#self.eqstat].first.." seconds delayed")
				else
					if self:pushRow(data) ~= "ok" then
						log(self.app_name, "DETAIL", "->Event not pushed '"..tostring(data).."'")
					end
--[[
					if log.musttrace(self.app_name, "DEBUG") then
						local msg = "Sent Event: "
						for k,v in pairs(data) do msg = msg..k.."="..tostring(v)..", " end
						log(self.app_name, "DEBUG", msg)
					end
--]]
				end
			end

		elseif data ~= nil then
			log(self.app_name, "DETAIL", "->Event not pushed '"..tostring(data).."'")
		end
		self.lock:release()
	end
end

------------------------------------------------------------------
-- pushRow - push row of data to AMM
-- 
-- Inputs:
--	data		- table of DELS data parameters to send:
--					timestamp
--					eventid
--					messageid
--					param0
--					param1
--					extra
--
-- Outputs:
--
-- Returns:
--
-- Notes
--	Replacement self:pushRow for (table:pushRow) is used to provide
--	limited functionality of persisted table:pushData() functionality,
--	but with much less flash wear. Attempts to duplicate alogrithm used
--	by oMG:
--	- Send pushed data into a queue (RAM) and start latency timer (via signal) if not already running.
--	- When latency timer triggers, all available messages are 'peeked' from the queue and sent immediately (synchronous)
--	- If the send was successful then the 'peeked' messages are popped and discarded.
--	- If the send failed, send data is cleared, and re-loaded from queue on next send
--	- Peridically, if there are at least n messages in the queue, the queue is persisted to disk.
--	- Upon orderly shutdown the queue is persisted to disk.
--	- On startup, the queue is restored from persist.
------------------------------------------------------------------
local function m_pushRow(self, data)
	-- If max queue size reached send a inform warning. Can infer restoration by next message
	if self.pushqueue:len() == self.maxqueue - 1 then
-- disable sending notification for now - it bypasses msg config & rename. -1 prevents log spamming
--		self.pushqueue:push {
--								timestamp = getSystimeMs(self),
-- --								eventid = getevtid(self),
--								warning = "Max Tx queue size of " .. self.maxqueue .. " reached" }

		log(self.app_name, "WARNING", "Max Tx queue size of %d reached", self.maxqueue)
	end
	
	if self.pushqueue:len() < self.maxqueue then
		self.pushqueue:push(data)
	end
	sched.signal(self.app_name, "SEND")

	return "ok"
end

------------------------------------------------------------------
-- inform - Inform AMM of internal conditions/errors
-- 
-- Inputs:
--	status	- "warning", "error", or "info"
--	fmt		- string format that holds the log text the same way as string.format does
--	varargs	- additional arguments can be provided (as with string.format)
--
-- Outputs:
--
-- Returns:
--
-- Notes
------------------------------------------------------------------
local function m_inform(self, status, fmt, ...)
	local msg = string.format(fmt, ...)
	local evts = {info=DEVT_INFO, warning=DEVT_WARN, error=DEVT_ERROR}
	
	if evts[status] then
		self:send {[evts[status]] = msg}
		log(self.app_name, "INFO", "Inform AMM %s: %s", status, msg)
	else
		log(self.app_name, "WARNING", "Unrecognized inform status %s: %s", status, msg)
	end
end

-- Timer callback to peridically save the Tx queue in case of unexpected power loss.
-- Only save if events are past threshold to prevent minimal unsent data from creating
-- excessive saves. Note that queue save has built-in checks to eleminate redundant writes, too.
local function savequeue(self)
	local size = self.pushqueue:len()
	if (size == 0 or size >= self.persistthresh) and self.pushqueue:save(self.persistname, false) then
		log(self.app_name, "DETAIL", "Persisted %d records to Tx queue", self.pushqueue:len())
	else
		log(self.app_name, "DEBUG", "queue not saved")
	end
end

------------------------------------------------------------------
-- run - Start transport send operations
-- 
-- Inputs:
--
-- Outputs:
--
-- Returns:
--
-- Notes
--	Start transport send operations coroutine.
--	When triggered by "SEND" signal, wait for latency time, and then
--	send all available data from the Tx queue and trigger the
--	send policy to send it. If data is successfully sent, remove it
--	from the Tx queue.
------------------------------------------------------------------
local function m_run(self)
	if self.running then
		log(self.app_name, "WARNING", "Attempted to start already-running transport task")
		return
	end

    self.running = true
	
	self.pushqueue:restore(self.persistname)
	log(self.app_name, "INFO", "Restored %d Persisted records to Tx queue", self.pushqueue:len())

	if self.persistrate > 0 then
		self.persisttimer = timer.periodic(self.persistrate, savequeue, self)
	end
	
	-- Init timestamp check
	self:send(nil)

	self.connect_task = sched.run (
        function()
            repeat
                local signal = sched.wait(self.app_name, "*")
				log(self.app_name, "DEBUG", "transport executing on signal " .. signal)
				if signal == "SEND" then
					sched.wait(self.latency)
					repeat
						local data = self.pushqueue:get(self.maxsend)
						log(self.app_name, "DETAIL", "Pushing %d / %d Records ...", #data, self.pushqueue:len())
						local success, err = self.protocol:send(data)
						if success == "ok" then
							log(self.app_name, "DEBUG", "Send success - drop " .. #data .. " records")
							for i = 1, #data do self.pushqueue:pop() end
						else
							log(self.app_name, "DETAIL", "Send failure (" .. #data .. " events) - " .. err)
							break
						end
					until self.pushqueue:len() == 0 or not self.running
				elseif signal == "SAVE" then
					-- persist queue
				elseif signal == "STOP" then
					self.running = false
				else
					log(self.app_name, "DEBUG", "transport no action on signal " .. signal)
				end
				log(self.app_name, "DEBUG", "transport action complete")
            until not self.running
			self:send("Stopped")
			log(self.app_name, "INFO", "transport task exiting ...")
        end
    )
	sched.wait()	-- yield to start thread
end

------------------------------------------------------------------
-- stop - Stop transport send operations
-- 
-- Inputs:
--
-- Outputs:
--
-- Returns:
--
-- Notes
--	Stops all transport send processes ands immediately persists the
--	Tx queue to flash.
------------------------------------------------------------------
local function m_stop (self)
	if self.running then
		self.running = false
		sched.signal(self.app_name, "STOP")
		if self.persisttimer then
			timer.cancel(self.persisttimer)
			self.persisttimer = nil
		end
		-- Force queue persist on stop. Do here in case triggerPolicy is
		-- blocked trying to send.
		log(self.app_name, "INFO", "Persisting %d records in Tx queue", self.pushqueue:len())
		self.pushqueue:save(self.persistname, true)
	end
end

------------------------------------------------------------------
-- configure - Configure transport parameters
-- 
-- Inputs:
--	config			- table of parameter / value pairs
--
-- Outputs:
--
-- Returns:
--
-- Notes
------------------------------------------------------------------
local function m_configure(self, config)
	-- Object to send data to target endpoint
	if type(config.protocol) == "table" and type(config.protocol.send) == "function" then
		self.protocol = config.protocol
	end
	
	-- Maximum queue size
	if tonumber(config.maxqueue) then
		self.maxqueue = tonumber(config.maxqueue)
	end
	
	-- Maximum events in a message
	if tonumber(config.maxsend) then
		self.maxsend = tonumber(config.maxsend)
	end
	
	-- latency is the send latency in seconds. Items sent within this period will
	-- be sent with the one initiating the send.
	if tonumber(config.latency) then
		self.latency = tonumber(config.latency)
	end
	
	-- set persist rate. 0 cancels automatic persist and triggers persist now
	if tonumber(config.persistrate) then
		self.persistrate = tonumber(config.persistrate)
		
		if self.persisttimer then
			timer.cancel(self.persisttimer)
			self.persisttimer = nil
		end
		if self.running then
			if tonumber(config.persistrate) > 0 then
				self.persisttimer = timer.periodic(tonumber(config.persistrate), savequeue, self)
			else
				savequeue(self)
			end
		end
	end
	
	-- persistthresh sets number of queued items that need to be present to save
	if tonumber(config.persistthresh) then
		self.persistthresh = tonumber(config.persistthresh)
	end
	
	-- persistname is name of persist to save queue
	if config.persistname then
		self.persistname = self.app_name .. tostring(config.persistname)
	end

	-- queuetail is size of queue debug tail (retained stale data)
	if config.queuetrail then
		self.pushqueue:settrail(tonumber(config.queuetrail))
	end
end

------------------------------------------------------------------
-- create - Create an instance of transport
-- 
-- Inputs:
--	app_name		- Name of the app (for logging etc)
--
-- Outputs:
--
-- Returns:
--	transport instance	- sucess
--	nil, error		- failure
--
-- Notes
------------------------------------------------------------------
local function m_create(app_name)
    local m = {}
	local err

    m.app_name = app_name
    m.configure = m_configure
    m.run = m_run
    m.stop = m_stop
    m.send = m_send
	m.inform = m_inform
	m.pushRow = m_pushRow
	
	-- debugging
	m.getSystimeMs = getSystimeMs
	
	m.last_now = 0
	m.tsqueue = queue_new()
	
	-- A default protocol object that just displays data to send
	-- (expected to be replaced via configure).
	-- The protocol object is expected to accept list of messages.
	-- This is expected to have at least the following methods. Each message
	-- is a table of one or more key=value pairs, which will be formatted
	-- in a way sutable for that protocol. There is no restriction on the
	-- format of this data item with one exception: it will contain at least
	-- one parameter, timestamp, which is the time the message was created
	-- (system time, in mS) as a number. All pending messages will be passed
	-- in this list, and the protocol object is expected to return "ok" on
	-- successful send of the entire list, or nil, reason of failure.
	m.protocol = {
		send = function(self, data)
			p(self.app_name .. " stub protocol send:", data)
			return "ok"
		end,
		stop = function(self) end,
		run = function(self) end,
		app_name = app_name
	}

	-- persisted table to hold list of stats describing occurrance of
	-- lost events due to lack of known time prior to reboot.
	-- Upon startup, last stat in the list is initially empty (either
	-- newly created if existing stats or re-used if last is empty).
	-- This last stat will be discarded when the respective queued events
	-- are time-adjusted and sent, or sent after next boot as warning message.
	m.eqstat = persist.load(EVTQ_PERSIST)
	if m.eqstat == nil or #m.eqstat < 1  then
		m.eqstat = { {} }
		persist.save(EVTQ_PERSIST, m.eqstat)
		log(app_name, "INFO", "EVTQ_PERSIST initialized")
	end
	-- Ensure there's a fresh instance slot on top at boot.
	if m.eqstat[#m.eqstat].count then
		table.insert(m.eqstat, {})
		persist.save(EVTQ_PERSIST, m.eqstat)
		log(app_name, "INFO", "EVTQ_PERSIST new tip added")
	end

    m.running = false
	
	m.lock = lock:new()
	
	-- Upper part of event id is persisted and incremented on boot
	-- or when the lower part exceeds the base. Guarantees unique
	-- event id for every message without too much flash wear.
	m.eventid_base = persist.load(EVTID_PERSIST)
	if type(m.eventid_base) ~= 'number' then
		m.eventid_base = 0	-- ToDo: make event id persistent
	else
		m.eventid_base = m.eventid_base + EVTID_BASE
	end
	persist.save(EVTID_PERSIST, m.eventid_base)
	m.eventid = 0

	-- table for transfer (Tx queue)
	m.pushqueue = queue_new()

	-- Default transport send parameters
	m.maxqueue = 10000
	m.persistthresh = 1
	m.persistrate = 0
	m.latency = 5
	m.maxsend = 0
	m.persistname = app_name .. "_send_queue"
	m.reftime = REFTIME
    return m
end

return m_create

