#pragma once

#include <deque>
#include <vector>
#include <mutex>
#include <condition_variable>
#include <optional>
#include <chrono>
#include <thread>

namespace psemek::util
{

	template <typename T>
	struct synchronized_queue
	{
		using clock = std::chrono::high_resolution_clock;

		synchronized_queue(std::size_t max_size = std::numeric_limits<std::size_t>::max()) noexcept
			: max_size_(max_size)
		{}

		std::size_t max_size() const noexcept
		{
			return max_size_;
		}

		void push(T const & x);
		void push(T && x);

		template <typename TimePoint>
		void push_at(TimePoint event_time, T const & x);
		template <typename TimePoint>
		void push_at(TimePoint event_time, T && x);

		bool try_push(T const & x);
		bool try_push(T && x);

		template <typename Duration>
		bool try_push_for(T const & x, Duration wait_period);
		template <typename Duration>
		bool try_push_for(T && x, Duration wait_period);

		template <typename TimePoint>
		bool try_push_until(T const & x, TimePoint wait_time);
		template <typename TimePoint>
		bool try_push_until(T && x, TimePoint wait_time);

		template <typename TimePoint>
		bool try_push_at(TimePoint event_time, T const & x);
		template <typename TimePoint>
		bool try_push_at(TimePoint event_time, T && x);

		template <typename TimePoint, typename Duration>
		bool try_push_at_for(TimePoint event_time, T const & x, Duration wait_period);
		template <typename TimePoint, typename Duration>
		bool try_push_at_for(TimePoint event_time, T && x, Duration wait_period);

		template <typename EventTimePoint, typename WaitTimePoint>
		bool try_push_at_until(EventTimePoint event_time, T const & x, WaitTimePoint wait_time);
		template <typename EventTimePoint, typename WaitTimePoint>
		bool try_push_at_until(EventTimePoint event_time, T && x, WaitTimePoint wait_time);

		T pop();

		std::optional<T> try_pop();
		template <typename Duration>
		std::optional<T> try_pop_for(Duration wait_period);
		template <typename TimePoint>
		std::optional<T> try_pop_until(TimePoint wait_time);

		void clear();

		// Wait for the queue to become empty
		// e.g. when no new items are going to be pushed
		void wait();
		template <typename Duration>
		bool wait_for(Duration wait_period);
		template <typename TimePoint>
		bool wait_until(TimePoint wait_time);

		std::size_t size() const;
		bool empty() const;

	private:
		struct deferred
		{
			clock::time_point time;
			T value;
		};

		mutable std::mutex mutex_;
		std::condition_variable push_cv_, pop_cv_;
		std::deque<T> queue_;
		std::vector<deferred> deferred_heap_;
		std::size_t const max_size_;

		static auto heap_compare();

		void flush_deferred();

		std::size_t size_internal() const;
		bool empty_internal() const;
	};

	template <typename T>
	auto synchronized_queue<T>::heap_compare()
	{
		return [](deferred const & d1, deferred const & d2){ return d1.time > d2.time; };
	}

	template <typename T>
	void synchronized_queue<T>::push(T const & x)
	{
		std::unique_lock lock{mutex_};
		push_cv_.wait(lock, [this]{ return size_internal() < max_size(); });
		queue_.push_back(x);
		lock.unlock();
		pop_cv_.notify_one();
	}

	template <typename T>
	void synchronized_queue<T>::push(T && x)
	{
		std::unique_lock lock{mutex_};
		push_cv_.wait(lock, [this]{ return size_internal() < max_size(); });
		queue_.push_back(std::move(x));
		lock.unlock();
		pop_cv_.notify_one();
	}

	template <typename T>
	template <typename TimePoint>
	void synchronized_queue<T>::push_at(TimePoint event_time, T const & x)
	{
		std::unique_lock lock{mutex_};
		push_cv_.wait(lock, [this]{ return size_internal() < max_size(); });
		if (event_time > clock::now())
		{
			deferred_heap_.push_back({std::chrono::time_point_cast<clock::duration>(event_time), x});
			std::push_heap(deferred_heap_.begin(), deferred_heap_.end(), heap_compare());
		}
		else
		{
			queue_.push_back(x);
		}
		lock.unlock();
		pop_cv_.notify_one();
	}

	template <typename T>
	template <typename TimePoint>
	void synchronized_queue<T>::push_at(TimePoint event_time, T && x)
	{
		std::unique_lock lock{mutex_};
		push_cv_.wait(lock, [this]{ return size_internal() < max_size(); });
		if (event_time > clock::now())
		{
			deferred_heap_.push_back({std::chrono::time_point_cast<clock::duration>(event_time), std::move(x)});
			std::push_heap(deferred_heap_.begin(), deferred_heap_.end(), heap_compare());
		}
		else
		{
			queue_.push_back(std::move(x));
		}
		lock.unlock();
		pop_cv_.notify_one();
	}

	template <typename T>
	bool synchronized_queue<T>::try_push(T const & x)
	{
		std::unique_lock lock{mutex_};
		if (size_internal() >= max_size())
			return false;

		queue_.push_back(x);
		lock.unlock();
		pop_cv_.notify_one();
		return true;
	}

	template <typename T>
	bool synchronized_queue<T>::try_push(T && x)
	{
		std::unique_lock lock{mutex_};
		if (size_internal() >= max_size())
			return false;

		queue_.push_back(std::move(x));
		lock.unlock();
		pop_cv_.notify_one();
		return true;
	}

	template <typename T>
	template <typename Duration>
	bool synchronized_queue<T>::try_push_for(T const & x, Duration wait_period)
	{
		std::unique_lock lock{mutex_};
		push_cv_.wait_for(lock, wait_period, [this]{ return size_internal() < max_size(); });
		if (size_internal() >= max_size())
			return false;

		queue_.push_back(x);
		lock.unlock();
		pop_cv_.notify_one();
		return true;
	}

	template <typename T>
	template <typename Duration>
	bool synchronized_queue<T>::try_push_for(T && x, Duration wait_period)
	{
		std::unique_lock lock{mutex_};
		push_cv_.wait_for(lock, wait_period, [this]{ return size_internal() < max_size(); });
		if (size_internal() >= max_size())
			return false;

		queue_.push_back(std::move(x));
		lock.unlock();
		pop_cv_.notify_one();
		return true;
	}

	template <typename T>
	template <typename TimePoint>
	bool synchronized_queue<T>::try_push_until(T const & x, TimePoint wait_time)
	{
		std::unique_lock lock{mutex_};
		push_cv_.wait_until(lock, wait_time, [this]{ return size_internal() < max_size(); });
		if (size_internal() >= max_size())
			return false;

		queue_.push_back(x);
		lock.unlock();
		pop_cv_.notify_one();
		return true;
	}

	template <typename T>
	template <typename TimePoint>
	bool synchronized_queue<T>::try_push_until(T && x, TimePoint wait_time)
	{
		std::unique_lock lock{mutex_};
		push_cv_.wait_until(lock, wait_time, [this]{ return size_internal() < max_size(); });
		if (size_internal() >= max_size())
			return false;

		queue_.push_back(std::move(x));
		lock.unlock();
		pop_cv_.notify_one();
		return true;
	}

	template <typename T>
	template <typename TimePoint>
	bool synchronized_queue<T>::try_push_at(TimePoint event_time, T const & x)
	{
		std::unique_lock lock{mutex_};
		if (size_internal() >= max_size())
			return false;

		if (event_time > clock::now())
		{
			deferred_heap_.push_back({std::chrono::time_point_cast<clock::duration>(event_time), x});
			std::push_heap(deferred_heap_.begin(), deferred_heap_.end(), heap_compare());
		}
		else
		{
			queue_.push_back(x);
		}
		lock.unlock();
		pop_cv_.notify_one();
		return true;
	}

	template <typename T>
	template <typename TimePoint>
	bool synchronized_queue<T>::try_push_at(TimePoint event_time, T && x)
	{
		std::unique_lock lock{mutex_};
		if (size_internal() >= max_size())
			return false;

		if (event_time > clock::now())
		{
			deferred_heap_.push_back({std::chrono::time_point_cast<clock::duration>(event_time), std::move(x)});
			std::push_heap(deferred_heap_.begin(), deferred_heap_.end(), heap_compare());
		}
		else
		{
			queue_.push_back(std::move(x));
		}
		lock.unlock();
		pop_cv_.notify_one();
		return true;
	}

	template <typename T>
	template <typename TimePoint, typename Duration>
	bool synchronized_queue<T>::try_push_at_for(TimePoint event_time, T const & x, Duration wait_period)
	{
		std::unique_lock lock{mutex_};
		push_cv_.wait_for(lock, wait_period, [this]{ return size_internal() < max_size(); });
		if (size_internal() >= max_size())
			return false;

		if (event_time > clock::now())
		{
			deferred_heap_.push_back({std::chrono::time_point_cast<clock::duration>(event_time), x});
			std::push_heap(deferred_heap_.begin(), deferred_heap_.end(), heap_compare());
		}
		else
		{
			queue_.push_back(x);
		}
		lock.unlock();
		pop_cv_.notify_one();
		return true;
	}

	template <typename T>
	template <typename TimePoint, typename Duration>
	bool synchronized_queue<T>::try_push_at_for(TimePoint event_time, T && x, Duration wait_period)
	{
		std::unique_lock lock{mutex_};
		push_cv_.wait_for(lock, wait_period, [this]{ return size_internal() < max_size(); });
		if (size_internal() >= max_size())
			return false;

		if (event_time > clock::now())
		{
			deferred_heap_.push_back({std::chrono::time_point_cast<clock::duration>(event_time), std::move(x)});
			std::push_heap(deferred_heap_.begin(), deferred_heap_.end(), heap_compare());
		}
		else
		{
			queue_.push_back(std::move(x));
		}
		lock.unlock();
		pop_cv_.notify_one();
		return true;
	}

	template <typename T>
	template <typename EventTimePoint, typename WaitTimePoint>
	bool synchronized_queue<T>::try_push_at_until(EventTimePoint event_time, T const & x, WaitTimePoint wait_time)
	{
		std::unique_lock lock{mutex_};
		push_cv_.wait_until(lock, wait_time, [this]{ return size_internal() < max_size(); });
		if (size_internal() >= max_size())
			return false;

		if (event_time > clock::now())
		{
			deferred_heap_.push_back({std::chrono::time_point_cast<clock::duration>(event_time), x});
			std::push_heap(deferred_heap_.begin(), deferred_heap_.end(), heap_compare());
		}
		else
		{
			queue_.push_back(x);
		}
		lock.unlock();
		pop_cv_.notify_one();
		return true;
	}

	template <typename T>
	template <typename EventTimePoint, typename WaitTimePoint>
	bool synchronized_queue<T>::try_push_at_until(EventTimePoint event_time, T && x, WaitTimePoint wait_time)
	{
		std::unique_lock lock{mutex_};
		push_cv_.wait_until(lock, wait_time, [this]{ return size_internal() < max_size(); });
		if (size_internal() >= max_size())
			return false;

		if (event_time > clock::now())
		{
			deferred_heap_.push_back({std::chrono::time_point_cast<clock::duration>(event_time), std::move(x)});
			std::push_heap(deferred_heap_.begin(), deferred_heap_.end(), heap_compare());
		}
		else
		{
			queue_.push_back(std::move(x));
		}
		lock.unlock();
		pop_cv_.notify_one();
		return true;
	}

	template <typename T>
	T synchronized_queue<T>::pop()
	{
		std::unique_lock lock{mutex_};
		while (true)
		{
			pop_cv_.wait(lock, [this]{ return !empty_internal(); });
			flush_deferred();
			if (!queue_.empty())
			{
				T x = std::move(queue_.front());
				queue_.pop_front();
				lock.unlock();
				push_cv_.notify_one();
				return x;
			}

			pop_cv_.wait_until(lock, deferred_heap_.front().time);
		}
	}

	template <typename T>
	std::optional<T> synchronized_queue<T>::try_pop()
	{
		std::unique_lock lock{mutex_};
		flush_deferred();
		if (queue_.empty())
			return std::nullopt;
		T x = std::move(queue_.front());
		queue_.pop_front();
		lock.unlock();
		push_cv_.notify_one();
		return { std::move(x) };
	}

	template <typename T>
	template <typename Duration>
	std::optional<T> synchronized_queue<T>::try_pop_for(Duration wait_period)
	{
		return try_pop_until(clock::now() + wait_period);
	}

	template <typename T>
	template <typename TimePoint>
	std::optional<T> synchronized_queue<T>::try_pop_until(TimePoint wait_time)
	{
		std::unique_lock lock{mutex_};
		while (clock::now() < wait_time)
		{
			pop_cv_.wait_until(lock, wait_time, [this]{ return !empty_internal(); });
			flush_deferred();
			if (!queue_.empty())
			{
				T x = std::move(queue_.front());
				queue_.pop_front();
				lock.unlock();
				push_cv_.notify_one();
				return x;
			}

			// here, queue_.empty()

			if (deferred_heap_.empty())
			{
				// queue_.empty() && deferred_heap_.empty() mean than the wait_until exited by timeout
				return std::nullopt;
			}

			pop_cv_.wait_until(lock, std::min(deferred_heap_.front().time, wait_time));
		}
		return std::nullopt;
	}

	template <typename T>
	void synchronized_queue<T>::clear()
	{
		std::unique_lock lock{mutex_};
		queue_.clear();
		deferred_heap_.clear();
		lock.unlock();
		push_cv_.notify_all();
	}

	template <typename T>
	void synchronized_queue<T>::wait()
	{
		std::unique_lock lock{mutex_};
		push_cv_.wait(lock, [this]{ return empty_internal(); });
	}

	template <typename T>
	template <typename Duration>
	bool synchronized_queue<T>::wait_for(Duration wait_period)
	{
		std::unique_lock lock{mutex_};
		return push_cv_.wait_for(lock, wait_period, [this]{ return empty_internal(); });
	}

	template <typename T>
	template <typename TimePoint>
	bool synchronized_queue<T>::wait_until(TimePoint wait_time)
	{
		std::unique_lock lock{mutex_};
		return push_cv_.wait_until(lock, wait_time, [this]{ return empty_internal(); });
	}

	template <typename T>
	std::size_t synchronized_queue<T>::size() const
	{
		std::lock_guard lock{mutex_};
		return size_internal();
	}

	template <typename T>
	bool synchronized_queue<T>::empty() const
	{
		std::lock_guard lock{mutex_};
		return empty_internal();
	}

	template <typename T>
	void synchronized_queue<T>::flush_deferred()
	{
		auto const now = clock::now();
		while (!deferred_heap_.empty() && deferred_heap_.front().time <= now)
		{
			queue_.push_back(std::move(deferred_heap_.front().value));
			std::pop_heap(deferred_heap_.begin(), deferred_heap_.end(), heap_compare());
			deferred_heap_.pop_back();
		}
	}


	template <typename T>
	std::size_t synchronized_queue<T>::size_internal() const
	{
		return queue_.size() + deferred_heap_.size();
	}

	template <typename T>
	bool synchronized_queue<T>::empty_internal() const
	{
		return queue_.empty() && deferred_heap_.empty();
	}

}
