"""Dashboard aggregation service — INDL-40.

Aggregation queries run sequentially against a single AsyncSession rather than
via asyncio.gather(): SQLAlchemy's AsyncSession is not safe for concurrent use
by multiple coroutines sharing one connection, so running them "in parallel"
would risk interleaved/corrupted results on the same connection. Each
aggregation is instead expressed as a single grouped query to keep round
trips low.
"""
from __future__ import annotations

import json
import logging
from datetime import date, datetime, timedelta, timezone
from typing import Optional

from sqlalchemy import case, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from src.apps.billing.models.invoice_payment import InvoicePayment
from src.apps.dashboard.schemas.dashboard import (
    CapacityBreakdown,
    DashboardResponse,
    KpiBlock,
    SectionCapacity,
    ServiceEventBlock,
    WeeklyServiceDay,
)
from src.apps.memorials.models.tribute import Tribute
from src.apps.plots.models.plot import Plot
from src.apps.scheduling.models.service_event import ServiceEvent
from src.apps.sections.models.section import Section
from src.apps.tenants.models.account import Account
from src.core.config import settings
from src.core.constants import PlotStatus, TributeStatus

logger = logging.getLogger(__name__)

_CACHE_TTL_SECONDS = 60
_DAY_LABELS = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]


class DashboardService:
    def __init__(self, db: AsyncSession):
        self.db = db
        self.last_cache_hit = False

    async def get_dashboard(self, tenant_id: str) -> DashboardResponse:
        cached = await self._cache_get(tenant_id)
        if cached is not None:
            try:
                self.last_cache_hit = True
                return DashboardResponse.model_validate(cached)
            except Exception:
                # A stale cache entry from a prior response shape (e.g. mid-deploy)
                # must not 500 the request — treat it as a cache miss instead.
                logger.warning("dashboard cache entry failed validation for tenant %s", tenant_id, exc_info=True)
                self.last_cache_hit = False
        else:
            self.last_cache_hit = False

        now = datetime.now(timezone.utc)
        today = now.date()

        plot_counts = await self._get_plot_counts(tenant_id)
        revenue_curr, revenue_prev = await self._get_revenue(tenant_id, today)
        plots_delta_7d = await self._get_plots_delta_7d(tenant_id, now)
        upcoming_services_7d = await self._get_upcoming_services_count(tenant_id, today)
        pending_approvals = await self._get_pending_approvals(tenant_id)
        weekly_services = await self._get_weekly_services(tenant_id, today)
        sections = await self._get_section_capacity(tenant_id)
        public_site_url = await self._get_public_site_url(tenant_id)

        total = sum(plot_counts.values())
        interred = plot_counts.get(PlotStatus.OCCUPIED.value, 0)
        reserved = plot_counts.get(PlotStatus.RESERVED.value, 0)
        available = plot_counts.get(PlotStatus.VACANT.value, 0)
        interred_pct = round((interred / total) * 100) if total > 0 else 0
        revenue_delta_pct = self._pct_change(revenue_curr, revenue_prev)

        response = DashboardResponse(
            public_site_url=public_site_url,
            kpis=[
                KpiBlock(
                    key="plots_available",
                    label="Plots available",
                    value=available,
                    delta=self._plots_delta_label(plots_delta_7d),
                    delta_direction="down" if plots_delta_7d < 0 else ("up" if plots_delta_7d > 0 else "neutral"),
                    link="/portal/map",
                ),
                KpiBlock(
                    key="revenue_this_month",
                    label="Revenue this month",
                    value=float(revenue_curr),
                    delta=self._revenue_delta_label(revenue_delta_pct),
                    delta_direction="up" if revenue_delta_pct > 0 else ("down" if revenue_delta_pct < 0 else "neutral"),
                    link="/portal/sales",
                ),
                KpiBlock(
                    key="upcoming_services_7d",
                    label="Upcoming services (7d)",
                    value=upcoming_services_7d,
                    delta="View calendar →",
                    delta_direction="neutral",
                    link="/portal/scheduling",
                ),
                KpiBlock(
                    key="pending_memorial_approvals",
                    label="Pending memorial approvals",
                    value=pending_approvals,
                    delta="Needs review",
                    delta_direction="neutral",
                    link="/portal/memorials?status=pending",
                ),
            ],
            weekly_services=weekly_services,
            capacity=CapacityBreakdown(
                total=total,
                interred=interred,
                reserved=reserved,
                available=available,
                interred_pct=interred_pct,
                sections=sections,
            ),
        )

        await self._cache_set(tenant_id, response)
        return response

    # ── Individual aggregations ──────────────────────────────────────────

    async def _get_plot_counts(self, tenant_id: str) -> dict:
        result = await self.db.execute(
            select(Plot.status, func.count(Plot.id))
            .where(Plot.tenant_id == tenant_id)
            .group_by(Plot.status)
        )
        return {status: count for status, count in result.all()}

    async def _get_plots_delta_7d(self, tenant_id: str, now: datetime) -> int:
        """Net change in available plots over the last 7 days.

        There's no historical snapshot of plot status, so this approximates
        the delta by counting plots that moved OUT of `vacant` in the last 7
        days (a status change bumps `updated_at`), treated as a reduction in
        availability. This is an approximation, not an exact ledger — it can
        over-count if a non-status edit (e.g. a `plot_ref` correction) touches
        an already-non-vacant plot within the window.
        """
        week_ago = now - timedelta(days=7)
        result = await self.db.execute(
            select(func.count(Plot.id)).where(
                Plot.tenant_id == tenant_id,
                Plot.status != PlotStatus.VACANT.value,
                Plot.updated_at >= week_ago,
            )
        )
        return -int(result.scalar_one())

    async def _get_revenue(self, tenant_id: str, today: date) -> tuple:
        curr_month_start = today.replace(day=1)
        prev_month_end = curr_month_start - timedelta(days=1)
        prev_month_start = prev_month_end.replace(day=1)

        result = await self.db.execute(
            select(
                func.coalesce(
                    func.sum(
                        case(
                            (InvoicePayment.received_on >= curr_month_start, InvoicePayment.amount),
                            else_=0,
                        )
                    ),
                    0,
                ),
                func.coalesce(
                    func.sum(
                        case(
                            (
                                InvoicePayment.received_on.between(prev_month_start, prev_month_end),
                                InvoicePayment.amount,
                            ),
                            else_=0,
                        )
                    ),
                    0,
                ),
            ).where(
                InvoicePayment.tenant_id == tenant_id,
                InvoicePayment.received_on >= prev_month_start,
            )
        )
        curr, prev = result.one()
        return float(curr), float(prev)

    async def _get_upcoming_services_count(self, tenant_id: str, today: date) -> int:
        result = await self.db.execute(
            select(func.count(ServiceEvent.id)).where(
                ServiceEvent.tenant_id == tenant_id,
                ServiceEvent.deleted_at.is_(None),
                ServiceEvent.scheduled_date >= today,
                ServiceEvent.scheduled_date <= today + timedelta(days=7),
            )
        )
        return int(result.scalar_one())

    async def _get_pending_approvals(self, tenant_id: str) -> int:
        result = await self.db.execute(
            select(func.count(Tribute.id)).where(
                Tribute.tenant_id == tenant_id,
                Tribute.status == TributeStatus.PENDING.value,
            )
        )
        return int(result.scalar_one())

    async def _get_weekly_services(self, tenant_id: str, today: date) -> list:
        monday = today - timedelta(days=today.weekday())
        sunday = monday + timedelta(days=6)

        result = await self.db.execute(
            select(ServiceEvent)
            .where(
                ServiceEvent.tenant_id == tenant_id,
                ServiceEvent.deleted_at.is_(None),
                ServiceEvent.scheduled_date >= monday,
                ServiceEvent.scheduled_date <= sunday,
            )
            .order_by(ServiceEvent.scheduled_date, ServiceEvent.scheduled_time)
        )
        events_by_date: dict = {}
        for event in result.scalars().all():
            events_by_date.setdefault(event.scheduled_date, []).append(
                ServiceEventBlock(
                    id=str(event.id),
                    time=event.scheduled_time.strftime("%H:%M") if event.scheduled_time else "00:00",
                    type=event.service_type.replace("_", " ").title(),
                    name=event.family_contact_name or "—",
                    service_detail_url=f"/portal/scheduling/{event.id}",
                )
            )

        days = []
        for i in range(7):
            day = monday + timedelta(days=i)
            days.append(
                WeeklyServiceDay(
                    date=day,
                    day_label=_DAY_LABELS[i],
                    day_number=day.day,
                    is_past=day < today,
                    services=events_by_date.get(day, []),
                )
            )
        return days

    async def _get_section_capacity(self, tenant_id: str) -> list:
        result = await self.db.execute(
            select(
                Section.name,
                func.count(Plot.id).label("total"),
                func.count(Plot.id).filter(Plot.status != PlotStatus.VACANT.value).label("filled"),
            )
            .join(Plot, Plot.section_id == Section.id, isouter=True)
            .where(Section.tenant_id == tenant_id)
            .group_by(Section.id, Section.name)
            .order_by(Section.name.asc())
        )
        sections = []
        for name, total, filled in result.all():
            pct = round((filled / total) * 100) if total else 0
            sections.append(SectionCapacity(name=name, fill_pct=pct))
        return sections

    async def _get_public_site_url(self, tenant_id: str) -> str:
        result = await self.db.execute(select(Account.subdomain).where(Account.id == tenant_id))
        subdomain = result.scalar_one_or_none()
        if subdomain is None:
            # A resolved tenant_id should always have a matching Account row —
            # this indicates a tenant-resolution bug elsewhere, not a normal case.
            logger.warning("dashboard: no Account found for resolved tenant_id %s", tenant_id)
            subdomain = "yourcemetery"
        if settings.PUBLIC_SITE_URL_TEMPLATE:
            return settings.PUBLIC_SITE_URL_TEMPLATE.format(subdomain=subdomain)
        return f"https://{subdomain}.{settings.APP_DOMAIN}"

    # ── Formatting helpers ───────────────────────────────────────────────

    @staticmethod
    def _pct_change(curr: float, prev: float) -> float:
        if prev <= 0:
            return 100.0 if curr > 0 else 0.0
        return round(((curr - prev) / prev) * 100)

    @staticmethod
    def _plots_delta_label(delta: int) -> str:
        if delta == 0:
            return "No change this week"
        arrow = "↓" if delta < 0 else "↑"
        return f"{arrow} {abs(delta)} this week"

    @staticmethod
    def _revenue_delta_label(pct: float) -> str:
        if pct == 0:
            return "No change vs. last month"
        arrow = "↑" if pct > 0 else "↓"
        return f"{arrow} {abs(int(pct))}% vs. last month"

    # ── Redis cache (cache-aside, fail-open, 60s TTL — AC-15) ────────────

    @staticmethod
    def _cache_key(tenant_id: str) -> str:
        return f"dashboard:{tenant_id}"

    async def _cache_get(self, tenant_id: str) -> Optional[dict]:
        from src.database.session import get_redis

        client = await get_redis()
        if client is None:
            return None
        try:
            raw = await client.get(self._cache_key(tenant_id))
            return json.loads(raw) if raw else None
        except Exception:
            logger.warning("dashboard cache read failed for tenant %s", tenant_id, exc_info=True)
            return None
        finally:
            try:
                await client.aclose()
            except Exception:
                pass

    async def _cache_set(self, tenant_id: str, response: DashboardResponse) -> None:
        from src.database.session import get_redis

        client = await get_redis()
        if client is None:
            return
        try:
            await client.set(
                self._cache_key(tenant_id),
                response.model_dump_json(),
                ex=_CACHE_TTL_SECONDS,
            )
        except Exception:
            logger.warning("dashboard cache write failed for tenant %s", tenant_id, exc_info=True)
        finally:
            try:
                await client.aclose()
            except Exception:
                pass