"""Durable job coordinator: idempotent submission, atomic claim, locks, progress
from durable item state, cancellation, stale recovery, and restart persistence."""

from concurrent.futures import ThreadPoolExecutor
from datetime import timedelta

import pytest

from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
from photo_pipeline.services.jobs import (
    ItemState,
    JobBlocked,
    JobService,
    JobState,
    _now,
)


@pytest.fixture
def db_url(tmp_path):
    url = f"sqlite:///{tmp_path / 'jobs.db'}"
    run_migrations(url)
    return url


@pytest.fixture
def make_factory():
    engines = []

    def _make(url):
        engine = create_db_engine(url)
        engines.append(engine)
        return create_session_factory(engine)

    yield _make
    for engine in engines:
        engine.dispose()


@pytest.fixture
def jobs(db_url, make_factory):
    return JobService(make_factory(db_url))


def test_enqueue_persists_items_and_event(jobs):
    job = jobs.enqueue("scan", items=["a", "b", "c"])
    assert job["state"] == JobState.QUEUED
    assert jobs.progress(job["id"]) == {
        "total": 3,
        "done": 0,
        "by_state": {ItemState.QUEUED: 3},
    }
    assert any(e["type"] == "queued" for e in jobs.events(job["id"]))


def test_idempotency_key_returns_same_job(jobs):
    a = jobs.enqueue("scan", idempotency_key="key-1")
    b = jobs.enqueue("scan", idempotency_key="key-1")
    assert a["id"] == b["id"]


def test_concurrent_idempotent_submission_creates_one_job(db_url, make_factory):
    service = JobService(make_factory(db_url))
    with ThreadPoolExecutor(max_workers=8) as pool:
        ids = list(
            pool.map(lambda _: service.enqueue("scan", idempotency_key="race")["id"], range(8))
        )
    assert len(set(ids)) == 1


def test_one_mutating_job_lock_blocks(jobs):
    jobs.enqueue("scan", lock="library_write")
    with pytest.raises(JobBlocked) as excinfo:
        jobs.enqueue("rename", lock="library_write")
    assert excinfo.value.code == "lock_held"
    assert jobs.blockers("library_write")


def test_lock_frees_after_terminal(jobs):
    first = jobs.enqueue("scan", lock="library_write")
    jobs.claim(["scan"], "w1")
    jobs.transition(first["id"], JobState.SUCCEEDED, worker_id="w1")
    # Lock is free now; a new mutating job enqueues fine.
    assert jobs.enqueue("rename", lock="library_write")["state"] == JobState.QUEUED


def test_atomic_claim_gives_job_to_one_worker(db_url, make_factory):
    service = JobService(make_factory(db_url))
    service.enqueue("scan")
    with ThreadPoolExecutor(max_workers=4) as pool:
        claims = list(pool.map(lambda i: service.claim(["scan"], f"w{i}"), range(4)))
    winners = [c for c in claims if c is not None]
    assert len(winners) == 1
    assert winners[0]["state"] == JobState.RUNNING
    assert winners[0]["fencing_token"] == 1


def test_progress_derives_from_item_state(jobs):
    job = jobs.enqueue("scan", items=["a", "b"])
    jobs.set_item(job["id"], "a", ItemState.RUNNING)
    jobs.set_item(job["id"], "a", ItemState.SUCCEEDED)
    jobs.set_item(job["id"], "b", ItemState.RUNNING)
    jobs.set_item(job["id"], "b", ItemState.FAILED, error=("boom", "kaboom"))
    progress = jobs.progress(job["id"])
    assert progress["total"] == 2 and progress["done"] == 2
    assert progress["by_state"] == {ItemState.SUCCEEDED: 1, ItemState.FAILED: 1}


def test_cancel_running_job_drains_via_cancelling(jobs):
    job = jobs.enqueue("scan")
    jobs.claim(["scan"], "w1")
    after = jobs.cancel(job["id"])
    assert after["state"] == JobState.CANCELLING


def test_cancel_queued_job_is_immediate(jobs):
    job = jobs.enqueue("scan")
    assert jobs.cancel(job["id"])["state"] == JobState.CANCELLED


def test_recover_stale_requeues_expired_leases(jobs):
    job = jobs.enqueue("scan")
    jobs.claim(["scan"], "w1", lease_seconds=30)
    recovered = jobs.recover_stale(now=_now() + timedelta(hours=1))
    assert job["id"] in recovered
    assert jobs.get(job["id"])["state"] == JobState.RETRY_QUEUED
    assert jobs.get(job["id"])["lease_owner"] is None


def test_heartbeat_only_from_lease_owner(jobs):
    job = jobs.enqueue("scan")
    jobs.claim(["scan"], "w1")
    assert jobs.heartbeat(job["id"], "w1") is True
    assert jobs.heartbeat(job["id"], "intruder") is False


def test_state_durable_across_restart(db_url, make_factory):
    first = JobService(make_factory(db_url))
    job = first.enqueue("scan", items=["a", "b"])
    first.claim(["scan"], "w1")
    first.set_item(job["id"], "a", ItemState.RUNNING)
    first.set_item(job["id"], "a", ItemState.SUCCEEDED)

    # Fresh service/engine against the same database.
    reopened = JobService(make_factory(db_url))
    assert reopened.get(job["id"])["state"] == JobState.RUNNING
    assert reopened.progress(job["id"])["done"] == 1