package ru.dbotthepony.kstarbound.util

import java.util.*
import java.util.concurrent.Callable
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.Delayed
import java.util.concurrent.ExecutionException
import java.util.concurrent.Future
import java.util.concurrent.FutureTask
import java.util.concurrent.RejectedExecutionException
import java.util.concurrent.ScheduledExecutorService
import java.util.concurrent.ScheduledFuture
import java.util.concurrent.TimeUnit
import java.util.concurrent.locks.LockSupport
import java.util.function.Consumer
import kotlin.NoSuchElementException
import kotlin.collections.ArrayList

private fun <E : Comparable<E>> LinkedList<E>.enqueue(value: E) {
	if (isEmpty()) {
		add(value)
	} else if (first >= value) {
		addFirst(value)
	} else if (last <= value) {
		addLast(value)
	} else {
		val iterator = listIterator()

		while (iterator.hasNext()) {
			val i = iterator.next()

			if (i >= value) {
				iterator.previous()
				iterator.add(value)
				break
			}
		}
	}
}

/**
 * [ScheduledExecutorService] which act as a mailbox, [executeQueuedTasks] must be called from main thread.
 *
 * [submit], [execute], etc can be called on any thread. If any of enqueueing methods are called on the same thread
 * as where [executeQueuedTasks] was called, executes provided lambda immediately and returns completed future.
 */
class MailboxExecutorService(@Volatile var thread: Thread = Thread.currentThread()) : ScheduledExecutorService {
	private val futureQueue = ConcurrentLinkedQueue<FutureTask<*>>()

	private val timers = LinkedList<Timer<*>>()
	private val repeatableTimers = LinkedList<RepeatableTimer>()

	@Volatile
	private var isShutdown = false
	@Volatile
	private var isTerminated = false

	private val timeOrigin = JVMClock()

	var exceptionHandler: Consumer<Throwable>? = null

	private inner class Timer<T>(task: Callable<T>, val executeAt: Long) : FutureTask<T>(task), ScheduledFuture<T> {
		override fun compareTo(other: Delayed): Int {
			return getDelay(TimeUnit.NANOSECONDS).compareTo(other.getDelay(TimeUnit.NANOSECONDS))
		}

		override fun getDelay(unit: TimeUnit): Long {
			return unit.convert(executeAt, TimeUnit.NANOSECONDS) - timeOrigin.nanos
		}
	}

	private data class CompletedFuture<T>(private val value: T) : Future<T> {
		override fun cancel(mayInterruptIfRunning: Boolean): Boolean {
			return false
		}

		override fun isCancelled(): Boolean {
			return false
		}

		override fun isDone(): Boolean {
			return true
		}

		override fun get(): T {
			return value
		}

		override fun get(timeout: Long, unit: TimeUnit): T {
			return value
		}

		companion object {
			val VOID = CompletedFuture(Unit)
		}
	}

	private inner class RepeatableTimer(
		task: Runnable,
		initialDelay: Long,
		val period: Long,
		val fixedDelay: Boolean,
	): FutureTask<Unit>({ task.run() }), ScheduledFuture<Unit> {
		var next = initialDelay
			private set

		public override fun runAndReset(): Boolean {
			if (fixedDelay) {
				next += period
				return super.runAndReset()
			} else {
				try {
					return super.runAndReset()
				} finally {
					next += period
				}
			}
		}

		override fun compareTo(other: Delayed): Int {
			return getDelay(TimeUnit.NANOSECONDS).compareTo(other.getDelay(TimeUnit.NANOSECONDS))
		}

		override fun getDelay(unit: TimeUnit): Long {
			return unit.convert(next, TimeUnit.NANOSECONDS) - timeOrigin.nanos
		}
	}

	fun isSameThread(): Boolean {
		return Thread.currentThread() === thread
	}

	fun executeQueuedTasks() {
		thread = Thread.currentThread()

		if (isShutdown) {
			if (!isTerminated) {
				isTerminated = true

				futureQueue.forEach {
					it.cancel(false)
				}

				futureQueue.clear()
				timers.clear()
				repeatableTimers.clear()

				return
			}
		}

		var next = futureQueue.poll()

		while (next != null) {
			if (isTerminated) return
			next.run()
			Thread.interrupted()

			try {
				next.get()
			} catch (err: ExecutionException) {
				exceptionHandler?.accept(err)
			}

			next = futureQueue.poll()
		}

		while (!timers.isEmpty()) {
			if (isTerminated) return
			val first = timers.first

			if (first.isCancelled) {
				timers.removeFirst()
			} else if (first.executeAt <= timeOrigin.nanos) {
				first.run()
				Thread.interrupted()

				try {
					first.get()
				} catch (err: ExecutionException) {
					exceptionHandler?.accept(err)
				}

				timers.removeFirst()
			} else {
				break
			}
		}

		if (repeatableTimers.isNotEmpty()) {
			val executed = LinkedList<RepeatableTimer>()

			while (repeatableTimers.isNotEmpty()) {
				if (isTerminated) return
				val first = repeatableTimers.first

				if (first.isDone) {
					repeatableTimers.removeFirst()
				} else if (first.next <= timeOrigin.nanos) {
					if (first.runAndReset()) {
						executed.add(first)
					}

					repeatableTimers.removeFirst()
				} else {
					break
				}
			}

			executed.forEach { repeatableTimers.enqueue(it) }
		}
	}

	override fun execute(command: Runnable) {
		if (isShutdown) throw RejectedExecutionException("This mailbox is shutting down")

		if (isSameThread()) {
			command.run()
		} else {
			futureQueue.add(FutureTask(command, Unit))
			LockSupport.unpark(thread)
		}
	}

	override fun shutdown() {
		isShutdown = true
	}

	override fun shutdownNow(): List<Runnable> {
		if (isTerminated) return listOf()
		isShutdown = true
		isTerminated = true

		val result = ArrayList<Runnable>()

		futureQueue.forEach {
			it.cancel(false)
			result.add(it)
		}

		futureQueue.clear()

		timers.forEach { it.cancel(false) }
		repeatableTimers.forEach { it.cancel(false) }

		timers.clear()
		repeatableTimers.clear()

		return result
	}

	override fun isShutdown(): Boolean {
		return isShutdown
	}

	override fun isTerminated(): Boolean {
		return isTerminated
	}

	override fun awaitTermination(timeout: Long, unit: TimeUnit): Boolean {
		throw UnsupportedOperationException()
	}

	override fun <T : Any?> submit(task: Callable<T>): Future<T> {
		if (isShutdown) throw RejectedExecutionException("This mailbox is shutting down")
		if (isSameThread()) return CompletedFuture(task.call())
		return FutureTask(task).also { futureQueue.add(it); LockSupport.unpark(thread) }
	}

	override fun <T : Any?> submit(task: Runnable, result: T): Future<T> {
		if (isShutdown) throw RejectedExecutionException("This mailbox is shutting down")
		if (isSameThread()) { task.run(); return CompletedFuture(result) }
		return FutureTask { task.run(); result }.also { futureQueue.add(it); LockSupport.unpark(thread) }
	}

	override fun submit(task: Runnable): Future<*> {
		if (isShutdown) throw RejectedExecutionException("This mailbox is shutting down")
		if (isSameThread()) { task.run(); return CompletedFuture.VOID }
		return FutureTask { task.run() }.also { futureQueue.add(it); LockSupport.unpark(thread) }
	}

	override fun <T : Any?> invokeAll(tasks: Collection<Callable<T>>): List<Future<T>> {
		if (isShutdown) throw RejectedExecutionException("This mailbox is shutting down")

		if (isSameThread()) {
			return tasks.map { CompletedFuture(it.call()) }
		} else {
			return tasks.map { submit(it) }.onEach { it.get() }
		}
	}

	override fun <T : Any?> invokeAll(
		tasks: Collection<Callable<T>>,
		timeout: Long,
		unit: TimeUnit
	): List<Future<T>> {
		if (isShutdown) throw RejectedExecutionException("This mailbox is shutting down")

		if (isSameThread()) {
			return tasks.map { CompletedFuture(it.call()) }
		} else {
			return tasks.map { submit(it) }.onEach { it.get(timeout, unit) }
		}
	}

	override fun <T : Any?> invokeAny(tasks: Collection<Callable<T>>): T {
		if (tasks.isEmpty())
			throw NoSuchElementException("Provided task list is empty")

		if (isShutdown) throw RejectedExecutionException("This mailbox is shutting down")

		if (isSameThread()) {
			return tasks.first().call()
		} else {
			return submit(tasks.first()).get()
		}
	}

	override fun <T : Any?> invokeAny(tasks: Collection<Callable<T>>, timeout: Long, unit: TimeUnit): T {
		if (tasks.isEmpty())
			throw NoSuchElementException("Provided task list is empty")

		if (isShutdown) throw RejectedExecutionException("This mailbox is shutting down")

		if (isSameThread()) {
			return tasks.first().call()
		} else {
			return submit(tasks.first()).get(timeout, unit)
		}
	}

	fun <V> join(future: Future<V>): V {
		if (isShutdown) throw RejectedExecutionException("This mailbox is shutting down")

		if (!isSameThread())
			return future.get()

		while (!future.isDone) {
			executeQueuedTasks()
			LockSupport.parkNanos(1_000_000L)
		}

		return future.get()
	}

	override fun schedule(command: Runnable, delay: Long, unit: TimeUnit): ScheduledFuture<*> {
		if (isShutdown) throw RejectedExecutionException("This mailbox is shutting down")
		val timer = Timer({ command.run() }, timeOrigin.nanos + TimeUnit.NANOSECONDS.convert(delay, unit))

		if (isSameThread() && delay <= 0L) {
			timer.run()
			Thread.interrupted()
		} else if (isSameThread()) {
			timers.enqueue(timer)
		} else {
			execute {
				if (timer.isCancelled) {
					// do nothing
				} else if (timer.executeAt <= timeOrigin.nanos) {
					timer.run()
					Thread.interrupted()
				} else {
					timers.enqueue(timer)
				}
			}
		}

		return timer
	}

	override fun <V : Any?> schedule(callable: Callable<V>, delay: Long, unit: TimeUnit): ScheduledFuture<V> {
		if (isShutdown) throw RejectedExecutionException("This mailbox is shutting down")

		val timer = Timer(callable, timeOrigin.nanos + TimeUnit.NANOSECONDS.convert(delay, unit))

		if (isSameThread() && delay <= 0L) {
			timer.run()
			Thread.interrupted()
		} else if (isSameThread()) {
			timers.enqueue(timer)
		} else {
			execute {
				if (timer.isCancelled) {
					// do nothing
				} else if (timer.executeAt <= timeOrigin.nanos) {
					timer.run()
					Thread.interrupted()
				} else {
					timers.enqueue(timer)
				}
			}
		}

		return timer
	}

	override fun scheduleAtFixedRate(
		command: Runnable,
		initialDelay: Long,
		period: Long,
		unit: TimeUnit
	): ScheduledFuture<*> {
		if (isShutdown) throw RejectedExecutionException("This mailbox is shutting down")

		return RepeatableTimer(
			command,
			timeOrigin.nanos + TimeUnit.NANOSECONDS.convert(initialDelay, unit),
			TimeUnit.NANOSECONDS.convert(period, unit), true)
			.also {
				execute {
					if (it.isCancelled) {
						// do nothing
					} else {
						repeatableTimers.enqueue(it)
					}
				}
			}
	}

	override fun scheduleWithFixedDelay(
		command: Runnable,
		initialDelay: Long,
		delay: Long,
		unit: TimeUnit
	): ScheduledFuture<*> {
		if (isShutdown) throw RejectedExecutionException("This mailbox is shutting down")

		return RepeatableTimer(
			command,
			timeOrigin.nanos + TimeUnit.NANOSECONDS.convert(initialDelay, unit),
			TimeUnit.NANOSECONDS.convert(delay, unit), false)
			.also {
				execute {
					if (it.isCancelled) {
						// do nothing
					} else {
						repeatableTimers.enqueue(it)
					}
				}
			}
	}
}