"""The job queue (app/core/jobs.py): a job exists only once the transaction
that queued it commits, runs once at a time however many workers claim,
is retried with backoff, fails for good after its attempts (running its
on_failure hook), and is handed out again when its worker stops renewing
the lease. Plus the scheduler that feeds it and the API that shows it."""

import asyncio
import uuid
from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import text

from app.core import jobs, scheduler
from tests.conftest import AsyncSessionLocal
from app.models.scheduling import ScheduledJob

calls: list[tuple[str, dict]] = []
failures: list[str] = []


@jobs.handler("test.ok")
async def _ok(job: jobs.Job) -> str:
    calls.append(("ok", job.payload))
    return f"did {job.payload.get('n')}"


@jobs.handler("test.flaky", max_attempts=3)
async def _flaky(job: jobs.Job) -> str:
    calls.append(("flaky", {"attempt": job.attempt}))
    if job.attempt < job.payload["succeed_on"]:
        raise RuntimeError(f"attempt {job.attempt} failed")
    return "finally"


@jobs.on_failure("test.flaky")
async def _flaky_failed(job: jobs.Job, exc: BaseException) -> None:
    failures.append(str(exc))


@jobs.handler("test.permanent")
async def _permanent(job: jobs.Job) -> str:
    raise jobs.Permanent("can't ever work")


@jobs.handler("test.slow")
async def _slow(job: jobs.Job) -> str:
    await asyncio.sleep(0.3)
    calls.append(("slow", job.payload))
    return "slow"


@pytest.fixture(autouse=True)
def _reset():
    calls.clear()
    failures.clear()


async def _enqueue(tenant_id, kind, payload=None, **kw) -> uuid.UUID:
    async with AsyncSessionLocal() as db:
        async with db.begin():
            return await jobs.enqueue(db, kind, payload or {}, tenant_id=tenant_id, **kw)


async def _row(job_id):
    async with AsyncSessionLocal() as db:
        return (await db.execute(text("SELECT * FROM jobs WHERE id = :id"), {"id": job_id})).one()


async def _make_due(job_id):
    async with AsyncSessionLocal() as db:
        async with db.begin():
            await db.execute(text("UPDATE jobs SET run_after = now() WHERE id = :id"), {"id": job_id})


async def test_a_job_exists_only_once_its_transaction_commits(test_tenant):
    tenant_id = test_tenant["tenant_id"]
    async with AsyncSessionLocal() as db:
        async with db.begin():
            await jobs.enqueue(db, "test.ok", {"n": 1}, tenant_id=tenant_id)
            await db.rollback()
    assert await jobs.run_due(["test.ok"], tenant_id=tenant_id) == 0

    job_id = await _enqueue(tenant_id, "test.ok", {"n": 2})
    assert await jobs.run_due(["test.ok"], tenant_id=tenant_id) == 1
    assert calls == [("ok", {"n": 2})]
    row = await _row(job_id)
    assert (row.status, row.attempts, row.result, row.locked_by) == ("done", 1, "did 2", None)
    assert row.finished_at is not None


async def test_a_failing_job_is_retried_with_backoff_then_succeeds(test_tenant):
    tenant_id = test_tenant["tenant_id"]
    job_id = await _enqueue(tenant_id, "test.flaky", {"succeed_on": 2})
    await jobs.run_due(["test.flaky"], tenant_id=tenant_id)
    row = await _row(job_id)
    assert (row.status, row.attempts, row.last_error) == ("queued", 1, "RuntimeError: attempt 1 failed")
    wait = (row.run_after - datetime.now(timezone.utc)).total_seconds()
    assert 5 < wait <= 12  # 10s ±20%, less the time since

    assert await jobs.run_due(["test.flaky"], tenant_id=tenant_id) == 0  # not due yet
    await _make_due(job_id)
    await jobs.run_due(["test.flaky"], tenant_id=tenant_id)
    row = await _row(job_id)
    assert (row.status, row.attempts, row.result) == ("done", 2, "finally")
    assert failures == []


async def test_a_job_fails_for_good_after_its_attempts(test_tenant):
    tenant_id = test_tenant["tenant_id"]
    job_id = await _enqueue(tenant_id, "test.flaky", {"succeed_on": 99})
    for _ in range(3):
        await _make_due(job_id)
        await jobs.run_due(["test.flaky"], tenant_id=tenant_id)
    row = await _row(job_id)
    assert (row.status, row.attempts) == ("failed", 3)
    assert failures == ["attempt 3 failed"]  # the hook ran once, at the end
    await _make_due(job_id)
    assert await jobs.run_due(["test.flaky"], tenant_id=tenant_id) == 0


async def test_permanent_fails_at_once(test_tenant):
    job_id = await _enqueue(test_tenant["tenant_id"], "test.permanent")
    await jobs.run_due(["test.permanent"], tenant_id=test_tenant["tenant_id"])
    row = await _row(job_id)
    assert (row.status, row.attempts, row.last_error) == ("failed", 1, "Permanent: can't ever work")


def test_backoff_doubles_and_is_capped():
    assert 8 <= jobs.backoff(1) <= 12
    assert 16 <= jobs.backoff(2) <= 24
    assert jobs.backoff(30) <= 3600 * 1.2


async def test_a_dedupe_key_keeps_one_waiting_job_with_the_latest_payload(test_tenant):
    tenant_id = test_tenant["tenant_id"]
    first = await _enqueue(tenant_id, "test.ok", {"n": 1}, dedupe_key=f"test:{tenant_id}", delay=60)
    second = await _enqueue(tenant_id, "test.ok", {"n": 2}, dedupe_key=f"test:{tenant_id}")
    assert first == second
    row = await _row(first)
    assert row.payload == {"n": 2} and row.run_after <= datetime.now(timezone.utc)  # runs no later than the sooner one
    await jobs.run_due(["test.ok"], tenant_id=tenant_id)
    assert calls == [("ok", {"n": 2})]
    # Once it has run, the key is free again.
    third = await _enqueue(tenant_id, "test.ok", {"n": 3}, dedupe_key=f"test:{tenant_id}")
    assert third != first


async def test_concurrent_workers_never_claim_the_same_job(test_tenant):
    tenant_id = test_tenant["tenant_id"]
    ids = {await _enqueue(tenant_id, "test.slow", {"n": i}) for i in range(6)}
    # Claimers racing each other: a round can come back short (a row another
    # claimer just took is re-checked and dropped) — never with a duplicate.
    claimed: list[jobs.Job] = []
    while True:
        batches = await asyncio.gather(*(jobs._claim(2, ["test.slow"], tenant_id) for _ in range(5)))
        got = [j for batch in batches for j in batch]
        if not got:
            break
        claimed += got
    assert len(claimed) == len({j.id for j in claimed}) == 6 and {j.id for j in claimed} == ids
    await asyncio.gather(*(jobs.run_job(j) for j in claimed))
    assert len(calls) == 6


async def test_an_expired_lease_puts_the_job_back(test_tenant):
    tenant_id = test_tenant["tenant_id"]
    job_id = await _enqueue(tenant_id, "test.ok", {"n": 1}, max_attempts=2)
    [claimed] = await jobs._claim(1, ["test.ok"], tenant_id)
    # Its worker dies: the lease runs out unrenewed.
    async with AsyncSessionLocal() as db:
        async with db.begin():
            await db.execute(text("UPDATE jobs SET locked_until = now() - interval '1 second', locked_by = 'dead-worker' WHERE id = :id"), {"id": job_id})
    assert await jobs.reclaim_expired() >= 1
    row = await _row(job_id)
    assert (row.status, row.attempts, row.locked_by) == ("queued", 1, None)
    assert "stopped responding" in row.last_error

    # The dead worker's late finish doesn't overwrite the job's new state.
    await jobs._finish(claimed, "done", result="late")
    assert (await _row(job_id)).status == "queued"

    await jobs.run_due(["test.ok"], tenant_id=tenant_id)
    assert (await _row(job_id)).status == "done"


async def test_the_scheduler_queues_a_due_job_once_and_the_queue_runs_it(test_tenant):
    tenant_id = test_tenant["tenant_id"]
    async with AsyncSessionLocal() as db:
        async with db.begin():
            scheduled = ScheduledJob(
                tenant_id=tenant_id, name="Pytest SLA scan", job_type="sla_overdue_scan", frequency="hourly",
                next_run_at=datetime.now(timezone.utc) - timedelta(minutes=1),
            )
            db.add(scheduled)
            await db.flush()
            scheduled_id = scheduled.id

    # Two instances ticking together queue it once.
    await asyncio.gather(scheduler.tick(), scheduler.tick())
    async with AsyncSessionLocal() as db:
        queued = (await db.execute(text("SELECT count(*) FROM jobs WHERE tenant_id = :t AND kind = :k"), {"t": tenant_id, "k": scheduler.RUN_JOB})).scalar()
        job = await db.get(ScheduledJob, scheduled_id)
    assert queued == 1
    assert job.next_run_at > datetime.now(timezone.utc) + timedelta(minutes=50)
    assert job.last_run_at is None  # queued, not run

    assert await jobs.run_due([scheduler.RUN_JOB], tenant_id=tenant_id) == 1
    async with AsyncSessionLocal() as db:
        job = await db.get(ScheduledJob, scheduled_id)
    assert job.last_run_at is not None and job.last_run_result.startswith("scanned 0 in-progress")


async def test_administrators_see_retry_and_cancel_jobs(client, auth_headers, test_tenant):
    tenant_id = test_tenant["tenant_id"]
    failed = await _enqueue(tenant_id, "test.permanent")
    await jobs.run_due(["test.permanent"], tenant_id=tenant_id)
    waiting = await _enqueue(tenant_id, "test.ok", {"n": 1}, delay=3600)

    resp = await client.get("/jobs", headers=auth_headers)
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["counts"]["failed"] == 1 and body["counts"]["queued"] == 1
    by_id = {j["id"]: j for j in body["jobs"]}
    assert by_id[str(failed)]["last_error"] == "Permanent: can't ever work" and "payload" not in by_id[str(failed)]
    assert [j["id"] for j in (await client.get("/jobs?status=failed", headers=auth_headers)).json()["jobs"]] == [str(failed)]

    assert (await client.post(f"/jobs/{waiting}/retry", headers=auth_headers)).status_code == 409
    assert (await client.post(f"/jobs/{failed}/retry", headers=auth_headers)).status_code == 204
    row = await _row(failed)
    assert (row.status, row.attempts, row.last_error) == ("queued", 0, None)

    assert (await client.post(f"/jobs/{waiting}/cancel", headers=auth_headers)).status_code == 204
    assert (await _row(waiting)).status == "cancelled"
    assert (await client.post(f"/jobs/{waiting}/cancel", headers=auth_headers)).status_code == 409
    # Another tenant's job isn't reachable.
    assert (await client.post(f"/jobs/{uuid.uuid4()}/retry", headers=auth_headers)).status_code == 409