from __future__ import annotations

import json
from datetime import datetime, timezone
from pathlib import Path
from types import SimpleNamespace

import pytest

from osm_lead_source_service.errors import (
    ContractViolation,
    InvalidConfiguration,
    PartialCycleError,
    PbfStageContractError,
)
from osm_lead_source_service.markets import get_english_market
from osm_lead_source_service.pbf.catalog import CatalogFeature, MarketSourceShard
from osm_lead_source_service.worker import (
    WorkerConfig,
    build_market_plan,
    dashboard_snapshot,
    next_monthly_run,
    run_worker_once,
    start_scheduler_from_env,
)


def _feature(feature_id: str) -> CatalogFeature:
    return CatalogFeature(
        feature_id,
        None,
        feature_id,
        f"https://download.geofabrik.de/test/{feature_id}-latest.osm.pbf",
    )


def _shard(country_code: str, feature_id: str) -> MarketSourceShard:
    return MarketSourceShard(get_english_market(country_code), _feature(feature_id))


def _identity(osm_id: int) -> SimpleNamespace:
    return SimpleNamespace(
        osm_type=SimpleNamespace(value="node"),
        osm_id=osm_id,
    )


def _fake_result(
    shard: MarketSourceShard,
    *,
    report_hash: str,
    source_sha256: str,
    lead_ids: tuple[int, ...],
    review_ids: tuple[int, ...] = (),
) -> SimpleNamespace:
    leads = tuple(
        SimpleNamespace(
            identity=_identity(osm_id),
            category="Restaurant / cafe",
        )
        for osm_id in lead_ids
    )
    reviews = tuple(SimpleNamespace(identity=_identity(osm_id)) for osm_id in review_ids)
    report = SimpleNamespace(
        state_counts=(
            ("active", len(leads)),
            ("excluded", 2),
            ("out_of_scope", 100),
            ("pending_review", len(reviews)),
        ),
        category_counts=(("Restaurant / cafe", len(leads)),),
        region_id=shard.market.country_code.casefold(),
        profile=SimpleNamespace(profile_id="luxeillum", version="0.1.0-draft"),
        report_hash=report_hash,
        source_sha256=source_sha256,
        source_size_bytes=123,
        leads=leads,
        review_items=reviews,
        as_mapping=lambda: {"report_hash": report_hash, "leads": []},
    )
    acquisition = SimpleNamespace(cache_hit=False)
    return SimpleNamespace(
        market=shard.market,
        shard=shard,
        report=report,
        acquisition=acquisition,
        extraction=None,
        evidence_mapping=lambda: {"report_hash": report_hash},
    )


def _config(tmp_path: Path, countries: str = "IE,NZ") -> WorkerConfig:
    return WorkerConfig.from_env(
        {
            "OSM_LEAD_SOURCE_COUNTRIES": countries,
            "OSM_LEAD_SOURCE_CACHE_DIR": str(tmp_path / "cache"),
            "OSM_LEAD_SOURCE_REPORT_DIR": str(tmp_path / "reports"),
            "OSM_LEAD_SOURCE_DASHBOARD_SNAPSHOT_PATH": str(tmp_path / "dashboard.json"),
        }
    )


def test_worker_config_defaults_to_all_english_markets() -> None:
    config = WorkerConfig.from_env({})

    assert config.cache_dir == Path("/tmp/osm-lead-source/cache")
    assert config.dashboard_snapshot_path == Path("/tmp/osm-lead-source/dashboard.json")
    assert len(config.markets) == 58
    assert config.run_on_start is False
    assert config.schedule_day == 1
    assert config.schedule_hour_utc == 2


def test_worker_config_accepts_country_subset() -> None:
    config = WorkerConfig.from_env({"OSM_LEAD_SOURCE_COUNTRIES": "NZ,UK,AU"})

    assert tuple(market.country_code for market in config.markets) == (
        "AU",
        "GB",
        "NZ",
    )


def test_scheduler_start_requires_request_and_approval(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from osm_lead_source_service import worker

    class FakeThread:
        def __init__(self, **kwargs: object) -> None:
            self.kwargs = kwargs
            self.started = False

        def start(self) -> None:
            self.started = True

    monkeypatch.setattr(worker.threading, "Thread", FakeThread)

    assert start_scheduler_from_env({"OSM_LEAD_SOURCE_SCHEDULER_ENABLED": "true"}) is None
    assert start_scheduler_from_env({"OSM_LEAD_SOURCE_SCHEDULER_APPROVED": "true"}) is None

    thread = start_scheduler_from_env(
        {
            "OSM_LEAD_SOURCE_SCHEDULER_ENABLED": "true",
            "OSM_LEAD_SOURCE_SCHEDULER_APPROVED": "true",
        }
    )
    assert isinstance(thread, FakeThread)
    assert thread.started is True
    args = thread.kwargs["args"]
    assert isinstance(args, tuple)
    config = args[0]
    assert isinstance(config, WorkerConfig)
    assert config.run_on_start is False


@pytest.mark.parametrize(
    "variable",
    [
        "OSM_LEAD_SOURCE_NOCO_WRITE_ENABLED",
        "OSM_LEAD_SOURCE_WRITER_ENABLED",
        "OSM_LEAD_SOURCE_STALE_DELETION_ENABLED",
    ],
)
def test_worker_refuses_write_capabilities(variable: str) -> None:
    with pytest.raises(InvalidConfiguration):
        WorkerConfig.from_env({variable: "true"})


def test_next_monthly_run_rolls_to_next_month() -> None:
    now = datetime(2026, 7, 19, 12, 0, tzinfo=timezone.utc)

    assert next_monthly_run(now, day=1, hour_utc=2) == datetime(
        2026,
        8,
        1,
        2,
        0,
        tzinfo=timezone.utc,
    )


def test_build_market_plan_is_sorted_and_bounded(tmp_path: Path) -> None:
    config = _config(tmp_path)

    class FakeCatalog:
        def plan_market_leaf_shards(self, market: object) -> tuple[MarketSourceShard, ...]:
            code = market.country_code
            return (_shard(code, f"{code.casefold()}-leaf"),)

    plan = build_market_plan(config, catalog=FakeCatalog())

    assert tuple(item.market.country_code for item in plan) == ("IE", "NZ")


def test_dashboard_snapshot_contains_one_shard_coverage() -> None:
    shard = _shard("IE", "ireland")
    result = _fake_result(
        shard,
        report_hash="a" * 64,
        source_sha256="b" * 64,
        lead_ids=(1, 2),
        review_ids=(3,),
    )
    started = datetime(2026, 7, 19, 1, 0, tzinfo=timezone.utc)
    completed = datetime(2026, 7, 19, 1, 5, tzinfo=timezone.utc)

    snapshot = dashboard_snapshot(result, started_at=started, completed_at=completed)

    assert snapshot["counts"]["active"] == 2
    assert snapshot["coverage"] == {
        "configured_markets": 1,
        "completed_markets": 1,
        "failed_markets": 0,
        "configured_shards": 1,
        "completed_shards": 1,
        "failed_shards": 0,
    }
    assert snapshot["noco"]["comparison_completed"] is False


def test_run_worker_once_aggregates_and_deduplicates(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from osm_lead_source_service import worker

    ie = _shard("IE", "ireland-a")
    ie_overlap = _shard("IE", "ireland-b")
    monkeypatch.setattr(worker, "build_market_plan", lambda _config: (ie, ie_overlap))
    results = {
        "ireland-a": _fake_result(
            ie,
            report_hash="a" * 64,
            source_sha256="b" * 64,
            lead_ids=(1, 2),
            review_ids=(3,),
        ),
        "ireland-b": _fake_result(
            ie_overlap,
            report_hash="c" * 64,
            source_sha256="d" * 64,
            lead_ids=(2, 3),
        ),
    }
    monkeypatch.setattr(
        worker,
        "run_market_shard",
        lambda shard, *_args, **_kwargs: results[shard.feature.feature_id],
    )
    monkeypatch.setattr(worker, "_store_shard_result", lambda *_args: None)
    config = _config(tmp_path, "IE")
    moment = datetime(2026, 7, 20, 2, 0, tzinfo=timezone.utc)

    summary = run_worker_once(config, clock=lambda: moment)
    snapshot = json.loads(config.dashboard_snapshot_path.read_text())

    assert summary["status"] == "completed"
    assert summary["lead_count"] == 3
    assert snapshot["counts"]["active"] == 3
    assert snapshot["counts"]["review"] == 0
    assert snapshot["coverage"]["completed_markets"] == 1
    assert snapshot["coverage"]["completed_shards"] == 2
    assert snapshot["noco"]["comparison_completed"] is False


def test_run_worker_once_keeps_partial_progress_after_shard_failure(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from osm_lead_source_service import worker

    ie = _shard("IE", "ireland")
    nz = _shard("NZ", "new-zealand")
    monkeypatch.setattr(worker, "build_market_plan", lambda _config: (ie, nz))
    result = _fake_result(
        ie,
        report_hash="a" * 64,
        source_sha256="b" * 64,
        lead_ids=(1,),
    )

    def run(shard: MarketSourceShard, *_args: object, **_kwargs: object) -> object:
        if shard.market.country_code == "NZ":
            raise RuntimeError("isolated test failure")
        return result

    monkeypatch.setattr(worker, "run_market_shard", run)
    monkeypatch.setattr(worker, "_store_shard_result", lambda *_args: None)
    config = _config(tmp_path)
    moment = datetime(2026, 7, 20, 2, 0, tzinfo=timezone.utc)

    with pytest.raises(PartialCycleError) as raised:
        run_worker_once(config, clock=lambda: moment)
    summary = raised.value.summary
    snapshot = json.loads(config.dashboard_snapshot_path.read_text())

    assert summary["event"] == "read_only_cycle_partial"
    assert summary["status"] == "partial"
    assert summary["failed_shards"] == 1
    assert summary["failure_codes"] == {"RuntimeError": 1}
    assert snapshot["coverage"] == {
        "configured_markets": 2,
        "completed_markets": 1,
        "failed_markets": 1,
        "configured_shards": 2,
        "completed_shards": 1,
        "failed_shards": 1,
    }
    assert snapshot["failures"] == {"by_code": {"RuntimeError": 1}}
    assert snapshot["run"]["status"] == "partial"
    assert snapshot["counts"]["active"] == 1


def test_run_worker_once_records_bounded_parser_cause_chain(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    from osm_lead_source_service import worker

    ie = _shard("IE", "ireland")
    monkeypatch.setattr(worker, "build_market_plan", lambda _config: (ie,))

    def fail(*_args: object, **_kwargs: object) -> object:
        try:
            raise ContractViolation("parsed object contains a duplicate tag key")
        except ContractViolation as exc:
            raise PbfStageContractError("sink_accept", exc) from exc

    monkeypatch.setattr(worker, "run_market_shard", fail)
    config = _config(tmp_path, "IE")
    moment = datetime(2026, 7, 25, 10, 0, tzinfo=timezone.utc)

    with pytest.raises(PartialCycleError) as raised:
        run_worker_once(config, clock=lambda: moment)

    expected_code = "PbfStageContractError:sink_accept:duplicate_normalized_tag_key"
    assert raised.value.summary["failure_codes"] == {expected_code: 1}
    output = capsys.readouterr().out
    assert f'"error_code":"{expected_code}"' in output
    assert "duplicate tag key" not in output