module executor

import time

fn test_run_sync_from_foreign_thread_waits_until_owner_executes_job() {
    mut ex := new(queue_size: 4)!
    sync_started := chan bool{cap: 1}
    sync_ran := chan bool{cap: 1}
    sync_done := chan bool{cap: 1}
    run_result := chan string{cap: 1}

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

    caller := spawn fn [mut ex, sync_started, sync_ran, sync_done] () {
        sync_started <- true
        ex.run_sync(fn [sync_ran] () ! {
            sync_ran <- true
        }) or {
            sync_done <- false
            return
        }
        sync_done <- true
    }()

    assert wait_for_bool(sync_started, 'run_sync caller did not start')
    assert wait_for_bool(sync_ran, 'run_sync job did not run on owner pump')
    assert wait_for_bool(sync_done, 'run_sync caller did not finish')
    ex.stop()
    assert wait_for_string(run_result, 'run did not stop') == 'ok'
    ex.wait()!
    caller.wait()
    runner.wait()
}

fn test_run_sync_returns_job_error_to_foreign_caller() {
    mut ex := new(queue_size: 2)!
    run_result := chan string{cap: 1}
    sync_result := chan string{cap: 1}

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

    caller := spawn fn [mut ex, sync_result] () {
        ex.run_sync(fn () ! {
            return error('sync job failed')
        }) or {
            sync_result <- err.msg()
            return
        }
        sync_result <- 'ok'
    }()

    assert wait_for_string(sync_result, 'run_sync error was not returned') == 'sync job failed'
    assert wait_for_string(run_result, 'run did not return after sync error') == 'sync job failed'
    ex.wait() or { assert err.msg() == 'sync job failed' }
    caller.wait()
    runner.wait()
}

fn test_run_sync_from_owner_runs_inline_without_deadlock() {
    mut ex := new(queue_size: 4)!
    events := chan int{cap: 4}

    ex.try_post(fn [mut ex, events] () ! {
        events <- 1
        ex.run_sync(fn [events] () ! {
            events <- 100
        })!
        events <- 2
    })!
    ex.try_post(fn [events] () ! {
        events <- 3
    })!

    assert ex.drain_pending(2)! == 2
    assert wait_for_int(events, 'outer job did not start') == 1
    assert wait_for_int(events, 'inline run_sync job did not run') == 100
    assert wait_for_int(events, 'outer job did not finish') == 2
    assert wait_for_int(events, 'second queued job did not run') == 3

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

fn test_run_sync_owner_inline_error_caught_closes_admission_and_wait_returns_error() {
    mut ex := new(queue_size: 4)!
    events := chan int{cap: 3}
    post_result := chan string{cap: 1}
    run_result := chan string{cap: 1}

    ex.try_post(fn [mut ex, events, post_result] () ! {
        events <- 1
        ex.run_sync(fn () ! {
            return error('inline sync failed')
        }) or { events <- 2 }
        ex.try_post(fn () ! {}) or {
            post_result <- err.msg()
            return
        }
        post_result <- 'accepted'
        ex.stop()
    })!
    ex.try_post(fn [events] () ! {
        events <- 3
    })!

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

    assert wait_for_int(events, 'outer job did not start') == 1
    assert wait_for_int(events, 'inline run_sync error was not caught') == 2
    assert wait_for_string(post_result, 'try_post after inline error did not return') == 'executor: executor is closed'
    assert wait_for_int(events, 'accepted job after inline error was not drained') == 3
    assert wait_for_string(run_result, 'run did not return inline first error') == 'inline sync failed'
    ex.wait() or {
        assert err.msg() == 'inline sync failed'
        runner.wait()
        return
    }
    assert false, 'wait returned successfully after inline run_sync error'
}

fn test_run_one_returns_owner_inline_run_sync_error_caught_by_outer_job() {
    mut ex := new(queue_size: 4)!
    events := chan int{cap: 2}

    ex.try_post(fn [mut ex, events] () ! {
        ex.run_sync(fn () ! {
            return error('inline sync failed')
        }) or { events <- 1 }
    })!
    ex.try_post(fn [events] () ! {
        events <- 2
    })!

    mut saw_inline_error := false
    ex.run_one() or {
        assert err.msg() == 'inline sync failed'
        saw_inline_error = true
    }
    assert saw_inline_error
    assert wait_for_int(events, 'outer job did not catch inline run_sync error') == 1

    mut saw_closed := false
    ex.try_post(fn () ! {}) or {
        assert err.msg() == 'executor: executor is closed'
        saw_closed = true
    }
    assert saw_closed

    ex.run_one() or { assert err.msg() == 'inline sync failed' }
    assert wait_for_int(events, 'accepted job after inline error was not drained') == 2
    ex.wait() or {
        assert err.msg() == 'inline sync failed'
        return
    }
    assert false, 'wait returned successfully after inline run_sync error'
}

fn test_drain_pending_limit_returns_owner_inline_run_sync_error_caught_by_outer_job() {
    mut ex := new(queue_size: 4)!
    events := chan int{cap: 2}

    ex.try_post(fn [mut ex, events] () ! {
        ex.run_sync(fn () ! {
            return error('inline sync failed')
        }) or { events <- 1 }
    })!
    ex.try_post(fn [events] () ! {
        events <- 2
    })!

    ex.drain_pending(1) or { assert err.msg() == 'inline sync failed' }
    assert wait_for_int(events, 'outer job did not catch inline run_sync error') == 1

    ex.drain_pending(8) or { assert err.msg() == 'inline sync failed' }
    assert wait_for_int(events, 'accepted job after inline error was not drained') == 2
    ex.wait() or {
        assert err.msg() == 'inline sync failed'
        return
    }
    assert false, 'wait returned successfully after inline run_sync error'
}

fn test_run_sync_from_owner_after_stop_returns_closed_without_running_nested_job() {
    mut ex := new(queue_size: 2)!
    result := chan string{cap: 1}
    nested_ran := chan bool{cap: 1}

    ex.try_post(fn [mut ex, result, nested_ran] () ! {
        ex.stop()
        ex.run_sync(fn [nested_ran] () ! {
            nested_ran <- true
        }) or {
            result <- err.msg()
            return
        }
        result <- 'ok'
    })!

    assert ex.run_one()!
    assert wait_for_string(result, 'owner run_sync after stop did not return') == 'executor: executor is closed'
    select {
        _ := <-nested_ran {
            assert false, 'nested run_sync job ran after stop'
        }
        else {}
    }
    ex.wait()!
}

fn test_run_sync_from_owner_after_prior_job_error_is_refused_without_nested_execution() {
    mut ex := new(queue_size: 3)!
    result := chan string{cap: 1}
    nested_ran := chan bool{cap: 1}

    ex.try_post(fn () ! {
        return error('first job failed')
    })!
    ex.try_post(fn [mut ex, result, nested_ran] () ! {
        ex.run_sync(fn [nested_ran] () ! {
            nested_ran <- true
        }) or {
            result <- err.msg()
            return
        }
        result <- 'ok'
    })!

    mut run_result := 'ok'
    ex.run() or { run_result = err.msg() }
    assert run_result == 'first job failed'
    assert wait_for_string(result, 'owner run_sync after first error did not return') == 'executor: executor is closed'
    select {
        _ := <-nested_ran {
            assert false, 'nested run_sync job ran after first error'
        }
        else {}
    }
    ex.wait() or {
        assert err.msg() == 'first job failed'
        return
    }
    assert false, 'wait returned successfully after first job error'
}

fn test_run_sync_after_stop_returns_error() {
    mut ex := new(queue_size: 1)!
    ex.stop()
    ex.run_sync(fn () ! {}) or {
        assert err.msg() == 'executor: executor is closed'
        return
    }
    assert false, 'run_sync accepted after stop'
}

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

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_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 ''
}