from __future__ import annotations

from collections.abc import Iterator
from hashlib import sha256

import pytest

from osm_lead_source_service.adoption import (
    AdoptionCandidate,
    AdoptionScanReport,
    AdoptionScanRequest,
    ExistingActiveMapping,
    scan_adoption_candidates,
)
from osm_lead_source_service.domain.enums import CandidateReviewOutcome, OsmObjectType
from osm_lead_source_service.domain.models import RegionScopedIdentity
from osm_lead_source_service.errors import ContractViolation
from osm_lead_source_service.noco import (
    FixtureNocoReadRepository,
    NocoColumn,
    NocoReadPage,
    NocoReadRequest,
    NocoSchema,
    NocoSourceRecord,
)
from osm_lead_source_service.normalization.models import NormalizationContext, NormalizedBusiness

_TAG_HASH = sha256(b"tags").hexdigest()
_ID_HASH = sha256(b"identity").hexdigest()


def schema(extra: tuple[NocoColumn, ...] = ()) -> NocoSchema:
    return NocoSchema(
        table_id="table-fixture",
        physical_schema="fixture_schema",
        physical_table="Scraper Leads",
        columns=(
            NocoColumn("id", "integer", False),
            NocoColumn("source", "text", True),
            NocoColumn("osm_type", "text", True),
            NocoColumn("osm_id", "text", True),
            NocoColumn("osm_url", "text", True),
            NocoColumn("name", "text", True),
            NocoColumn("website", "text", True),
            NocoColumn("phone", "varchar", True),
            NocoColumn("address", "text", True),
            NocoColumn("city", "text", True),
            NocoColumn("country_code", "text", True),
            NocoColumn("Blocked", "boolean", True),
        )
        + extra,
    )


def source(
    osm_type: OsmObjectType,
    osm_id: int,
    *,
    name: str = "fixture cafe",
    domain: str | None = "fixture.example",
    phone: str | None = "+6621234567",
    address: str | None = "1 fixture road",
    city: str | None = "chiang mai",
) -> NormalizedBusiness:
    return NormalizedBusiness(
        identity=RegionScopedIdentity("thailand", osm_type, osm_id),
        name=name,
        domain=domain,
        phone=phone,
        address=address,
        city=city,
        state=None,
        postcode=None,
        country="TH",
        longitude_e7=None,
        latitude_e7=None,
        source_tags_hash=_TAG_HASH,
        identity_hash=_ID_HASH,
    )


def request(current_schema: NocoSchema) -> AdoptionScanRequest:
    return AdoptionScanRequest(
        run_id="fixture-run",
        region_id="thailand",
        table_id=current_schema.table_id,
        expected_schema_fingerprint=current_schema.fingerprint,
        normalization_context=NormalizationContext(country_code="TH", calling_code="66"),
        read_request=NocoReadRequest(page_size=2, max_rows=100),
    )


def scan(rows: tuple[NocoSourceRecord, ...], sources: tuple[NormalizedBusiness, ...]):
    current_schema = schema()
    repository = FixtureNocoReadRepository(schema=current_schema, rows=rows)
    return scan_adoption_candidates(
        repository=repository,
        request=request(current_schema),
        sources=sources,
    )


def test_exact_type_and_id_is_reported_without_mutation() -> None:
    report = scan(
        (
            NocoSourceRecord(
                10,
                {
                    "source": "OSM",
                    "osm_type": "node",
                    "osm_id": "123",
                    "name": "Fixture Cafe",
                    "Blocked": True,
                },
            ),
        ),
        (source(OsmObjectType.NODE, 123),),
    )
    candidate = report.candidates[0]
    assert candidate.result_status is CandidateReviewOutcome.ACCEPTED
    assert candidate.proposed_outcome.value == "exact"
    assert candidate.protection is not None and candidate.protection.protected
    assert "workflow_field:Blocked" in candidate.protection.reasons
    assert "source_baseline:absent" in candidate.protection.reasons
    assert not hasattr(candidate, "execute")
    assert not hasattr(candidate, "executable")
    assert not hasattr(report, "execute")
    assert not hasattr(report, "executable")


def test_osm_url_adopts_when_identity_columns_are_blank() -> None:
    report = scan(
        (NocoSourceRecord(10, {"osm_url": "https://www.openstreetmap.org/way/456"}),),
        (source(OsmObjectType.WAY, 456),),
    )
    assert report.candidates[0].proposed_outcome.value == "exact"


def test_same_numeric_node_and_way_do_not_collide() -> None:
    report = scan(
        (
            NocoSourceRecord(1, {"source": "OSM", "osm_type": "node", "osm_id": "99"}),
            NocoSourceRecord(2, {"source": "OSM", "osm_type": "way", "osm_id": "99"}),
        ),
        (
            source(OsmObjectType.NODE, 99, name="node place", domain=None, phone=None),
            source(OsmObjectType.WAY, 99, name="way place", domain=None, phone=None),
        ),
    )
    assert [candidate.noco_id for candidate in report.candidates] == [1, 2]
    assert all(
        candidate.result_status is CandidateReviewOutcome.ACCEPTED
        for candidate in report.candidates
    )


def test_unique_domain_name_address_is_high_confidence() -> None:
    report = scan(
        (
            NocoSourceRecord(
                7,
                {
                    "name": "Fixture Cafe",
                    "website": "https://www.fixture.example/menu",
                    "address": "1 Fixture Road",
                    "country_code": "TH",
                },
            ),
        ),
        (source(OsmObjectType.NODE, 1, phone=None),),
    )
    candidate = report.candidates[0]
    assert candidate.result_status is CandidateReviewOutcome.ACCEPTED
    assert candidate.proposed_outcome.value == "high_confidence"
    assert candidate.rule == "domain_name_address"


def test_multiple_accepted_rows_create_collision_evidence() -> None:
    rows = (
        NocoSourceRecord(1, {"source": "OSM", "osm_type": "node", "osm_id": "5"}),
        NocoSourceRecord(2, {"source": "OSM", "osm_type": "node", "osm_id": "5"}),
    )
    report = scan(rows, (source(OsmObjectType.NODE, 5),))
    assert len(report.candidates) == 2
    assert all(
        candidate.result_status is CandidateReviewOutcome.COLLISION
        for candidate in report.candidates
    )
    assert all(
        "multiple_accepted_noco_matches" in candidate.collision_reasons
        for candidate in report.candidates
    )


def test_existing_mapping_conflict_is_collision() -> None:
    current_schema = schema()
    repository = FixtureNocoReadRepository(
        schema=current_schema,
        rows=(NocoSourceRecord(8, {"source": "OSM", "osm_type": "node", "osm_id": "8"}),),
    )
    report = scan_adoption_candidates(
        repository=repository,
        request=request(current_schema),
        sources=(source(OsmObjectType.NODE, 8),),
        existing_mappings=(
            ExistingActiveMapping(9, RegionScopedIdentity("thailand", OsmObjectType.NODE, 8)),
        ),
    )
    assert report.candidates[0].result_status is CandidateReviewOutcome.COLLISION
    assert "source_already_maps_to_other_noco_row" in report.candidates[0].collision_reasons


def test_accepted_match_does_not_hide_mapping_collision() -> None:
    current_schema = schema()
    repository = FixtureNocoReadRepository(
        schema=current_schema,
        rows=(
            NocoSourceRecord(
                8,
                {"source": "OSM", "osm_type": "node", "osm_id": "8"},
            ),
            NocoSourceRecord(
                9,
                {
                    "name": "Fixture Cafe",
                    "website": "https://fixture.example/menu",
                    "phone": "+6621234567",
                    "country_code": "TH",
                },
            ),
        ),
    )
    report = scan_adoption_candidates(
        repository=repository,
        request=request(current_schema),
        sources=(source(OsmObjectType.NODE, 8),),
        existing_mappings=(
            ExistingActiveMapping(8, RegionScopedIdentity("thailand", OsmObjectType.NODE, 8)),
        ),
    )
    assert [(candidate.noco_id, candidate.result_status) for candidate in report.candidates] == [
        (8, CandidateReviewOutcome.ACCEPTED),
        (9, CandidateReviewOutcome.COLLISION),
    ]
    assert "source_already_maps_to_other_noco_row" in report.candidates[1].collision_reasons


def test_ambiguous_partial_signals_never_propose_mapping() -> None:
    report = scan(
        (NocoSourceRecord(3, {"name": "Fixture Cafe", "city": "Chiang Mai"}),),
        (source(OsmObjectType.NODE, 3, domain=None, phone=None, address=None),),
    )
    candidate = report.candidates[0]
    assert candidate.result_status is CandidateReviewOutcome.AMBIGUOUS
    assert candidate.proposed_outcome is None


def test_name_address_country_city_without_category_is_ambiguous() -> None:
    report = scan(
        (
            NocoSourceRecord(
                4,
                {
                    "name": "Fixture Cafe",
                    "address": "1 Fixture Road",
                    "city": "Chiang Mai",
                    "country_code": "TH",
                },
            ),
        ),
        (source(OsmObjectType.NODE, 4, domain=None, phone=None),),
    )
    candidate = report.candidates[0]
    assert candidate.result_status is CandidateReviewOutcome.AMBIGUOUS
    assert candidate.proposed_outcome is None
    assert candidate.rule == "multiple_partial_signals"


def test_unmatched_source_is_explicit() -> None:
    report = scan((), (source(OsmObjectType.NODE, 404),))
    assert report.candidates[0].result_status is CandidateReviewOutcome.UNMATCHED
    assert report.candidates[0].noco_id is None


def test_schema_drift_fails_before_rows_are_scanned() -> None:
    actual = schema((NocoColumn("new_column", "text", True),))
    expected = schema()
    repository = FixtureNocoReadRepository(schema=actual, rows=())
    with pytest.raises(ContractViolation, match="fingerprint"):
        scan_adoption_candidates(
            repository=repository,
            request=request(expected),
            sources=(source(OsmObjectType.NODE, 1),),
        )


def test_reports_are_deterministic_and_secret_free() -> None:
    rows = (
        NocoSourceRecord(
            1,
            {
                "source": "OSM",
                "osm_type": "node",
                "osm_id": "1",
                "email": "private@example.test",
                "raw_tags_json": "secret-like-raw-payload",
            },
        ),
    )
    first = scan(rows, (source(OsmObjectType.NODE, 1),))
    second = scan(rows, (source(OsmObjectType.NODE, 1),))
    assert first.report_hash == second.report_hash
    rendered = repr(first)
    assert "private@example.test" not in rendered
    assert "secret-like-raw-payload" not in rendered
    assert first.read_audit.rows_read == 1
    assert first.read_audit.pages_read == 1


def test_source_and_existing_mapping_bounds_fail_closed() -> None:
    current_schema = schema()
    repository = FixtureNocoReadRepository(schema=current_schema, rows=())
    limited = AdoptionScanRequest(
        run_id="run",
        region_id="thailand",
        table_id=current_schema.table_id,
        expected_schema_fingerprint=current_schema.fingerprint,
        normalization_context=NormalizationContext(country_code="TH"),
        max_sources=1,
    )
    with pytest.raises(ContractViolation, match="max_sources"):
        scan_adoption_candidates(
            repository=repository,
            request=limited,
            sources=(source(OsmObjectType.NODE, 1), source(OsmObjectType.NODE, 2)),
        )
    with pytest.raises(ContractViolation, match="uniqueness"):
        scan_adoption_candidates(
            repository=repository,
            request=request(current_schema),
            sources=(source(OsmObjectType.NODE, 1),),
            existing_mappings=(
                ExistingActiveMapping(1, RegionScopedIdentity("thailand", OsmObjectType.NODE, 1)),
                ExistingActiveMapping(1, RegionScopedIdentity("thailand", OsmObjectType.WAY, 2)),
            ),
        )


class ScriptedRepository:
    def __init__(self, current_schema: NocoSchema, pages: tuple[NocoReadPage, ...]) -> None:
        self._schema = current_schema
        self._pages = pages

    def read_schema(self) -> NocoSchema:
        return self._schema

    def iter_pages(self, read_request: NocoReadRequest) -> Iterator[NocoReadPage]:
        assert isinstance(read_request, NocoReadRequest)
        yield from self._pages


def test_source_iterable_stops_at_the_declared_boundary() -> None:
    current_schema = schema()
    repository = FixtureNocoReadRepository(schema=current_schema, rows=())
    consumed: list[int] = []

    def source_rows() -> Iterator[NormalizedBusiness]:
        for osm_id in (1, 2, 3):
            consumed.append(osm_id)
            yield source(OsmObjectType.NODE, osm_id)

    limited = AdoptionScanRequest(
        run_id="bounded-source-run",
        region_id="thailand",
        table_id=current_schema.table_id,
        expected_schema_fingerprint=current_schema.fingerprint,
        normalization_context=NormalizationContext(country_code="TH"),
        max_sources=1,
    )
    with pytest.raises(ContractViolation, match="max_sources"):
        scan_adoption_candidates(
            repository=repository,
            request=limited,
            sources=source_rows(),
        )
    assert consumed == [1, 2]


@pytest.mark.parametrize(
    ("pages", "message"),
    (
        (
            (
                NocoReadPage((NocoSourceRecord(1, {}),), None),
                NocoReadPage((NocoSourceRecord(2, {}),), None),
            ),
            "after a terminal page",
        ),
        (
            (NocoReadPage((NocoSourceRecord(1, {}),), 1),),
            "must fill its bounded limit",
        ),
        (
            (
                NocoReadPage(
                    (NocoSourceRecord(1, {}), NocoSourceRecord(2, {})),
                    2,
                ),
                NocoReadPage((NocoSourceRecord(2, {}),), None),
            ),
            "cross-page keyset ordering",
        ),
        (
            (
                NocoReadPage(
                    (NocoSourceRecord(1, {}), NocoSourceRecord(2, {})),
                    2,
                ),
            ),
            "ended before its declared continuation",
        ),
    ),
)
def test_malicious_repository_page_sequences_fail_closed(
    pages: tuple[NocoReadPage, ...],
    message: str,
) -> None:
    current_schema = schema()
    repository = ScriptedRepository(current_schema, pages)
    with pytest.raises(ContractViolation, match=message):
        scan_adoption_candidates(
            repository=repository,
            request=request(current_schema),
            sources=(),
        )


def test_per_row_source_fanout_is_bounded() -> None:
    current_schema = schema()
    repository = FixtureNocoReadRepository(
        schema=current_schema,
        rows=(NocoSourceRecord(1, {"name": "Fixture Cafe"}),),
    )
    limited = AdoptionScanRequest(
        run_id="bounded-fanout-run",
        region_id="thailand",
        table_id=current_schema.table_id,
        expected_schema_fingerprint=current_schema.fingerprint,
        normalization_context=NormalizationContext(country_code="TH"),
        max_sources_per_noco_row=1,
    )
    with pytest.raises(ContractViolation, match="max_sources_per_noco_row"):
        scan_adoption_candidates(
            repository=repository,
            request=limited,
            sources=(
                source(
                    OsmObjectType.NODE,
                    1,
                    domain=None,
                    phone=None,
                    address=None,
                ),
                source(
                    OsmObjectType.NODE,
                    2,
                    domain=None,
                    phone=None,
                    address=None,
                ),
            ),
        )


def test_report_hash_covers_workflow_protection_reasons() -> None:
    protected = scan(
        (
            NocoSourceRecord(
                1,
                {
                    "source": "OSM",
                    "osm_type": "node",
                    "osm_id": "1",
                    "Blocked": True,
                },
            ),
        ),
        (source(OsmObjectType.NODE, 1),),
    )
    baseline = scan(
        (
            NocoSourceRecord(
                1,
                {"source": "OSM", "osm_type": "node", "osm_id": "1"},
            ),
        ),
        (source(OsmObjectType.NODE, 1),),
    )
    assert protected.report_hash != baseline.report_hash


def test_duplicate_report_evidence_is_rejected() -> None:
    original = scan(
        (
            NocoSourceRecord(
                1,
                {"source": "OSM", "osm_type": "node", "osm_id": "1"},
            ),
        ),
        (source(OsmObjectType.NODE, 1),),
    )
    with pytest.raises(ContractViolation, match="duplicate candidate evidence"):
        AdoptionScanReport(
            run_id=original.run_id,
            region_id=original.region_id,
            table_id=original.table_id,
            schema_fingerprint=original.schema_fingerprint,
            candidates=original.candidates + original.candidates,
            read_audit=original.read_audit,
        )


def test_candidate_evidence_iterables_are_bounded() -> None:
    identity = RegionScopedIdentity("thailand", OsmObjectType.NODE, 1)
    with pytest.raises(ContractViolation, match="matched_fields"):
        AdoptionCandidate(
            source_identity=identity,
            noco_id=1,
            result_status=CandidateReviewOutcome.AMBIGUOUS,
            proposed_outcome=None,
            rule="bounded-fields",
            confidence_milli=500,
            matched_fields=tuple(f"field-{index}" for index in range(129)),
            protection=None,
        )
    with pytest.raises(ContractViolation, match="collision_reasons"):
        AdoptionCandidate(
            source_identity=identity,
            noco_id=1,
            result_status=CandidateReviewOutcome.COLLISION,
            proposed_outcome=None,
            rule="bounded-collisions",
            confidence_milli=500,
            matched_fields=(),
            protection=None,
            collision_reasons=tuple(f"reason-{index}" for index in range(129)),
        )