#!/usr/bin/lua

-- Created by jiaqiang.ye@lnxall.com
-- 2022/08/09

local cjson = require "cjson"
local posix = require "posix"
local invoker = require "invoker"
local mosq = require "mosquitto"
local base64 = require "base64"

local g_int_hdl = nil           -- MQTT connection to internal broker
local g_int_con = false         -- true if successfully connected to internal broker
local g_dtab = {}               -- global docker table
local g_dfuncs = {}             -- docker operation functions
local g_dockercfg = "docker-compose.yaml"

local function load_json(fpath)
    local filp = io.open(fpath, "rb")
    if not filp then return nil end
    local fild = filp:read("*a")
    filp:close(); filp = nil
    if not fild or string.len(fild) == 0 then return nil end
    local okay, jdat = pcall(cjson.decode, fild)
    if not okay then return nil end
    return jdat
end

local function goto_dir(dirp)
	if not posix.chdir(dirp) then
		io.stderr:write(string.format("Error, chdir(%s) has failed.\n", dirp))
		io.stderr:flush()
		return false
	end
	return true
end

local function service_reload(command)
	if not command then command = "restart" end
	local rval = invoker.invoke(invoker.CLOSEFD, "/etc/init.d/dockerd", command)
	if rval == 0 then
		posix.nanosleep{tv_sec = 1, tv_nsec = 0}
		return true
	end
	return false
end

local function load_node_cfgs()
	-- local objs = load_json("/app/node/object_model_cfg.json")
	-- if not objs then return false end
	local ncfgs = load_json("/app/node/nodes_cfg.json")
	if not ncfgs then return false end

    local emptab, idx, jdx = {}, 0, 0
    ncfgs = ncfgs["nodes_cfg"] or emptab
    if not ncfgs then return false end

    while true do
        idx = idx + 1
        local ncfg = ncfgs[idx]
        if type(ncfg) ~= "table" then break end
        local cport = ncfg["connect_port"]
		if cport == "VIRTUAL_1" then
			local docker_sn = ncfg["sn"]
			if type(docker_sn) ~= "string" or g_dtab[docker_sn] then
				io.stderr:write(string.format("Error, sn already exits: %s\n", docker_sn))
				io.stderr:flush()
				break
			end
			local durl = ncfg["ext_data"]
			if type(durl) == "string" and string.len(durl) > 0 then
				jdx = jdx + 1
				local dirp = string.format("/opt/docker_%s", docker_sn)
				if not goto_dir(dirp) then posix.mkdir(dirp) end
				g_dtab[docker_sn] = { ["sn"] = docker_sn,
					["durl"] = durl, ["dirp"] = dirp }
			end
		end
	end

	goto_dir("/")
	if jdx == 0 then
		io.stderr:write("Error, no virtual device with ext_data found.\n")
		io.stderr:flush()
		return false
	end
	return true
end

local function curldl(dlurl, outfile, curltime, totaltime, maxsize, md5sum)
	local args, argc = {}, 0
	-- setup arguments
	argc = argc + 1; args[argc] = "curl"
	argc = argc + 1; args[argc] = "--insecure"
	argc = argc + 1; args[argc] = "--silent"
	argc = argc + 1; args[argc] = '--output'
	argc = argc + 1; args[argc] = outfile
	if curltime then
		argc = argc + 1; args[argc] = '--connect-timeout'
		argc = argc + 1; args[argc] = type(curltime) == "number" and tostring(curltime) or "5"
	end
	argc = argc + 1; args[argc] = dlurl

	-- remove existing output file if already exist
	if posix.access(outfile) then posix.unlink(outfile) end

	-- invoke `curl application
	local curlpid = invoker.invoke(invoker.NOWAIT + invoker.NOSTDIO + invoker.CLOSEFD, args)
	if not curlpid then return nil, "failed to invoke curl" end
	local nowt = invoker.uptime()
	while true do
		local exited, exst = invoker.waitpid(curlpid, true)
		if exited then
			if exst ~= 0 then
				posix.unlink(outfile); posix.sync()
				return nil, "failed to download via curl: " .. dlurl
			end
			break
		end
		if totaltime and totaltime > 0 then
			local thetime = invoker.uptime()
			if (thetime - nowt) > totaltime then
				-- download time has expired
				if invoker.kill(curlpid, posix.SIGTERM) then
					invoker.waitpid(curlpid, false)
				end
				posix.unlink(outfile); posix.sync()
				return nil, "download has expired: " .. dlurl
			end
		end
		if maxsize and maxsize > 0 then
			local filest = posix.stat(outfile)
			if type(filest) == "table" then
				local cursize = filest.st_size
				if type(cursize) == "number" then
					io.stdout:write(string.format("\rDownloading to %s: %.02f%%",
						outfile, (cursize / maxsize) * 100.0))
					io.stdout:flush()
				end
			end
		end
		posix.sync(); posix.nanosleep{tv_sec = 0, tv_nsec = 500000000}
	end

	local outst = posix.stat(outfile)
	if not outst then return nil, "curl has not written data to " .. outfile end
	if md5sum then
		local _, md5buf = invoker.invoke(invoker.OUTPUT + invoker.CLOSEFD, "md5sum", outfile)
		if type(md5buf) ~= "string" or #md5buf == 0 then
			-- fatal error, cannot calculate the md5sum for `outfile
			posix.unlink(outfile)
			return nil, "cannot get md5sum for " .. outfile
		end
		md5buf = string.match(md5buf, "^(%x+)")
		if md5buf ~= md5sum then
			posix.unlink(outfile)
			return nil, "md5sum values do not match for " .. outfile
		end
	end
	return outst.st_size
end

local function exec_containers(clist, cmdlist)
	local idx, argc = 0, 1

	local cmds = {}
	cmds[argc] = "docker-compose"
	while true do
		idx = idx + 1
		local cmd = cmdlist[idx]
		if not cmd then break end
		argc = argc + 1
		cmds[argc] = cmd
	end

	idx = 0
	argc = argc + 1
	while true do
		idx = idx + 1
		local cid = clist[idx]
		if not cid then break end
		cmds[argc] = cid["name"]
		local rval = invoker.invoke(invoker.CLOSEFD, cmds)
		if type(rval) == "number" and rval ~= 0 then
			io.stderr:write(string.format("command has failed: '%s' with %d\n",
				table.concat(cmds, " "), rval))
			io.stderr:flush()
		end
	end
	return true
end

local function list_containers()
	local ccnt, containers = 0, {}
	local flags = invoker.CLOSEFD + invoker.OUTPUT + invoker.NOSTDIO
	local okay, cbuf = invoker.invoke(flags, "docker-compose", "ps", "-a")
	if not okay or okay ~= 0 then
		return ccnt, containers
	end

	local sepflag, state_pos = false, 0
	for line in string.gmatch(cbuf, "([^\r\n]+)") do
		if not sepflag then
			if string.find(line, "^%-%-") then
				sepflag = true
			else
				local spos = string.find(line, "State", 1, true)
				if spos then state_pos = spos end
			end
		else
			local name = string.match(line, "^([^%s]+)%s+")
			local state = string.match(line:sub(state_pos), "^([^%s]+)%s+")
			if not name or not state then
				io.stderr:write(string.format("Error, invalid container: %s\n", line))
				io.stderr:flush()
			else
				ccnt = ccnt + 1
				containers[ccnt] = { ["name"] = name, ["state"] = state }
			end
		end
	end
	return ccnt, containers
end

g_dfuncs["create"] = function (dsn)
	local dinfo = g_dtab[dsn]
	if not dinfo then
		return nil, string.format("No docker information found for %s", dsn)
	end

	-- stop running containers
	local count, contains = list_containers()
	if count > 0 then
		local commands = { [1] = "stop", }
		exec_containers(contains, commands)
		commands[1] = "rm"; commands[2] = "--force"
		exec_containers(contains, commands)
		invoker.invoke(invoker.CLOSEFD,
			"docker-compose", "down", "--rmi", "all")
		posix.sync()
	end

	local okay, errmsg = curldl(dinfo["durl"], g_dockercfg, 2)
	if not okay then
		io.stderr:write(string.format("Error, failed to fetch '%s'\n", dinfo["durl"]))
		if type(errmsg) == "string" then
			io.stderr:write(errmsg)
			io.stderr:write("\n")
		end
		io.stderr:flush()
		return nil, string.format("failed to download from '%s'", dinfo["durl"])
	end

	okay, errmsg = invoker.invoke(invoker.CLOSEFD, "docker-compose", "pull")
	if okay and okay == 0 then
		okay, errmsg = invoker.invoke(invoker.CLOSEFD, "docker-compose", "up", "-d")
		if okay ~= 0 then
			service_reload()
			okay, errmsg = invoker.invoke(invoker.CLOSEFD, "docker-compose", "up", "-d")
		end
	end
	if okay and okay == 0 then return true end
	if type(errmsg) ~= "string" then
		errmsg = "Error, failed to add docker image"
	end
	return nil, errmsg
end

g_dfuncs["remove"] = function (dsn)
	-- stop running containers
	local count, contains = list_containers()
	if count > 0 then
		local commands = { [1] = "stop", }
		exec_containers(contains, commands)
		commands[1] = "rm"; commands[2] = "--force"
		exec_containers(contains, commands)
		invoker.invoke(invoker.CLOSEFD, "docker-compose",
			"down", "--rmi", "all")
		posix.sync()
	end
	posix.unlink(g_dockercfg)
	local ddir = g_dtab[dsn]
	if ddir then ddir = ddir["dirp"] end
	if ddir then
		posix.chdir("/")
		posix.rmdir(ddir)
	end
	return true, { ["sn"] = dsn, ["message"] = "remove ok" }
end

g_dfuncs["ps"] = function (dsn)
	local okay, outbuf = invoker.invoke(invoker.CLOSEFD + invoker.OUTPUT,
		"docker-compose", "ps", "-a")
	if okay and okay == 0 then
		local stats, found = {}, false
		for line in string.gmatch(outbuf, "([^\r\n]+)") do
			if found then
				stats[#stats + 1] = line
			elseif string.find(line, "^%-%-") then
				found = true
			end
		end
		return true, { ["sn"] = dsn, ["message"] = stats }
	end
	if type(outbuf) ~= "string" then
		outbuf = "Error, failed to get docker status"
	end
	return nil, outbuf
end

g_dfuncs["pull"] = function (dsn)
	-- stop running containers
	local count, contains = list_containers()
	if count > 0 then
		local commands = { [1] = "stop", }
		exec_containers(contains, commands)
		posix.sync()
	end

	local okay, errmsg = invoker.invoke(invoker.CLOSEFD, "docker-compose", "pull")
	if okay == 0 then
		okay, errmsg = invoker.invoke(invoker.CLOSEFD, "docker-compose", "up", "-d")
		if okay ~= 0 then
			service_reload()
			okay, errmsg = invoker.invoke(invoker.CLOSEFD, "docker-compose", "up", "-d")
		end
	end
	if okay == 0 then
		return true, { ["sn"] = dsn, ["message"] = "pull ok" }
	end
	if type(errmsg) ~= "string" then
		errmsg = "Error, failed to add docker image"
	end
	return nil, errmsg
end

g_dfuncs["stop"] = function (dsn)
	local okay, errmsg = invoker.invoke(invoker.CLOSEFD,
		"docker-compose", "stop")
	if okay and okay == 0 then
		return true, { ["sn"] = dsn, ["message"] = "stop ok" }
	end
	if type(outbuf) ~= "string" then
		outbuf = "Error, failed to stop docker"
	end
	return nil, outbuf
end

g_dfuncs["start"] = function (dsn)
	local okay, errmsg = invoker.invoke(invoker.CLOSEFD,
		"docker-compose", "start")
	if okay ~= 0 then
		service_reload()
		okay, errmsg = invoker.invoke(invoker.CLOSEFD,
			"docker-compose", "start")
	end
	if okay == 0 then
		return true, { ["sn"] = dsn, ["message"] = "start ok" }
	end
	if type(outbuf) ~= "string" then
		outbuf = string.format("Error, failed to start docker: %d", okay or -1)
	end
	return nil, outbuf
end

g_dfuncs["logs"] = function (dsn, data)
	local typn = type(data)
	if typn ~= "string" then data = "" end
	if string.len(data) == 0 then
		return nil, "Error, invalid command to run"
	end
	-- io.stdout:write("data: ["); io.stdout:write(data)
	-- io.stdout:write("]\n======================================\n")
	local lines, name = string.match(data, "^([^\n]*)\n([^\n]*)$")
	if lines then lines = tonumber(lines) end
	if not lines then lines = 0 end
	if not name or #name == 0 then
		name = nil
		local cnt, lst = list_containers()
		if cnt > 0 then name = lst[1]["name"] end
	end
	if not name then
		return false, { ["sn"] = dsn, ["return"] = 1,
			["message"] = string.format("container not found"), }
	end

	local cmds = nil
	if lines > 0 then
		cmds = string.format("exec docker logs -n %d '%s' 2>&1",
			lines, name)
	else
		cmds = string.format("exec docker logs '%s' 2>&1", name)
	end
	-- io.stdout:write(cmds)
	-- io.stdout:write("\n--------------------------------------\n")
	local okay, outbuf = invoker.invoke(invoker.OUTPUT + invoker.CLOSEFD + 0xFF0000,
		"/bin/sh", "-c", cmds)
	if not okay then
		return nil, string.format("Error, failed to run command: '%s'", data)
	end
	if type(outbuf) == "string" and #outbuf > 0 then
		local cnt, outb = 0, {}
		for line in string.gmatch(outbuf, "([^\r\n]+)") do
			cnt = cnt + 1
			outb[cnt] = line
		end
		if cnt > 0 then outbuf = outb end
	end
	return true, { ["sn"] = dsn, ["message"] = outbuf,
		["return"] = okay, ["container"] = name }
end

g_dfuncs["login"] = function (dsn, data)
	if type(data) ~= "string" then
		return nil, "Error, no login parameters given"
	end
	local params, cntp = {}, 0
	for param in string.gmatch(data, "([^\r\n]+)") do
		cntp = cntp + 1
		params[cntp] = param
		-- io.stdout:write(string.format("login param[%d]: %s\n", cntp, param))
		-- io.stdout:flush()
	end
	if cntp ~= 0x3 then
		return nil, string.format("Error, invalid number of login parameters: %d", cntp)
	end
	posix.unlink("/root/.docker/config.json")
	local okay, outbuf = invoker.invoke(invoker.CLOSEFD + invoker.OUTPUT,
		"docker", "login", params[1], "-u", params[2], "-p", params[3])
	if not okay then
		return nil, "Error, failed to run docker login"
	end
	if okay ~= 0 then
		return nil, { ["sn"] = dsn,
			["message"] = string.format("docker login has failed with %d", okay), }
	end
	return true
end

local function mqtt_cb_conn(okay, errmsg)
	g_int_con = okay and true or false
    if not okay then
        io.stderr:write("Error, failed to connect to internal broker!\n")
        io.stderr:flush()
        return false
    end
    io.stdout:write("Internal broker connected!\n")
	io.stdout:flush()
	local mqtt = g_int_hdl
	if not mqtt then
		io.stderr:write("Error, invalid mqtt handle!\n")
		io.stderr:flush()
		return false
	end
	okay = mqtt:subscribe("ipc/+/VIRTUAL_1/device/+/data/Set_Rglt_Raw")
	if not okay then
		io.stderr:write("Error, failed to subscribe topic!\n")
		io.stderr:flush()
		return false
	end
	return true
end

local function mqtt_cb_discon()
	g_int_con = false
	io.stdout:write("Internal broker disconnected!\n")
	io.stdout:flush()
	return true
end

local function mqtt_cb_msg(msgid, mtopic, mpayload)
	local okay, pload = pcall(cjson.decode, mpayload)
	if not okay then
		io.stderr:write(string.format("Error, invalid payload:\n%s\n", mpayload))
		io.stderr:flush()
		return nil
	end
	local dsn, tagnode = pload["sn"], nil
	if type(dsn) ~= "string" or #dsn == 0 then
		io.stderr:write(string.format("Error, sn not found in topic: [%s]\n", mtopic))
		io.stderr:flush()
		return nil
	end
	local dinfo = g_dtab[dsn]
	if not dinfo then
		io.stderr:write(string.format("Docker with SN not found: %s\n", dsn))
		io.stderr:flush()
		return nil
	end
	if not goto_dir(dinfo["dirp"]) then
		posix.mkdir(dinfo["dirp"])
		if not goto_dir(dinfo["dirp"]) then return nil end
	end
	dinfo = nil -- release local reference to docker info

	local srcid = pload["src_identifier"]
	if type(srcid) ~= "string" then srcid = "WHAT_THE_HELL" end
	local dfunc = g_dfuncs[srcid]
	if not dfunc then
		io.stderr:write(string.format("Error, cannot handle src_identifier: %s\n", srcid))
		io.stderr:flush()
		return nil
	end
	local db64 = pload["data_b64"]
	if type(db64) ~= "string" or #db64 == 0 then
		db64 = nil
	else
		okay, db64 = pcall(base64.decode, db64)
		if not okay then db64 = nil end
	end
	okay, tags = dfunc(dsn, db64)
	goto_dir("/opt")
	if okay ~= false then
		local mqtt, typn = g_int_hdl, type(tags)
		if typn == "string" then
			tags = { ["sn"] = dsn, ["message"] = tags }
		elseif typn ~= "table" then
			if tags == nil then
				tags = { ["sn"] = dsn, ["message"] = "success" }
			else
				tags = { ["sn"] = dsn, ["message"] = "failed" }
			end
		end
		local newtop = string.gsub(mtopic,
			"/data/Set_Rglt_Raw$", "/data_filtered/service/") .. srcid
		local reply = { ["mi"] = pload["mi"], ["sn"] = dsn,
			["identifier"] = srcid, ["tags"] = tags, ["time"] = os.time() }
		if mqtt then
			reply = cjson.encode(reply)
			mqtt:publish(newtop, reply)
		end
	end
	return okay and true or false
end

-- try to connect to internal broker
local function mqtt_int_connect(ipaddr, portno)
    local mqtt = mosq.new()
    mqtt.ON_CONNECT = mqtt_cb_conn
    mqtt.ON_DISCONNECT = mqtt_cb_discon
	mqtt.ON_MESSAGE = mqtt_cb_msg
	g_int_hdl = mqtt -- export internal mqtt handle
    if g_int_con then g_int_con = false end
    if not mqtt:connect_async(ipaddr, portno) then
        mqtt:destory()
		mqtt = nil; g_int_hdl = nil
        io.stderr:write("Error, failed to connect to internal broker!\n")
        io.stderr:flush()
        return false
	end

	local nowt = invoker.uptime()
	while not g_int_con do
		local thetime = invoker.uptime()
		if (thetime - nowt) > 10 then
			mqtt:destory()
			mqtt = nil; g_int_hdl = nil
			io.stderr:write("Error, failed to connect to internal broker!\n")
			io.stderr:flush()
			return false
		end
		mqtt:loop(1000, 10)
	end
	return true
end

mosq.init() -- initialize mosquitto library
invoker.setname("DOCKERMAN")
posix.setenv("HOME", "/root", true)
if not load_node_cfgs() then
	-- hange process
	while true do posix.nanosleep{tv_sec = 60, tv_nsec = 0} end
end
if not goto_dir("/") then os.exit(2) end
if not mqtt_int_connect("127.0.0.1", 1883) then os.exit(3) end
while g_int_con do g_int_hdl:loop(5000, 10) end
g_int_hdl:destroy()
g_int_hdl = nil
os.exit(4)