from __future__ import annotations

from collections.abc import Mapping, Sequence

import pytest

from osm_lead_source_service.errors import ContractViolation
from osm_lead_source_service.noco import (
    FixtureNocoReadRepository,
    NocoColumn,
    NocoReadRequest,
    NocoSchema,
    NocoSourceRecord,
    PostgresNocoReadRepository,
    WorkflowProtectionEvidence,
)


def schema() -> 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),
        ),
    )


def test_schema_fingerprint_is_order_independent_and_drift_sensitive() -> None:
    first = schema()
    second = NocoSchema(
        table_id=first.table_id,
        physical_schema=first.physical_schema,
        physical_table=first.physical_table,
        columns=tuple(reversed(first.columns)),
    )
    assert first.fingerprint == second.fingerprint
    changed = NocoSchema(
        table_id=first.table_id,
        physical_schema=first.physical_schema,
        physical_table=first.physical_table,
        columns=first.columns + (NocoColumn("website", "text", True),),
    )
    assert changed.fingerprint != first.fingerprint


def test_schema_rejects_case_insensitive_duplicate_columns() -> None:
    with pytest.raises(ContractViolation, match="unique"):
        NocoSchema(
            table_id="table",
            physical_schema="fixture",
            physical_table="Leads",
            columns=(NocoColumn("source", "text", True), NocoColumn("SOURCE", "text", True)),
        )


def test_schema_and_workflow_reason_iterables_are_bounded() -> None:
    with pytest.raises(ContractViolation, match="bounded nonempty set"):
        NocoSchema(
            table_id="table",
            physical_schema="fixture",
            physical_table="Leads",
            columns=tuple(NocoColumn(f"column_{index}", "text", True) for index in range(513)),
        )
    with pytest.raises(ContractViolation, match="reasons exceed"):
        WorkflowProtectionEvidence(
            noco_id=1,
            protected=True,
            reasons=tuple(f"reason-{index}" for index in range(513)),
        )


def test_fixture_uses_stable_bounded_keyset_pages() -> None:
    repository = FixtureNocoReadRepository(
        schema=schema(),
        rows=(
            NocoSourceRecord(3, {"name": "C"}),
            NocoSourceRecord(1, {"name": "A"}),
            NocoSourceRecord(2, {"name": "B"}),
        ),
    )
    pages = tuple(repository.iter_pages(NocoReadRequest(page_size=1, max_rows=2)))
    assert [page.records[0].noco_id for page in pages] == [1, 2]
    assert pages[0].next_after_id == 1
    assert pages[1].next_after_id is None


def test_fixture_has_no_mutation_surface() -> None:
    repository = FixtureNocoReadRepository(schema=schema(), rows=())
    for name in ("create", "update", "delete", "execute", "commit"):
        assert not hasattr(repository, name)


def test_osm_identity_requires_type_and_preserves_primitive() -> None:
    node = NocoSourceRecord(1, {"source": "OSM", "osm_type": "node", "osm_id": "123"})
    way = NocoSourceRecord(2, {"source": "OSM", "osm_type": "way", "osm_id": "123"})
    assert node.osm_identity() != way.osm_identity()
    via_url = NocoSourceRecord(3, {"osm_url": "https://www.openstreetmap.org/relation/99"})
    assert via_url.osm_identity() is not None
    assert via_url.osm_identity().osm_type.value == "relation"


class FakeExecutor:
    def __init__(self) -> None:
        self.read_only_checks = 0
        self.queries: list[tuple[str, Sequence[object]]] = []
        self.pages = [
            [{"id": 2, "source": "OSM"}],
            [],
        ]

    def assert_transaction_read_only(self) -> None:
        self.read_only_checks += 1

    def fetch_all(
        self, query: str, parameters: Sequence[object] = ()
    ) -> Sequence[Mapping[str, object]]:
        self.queries.append((query, parameters))
        if "information_schema.columns" in query:
            return (
                {"column_name": "id", "data_type": "integer", "is_nullable": "NO"},
                {"column_name": "source", "data_type": "text", "is_nullable": "YES"},
            )
        return self.pages.pop(0)


def test_postgres_adapter_uses_only_read_only_fixed_selects() -> None:
    executor = FakeExecutor()
    repository = PostgresNocoReadRepository(
        executor=executor,
        table_id="table",
        physical_schema="fixture_schema",
        physical_table="Scraper Leads",
    )
    assert repository.read_schema().columns[0].name == "id"
    pages = tuple(repository.iter_pages(NocoReadRequest(page_size=10, max_rows=10)))
    assert pages[0].records[0].noco_id == 2
    assert executor.read_only_checks == 2
    assert all(query.lstrip().upper().startswith("SELECT") for query, _ in executor.queries)
    assert any('ORDER BY "id" ASC LIMIT %s' in query for query, _ in executor.queries)
    for name in ("create", "update", "delete", "execute", "commit"):
        assert not hasattr(repository, name)


class BadOrderExecutor(FakeExecutor):
    def __init__(self) -> None:
        super().__init__()
        self.pages = [[{"id": 2}, {"id": 1}]]


def test_postgres_adapter_rejects_keyset_order_violation() -> None:
    repository = PostgresNocoReadRepository(
        executor=BadOrderExecutor(),
        table_id="table",
        physical_schema="fixture_schema",
        physical_table="Scraper Leads",
    )
    with pytest.raises(ContractViolation, match="keyset"):
        tuple(repository.iter_pages(NocoReadRequest(page_size=10, max_rows=10)))


class ExactBoundaryExecutor(FakeExecutor):
    def __init__(self) -> None:
        super().__init__()
        self.pages = [[{"id": 1}, {"id": 2}]]


class LookaheadExecutor(FakeExecutor):
    def __init__(self) -> None:
        super().__init__()
        self.pages = [
            [{"id": 1}, {"id": 2}, {"id": 3}],
            [{"id": 3}],
        ]


def test_exact_page_boundary_is_terminal_without_an_empty_followup() -> None:
    executor = ExactBoundaryExecutor()
    repository = PostgresNocoReadRepository(
        executor=executor,
        table_id="table",
        physical_schema="fixture_schema",
        physical_table="Scraper Leads",
    )
    pages = tuple(repository.iter_pages(NocoReadRequest(page_size=2, max_rows=10)))
    assert [[record.noco_id for record in page.records] for page in pages] == [[1, 2]]
    assert pages[0].next_after_id is None
    assert executor.read_only_checks == 2


def test_lookahead_row_drives_continuation_without_being_lost() -> None:
    executor = LookaheadExecutor()
    repository = PostgresNocoReadRepository(
        executor=executor,
        table_id="table",
        physical_schema="fixture_schema",
        physical_table="Scraper Leads",
    )
    pages = tuple(repository.iter_pages(NocoReadRequest(page_size=2, max_rows=10)))
    assert [[record.noco_id for record in page.records] for page in pages] == [[1, 2], [3]]
    assert pages[0].next_after_id == 2
    assert pages[1].next_after_id is None
    row_parameters = [
        parameters for query, parameters in executor.queries if 'WHERE "id" > %s' in query
    ]
    assert row_parameters == [(0, 3), (2, 3)]