module executor

import time

fn test_first_job_error_is_returned_after_accepted_jobs_drain() {
	mut ex := new(queue_size: 4)!
	ran := chan int{cap: 2}
	result := chan string{cap: 1}

	ex.try_post(fn [ran] () ! {
		ran <- 1
		return error('first owner failure')
	})!
	ex.try_post(fn [ran] () ! {
		ran <- 2
	})!

	runner := spawn fn [mut ex, result] () {
		ex.run() or {
			result <- err.msg()
			return
		}
		result <- 'ok'
	}()

	assert wait_for_int(ran, 'first job did not run') == 1
	assert wait_for_int(ran, 'accepted job after error did not drain') == 2
	assert wait_for_string(result, 'run did not return first job error') == 'first owner failure'
	ex.wait() or {
		assert err.msg() == 'first owner failure'
		runner.wait()
		return
	}
	assert false, 'wait did not return first job error'
}

fn test_admission_closes_after_first_job_error() {
	mut ex := new(queue_size: 1)!
	ex.try_post(fn () ! {
		return error('owner failure')
	})!
	mut saw_job_error := false
	ex.run_one() or {
		assert err.msg() == 'owner failure'
		saw_job_error = true
	}
	assert saw_job_error

	ex.try_post(fn () ! {}) or {
		assert err.msg() == 'executor: executor is closed'
		ex.wait() or { assert err.msg() == 'owner failure' }
		return
	}
	assert false, 'try_post accepted after first job error'
}

fn test_run_one_after_job_error_requires_continued_pumping_to_drain() {
	mut ex := new(queue_size: 3)!
	ran := chan int{cap: 2}

	ex.try_post(fn [ran] () ! {
		ran <- 1
		return error('owner failure')
	})!
	ex.try_post(fn [ran] () ! {
		ran <- 2
	})!

	ex.run_one() or { assert err.msg() == 'owner failure' }
	assert wait_for_int(ran, 'erroring job did not run') == 1

	ex.run_one() or { assert err.msg() == 'owner failure' }
	assert wait_for_int(ran, 'accepted job after error was not drained') == 2

	ex.wait() or {
		assert err.msg() == 'owner failure'
		return
	}
	assert false, 'wait did not return the stored owner failure'
}

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

	ex.try_post(fn [ran] () ! {
		ran <- 1
		return error('owner failure')
	})!
	ex.try_post(fn [ran] () ! {
		ran <- 2
	})!
	ex.try_post(fn [ran] () ! {
		ran <- 3
	})!

	ex.drain_pending(1) or { assert err.msg() == 'owner failure' }
	assert wait_for_int(ran, 'erroring drain job did not run') == 1

	ex.drain_pending(8) or { assert err.msg() == 'owner failure' }
	assert wait_for_int(ran, 'second accepted job was not drained') == 2
	assert wait_for_int(ran, 'third accepted job was not drained') == 3

	ex.wait() or {
		assert err.msg() == 'owner failure'
		return
	}
	assert false, 'wait did not return the stored owner failure'
}

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

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