module executor

import context
import time

fn test_try_post_returns_backpressure_when_queue_is_full() {
	mut ex := new(queue_size: 1)!
	ex.try_post(fn () ! {})!
	ex.try_post(fn () ! {}) or {
		assert err.msg() == 'executor: queue is full'
		ex.stop()
		return
	}
	assert false, 'try_post accepted a job beyond queue capacity'
}

fn test_post_with_timeout_returns_timeout_when_queue_stays_full() {
	mut ex := new(queue_size: 1)!
	ex.try_post(fn () ! {})!
	ex.post_with_timeout(20 * time.millisecond, fn () ! {}) or {
		assert err.msg() == 'executor: timeout'
		ex.stop()
		return
	}
	assert false, 'post_with_timeout accepted a job while the queue stayed full'
}

fn test_post_with_timeout_zero_timeout_rejects_available_capacity() {
	mut ex := new(queue_size: 1)!
	ran := chan bool{cap: 1}
	ex.post_with_timeout(time.Duration(0), fn [ran] () ! {
		ran <- true
		return
	}) or {
		assert err.msg() == 'executor: timeout'
		assert ex.drain_pending(1)! == 0
		assert_no_bool(ran, 'zero-timeout admission job ran')
		ex.stop()
		ex.wait()!
		return
	}
	assert false, 'post_with_timeout accepted a job with zero timeout'
}

fn test_post_with_timeout_negative_timeout_rejects_available_capacity() {
	mut ex := new(queue_size: 1)!
	ran := chan bool{cap: 1}
	ex.post_with_timeout(-1 * time.nanosecond, fn [ran] () ! {
		ran <- true
		return
	}) or {
		assert err.msg() == 'executor: timeout'
		assert ex.drain_pending(1)! == 0
		assert_no_bool(ran, 'negative-timeout admission job ran')
		ex.stop()
		ex.wait()!
		return
	}
	assert false, 'post_with_timeout accepted a job with negative timeout'
}

fn test_post_with_timeout_long_timeout_accepts_available_capacity() {
	jobs := 4
	mut ex := new(queue_size: jobs)!
	ran := chan int{cap: jobs}

	for i in 0 .. jobs {
		ex.post_with_timeout(1 * time.hour, fn [ran, i] () ! {
			value := i + 1
			ran <- value
			return
		})!
	}

	assert ex.drain_pending(jobs)! == jobs
	mut total := 0
	for _ in 0 .. jobs {
		total += <-ran
	}
	assert total == 10

	ex.stop()
	assert !ex.run_one()!
	ex.wait()!
}

fn test_post_with_context_returns_parent_cancellation_without_accepting_job() {
	mut ex := new(queue_size: 1)!
	ex.try_post(fn () ! {})!
	mut background := context.background()
	mut parent, cancel := context.with_cancel(mut background)
	cancel()
	ran := chan bool{cap: 1}

	ex.post_with_context(parent, fn [ran] () ! {
		ran <- true
	}) or {
		assert err.msg() == 'context canceled'
		ex.stop()
		assert_no_bool(ran, 'canceled admission job ran')
		return
	}
	assert false, 'post_with_context accepted after parent cancellation'
}

fn test_post_with_context_rejects_nil_job() {
	mut ex := new(queue_size: 1)!
	nil_job := unsafe { JobFn(nil) }
	ex.post_with_context(context.background(), nil_job) or {
		assert err.msg() == 'executor: job function is nil'
		ex.stop()
		return
	}
	assert false, 'post_with_context accepted a nil job'
}

fn test_post_with_timeout_rejects_nil_job() {
	mut ex := new(queue_size: 1)!
	nil_job := unsafe { JobFn(nil) }
	ex.post_with_timeout(1 * time.second, nil_job) or {
		assert err.msg() == 'executor: job function is nil'
		ex.stop()
		return
	}
	assert false, 'post_with_timeout accepted a nil job'
}

fn test_owner_post_with_context_full_queue_returns_owner_capacity_error() {
	mut ex := new(queue_size: 1)!
	owner_started := chan bool{cap: 1}
	filler_accepted := chan bool{cap: 1}
	owner_result := chan string{cap: 1}
	filler_ran := chan bool{cap: 1}
	nested_ran := chan bool{cap: 1}

	ex.try_post(fn [mut ex, owner_started, filler_accepted, owner_result, nested_ran] () ! {
		owner_started <- true
		assert wait_for_bool(filler_accepted, 'filler job was not accepted')
		ex.post_with_context(context.background(), fn [nested_ran] () ! {
			nested_ran <- true
		}) or {
			owner_result <- err.msg()
			return
		}
		owner_result <- 'accepted'
	})!

	filler := spawn fn [mut ex, owner_started, filler_accepted, filler_ran] () {
		assert wait_for_bool(owner_started, 'owner callback did not start')
		ex.try_post(fn [filler_ran] () ! {
			filler_ran <- true
		}) or {
			filler_accepted <- false
			return
		}
		filler_accepted <- true
	}()

	runner := spawn fn [mut ex, owner_result] () {
		ex.run_one() or {
			owner_result <- err.msg()
			return
		}
	}()

	msg := wait_for_string(owner_result, 'owner post_with_context did not return')
	assert msg == 'executor: owner thread cannot wait for queue capacity'

	ex.stop()
	runner.wait()
	filler.wait()
	assert ex.run_one()!
	assert wait_for_bool(filler_ran, 'filler job did not run after owner rejection')
	assert_no_bool(nested_ran, 'owner post_with_context job ran after rejection')
	ex.wait()!
}

fn test_owner_post_with_timeout_full_queue_returns_owner_capacity_error() {
	mut ex := new(queue_size: 1)!
	owner_started := chan bool{cap: 1}
	filler_accepted := chan bool{cap: 1}
	owner_result := chan string{cap: 1}
	filler_ran := chan bool{cap: 1}
	nested_ran := chan bool{cap: 1}

	ex.try_post(fn [mut ex, owner_started, filler_accepted, owner_result, nested_ran] () ! {
		owner_started <- true
		assert wait_for_bool(filler_accepted, 'filler job was not accepted')
		ex.post_with_timeout(1 * time.hour, fn [nested_ran] () ! {
			nested_ran <- true
		}) or {
			owner_result <- err.msg()
			return
		}
		owner_result <- 'accepted'
	})!

	filler := spawn fn [mut ex, owner_started, filler_accepted, filler_ran] () {
		assert wait_for_bool(owner_started, 'owner callback did not start')
		ex.try_post(fn [filler_ran] () ! {
			filler_ran <- true
		}) or {
			filler_accepted <- false
			return
		}
		filler_accepted <- true
	}()

	runner := spawn fn [mut ex, owner_result] () {
		ex.run_one() or {
			owner_result <- err.msg()
			return
		}
	}()

	msg := wait_for_string(owner_result, 'owner post_with_timeout did not return')
	assert msg == 'executor: owner thread cannot wait for queue capacity'

	ex.stop()
	runner.wait()
	filler.wait()
	assert ex.run_one()!
	assert wait_for_bool(filler_ran, 'filler job did not run after owner rejection')
	assert_no_bool(nested_ran, 'owner post_with_timeout job ran after rejection')
	ex.wait()!
}

fn test_owner_try_post_full_queue_returns_queue_full_and_pump_continues() {
	mut ex := new(queue_size: 1)!
	owner_started := chan bool{cap: 1}
	filler_accepted := chan bool{cap: 1}
	owner_result := chan string{cap: 1}
	filler_ran := chan bool{cap: 1}
	nested_ran := chan bool{cap: 1}

	ex.try_post(fn [mut ex, owner_started, filler_accepted, owner_result, nested_ran] () ! {
		owner_started <- true
		assert wait_for_bool(filler_accepted, 'filler job was not accepted')
		ex.try_post(fn [nested_ran] () ! {
			nested_ran <- true
		}) or {
			owner_result <- err.msg()
			return
		}
		owner_result <- 'accepted'
	})!

	filler := spawn fn [mut ex, owner_started, filler_accepted, filler_ran] () {
		assert wait_for_bool(owner_started, 'owner callback did not start')
		ex.try_post(fn [filler_ran] () ! {
			filler_ran <- true
		}) or {
			filler_accepted <- false
			return
		}
		filler_accepted <- true
	}()

	assert ex.run_one()!
	assert wait_for_string(owner_result, 'owner try_post did not return') == 'executor: queue is full'
	filler.wait()
	assert ex.run_one()!
	assert wait_for_bool(filler_ran, 'pump did not continue to filler job')
	assert_no_bool(nested_ran, 'owner try_post job ran after queue-full rejection')

	ex.stop()
	assert !ex.run_one()!
	ex.wait()!
}

fn test_owner_bounded_posts_accept_when_capacity_is_available() {
	mut ex := new(queue_size: 2)!
	owner_result := chan string{cap: 1}
	context_ran := chan bool{cap: 1}
	timeout_ran := chan bool{cap: 1}

	ex.try_post(fn [mut ex, owner_result, context_ran, timeout_ran] () ! {
		ex.post_with_context(context.background(), fn [context_ran] () ! {
			context_ran <- true
		}) or {
			owner_result <- err.msg()
			return
		}
		ex.post_with_timeout(1 * time.hour, fn [timeout_ran] () ! {
			timeout_ran <- true
		}) or {
			owner_result <- err.msg()
			return
		}
		owner_result <- 'accepted'
	})!

	assert ex.run_one()!
	assert wait_for_string(owner_result, 'owner bounded posts did not return') == 'accepted'
	assert ex.run_one()!
	assert ex.run_one()!
	assert wait_for_bool(context_ran, 'owner post_with_context job did not run')
	assert wait_for_bool(timeout_ran, 'owner post_with_timeout job did not run')

	ex.stop()
	assert !ex.run_one()!
	ex.wait()!
}

fn test_post_with_context_waits_until_capacity_opens() {
	mut ex := new(queue_size: 1)!
	release := chan bool{cap: 1}
	started := chan bool{cap: 1}
	attempting := chan bool{cap: 1}
	accepted := chan bool{cap: 1}
	ran := chan bool{cap: 1}
	owner_result := chan string{cap: 1}

	ex.try_post(fn [started, release] () ! {
		started <- true
		_ := <-release
	})!

	submitter := spawn fn [mut ex, attempting, accepted, ran] () {
		attempting <- true
		ex.post_with_context(context.background(), fn [ran] () ! {
			ran <- true
		}) or {
			accepted <- false
			return
		}
		accepted <- true
	}()
	assert wait_for_bool(attempting, 'bounded submitter did not start')
	assert_no_bool(accepted, 'post_with_context returned before capacity opened')

	owner := spawn fn [mut ex, owner_result] () {
		ran_one := ex.run_one() or {
			owner_result <- err.msg()
			return
		}
		if ran_one {
			owner_result <- 'ran'
		} else {
			owner_result <- 'empty'
		}
	}()
	assert wait_for_bool(started, 'blocking job did not start')
	assert wait_for_bool(accepted, 'post_with_context did not accept after capacity opened')
	release <- true
	assert wait_for_string(owner_result, 'owner run_one did not return') == 'ran'
	assert ex.run_one()!
	assert wait_for_bool(ran, 'bounded admission job did not run')

	ex.stop()
	assert !ex.run_one()!
	ex.wait()!
	owner.wait()
	submitter.wait()
}

fn wait_for_bool(signal chan bool, message string) bool {
	select {
		value := <-signal {
			return value
		}
		1 * time.second {
			assert false, message
		}
	}
	return false
}

fn wait_for_string(signal chan string, message string) string {
	select {
		value := <-signal {
			return value
		}
		1 * time.second {
			assert false, message
		}
	}
	return ''
}

fn assert_no_bool(signal chan bool, message string) {
	select {
		_ := <-signal {
			assert false, message
		}
		50 * time.millisecond {}
	}
}