#!/opt/lnxall_app/bin/lua

-- Created by jiaqiang.ye@lnxall.com
-- Simple Command Server for Non-openwrt Systems
-- 2024/02/04

local posix = require "posix"
local cjson = require "cjson"
local invoker = require "invoker"
local ubuslog = require "ubuslog"
local mosquitto = require "mosquitto"

local gfmt = string.format
-- global variables
local g_mosq = nil        -- mosquitto connection handle
local g_mosq_okay = false -- whether MQTT connection established
local g_cmdlist = {}      -- command execute list

local function mosq_on_connect(okay)
	if not okay then
		g_mosq_okay = false
		io.stderr:write("Error, failed to connect to MQTT broker.\n")
		io.stderr:flush()
		return false
	end

	g_mosq_okay = true
	io.stderr:write("MQTT Broker connected successfully.\n")
	io.stderr:flush()
	return true
end

local function mosq_on_disconnect()
	g_mosq_okay = false
	io.stderr:write("Error, MQTT connection has terminated.\n")
	io.stderr:flush()
	return true
end

local function check_app(info)
	local exited, eval = invoker.waitpid(info.pid, true)
	if exited then
		local resp = nil
		if info.outfd > 0 then
			resp = invoker.readfd(info.outfd, 65535)
			posix.close(info.outfd); info.outfd = -1
		end
		local topic = gfmt("ipc/cmdsvr/resp/%s", info.cmd)
		local payload = { name = info.cmd, seq = info.seq,
			timedout = false, status = eval, resp = resp }
		if g_mosq then g_mosq:publish(topic, cjson.encode(payload)) end
		return true
	end

	if info.timeout > 0 then
		local uptim = invoker.uptimsec()
		local delta = uptim - info.begin
		if delta >= (info.timeout * 1000) then
			if info.outfd > 0 then
				posix.close(info.outfd)
				info.outfd = -1;
			end
			local topic = gfmt("ipc/cmdsvr/resp/%s", info.cmd)
			local payload = { name = info.cmd, seq = info.seq, timedout = true }
			if g_mosq then g_mosq:publish(topic, cjson.encode(payload)) end

			invoker.kill(info.pid, posix.SIGKILL)
			invoker.waitpid(info.pid, false)
			return true
		end
	end
	return false
end

local function mosq_new_msg(mid, topic, payload)
	-- io.stdout:write(gfmt("Received message: %s\n%s\n", topic, payload))
	-- io.stdout:flush()

	local cmd = string.match(topic, "^ipc/cmdsvr/call/([^/\r\n]+)$")
	if not cmd then
		io.stderr:write("Error, command-name not found.\n")
		io.stderr:flush()
		return false
	end

	local okay, pload = pcall(cjson.decode, payload)
	if not okay or type(pload) ~= "table" then
		io.stderr:write("Error, invalid payload for CMDSVR.\n")
		io.stderr:flush()
		return false
	end

	if pload.name ~= cmd then
		io.stderr:write(gfmt("Error, invalid name found in payload, expected: %s\n", cmd))
		io.stderr:flush()
		return false
	end

	-- check the argument array
	if type(pload.argv) ~= "table" then
		io.stderr:write(gfmt("Error, invalid argument array found for '%s'\n", cmd))
		io.stderr:flush()
		return false
	end

	if type(pload.argv[1]) ~= "string" or string.len(pload.argv[1]) == 0 then
		io.stderr:write(gfmt("Error, no argument found for '%s'\n", cmd))
		io.stderr:flush()
		return false
	end

	if type(pload.timeout) ~= "number" then pload.timeout = -1 end -- in seconds
	local pid, outfd = invoker.invoke(invoker.NOWAIT + invoker.OUTPUT, pload.argv)
	if not pid or type(outfd) ~= "number" then
		io.stderr:write(gfmt("Error, failed to execute '%s' for '%s'\n", pload.argv[1], cmd))
		io.stderr:flush()
		return false
	end

	local begin = invoker.uptimsec()
	local cinfo = { ["timeout"] = pload.timeout, begin = begin,
		["cmd"] = cmd, pid = pid, outfd = outfd, ["seq"] = pload.seq }
	if not check_app(cinfo) then
		local randval = gfmt("%s_%s", cmd, invoker.randstr(0x8, true))
		g_cmdlist[randval] = cinfo
	end
	return true
end

local function check_cmdlist()
	local dlist = {}
	for randkey, cmdinfo in pairs(g_cmdlist) do
		if check_app(cmdinfo) then
			dlist[#dlist + 1] = randkey
		end
	end

	for _, rkey in ipairs(dlist) do
		g_cmdlist[rkey] = nil
	end
	return true
end

local function mosq_connect(host, port)
	local mosq = mosquitto.new()
	if not mosq then
		io.stderr:write("Error, failed to create new MQTT instance.\n")
		io.stderr:flush()
		return false
	end

	mosq.ON_CONNECT = mosq_on_connect
	mosq.ON_DISCONNECT = mosq_on_disconnect
	mosq.ON_MESSAGE = mosq_new_msg
	g_mosq = mosq
	local okay = mosq:connect_async(host, port, 20)
	if not okay then
		g_mosq = nil
		io.stderr:write(gfmt("Error, failed to connect to MQTT broker: %s:%d\n", host, port))
		io.stderr:flush()
		mosq:destroy(); msoq = nil
		return false
	end

	local now = invoker.uptime() + 0x4
	while true do
		mosq:loop(500)
		if g_mosq_okay then break end
		local uptim = invoker.uptime()
		if uptim >= now then break end
	end

	if not g_mosq_okay then
		g_mosq = nil
		io.stderr:write(gfmt("Error, connect to MQTT broker timedout: %s:%d\n", host, port))
		io.stderr:flush()
		mosq:destroy(); msoq = nil
		return false
	end

	if not mosq:subscribe("ipc/cmdsvr/call/#") then
		g_mosq = nil
		io.stderr:write("Error, failed to subscribe CMDSVR topic.\n")
		io.stderr:flush()
		mosq:destroy(); msoq = nil
		return false
	end

	return true
end

local function cmdsvr()
	mosquitto.init()
	invoker.setname("CMDSVR")
	posix.signal(posix.SIGCHLD, posix.SIG_DFL)

	while true do
		if mosq_connect('127.0.0.1', 1883) then
			break
		end
		invoker.waitsec(5)
	end
	ubuslog.reinit('CMDSVR')

	local lastchk = invoker.uptimsec()
	while g_mosq_okay do
		g_mosq:loop(165)
		local nowchk = invoker.uptimsec()
		if (nowchk - lastchk) >= 330 then
			check_cmdlist()
			lastchk = nowchk
		end
	end
	return true
end

invoker.waitsec(3)
cmdsvr()
invoker.waitsec(5)
os.exit(1)