from __future__ import annotations

import hashlib
import json
from typing import Mapping

import pytest

from osm_lead_source_service.errors import ContractViolation, InvalidConfiguration
from osm_lead_source_service.noco.write import (
    JsonTransport,
    NocoApiInsertRepository,
    NocoInsertConfig,
)
from osm_lead_source_service.writer.lifecycle import (
    LifecycleDecision,
    LifecycleGateError,
)
from osm_lead_source_service.writer.models import InsertCandidate, ShadowReportDocument
from osm_lead_source_service.writer.service import (
    ControlledInsertService,
    InMemoryInsertLedger,
)


def shadow_json(*, osm_id: int = 42) -> tuple[str, str]:
    document: dict[str, object] = {
        "category_counts": {"Hospitality/Restaurant or cafe": 1},
        "definition_digest": "d" * 64,
        "leads": [
            {
                "address": "1 fixture road",
                "category": "Hospitality",
                "city": "chiang mai",
                "country": "TH",
                "domain": "fixture.example",
                "evidence": ["amenity=restaurant"],
                "identity_hash": "b" * 64,
                "latitude_e7": 188000000,
                "longitude_e7": 989000000,
                "matched_rule_ids": ["hospitality.restaurant"],
                "name": "fixture cafe",
                "osm_id": osm_id,
                "osm_type": "node",
                "phone": "+6621234567",
                "postcode": "50000",
                "source_tags_hash": "c" * 64,
                "state": "chiang mai",
                "subcategory": "Restaurant or cafe",
            }
        ],
        "object_counts": {"nodes": 1, "relations": 0, "total": 1, "ways": 0},
        "profile": {"id": "luxeillum", "version": "0.1.0-draft"},
        "region_id": "thailand",
        "review_items": [],
        "source_sha256": "a" * 64,
        "source_size_bytes": 123,
        "state_counts": {"active": 1},
        "total_tags": 2,
    }
    canonical = json.dumps(document, sort_keys=True, separators=(",", ":")).encode()
    report_hash = hashlib.sha256(canonical).hexdigest()
    document["report_hash"] = report_hash
    return json.dumps(document), report_hash


class FakeTransport(JsonTransport):
    def __init__(
        self,
        *,
        duplicate_id: int | None = None,
        post_error: bool = False,
    ) -> None:
        self.duplicate_id = duplicate_id
        self.post_error = post_error
        self.requests: list[tuple[str, str, Mapping[str, object] | None]] = []

    def request(
        self,
        *,
        method: str,
        url: str,
        headers: Mapping[str, str],
        body: Mapping[str, object] | None,
        timeout_seconds: float,
    ) -> object:
        assert headers["xc-token"] == "fixture-token"
        assert timeout_seconds == 5
        self.requests.append((method, url, body))
        if method == "POST":
            if self.post_error:
                raise RuntimeError("response lost after POST")
            return {"Id": 9001}
        if self.duplicate_id is not None and "osm_identity" not in url:
            # URL contains an encoded where clause, not the local label. Return
            # duplicate evidence on the first GET only.
            duplicate = self.duplicate_id
            self.duplicate_id = None
            return {"list": [{"Id": duplicate}]}
        return {"list": []}


class FakeLifecycleGate:
    def __init__(
        self,
        decision: str = "ALLOW",
        *,
        observe_error: bool = False,
    ) -> None:
        self.decision = decision
        self.observe_error = observe_error
        self.evaluated: list[InsertCandidate] = []
        self.observations: list[tuple[int, bool]] = []

    def evaluate(self, candidate: InsertCandidate) -> LifecycleDecision:
        self.evaluated.append(candidate)
        return LifecycleDecision(
            decision=self.decision,
            reason="fixture-decision",
            scope="website_only" if self.decision == "SUPPRESS_WEBSITE" else None,
            entity_id=None,
            site_key=None,
            replacement_allowed=self.decision in {"SUPPRESS_WEBSITE", "RESTORE_CANDIDATE"},
        )

    def observe(
        self,
        candidate: InsertCandidate,
        *,
        record_id: int,
        include_website: bool,
    ) -> None:
        del candidate
        if self.observe_error:
            raise LifecycleGateError("fixture observation failure")
        self.observations.append((record_id, include_website))


def repository(transport: JsonTransport) -> NocoApiInsertRepository:
    return NocoApiInsertRepository(
        config=NocoInsertConfig(
            base_url="https://nocodb.example",
            table_id="mswfxq541pe6khe",
            api_token="fixture-token",
            timeout_seconds=5,
        ),
        transport=transport,
    )


def parsed_document() -> ShadowReportDocument:
    content, report_hash = shadow_json()
    return ShadowReportDocument.from_json(
        content,
        expected_report_hash=report_hash,
        max_candidates=10,
    )


def test_shadow_document_recomputes_hash_and_builds_deterministic_payload() -> None:
    document = parsed_document()
    candidate = document.candidates[0]

    assert candidate.osm_url == "https://www.openstreetmap.org/node/42"
    assert candidate.website == "https://fixture.example"
    assert len(candidate.idempotency_key) == 64
    payload = candidate.noco_payload(
        run_id="pbf:thailand:test",
        run_timestamp="2026-07-16T09:00:00+00:00",
    )
    assert payload["source"] == "OSM"
    assert payload["osm_id"] == "42"
    assert payload["website"] == "https://fixture.example"
    assert payload["selector_id"] == "luxeillum:0.1.0-draft"

    content, report_hash = shadow_json()
    tampered = content.replace("fixture cafe", "different cafe")
    with pytest.raises(ContractViolation, match="hash"):
        ShadowReportDocument.from_json(
            tampered,
            expected_report_hash=report_hash,
            max_candidates=10,
        )


def test_controlled_writer_inserts_once_and_is_idempotent_in_ledger() -> None:
    transport = FakeTransport()
    ledger = InMemoryInsertLedger()
    service = ControlledInsertService(
        repository=repository(transport),
        ledger=ledger,
        lifecycle_gate=FakeLifecycleGate(),
        writer_enabled=True,
        legacy_writer_disabled=True,
    )

    first = service.run(
        parsed_document(),
        run_id="pbf:thailand:test",
        run_timestamp="2026-07-16T09:00:00+00:00",
        max_inserts=5,
    )
    second = service.run(
        parsed_document(),
        run_id="pbf:thailand:test",
        run_timestamp="2026-07-16T09:00:00+00:00",
        max_inserts=5,
    )

    assert first.inserted_count == 1
    assert second.already_applied_count == 1
    assert [method for method, _, _ in transport.requests].count("POST") == 1


def test_final_duplicate_check_blocks_post() -> None:
    transport = FakeTransport(duplicate_id=77)
    service = ControlledInsertService(
        repository=repository(transport),
        ledger=InMemoryInsertLedger(),
        lifecycle_gate=FakeLifecycleGate(),
        writer_enabled=True,
        legacy_writer_disabled=True,
    )

    result = service.run(
        parsed_document(),
        run_id="pbf:thailand:test",
        run_timestamp="2026-07-16T09:00:00+00:00",
        max_inserts=5,
    )

    assert result.duplicate_count == 1
    assert all(method == "GET" for method, _, _ in transport.requests)


def test_ambiguous_post_failure_remains_claimed() -> None:
    ledger = InMemoryInsertLedger()
    service = ControlledInsertService(
        repository=repository(FakeTransport(post_error=True)),
        ledger=ledger,
        lifecycle_gate=FakeLifecycleGate(),
        writer_enabled=True,
        legacy_writer_disabled=True,
    )
    document = parsed_document()

    with pytest.raises(RuntimeError, match="response lost"):
        service.run(
            document,
            run_id="pbf:thailand:test",
            run_timestamp="2026-07-16T09:00:00+00:00",
            max_inserts=5,
        )

    candidate = document.candidates[0]
    assert ledger.states[candidate.idempotency_key] == "claimed"


def test_writer_fails_closed_until_cutover() -> None:
    current = parsed_document()
    with pytest.raises(InvalidConfiguration, match="disabled"):
        ControlledInsertService(
            repository=repository(FakeTransport()),
            ledger=InMemoryInsertLedger(),
            lifecycle_gate=FakeLifecycleGate(),
            writer_enabled=False,
            legacy_writer_disabled=True,
        ).run(
            current,
            run_id="fixture",
            run_timestamp="2026-07-16T09:00:00+00:00",
            max_inserts=1,
        )
    with pytest.raises(InvalidConfiguration, match="legacy"):
        ControlledInsertService(
            repository=repository(FakeTransport()),
            ledger=InMemoryInsertLedger(),
            lifecycle_gate=FakeLifecycleGate(),
            writer_enabled=True,
            legacy_writer_disabled=False,
        ).run(
            current,
            run_id="fixture",
            run_timestamp="2026-07-16T09:00:00+00:00",
            max_inserts=1,
        )


def test_lifecycle_business_suppression_blocks_insert() -> None:
    transport = FakeTransport()
    gate = FakeLifecycleGate("SUPPRESS")
    result = ControlledInsertService(
        repository=repository(transport),
        ledger=InMemoryInsertLedger(),
        lifecycle_gate=gate,
        writer_enabled=True,
        legacy_writer_disabled=True,
    ).run(
        parsed_document(),
        run_id="fixture",
        run_timestamp="2026-07-16T09:00:00+00:00",
        max_inserts=1,
    )

    assert result.suppressed_count == 1
    assert result.inserted_count == 0
    assert transport.requests == []


def test_lifecycle_website_suppression_inserts_without_website() -> None:
    transport = FakeTransport()
    gate = FakeLifecycleGate("SUPPRESS_WEBSITE")
    result = ControlledInsertService(
        repository=repository(transport),
        ledger=InMemoryInsertLedger(),
        lifecycle_gate=gate,
        writer_enabled=True,
        legacy_writer_disabled=True,
    ).run(
        parsed_document(),
        run_id="fixture",
        run_timestamp="2026-07-16T09:00:00+00:00",
        max_inserts=1,
    )

    post_bodies = [body for method, _, body in transport.requests if method == "POST"]
    assert result.website_suppressed_count == 1
    assert result.inserted_count == 1
    assert post_bodies[0]["website"] == ""
    assert gate.observations == [(9001, False)]


def test_lifecycle_restore_candidate_is_counted_and_inserted() -> None:
    transport = FakeTransport()
    gate = FakeLifecycleGate("RESTORE_CANDIDATE")
    result = ControlledInsertService(
        repository=repository(transport),
        ledger=InMemoryInsertLedger(),
        lifecycle_gate=gate,
        writer_enabled=True,
        legacy_writer_disabled=True,
    ).run(
        parsed_document(),
        run_id="fixture",
        run_timestamp="2026-07-16T09:00:00+00:00",
        max_inserts=1,
    )

    assert result.restore_candidate_count == 1
    assert result.inserted_count == 1
    assert gate.observations == [(9001, True)]


def test_lifecycle_observation_failure_does_not_retry_insert() -> None:
    transport = FakeTransport()
    gate = FakeLifecycleGate(observe_error=True)
    result = ControlledInsertService(
        repository=repository(transport),
        ledger=InMemoryInsertLedger(),
        lifecycle_gate=gate,
        writer_enabled=True,
        legacy_writer_disabled=True,
    ).run(
        parsed_document(),
        run_id="fixture",
        run_timestamp="2026-07-16T09:00:00+00:00",
        max_inserts=1,
    )

    assert result.inserted_count == 1
    assert result.observation_failure_count == 1
    assert [method for method, _, _ in transport.requests].count("POST") == 1