import logging
from typing import Optional
from uuid import UUID

from geoalchemy2 import WKTElement
from sqlalchemy import and_, func, select, text
from sqlalchemy.ext.asyncio import AsyncSession

from src.apps.plots.models.plot import Plot
from src.apps.sections.models.section import Section
from src.core.exceptions import ConflictError, NotFoundError, RateLimitError, ValidationError
from src.core.utils.geometry import build_plot_grid, geojson_to_wkt

logger = logging.getLogger(__name__)

# Per-tenant absolute plot ceiling and autogenerate rate limit (security A04 / API4).
_MAX_PLOTS_PER_TENANT = 50_000
_AUTOGEN_WINDOW_SECONDS = 600
_AUTOGEN_MAX_CALLS = 10


class SectionService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def list_sections(
        self,
        tenant_id: UUID,
        skip: int = 0,
        limit: int = 50,
    ) -> tuple[list, int]:
        conditions = [Section.tenant_id == tenant_id]

        count_q = await self.db.execute(
            select(func.count(Section.id)).where(and_(*conditions))
        )
        total = count_q.scalar_one()

        result = await self.db.execute(
            select(Section)
            .where(and_(*conditions))
            .order_by(Section.code.asc())
            .offset(skip)
            .limit(limit)
        )
        return result.scalars().all(), total

    async def get_by_id(
        self, section_id: UUID, tenant_id: UUID
    ) -> Optional[Section]:
        result = await self.db.execute(
            select(Section).where(
                and_(
                    Section.id == section_id,
                    Section.tenant_id == tenant_id,
                )
            )
        )
        return result.scalar_one_or_none()

    async def plot_count(self, section_id: UUID, tenant_id: UUID) -> int:
        result = await self.db.execute(
            select(func.count(Plot.id)).where(
                and_(Plot.section_id == section_id, Plot.tenant_id == tenant_id)
            )
        )
        return int(result.scalar_one())

    async def plot_counts(self, tenant_id: UUID) -> dict:
        """One grouped query → {section_id: plot_count} for the whole tenant (AC-05)."""
        result = await self.db.execute(
            select(Plot.section_id, func.count(Plot.id))
            .where(Plot.tenant_id == tenant_id)
            .group_by(Plot.section_id)
        )
        return {row[0]: int(row[1]) for row in result.all()}

    async def status_counts(self, tenant_id: UUID) -> dict:
        """One grouped query → {section_id: {"total": N, "available": M}} (INDL-41 AC-29).

        `available` counts vacant plots for the filter-bar "(N remaining)" chips;
        the count is server-side, never derived client-side from the full plot list.
        """
        result = await self.db.execute(
            select(Plot.section_id, Plot.status, func.count(Plot.id))
            .where(Plot.tenant_id == tenant_id)
            .group_by(Plot.section_id, Plot.status)
        )
        out: dict = {}
        for section_id, status, cnt in result.all():
            bucket = out.setdefault(section_id, {"total": 0, "available": 0})
            cnt = int(cnt)
            bucket["total"] += cnt
            if status == "vacant":
                bucket["available"] += cnt
        return out

    async def _area_of_wkt(self, wkt: str) -> Optional[float]:
        """Compute geodesic area (m²) of a WKT polygon via PostGIS ST_Area."""
        result = await self.db.execute(
            select(func.ST_Area(func.ST_GeogFromText(wkt)))
        )
        val = result.scalar_one_or_none()
        return round(float(val), 2) if val is not None else None

    async def _assert_within_account_boundary(self, tenant_id: UUID, wkt: str) -> None:
        """INDL-55 E-05 / AC-11 / BR-03: a section boundary must be spatially
        contained within the cemetery (account) boundary when one exists.

        Uses ``ST_CoveredBy`` (not strict ``ST_Contains``) so a section that shares
        an edge with the grounds outline is still accepted (R-02). When the account
        has no boundary yet, creation is *allowed* (BR-04 product choice:
        allow-with-warning; the client surfaces the warning). Cross-tenant leakage
        is impossible — the containment is tested only against the caller's own
        account row (`id = :aid`).
        """
        row = (await self.db.execute(
            text(
                "SELECT boundary IS NOT NULL AS has_boundary, "
                "CASE WHEN boundary IS NOT NULL THEN "
                "  ST_CoveredBy(ST_GeogFromText(:wkt)::geometry, boundary::geometry) "
                "  ELSE NULL END AS contained "
                "FROM accounts WHERE id = :aid"
            ),
            {"wkt": wkt, "aid": str(tenant_id)},
        )).first()
        if row and row[0] and not row[1]:
            raise ValidationError("Section must be inside the cemetery boundary")

    async def _centroid_of_geog(self, section: Section) -> Optional[tuple[float, float]]:
        """Return (lat, lng) of a section's boundary centroid, or None."""
        if section.boundary is None:
            return None
        row = (await self.db.execute(
            text(
                "SELECT ST_Y(ST_Centroid(boundary::geometry)) AS lat, "
                "ST_X(ST_Centroid(boundary::geometry)) AS lng "
                "FROM sections WHERE id = :sid"
            ),
            {"sid": str(section.id)},
        )).first()
        if row and row[0] is not None:
            return float(row[0]), float(row[1])
        return None

    async def create(self, tenant_id: UUID, data: dict) -> Section:
        # Enforce code uniqueness per tenant (case-insensitive).
        existing = await self.db.execute(
            select(Section).where(
                and_(
                    Section.tenant_id == tenant_id,
                    func.lower(Section.code) == data["code"].lower(),
                )
            )
        )
        if existing.scalar_one_or_none():
            raise ConflictError(
                f"A section with code '{data['code']}' already exists in this cemetery"
            )

        # Enforce name uniqueness per tenant (case-insensitive) — PRD validation rule.
        existing_name = await self.db.execute(
            select(Section).where(
                and_(
                    Section.tenant_id == tenant_id,
                    func.lower(Section.name) == data["name"].lower(),
                )
            )
        )
        if existing_name.scalar_one_or_none():
            raise ConflictError("Name already in use")

        section = Section(
            tenant_id=tenant_id,
            code=data["code"],
            name=data["name"],
            description=data.get("description"),
            is_religious=data.get("is_religious", False),
            religious_denomination=data.get("religious_denomination"),
            display_color=data.get("display_color"),
            plot_number_prefix=data.get("plot_number_prefix") or data["code"],
        )

        boundary = data.get("boundary")
        if boundary is not None:
            wkt = geojson_to_wkt(boundary)  # validates (422 on bad geometry)
            await self._assert_within_account_boundary(tenant_id, wkt)  # E-05
            section.boundary = WKTElement(wkt, srid=4326, extended=False)
            section.area_m2 = await self._area_of_wkt(wkt)

        self.db.add(section)
        await self.db.flush()
        return section

    async def update(self, section_id: UUID, tenant_id: UUID, data: dict) -> Section:
        section = await self.get_by_id(section_id, tenant_id)
        if not section:
            raise NotFoundError("Section not found")

        # Plain scalar fields.
        for field in ("name", "description", "is_religious",
                      "religious_denomination", "display_color", "plot_number_prefix"):
            if field in data:
                setattr(section, field, data[field])

        # Boundary: explicit null clears it (and its area); a value replaces it.
        if "boundary" in data:
            boundary = data["boundary"]
            if boundary is None:
                section.boundary = None
                section.area_m2 = None
            else:
                wkt = geojson_to_wkt(boundary)
                await self._assert_within_account_boundary(tenant_id, wkt)  # E-05
                section.boundary = WKTElement(wkt, srid=4326, extended=False)
                section.area_m2 = await self._area_of_wkt(wkt)

        await self.db.flush()
        return section

    async def delete(self, section_id: UUID, tenant_id: UUID) -> None:
        section = await self.get_by_id(section_id, tenant_id)
        if not section:
            raise NotFoundError("Section not found")
        await self.db.delete(section)
        await self.db.flush()

    # ── Grid autogeneration (INDL-49, AC-08/09/10) ──────────────────────────────

    async def _check_autogen_rate_limit(self, tenant_id: UUID) -> None:
        """Redis sliding-ish counter: max _AUTOGEN_MAX_CALLS per window per tenant.

        Fails open (no limit) if Redis is unavailable so the feature still works
        in environments without Redis (e.g. the test suite)."""
        from src.database.session import get_redis

        client = await get_redis()
        if client is None:
            return
        try:
            key = f"autogen_rl:{tenant_id}"
            count = await client.incr(key)
            if count == 1:
                await client.expire(key, _AUTOGEN_WINDOW_SECONDS)
            if count > _AUTOGEN_MAX_CALLS:
                ttl = await client.ttl(key)
                raise RateLimitError(
                    "Too many grid generations — please wait a moment and try again.",
                    retry_after=max(int(ttl), 1),
                )
        finally:
            try:
                await client.aclose()
            except Exception:
                pass

    async def _resolve_origin(
        self, section: Section, tenant_id: UUID,
        req_lat: Optional[float], req_lng: Optional[float],
    ) -> tuple[float, float]:
        if req_lat is not None and req_lng is not None:
            return req_lat, req_lng
        centroid = await self._centroid_of_geog(section)
        if centroid:
            return centroid
        # Fall back to account boundary centroid / last-searched location.
        row = (await self.db.execute(
            text(
                "SELECT ST_Y(ST_Centroid(boundary::geometry)) AS lat, "
                "ST_X(ST_Centroid(boundary::geometry)) AS lng, "
                "map_search_lat, map_search_lng "
                "FROM accounts WHERE id = :aid"
            ),
            {"aid": str(tenant_id)},
        )).first()
        if row:
            if row[0] is not None:
                return float(row[0]), float(row[1])
            if row[2] is not None and row[3] is not None:
                return float(row[2]), float(row[3])
        # No anchor at all — refuse rather than silently generating plots at
        # null-island (0,0). The admin must draw a boundary or pass an origin.
        raise ValidationError(
            "Draw the section's boundary (or the cemetery boundary) before "
            "generating a grid, or provide a grid origin."
        )

    @staticmethod
    def _grid_capacity(section_area_m2: Optional[float], data: dict) -> tuple[float, Optional[int]]:
        """INDL-55 E-07 / AC-16a / BR-07a: how many plots fit in a section by AREA.

        A single plot cell occupies ``(plot_width + gap) × (plot_length + gap)`` m²
        (spacing/padding-aware). ``max_plots_fit`` is the section's measured usable
        area divided by that cell footprint (floored) — an area budget, not a
        geometric-containment guarantee: a valid-area grid on a very elongated
        section could still extend past a narrow edge, so this caps the plot COUNT
        rather than proving every cell lies inside the boundary. Returns
        ``(cell_area, max_fit)``; ``max_fit`` is ``None`` only when the section has
        no measured area at all (no boundary drawn), in which case no cap applies.
        A measured area of exactly 0 (degenerate boundary) yields ``max_fit == 0``
        so any grid is rejected — it is NOT treated as "no cap" (falsy-zero guard).
        """
        gap = float(data.get("gap_m", 0.3) or 0.0)
        cell_area = (float(data["plot_width_m"]) + gap) * (float(data["plot_length_m"]) + gap)
        if section_area_m2 is not None and cell_area > 0:
            return cell_area, int(section_area_m2 // cell_area)
        return cell_area, None

    async def autogenerate_plots(self, section_id: UUID, tenant_id: UUID, data: dict) -> dict:
        rows = data["rows"]
        cols = data["columns"]
        total = rows * cols
        if total > 2000:
            raise ValidationError("Grid is too large — split into multiple generations")

        # ── E-07 preview / dry-run: compute capacity and return without creating.
        # Runs before the row lock and rate limit so previews are free and never
        # burn the tenant's autogenerate budget (security L-2).
        if data.get("preview"):
            section = await self.get_by_id(section_id, tenant_id)
            if not section:
                raise NotFoundError("Section not found")
            section_area = float(section.area_m2) if section.area_m2 is not None else None
            cell_area, max_fit = self._grid_capacity(section_area, data)
            return {
                "preview": True,
                "requested": total,
                "max_plots_fit": max_fit,
                "section_area_m2": section_area,
                "cell_area_m2": round(cell_area, 4),
                "fits": max_fit is None or total <= max_fit,
            }

        # Lock the section row so concurrent generations get disjoint seq ranges (S-08).
        locked = await self.db.execute(
            select(Section)
            .where(and_(Section.id == section_id, Section.tenant_id == tenant_id))
            .with_for_update()
        )
        section = locked.scalar_one_or_none()
        if not section:
            raise NotFoundError("Section not found")

        # ── E-07 / AC-16a / BR-07a: section-capacity cap. Reject a grid whose total
        # plot footprint + spacing exceeds what the section can physically hold, so
        # generated plots never spill outside the section boundary. Skipped when the
        # section has no measured area (no boundary drawn) — nothing to bound against.
        section_area = float(section.area_m2) if section.area_m2 is not None else None
        _, max_fit = self._grid_capacity(section_area, data)
        if max_fit is not None and total > max_fit:
            raise ValidationError(
                "Grid exceeds the section's capacity — reduce rows/columns or plot "
                f"size (section fits at most {max_fit} plots at this size)"
            )

        # Rate-limit only after ownership is confirmed, so a bad/foreign section_id
        # can't burn the tenant's autogenerate budget (security L-2).
        await self._check_autogen_rate_limit(tenant_id)

        # Absolute per-tenant plot ceiling (A04).
        current_total = await self.db.execute(
            select(func.count(Plot.id)).where(Plot.tenant_id == tenant_id)
        )
        if int(current_total.scalar_one()) + total > _MAX_PLOTS_PER_TENANT:
            raise RateLimitError(
                "Plot limit reached for this cemetery — contact support to raise the cap.",
                retry_after=3600,
            )

        prefix = section.plot_number_prefix or section.code
        start_seq = section.next_plot_seq or 1

        # Pre-check for plot_ref collisions in the range (manual labels, re-runs).
        refs = [f"{prefix}-{start_seq + i}" for i in range(total)]
        clash = await self.db.execute(
            select(Plot.plot_ref).where(
                and_(Plot.tenant_id == tenant_id, Plot.plot_ref.in_(refs))
            ).limit(1)
        )
        if clash.scalar_one_or_none():
            raise ConflictError(
                "Some generated plot IDs already exist — adjust the section's "
                "numbering sequence and try again."
            )

        origin_lat, origin_lng = await self._resolve_origin(
            section, tenant_id, data.get("origin_lat"), data.get("origin_lng")
        )

        polygons = build_plot_grid(
            origin_lat=origin_lat,
            origin_lng=origin_lng,
            rows=rows,
            cols=cols,
            plot_width_m=data["plot_width_m"],
            plot_length_m=data["plot_length_m"],
            gap_m=data.get("gap_m", 0.3),
            orientation_deg=data.get("orientation_deg", 0.0),
        )

        created_refs: list[str] = []
        for i, poly in enumerate(polygons):
            ref = refs[i]
            cx, cy = poly.centroid.x, poly.centroid.y  # lng, lat
            plot = Plot(
                tenant_id=tenant_id,
                plot_ref=ref,
                section_id=section.id,
                status="vacant",
                geometry=WKTElement(poly.wkt, srid=4326, extended=False),
                centroid=WKTElement(f"POINT({cx} {cy})", srid=4326, extended=False),
                latitude=round(cy, 7),
                longitude=round(cx, 7),
            )
            self.db.add(plot)
            created_refs.append(ref)

        section.next_plot_seq = start_seq + total
        await self.db.flush()

        ref_range = (
            f"{created_refs[0]} … {created_refs[-1]}"
            if len(created_refs) > 1
            else (created_refs[0] if created_refs else None)
        )
        logger.info(
            "autogenerate_plots tenant=%s section=%s created=%s range=%s",
            tenant_id, section_id, total, ref_range,
        )
        return {
            "created": total,
            "section_id": section_id,
            "plot_ref_range": ref_range,
            "plot_refs": created_refs,
        }