import context
import time
import x.async as xasync

fn test_pool_rejects_invalid_config() {
    xasync.new_pool(workers: 0, queue_size: 1) or {
        assert err.msg() == 'async: pool worker count must be positive'
        return
    }
    assert false
}

fn test_pool_rejects_invalid_queue_size() {
    xasync.new_pool(workers: 1, queue_size: 0) or {
        assert err.msg() == 'async: pool queue size must be positive'
        return
    }
    assert false
}

fn test_pool_respects_worker_count() {
    mut pool := xasync.new_pool(workers: 2, queue_size: 4)!
    started := chan bool{cap: 4}
    release := chan bool{cap: 4}
    for _ in 0 .. 4 {
        pool.try_submit(fn [started, release] (mut ctx context.Context) ! {
            _ = ctx
            started <- true
            _ := <-release
        })!
    }

    wait_for_bool_signal(started, 'first pool job did not start')
    wait_for_bool_signal(started, 'second pool job did not start')
    select {
        _ := <-started {
            assert false, 'pool started more jobs than worker count before release'
        }
        100 * time.millisecond {}
    }

    for _ in 0 .. 4 {
        release <- true
    }
    pool.close()!
}

fn test_pool_queue_full_returns_backpressure_error() {
    mut pool := xasync.new_pool(workers: 1, queue_size: 1)!
    started := chan bool{cap: 1}
    release := chan bool{cap: 2}
    blocking_job := fn [started, release] (mut ctx context.Context) ! {
        _ = ctx
        started <- true
        _ := <-release
    }

    pool.try_submit(blocking_job)!
    wait_for_bool_signal(started, 'blocking pool job did not start')
    pool.try_submit(blocking_job)!
    pool.try_submit(blocking_job) or {
        assert err.msg() == 'async: pool queue is full'
        release <- true
        release <- true
        pool.close()!
        return
    }
    assert false
}

fn test_pool_submit_after_close_is_refused() {
    mut pool := xasync.new_pool(workers: 1, queue_size: 1)!
    pool.close()!
    pool.try_submit(fn (mut ctx context.Context) ! {
        _ = ctx
    }) or {
        assert err.msg() == 'async: pool is closed'
        return
    }
    assert false
}

fn test_pool_wait_is_one_shot() {
    mut pool := xasync.new_pool(workers: 1, queue_size: 1)!
    pool.wait()!
    pool.wait() or {
        assert err.msg() == 'async: pool wait was already called'
        return
    }
    assert false
}

fn test_pool_refuses_nil_job() {
    mut pool := xasync.new_pool(workers: 1, queue_size: 1)!
    nil_job := unsafe { xasync.JobFn(nil) }
    pool.try_submit(nil_job) or {
        assert err.msg() == 'async: job function is nil'
        pool.close()!
        return
    }
    assert false
}

fn test_pool_close_waits_for_accepted_jobs() {
    mut pool := xasync.new_pool(workers: 1, queue_size: 1)!
    release := chan bool{cap: 1}
    closed := chan bool{cap: 1}
    pool.try_submit(fn [release] (mut ctx context.Context) ! {
        _ = ctx
        _ := <-release
    })!

    close_thread := spawn fn [mut pool, closed] () {
        pool.close() or {
            closed <- false
            return
        }
        closed <- true
    }()
    select {
        _ := <-closed {
            assert false, 'pool close returned before accepted job completed'
        }
        100 * time.millisecond {}
    }
    release <- true
    select {
        ok := <-closed {
            assert ok
        }
        1 * time.second {
            assert false, 'pool close did not return after accepted job completed'
        }
    }
    close_thread.wait()
}

fn test_pool_first_error_is_propagated() {
    mut pool := xasync.new_pool(workers: 1, queue_size: 2)!
    started := chan bool{cap: 1}
    release := chan bool{cap: 1}
    pool.try_submit(fn [started, release] (mut ctx context.Context) ! {
        _ = ctx
        started <- true
        _ := <-release
        return error('first pool failure')
    })!
    wait_for_bool_signal(started, 'first pool failure job did not start')
    pool.try_submit(fn (mut ctx context.Context) ! {
        _ = ctx
        return error('second pool failure')
    })!
    release <- true
    pool.close() or {
        assert err.msg() == 'first pool failure'
        return
    }
    assert false
}

fn test_pool_error_does_not_drop_accepted_jobs() {
    mut pool := xasync.new_pool(workers: 1, queue_size: 1)!
    accepted_job_ran := chan bool{cap: 1}
    pool.try_submit(fn (mut ctx context.Context) ! {
        _ = ctx
        return error('first pool failure')
    })!
    pool.try_submit(fn [accepted_job_ran] (mut ctx context.Context) ! {
        _ = ctx
        accepted_job_ran <- true
    })!
    pool.close() or {
        assert err.msg() == 'first pool failure'
        select {
            did_run := <-accepted_job_ran {
                assert did_run
            }
            1 * time.second {
                assert false, 'accepted pool job did not run after earlier error'
            }
        }
        return
    }
    assert false
}

fn test_pool_concurrent_errors_return_one_error_and_drain_accepted_jobs() {
    error_jobs := 4
    ok_jobs := 8
    mut pool := xasync.new_pool(workers: error_jobs, queue_size: ok_jobs)!
    started := chan bool{cap: error_jobs}
    release := chan bool{cap: error_jobs}
    completed_ok := chan bool{cap: ok_jobs}
    for i in 0 .. error_jobs {
        pool.try_submit(fn [started, release, i] (mut ctx context.Context) ! {
            _ = ctx
            started <- true
            _ := <-release
            return error('pool concurrent failure ${i}')
        })!
    }
    for _ in 0 .. ok_jobs {
        pool.try_submit(fn [completed_ok] (mut ctx context.Context) ! {
            _ = ctx
            completed_ok <- true
        })!
    }
    for _ in 0 .. error_jobs {
        wait_for_bool_signal(started, 'pool error job did not start')
    }
    for _ in 0 .. error_jobs {
        release <- true
    }
    pool.close() or {
        assert err.msg().starts_with('pool concurrent failure ')
        for _ in 0 .. ok_jobs {
            wait_for_bool_signal(completed_ok, 'accepted ok pool job did not complete')
        }
        return
    }
    assert false
}

fn test_pool_close_drains_many_accepted_jobs_while_finishing() {
    jobs := 12
    workers := 3
    mut pool := xasync.new_pool(workers: workers, queue_size: jobs - workers)!
    started := chan bool{cap: jobs}
    release := chan bool{cap: jobs}
    finished := chan bool{cap: jobs}
    closed := chan bool{cap: 1}
    for _ in 0 .. jobs {
        pool.try_submit(fn [started, release, finished] (mut ctx context.Context) ! {
            _ = ctx
            started <- true
            _ := <-release
            finished <- true
        })!
    }
    for _ in 0 .. workers {
        wait_for_bool_signal(started, 'initial pool job did not start')
    }
    close_thread := spawn fn [mut pool, closed] () {
        pool.close() or {
            closed <- false
            return
        }
        closed <- true
    }()
    wait_until_pool_rejects_as_closed(mut pool)
    assert_no_bool_signal(closed, 'pool close returned while accepted jobs were still blocked')

    for _ in 0 .. jobs - 1 {
        release <- true
    }
    for _ in 0 .. jobs - 1 {
        wait_for_bool_signal(finished, 'accepted pool job did not finish')
    }
    assert_no_bool_signal(closed, 'pool close returned before the last accepted job finished')

    release <- true
    wait_for_bool_signal(finished, 'last accepted pool job did not finish')
    wait_for_bool_signal(closed, 'pool close did not drain accepted jobs')
    close_thread.wait()
}

fn test_pool_parent_cancellation_is_observed_by_cooperative_job() {
    parent_ctx, cancel := xasync.with_cancel()
    mut pool := xasync.new_pool_with_context(parent_ctx, workers: 1, queue_size: 1)!
    started := chan bool{cap: 1}
    pool.try_submit(fn [started] (mut ctx context.Context) ! {
        started <- true
        done := ctx.done()
        select {
            _ := <-done {
                return ctx.err()
            }
            1 * time.second {
                return error('pool job did not observe parent cancellation')
            }
        }
        return error('unreachable')
    })!
    wait_for_bool_signal(started, 'pool job did not start before parent cancellation')
    cancel()
    pool.close() or {
        assert err.msg() == 'context canceled'
        return
    }
    assert false
}

fn test_pool_non_cooperative_job_finishes_naturally_after_parent_cancel() {
    parent_ctx, cancel := xasync.with_cancel()
    mut pool := xasync.new_pool_with_context(parent_ctx, workers: 1, queue_size: 1)!
    finished := chan bool{cap: 1}
    pool.try_submit(fn [finished] (mut ctx context.Context) ! {
        _ = ctx
        time.sleep(20 * time.millisecond)
        finished <- true
    })!
    cancel()
    pool.close()!
    assert <-finished
}

fn test_pool_short_stress_many_jobs() {
    jobs := 100
    mut pool := xasync.new_pool(workers: 4, queue_size: jobs)!
    done := chan int{cap: jobs}
    for i in 0 .. jobs {
        pool.try_submit(fn [done, i] (mut ctx context.Context) ! {
            _ = ctx
            done <- i
        })!
    }
    pool.close()!

    mut seen := []bool{len: jobs}
    for _ in 0 .. jobs {
        i := <-done
        assert i >= 0
        assert i < jobs
        assert !seen[i]
        seen[i] = true
    }
    for was_seen in seen {
        assert was_seen
    }
}

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

fn assert_no_bool_signal(signal chan bool, message string) {
    select {
        _ := <-signal {
            assert false, message
        }
        else {}
    }
}

fn wait_until_pool_rejects_as_closed(mut pool xasync.Pool) {
    probe := fn (mut ctx context.Context) ! {
        _ = ctx
    }
    for _ in 0 .. 100 {
        pool.try_submit(probe) or {
            if err.msg() == 'async: pool is closed' {
                return
            }
            assert err.msg() == 'async: pool queue is full'
            time.sleep(1 * time.millisecond)
            continue
        }
        assert false, 'pool accepted probe while backlog should be full'
    }
    assert false, 'pool close did not start'
}