local logger = require "prosody.util.logger";

local log = logger.init("util.async");

local new_id = require "prosody.util.id".short;

local xpcall = require "prosody.util.xpcall".xpcall;

local time_now = require "prosody.util.time".now;

local function checkthread()

	local thread, main = coroutine.running();

	if not thread or main then

		error("Not running in an async context, see https://prosody.im/doc/developers/util/async");

	end

	return thread;

end

-- Configurable functions

local schedule_task = nil; -- schedule_task(seconds, callback)

local next_tick = function (f)

	f();

end

local function runner_from_thread(thread)

	local level = 0;

	-- Find the 'level' of the top-most function (0 == current level, 1 == caller, ...)

	while debug.getinfo(thread, level, "") do level = level + 1; end

	local name, runner = debug.getlocal(thread, level-1, 1);

	if name ~= "self" or type(runner) ~= "table" or runner.thread ~= thread then

		return nil;

	end

	return runner;

end

local function call_watcher(runner, watcher_name, ...)

	local watcher = runner.watchers[watcher_name];

	if not watcher then

		return false;

	end

	runner:log("debug", "Calling '%s' watcher", watcher_name);

	local ok, err = xpcall(watcher, debug.traceback, runner, ...);

	if not ok then

		runner:log("error", "Error in '%s' watcher: %s", watcher_name, err);

		return nil, err;

	end

	return true;

end

local function runner_continue(thread)

	-- ASSUMPTION: runner is in 'waiting' state (but we don't have the runner to know for sure)

	if coroutine.status(thread) ~= "suspended" then -- This should suffice

		log("error", "unexpected async state: thread not suspended (%s, %s)", thread, coroutine.status(thread));

		-- Fetching the traceback is likely to *crash* if a C library is calling us while suspended

		--log("error", "coroutine stack: %s", debug.traceback());

		return false;

	end

	local ok, state, runner = coroutine.resume(thread);

	if not ok then

		local err = state;

		-- Running the coroutine failed, which means we have to find the runner manually,

		-- in order to inform the error handler

		runner = runner_from_thread(thread);

		if not runner then

			log("error", "unexpected async state: unable to locate runner during error handling");

			return false;

		end

		call_watcher(runner, "error", debug.traceback(thread, err));

		runner.state = "ready";

		return runner:run();

	elseif state == "ready" then

		-- If state is 'ready', it is our responsibility to update runner.state from 'waiting'.

		-- We also have to :run(), because the queue might have further items that will not be

		-- processed otherwise. FIXME: It's probably best to do this in a nexttick (0 timer).

		next_tick(function ()

			runner.state = "ready";

			runner:run();

		end);

	end

	return true;

end

local function waiter(num, allow_many)

	local thread = checkthread();

	num = num or 1;

	local waiting;

	return function ()

		if num == 0 then return; end -- already done

		waiting = true;

		coroutine.yield("wait");

	end, function ()

		num = num - 1;

		if num == 0 and waiting then

			runner_continue(thread);

		elseif not allow_many and num < 0 then

			error("done() called too many times");

		end

	end;

end

local function guarder()

	local guards = {};

	local default_id = {};

	return function (id, func)

		id = id or default_id;

		local thread = checkthread();

		local guard = guards[id];

		if not guard then

			guard = {};

			guards[id] = guard;

			log("debug", "New guard!");

		else

			table.insert(guard, thread);

			log("debug", "Guarded. %d threads waiting.", #guard)

			coroutine.yield("wait");

		end

		local function exit()

			local next_waiting = table.remove(guard, 1);

			if next_waiting then

				log("debug", "guard: Executing next waiting thread (%d left)", #guard)

				runner_continue(next_waiting);

			else

				log("debug", "Guard off duty.")

				guards[id] = nil;

			end

		end

		if func then

			func();

			exit();

			return;

		end

		return exit;

	end;

end

local function sleep(seconds)

	if not schedule_task then

		error("async.sleep() is not available - configure schedule function");

	end

	local wait, done = waiter();

	schedule_task(seconds, done);

	wait();

end

local runner_mt = {};

runner_mt.__index = runner_mt;

local waiting_runners = {};

local function runner_create_thread(func, self)

	local thread = coroutine.create(function (self) -- luacheck: ignore 432/self

		while true do

			func(coroutine.yield("ready", self));

		end

	end);

	debug.sethook(thread, debug.gethook());

	assert(coroutine.resume(thread, self)); -- Start it up, it will return instantly to wait for the first input

	return thread;

end

local function default_error_watcher(runner, err)

	runner:log("error", "Encountered error: %s", err);

	error(err);

end

local function default_func(f) f(); end

local function runner(func, watchers, data)

	local id = new_id();

	local _log = logger.init("runner" .. id);

	return setmetatable({ func = func or default_func, thread = false, state = "ready", notified_state = "ready",

		queue = {}, watchers = watchers or { error = default_error_watcher }, data = data, id = id, _log = _log; }

	, runner_mt);

end

-- Add a task item for the runner to process

function runner_mt:run(input)

	if input ~= nil then

		table.insert(self.queue, input);

		--self:log("debug", "queued new work item, %d items queued", #self.queue);

	end

	if self.state ~= "ready" then

		-- The runner is busy. Indicate that the task item has been

		-- queued, and return information about the current runner state

		return true, self.state, #self.queue;

	end

	local q, thread = self.queue, self.thread;

	if not thread or coroutine.status(thread) == "dead" then

		--luacheck: ignore 143/coroutine

		if thread and coroutine.close then

			coroutine.close(thread);

		end

		self:log("debug", "creating new coroutine");

		-- Create a new coroutine for this runner

		thread = runner_create_thread(self.func, self);

		self.thread = thread;

	end

	-- Process task item(s) while the queue is not empty, and we're not blocked

	local n, state, err = #q, self.state, nil;

	self.state = "running";

	--self:log("debug", "running main loop");

	while n > 0 and state == "ready" and not err do

		local consumed;

		-- Loop through queue items, and attempt to run them

		for i = 1,n do

			local queued_input = q[i];

			self:log("Resuming thread with new item [%s]", thread);

			self.current_item = queued_input;

			local ok, new_state = coroutine.resume(thread, queued_input);

			if not ok then

				-- There was an error running the coroutine, save the error, mark runner as ready to begin again

				consumed, state, err = i, "ready", debug.traceback(thread, new_state);

				self.thread = nil;

				break;

			elseif new_state == "wait" then

				 -- Runner is blocked on waiting for a task item to complete

				consumed, state = i, "waiting";

				break;

			end

		end

		-- Loop ended - either queue empty because all tasks passed without blocking (consumed == nil)

		-- or runner is blocked/errored, and consumed will contain the number of tasks processed so far

		if not consumed then consumed = n; end

		-- Remove consumed items from the queue array

		if q[n+1] ~= nil then

			n = #q;

		end

		for i = 1, n do

			q[i] = q[consumed+i];

		end

		n = #q;

	end

	-- Runner processed all items it can, so save current runner state

	self.state = state;

	if state == "ready" and self.current_item then

		self.current_item = nil;

	end

	if err or state ~= self.notified_state then

		self:log("debug", "changed state from %s to %s [%s %s]", self.notified_state, err and ("error (" .. state .. ")") or state, self.thread,

			self.thread and coroutine.status(self.thread));

		if err then

			state = "error"

		else

			self.notified_state = state;

		end

		local handler = self.watchers[state];

		if handler then handler(self, err); end

	end

	if n > 0 then

		return self:run();

	end

	waiting_runners[self] = state == "waiting" and time_now() or nil;

	return true, state, n;

end

-- Add a task item to the queue without invoking the runner, even if it is idle

function runner_mt:enqueue(input)

	table.insert(self.queue, input);

	self:log("debug", "queued new work item, %d items queued", #self.queue);

	return self;

end

function runner_mt:log(level, fmt, ...)

	return self._log(level, fmt, ...);

end

function runner_mt:onready(f)

	self.watchers.ready = f;

	return self;

end

function runner_mt:onwaiting(f)

	self.watchers.waiting = f;

	return self;

end

function runner_mt:onerror(f)

	self.watchers.error = f;

	return self;

end

local function ready()

	return pcall(checkthread);

end

local function wait_for(promise)

	local async_wait, async_done = waiter();

	local ret, err = nil, nil;

	promise:next(

		function (r) ret = r; end,

		function (e) err = e; end)

		:finally(async_done);

	async_wait();

	if ret then

		return ret;

	else

		return nil, err;

	end

end

return {

	ready = ready;

	waiter = waiter;

	guarder = guarder;

	runner = runner;

	wait = wait_for; -- COMPAT w/trunk pre-0.12

	wait_for = wait_for;

	sleep = sleep;

	set_nexttick = function(new_next_tick) next_tick = new_next_tick; end;

	set_schedule_function = function (new_schedule_function) schedule_task = new_schedule_function; end;

	waiting_runners = waiting_runners;

	default_runner_func = default_func;

};