"""Bounded Noco read adapters with no mutation or connection-creation surface."""

from __future__ import annotations

import re
from collections.abc import Iterable, Iterator, Mapping, Sequence
from itertools import islice
from typing import Protocol, runtime_checkable

from ..errors import ContractViolation
from .models import NocoColumn, NocoReadPage, NocoReadRequest, NocoSchema, NocoSourceRecord

_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_ ]{0,127}$")

# This projection contains matching evidence and all currently known downstream
# workflow/protection fields. Schema fingerprinting still covers every physical
# column returned by information_schema.
READ_PROJECTION = (
    "id",
    "source",
    "osm_type",
    "osm_id",
    "osm_url",
    "name",
    "category",
    "subcategory",
    "website",
    "phone",
    "email",
    "address",
    "city",
    "area",
    "state_region",
    "postcode",
    "lead_country",
    "country_code",
    "raw_tags_json",
    "place_id",
    "cid",
    "data_id",
    "maps_link",
    "foursquare_id",
    "foursquare_url",
    "Enriched_Luxeillum",
    "Email_Found",
    "Blocked",
    "Enriched_Saas",
    "Blocked_SaaS",
    "Website_Found_By_Scraper",
    "website_last_checked_at",
    "website_enrichment_status",
    "website_enrichment_reason",
    "Website_Review_Candidates",
    "Website_Review_Summary",
    "Website_Review_Status",
    "Website_Review_Selected_URL",
    "Offer_URL",
    "Prospect_URL",
)


@runtime_checkable
class ReadOnlyQueryExecutor(Protocol):
    """Minimal injected SQL read capability; no execute/commit method exists."""

    def assert_transaction_read_only(self) -> None:
        """Fail unless the current database transaction is read-only."""

    def fetch_all(
        self,
        query: str,
        parameters: Sequence[object] = (),
    ) -> Sequence[Mapping[str, object]]:
        """Run one bounded SELECT and return detached scalar mappings."""


def _quote_identifier(value: str) -> str:
    if not isinstance(value, str) or not _IDENTIFIER.fullmatch(value):
        raise ContractViolation("SQL identifier has an unsupported shape")
    return f'"{value}"'


def _record_from_mapping(row: Mapping[str, object]) -> NocoSourceRecord:
    if not isinstance(row, Mapping):
        raise ContractViolation("query row must be a mapping")
    noco_id = row.get("id")
    if isinstance(noco_id, bool) or not isinstance(noco_id, int):
        raise ContractViolation("query row id must be an integer")
    fields = {key: value for key, value in row.items() if key != "id"}
    return NocoSourceRecord(noco_id=noco_id, fields=fields)


class FixtureNocoReadRepository:
    """Deterministic fixture adapter used for safe Phase 2A.5 validation."""

    def __init__(self, *, schema: NocoSchema, rows: Iterable[NocoSourceRecord]) -> None:
        if not isinstance(schema, NocoSchema):
            raise ContractViolation("schema must be NocoSchema")
        records = tuple(islice(iter(rows), 100_001))
        if len(records) > 100_000:
            raise ContractViolation("fixture rows exceed the bounded count")
        if any(not isinstance(record, NocoSourceRecord) for record in records):
            raise ContractViolation("rows must contain NocoSourceRecord values")
        ordered = tuple(sorted(records, key=lambda record: record.noco_id))
        if len({record.noco_id for record in ordered}) != len(ordered):
            raise ContractViolation("fixture Noco ids must be unique")
        self._schema = schema
        self._rows = ordered

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

    def iter_pages(self, request: NocoReadRequest) -> Iterator[NocoReadPage]:
        if not isinstance(request, NocoReadRequest):
            raise ContractViolation("request must be NocoReadRequest")
        remaining = request.max_rows
        index = 0
        while index < len(self._rows) and self._rows[index].noco_id <= request.start_after_id:
            index += 1
        while index < len(self._rows) and remaining > 0:
            page_size = min(request.page_size, remaining)
            records = self._rows[index : index + page_size]
            if not records:
                break
            index += len(records)
            remaining -= len(records)
            has_more = index < len(self._rows) and remaining > 0
            yield NocoReadPage(
                records=records,
                next_after_id=records[-1].noco_id if has_more else None,
            )


class PostgresNocoReadRepository:
    """Injected read-only PostgreSQL adapter using bounded keyset pagination."""

    def __init__(
        self,
        *,
        executor: ReadOnlyQueryExecutor,
        table_id: str,
        physical_schema: str,
        physical_table: str,
    ) -> None:
        if not isinstance(executor, ReadOnlyQueryExecutor):
            raise ContractViolation("executor must implement ReadOnlyQueryExecutor")
        self._executor = executor
        self._table_id = table_id
        self._physical_schema = physical_schema
        self._physical_table = physical_table
        self._qualified_table = (
            f"{_quote_identifier(physical_schema)}.{_quote_identifier(physical_table)}"
        )
        self._projection: str | None = None

    def _assert_read_only(self) -> None:
        self._executor.assert_transaction_read_only()

    def read_schema(self) -> NocoSchema:
        self._assert_read_only()
        rows = self._executor.fetch_all(
            "SELECT column_name, data_type, is_nullable "
            "FROM information_schema.columns "
            "WHERE table_schema = %s AND table_name = %s "
            "ORDER BY ordinal_position",
            (self._physical_schema, self._physical_table),
        )
        if len(rows) > 512:
            raise ContractViolation("schema query returned too many columns")
        columns: list[NocoColumn] = []
        for row in rows:
            try:
                nullable_value = row["is_nullable"]
                columns.append(
                    NocoColumn(
                        name=str(row["column_name"]),
                        db_type=str(row["data_type"]),
                        nullable=str(nullable_value).casefold() == "yes",
                    )
                )
            except KeyError as exc:
                raise ContractViolation("schema query row is missing required metadata") from exc
        schema = NocoSchema(
            table_id=self._table_id,
            physical_schema=self._physical_schema,
            physical_table=self._physical_table,
            columns=tuple(columns),
        )
        available = {column.name for column in schema.columns}
        if "id" not in available:
            raise ContractViolation("fingerprinted Noco schema is missing id")
        selected = tuple(column for column in READ_PROJECTION if column in available)
        self._projection = ", ".join(_quote_identifier(column) for column in selected)
        return schema

    def iter_pages(self, request: NocoReadRequest) -> Iterator[NocoReadPage]:
        if not isinstance(request, NocoReadRequest):
            raise ContractViolation("request must be NocoReadRequest")
        if self._projection is None:
            self.read_schema()
        projection = self._projection
        if projection is None:
            raise ContractViolation("Noco schema projection is unavailable")
        remaining = request.max_rows
        after_id = request.start_after_id
        while remaining > 0:
            limit = min(request.page_size, remaining)
            self._assert_read_only()
            rows = self._executor.fetch_all(
                f"SELECT {projection} FROM {self._qualified_table} "
                f"WHERE {_quote_identifier('id')} > %s "
                f"ORDER BY {_quote_identifier('id')} ASC LIMIT %s",
                (after_id, limit + 1),
            )
            if len(rows) > limit + 1:
                raise ContractViolation("query executor returned more rows than requested")
            records = tuple(_record_from_mapping(row) for row in rows[:limit])
            if not records:
                break
            ids = [record.noco_id for record in records]
            if ids != sorted(ids) or ids[0] <= after_id or len(ids) != len(set(ids)):
                raise ContractViolation("query executor violated ascending keyset pagination")
            remaining -= len(records)
            after_id = records[-1].noco_id
            has_more = len(rows) > limit and remaining > 0
            yield NocoReadPage(records=records, next_after_id=after_id if has_more else None)
            if not has_more:
                break